services/datamanager/keep/keep.go

   1 /* Deals with getting Keep Server blocks from API Server and Keep Servers. */
   2
   3 package keep
   4
   5 import (
   6         "bufio"
   7         "flag"
   8         "fmt"
   9         //"git.curoverse.com/arvados.git/sdk/go/keepclient"
  10         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  11         "git.curoverse.com/arvados.git/sdk/go/blockdigest"
  12         "git.curoverse.com/arvados.git/sdk/go/manifest"
  13         "git.curoverse.com/arvados.git/sdk/go/util"
  14         "log"
  15         "io/ioutil"
  16         "net/http"
  17         "strconv"
  18         "strings"
  19         "sync"
  20 )
  21
  22 type ServerAddress struct {
  23         Host string `json:"service_host"`
  24         Port int `json:"service_port"`
  25 }
  26
  27 // Info about a particular block returned by the server
  28 type BlockInfo struct {
  29         Digest     blockdigest.BlockDigest
  30         Size       int
  31         Mtime      int  // TODO(misha): Replace this with a timestamp.
  32 }
  33
  34 // Info about a specified block given by a server
  35 type BlockServerInfo struct {
  36         ServerIndex int
  37         Size        int
  38         Mtime       int  // TODO(misha): Replace this with a timestamp.
  39 }
  40
  41 type ServerContents struct {
  42         BlockDigestToInfo map[blockdigest.BlockDigest]BlockInfo
  43 }
  44
  45 type ServerResponse struct {
  46         Address ServerAddress
  47         Contents ServerContents
  48 }
  49
  50 type ReadServers struct {
  51         ReadAllServers            bool
  52         KeepServerIndexToAddress  []ServerAddress
  53         KeepServerAddressToIndex  map[ServerAddress]int
  54         ServerToContents          map[ServerAddress]ServerContents
  55         BlockToServers            map[blockdigest.BlockDigest][]BlockServerInfo
  56         BlockReplicationCounts    map[int]int
  57 }
  58
  59 type GetKeepServersParams struct {
  60         Client arvadosclient.ArvadosClient
  61         Limit int
  62 }
  63
  64 type KeepServiceList struct {
  65         ItemsAvailable int `json:"items_available"`
  66         KeepServers []ServerAddress `json:"items"`
  67 }
  68
  69 // Methods to implement util.SdkListResponse Interface
  70 func (k KeepServiceList) NumItemsAvailable() (numAvailable int, err error) {
  71         return k.ItemsAvailable, nil
  72 }
  73
  74 func (k KeepServiceList) NumItemsContained() (numContained int, err error) {
  75         return len(k.KeepServers), nil
  76 }
  77
  78 var (
  79         // Don't access the token directly, use getDataManagerToken() to
  80         // make sure it's been read.
  81         dataManagerToken                string
  82         dataManagerTokenFile            string
  83         dataManagerTokenFileReadOnce    sync.Once
  84 )
  85
  86 func init() {
  87         flag.StringVar(&dataManagerTokenFile,
  88                 "data-manager-token-file",
  89                 "",
  90                 "File with the API token we should use to contact keep servers.")
  91 }
  92
  93 func getDataManagerToken() (string) {
  94         readDataManagerToken := func () {
  95                 if dataManagerTokenFile == "" {
  96                         flag.Usage()
  97                         log.Fatalf("Data Manager Token needed, but data manager token file not specified.")
  98                 } else {
  99                         rawRead, err := ioutil.ReadFile(dataManagerTokenFile)
 100                         if err != nil {
 101                                 log.Fatalf("Unexpected error reading token file %s: %v",
 102                                         dataManagerTokenFile,
 103                                         err)
 104                         }
 105                         dataManagerToken = strings.TrimSpace(string(rawRead))
 106                 }
 107         }
 108
 109         dataManagerTokenFileReadOnce.Do(readDataManagerToken)
 110         return dataManagerToken
 111 }
 112
 113 func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers) {
 114         results = GetKeepServers(params)
 115         log.Printf("Returned %d keep disks", len(results.ServerToContents))
 116
 117         ComputeBlockReplicationCounts(&results)
 118         log.Printf("Replication level distribution: %v",
 119                 results.BlockReplicationCounts)
 120
 121         return
 122 }
 123
 124 func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
 125         if &params.Client == nil {
 126                 log.Fatalf("params.Client passed to GetKeepServers() should " +
 127                         "contain a valid ArvadosClient, but instead it is nil.")
 128         }
 129
 130         sdkParams := arvadosclient.Dict{}
 131         if params.Limit > 0 {
 132                 sdkParams["limit"] = params.Limit
 133         }
 134
 135         var sdkResponse KeepServiceList
 136         err := params.Client.Call("GET", "keep_services", "", "accessible", sdkParams, &sdkResponse)
 137
 138         if err != nil {
 139                 log.Fatalf("Error requesting keep disks from API server: %v", err)
 140         }
 141
 142         log.Printf("Received keep services list: %v", sdkResponse)
 143
 144         {
 145                 var numReceived, numAvailable int
 146                 results.ReadAllServers, numReceived, numAvailable =
 147                         util.ContainsAllAvailableItems(sdkResponse)
 148
 149                 if (!results.ReadAllServers) {
 150                         log.Printf("ERROR: Did not receive all keep server addresses.")
 151                 }
 152                 log.Printf("Received %d of %d available keep server addresses.",
 153                         numReceived,
 154                         numAvailable)
 155         }
 156
 157         results.KeepServerIndexToAddress = sdkResponse.KeepServers
 158         results.KeepServerAddressToIndex = make(map[ServerAddress]int)
 159         for i, address := range results.KeepServerIndexToAddress {
 160                 results.KeepServerAddressToIndex[address] = i
 161         }
 162
 163         log.Printf("Got Server Addresses: %v", results)
 164
 165         // This is safe for concurrent use
 166         client := http.Client{}
 167
 168         // Send off all the index requests concurrently
 169         responseChan := make(chan ServerResponse)
 170         for _, keepServer := range sdkResponse.KeepServers {
 171                 go GetServerContents(keepServer, client, responseChan)
 172         }
 173
 174         results.ServerToContents = make(map[ServerAddress]ServerContents)
 175         results.BlockToServers = make(map[blockdigest.BlockDigest][]BlockServerInfo)
 176
 177         // Read all the responses
 178         for i := range sdkResponse.KeepServers {
 179                 _ = i  // Here to prevent go from complaining.
 180                 response := <- responseChan
 181                 log.Printf("Received channel response from %v containing %d files",
 182                         response.Address,
 183                         len(response.Contents.BlockDigestToInfo))
 184                 results.ServerToContents[response.Address] = response.Contents
 185                 serverIndex := results.KeepServerAddressToIndex[response.Address]
 186                 for _, blockInfo := range response.Contents.BlockDigestToInfo {
 187                         results.BlockToServers[blockInfo.Digest] = append(
 188                                 results.BlockToServers[blockInfo.Digest],
 189                                 BlockServerInfo{ServerIndex: serverIndex,
 190                                         Size: blockInfo.Size,
 191                                         Mtime: blockInfo.Mtime})
 192                 }
 193         }
 194         return
 195 }
 196
 197 // TODO(misha): Break this function apart into smaller, easier to
 198 // understand functions.
 199 func GetServerContents(keepServer ServerAddress,
 200         client http.Client,
 201         responseChan chan<- ServerResponse) () {
 202         // Create and send request.
 203         url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port)
 204         log.Println("About to fetch keep server contents from " + url)
 205
 206         req, err := http.NewRequest("GET", url, nil)
 207         if err != nil {
 208                 log.Fatalf("Error building http request for %s: %v", url, err)
 209         }
 210
 211         req.Header.Add("Authorization",
 212                 fmt.Sprintf("OAuth2 %s", getDataManagerToken()))
 213
 214         resp, err := client.Do(req)
 215         if err != nil {
 216                 log.Fatalf("Error fetching %s: %v", url, err)
 217         }
 218
 219         // Process response.
 220         if resp.StatusCode != 200 {
 221                 log.Fatalf("Received error code %d in response to request for %s: %s",
 222                         resp.StatusCode, url, resp.Status)
 223         }
 224
 225         response := ServerResponse{}
 226         response.Address = keepServer
 227         response.Contents.BlockDigestToInfo =
 228                 make(map[blockdigest.BlockDigest]BlockInfo)
 229         scanner := bufio.NewScanner(resp.Body)
 230         numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
 231         for scanner.Scan() {
 232                 numLines++
 233                 blockInfo, err := parseBlockInfoFromIndexLine(scanner.Text())
 234                 if err != nil {
 235                         log.Fatalf("Error parsing BlockInfo from index line received from %s: %v",
 236                                 url,
 237                                 err)
 238                 }
 239
 240                 if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
 241                         // This server returned multiple lines containing the same block digest.
 242                         numDuplicates += 1
 243                         if storedBlock.Size != blockInfo.Size {
 244                                 numSizeDisagreements += 1
 245                                 // TODO(misha): Consider failing here.
 246                                 log.Printf("Saw different sizes for the same block on %s: %v %v",
 247                                         url,
 248                                         storedBlock,
 249                                         blockInfo)
 250                         }
 251                         // Keep the block that is bigger, or the block that's newer in
 252                         // the case of a size tie.
 253                         if storedBlock.Size < blockInfo.Size ||
 254                                 (storedBlock.Size == blockInfo.Size &&
 255                                 storedBlock.Mtime < blockInfo.Mtime) {
 256                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 257                         }
 258                 } else {
 259                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 260                 }
 261         }
 262         if err := scanner.Err(); err != nil {
 263                 log.Fatalf("Received error scanning response from %s: %v", url, err)
 264         } else {
 265                 log.Printf("%s contained %d lines with %d duplicates with " +
 266                         "%d size disagreements",
 267                         url,
 268                         numLines,
 269                         numDuplicates,
 270                         numSizeDisagreements)
 271         }
 272         resp.Body.Close()
 273         responseChan <- response
 274 }
 275
 276 func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err error) {
 277         tokens := strings.Fields(indexLine)
 278         if len(tokens) != 2 {
 279                 err = fmt.Errorf("Expected 2 tokens per line but received a " +
 280                         "line containing %v instead.",
 281                         tokens)
 282         }
 283
 284         var locator manifest.BlockLocator
 285         if locator, err = manifest.ParseBlockLocator(tokens[0]); err != nil {
 286                 return
 287         }
 288         if len(locator.Hints) > 0 {
 289                 err = fmt.Errorf("Block locator in index line should not contain hints " +
 290                         "but it does: %v",
 291                         locator)
 292                 return
 293         }
 294
 295         blockInfo.Mtime, err = strconv.Atoi(tokens[1])
 296         if err != nil {
 297                 return
 298         }
 299         blockInfo.Digest = locator.Digest
 300         blockInfo.Size = locator.Size
 301         return
 302 }
 303
 304 func ComputeBlockReplicationCounts(readServers *ReadServers) {
 305         readServers.BlockReplicationCounts = make(map[int]int)
 306         for _, infos := range readServers.BlockToServers {
 307                 replication := len(infos)
 308                 readServers.BlockReplicationCounts[replication] += 1
 309         }
 310 }