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) BlockWrite(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                 go func() {
 153                         _, err := io.Copy(buf, HashCheckingReader{req.Reader, md5.New(), req.Hash})
 154                         buf.CloseWithError(err)
 155                 }()
 156                 getReader = buf.NewReader
 157         }
 158         if req.Hash == "" {
 159                 m := md5.New()
 160                 _, err := io.Copy(m, getReader())
 161                 if err != nil {
 162                         return resp, err
 163                 }
 164                 req.Hash = fmt.Sprintf("%x", m.Sum(nil))
 165         }
 166         if req.StorageClasses == nil {
 167                 req.StorageClasses = kc.StorageClasses
 168         }
 169         if req.Replicas == 0 {
 170                 req.Replicas = kc.Want_replicas
 171         }
 172         if req.RequestID == "" {
 173                 req.RequestID = kc.getRequestID()
 174         }
 175         if req.Attempts == 0 {
 176                 req.Attempts = 1 + kc.Retries
 177         }
 178
 179         // Calculate the ordering for uploading to servers
 180         sv := NewRootSorter(kc.WritableLocalRoots(), req.Hash).GetSortedRoots()
 181
 182         // The next server to try contacting
 183         nextServer := 0
 184
 185         // The number of active writers
 186         active := 0
 187
 188         // Used to communicate status from the upload goroutines
 189         uploadStatusChan := make(chan uploadStatus)
 190         defer func() {
 191                 // Wait for any abandoned uploads (e.g., we started
 192                 // two uploads and the first replied with replicas=2)
 193                 // to finish before closing the status channel.
 194                 go func() {
 195                         for active > 0 {
 196                                 <-uploadStatusChan
 197                         }
 198                         close(uploadStatusChan)
 199                 }()
 200         }()
 201
 202         replicasTodo := map[string]int{}
 203         for _, c := range req.StorageClasses {
 204                 replicasTodo[c] = req.Replicas
 205         }
 206
 207         replicasPerThread := kc.replicasPerService
 208         if replicasPerThread < 1 {
 209                 // unlimited or unknown
 210                 replicasPerThread = req.Replicas
 211         }
 212
 213         retriesRemaining := req.Attempts
 214         var retryServers []string
 215
 216         lastError := make(map[string]string)
 217         trackingClasses := len(replicasTodo) > 0
 218
 219         for retriesRemaining > 0 {
 220                 retriesRemaining--
 221                 nextServer = 0
 222                 retryServers = []string{}
 223                 for {
 224                         var classesTodo []string
 225                         var maxConcurrency int
 226                         for sc, r := range replicasTodo {
 227                                 classesTodo = append(classesTodo, sc)
 228                                 if maxConcurrency == 0 || maxConcurrency > r {
 229                                         // Having more than r
 230                                         // writes in flight
 231                                         // would overreplicate
 232                                         // class sc.
 233                                         maxConcurrency = r
 234                                 }
 235                         }
 236                         if !trackingClasses {
 237                                 maxConcurrency = req.Replicas - resp.Replicas
 238                         }
 239                         if maxConcurrency < 1 {
 240                                 // If there are no non-zero entries in
 241                                 // replicasTodo, we're done.
 242                                 break
 243                         }
 244                         for active*replicasPerThread < maxConcurrency {
 245                                 // Start some upload requests
 246                                 if nextServer < len(sv) {
 247                                         DebugPrintf("DEBUG: [%s] Begin upload %s to %s", req.RequestID, req.Hash, sv[nextServer])
 248                                         go kc.uploadToKeepServer(sv[nextServer], req.Hash, classesTodo, getReader(), uploadStatusChan, req.DataSize, req.RequestID)
 249                                         nextServer++
 250                                         active++
 251                                 } else {
 252                                         if active == 0 && retriesRemaining == 0 {
 253                                                 msg := "Could not write sufficient replicas: "
 254                                                 for _, resp := range lastError {
 255                                                         msg += resp + "; "
 256                                                 }
 257                                                 msg = msg[:len(msg)-2]
 258                                                 return resp, InsufficientReplicasError{error: errors.New(msg)}
 259                                         }
 260                                         break
 261                                 }
 262                         }
 263
 264                         DebugPrintf("DEBUG: [%s] Replicas remaining to write: %v active uploads: %v", req.RequestID, replicasTodo, active)
 265                         if active < 1 {
 266                                 break
 267                         }
 268
 269                         // Wait for something to happen.
 270                         status := <-uploadStatusChan
 271                         active--
 272
 273                         if status.statusCode == http.StatusOK {
 274                                 delete(lastError, status.url)
 275                                 resp.Replicas += status.replicasStored
 276                                 if len(status.classesStored) == 0 {
 277                                         // Server doesn't report
 278                                         // storage classes. Give up
 279                                         // trying to track which ones
 280                                         // are satisfied; just rely on
 281                                         // total # replicas.
 282                                         trackingClasses = false
 283                                 }
 284                                 for className, replicas := range status.classesStored {
 285                                         if replicasTodo[className] > replicas {
 286                                                 replicasTodo[className] -= replicas
 287                                         } else {
 288                                                 delete(replicasTodo, className)
 289                                         }
 290                                 }
 291                                 resp.Locator = status.response
 292                         } else {
 293                                 msg := fmt.Sprintf("[%d] %s", status.statusCode, status.response)
 294                                 if len(msg) > 100 {
 295                                         msg = msg[:100]
 296                                 }
 297                                 lastError[status.url] = msg
 298                         }
 299
 300                         if status.statusCode == 0 || status.statusCode == 408 || status.statusCode == 429 ||
 301                                 (status.statusCode >= 500 && status.statusCode != 503) {
 302                                 // Timeout, too many requests, or other server side failure
 303                                 // Do not retry when status code is 503, which means the keep server is full
 304                                 retryServers = append(retryServers, status.url[0:strings.LastIndex(status.url, "/")])
 305                         }
 306                 }
 307
 308                 sv = retryServers
 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 }