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