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