1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
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"
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{}) {}
32 if arvadosclient.StringBool(os.Getenv("ARVADOS_DEBUG")) {
33 DebugPrintf = log.Printf
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"`
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)))
52 Items []keepService `json:"items"`
55 type uploadStatus struct {
60 classesStored map[string]int
64 func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo []string, body io.Reader,
65 uploadStatusChan chan<- uploadStatus, expectedLength int, reqid string) {
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, ""}
76 req.ContentLength = int64(expectedLength)
77 if expectedLength > 0 {
78 req.Body = ioutil.NopCloser(body)
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.
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, ", "))
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()}
101 if xr := resp.Header.Get(XKeepReplicasStored); xr != "" {
102 fmt.Sscanf(xr, "%d", &rep)
104 scc := resp.Header.Get(XKeepStorageClassesConfirmed)
105 classesStored, err := parseStorageClassesConfirmedHeader(scc)
107 DebugPrintf("DEBUG: [%s] Ignoring invalid %s header %q: %s", reqid, XKeepStorageClassesConfirmed, scc, err)
110 defer resp.Body.Close()
111 defer io.Copy(ioutil.Discard, resp.Body)
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}
122 if resp.StatusCode >= 300 && response == "" {
123 response = resp.Status
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}
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")
136 if req.DataSize < 0 {
137 return resp, fmt.Errorf("invalid BlockWriteOptions: negative DataSize %d", req.DataSize)
139 if req.DataSize > BLOCKSIZE || len(req.Data) > BLOCKSIZE {
140 return resp, ErrOversizeBlock
143 if req.DataSize > len(req.Data) {
144 return resp, errors.New("invalid BlockWriteOptions: DataSize > len(Data)")
146 if req.DataSize == 0 {
147 req.DataSize = len(req.Data)
149 getReader = func() io.Reader { return bytes.NewReader(req.Data[:req.DataSize]) }
151 buf := asyncbuf.NewBuffer(make([]byte, 0, req.DataSize))
154 reader = HashCheckingReader{req.Reader, md5.New(), req.Hash}
157 _, err := io.Copy(buf, reader)
158 buf.CloseWithError(err)
160 getReader = buf.NewReader
164 _, err := io.Copy(m, getReader())
168 req.Hash = fmt.Sprintf("%x", m.Sum(nil))
170 if req.StorageClasses == nil {
171 if len(kc.StorageClasses) > 0 {
172 req.StorageClasses = kc.StorageClasses
174 req.StorageClasses = kc.DefaultStorageClasses
177 if req.Replicas == 0 {
178 req.Replicas = kc.Want_replicas
180 if req.RequestID == "" {
181 req.RequestID = kc.getRequestID()
183 if req.Attempts == 0 {
184 req.Attempts = 1 + kc.Retries
187 // Calculate the ordering for uploading to servers
188 sv := NewRootSorter(kc.WritableLocalRoots(), req.Hash).GetSortedRoots()
190 // The next server to try contacting
193 // The number of active writers
196 // Used to communicate status from the upload goroutines
197 uploadStatusChan := make(chan uploadStatus)
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.
206 close(uploadStatusChan)
210 replicasTodo := map[string]int{}
211 for _, c := range req.StorageClasses {
212 replicasTodo[c] = req.Replicas
215 replicasPerThread := kc.replicasPerService
216 if replicasPerThread < 1 {
217 // unlimited or unknown
218 replicasPerThread = req.Replicas
221 retriesRemaining := req.Attempts
222 var retryServers []string
224 lastError := make(map[string]string)
225 trackingClasses := len(replicasTodo) > 0
227 for retriesRemaining > 0 {
230 retryServers = []string{}
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
239 // would overreplicate
244 if !trackingClasses {
245 maxConcurrency = req.Replicas - resp.Replicas
247 if maxConcurrency < 1 {
248 // If there are no non-zero entries in
249 // replicasTodo, we're done.
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)
260 if active == 0 && retriesRemaining == 0 {
261 msg := "Could not write sufficient replicas: "
262 for _, resp := range lastError {
265 msg = msg[:len(msg)-2]
266 return resp, InsufficientReplicasError{error: errors.New(msg)}
272 DebugPrintf("DEBUG: [%s] Replicas remaining to write: %v active uploads: %v", req.RequestID, replicasTodo, active)
277 // Wait for something to happen.
278 status := <-uploadStatusChan
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
290 trackingClasses = false
292 for className, replicas := range status.classesStored {
293 if replicasTodo[className] > replicas {
294 replicasTodo[className] -= replicas
296 delete(replicasTodo, className)
299 resp.Locator = status.response
301 msg := fmt.Sprintf("[%d] %s", status.statusCode, status.response)
305 lastError[status.url] = msg
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, "/")])
322 func parseStorageClassesConfirmedHeader(hdr string) (map[string]int, error) {
326 classesStored := map[string]int{}
327 for _, cr := range strings.Split(hdr, ",") {
328 cr = strings.TrimSpace(cr)
332 fields := strings.SplitN(cr, "=", 2)
333 if len(fields) != 2 {
334 return nil, fmt.Errorf("expected exactly one '=' char in entry %q", cr)
336 className := fields[0]
338 return nil, fmt.Errorf("empty class name in entry %q", cr)
340 replicas, err := strconv.Atoi(fields[1])
341 if err != nil || replicas < 1 {
342 return nil, fmt.Errorf("invalid replica count %q", fields[1])
344 classesStored[className] = replicas
346 return classesStored, nil