services/keep-balance/keep_service.go

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