Merge branch '21766-disk-cache-size'
[arvados.git] / 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 }