1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
19 "git.arvados.org/arvados.git/sdk/go/arvadosclient"
22 // DebugPrintf emits debug messages. The easiest way to enable
23 // keepclient debug messages in your application is to assign
24 // log.Printf to DebugPrintf.
25 var DebugPrintf = func(string, ...interface{}) {}
28 if arvadosclient.StringBool(os.Getenv("ARVADOS_DEBUG")) {
29 DebugPrintf = log.Printf
33 type keepService struct {
34 Uuid string `json:"uuid"`
35 Hostname string `json:"service_host"`
36 Port int `json:"service_port"`
37 SSL bool `json:"service_ssl_flag"`
38 SvcType string `json:"service_type"`
39 ReadOnly bool `json:"read_only"`
42 // Md5String returns md5 hash for the bytes in the given string
43 func Md5String(s string) string {
44 return fmt.Sprintf("%x", md5.Sum([]byte(s)))
48 Items []keepService `json:"items"`
51 type uploadStatus struct {
56 classesStored map[string]int
60 func (kc *KeepClient) uploadToKeepServer(host string, hash string, classesTodo []string, body io.Reader,
61 uploadStatusChan chan<- uploadStatus, expectedLength int64, reqid string) {
65 var url = fmt.Sprintf("%s/%s", host, hash)
66 if req, err = http.NewRequest("PUT", url, nil); err != nil {
67 DebugPrintf("DEBUG: [%s] Error creating request PUT %v error: %v", reqid, url, err.Error())
68 uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, ""}
72 req.ContentLength = expectedLength
73 if expectedLength > 0 {
74 req.Body = ioutil.NopCloser(body)
76 // "For client requests, a value of 0 means unknown if
77 // Body is not nil." In this case we do want the body
78 // to be empty, so don't set req.Body.
81 req.Header.Add("X-Request-Id", reqid)
82 req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
83 req.Header.Add("Content-Type", "application/octet-stream")
84 req.Header.Add(XKeepDesiredReplicas, fmt.Sprint(kc.Want_replicas))
85 if len(classesTodo) > 0 {
86 req.Header.Add(XKeepStorageClasses, strings.Join(classesTodo, ", "))
89 var resp *http.Response
90 if resp, err = kc.httpClient().Do(req); err != nil {
91 DebugPrintf("DEBUG: [%s] Upload failed %v error: %v", reqid, url, err.Error())
92 uploadStatusChan <- uploadStatus{err, url, 0, 0, nil, err.Error()}
97 if xr := resp.Header.Get(XKeepReplicasStored); xr != "" {
98 fmt.Sscanf(xr, "%d", &rep)
100 scc := resp.Header.Get(XKeepStorageClassesConfirmed)
101 classesStored, err := parseStorageClassesConfirmedHeader(scc)
103 DebugPrintf("DEBUG: [%s] Ignoring invalid %s header %q: %s", reqid, XKeepStorageClassesConfirmed, scc, err)
106 defer resp.Body.Close()
107 defer io.Copy(ioutil.Discard, resp.Body)
109 respbody, err2 := ioutil.ReadAll(&io.LimitedReader{R: resp.Body, N: 4096})
110 response := strings.TrimSpace(string(respbody))
111 if err2 != nil && err2 != io.EOF {
112 DebugPrintf("DEBUG: [%s] Upload %v error: %v response: %v", reqid, url, err2.Error(), response)
113 uploadStatusChan <- uploadStatus{err2, url, resp.StatusCode, rep, classesStored, response}
114 } else if resp.StatusCode == http.StatusOK {
115 DebugPrintf("DEBUG: [%s] Upload %v success", reqid, url)
116 uploadStatusChan <- uploadStatus{nil, url, resp.StatusCode, rep, classesStored, response}
118 if resp.StatusCode >= 300 && response == "" {
119 response = resp.Status
121 DebugPrintf("DEBUG: [%s] Upload %v error: %v response: %v", reqid, url, resp.StatusCode, response)
122 uploadStatusChan <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode, rep, classesStored, response}
126 func (kc *KeepClient) putReplicas(
128 getReader func() io.Reader,
129 expectedLength int64) (locator string, replicas int, err error) {
131 reqid := kc.getRequestID()
133 // Calculate the ordering for uploading to servers
134 sv := NewRootSorter(kc.WritableLocalRoots(), hash).GetSortedRoots()
136 // The next server to try contacting
139 // The number of active writers
142 // Used to communicate status from the upload goroutines
143 uploadStatusChan := make(chan uploadStatus)
145 // Wait for any abandoned uploads (e.g., we started
146 // two uploads and the first replied with replicas=2)
147 // to finish before closing the status channel.
152 close(uploadStatusChan)
156 replicasWanted := kc.Want_replicas
157 replicasTodo := map[string]int{}
158 for _, c := range kc.StorageClasses {
159 replicasTodo[c] = replicasWanted
163 replicasPerThread := kc.replicasPerService
164 if replicasPerThread < 1 {
165 // unlimited or unknown
166 replicasPerThread = replicasWanted
169 retriesRemaining := 1 + kc.Retries
170 var retryServers []string
172 lastError := make(map[string]string)
173 trackingClasses := len(replicasTodo) > 0
175 for retriesRemaining > 0 {
178 retryServers = []string{}
180 var classesTodo []string
181 var maxConcurrency int
182 for sc, r := range replicasTodo {
183 classesTodo = append(classesTodo, sc)
184 if maxConcurrency == 0 || maxConcurrency > r {
185 // Having more than r
187 // would overreplicate
192 if !trackingClasses {
193 maxConcurrency = replicasWanted - replicasDone
195 if maxConcurrency < 1 {
196 // If there are no non-zero entries in
197 // replicasTodo, we're done.
200 for active*replicasPerThread < maxConcurrency {
201 // Start some upload requests
202 if nextServer < len(sv) {
203 DebugPrintf("DEBUG: [%s] Begin upload %s to %s", reqid, hash, sv[nextServer])
204 go kc.uploadToKeepServer(sv[nextServer], hash, classesTodo, getReader(), uploadStatusChan, expectedLength, reqid)
208 if active == 0 && retriesRemaining == 0 {
209 msg := "Could not write sufficient replicas: "
210 for _, resp := range lastError {
213 msg = msg[:len(msg)-2]
214 return locator, replicasDone, InsufficientReplicasError(errors.New(msg))
220 DebugPrintf("DEBUG: [%s] Replicas remaining to write: %v active uploads: %v", reqid, replicasTodo, active)
225 // Wait for something to happen.
226 status := <-uploadStatusChan
229 if status.statusCode == http.StatusOK {
230 delete(lastError, status.url)
231 replicasDone += status.replicasStored
232 if len(status.classesStored) == 0 {
233 // Server doesn't report
234 // storage classes. Give up
235 // trying to track which ones
236 // are satisfied; just rely on
238 trackingClasses = false
240 for className, replicas := range status.classesStored {
241 if replicasTodo[className] > replicas {
242 replicasTodo[className] -= replicas
244 delete(replicasTodo, className)
247 locator = status.response
249 msg := fmt.Sprintf("[%d] %s", status.statusCode, status.response)
253 lastError[status.url] = msg
256 if status.statusCode == 0 || status.statusCode == 408 || status.statusCode == 429 ||
257 (status.statusCode >= 500 && status.statusCode != 503) {
258 // Timeout, too many requests, or other server side failure
259 // Do not retry when status code is 503, which means the keep server is full
260 retryServers = append(retryServers, status.url[0:strings.LastIndex(status.url, "/")])
267 return locator, replicasDone, nil
270 func parseStorageClassesConfirmedHeader(hdr string) (map[string]int, error) {
274 classesStored := map[string]int{}
275 for _, cr := range strings.Split(hdr, ",") {
276 cr = strings.TrimSpace(cr)
280 fields := strings.SplitN(cr, "=", 2)
281 if len(fields) != 2 {
282 return nil, fmt.Errorf("expected exactly one '=' char in entry %q", cr)
284 className := fields[0]
286 return nil, fmt.Errorf("empty class name in entry %q", cr)
288 replicas, err := strconv.Atoi(fields[1])
289 if err != nil || replicas < 1 {
290 return nil, fmt.Errorf("invalid replica count %q", fields[1])
292 classesStored[className] = replicas
294 return classesStored, nil