13 "git.curoverse.com/arvados.git/sdk/go/arvados"
16 // Config specifies site configuration, like API credentials and the
17 // choice of which servers are to be balanced.
19 // Config is loaded from a JSON config file (see usage()).
21 // Arvados API endpoint and credentials.
24 // List of service types (e.g., "disk") to balance.
25 KeepServiceTypes []string
27 KeepServiceList arvados.KeepServiceList
30 RunPeriod arvados.Duration
32 // Number of collections to request in each API call
33 CollectionBatchSize int
35 // Max collections to buffer in memory (bigger values consume
36 // more memory, but can reduce store-and-forward latency when
41 // RunOptions controls runtime behavior. The flags/options that belong
42 // here are the ones that are useful for interactive use. For example,
43 // "CommitTrash" is a runtime option rather than a config item because
44 // it invokes a troubleshooting feature rather than expressing how
45 // balancing is meant to be done at a given site.
47 // RunOptions fields are controlled by command line flags.
48 type RunOptions struct {
55 // SafeRendezvousState from the most recent balance operation,
56 // or "" if unknown. If this changes from one run to the next,
57 // we need to watch out for races. See
58 // (*Balancer)ClearTrashLists.
59 SafeRendezvousState string
62 var debugf = func(string, ...interface{}) {}
66 var runOptions RunOptions
68 configPath := flag.String("config", "",
69 "`path` of json configuration file")
70 serviceListPath := flag.String("config.KeepServiceList", "",
71 "`path` of json file with list of keep services to balance, as given by \"arv keep_service list\" "+
72 "(default: config[\"KeepServiceList\"], or if none given, get all available services and filter by config[\"KeepServiceTypes\"])")
73 flag.BoolVar(&runOptions.Once, "once", false,
74 "balance once and then exit")
75 flag.BoolVar(&runOptions.CommitPulls, "commit-pulls", false,
76 "send pull requests (make more replicas of blocks that are underreplicated or are not in optimal rendezvous probe order)")
77 flag.BoolVar(&runOptions.CommitTrash, "commit-trash", false,
78 "send trash requests (delete unreferenced old blocks, and excess replicas of overreplicated blocks)")
79 dumpFlag := flag.Bool("dump", false, "dump details for each block to stdout")
80 debugFlag := flag.Bool("debug", false, "enable debug messages")
84 if *configPath == "" {
85 log.Fatal("You must specify a config file (see `keep-balance -help`)")
87 mustReadJSON(&config, *configPath)
88 if *serviceListPath != "" {
89 mustReadJSON(&config.KeepServiceList, *serviceListPath)
94 if j, err := json.Marshal(config); err != nil {
97 log.Printf("config is %s", j)
101 runOptions.Dumper = log.New(os.Stdout, "", log.LstdFlags)
103 err := CheckConfig(config, runOptions)
106 } else if runOptions.Once {
107 _, err = (&Balancer{}).Run(config, runOptions)
109 err = RunForever(config, runOptions, nil)
116 func mustReadJSON(dst interface{}, path string) {
117 if buf, err := ioutil.ReadFile(path); err != nil {
118 log.Fatalf("Reading %q: %v", path, err)
119 } else if err = json.Unmarshal(buf, dst); err != nil {
120 log.Fatalf("Decoding %q: %v", path, err)
124 // RunForever runs forever, or (for testing purposes) until the given
125 // stop channel is ready to receive.
126 func RunForever(config Config, runOptions RunOptions, stop <-chan interface{}) error {
127 if runOptions.Logger == nil {
128 runOptions.Logger = log.New(os.Stderr, "", log.LstdFlags)
130 logger := runOptions.Logger
132 ticker := time.NewTicker(time.Duration(config.RunPeriod))
134 // The unbuffered channel here means we only hear SIGUSR1 if
135 // it arrives while we're waiting in select{}.
136 sigUSR1 := make(chan os.Signal)
137 signal.Notify(sigUSR1, syscall.SIGUSR1)
139 logger.Printf("starting up: will scan every %v and on SIGUSR1", config.RunPeriod)
142 if !runOptions.CommitPulls && !runOptions.CommitTrash {
143 logger.Print("WARNING: Will scan periodically, but no changes will be committed.")
144 logger.Print("======= Consider using -commit-pulls and -commit-trash flags.")
149 runOptions, err = bal.Run(config, runOptions)
151 logger.Print("run failed: ", err)
153 logger.Print("run succeeded")
161 logger.Print("timer went off")
163 logger.Print("received SIGUSR1, resetting timer")
164 // Reset the timer so we don't start the N+1st
165 // run too soon after the Nth run is triggered
168 ticker = time.NewTicker(time.Duration(config.RunPeriod))
170 logger.Print("starting next run")