sdk/go/arvados/keep_service.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package arvados
   6
   7 import (
   8         "bufio"
   9         "context"
  10         "fmt"
  11         "io/ioutil"
  12         "net/http"
  13         "strconv"
  14         "strings"
  15         "time"
  16 )
  17
  18 // KeepService is an arvados#keepService record
  19 type KeepService struct {
  20         UUID           string    `json:"uuid"`
  21         ServiceHost    string    `json:"service_host"`
  22         ServicePort    int       `json:"service_port"`
  23         ServiceSSLFlag bool      `json:"service_ssl_flag"`
  24         ServiceType    string    `json:"service_type"`
  25         ReadOnly       bool      `json:"read_only"`
  26         CreatedAt      time.Time `json:"created_at"`
  27         ModifiedAt     time.Time `json:"modified_at"`
  28 }
  29
  30 type KeepMount struct {
  31         UUID           string          `json:"uuid"`
  32         DeviceID       string          `json:"device_id"`
  33         ReadOnly       bool            `json:"read_only"`
  34         Replication    int             `json:"replication"`
  35         StorageClasses map[string]bool `json:"storage_classes"`
  36 }
  37
  38 // KeepServiceList is an arvados#keepServiceList record
  39 type KeepServiceList struct {
  40         Items          []KeepService `json:"items"`
  41         ItemsAvailable int           `json:"items_available"`
  42         Offset         int           `json:"offset"`
  43         Limit          int           `json:"limit"`
  44 }
  45
  46 // KeepServiceIndexEntry is what a keep service's index response tells
  47 // us about a stored block.
  48 type KeepServiceIndexEntry struct {
  49         SizedDigest
  50         // Time of last write, in nanoseconds since Unix epoch
  51         Mtime int64
  52 }
  53
  54 // EachKeepService calls f once for every readable
  55 // KeepService. EachKeepService stops if it encounters an
  56 // error, such as f returning a non-nil error.
  57 func (c *Client) EachKeepService(f func(KeepService) error) error {
  58         params := ResourceListParams{}
  59         for {
  60                 var page KeepServiceList
  61                 err := c.RequestAndDecode(&page, "GET", "arvados/v1/keep_services", nil, params)
  62                 if err != nil {
  63                         return err
  64                 }
  65                 for _, item := range page.Items {
  66                         err = f(item)
  67                         if err != nil {
  68                                 return err
  69                         }
  70                 }
  71                 params.Offset = params.Offset + len(page.Items)
  72                 if params.Offset >= page.ItemsAvailable {
  73                         return nil
  74                 }
  75         }
  76 }
  77
  78 func (s *KeepService) url(path string) string {
  79         var f string
  80         if s.ServiceSSLFlag {
  81                 f = "https://%s:%d/%s"
  82         } else {
  83                 f = "http://%s:%d/%s"
  84         }
  85         return fmt.Sprintf(f, s.ServiceHost, s.ServicePort, path)
  86 }
  87
  88 // String implements fmt.Stringer
  89 func (s *KeepService) String() string {
  90         return s.UUID
  91 }
  92
  93 func (s *KeepService) Mounts(c *Client) ([]KeepMount, error) {
  94         url := s.url("mounts")
  95         req, err := http.NewRequest("GET", url, nil)
  96         if err != nil {
  97                 return nil, err
  98         }
  99         var mounts []KeepMount
 100         err = c.DoAndDecode(&mounts, req)
 101         if err != nil {
 102                 return nil, fmt.Errorf("GET %v: %v", url, err)
 103         }
 104         return mounts, nil
 105 }
 106
 107 // Touch updates the timestamp on the given block.
 108 func (s *KeepService) Touch(ctx context.Context, c *Client, blk string) error {
 109         req, err := http.NewRequest("TOUCH", s.url(blk), nil)
 110         if err != nil {
 111                 return err
 112         }
 113         resp, err := c.Do(req.WithContext(ctx))
 114         if err != nil {
 115                 return err
 116         }
 117         defer resp.Body.Close()
 118         if resp.StatusCode != http.StatusOK {
 119                 body, _ := ioutil.ReadAll(resp.Body)
 120                 return fmt.Errorf("%s %s: %s", resp.Proto, resp.Status, body)
 121         }
 122         return nil
 123 }
 124
 125 // Untrash moves/copies the given block out of trash.
 126 func (s *KeepService) Untrash(ctx context.Context, c *Client, blk string) error {
 127         req, err := http.NewRequest("PUT", s.url("untrash/"+blk), nil)
 128         if err != nil {
 129                 return err
 130         }
 131         resp, err := c.Do(req.WithContext(ctx))
 132         if err != nil {
 133                 return err
 134         }
 135         defer resp.Body.Close()
 136         if resp.StatusCode != http.StatusOK {
 137                 body, _ := ioutil.ReadAll(resp.Body)
 138                 return fmt.Errorf("%s %s: %s", resp.Proto, resp.Status, body)
 139         }
 140         return nil
 141 }
 142
 143 // IndexMount returns an unsorted list of blocks at the given mount point.
 144 func (s *KeepService) IndexMount(ctx context.Context, c *Client, mountUUID string, prefix string) ([]KeepServiceIndexEntry, error) {
 145         return s.index(ctx, c, s.url("mounts/"+mountUUID+"/blocks?prefix="+prefix))
 146 }
 147
 148 // Index returns an unsorted list of blocks that can be retrieved from
 149 // this server.
 150 func (s *KeepService) Index(ctx context.Context, c *Client, prefix string) ([]KeepServiceIndexEntry, error) {
 151         return s.index(ctx, c, s.url("index/"+prefix))
 152 }
 153
 154 func (s *KeepService) index(ctx context.Context, c *Client, url string) ([]KeepServiceIndexEntry, error) {
 155         req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 156         if err != nil {
 157                 return nil, fmt.Errorf("NewRequestWithContext(%v): %v", url, err)
 158         }
 159         resp, err := c.Do(req)
 160         if err != nil {
 161                 return nil, fmt.Errorf("Do(%v): %v", url, err)
 162         } else if resp.StatusCode != 200 {
 163                 return nil, fmt.Errorf("%v: %d %v", url, resp.StatusCode, resp.Status)
 164         }
 165         defer resp.Body.Close()
 166
 167         var entries []KeepServiceIndexEntry
 168         scanner := bufio.NewScanner(resp.Body)
 169         sawEOF := false
 170         for scanner.Scan() {
 171                 if scanner.Err() != nil {
 172                         // If we encounter a read error (timeout,
 173                         // connection failure), stop now and return it
 174                         // below, so it doesn't get masked by the
 175                         // ensuing "badly formatted response" error.
 176                         break
 177                 }
 178                 if sawEOF {
 179                         return nil, fmt.Errorf("Index response contained non-terminal blank line")
 180                 }
 181                 line := scanner.Text()
 182                 if line == "" {
 183                         sawEOF = true
 184                         continue
 185                 }
 186                 fields := strings.Split(line, " ")
 187                 if len(fields) != 2 {
 188                         return nil, fmt.Errorf("Malformed index line %q: %d fields", line, len(fields))
 189                 }
 190                 mtime, err := strconv.ParseInt(fields[1], 10, 64)
 191                 if err != nil {
 192                         return nil, fmt.Errorf("Malformed index line %q: mtime: %v", line, err)
 193                 }
 194                 if mtime < 1e12 {
 195                         // An old version of keepstore is giving us
 196                         // timestamps in seconds instead of
 197                         // nanoseconds. (This threshold correctly
 198                         // handles all times between 1970-01-02 and
 199                         // 33658-09-27.)
 200                         mtime = mtime * 1e9
 201                 }
 202                 entries = append(entries, KeepServiceIndexEntry{
 203                         SizedDigest: SizedDigest(fields[0]),
 204                         Mtime:       mtime,
 205                 })
 206         }
 207         if err := scanner.Err(); err != nil {
 208                 return nil, fmt.Errorf("Error scanning index response: %v", err)
 209         }
 210         if !sawEOF {
 211                 return nil, fmt.Errorf("Index response had no EOF marker")
 212         }
 213         return entries, nil
 214 }