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 }
  28
  29 // Info about a particular block returned by the server
  30 type BlockInfo struct {
  31         Digest blockdigest.BlockDigest
  32         Size   int
  33         Mtime  int64 // 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       int64 // 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 var (
  73         // Don't access the token directly, use getDataManagerToken() to
  74         // make sure it's been read.
  75         dataManagerToken             string
  76         dataManagerTokenFile         string
  77         dataManagerTokenFileReadOnce sync.Once
  78 )
  79
  80 func init() {
  81         flag.StringVar(&dataManagerTokenFile,
  82                 "data-manager-token-file",
  83                 "",
  84                 "File with the API token we should use to contact keep servers.")
  85 }
  86
  87 func (s ServerAddress) String() string {
  88         return fmt.Sprintf("%s:%d", s.Host, s.Port)
  89 }
  90
  91 func getDataManagerToken(arvLogger *logger.Logger) string {
  92         readDataManagerToken := func() {
  93                 if dataManagerTokenFile == "" {
  94                         flag.Usage()
  95                         loggerutil.FatalWithMessage(arvLogger,
  96                                 "Data Manager Token needed, but data manager token file not specified.")
  97                 } else {
  98                         rawRead, err := ioutil.ReadFile(dataManagerTokenFile)
  99                         if err != nil {
 100                                 loggerutil.FatalWithMessage(arvLogger,
 101                                         fmt.Sprintf("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                 "filters": [][]string{[]string{"service_type", "=", "disk"}},
 132         }
 133         if params.Limit > 0 {
 134                 sdkParams["limit"] = params.Limit
 135         }
 136
 137         var sdkResponse KeepServiceList
 138         err := params.Client.List("keep_services", sdkParams, &sdkResponse)
 139
 140         if err != nil {
 141                 loggerutil.FatalWithMessage(params.Logger,
 142                         fmt.Sprintf("Error requesting keep disks from API server: %v", err))
 143         }
 144
 145         if params.Logger != nil {
 146                 params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 147                         keepInfo := make(map[string]interface{})
 148
 149                         keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable
 150                         keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers)
 151                         keepInfo["keep_servers"] = sdkResponse.KeepServers
 152
 153                         p["keep_info"] = keepInfo
 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.BlockDigest][]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                                         Size:  blockInfo.Size,
 208                                         Mtime: blockInfo.Mtime})
 209                 }
 210         }
 211         return
 212 }
 213
 214 func GetServerContents(arvLogger *logger.Logger,
 215         keepServer ServerAddress,
 216         client http.Client) (response ServerResponse) {
 217
 218         GetServerStatus(arvLogger, keepServer, client)
 219
 220         req := CreateIndexRequest(arvLogger, keepServer)
 221         resp, err := client.Do(req)
 222         if err != nil {
 223                 loggerutil.FatalWithMessage(arvLogger,
 224                         fmt.Sprintf("Error fetching %s: %v", req.URL.String(), err))
 225         }
 226
 227         return ReadServerResponse(arvLogger, keepServer, resp)
 228 }
 229
 230 func GetServerStatus(arvLogger *logger.Logger,
 231         keepServer ServerAddress,
 232         client http.Client) {
 233         url := fmt.Sprintf("http://%s:%d/status.json",
 234                 keepServer.Host,
 235                 keepServer.Port)
 236
 237         if arvLogger != nil {
 238                 now := time.Now()
 239                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 240                         keepInfo := p["keep_info"].(map[string]interface{})
 241                         serverInfo := make(map[string]interface{})
 242                         serverInfo["time_status_request_sent"] = now
 243
 244                         keepInfo[keepServer.String()] = serverInfo
 245                 })
 246         }
 247
 248         resp, err := client.Get(url)
 249         if err != nil {
 250                 loggerutil.FatalWithMessage(arvLogger,
 251                         fmt.Sprintf("Error getting keep status from %s: %v", url, err))
 252         } else if resp.StatusCode != 200 {
 253                 loggerutil.FatalWithMessage(arvLogger,
 254                         fmt.Sprintf("Received error code %d in response to request "+
 255                                 "for %s status: %s",
 256                                 resp.StatusCode, url, resp.Status))
 257         }
 258
 259         var keepStatus map[string]interface{}
 260         decoder := json.NewDecoder(resp.Body)
 261         decoder.UseNumber()
 262         err = decoder.Decode(&keepStatus)
 263         if err != nil {
 264                 loggerutil.FatalWithMessage(arvLogger,
 265                         fmt.Sprintf("Error decoding keep status from %s: %v", url, err))
 266         }
 267
 268         if arvLogger != nil {
 269                 now := time.Now()
 270                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 271                         keepInfo := p["keep_info"].(map[string]interface{})
 272                         serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
 273                         serverInfo["time_status_response_processed"] = now
 274                         serverInfo["status"] = keepStatus
 275                 })
 276         }
 277 }
 278
 279 func CreateIndexRequest(arvLogger *logger.Logger,
 280         keepServer ServerAddress) (req *http.Request) {
 281         url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port)
 282         log.Println("About to fetch keep server contents from " + url)
 283
 284         if arvLogger != nil {
 285                 now := time.Now()
 286                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 287                         keepInfo := p["keep_info"].(map[string]interface{})
 288                         serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
 289                         serverInfo["time_index_request_sent"] = now
 290                 })
 291         }
 292
 293         req, err := http.NewRequest("GET", url, nil)
 294         if err != nil {
 295                 loggerutil.FatalWithMessage(arvLogger,
 296                         fmt.Sprintf("Error building http request for %s: %v", url, err))
 297         }
 298
 299         req.Header.Add("Authorization",
 300                 fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger)))
 301         return
 302 }
 303
 304 func ReadServerResponse(arvLogger *logger.Logger,
 305         keepServer ServerAddress,
 306         resp *http.Response) (response ServerResponse) {
 307
 308         if resp.StatusCode != 200 {
 309                 loggerutil.FatalWithMessage(arvLogger,
 310                         fmt.Sprintf("Received error code %d in response to request "+
 311                                 "for %s index: %s",
 312                                 resp.StatusCode, keepServer.String(), resp.Status))
 313         }
 314
 315         if arvLogger != nil {
 316                 now := time.Now()
 317                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 318                         keepInfo := p["keep_info"].(map[string]interface{})
 319                         serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
 320                         serverInfo["time_index_response_received"] = now
 321                 })
 322         }
 323
 324         response.Address = keepServer
 325         response.Contents.BlockDigestToInfo =
 326                 make(map[blockdigest.BlockDigest]BlockInfo)
 327         scanner := bufio.NewScanner(resp.Body)
 328         numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
 329         for scanner.Scan() {
 330                 numLines++
 331                 blockInfo, err := parseBlockInfoFromIndexLine(scanner.Text())
 332                 if err != nil {
 333                         loggerutil.FatalWithMessage(arvLogger,
 334                                 fmt.Sprintf("Error parsing BlockInfo from index line "+
 335                                         "received from %s: %v",
 336                                         keepServer.String(),
 337                                         err))
 338                 }
 339
 340                 if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
 341                         // This server returned multiple lines containing the same block digest.
 342                         numDuplicates += 1
 343                         if storedBlock.Size != blockInfo.Size {
 344                                 numSizeDisagreements += 1
 345                                 // TODO(misha): Consider failing here.
 346                                 message := fmt.Sprintf("Saw different sizes for the same block "+
 347                                         "on %s: %+v %+v",
 348                                         keepServer.String(),
 349                                         storedBlock,
 350                                         blockInfo)
 351                                 log.Println(message)
 352                                 if arvLogger != nil {
 353                                         arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 354                                                 keepInfo := p["keep_info"].(map[string]interface{})
 355                                                 serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
 356                                                 var error_list []string
 357                                                 read_error_list, has_list := serverInfo["error_list"]
 358                                                 if has_list {
 359                                                         error_list = read_error_list.([]string)
 360                                                 } // If we didn't have the list, error_list is already an empty list
 361                                                 serverInfo["error_list"] = append(error_list, message)
 362                                         })
 363                                 }
 364                         }
 365                         // Keep the block that is bigger, or the block that's newer in
 366                         // the case of a size tie.
 367                         if storedBlock.Size < blockInfo.Size ||
 368                                 (storedBlock.Size == blockInfo.Size &&
 369                                         storedBlock.Mtime < blockInfo.Mtime) {
 370                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 371                         }
 372                 } else {
 373                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 374                 }
 375         }
 376         if err := scanner.Err(); err != nil {
 377                 loggerutil.FatalWithMessage(arvLogger,
 378                         fmt.Sprintf("Received error scanning index response from %s: %v",
 379                                 keepServer.String(),
 380                                 err))
 381         } else {
 382                 log.Printf("%s index contained %d lines with %d duplicates with "+
 383                         "%d size disagreements",
 384                         keepServer.String(),
 385                         numLines,
 386                         numDuplicates,
 387                         numSizeDisagreements)
 388
 389                 if arvLogger != nil {
 390                         now := time.Now()
 391                         arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 392                                 keepInfo := p["keep_info"].(map[string]interface{})
 393                                 serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
 394
 395                                 serverInfo["time_processing_finished"] = now
 396                                 serverInfo["lines_received"] = numLines
 397                                 serverInfo["duplicates_seen"] = numDuplicates
 398                                 serverInfo["size_disagreements_seen"] = numSizeDisagreements
 399                         })
 400                 }
 401         }
 402         resp.Body.Close()
 403         return
 404 }
 405
 406 func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err error) {
 407         tokens := strings.Fields(indexLine)
 408         if len(tokens) != 2 {
 409                 err = fmt.Errorf("Expected 2 tokens per line but received a "+
 410                         "line containing %v instead.",
 411                         tokens)
 412         }
 413
 414         var locator manifest.BlockLocator
 415         if locator, err = manifest.ParseBlockLocator(tokens[0]); err != nil {
 416                 return
 417         }
 418         if len(locator.Hints) > 0 {
 419                 err = fmt.Errorf("Block locator in index line should not contain hints "+
 420                         "but it does: %v",
 421                         locator)
 422                 return
 423         }
 424
 425         blockInfo.Mtime, err = strconv.ParseInt(tokens[1], 10, 64)
 426         if err != nil {
 427                 return
 428         }
 429         blockInfo.Digest = locator.Digest
 430         blockInfo.Size = locator.Size
 431         return
 432 }
 433
 434 func ComputeBlockReplicationCounts(readServers *ReadServers) {
 435         readServers.BlockReplicationCounts = make(map[int]int)
 436         for _, infos := range readServers.BlockToServers {
 437                 replication := len(infos)
 438                 readServers.BlockReplicationCounts[replication] += 1
 439         }
 440 }