sdk/go/src/arvados.org/keepclient/support.go

   1 /* Internal methods to support keepclient.go */
   2 package keepclient
   3
   4 import (
   5         "arvados.org/buffer"
   6         "encoding/json"
   7         "errors"
   8         "fmt"
   9         "io"
  10         "log"
  11         "net/http"
  12         "sort"
  13         "strconv"
  14 )
  15
  16 type keepDisk struct {
  17         Hostname string `json:"service_host"`
  18         Port     int    `json:"service_port"`
  19         SSL      bool   `json:"service_ssl_flag"`
  20 }
  21
  22 func (this *KeepClient) discoverKeepServers() error {
  23         // Construct request of keep disk list
  24         var req *http.Request
  25         var err error
  26         if req, err = http.NewRequest("GET", fmt.Sprintf("https://%s/arvados/v1/keep_disks", this.ApiServer), nil); err != nil {
  27                 return err
  28         }
  29
  30         // Add api token header
  31         req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.ApiToken))
  32
  33         // Make the request
  34         var resp *http.Response
  35         if resp, err = this.Client.Do(req); err != nil {
  36                 return err
  37         }
  38
  39         type svcList struct {
  40                 Items []keepDisk `json:"items"`
  41         }
  42
  43         // Decode json reply
  44         dec := json.NewDecoder(resp.Body)
  45         var m svcList
  46         if err := dec.Decode(&m); err != nil {
  47                 return err
  48         }
  49
  50         listed := make(map[string]bool)
  51         this.Service_roots = make([]string, 0, len(m.Items))
  52
  53         for _, element := range m.Items {
  54                 n := ""
  55                 if element.SSL {
  56                         n = "s"
  57                 }
  58
  59                 // Construct server URL
  60                 url := fmt.Sprintf("http%s://%s:%d", n, element.Hostname, element.Port)
  61
  62                 // Skip duplicates
  63                 if !listed[url] {
  64                         listed[url] = true
  65                         this.Service_roots = append(this.Service_roots, url)
  66                 }
  67         }
  68
  69         // Must be sorted for ShuffledServiceRoots() to produce consistent
  70         // results.
  71         sort.Strings(this.Service_roots)
  72
  73         return nil
  74 }
  75
  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
  80
  81         // seed used to calculate the next keep server from 'pool' to be added
  82         // to 'pseq'
  83         seed := hash
  84
  85         // Keep servers still to be added to the ordering
  86         pool := make([]string, len(this.Service_roots))
  87         copy(pool, this.Service_roots)
  88
  89         // output probe sequence
  90         pseq = make([]string, 0, len(this.Service_roots))
  91
  92         // iterate while there are servers left to be assigned
  93         for len(pool) > 0 {
  94
  95                 if len(seed) < 8 {
  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
 101                                 // last 4 digits.
 102                                 seed = hash[len(hash)-4:]
 103                         }
 104                         seed += hash
 105                 }
 106
 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
 109                 // selected server.
 110                 probe, _ := strconv.ParseUint(seed[0:8], 16, 32)
 111                 probe %= uint64(len(pool))
 112
 113                 // Append the selected server to the probe sequence and remove it
 114                 // from the pool.
 115                 pseq = append(pseq, pool[probe])
 116                 pool = append(pool[:probe], pool[probe+1:]...)
 117
 118                 // Remove the digits just used from the seed
 119                 seed = seed[8:]
 120         }
 121         return pseq
 122 }
 123
 124 type uploadStatus struct {
 125         Err        error
 126         Url        string
 127         StatusCode int
 128 }
 129
 130 func (this KeepClient) uploadToKeepServer(host string, hash string, body io.ReadCloser,
 131         upload_status chan<- uploadStatus, expectedLength int64) {
 132
 133         log.Printf("Uploading to %s", host)
 134
 135         var req *http.Request
 136         var err error
 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}
 140                 return
 141         }
 142
 143         if expectedLength > 0 {
 144                 req.ContentLength = expectedLength
 145         }
 146
 147         req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", this.ApiToken))
 148         req.Header.Add("Content-Type", "application/octet-stream")
 149         req.Body = body
 150
 151         var resp *http.Response
 152         if resp, err = this.Client.Do(req); err != nil {
 153                 upload_status <- uploadStatus{err, url, 0}
 154                 return
 155         }
 156
 157         if resp.StatusCode == http.StatusOK {
 158                 upload_status <- uploadStatus{nil, url, resp.StatusCode}
 159         } else {
 160                 upload_status <- uploadStatus{errors.New(resp.Status), url, resp.StatusCode}
 161         }
 162 }
 163
 164 func (this KeepClient) putReplicas(
 165         hash string,
 166         tr buffer.TransferBuffer,
 167         expectedLength int64) (replicas int, err error) {
 168
 169         // Calculate the ordering for uploading to servers
 170         sv := this.shuffledServiceRoots(hash)
 171
 172         // The next server to try contacting
 173         next_server := 0
 174
 175         // The number of active writers
 176         active := 0
 177
 178         // Used to communicate status from the upload goroutines
 179         upload_status := make(chan uploadStatus)
 180         defer close(upload_status)
 181
 182         // Desired number of replicas
 183         remaining_replicas := this.Want_replicas
 184
 185         for remaining_replicas > 0 {
 186                 for active < remaining_replicas {
 187                         // Start some upload requests
 188                         if next_server < len(sv) {
 189                                 go this.uploadToKeepServer(sv[next_server], hash, tr.MakeBufferReader(), upload_status, expectedLength)
 190                                 next_server += 1
 191                                 active += 1
 192                         } else {
 193                                 return (this.Want_replicas - remaining_replicas), InsufficientReplicasError
 194                         }
 195                 }
 196
 197                 // Now wait for something to happen.
 198                 select {
 199                 case status := <-tr.Reader_status:
 200                         if status == io.EOF {
 201                                 // good news!
 202                         } else {
 203                                 // bad news
 204                                 return (this.Want_replicas - remaining_replicas), status
 205                         }
 206                 case status := <-upload_status:
 207                         if status.StatusCode == 200 {
 208                                 // good news!
 209                                 remaining_replicas -= 1
 210                         } else {
 211                                 // writing to keep server failed for some reason
 212                                 log.Printf("Keep server put to %v failed with '%v'",
 213                                         status.Url, status.Err)
 214                         }
 215                         active -= 1
 216                         log.Printf("Upload status %v %v %v", status.StatusCode, remaining_replicas, active)
 217                 }
 218         }
 219
 220         return (this.Want_replicas - remaining_replicas), nil
 221 }