1 /* Internal methods to support keepclient.go */
16 type keepDisk struct {
17 Hostname string `json:"service_host"`
18 Port int `json:"service_port"`
19 SSL bool `json:"service_ssl_flag"`
22 func (this *KeepClient) discoverKeepServers() error {
23 // Construct request of keep disk list
26 if req, err = http.NewRequest("GET", fmt.Sprintf("https://%s/arvados/v1/keep_disks", this.ApiServer), nil); err != nil {
30 // Add api token header
31 req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.ApiToken))
34 var resp *http.Response
35 if resp, err = this.Client.Do(req); err != nil {
40 Items []keepDisk `json:"items"`
44 dec := json.NewDecoder(resp.Body)
46 if err := dec.Decode(&m); err != nil {
50 listed := make(map[string]bool)
51 this.Service_roots = make([]string, 0, len(m.Items))
53 for _, element := range m.Items {
59 // Construct server URL
60 url := fmt.Sprintf("http%s://%s:%d", n, element.Hostname, element.Port)
65 this.Service_roots = append(this.Service_roots, url)
69 // Must be sorted for ShuffledServiceRoots() to produce consistent
71 sort.Strings(this.Service_roots)
76 func (this KeepClient) shuffledServiceRoots(hash string) (pseq []string) {
77 // Build an ordering with which to query the Keep servers based on the
78 // contents of the hash. "hash" is a hex-encoded number at least 8
79 // digits (32 bits) long
81 // seed used to calculate the next keep server from 'pool' to be added
85 // Keep servers still to be added to the ordering
86 pool := make([]string, len(this.Service_roots))
87 copy(pool, this.Service_roots)
89 // output probe sequence
90 pseq = make([]string, 0, len(this.Service_roots))
92 // iterate while there are servers left to be assigned
96 // ran out of digits in the seed
97 if len(pseq) < (len(hash) / 4) {
98 // the number of servers added to the probe
99 // sequence is less than the number of 4-digit
100 // slices in 'hash' so refill the seed with the
102 seed = hash[len(hash)-4:]
107 // Take the next 8 digits (32 bytes) and interpret as an integer,
108 // then modulus with the size of the remaining pool to get the next
110 probe, _ := strconv.ParseUint(seed[0:8], 16, 32)
111 probe %= uint64(len(pool))
113 // Append the selected server to the probe sequence and remove it
115 pseq = append(pseq, pool[probe])
116 pool = append(pool[:probe], pool[probe+1:]...)
118 // Remove the digits just used from the seed
124 type uploadStatus struct {
130 func (this KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser,
131 upload_status chan<- uploadStatus, expectedLength int64) {
133 log.Printf("Uploading to %s", host)
135 var req *http.Request
137 var url = fmt.Sprintf("%s/%s", host, hash)
138 if req, err = http.NewRequest("PUT", url, nil); err != nil {
139 upload_status <- uploadStatus{err, url, 0}
144 if expectedLength > 0 {
145 req.ContentLength = expectedLength
148 req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.ApiToken))
149 req.Header.Add("Content-Type", "application/octet-stream")
152 var resp *http.Response
153 if resp, err = this.Client.Do(req); err != nil {
154 upload_status <- uploadStatus{err, url, 0}
158 if resp.StatusCode == http.StatusOK {
159 upload_status <- uploadStatus{nil, url, resp.StatusCode}
161 upload_status <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode}
165 func (this KeepClient) putReplicas(
167 tr *streamer.AsyncStream,
168 expectedLength int64) (replicas int, err error) {
170 // Calculate the ordering for uploading to servers
171 sv := this.shuffledServiceRoots(hash)
173 // The next server to try contacting
176 // The number of active writers
179 // Used to communicate status from the upload goroutines
180 upload_status := make(chan uploadStatus)
181 defer close(upload_status)
183 // Desired number of replicas
184 remaining_replicas := this.Want_replicas
186 for remaining_replicas > 0 {
187 for active < remaining_replicas {
188 // Start some upload requests
189 if next_server < len(sv) {
190 go this.uploadToKeepServer(sv[next_server], hash, tr.MakeStreamReader(), upload_status, expectedLength)
194 return (this.Want_replicas - remaining_replicas), InsufficientReplicasError
198 // Now wait for something to happen.
199 status := <-upload_status
200 if status.StatusCode == 200 {
202 remaining_replicas -= 1
204 // writing to keep server failed for some reason
205 log.Printf("Keep server put to %v failed with '%v'",
206 status.Url, status.Err)
209 log.Printf("Upload status %v %v %v", status.StatusCode, remaining_replicas, active)
212 return (this.Want_replicas - remaining_replicas), nil