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