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/services/datamanager/loggerutil"
  14         "io"
  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         Uuid string `json:"uuid"`
  28 }
  29
  30 // Info about a particular block returned by the server
  31 type BlockInfo struct {
  32         Digest blockdigest.DigestWithSize
  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         Mtime       int64 // TODO(misha): Replace this with a timestamp.
  40 }
  41
  42 type ServerContents struct {
  43         BlockDigestToInfo map[blockdigest.DigestWithSize]BlockInfo
  44 }
  45
  46 type ServerResponse struct {
  47         Address  ServerAddress
  48         Contents ServerContents
  49 }
  50
  51 type ReadServers struct {
  52         ReadAllServers           bool
  53         KeepServerIndexToAddress []ServerAddress
  54         KeepServerAddressToIndex map[ServerAddress]int
  55         ServerToContents         map[ServerAddress]ServerContents
  56         BlockToServers           map[blockdigest.DigestWithSize][]BlockServerInfo
  57         BlockReplicationCounts   map[int]int
  58 }
  59
  60 type GetKeepServersParams struct {
  61         Client arvadosclient.ArvadosClient
  62         Logger *logger.Logger
  63         Limit  int
  64 }
  65
  66 type KeepServiceList struct {
  67         ItemsAvailable int             `json:"items_available"`
  68         KeepServers    []ServerAddress `json:"items"`
  69 }
  70
  71 var (
  72         // Don't access the token directly, use getDataManagerToken() to
  73         // make sure it's been read.
  74         dataManagerToken             string
  75         dataManagerTokenFile         string
  76         dataManagerTokenFileReadOnce sync.Once
  77 )
  78
  79 func init() {
  80         flag.StringVar(&dataManagerTokenFile,
  81                 "data-manager-token-file",
  82                 "",
  83                 "File with the API token we should use to contact keep servers.")
  84 }
  85
  86 // TODO(misha): Change this to include the UUID as well.
  87 func (s ServerAddress) String() string {
  88         return s.HostPort()
  89 }
  90
  91 func (s ServerAddress) HostPort() string {
  92         return fmt.Sprintf("%s:%d", s.Host, s.Port)
  93 }
  94
  95 func getDataManagerToken(arvLogger *logger.Logger) string {
  96         readDataManagerToken := func() {
  97                 if dataManagerTokenFile == "" {
  98                         flag.Usage()
  99                         loggerutil.FatalWithMessage(arvLogger,
 100                                 "Data Manager Token needed, but data manager token file not specified.")
 101                 } else {
 102                         rawRead, err := ioutil.ReadFile(dataManagerTokenFile)
 103                         if err != nil {
 104                                 loggerutil.FatalWithMessage(arvLogger,
 105                                         fmt.Sprintf("Unexpected error reading token file %s: %v",
 106                                                 dataManagerTokenFile,
 107                                                 err))
 108                         }
 109                         dataManagerToken = strings.TrimSpace(string(rawRead))
 110                 }
 111         }
 112
 113         dataManagerTokenFileReadOnce.Do(readDataManagerToken)
 114         return dataManagerToken
 115 }
 116
 117 func GetKeepServersAndSummarize(params GetKeepServersParams) (results ReadServers) {
 118         results = GetKeepServers(params)
 119         log.Printf("Returned %d keep disks", len(results.ServerToContents))
 120
 121         results.Summarize(params.Logger)
 122         log.Printf("Replication level distribution: %v",
 123                 results.BlockReplicationCounts)
 124
 125         return
 126 }
 127
 128 func GetKeepServers(params GetKeepServersParams) (results ReadServers) {
 129         if &params.Client == nil {
 130                 log.Fatalf("params.Client passed to GetKeepServers() should " +
 131                         "contain a valid ArvadosClient, but instead it is nil.")
 132         }
 133
 134         sdkParams := arvadosclient.Dict{
 135                 "filters": [][]string{[]string{"service_type", "=", "disk"}},
 136         }
 137         if params.Limit > 0 {
 138                 sdkParams["limit"] = params.Limit
 139         }
 140
 141         var sdkResponse KeepServiceList
 142         err := params.Client.List("keep_services", sdkParams, &sdkResponse)
 143
 144         if err != nil {
 145                 loggerutil.FatalWithMessage(params.Logger,
 146                         fmt.Sprintf("Error requesting keep disks from API server: %v", err))
 147         }
 148
 149         if params.Logger != nil {
 150                 params.Logger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 151                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 152                         keepInfo["num_keep_servers_available"] = sdkResponse.ItemsAvailable
 153                         keepInfo["num_keep_servers_received"] = len(sdkResponse.KeepServers)
 154                         keepInfo["keep_servers"] = sdkResponse.KeepServers
 155                 })
 156         }
 157
 158         log.Printf("Received keep services list: %+v", sdkResponse)
 159
 160         if len(sdkResponse.KeepServers) < sdkResponse.ItemsAvailable {
 161                 loggerutil.FatalWithMessage(params.Logger,
 162                         fmt.Sprintf("Did not receive all available keep servers: %+v", sdkResponse))
 163         }
 164
 165         results.KeepServerIndexToAddress = sdkResponse.KeepServers
 166         results.KeepServerAddressToIndex = make(map[ServerAddress]int)
 167         for i, address := range results.KeepServerIndexToAddress {
 168                 results.KeepServerAddressToIndex[address] = i
 169         }
 170
 171         log.Printf("Got Server Addresses: %v", results)
 172
 173         // This is safe for concurrent use
 174         client := http.Client{}
 175
 176         // Send off all the index requests concurrently
 177         responseChan := make(chan ServerResponse)
 178         for _, keepServer := range sdkResponse.KeepServers {
 179                 // The above keepsServer variable is reused for each iteration, so
 180                 // it would be shared across all goroutines. This would result in
 181                 // us querying one server n times instead of n different servers
 182                 // as we intended. To avoid this we add it as an explicit
 183                 // parameter which gets copied. This bug and solution is described
 184                 // in https://golang.org/doc/effective_go.html#channels
 185                 go func(keepServer ServerAddress) {
 186                         responseChan <- GetServerContents(params.Logger,
 187                                 keepServer,
 188                                 client)
 189                 }(keepServer)
 190         }
 191
 192         results.ServerToContents = make(map[ServerAddress]ServerContents)
 193         results.BlockToServers = make(map[blockdigest.DigestWithSize][]BlockServerInfo)
 194
 195         // Read all the responses
 196         for i := range sdkResponse.KeepServers {
 197                 _ = i // Here to prevent go from complaining.
 198                 response := <-responseChan
 199                 log.Printf("Received channel response from %v containing %d files",
 200                         response.Address,
 201                         len(response.Contents.BlockDigestToInfo))
 202                 results.ServerToContents[response.Address] = response.Contents
 203                 serverIndex := results.KeepServerAddressToIndex[response.Address]
 204                 for _, blockInfo := range response.Contents.BlockDigestToInfo {
 205                         results.BlockToServers[blockInfo.Digest] = append(
 206                                 results.BlockToServers[blockInfo.Digest],
 207                                 BlockServerInfo{ServerIndex: serverIndex,
 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. Response was %+v",
 225                                 req.URL.String(),
 226                                 err,
 227                                 resp))
 228         }
 229
 230         return ReadServerResponse(arvLogger, keepServer, resp)
 231 }
 232
 233 func GetServerStatus(arvLogger *logger.Logger,
 234         keepServer ServerAddress,
 235         client http.Client) {
 236         url := fmt.Sprintf("http://%s:%d/status.json",
 237                 keepServer.Host,
 238                 keepServer.Port)
 239
 240         if arvLogger != nil {
 241                 now := time.Now()
 242                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 243                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 244                         serverInfo := make(map[string]interface{})
 245                         serverInfo["status_request_sent_at"] = now
 246                         serverInfo["host"] = keepServer.Host
 247                         serverInfo["port"] = keepServer.Port
 248
 249                         keepInfo[keepServer.Uuid] = serverInfo
 250                 })
 251         }
 252
 253         resp, err := client.Get(url)
 254         if err != nil {
 255                 loggerutil.FatalWithMessage(arvLogger,
 256                         fmt.Sprintf("Error getting keep status from %s: %v", url, err))
 257         } else if resp.StatusCode != 200 {
 258                 loggerutil.FatalWithMessage(arvLogger,
 259                         fmt.Sprintf("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                 loggerutil.FatalWithMessage(arvLogger,
 270                         fmt.Sprintf("Error decoding keep status from %s: %v", url, err))
 271         }
 272
 273         if arvLogger != nil {
 274                 now := time.Now()
 275                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 276                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 277                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 278                         serverInfo["status_response_processed_at"] = now
 279                         serverInfo["status"] = keepStatus
 280                 })
 281         }
 282 }
 283
 284 func CreateIndexRequest(arvLogger *logger.Logger,
 285         keepServer ServerAddress) (req *http.Request) {
 286         url := fmt.Sprintf("http://%s:%d/index", keepServer.Host, keepServer.Port)
 287         log.Println("About to fetch keep server contents from " + url)
 288
 289         if arvLogger != nil {
 290                 now := time.Now()
 291                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 292                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 293                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 294                         serverInfo["index_request_sent_at"] = now
 295                 })
 296         }
 297
 298         req, err := http.NewRequest("GET", url, nil)
 299         if err != nil {
 300                 loggerutil.FatalWithMessage(arvLogger,
 301                         fmt.Sprintf("Error building http request for %s: %v", url, err))
 302         }
 303
 304         req.Header.Add("Authorization",
 305                 fmt.Sprintf("OAuth2 %s", getDataManagerToken(arvLogger)))
 306         return
 307 }
 308
 309 func ReadServerResponse(arvLogger *logger.Logger,
 310         keepServer ServerAddress,
 311         resp *http.Response) (response ServerResponse) {
 312
 313         if resp.StatusCode != 200 {
 314                 loggerutil.FatalWithMessage(arvLogger,
 315                         fmt.Sprintf("Received error code %d in response to request "+
 316                                 "for %s index: %s",
 317                                 resp.StatusCode, keepServer.String(), resp.Status))
 318         }
 319
 320         if arvLogger != nil {
 321                 now := time.Now()
 322                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 323                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 324                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 325                         serverInfo["index_response_received_at"] = now
 326                 })
 327         }
 328
 329         response.Address = keepServer
 330         response.Contents.BlockDigestToInfo =
 331                 make(map[blockdigest.DigestWithSize]BlockInfo)
 332         reader := bufio.NewReader(resp.Body)
 333         numLines, numDuplicates, numSizeDisagreements := 0, 0, 0
 334         for {
 335                 numLines++
 336                 line, err := reader.ReadString('\n')
 337                 if err == io.EOF {
 338                         loggerutil.FatalWithMessage(arvLogger,
 339                                 fmt.Sprintf("Index from %s truncated at line %d",
 340                                         keepServer.String(), numLines))
 341                 } else if err != nil {
 342                         loggerutil.FatalWithMessage(arvLogger,
 343                                 fmt.Sprintf("Error reading index response from %s at line %d: %v",
 344                                         keepServer.String(), numLines, err))
 345                 }
 346                 if line == "\n" {
 347                         if _, err := reader.Peek(1); err == nil {
 348                                 extra, _ := reader.ReadString('\n')
 349                                 loggerutil.FatalWithMessage(arvLogger,
 350                                         fmt.Sprintf("Index from %s had trailing data at line %d after EOF marker: %s",
 351                                                 keepServer.String(), numLines+1, extra))
 352                         } else if err != io.EOF {
 353                                 loggerutil.FatalWithMessage(arvLogger,
 354                                         fmt.Sprintf("Index from %s had read error after EOF marker at line %d: %v",
 355                                                 keepServer.String(), numLines, err))
 356                         }
 357                         numLines--
 358                         break
 359                 }
 360                 blockInfo, err := parseBlockInfoFromIndexLine(line)
 361                 if err != nil {
 362                         loggerutil.FatalWithMessage(arvLogger,
 363                                 fmt.Sprintf("Error parsing BlockInfo from index line "+
 364                                         "received from %s: %v",
 365                                         keepServer.String(),
 366                                         err))
 367                 }
 368
 369                 if storedBlock, ok := response.Contents.BlockDigestToInfo[blockInfo.Digest]; ok {
 370                         // This server returned multiple lines containing the same block digest.
 371                         numDuplicates += 1
 372                         // Keep the block that's newer.
 373                         if storedBlock.Mtime < blockInfo.Mtime {
 374                                 response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 375                         }
 376                 } else {
 377                         response.Contents.BlockDigestToInfo[blockInfo.Digest] = blockInfo
 378                 }
 379         }
 380
 381         log.Printf("%s index contained %d lines with %d duplicates with "+
 382                 "%d size disagreements",
 383                 keepServer.String(),
 384                 numLines,
 385                 numDuplicates,
 386                 numSizeDisagreements)
 387
 388         if arvLogger != nil {
 389                 now := time.Now()
 390                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 391                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 392                         serverInfo := keepInfo[keepServer.Uuid].(map[string]interface{})
 393
 394                         serverInfo["processing_finished_at"] = now
 395                         serverInfo["lines_received"] = numLines
 396                         serverInfo["duplicates_seen"] = numDuplicates
 397                         serverInfo["size_disagreements_seen"] = numSizeDisagreements
 398                 })
 399         }
 400         resp.Body.Close()
 401         return
 402 }
 403
 404 func parseBlockInfoFromIndexLine(indexLine string) (blockInfo BlockInfo, err error) {
 405         tokens := strings.Fields(indexLine)
 406         if len(tokens) != 2 {
 407                 err = fmt.Errorf("Expected 2 tokens per line but received a "+
 408                         "line containing %v instead.",
 409                         tokens)
 410         }
 411
 412         var locator blockdigest.BlockLocator
 413         if locator, err = blockdigest.ParseBlockLocator(tokens[0]); err != nil {
 414                 err = fmt.Errorf("%v Received error while parsing line \"%s\"",
 415                         err, indexLine)
 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 =
 430                 blockdigest.DigestWithSize{Digest: locator.Digest,
 431                         Size: uint32(locator.Size)}
 432         return
 433 }
 434
 435 func (readServers *ReadServers) Summarize(arvLogger *logger.Logger) {
 436         readServers.BlockReplicationCounts = make(map[int]int)
 437         for _, infos := range readServers.BlockToServers {
 438                 replication := len(infos)
 439                 readServers.BlockReplicationCounts[replication] += 1
 440         }
 441
 442         if arvLogger != nil {
 443                 arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
 444                         keepInfo := logger.GetOrCreateMap(p, "keep_info")
 445                         keepInfo["distinct_blocks_stored"] = len(readServers.BlockToServers)
 446                 })
 447         }
 448
 449 }