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