sdk/go/keepclient/support.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package keepclient
   6
   7 import (
   8         "bytes"
   9         "context"
  10         "crypto/md5"
  11         "errors"
  12         "fmt"
  13         "io"
  14         "io/ioutil"
  15         "math/rand"
  16         "net/http"
  17         "strconv"
  18         "strings"
  19         "time"
  20
  21         "git.arvados.org/arvados.git/sdk/go/arvados"
  22         "git.arvados.org/arvados.git/sdk/go/asyncbuf"
  23 )
  24
  25 type keepService struct {
  26         Uuid     string `json:"uuid"`
  27         Hostname string `json:"service_host"`
  28         Port     int    `json:"service_port"`
  29         SSL      bool   `json:"service_ssl_flag"`
  30         SvcType  string `json:"service_type"`
  31         ReadOnly bool   `json:"read_only"`
  32 }
  33
  34 // Md5String returns md5 hash for the bytes in the given string
  35 func Md5String(s string) string {
  36         return fmt.Sprintf("%x", md5.Sum([]byte(s)))
  37 }
  38
  39 type svcList struct {
  40         Items []keepService `json:"items"`
  41 }
  42
  43 type uploadStatus struct {
  44         err            error
  45         url            string
  46         statusCode     int
  47         replicasStored int
  48         classesStored  map[string]int
  49         response       string
  50 }
  51
  52 func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo []string, body io.Reader,
  53         uploadStatusChan chan<- uploadStatus, expectedLength int, reqid string) {
  54
  55         var req *http.Request
  56         var err error
  57         var url = fmt.Sprintf("%s/%s", host, hash)
  58         if req, err = http.NewRequest("PUT", url, nil); err != nil {
  59                 kc.debugf("[%s] Error creating request: PUT %s error: %s", reqid, url, err)
  60                 uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, ""}
  61                 return
  62         }
  63
  64         req.ContentLength = int64(expectedLength)
  65         if expectedLength > 0 {
  66                 req.Body = ioutil.NopCloser(body)
  67         } else {
  68                 // "For client requests, a value of 0 means unknown if
  69                 // Body is not nil."  In this case we do want the body
  70                 // to be empty, so don't set req.Body.
  71         }
  72
  73         req.Header.Add("X-Request-Id", reqid)
  74         req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
  75         req.Header.Add("Content-Type", "application/octet-stream")
  76         req.Header.Add(XKeepDesiredReplicas, fmt.Sprint(kc.Want_replicas))
  77         if len(classesTodo) > 0 {
  78                 req.Header.Add(XKeepStorageClasses, strings.Join(classesTodo, ", "))
  79         }
  80
  81         var resp *http.Response
  82         if resp, err = kc.httpClient().Do(req); err != nil {
  83                 kc.debugf("[%s] Upload failed: %s error: %s", reqid, url, err)
  84                 uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, err.Error()}
  85                 return
  86         }
  87
  88         rep := 1
  89         if xr := resp.Header.Get(XKeepReplicasStored); xr != "" {
  90                 fmt.Sscanf(xr, "%d", &rep)
  91         }
  92         scc := resp.Header.Get(XKeepStorageClassesConfirmed)
  93         classesStored, err := parseStorageClassesConfirmedHeader(scc)
  94         if err != nil {
  95                 kc.debugf("[%s] Ignoring invalid %s header %q: %s", reqid, XKeepStorageClassesConfirmed, scc, err)
  96         }
  97
  98         defer resp.Body.Close()
  99         defer io.Copy(ioutil.Discard, resp.Body)
 100
 101         respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
 102         response := strings.TrimSpace(string(respbody))
 103         if err2 != nil && err2 != io.EOF {
 104                 kc.debugf("[%s] Upload %s error: %s response: %s", reqid, url, err2, response)
 105                 uploadStatusChan <- uploadStatus{err2, url, resp.StatusCode, rep, classesStored, response}
 106         } else if resp.StatusCode == http.StatusOK {
 107                 kc.debugf("[%s] Upload %s success", reqid, url)
 108                 uploadStatusChan <- uploadStatus{nil, url, resp.StatusCode, rep, classesStored, response}
 109         } else {
 110                 if resp.StatusCode >= 300 && response == "" {
 111                         response = resp.Status
 112                 }
 113                 kc.debugf("[%s] Upload %s status: %d %s", reqid, url, resp.StatusCode, response)
 114                 uploadStatusChan <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode, rep, classesStored, response}
 115         }
 116 }
 117
 118 func (kc *KeepClient) httpBlockWrite(ctx context.Context, req arvados.BlockWriteOptions) (arvados.BlockWriteResponse, error) {
 119         var resp arvados.BlockWriteResponse
 120         var getReader func() io.Reader
 121         if req.Data == nil && req.Reader == nil {
 122                 return resp, errors.New("invalid BlockWriteOptions: Data and Reader are both nil")
 123         }
 124         if req.DataSize < 0 {
 125                 return resp, fmt.Errorf("invalid BlockWriteOptions: negative DataSize %d", req.DataSize)
 126         }
 127         if req.DataSize > BLOCKSIZE || len(req.Data) > BLOCKSIZE {
 128                 return resp, ErrOversizeBlock
 129         }
 130         if req.Data != nil {
 131                 if req.DataSize > len(req.Data) {
 132                         return resp, errors.New("invalid BlockWriteOptions: DataSize > len(Data)")
 133                 }
 134                 if req.DataSize == 0 {
 135                         req.DataSize = len(req.Data)
 136                 }
 137                 getReader = func() io.Reader { return bytes.NewReader(req.Data[:req.DataSize]) }
 138         } else {
 139                 buf := asyncbuf.NewBuffer(make([]byte, 0, req.DataSize))
 140                 reader := req.Reader
 141                 if req.Hash != "" {
 142                         reader = HashCheckingReader{req.Reader, md5.New(), req.Hash}
 143                 }
 144                 go func() {
 145                         _, err := io.Copy(buf, reader)
 146                         buf.CloseWithError(err)
 147                 }()
 148                 getReader = buf.NewReader
 149         }
 150         if req.Hash == "" {
 151                 m := md5.New()
 152                 _, err := io.Copy(m, getReader())
 153                 if err != nil {
 154                         return resp, err
 155                 }
 156                 req.Hash = fmt.Sprintf("%x", m.Sum(nil))
 157         }
 158         if req.StorageClasses == nil {
 159                 if len(kc.StorageClasses) > 0 {
 160                         req.StorageClasses = kc.StorageClasses
 161                 } else {
 162                         req.StorageClasses = kc.DefaultStorageClasses
 163                 }
 164         }
 165         if req.Replicas == 0 {
 166                 req.Replicas = kc.Want_replicas
 167         }
 168         if req.RequestID == "" {
 169                 req.RequestID = kc.getRequestID()
 170         }
 171         if req.Attempts == 0 {
 172                 req.Attempts = 1 + kc.Retries
 173         }
 174
 175         // Calculate the ordering for uploading to servers
 176         sv := NewRootSorter(kc.WritableLocalRoots(), req.Hash).GetSortedRoots()
 177
 178         // The next server to try contacting
 179         nextServer := 0
 180
 181         // The number of active writers
 182         active := 0
 183
 184         // Used to communicate status from the upload goroutines
 185         uploadStatusChan := make(chan uploadStatus)
 186         defer func() {
 187                 // Wait for any abandoned uploads (e.g., we started
 188                 // two uploads and the first replied with replicas=2)
 189                 // to finish before closing the status channel.
 190                 go func() {
 191                         for active > 0 {
 192                                 <-uploadStatusChan
 193                         }
 194                         close(uploadStatusChan)
 195                 }()
 196         }()
 197
 198         replicasTodo := map[string]int{}
 199         for _, c := range req.StorageClasses {
 200                 replicasTodo[c] = req.Replicas
 201         }
 202
 203         replicasPerThread := kc.replicasPerService
 204         if replicasPerThread < 1 {
 205                 // unlimited or unknown
 206                 replicasPerThread = req.Replicas
 207         }
 208
 209         delay := delayCalculator{InitialMaxDelay: kc.RetryDelay}
 210         retriesRemaining := req.Attempts
 211         var retryServers []string
 212
 213         lastError := make(map[string]string)
 214         trackingClasses := len(replicasTodo) > 0
 215
 216         for retriesRemaining > 0 {
 217                 retriesRemaining--
 218                 nextServer = 0
 219                 retryServers = []string{}
 220                 for {
 221                         var classesTodo []string
 222                         var maxConcurrency int
 223                         for sc, r := range replicasTodo {
 224                                 classesTodo = append(classesTodo, sc)
 225                                 if maxConcurrency == 0 || maxConcurrency > r {
 226                                         // Having more than r
 227                                         // writes in flight
 228                                         // would overreplicate
 229                                         // class sc.
 230                                         maxConcurrency = r
 231                                 }
 232                         }
 233                         if !trackingClasses {
 234                                 maxConcurrency = req.Replicas - resp.Replicas
 235                         }
 236                         if maxConcurrency < 1 {
 237                                 // If there are no non-zero entries in
 238                                 // replicasTodo, we're done.
 239                                 break
 240                         }
 241                         for active*replicasPerThread < maxConcurrency {
 242                                 // Start some upload requests
 243                                 if nextServer < len(sv) {
 244                                         kc.debugf("[%s] Begin upload %s to %s", req.RequestID, req.Hash, sv[nextServer])
 245                                         go kc.uploadToKeepServer(sv[nextServer], req.Hash, classesTodo, getReader(), uploadStatusChan, req.DataSize, req.RequestID)
 246                                         nextServer++
 247                                         active++
 248                                 } else {
 249                                         if active == 0 && retriesRemaining == 0 {
 250                                                 msg := "Could not write sufficient replicas: "
 251                                                 for _, resp := range lastError {
 252                                                         msg += resp + "; "
 253                                                 }
 254                                                 msg = msg[:len(msg)-2]
 255                                                 return resp, InsufficientReplicasError{error: errors.New(msg)}
 256                                         }
 257                                         break
 258                                 }
 259                         }
 260
 261                         kc.debugf("[%s] Replicas remaining to write: %d active uploads: %d", req.RequestID, replicasTodo, active)
 262                         if active < 1 {
 263                                 break
 264                         }
 265
 266                         // Wait for something to happen.
 267                         status := <-uploadStatusChan
 268                         active--
 269
 270                         if status.statusCode == http.StatusOK {
 271                                 delete(lastError, status.url)
 272                                 resp.Replicas += status.replicasStored
 273                                 if len(status.classesStored) == 0 {
 274                                         // Server doesn't report
 275                                         // storage classes. Give up
 276                                         // trying to track which ones
 277                                         // are satisfied; just rely on
 278                                         // total # replicas.
 279                                         trackingClasses = false
 280                                 }
 281                                 for className, replicas := range status.classesStored {
 282                                         if replicasTodo[className] > replicas {
 283                                                 replicasTodo[className] -= replicas
 284                                         } else {
 285                                                 delete(replicasTodo, className)
 286                                         }
 287                                 }
 288                                 resp.Locator = status.response
 289                         } else {
 290                                 msg := fmt.Sprintf("[%d] %s", status.statusCode, status.response)
 291                                 if len(msg) > 100 {
 292                                         msg = msg[:100]
 293                                 }
 294                                 lastError[status.url] = msg
 295                         }
 296
 297                         if status.statusCode == 0 || status.statusCode == 408 || status.statusCode == 429 ||
 298                                 (status.statusCode >= 500 && status.statusCode != http.StatusInsufficientStorage) {
 299                                 // Timeout, too many requests, or other server side failure
 300                                 // (do not auto-retry status 507 "full")
 301                                 retryServers = append(retryServers, status.url[0:strings.LastIndex(status.url, "/")])
 302                         }
 303                 }
 304
 305                 sv = retryServers
 306                 if len(sv) > 0 {
 307                         time.Sleep(delay.Next())
 308                 }
 309         }
 310
 311         return resp, nil
 312 }
 313
 314 func parseStorageClassesConfirmedHeader(hdr string) (map[string]int, error) {
 315         if hdr == "" {
 316                 return nil, nil
 317         }
 318         classesStored := map[string]int{}
 319         for _, cr := range strings.Split(hdr, ",") {
 320                 cr = strings.TrimSpace(cr)
 321                 if cr == "" {
 322                         continue
 323                 }
 324                 fields := strings.SplitN(cr, "=", 2)
 325                 if len(fields) != 2 {
 326                         return nil, fmt.Errorf("expected exactly one '=' char in entry %q", cr)
 327                 }
 328                 className := fields[0]
 329                 if className == "" {
 330                         return nil, fmt.Errorf("empty class name in entry %q", cr)
 331                 }
 332                 replicas, err := strconv.Atoi(fields[1])
 333                 if err != nil || replicas < 1 {
 334                         return nil, fmt.Errorf("invalid replica count %q", fields[1])
 335                 }
 336                 classesStored[className] = replicas
 337         }
 338         return classesStored, nil
 339 }
 340
 341 // delayCalculator calculates a series of delays for implementing
 342 // exponential backoff with jitter.  The first call to Next() returns
 343 // a random duration between MinimumRetryDelay and the specified
 344 // InitialMaxDelay (or DefaultRetryDelay if 0).  The max delay is
 345 // doubled on each subsequent call to Next(), up to 10x the initial
 346 // max delay.
 347 type delayCalculator struct {
 348         InitialMaxDelay time.Duration
 349         n               int // number of delays returned so far
 350         nextmax         time.Duration
 351         limit           time.Duration
 352 }
 353
 354 func (dc *delayCalculator) Next() time.Duration {
 355         if dc.nextmax <= MinimumRetryDelay {
 356                 // initialize
 357                 if dc.InitialMaxDelay > 0 {
 358                         dc.nextmax = dc.InitialMaxDelay
 359                 } else {
 360                         dc.nextmax = DefaultRetryDelay
 361                 }
 362                 dc.limit = 10 * dc.nextmax
 363         }
 364         d := time.Duration(rand.Float64() * float64(dc.nextmax))
 365         if d < MinimumRetryDelay {
 366                 d = MinimumRetryDelay
 367         }
 368         dc.nextmax *= 2
 369         if dc.nextmax > dc.limit {
 370                 dc.nextmax = dc.limit
 371         }
 372         return d
 373 }