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