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/services/datamanager/loggerutil"
  14         "io/ioutil"
  15         "log"
  16         "net/http"
  17         "strconv"
  18         "strings"
  19         "sync"
  20         "time"
  21 )
  22
  23 type ServerAddress struct {
  24         Host string `json:"service_host"`
  25         Port int    `json:"service_port"`
  26         Uuid string `json:"uuid"`
  27 }
  28
  29 // Info about a particular block returned by the server
  30 type BlockInfo struct {
  31         Digest blockdigest.DigestWithSize
  32         Mtime  int64 // TODO(misha): Replace this with a timestamp.
  33 }
  34
  35 // Info about a specified block given by a server
  36 type BlockServerInfo struct {
  37         ServerIndex int
  38         Mtime       int64 // TODO(misha): Replace this with a timestamp.
  39 }
  40
  41 type ServerContents struct {
  42         BlockDigestToInfo map[blockdigest.DigestWithSize]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.DigestWithSize][]BlockServerInfo
  56         BlockReplicationCounts   map[int]int
  57 }
  58
  59 type GetKeepServersParams struct {
  60         Client arvadosclient.ArvadosClient
  61         Logger *logger.Logger
  62         Limit  int
  63 }
  64
  65 type KeepServiceList struct {
  66         ItemsAvailable int             `json:"items_available"`
  67         KeepServers    []ServerAddress `json:"items"`
  68 }
  69
  70 var (
  71         // Don't access the token directly, use getDataManagerToken() to
  72         // make sure it's been read.
  73         dataManagerToken             string
  74         dataManagerTokenFile         string
  75         dataManagerTokenFileReadOnce sync.Once
  76 )
  77
  78 func init() {
  79         flag.StringVar(&dataManagerTokenFile,
  80                 "data-manager-token-file",
  81                 "",
  82                 "File with the API token we should use to contact keep servers.")
  83 }
  84
  85 // TODO(misha): Change this to include the UUID as well.
  86 func (s ServerAddress) String() string {
  87         return s.HostPort()
  88 }
  89
  90 func (s ServerAddress) HostPort() string {
  91         return fmt.Sprintf("%s:%d", s.Host, s.Port)
  92 }
  93
  94 func getDataManagerToken(arvLogger *logger.Logger) string {
  95         readDataManagerToken := func() {
  96                 if dataManagerTokenFile == "" {
  97                         flag.Usage()
  98                         loggerutil.FatalWithMessage(arvLogger,
  99                                 "Data Manager Token needed, but data manager token file not specified.")
 100                 } else {
 101                         rawRead, err := ioutil.ReadFile(dataManagerTokenFile)
 102                         if err != nil {
 103                                 loggerutil.FatalWithMessage(arvLogger,
 104                                         fmt.Sprintf("Unexpected error reading token file %s: %v",
 105                                                 dataManagerTokenFile,
 106                                                 err))
 107                         }
 108                         dataManagerToken = strings.TrimSpace(string(rawRead))
 109                 }
 110         }
 111
 112         dataManagerTokenFileReadOnce.Do(readDataManagerToken)
 113         return dataManagerToken
 114 }
 115
 116 func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers) {
 117         results = GetKeepServers(params)
 118         log.Printf("Returned %d keep disks", len(results.ServerToContents))
 119
 120         results.Summarize(params.Logger)
 121         log.Printf("Replication level distribution: %v",
 122                 results.BlockReplicationCounts)
 123
 124         return
 125 }
 126
 127 func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
 128         if &params.Client == nil {
 129                 log.Fatalf("params.Client passed to GetKeepServers() should " +
 130                         "contain a valid ArvadosClient, but instead it is nil.")
 131         }
 132
 133         sdkParams := arvadosclient.Dict{
 134                 "filters": [][]string{[]string{"service_type", "=", "disk"}},
 135         }
 136         if params.Limit > 0 {
 137                 sdkParams["limit"] = params.Limit
 138         }
 139
 140         var sdkResponse KeepServiceList
 141         err := params.Client.List("keep_services", sdkParams, &sdkResponse)
 142
 143         if err != nil {
 144                 loggerutil.FatalWithMessage(params.Logger,
 145                         fmt.Sprintf("Error requesting keep disks from API server: %v", err))
 146         }
 147
 148         if params.Logger != nil {
 149                 params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 150                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 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         }
 156
 157         log.Printf("Received keep services list: %+v", sdkResponse)
 158
 159         if len(sdkResponse.KeepServers) < sdkResponse.ItemsAvailable {
 160                 loggerutil.FatalWithMessage(params.Logger,
 161                         fmt.Sprintf("Did not receive all available keep servers: %+v", sdkResponse))
 162         }
 163
 164         results.KeepServerIndexToAddress = sdkResponse.KeepServers
 165         results.KeepServerAddressToIndex = make(map[ServerAddress]int)
 166         for i, address := range results.KeepServerIndexToAddress {
 167                 results.KeepServerAddressToIndex[address] = i
 168         }
 169
 170         log.Printf("Got Server Addresses: %v", results)
 171
 172         // This is safe for concurrent use
 173         client := http.Client{}
 174
 175         // Send off all the index requests concurrently
 176         responseChan := make(chan ServerResponse)
 177         for _, keepServer := range sdkResponse.KeepServers {
 178                 // The above keepsServer variable is reused for each iteration, so
 179                 // it would be shared across all goroutines. This would result in
 180                 // us querying one server n times instead of n different servers
 181                 // as we intended. To avoid this we add it as an explicit
 182                 // parameter which gets copied. This bug and solution is described
 183                 // in https://golang.org/doc/effective_go.html#channels
 184                 go func(keepServer ServerAddress) {
 185                         responseChan <- GetServerContents(params.Logger,
 186                                 keepServer,
 187                                 client)
 188                 }(keepServer)
 189         }
 190
 191         results.ServerToContents = make(map[ServerAddress]ServerContents)
 192         results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo)
 193
 194         // Read all the responses
 195         for i := range sdkResponse.KeepServers {
 196                 _ = i // Here to prevent go from complaining.
 197                 response := <-responseChan
 198                 log.Printf("Received channel response from %v containing %d files",
 199                         response.Address,
 200                         len(response.Contents.BlockDigestToInfo))
 201                 results.ServerToContents[response.Address] = response.Contents
 202                 serverIndex := results.KeepServerAddressToIndex[response.Address]
 203                 for _, blockInfo := range response.Contents.BlockDigestToInfo {
 204                         results.BlockToServers[blockInfo.Digest] = append(
 205                                 results.BlockToServers[blockInfo.Digest],
 206                                 BlockServerInfo{ServerIndex: serverIndex,
 207                                         Mtime: blockInfo.Mtime})
 208                 }
 209         }
 210         return
 211 }
 212
 213 func GetServerContents(arvLogger *logger.Logger,
 214         keepServer ServerAddress,
 215         client http.Client) (response ServerResponse) {
 216
 217         GetServerStatus(arvLogger, keepServer, client)
 218
 219         req := CreateIndexRequest(arvLogger, keepServer)
 220         resp, err := client.Do(req)
 221         if err != nil {
 222                 loggerutil.FatalWithMessage(arvLogger,
 223                         fmt.Sprintf("Error fetching %s: %v. Response was %+v",
 224                                 req.URL.String(),
 225                                 err,
 226                                 resp))
 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 := logger.GetOrCreateMap(p, "keep_info")
 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 := logger.GetOrCreateMap(p, "keep_info")
 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 := logger.GetOrCreateMap(p, "keep_info")
 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 := logger.GetOrCreateMap(p, "keep_info")
 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.DigestWithSize]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                         // Keep the block that's newer.
 348                         if storedBlock.Mtime < blockInfo.Mtime {
 349                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 350                         }
 351                 } else {
 352                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 353                 }
 354         }
 355         if err := scanner.Err(); err != nil {
 356                 loggerutil.FatalWithMessage(arvLogger,
 357                         fmt.Sprintf("Received error scanning index response from %s: %v",
 358                                 keepServer.String(),
 359                                 err))
 360         } else {
 361                 log.Printf("%s index contained %d lines with %d duplicates with "+
 362                         "%d size disagreements",
 363                         keepServer.String(),
 364                         numLines,
 365                         numDuplicates,
 366                         numSizeDisagreements)
 367
 368                 if arvLogger != nil {
 369                         now := time.Now()
 370                         arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 371                                 keepInfo := logger.GetOrCreateMap(p, "keep_info")
 372                                 serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 373
 374                                 serverInfo["processing_finished_at"] = now
 375                                 serverInfo["lines_received"] = numLines
 376                                 serverInfo["duplicates_seen"] = numDuplicates
 377                                 serverInfo["size_disagreements_seen"] = numSizeDisagreements
 378                         })
 379                 }
 380         }
 381         resp.Body.Close()
 382         return
 383 }
 384
 385 func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err error) {
 386         tokens := strings.Fields(indexLine)
 387         if len(tokens) != 2 {
 388                 err = fmt.Errorf("Expected 2 tokens per line but received a "+
 389                         "line containing %v instead.",
 390                         tokens)
 391         }
 392
 393         var locator blockdigest.BlockLocator
 394         if locator, err = blockdigest.ParseBlockLocator(tokens[0]); err != nil {
 395                 err = fmt.Errorf("%v Received error while parsing line \"%s\"",
 396                         err, indexLine)
 397                 return
 398         }
 399         if len(locator.Hints) > 0 {
 400                 err = fmt.Errorf("Block locator in index line should not contain hints "+
 401                         "but it does: %v",
 402                         locator)
 403                 return
 404         }
 405
 406         blockInfo.Mtime, err = strconv.ParseInt(tokens[1], 10, 64)
 407         if err != nil {
 408                 return
 409         }
 410         blockInfo.Digest =
 411                 blockdigest.DigestWithSize{Digest: locator.Digest,
 412                         Size: uint32(locator.Size)}
 413         return
 414 }
 415
 416 func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) {
 417         readServers.BlockReplicationCounts = make(map[int]int)
 418         for _, infos := range readServers.BlockToServers {
 419                 replication := len(infos)
 420                 readServers.BlockReplicationCounts[replication] += 1
 421         }
 422
 423         if arvLogger != nil {
 424                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 425                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 426                         keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers)
 427                 })
 428         }
 429
 430 }