services/keep-balance/main.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "encoding/json"
   9         "flag"
  10         "fmt"
  11         "log"
  12         "os"
  13         "os/signal"
  14         "syscall"
  15         "time"
  16
  17         "git.curoverse.com/arvados.git/sdk/go/arvados"
  18         "git.curoverse.com/arvados.git/sdk/go/config"
  19         arvadosVersion "git.curoverse.com/arvados.git/sdk/go/version"
  20 )
  21
  22 const defaultConfigPath = "/etc/arvados/keep-balance/keep-balance.yml"
  23
  24 // Config specifies site configuration, like API credentials and the
  25 // choice of which servers are to be balanced.
  26 //
  27 // Config is loaded from a JSON config file (see usage()).
  28 type Config struct {
  29         // Arvados API endpoint and credentials.
  30         Client arvados.Client
  31
  32         // List of service types (e.g., "disk") to balance.
  33         KeepServiceTypes []string
  34
  35         KeepServiceList arvados.KeepServiceList
  36
  37         // How often to check
  38         RunPeriod arvados.Duration
  39
  40         // Number of collections to request in each API call
  41         CollectionBatchSize int
  42
  43         // Max collections to buffer in memory (bigger values consume
  44         // more memory, but can reduce store-and-forward latency when
  45         // fetching pages)
  46         CollectionBuffers int
  47 }
  48
  49 // RunOptions controls runtime behavior. The flags/options that belong
  50 // here are the ones that are useful for interactive use. For example,
  51 // "CommitTrash" is a runtime option rather than a config item because
  52 // it invokes a troubleshooting feature rather than expressing how
  53 // balancing is meant to be done at a given site.
  54 //
  55 // RunOptions fields are controlled by command line flags.
  56 type RunOptions struct {
  57         Once        bool
  58         CommitPulls bool
  59         CommitTrash bool
  60         Logger      *log.Logger
  61         Dumper      *log.Logger
  62
  63         // SafeRendezvousState from the most recent balance operation,
  64         // or "" if unknown. If this changes from one run to the next,
  65         // we need to watch out for races. See
  66         // (*Balancer)ClearTrashLists.
  67         SafeRendezvousState string
  68 }
  69
  70 var debugf = func(string, ...interface{}) {}
  71
  72 func main() {
  73         var cfg Config
  74         var runOptions RunOptions
  75
  76         configPath := flag.String("config", defaultConfigPath,
  77                 "`path` of JSON or YAML configuration file")
  78         serviceListPath := flag.String("config.KeepServiceList", "",
  79                 "`path` of JSON or YAML file with list of keep services to balance, as given by \"arv keep_service list\" "+
  80                         "(default: config[\"KeepServiceList\"], or if none given, get all available services and filter by config[\"KeepServiceTypes\"])")
  81         flag.BoolVar(&runOptions.Once, "once", false,
  82                 "balance once and then exit")
  83         flag.BoolVar(&runOptions.CommitPulls, "commit-pulls", false,
  84                 "send pull requests (make more replicas of blocks that are underreplicated or are not in optimal rendezvous probe order)")
  85         flag.BoolVar(&runOptions.CommitTrash, "commit-trash", false,
  86                 "send trash requests (delete unreferenced old blocks, and excess replicas of overreplicated blocks)")
  87         dumpConfig := flag.Bool("dump-config", false, "write current configuration to stdout and exit")
  88         dumpFlag := flag.Bool("dump", false, "dump details for each block to stdout")
  89         debugFlag := flag.Bool("debug", false, "enable debug messages")
  90         getVersion := flag.Bool("version", false, "Print version information and exit.")
  91         flag.Usage = usage
  92         flag.Parse()
  93
  94         // Print version information if requested
  95         if *getVersion {
  96                 fmt.Printf("Version: %s\n", arvadosVersion.GetVersion())
  97                 os.Exit(0)
  98         }
  99
 100         mustReadConfig(&cfg, *configPath)
 101         if *serviceListPath != "" {
 102                 mustReadConfig(&cfg.KeepServiceList, *serviceListPath)
 103         }
 104
 105         if *dumpConfig {
 106                 log.Fatal(config.DumpAndExit(cfg))
 107         }
 108
 109         log.Printf("keep-balance %q started", arvadosVersion.GetVersion())
 110
 111         if *debugFlag {
 112                 debugf = log.Printf
 113                 if j, err := json.Marshal(cfg); err != nil {
 114                         log.Fatal(err)
 115                 } else {
 116                         log.Printf("config is %s", j)
 117                 }
 118         }
 119         if *dumpFlag {
 120                 runOptions.Dumper = log.New(os.Stdout, "", log.LstdFlags)
 121         }
 122         err := CheckConfig(cfg, runOptions)
 123         if err != nil {
 124                 // (don't run)
 125         } else if runOptions.Once {
 126                 _, err = (&Balancer{}).Run(cfg, runOptions)
 127         } else {
 128                 err = RunForever(cfg, runOptions, nil)
 129         }
 130         if err != nil {
 131                 log.Fatal(err)
 132         }
 133 }
 134
 135 func mustReadConfig(dst interface{}, path string) {
 136         if err := config.LoadFile(dst, path); err != nil {
 137                 log.Fatal(err)
 138         }
 139 }
 140
 141 // RunForever runs forever, or (for testing purposes) until the given
 142 // stop channel is ready to receive.
 143 func RunForever(config Config, runOptions RunOptions, stop <-chan interface{}) error {
 144         if runOptions.Logger == nil {
 145                 runOptions.Logger = log.New(os.Stderr, "", log.LstdFlags)
 146         }
 147         logger := runOptions.Logger
 148
 149         ticker := time.NewTicker(time.Duration(config.RunPeriod))
 150
 151         // The unbuffered channel here means we only hear SIGUSR1 if
 152         // it arrives while we're waiting in select{}.
 153         sigUSR1 := make(chan os.Signal)
 154         signal.Notify(sigUSR1, syscall.SIGUSR1)
 155
 156         logger.Printf("starting up: will scan every %v and on SIGUSR1", config.RunPeriod)
 157
 158         for {
 159                 if !runOptions.CommitPulls && !runOptions.CommitTrash {
 160                         logger.Print("WARNING: Will scan periodically, but no changes will be committed.")
 161                         logger.Print("=======  Consider using -commit-pulls and -commit-trash flags.")
 162                 }
 163
 164                 bal := &Balancer{}
 165                 var err error
 166                 runOptions, err = bal.Run(config, runOptions)
 167                 if err != nil {
 168                         logger.Print("run failed: ", err)
 169                 } else {
 170                         logger.Print("run succeeded")
 171                 }
 172
 173                 select {
 174                 case <-stop:
 175                         signal.Stop(sigUSR1)
 176                         return nil
 177                 case <-ticker.C:
 178                         logger.Print("timer went off")
 179                 case <-sigUSR1:
 180                         logger.Print("received SIGUSR1, resetting timer")
 181                         // Reset the timer so we don't start the N+1st
 182                         // run too soon after the Nth run is triggered
 183                         // by SIGUSR1.
 184                         ticker.Stop()
 185                         ticker = time.NewTicker(time.Duration(config.RunPeriod))
 186                 }
 187                 logger.Print("starting next run")
 188         }
 189 }