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