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         "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/keepclient"
  13         "git.curoverse.com/arvados.git/sdk/go/logger"
  14         "io"
  15         "io/ioutil"
  16         "log"
  17         "net/http"
  18         "os"
  19         "strconv"
  20         "strings"
  21         "time"
  22 )
  23
  24 // ServerAddress struct
  25 type ServerAddress struct {
  26         SSL         bool   `json:"service_ssl_flag"`
  27         Host        string `json:"service_host"`
  28         Port        int    `json:"service_port"`
  29         UUID        string `json:"uuid"`
  30         ServiceType string `json:"service_type"`
  31 }
  32
  33 // BlockInfo is 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 // BlockServerInfo is 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 // ServerContents struct
  46 type ServerContents struct {
  47         BlockDigestToInfo map[blockdigest.DigestWithSize]BlockInfo
  48 }
  49
  50 // ServerResponse struct
  51 type ServerResponse struct {
  52         Address  ServerAddress
  53         Contents ServerContents
  54         Err      error
  55 }
  56
  57 // ReadServers struct
  58 type ReadServers struct {
  59         ReadAllServers           bool
  60         KeepServerIndexToAddress []ServerAddress
  61         KeepServerAddressToIndex map[ServerAddress]int
  62         ServerToContents         map[ServerAddress]ServerContents
  63         BlockToServers           map[blockdigest.DigestWithSize][]BlockServerInfo
  64         BlockReplicationCounts   map[int]int
  65 }
  66
  67 // GetKeepServersParams struct
  68 type GetKeepServersParams struct {
  69         Client arvadosclient.ArvadosClient
  70         Logger *logger.Logger
  71         Limit  int
  72 }
  73
  74 // ServiceList consists of the addresses of all the available kee servers
  75 type ServiceList struct {
  76         ItemsAvailable int             `json:"items_available"`
  77         KeepServers    []ServerAddress `json:"items"`
  78 }
  79
  80 // String
  81 // TODO(misha): Change this to include the UUID as well.
  82 func (s ServerAddress) String() string {
  83         return s.URL()
  84 }
  85
  86 // URL of the keep server
  87 func (s ServerAddress) URL() string {
  88         if s.SSL {
  89                 return fmt.Sprintf("https://%s:%d", s.Host, s.Port)
  90         }
  91         return fmt.Sprintf("http://%s:%d", s.Host, s.Port)
  92 }
  93
  94 // GetKeepServersAndSummarize gets keep servers from api
  95 func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers, err error) {
  96         results, err = GetKeepServers(params)
  97         if err != nil {
  98                 return
  99         }
 100         log.Printf("Returned %d keep disks", len(results.ServerToContents))
 101
 102         results.Summarize(params.Logger)
 103         log.Printf("Replication level distribution: %v",
 104                 results.BlockReplicationCounts)
 105
 106         return
 107 }
 108
 109 // GetKeepServers from api server
 110 func GetKeepServers(params GetKeepServersParams) (results ReadServers, err error) {
 111         sdkParams := arvadosclient.Dict{
 112                 "filters": [][]string{[]string{"service_type", "!=", "proxy"}},
 113         }
 114         if params.Limit > 0 {
 115                 sdkParams["limit"] = params.Limit
 116         }
 117
 118         var sdkResponse ServiceList
 119         err = params.Client.List("keep_services", sdkParams, &sdkResponse)
 120
 121         if err != nil {
 122                 return
 123         }
 124
 125         // Currently, only "disk" types are supported. Stop if any other service types are found.
 126         for _, server := range sdkResponse.KeepServers {
 127                 if server.ServiceType != "disk" {
 128                         return results, fmt.Errorf("Unsupported service type %q found for: %v", server.ServiceType, server)
 129                 }
 130         }
 131
 132         if params.Logger != nil {
 133                 params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 134                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 135                         keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable
 136                         keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers)
 137                         keepInfo["keep_servers"] = sdkResponse.KeepServers
 138                 })
 139         }
 140
 141         log.Printf("Received keep services list: %+v", sdkResponse)
 142
 143         if len(sdkResponse.KeepServers) < sdkResponse.ItemsAvailable {
 144                 return results, fmt.Errorf("Did not receive all available keep servers: %+v", sdkResponse)
 145         }
 146
 147         results.KeepServerIndexToAddress = sdkResponse.KeepServers
 148         results.KeepServerAddressToIndex = make(map[ServerAddress]int)
 149         for i, address := range results.KeepServerIndexToAddress {
 150                 results.KeepServerAddressToIndex[address] = i
 151         }
 152
 153         log.Printf("Got Server Addresses: %v", results)
 154
 155         // Send off all the index requests concurrently
 156         responseChan := make(chan ServerResponse)
 157         for _, keepServer := range sdkResponse.KeepServers {
 158                 // The above keepsServer variable is reused for each iteration, so
 159                 // it would be shared across all goroutines. This would result in
 160                 // us querying one server n times instead of n different servers
 161                 // as we intended. To avoid this we add it as an explicit
 162                 // parameter which gets copied. This bug and solution is described
 163                 // in https://golang.org/doc/effective_go.html#channels
 164                 go func(keepServer ServerAddress) {
 165                         responseChan <- GetServerContents(params.Logger,
 166                                 keepServer,
 167                                 params.Client)
 168                 }(keepServer)
 169         }
 170
 171         results.ServerToContents = make(map[ServerAddress]ServerContents)
 172         results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo)
 173
 174         // Read all the responses
 175         for i := range sdkResponse.KeepServers {
 176                 _ = i // Here to prevent go from complaining.
 177                 response := <-responseChan
 178
 179                 // Check if there were any errors during GetServerContents
 180                 if response.Err != nil {
 181                         return results, response.Err
 182                 }
 183
 184                 log.Printf("Received channel response from %v containing %d files",
 185                         response.Address,
 186                         len(response.Contents.BlockDigestToInfo))
 187                 results.ServerToContents[response.Address] = response.Contents
 188                 serverIndex := results.KeepServerAddressToIndex[response.Address]
 189                 for _, blockInfo := range response.Contents.BlockDigestToInfo {
 190                         results.BlockToServers[blockInfo.Digest] = append(
 191                                 results.BlockToServers[blockInfo.Digest],
 192                                 BlockServerInfo{ServerIndex: serverIndex,
 193                                         Mtime: blockInfo.Mtime})
 194                 }
 195         }
 196         return
 197 }
 198
 199 // GetServerContents of the keep server
 200 func GetServerContents(arvLogger *logger.Logger,
 201         keepServer ServerAddress,
 202         arv arvadosclient.ArvadosClient) (response ServerResponse) {
 203
 204         err := GetServerStatus(arvLogger, keepServer, arv)
 205         if err != nil {
 206                 response.Err = err
 207                 return
 208         }
 209
 210         req, err := CreateIndexRequest(arvLogger, keepServer, arv)
 211         if err != nil {
 212                 response.Err = err
 213                 return
 214         }
 215
 216         resp, err := arv.Client.Do(req)
 217         if err != nil {
 218                 response.Err = err
 219                 return
 220         }
 221
 222         response, err = ReadServerResponse(arvLogger, keepServer, resp)
 223         if err != nil {
 224                 response.Err = err
 225                 return
 226         }
 227
 228         return
 229 }
 230
 231 // GetServerStatus get keep server status by invoking /status.json
 232 func GetServerStatus(arvLogger *logger.Logger,
 233         keepServer ServerAddress,
 234         arv arvadosclient.ArvadosClient) error {
 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                         runInfo := logger.GetOrCreateMap(p, "run_info")
 251                         runInfo["args"] = os.Args
 252                 })
 253         }
 254
 255         resp, err := arv.Client.Get(url)
 256         if err != nil {
 257                 return fmt.Errorf("Error getting keep status from %s: %v", url, err)
 258         } else if resp.StatusCode != 200 {
 259                 return fmt.Errorf("Received error code %d in response to request "+
 260                         "for %s status: %s",
 261                         resp.StatusCode, url, resp.Status)
 262         }
 263
 264         var keepStatus map[string]interface{}
 265         decoder := json.NewDecoder(resp.Body)
 266         decoder.UseNumber()
 267         err = decoder.Decode(&keepStatus)
 268         if err != nil {
 269                 return fmt.Errorf("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                         runInfo := logger.GetOrCreateMap(p, "run_info")
 281                         runInfo["args"] = os.Args
 282                 })
 283         }
 284
 285         return nil
 286 }
 287
 288 // CreateIndexRequest to the keep server
 289 func CreateIndexRequest(arvLogger *logger.Logger,
 290         keepServer ServerAddress,
 291         arv arvadosclient.ArvadosClient) (req *http.Request, err error) {
 292         url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port)
 293         log.Println("About to fetch keep server contents from " + url)
 294
 295         if arvLogger != nil {
 296                 now := time.Now()
 297                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 298                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 299                         serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
 300                         serverInfo["index_request_sent_at"] = now
 301
 302                         runInfo := logger.GetOrCreateMap(p, "run_info")
 303                         runInfo["args"] = os.Args
 304                 })
 305         }
 306
 307         req, err = http.NewRequest("GET", url, nil)
 308         if err != nil {
 309                 return req, fmt.Errorf("Error building http request for %s: %v", url, err)
 310         }
 311
 312         req.Header.Add("Authorization", "OAuth2 "+arv.ApiToken)
 313         return req, err
 314 }
 315
 316 // ReadServerResponse reads reasponse from keep server
 317 func ReadServerResponse(arvLogger *logger.Logger,
 318         keepServer ServerAddress,
 319         resp *http.Response) (response ServerResponse, err error) {
 320
 321         if resp.StatusCode != 200 {
 322                 return response, fmt.Errorf("Received error code %d in response to index request for %s: %s",
 323                         resp.StatusCode, keepServer.String(), resp.Status)
 324         }
 325
 326         if arvLogger != nil {
 327                 now := time.Now()
 328                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 329                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 330                         serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
 331                         serverInfo["index_response_received_at"] = now
 332
 333                         runInfo := logger.GetOrCreateMap(p, "run_info")
 334                         runInfo["args"] = os.Args
 335                 })
 336         }
 337
 338         response.Address = keepServer
 339         response.Contents.BlockDigestToInfo =
 340                 make(map[blockdigest.DigestWithSize]BlockInfo)
 341         reader := bufio.NewReader(resp.Body)
 342         numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
 343         for {
 344                 numLines++
 345                 line, err := reader.ReadString('\n')
 346                 if err == io.EOF {
 347                         return response, fmt.Errorf("Index from %s truncated at line %d",
 348                                 keepServer.String(), numLines)
 349                 } else if err != nil {
 350                         return response, fmt.Errorf("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                                 return response, fmt.Errorf("Index from %s had trailing data at line %d after EOF marker: %s",
 357                                         keepServer.String(), numLines+1, extra)
 358                         } else if err != io.EOF {
 359                                 return response, fmt.Errorf("Index from %s had read error after EOF marker at line %d: %v",
 360                                         keepServer.String(), numLines, err)
 361                         }
 362                         numLines--
 363                         break
 364                 }
 365                 blockInfo, err := parseBlockInfoFromIndexLine(line)
 366                 if err != nil {
 367                         return response, fmt.Errorf("Error parsing BlockInfo from index line "+
 368                                 "received from %s: %v",
 369                                 keepServer.String(),
 370                                 err)
 371                 }
 372
 373                 if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
 374                         // This server returned multiple lines containing the same block digest.
 375                         numDuplicates++
 376                         // Keep the block that's newer.
 377                         if storedBlock.Mtime < blockInfo.Mtime {
 378                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 379                         }
 380                 } else {
 381                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 382                 }
 383         }
 384
 385         log.Printf("%s index contained %d lines with %d duplicates with "+
 386                 "%d size disagreements",
 387                 keepServer.String(),
 388                 numLines,
 389                 numDuplicates,
 390                 numSizeDisagreements)
 391
 392         if arvLogger != nil {
 393                 now := time.Now()
 394                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 395                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 396                         serverInfo := keepInfo[keepServer.UUID].(map[string]interface{})
 397
 398                         serverInfo["processing_finished_at"] = now
 399                         serverInfo["lines_received"] = numLines
 400                         serverInfo["duplicates_seen"] = numDuplicates
 401                         serverInfo["size_disagreements_seen"] = numSizeDisagreements
 402
 403                         runInfo := logger.GetOrCreateMap(p, "run_info")
 404                         runInfo["args"] = os.Args
 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 %#q 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 \"%#q\"",
 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: %#q",
 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 // Summarize results from keep server
 443 func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) {
 444         readServers.BlockReplicationCounts = make(map[int]int)
 445         for _, infos := range readServers.BlockToServers {
 446                 replication := len(infos)
 447                 readServers.BlockReplicationCounts[replication]++
 448         }
 449
 450         if arvLogger != nil {
 451                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 452                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 453                         keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers)
 454
 455                         runInfo := logger.GetOrCreateMap(p, "run_info")
 456                         runInfo["args"] = os.Args
 457                 })
 458         }
 459 }
 460
 461 // TrashRequest struct
 462 type TrashRequest struct {
 463         Locator    string `json:"locator"`
 464         BlockMtime int64  `json:"block_mtime"`
 465 }
 466
 467 // TrashList is an array of TrashRequest objects
 468 type TrashList []TrashRequest
 469
 470 // SendTrashLists to trash queue
 471 func SendTrashLists(kc *keepclient.KeepClient, spl map[string]TrashList) (errs []error) {
 472         count := 0
 473         barrier := make(chan error)
 474
 475         client := kc.Client
 476
 477         for url, v := range spl {
 478                 count++
 479                 log.Printf("Sending trash list to %v", url)
 480
 481                 go (func(url string, v TrashList) {
 482                         pipeReader, pipeWriter := io.Pipe()
 483                         go (func() {
 484                                 enc := json.NewEncoder(pipeWriter)
 485                                 enc.Encode(v)
 486                                 pipeWriter.Close()
 487                         })()
 488
 489                         req, err := http.NewRequest("PUT", fmt.Sprintf("%s/trash", url), pipeReader)
 490                         if err != nil {
 491                                 log.Printf("Error creating trash list request for %v error: %v", url, err.Error())
 492                                 barrier <- err
 493                                 return
 494                         }
 495
 496                         req.Header.Add("Authorization", "OAuth2 "+kc.Arvados.ApiToken)
 497
 498                         // Make the request
 499                         var resp *http.Response
 500                         if resp, err = client.Do(req); err != nil {
 501                                 log.Printf("Error sending trash list to %v error: %v", url, err.Error())
 502                                 barrier <- err
 503                                 return
 504                         }
 505
 506                         log.Printf("Sent trash list to %v: response was HTTP %v", url, resp.Status)
 507
 508                         io.Copy(ioutil.Discard, resp.Body)
 509                         resp.Body.Close()
 510
 511                         if resp.StatusCode != 200 {
 512                                 barrier <- errors.New(fmt.Sprintf("Got HTTP code %v", resp.StatusCode))
 513                         } else {
 514                                 barrier <- nil
 515                         }
 516                 })(url, v)
 517
 518         }
 519
 520         for i := 0; i < count; i++ {
 521                 b := <-barrier
 522                 if b != nil {
 523                         errs = append(errs, b)
 524                 }
 525         }
 526
 527         return errs
 528 }