services/datamanager/datamanager.go

   1 /* Keep Datamanager. Responsible for checking on and reporting on Keep Storage */
   2
   3 package main
   4
   5 import (
   6         "flag"
   7         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
   8         "git.curoverse.com/arvados.git/sdk/go/logger"
   9         "git.curoverse.com/arvados.git/sdk/go/util"
  10         "git.curoverse.com/arvados.git/services/datamanager/collection"
  11         "git.curoverse.com/arvados.git/services/datamanager/keep"
  12         "git.curoverse.com/arvados.git/services/datamanager/loggerutil"
  13         "log"
  14         "time"
  15 )
  16
  17 var (
  18         logEventTypePrefix  string
  19         logFrequencySeconds int
  20         minutesBetweenRuns  int
  21 )
  22
  23 func init() {
  24         flag.StringVar(&logEventTypePrefix,
  25                 "log-event-type-prefix",
  26                 "experimental-data-manager",
  27                 "Prefix to use in the event_type of our arvados log entries. Set to empty to turn off logging")
  28         flag.IntVar(&logFrequencySeconds,
  29                 "log-frequency-seconds",
  30                 20,
  31                 "How frequently we'll write log entries in seconds.")
  32         flag.IntVar(&minutesBetweenRuns,
  33                 "minutes-between-runs",
  34                 0,
  35                 "How many minutes we wait betwen data manager runs. 0 means run once and exit.")
  36 }
  37
  38 func main() {
  39         flag.Parse()
  40         if minutesBetweenRuns == 0 {
  41                 singlerun()
  42         } else {
  43                 waitTime := time.Minute * time.Duration(minutesBetweenRuns)
  44                 for {
  45                         log.Println("Beginning Run")
  46                         singlerun()
  47                         log.Printf("Sleeping for %d minutes", minutesBetweenRuns)
  48                         time.Sleep(waitTime)
  49                 }
  50         }
  51 }
  52
  53 func singlerun() {
  54         arv, err := arvadosclient.MakeArvadosClient()
  55         if err != nil {
  56                 log.Fatalf("Error setting up arvados client %s", err.Error())
  57         }
  58
  59         if is_admin, err := util.UserIsAdmin(arv); err != nil {
  60                 log.Fatalf("Error querying current arvados user %s", err.Error())
  61         } else if !is_admin {
  62                 log.Fatalf("Current user is not an admin. Datamanager can only be run by admins.")
  63         }
  64
  65         var arvLogger *logger.Logger
  66         if logEventTypePrefix != "" {
  67                 arvLogger = logger.NewLogger(logger.LoggerParams{Client: arv,
  68                         EventTypePrefix: logEventTypePrefix,
  69                         WriteInterval:   time.Second * time.Duration(logFrequencySeconds)})
  70         }
  71
  72         loggerutil.LogRunInfo(arvLogger)
  73         if arvLogger != nil {
  74                 arvLogger.AddWriteHook(loggerutil.LogMemoryAlloc)
  75         }
  76
  77         collectionChannel := make(chan collection.ReadCollections)
  78
  79         go func() {
  80                 collectionChannel <- collection.GetCollectionsAndSummarize(
  81                         collection.GetCollectionsParams{
  82                                 Client: arv, Logger: arvLogger, BatchSize: 50})
  83         }()
  84
  85         keepServerInfo := keep.GetKeepServersAndSummarize(
  86                 keep.GetKeepServersParams{Client: arv, Logger: arvLogger, Limit: 1000})
  87
  88         readCollections := <-collectionChannel
  89
  90         // TODO(misha): Use these together to verify replication.
  91         _ = readCollections
  92         _ = keepServerInfo
  93
  94         // Log that we're finished. We force the recording, since go will
  95         // not wait for the timer before exiting.
  96         if arvLogger != nil {
  97                 arvLogger.FinalUpdate(func(p map[string]interface{}, e map[string]interface{}) {
  98                         p["run_info"].(map[string]interface{})["finished_at"] = time.Now()
  99                 })
 100         }
 101 }