services/keep-balance/keep_service.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "io"
   7         "io/ioutil"
   8         "net/http"
   9
  10         "git.curoverse.com/arvados.git/sdk/go/arvados"
  11 )
  12
  13 // KeepService represents a keepstore server that is being rebalanced.
  14 type KeepService struct {
  15         arvados.KeepService
  16         *ChangeSet
  17 }
  18
  19 // String implements fmt.Stringer.
  20 func (srv *KeepService) String() string {
  21         return fmt.Sprintf("%s (%s:%d, %s)", srv.UUID, srv.ServiceHost, srv.ServicePort, srv.ServiceType)
  22 }
  23
  24 var ksSchemes = map[bool]string{false: "http", true: "https"}
  25
  26 // URLBase returns scheme://host:port for this server.
  27 func (srv *KeepService) URLBase() string {
  28         return fmt.Sprintf("%s://%s:%d", ksSchemes[srv.ServiceSSLFlag], srv.ServiceHost, srv.ServicePort)
  29 }
  30
  31 // CommitPulls sends the current list of pull requests to the storage
  32 // server (even if the list is empty).
  33 func (srv *KeepService) CommitPulls(c *arvados.Client) error {
  34         return srv.put(c, "pull", srv.ChangeSet.Pulls)
  35 }
  36
  37 // CommitTrash sends the current list of trash requests to the storage
  38 // server (even if the list is empty).
  39 func (srv *KeepService) CommitTrash(c *arvados.Client) error {
  40         return srv.put(c, "trash", srv.ChangeSet.Trashes)
  41 }
  42
  43 // Perform a PUT request at path, with data (as JSON) in the request
  44 // body.
  45 func (srv *KeepService) put(c *arvados.Client, path string, data interface{}) error {
  46         // We'll start a goroutine to do the JSON encoding, so we can
  47         // stream it to the http client through a Pipe, rather than
  48         // keeping the entire encoded version in memory.
  49         jsonR, jsonW := io.Pipe()
  50
  51         // errC communicates any encoding errors back to our main
  52         // goroutine.
  53         errC := make(chan error, 1)
  54
  55         go func() {
  56                 enc := json.NewEncoder(jsonW)
  57                 errC <- enc.Encode(data)
  58                 jsonW.Close()
  59         }()
  60
  61         url := srv.URLBase() + "/" + path
  62         req, err := http.NewRequest("PUT", url, ioutil.NopCloser(jsonR))
  63         if err != nil {
  64                 return fmt.Errorf("building request for %s: %v", url, err)
  65         }
  66         err = c.DoAndDecode(nil, req)
  67
  68         // If there was an error encoding the request body, report
  69         // that instead of the response: obviously we won't get a
  70         // useful response if our request wasn't properly encoded.
  71         if encErr := <-errC; encErr != nil {
  72                 return fmt.Errorf("encoding data for %s: %v", url, encErr)
  73         }
  74
  75         return err
  76 }