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