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         "errors"
   9         "flag"
  10         "fmt"
  11         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  12         "git.curoverse.com/arvados.git/sdk/go/blockdigest"
  13         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  14         "git.curoverse.com/arvados.git/sdk/go/logger"
  15         "git.curoverse.com/arvados.git/services/datamanager/loggerutil"
  16         "io"
  17         "io/ioutil"
  18         "log"
  19         "net/http"
  20         "strconv"
  21         "strings"
  22         "sync"
  23         "time"
  24 )
  25
  26 type ServerAddress struct {
  27         SSL  bool   `json:service_ssl_flag`
  28         Host string `json:"service_host"`
  29         Port int    `json:"service_port"`
  30         Uuid string `json:"uuid"`
  31 }
  32
  33 // Info about a particular block returned by the server
  34 type BlockInfo struct {
  35         Digest blockdigest.DigestWithSize
  36         Mtime  int64 // TODO(misha): Replace this with a timestamp.
  37 }
  38
  39 // Info about a specified block given by a server
  40 type BlockServerInfo struct {
  41         ServerIndex int
  42         Mtime       int64 // TODO(misha): Replace this with a timestamp.
  43 }
  44
  45 type ServerContents struct {
  46         BlockDigestToInfo map[blockdigest.DigestWithSize]BlockInfo
  47 }
  48
  49 type ServerResponse struct {
  50         Address  ServerAddress
  51         Contents ServerContents
  52 }
  53
  54 type ReadServers struct {
  55         ReadAllServers           bool
  56         KeepServerIndexToAddress []ServerAddress
  57         KeepServerAddressToIndex map[ServerAddress]int
  58         ServerToContents         map[ServerAddress]ServerContents
  59         BlockToServers           map[blockdigest.DigestWithSize][]BlockServerInfo
  60         BlockReplicationCounts   map[int]int
  61 }
  62
  63 type GetKeepServersParams struct {
  64         Client arvadosclient.ArvadosClient
  65         Logger *logger.Logger
  66         Limit  int
  67 }
  68
  69 type KeepServiceList struct {
  70         ItemsAvailable int             `json:"items_available"`
  71         KeepServers    []ServerAddress `json:"items"`
  72 }
  73
  74 var (
  75         // Don't access the token directly, use getDataManagerToken() to
  76         // make sure it's been read.
  77         dataManagerToken             string
  78         dataManagerTokenFile         string
  79         dataManagerTokenFileReadOnce sync.Once
  80 )
  81
  82 func init() {
  83         flag.StringVar(&dataManagerTokenFile,
  84                 "data-manager-token-file",
  85                 "",
  86                 "File with the API token we should use to contact keep servers.")
  87 }
  88
  89 // TODO(misha): Change this to include the UUID as well.
  90 func (s ServerAddress) String() string {
  91         return s.URL()
  92 }
  93
  94 func (s ServerAddress) URL() string {
  95         if s.SSL {
  96                 return fmt.Sprintf("https://%s:%d", s.Host, s.Port)
  97         } else {
  98                 return fmt.Sprintf("http://%s:%d", s.Host, s.Port)
  99         }
 100 }
 101
 102 func GetDataManagerToken(arvLogger *logger.Logger) string {
 103         readDataManagerToken := func() {
 104                 if dataManagerTokenFile == "" {
 105                         flag.Usage()
 106                         loggerutil.FatalWithMessage(arvLogger,
 107                                 "Data Manager Token needed, but data manager token file not specified.")
 108                 } else {
 109                         rawRead, err := ioutil.ReadFile(dataManagerTokenFile)
 110                         if err != nil {
 111                                 loggerutil.FatalWithMessage(arvLogger,
 112                                         fmt.Sprintf("Unexpected error reading token file %s: %v",
 113                                                 dataManagerTokenFile,
 114                                                 err))
 115                         }
 116                         dataManagerToken = strings.TrimSpace(string(rawRead))
 117                 }
 118         }
 119
 120         dataManagerTokenFileReadOnce.Do(readDataManagerToken)
 121         return dataManagerToken
 122 }
 123
 124 func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers) {
 125         results = GetKeepServers(params)
 126         log.Printf("Returned %d keep disks", len(results.ServerToContents))
 127
 128         results.Summarize(params.Logger)
 129         log.Printf("Replication level distribution: %v",
 130                 results.BlockReplicationCounts)
 131
 132         return
 133 }
 134
 135 func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
 136         if &params.Client == nil {
 137                 log.Fatalf("params.Client passed to GetKeepServers() should " +
 138                         "contain a valid ArvadosClient, but instead it is nil.")
 139         }
 140
 141         sdkParams := arvadosclient.Dict{
 142                 "filters": [][]string{[]string{"service_type", "=", "disk"}},
 143         }
 144         if params.Limit > 0 {
 145                 sdkParams["limit"] = params.Limit
 146         }
 147
 148         var sdkResponse KeepServiceList
 149         err := params.Client.List("keep_services", sdkParams, &sdkResponse)
 150
 151         if err != nil {
 152                 loggerutil.FatalWithMessage(params.Logger,
 153                         fmt.Sprintf("Error requesting keep disks from API server: %v", err))
 154         }
 155
 156         if params.Logger != nil {
 157                 params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 158                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 159                         keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable
 160                         keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers)
 161                         keepInfo["keep_servers"] = sdkResponse.KeepServers
 162                 })
 163         }
 164
 165         log.Printf("Received keep services list: %+v", sdkResponse)
 166
 167         if len(sdkResponse.KeepServers) < sdkResponse.ItemsAvailable {
 168                 loggerutil.FatalWithMessage(params.Logger,
 169                         fmt.Sprintf("Did not receive all available keep servers: %+v", sdkResponse))
 170         }
 171
 172         results.KeepServerIndexToAddress = sdkResponse.KeepServers
 173         results.KeepServerAddressToIndex = make(map[ServerAddress]int)
 174         for i, address := range results.KeepServerIndexToAddress {
 175                 results.KeepServerAddressToIndex[address] = i
 176         }
 177
 178         log.Printf("Got Server Addresses: %v", results)
 179
 180         // This is safe for concurrent use
 181         client := http.Client{}
 182
 183         // Send off all the index requests concurrently
 184         responseChan := make(chan ServerResponse)
 185         for _, keepServer := range sdkResponse.KeepServers {
 186                 // The above keepsServer variable is reused for each iteration, so
 187                 // it would be shared across all goroutines. This would result in
 188                 // us querying one server n times instead of n different servers
 189                 // as we intended. To avoid this we add it as an explicit
 190                 // parameter which gets copied. This bug and solution is described
 191                 // in https://golang.org/doc/effective_go.html#channels
 192                 go func(keepServer ServerAddress) {
 193                         responseChan <- GetServerContents(params.Logger,
 194                                 keepServer,
 195                                 client)
 196                 }(keepServer)
 197         }
 198
 199         results.ServerToContents = make(map[ServerAddress]ServerContents)
 200         results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo)
 201
 202         // Read all the responses
 203         for i := range sdkResponse.KeepServers {
 204                 _ = i // Here to prevent go from complaining.
 205                 response := <-responseChan
 206                 log.Printf("Received channel response from %v containing %d files",
 207                         response.Address,
 208                         len(response.Contents.BlockDigestToInfo))
 209                 results.ServerToContents[response.Address] = response.Contents
 210                 serverIndex := results.KeepServerAddressToIndex[response.Address]
 211                 for _, blockInfo := range response.Contents.BlockDigestToInfo {
 212                         results.BlockToServers[blockInfo.Digest] = append(
 213                                 results.BlockToServers[blockInfo.Digest],
 214                                 BlockServerInfo{ServerIndex: serverIndex,
 215                                         Mtime: blockInfo.Mtime})
 216                 }
 217         }
 218         return
 219 }
 220
 221 func GetServerContents(arvLogger *logger.Logger,
 222         keepServer ServerAddress,
 223         client http.Client) (response ServerResponse) {
 224
 225         GetServerStatus(arvLogger, keepServer, client)
 226
 227         req := CreateIndexRequest(arvLogger, keepServer)
 228         resp, err := client.Do(req)
 229         if err != nil {
 230                 loggerutil.FatalWithMessage(arvLogger,
 231                         fmt.Sprintf("Error fetching %s: %v. Response was %+v",
 232                                 req.URL.String(),
 233                                 err,
 234                                 resp))
 235         }
 236
 237         return ReadServerResponse(arvLogger, keepServer, resp)
 238 }
 239
 240 func GetServerStatus(arvLogger *logger.Logger,
 241         keepServer ServerAddress,
 242         client http.Client) {
 243         url := fmt.Sprintf("http://%s:%d/status.json",
 244                 keepServer.Host,
 245                 keepServer.Port)
 246
 247         if arvLogger != nil {
 248                 now := time.Now()
 249                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 250                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 251                         serverInfo := make(map[string]interface{})
 252                         serverInfo["status_request_sent_at"] = now
 253                         serverInfo["host"] = keepServer.Host
 254                         serverInfo["port"] = keepServer.Port
 255
 256                         keepInfo[keepServer.Uuid] = serverInfo
 257                 })
 258         }
 259
 260         resp, err := client.Get(url)
 261         if err != nil {
 262                 loggerutil.FatalWithMessage(arvLogger,
 263                         fmt.Sprintf("Error getting keep status from %s: %v", url, err))
 264         } else if resp.StatusCode != 200 {
 265                 loggerutil.FatalWithMessage(arvLogger,
 266                         fmt.Sprintf("Received error code %d in response to request "+
 267                                 "for %s status: %s",
 268                                 resp.StatusCode, url, resp.Status))
 269         }
 270
 271         var keepStatus map[string]interface{}
 272         decoder := json.NewDecoder(resp.Body)
 273         decoder.UseNumber()
 274         err = decoder.Decode(&keepStatus)
 275         if err != nil {
 276                 loggerutil.FatalWithMessage(arvLogger,
 277                         fmt.Sprintf("Error decoding keep status from %s: %v", url, err))
 278         }
 279
 280         if arvLogger != nil {
 281                 now := time.Now()
 282                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 283                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 284                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 285                         serverInfo["status_response_processed_at"] = now
 286                         serverInfo["status"] = keepStatus
 287                 })
 288         }
 289 }
 290
 291 func CreateIndexRequest(arvLogger *logger.Logger,
 292         keepServer ServerAddress) (req *http.Request) {
 293         url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port)
 294         log.Println("About to fetch keep server contents from " + url)
 295
 296         if arvLogger != nil {
 297                 now := time.Now()
 298                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 299                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 300                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 301                         serverInfo["index_request_sent_at"] = now
 302                 })
 303         }
 304
 305         req, err := http.NewRequest("GET", url, nil)
 306         if err != nil {
 307                 loggerutil.FatalWithMessage(arvLogger,
 308                         fmt.Sprintf("Error building http request for %s: %v", url, err))
 309         }
 310
 311         req.Header.Add("Authorization",
 312                 fmt.Sprintf("OAuth2 %s", GetDataManagerToken(arvLogger)))
 313         return
 314 }
 315
 316 func ReadServerResponse(arvLogger *logger.Logger,
 317         keepServer ServerAddress,
 318         resp *http.Response) (response ServerResponse) {
 319
 320         if resp.StatusCode != 200 {
 321                 loggerutil.FatalWithMessage(arvLogger,
 322                         fmt.Sprintf("Received error code %d in response to request "+
 323                                 "for %s index: %s",
 324                                 resp.StatusCode, keepServer.String(), resp.Status))
 325         }
 326
 327         if arvLogger != nil {
 328                 now := time.Now()
 329                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 330                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 331                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 332                         serverInfo["index_response_received_at"] = now
 333                 })
 334         }
 335
 336         response.Address = keepServer
 337         response.Contents.BlockDigestToInfo =
 338                 make(map[blockdigest.DigestWithSize]BlockInfo)
 339         reader := bufio.NewReader(resp.Body)
 340         numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
 341         for {
 342                 numLines++
 343                 line, err := reader.ReadString('\n')
 344                 if err == io.EOF {
 345                         loggerutil.FatalWithMessage(arvLogger,
 346                                 fmt.Sprintf("Index from %s truncated at line %d",
 347                                         keepServer.String(), numLines))
 348                 } else if err != nil {
 349                         loggerutil.FatalWithMessage(arvLogger,
 350                                 fmt.Sprintf("Error reading index response from %s at line %d: %v",
 351                                         keepServer.String(), numLines, err))
 352                 }
 353                 if line == "\n" {
 354                         if _, err := reader.Peek(1); err == nil {
 355                                 extra, _ := reader.ReadString('\n')
 356                                 loggerutil.FatalWithMessage(arvLogger,
 357                                         fmt.Sprintf("Index from %s had trailing data at line %d after EOF marker: %s",
 358                                                 keepServer.String(), numLines+1, extra))
 359                         } else if err != io.EOF {
 360                                 loggerutil.FatalWithMessage(arvLogger,
 361                                         fmt.Sprintf("Index from %s had read error after EOF marker at line %d: %v",
 362                                                 keepServer.String(), numLines, err))
 363                         }
 364                         numLines--
 365                         break
 366                 }
 367                 blockInfo, err := parseBlockInfoFromIndexLine(line)
 368                 if err != nil {
 369                         loggerutil.FatalWithMessage(arvLogger,
 370                                 fmt.Sprintf("Error parsing BlockInfo from index line "+
 371                                         "received from %s: %v",
 372                                         keepServer.String(),
 373                                         err))
 374                 }
 375
 376                 if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
 377                         // This server returned multiple lines containing the same block digest.
 378                         numDuplicates += 1
 379                         // Keep the block that's newer.
 380                         if storedBlock.Mtime < blockInfo.Mtime {
 381                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 382                         }
 383                 } else {
 384                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 385                 }
 386         }
 387
 388         log.Printf("%s index contained %d lines with %d duplicates with "+
 389                 "%d size disagreements",
 390                 keepServer.String(),
 391                 numLines,
 392                 numDuplicates,
 393                 numSizeDisagreements)
 394
 395         if arvLogger != nil {
 396                 now := time.Now()
 397                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 398                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 399                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 400
 401                         serverInfo["processing_finished_at"] = now
 402                         serverInfo["lines_received"] = numLines
 403                         serverInfo["duplicates_seen"] = numDuplicates
 404                         serverInfo["size_disagreements_seen"] = numSizeDisagreements
 405                 })
 406         }
 407         resp.Body.Close()
 408         return
 409 }
 410
 411 func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err error) {
 412         tokens := strings.Fields(indexLine)
 413         if len(tokens) != 2 {
 414                 err = fmt.Errorf("Expected 2 tokens per line but received a "+
 415                         "line containing %v instead.",
 416                         tokens)
 417         }
 418
 419         var locator blockdigest.BlockLocator
 420         if locator, err = blockdigest.ParseBlockLocator(tokens[0]); err != nil {
 421                 err = fmt.Errorf("%v Received error while parsing line \"%s\"",
 422                         err, indexLine)
 423                 return
 424         }
 425         if len(locator.Hints) > 0 {
 426                 err = fmt.Errorf("Block locator in index line should not contain hints "+
 427                         "but it does: %v",
 428                         locator)
 429                 return
 430         }
 431
 432         blockInfo.Mtime, err = strconv.ParseInt(tokens[1], 10, 64)
 433         if err != nil {
 434                 return
 435         }
 436         blockInfo.Digest =
 437                 blockdigest.DigestWithSize{Digest: locator.Digest,
 438                         Size: uint32(locator.Size)}
 439         return
 440 }
 441
 442 func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) {
 443         readServers.BlockReplicationCounts = make(map[int]int)
 444         for _, infos := range readServers.BlockToServers {
 445                 replication := len(infos)
 446                 readServers.BlockReplicationCounts[replication] += 1
 447         }
 448
 449         if arvLogger != nil {
 450                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 451                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 452                         keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers)
 453                 })
 454         }
 455
 456 }
 457
 458 type TrashRequest struct {
 459         Locator    string `json:"locator"`
 460         BlockMtime int64  `json:"block_mtime"`
 461 }
 462
 463 type TrashList []TrashRequest
 464
 465 func SendTrashLists(dataManagerToken string, kc *keepclient.KeepClient, spl map[string]TrashList) (errs []error) {
 466         count := 0
 467         barrier := make(chan error)
 468
 469         client := kc.Client
 470
 471         for url, v := range spl {
 472                 count += 1
 473                 log.Printf("Sending trash list to %v", url)
 474
 475                 go (func(url string, v TrashList) {
 476                         pipeReader, pipeWriter := io.Pipe()
 477                         go (func() {
 478                                 enc := json.NewEncoder(pipeWriter)
 479                                 enc.Encode(v)
 480                                 pipeWriter.Close()
 481                         })()
 482
 483                         req, err := http.NewRequest("PUT", fmt.Sprintf("%s/trash", url), pipeReader)
 484                         if err != nil {
 485                                 log.Printf("Error creating trash list request for %v error: %v", url, err.Error())
 486                                 barrier <- err
 487                                 return
 488                         }
 489
 490                         // Add api token header
 491                         req.Header.Add("Authorization", fmt.Sprintf("OAuth2 %s", dataManagerToken))
 492
 493                         // Make the request
 494                         var resp *http.Response
 495                         if resp, err = client.Do(req); err != nil {
 496                                 log.Printf("Error sending trash list to %v error: %v", url, err.Error())
 497                                 barrier <- err
 498                                 return
 499                         }
 500
 501                         log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status)
 502
 503                         io.Copy(ioutil.Discard, resp.Body)
 504                         resp.Body.Close()
 505
 506                         if resp.StatusCode != 200 {
 507                                 barrier <- errors.New(fmt.Sprintf("Got HTTP code %v", resp.StatusCode))
 508                         } else {
 509                                 barrier <- nil
 510                         }
 511                 })(url, v)
 512
 513         }
 514
 515         for i := 0; i < count; i += 1 {
 516                 b := <-barrier
 517                 if b != nil {
 518                         errs = append(errs, b)
 519                 }
 520         }
 521
 522         return errs
 523 }