13427: Ignore readonly devices mounted read-write elsewhere.
[arvados.git] / services / keep-balance / balance.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         "bytes"
9         "crypto/md5"
10         "fmt"
11         "log"
12         "math"
13         "os"
14         "runtime"
15         "sort"
16         "strings"
17         "sync"
18         "time"
19
20         "git.curoverse.com/arvados.git/sdk/go/arvados"
21         "git.curoverse.com/arvados.git/sdk/go/keepclient"
22 )
23
24 // CheckConfig returns an error if anything is wrong with the given
25 // config and runOptions.
26 func CheckConfig(config Config, runOptions RunOptions) error {
27         if len(config.KeepServiceList.Items) > 0 && config.KeepServiceTypes != nil {
28                 return fmt.Errorf("cannot specify both KeepServiceList and KeepServiceTypes in config")
29         }
30         if !runOptions.Once && config.RunPeriod == arvados.Duration(0) {
31                 return fmt.Errorf("you must either use the -once flag, or specify RunPeriod in config")
32         }
33         return nil
34 }
35
36 // Balancer compares the contents of keepstore servers with the
37 // collections stored in Arvados, and issues pull/trash requests
38 // needed to get (closer to) the optimal data layout.
39 //
40 // In the optimal data layout: every data block referenced by a
41 // collection is replicated at least as many times as desired by the
42 // collection; there are no unreferenced data blocks older than
43 // BlobSignatureTTL; and all N existing replicas of a given data block
44 // are in the N best positions in rendezvous probe order.
45 type Balancer struct {
46         *BlockStateMap
47         KeepServices       map[string]*KeepService
48         DefaultReplication int
49         Logger             *log.Logger
50         Dumper             *log.Logger
51         MinMtime           int64
52
53         classes       []string
54         mounts        int
55         mountsByClass map[string]map[*KeepMount]bool
56         collScanned   int
57         serviceRoots  map[string]string
58         errors        []error
59         stats         balancerStats
60         mutex         sync.Mutex
61 }
62
63 // Run performs a balance operation using the given config and
64 // runOptions, and returns RunOptions suitable for passing to a
65 // subsequent balance operation.
66 //
67 // Run should only be called once on a given Balancer object.
68 //
69 // Typical usage:
70 //
71 //   runOptions, err = (&Balancer{}).Run(config, runOptions)
72 func (bal *Balancer) Run(config Config, runOptions RunOptions) (nextRunOptions RunOptions, err error) {
73         nextRunOptions = runOptions
74
75         bal.Dumper = runOptions.Dumper
76         bal.Logger = runOptions.Logger
77         if bal.Logger == nil {
78                 bal.Logger = log.New(os.Stderr, "", log.LstdFlags)
79         }
80
81         defer timeMe(bal.Logger, "Run")()
82
83         if len(config.KeepServiceList.Items) > 0 {
84                 err = bal.SetKeepServices(config.KeepServiceList)
85         } else {
86                 err = bal.DiscoverKeepServices(&config.Client, config.KeepServiceTypes)
87         }
88         if err != nil {
89                 return
90         }
91
92         for _, srv := range bal.KeepServices {
93                 err = srv.discoverMounts(&config.Client)
94                 if err != nil {
95                         return
96                 }
97         }
98         bal.dedupDevices()
99
100         if err = bal.CheckSanityEarly(&config.Client); err != nil {
101                 return
102         }
103         rs := bal.rendezvousState()
104         if runOptions.CommitTrash && rs != runOptions.SafeRendezvousState {
105                 if runOptions.SafeRendezvousState != "" {
106                         bal.logf("notice: KeepServices list has changed since last run")
107                 }
108                 bal.logf("clearing existing trash lists, in case the new rendezvous order differs from previous run")
109                 if err = bal.ClearTrashLists(&config.Client); err != nil {
110                         return
111                 }
112                 // The current rendezvous state becomes "safe" (i.e.,
113                 // OK to compute changes for that state without
114                 // clearing existing trash lists) only now, after we
115                 // succeed in clearing existing trash lists.
116                 nextRunOptions.SafeRendezvousState = rs
117         }
118         if err = bal.GetCurrentState(&config.Client, config.CollectionBatchSize, config.CollectionBuffers); err != nil {
119                 return
120         }
121         bal.ComputeChangeSets()
122         bal.PrintStatistics()
123         if err = bal.CheckSanityLate(); err != nil {
124                 return
125         }
126         if runOptions.CommitPulls {
127                 err = bal.CommitPulls(&config.Client)
128                 if err != nil {
129                         // Skip trash if we can't pull. (Too cautious?)
130                         return
131                 }
132         }
133         if runOptions.CommitTrash {
134                 err = bal.CommitTrash(&config.Client)
135         }
136         return
137 }
138
139 // SetKeepServices sets the list of KeepServices to operate on.
140 func (bal *Balancer) SetKeepServices(srvList arvados.KeepServiceList) error {
141         bal.KeepServices = make(map[string]*KeepService)
142         for _, srv := range srvList.Items {
143                 bal.KeepServices[srv.UUID] = &KeepService{
144                         KeepService: srv,
145                         ChangeSet:   &ChangeSet{},
146                 }
147         }
148         return nil
149 }
150
151 // DiscoverKeepServices sets the list of KeepServices by calling the
152 // API to get a list of all services, and selecting the ones whose
153 // ServiceType is in okTypes.
154 func (bal *Balancer) DiscoverKeepServices(c *arvados.Client, okTypes []string) error {
155         bal.KeepServices = make(map[string]*KeepService)
156         ok := make(map[string]bool)
157         for _, t := range okTypes {
158                 ok[t] = true
159         }
160         return c.EachKeepService(func(srv arvados.KeepService) error {
161                 if ok[srv.ServiceType] {
162                         bal.KeepServices[srv.UUID] = &KeepService{
163                                 KeepService: srv,
164                                 ChangeSet:   &ChangeSet{},
165                         }
166                 } else {
167                         bal.logf("skipping %v with service type %q", srv.UUID, srv.ServiceType)
168                 }
169                 return nil
170         })
171 }
172
173 func (bal *Balancer) dedupDevices() {
174         rwdev := map[string]*KeepService{}
175         for _, srv := range bal.KeepServices {
176                 for _, mnt := range srv.mounts {
177                         if !mnt.ReadOnly && mnt.DeviceID != "" {
178                                 rwdev[mnt.DeviceID] = srv
179                         }
180                 }
181         }
182         // Drop the readonly mounts whose device is mounted RW
183         // elsewhere.
184         for _, srv := range bal.KeepServices {
185                 var dedup []*KeepMount
186                 for _, mnt := range srv.mounts {
187                         if mnt.ReadOnly && rwdev[mnt.DeviceID] != nil {
188                                 bal.logf("skipping srv %s readonly mount %q because same device %q is mounted read-write on srv %s", srv, mnt.UUID, mnt.DeviceID, rwdev[mnt.DeviceID])
189                         } else {
190                                 dedup = append(dedup, mnt)
191                         }
192                 }
193                 srv.mounts = dedup
194         }
195 }
196
197 // CheckSanityEarly checks for configuration and runtime errors that
198 // can be detected before GetCurrentState() and ComputeChangeSets()
199 // are called.
200 //
201 // If it returns an error, it is pointless to run GetCurrentState or
202 // ComputeChangeSets: after doing so, the statistics would be
203 // meaningless and it would be dangerous to run any Commit methods.
204 func (bal *Balancer) CheckSanityEarly(c *arvados.Client) error {
205         u, err := c.CurrentUser()
206         if err != nil {
207                 return fmt.Errorf("CurrentUser(): %v", err)
208         }
209         if !u.IsActive || !u.IsAdmin {
210                 return fmt.Errorf("current user (%s) is not an active admin user", u.UUID)
211         }
212         for _, srv := range bal.KeepServices {
213                 if srv.ServiceType == "proxy" {
214                         return fmt.Errorf("config error: %s: proxy servers cannot be balanced", srv)
215                 }
216         }
217         return nil
218 }
219
220 // rendezvousState returns a fingerprint (e.g., a sorted list of
221 // UUID+host+port) of the current set of keep services.
222 func (bal *Balancer) rendezvousState() string {
223         srvs := make([]string, 0, len(bal.KeepServices))
224         for _, srv := range bal.KeepServices {
225                 srvs = append(srvs, srv.String())
226         }
227         sort.Strings(srvs)
228         return strings.Join(srvs, "; ")
229 }
230
231 // ClearTrashLists sends an empty trash list to each keep
232 // service. Calling this before GetCurrentState avoids races.
233 //
234 // When a block appears in an index, we assume that replica will still
235 // exist after we delete other replicas on other servers. However,
236 // it's possible that a previous rebalancing operation made different
237 // decisions (e.g., servers were added/removed, and rendezvous order
238 // changed). In this case, the replica might already be on that
239 // server's trash list, and it might be deleted before we send a
240 // replacement trash list.
241 //
242 // We avoid this problem if we clear all trash lists before getting
243 // indexes. (We also assume there is only one rebalancing process
244 // running at a time.)
245 func (bal *Balancer) ClearTrashLists(c *arvados.Client) error {
246         for _, srv := range bal.KeepServices {
247                 srv.ChangeSet = &ChangeSet{}
248         }
249         return bal.CommitTrash(c)
250 }
251
252 // GetCurrentState determines the current replication state, and the
253 // desired replication level, for every block that is either
254 // retrievable or referenced.
255 //
256 // It determines the current replication state by reading the block index
257 // from every known Keep service.
258 //
259 // It determines the desired replication level by retrieving all
260 // collection manifests in the database (API server).
261 //
262 // It encodes the resulting information in BlockStateMap.
263 func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) error {
264         defer timeMe(bal.Logger, "GetCurrentState")()
265         bal.BlockStateMap = NewBlockStateMap()
266
267         dd, err := c.DiscoveryDocument()
268         if err != nil {
269                 return err
270         }
271         bal.DefaultReplication = dd.DefaultCollectionReplication
272         bal.MinMtime = time.Now().UnixNano() - dd.BlobSignatureTTL*1e9
273
274         errs := make(chan error, 2+len(bal.KeepServices))
275         wg := sync.WaitGroup{}
276
277         // Start one goroutine for each KeepService: retrieve the
278         // index, and add the returned blocks to BlockStateMap.
279         for _, srv := range bal.KeepServices {
280                 wg.Add(1)
281                 go func(srv *KeepService) {
282                         defer wg.Done()
283                         bal.logf("%s: retrieve indexes", srv)
284                         for _, mount := range srv.mounts {
285                                 bal.logf("%s: retrieve index", mount)
286                                 idx, err := srv.IndexMount(c, mount.UUID, "")
287                                 if err != nil {
288                                         errs <- fmt.Errorf("%s: retrieve index: %v", mount, err)
289                                         return
290                                 }
291                                 if len(errs) > 0 {
292                                         // Some other goroutine encountered an
293                                         // error -- any further effort here
294                                         // will be wasted.
295                                         return
296                                 }
297                                 bal.logf("%s: add %d replicas to map", mount, len(idx))
298                                 bal.BlockStateMap.AddReplicas(mount, idx)
299                                 bal.logf("%s: done", mount)
300                         }
301                         bal.logf("%s: done", srv)
302                 }(srv)
303         }
304
305         // collQ buffers incoming collections so we can start fetching
306         // the next page without waiting for the current page to
307         // finish processing.
308         collQ := make(chan arvados.Collection, bufs)
309
310         // Start a goroutine to process collections. (We could use a
311         // worker pool here, but even with a single worker we already
312         // process collections much faster than we can retrieve them.)
313         wg.Add(1)
314         go func() {
315                 defer wg.Done()
316                 for coll := range collQ {
317                         err := bal.addCollection(coll)
318                         if err != nil {
319                                 errs <- err
320                                 for range collQ {
321                                 }
322                                 return
323                         }
324                         bal.collScanned++
325                 }
326         }()
327
328         // Start a goroutine to retrieve all collections from the
329         // Arvados database and send them to collQ for processing.
330         wg.Add(1)
331         go func() {
332                 defer wg.Done()
333                 err = EachCollection(c, pageSize,
334                         func(coll arvados.Collection) error {
335                                 collQ <- coll
336                                 if len(errs) > 0 {
337                                         // some other GetCurrentState
338                                         // error happened: no point
339                                         // getting any more
340                                         // collections.
341                                         return fmt.Errorf("")
342                                 }
343                                 return nil
344                         }, func(done, total int) {
345                                 bal.logf("collections: %d/%d", done, total)
346                         })
347                 close(collQ)
348                 if err != nil {
349                         errs <- err
350                 }
351         }()
352
353         wg.Wait()
354         if len(errs) > 0 {
355                 return <-errs
356         }
357         return nil
358 }
359
360 func (bal *Balancer) addCollection(coll arvados.Collection) error {
361         blkids, err := coll.SizedDigests()
362         if err != nil {
363                 bal.mutex.Lock()
364                 bal.errors = append(bal.errors, fmt.Errorf("%v: %v", coll.UUID, err))
365                 bal.mutex.Unlock()
366                 return nil
367         }
368         repl := bal.DefaultReplication
369         if coll.ReplicationDesired != nil {
370                 repl = *coll.ReplicationDesired
371         }
372         debugf("%v: %d block x%d", coll.UUID, len(blkids), repl)
373         bal.BlockStateMap.IncreaseDesired(coll.StorageClassesDesired, repl, blkids)
374         return nil
375 }
376
377 // ComputeChangeSets compares, for each known block, the current and
378 // desired replication states. If it is possible to get closer to the
379 // desired state by copying or deleting blocks, it adds those changes
380 // to the relevant KeepServices' ChangeSets.
381 //
382 // It does not actually apply any of the computed changes.
383 func (bal *Balancer) ComputeChangeSets() {
384         // This just calls balanceBlock() once for each block, using a
385         // pool of worker goroutines.
386         defer timeMe(bal.Logger, "ComputeChangeSets")()
387         bal.setupLookupTables()
388
389         type balanceTask struct {
390                 blkid arvados.SizedDigest
391                 blk   *BlockState
392         }
393         workers := runtime.GOMAXPROCS(-1)
394         todo := make(chan balanceTask, workers)
395         go func() {
396                 bal.BlockStateMap.Apply(func(blkid arvados.SizedDigest, blk *BlockState) {
397                         todo <- balanceTask{
398                                 blkid: blkid,
399                                 blk:   blk,
400                         }
401                 })
402                 close(todo)
403         }()
404         results := make(chan balanceResult, workers)
405         go func() {
406                 var wg sync.WaitGroup
407                 for i := 0; i < workers; i++ {
408                         wg.Add(1)
409                         go func() {
410                                 for work := range todo {
411                                         results <- bal.balanceBlock(work.blkid, work.blk)
412                                 }
413                                 wg.Done()
414                         }()
415                 }
416                 wg.Wait()
417                 close(results)
418         }()
419         bal.collectStatistics(results)
420 }
421
422 func (bal *Balancer) setupLookupTables() {
423         bal.serviceRoots = make(map[string]string)
424         bal.classes = []string{"default"}
425         bal.mountsByClass = map[string]map[*KeepMount]bool{"default": {}}
426         bal.mounts = 0
427         for _, srv := range bal.KeepServices {
428                 bal.serviceRoots[srv.UUID] = srv.UUID
429                 for _, mnt := range srv.mounts {
430                         bal.mounts++
431
432                         // All mounts on a read-only service are
433                         // effectively read-only.
434                         mnt.ReadOnly = mnt.ReadOnly || srv.ReadOnly
435
436                         if len(mnt.StorageClasses) == 0 {
437                                 bal.mountsByClass["default"][mnt] = true
438                                 continue
439                         }
440                         for _, class := range mnt.StorageClasses {
441                                 if mbc := bal.mountsByClass[class]; mbc == nil {
442                                         bal.classes = append(bal.classes, class)
443                                         bal.mountsByClass[class] = map[*KeepMount]bool{mnt: true}
444                                 } else {
445                                         mbc[mnt] = true
446                                 }
447                         }
448                 }
449         }
450         // Consider classes in lexicographic order to avoid flapping
451         // between balancing runs.  The outcome of the "prefer a mount
452         // we're already planning to use for a different storage
453         // class" case in balanceBlock depends on the order classes
454         // are considered.
455         sort.Strings(bal.classes)
456 }
457
458 const (
459         changeStay = iota
460         changePull
461         changeTrash
462         changeNone
463 )
464
465 var changeName = map[int]string{
466         changeStay:  "stay",
467         changePull:  "pull",
468         changeTrash: "trash",
469         changeNone:  "none",
470 }
471
472 type balanceResult struct {
473         blk        *BlockState
474         blkid      arvados.SizedDigest
475         have       int
476         want       int
477         classState map[string]balancedBlockState
478 }
479
480 // balanceBlock compares current state to desired state for a single
481 // block, and makes the appropriate ChangeSet calls.
482 func (bal *Balancer) balanceBlock(blkid arvados.SizedDigest, blk *BlockState) balanceResult {
483         debugf("balanceBlock: %v %+v", blkid, blk)
484
485         type slot struct {
486                 mnt  *KeepMount // never nil
487                 repl *Replica   // replica already stored here (or nil)
488                 want bool       // we should pull/leave a replica here
489         }
490
491         // Build a list of all slots (one per mounted volume).
492         slots := make([]slot, 0, bal.mounts)
493         for _, srv := range bal.KeepServices {
494                 for _, mnt := range srv.mounts {
495                         var repl *Replica
496                         for r := range blk.Replicas {
497                                 if blk.Replicas[r].KeepMount == mnt {
498                                         repl = &blk.Replicas[r]
499                                 }
500                         }
501                         // Initial value of "want" is "have, and can't
502                         // delete". These untrashable replicas get
503                         // prioritized when sorting slots: otherwise,
504                         // non-optimal readonly copies would cause us
505                         // to overreplicate.
506                         slots = append(slots, slot{
507                                 mnt:  mnt,
508                                 repl: repl,
509                                 want: repl != nil && (mnt.ReadOnly || repl.Mtime >= bal.MinMtime),
510                         })
511                 }
512         }
513
514         uuids := keepclient.NewRootSorter(bal.serviceRoots, string(blkid[:32])).GetSortedRoots()
515         srvRendezvous := make(map[*KeepService]int, len(uuids))
516         for i, uuid := range uuids {
517                 srv := bal.KeepServices[uuid]
518                 srvRendezvous[srv] = i
519         }
520
521         // Below we set underreplicated=true if we find any storage
522         // class that's currently underreplicated -- in that case we
523         // won't want to trash any replicas.
524         underreplicated := false
525
526         classState := make(map[string]balancedBlockState, len(bal.classes))
527         unsafeToDelete := make(map[int64]bool, len(slots))
528         for _, class := range bal.classes {
529                 desired := blk.Desired[class]
530
531                 have := 0
532                 for _, slot := range slots {
533                         if slot.repl != nil && bal.mountsByClass[class][slot.mnt] {
534                                 have++
535                         }
536                 }
537                 classState[class] = balancedBlockState{
538                         desired: desired,
539                         surplus: have - desired,
540                 }
541
542                 if desired == 0 {
543                         continue
544                 }
545
546                 // Sort the slots by desirability.
547                 sort.Slice(slots, func(i, j int) bool {
548                         si, sj := slots[i], slots[j]
549                         if classi, classj := bal.mountsByClass[class][si.mnt], bal.mountsByClass[class][sj.mnt]; classi != classj {
550                                 // Prefer a mount that satisfies the
551                                 // desired class.
552                                 return bal.mountsByClass[class][si.mnt]
553                         } else if wanti, wantj := si.want, si.want; wanti != wantj {
554                                 // Prefer a mount that will have a
555                                 // replica no matter what we do here
556                                 // -- either because it already has an
557                                 // untrashable replica, or because we
558                                 // already need it to satisfy a
559                                 // different storage class.
560                                 return slots[i].want
561                         } else if orderi, orderj := srvRendezvous[si.mnt.KeepService], srvRendezvous[sj.mnt.KeepService]; orderi != orderj {
562                                 // Prefer a better rendezvous
563                                 // position.
564                                 return orderi < orderj
565                         } else if repli, replj := si.repl != nil, sj.repl != nil; repli != replj {
566                                 // Prefer a mount that already has a
567                                 // replica.
568                                 return repli
569                         } else {
570                                 // If pull/trash turns out to be
571                                 // needed, distribute the
572                                 // new/remaining replicas uniformly
573                                 // across qualifying mounts on a given
574                                 // server.
575                                 return rendezvousLess(si.mnt.DeviceID, sj.mnt.DeviceID, blkid)
576                         }
577                 })
578
579                 // Servers and mounts (with or without existing
580                 // replicas) that are part of the best achievable
581                 // layout for this storage class.
582                 wantSrv := map[*KeepService]bool{}
583                 wantMnt := map[*KeepMount]bool{}
584                 // Positions (with existing replicas) that have been
585                 // protected (via unsafeToDelete) to ensure we don't
586                 // reduce replication below desired level when
587                 // trashing replicas that aren't optimal positions for
588                 // any storage class.
589                 protMnt := map[*KeepMount]bool{}
590
591                 // trySlot tries using a slot to meet requirements,
592                 // and returns true if all requirements are met.
593                 trySlot := func(i int) bool {
594                         slot := slots[i]
595                         if len(protMnt) < desired && slot.repl != nil {
596                                 unsafeToDelete[slot.repl.Mtime] = true
597                                 protMnt[slot.mnt] = true
598                         }
599                         if len(wantMnt) < desired && (slot.repl != nil || !slot.mnt.ReadOnly) {
600                                 slots[i].want = true
601                                 wantSrv[slot.mnt.KeepService] = true
602                                 wantMnt[slot.mnt] = true
603                         }
604                         return len(protMnt) >= desired && len(wantMnt) >= desired
605                 }
606
607                 // First try to achieve desired replication without
608                 // using the same server twice.
609                 done := false
610                 for i := 0; i < len(slots) && !done; i++ {
611                         if !wantSrv[slots[i].mnt.KeepService] {
612                                 done = trySlot(i)
613                         }
614                 }
615
616                 // If that didn't suffice, do another pass without the
617                 // "distinct services" restriction. (Achieving the
618                 // desired volume replication on fewer than the
619                 // desired number of services is better than
620                 // underreplicating.)
621                 for i := 0; i < len(slots) && !done; i++ {
622                         done = trySlot(i)
623                 }
624
625                 if !underreplicated {
626                         safe := 0
627                         for _, slot := range slots {
628                                 if slot.repl == nil || !bal.mountsByClass[class][slot.mnt] {
629                                         continue
630                                 }
631                                 if safe++; safe >= desired {
632                                         break
633                                 }
634                         }
635                         underreplicated = safe < desired
636                 }
637
638                 // set the unachievable flag if there aren't enough
639                 // slots offering the relevant storage class. (This is
640                 // as easy as checking slots[desired] because we
641                 // already sorted the qualifying slots to the front.)
642                 if desired >= len(slots) || !bal.mountsByClass[class][slots[desired].mnt] {
643                         cs := classState[class]
644                         cs.unachievable = true
645                         classState[class] = cs
646                 }
647         }
648
649         // TODO: If multiple replicas are trashable, prefer the oldest
650         // replica that doesn't have a timestamp collision with
651         // others.
652
653         var have, want int
654         for _, slot := range slots {
655                 if slot.want {
656                         want++
657                 }
658                 if slot.repl != nil {
659                         have++
660                 }
661         }
662
663         var changes []string
664         for _, slot := range slots {
665                 // TODO: request a Touch if Mtime is duplicated.
666                 var change int
667                 switch {
668                 case !underreplicated && slot.repl != nil && !slot.want && !unsafeToDelete[slot.repl.Mtime]:
669                         slot.mnt.KeepService.AddTrash(Trash{
670                                 SizedDigest: blkid,
671                                 Mtime:       slot.repl.Mtime,
672                                 From:        slot.mnt,
673                         })
674                         change = changeTrash
675                 case len(blk.Replicas) == 0:
676                         change = changeNone
677                 case slot.repl == nil && slot.want && !slot.mnt.ReadOnly:
678                         slot.mnt.KeepService.AddPull(Pull{
679                                 SizedDigest: blkid,
680                                 From:        blk.Replicas[0].KeepMount.KeepService,
681                                 To:          slot.mnt,
682                         })
683                         change = changePull
684                 default:
685                         change = changeStay
686                 }
687                 if bal.Dumper != nil {
688                         var mtime int64
689                         if slot.repl != nil {
690                                 mtime = slot.repl.Mtime
691                         }
692                         srv := slot.mnt.KeepService
693                         changes = append(changes, fmt.Sprintf("%s:%d/%s=%s,%d", srv.ServiceHost, srv.ServicePort, slot.mnt.UUID, changeName[change], mtime))
694                 }
695         }
696         if bal.Dumper != nil {
697                 bal.Dumper.Printf("%s have=%d want=%v %s", blkid, have, want, strings.Join(changes, " "))
698         }
699         return balanceResult{
700                 blk:        blk,
701                 blkid:      blkid,
702                 have:       have,
703                 want:       want,
704                 classState: classState,
705         }
706 }
707
708 type blocksNBytes struct {
709         replicas int
710         blocks   int
711         bytes    int64
712 }
713
714 func (bb blocksNBytes) String() string {
715         return fmt.Sprintf("%d replicas (%d blocks, %d bytes)", bb.replicas, bb.blocks, bb.bytes)
716 }
717
718 type balancerStats struct {
719         lost          blocksNBytes
720         overrep       blocksNBytes
721         unref         blocksNBytes
722         garbage       blocksNBytes
723         underrep      blocksNBytes
724         unachievable  blocksNBytes
725         justright     blocksNBytes
726         desired       blocksNBytes
727         current       blocksNBytes
728         pulls         int
729         trashes       int
730         replHistogram []int
731         classStats    map[string]replicationStats
732 }
733
734 type replicationStats struct {
735         desired      blocksNBytes
736         surplus      blocksNBytes
737         short        blocksNBytes
738         unachievable blocksNBytes
739 }
740
741 type balancedBlockState struct {
742         desired      int
743         surplus      int
744         unachievable bool
745 }
746
747 func (bal *Balancer) collectStatistics(results <-chan balanceResult) {
748         var s balancerStats
749         s.replHistogram = make([]int, 2)
750         s.classStats = make(map[string]replicationStats, len(bal.classes))
751         for result := range results {
752                 surplus := result.have - result.want
753                 bytes := result.blkid.Size()
754
755                 for class, state := range result.classState {
756                         cs := s.classStats[class]
757                         if state.unachievable {
758                                 cs.unachievable.blocks++
759                                 cs.unachievable.bytes += bytes
760                         }
761                         if state.desired > 0 {
762                                 cs.desired.replicas += state.desired
763                                 cs.desired.blocks++
764                                 cs.desired.bytes += bytes * int64(state.desired)
765                         }
766                         if state.surplus > 0 {
767                                 cs.surplus.replicas += state.surplus
768                                 cs.surplus.blocks++
769                                 cs.surplus.bytes += bytes * int64(state.surplus)
770                         } else if state.surplus < 0 {
771                                 cs.short.replicas += -state.surplus
772                                 cs.short.blocks++
773                                 cs.short.bytes += bytes * int64(-state.surplus)
774                         }
775                         s.classStats[class] = cs
776                 }
777
778                 switch {
779                 case result.have == 0 && result.want > 0:
780                         s.lost.replicas -= surplus
781                         s.lost.blocks++
782                         s.lost.bytes += bytes * int64(-surplus)
783                 case surplus < 0:
784                         s.underrep.replicas -= surplus
785                         s.underrep.blocks++
786                         s.underrep.bytes += bytes * int64(-surplus)
787                 case surplus > 0 && result.want == 0:
788                         counter := &s.garbage
789                         for _, r := range result.blk.Replicas {
790                                 if r.Mtime >= bal.MinMtime {
791                                         counter = &s.unref
792                                         break
793                                 }
794                         }
795                         counter.replicas += surplus
796                         counter.blocks++
797                         counter.bytes += bytes * int64(surplus)
798                 case surplus > 0:
799                         s.overrep.replicas += surplus
800                         s.overrep.blocks++
801                         s.overrep.bytes += bytes * int64(len(result.blk.Replicas)-result.want)
802                 default:
803                         s.justright.replicas += result.want
804                         s.justright.blocks++
805                         s.justright.bytes += bytes * int64(result.want)
806                 }
807
808                 if result.want > 0 {
809                         s.desired.replicas += result.want
810                         s.desired.blocks++
811                         s.desired.bytes += bytes * int64(result.want)
812                 }
813                 if len(result.blk.Replicas) > 0 {
814                         s.current.replicas += len(result.blk.Replicas)
815                         s.current.blocks++
816                         s.current.bytes += bytes * int64(len(result.blk.Replicas))
817                 }
818
819                 for len(s.replHistogram) <= len(result.blk.Replicas) {
820                         s.replHistogram = append(s.replHistogram, 0)
821                 }
822                 s.replHistogram[len(result.blk.Replicas)]++
823         }
824         for _, srv := range bal.KeepServices {
825                 s.pulls += len(srv.ChangeSet.Pulls)
826                 s.trashes += len(srv.ChangeSet.Trashes)
827         }
828         bal.stats = s
829 }
830
831 // PrintStatistics writes statistics about the computed changes to
832 // bal.Logger. It should not be called until ComputeChangeSets has
833 // finished.
834 func (bal *Balancer) PrintStatistics() {
835         bal.logf("===")
836         bal.logf("%s lost (0=have<want)", bal.stats.lost)
837         bal.logf("%s underreplicated (0<have<want)", bal.stats.underrep)
838         bal.logf("%s just right (have=want)", bal.stats.justright)
839         bal.logf("%s overreplicated (have>want>0)", bal.stats.overrep)
840         bal.logf("%s unreferenced (have>want=0, new)", bal.stats.unref)
841         bal.logf("%s garbage (have>want=0, old)", bal.stats.garbage)
842         for _, class := range bal.classes {
843                 cs := bal.stats.classStats[class]
844                 bal.logf("===")
845                 bal.logf("storage class %q: %s desired", class, cs.desired)
846                 bal.logf("storage class %q: %s short", class, cs.short)
847                 bal.logf("storage class %q: %s surplus", class, cs.surplus)
848                 bal.logf("storage class %q: %s unachievable", class, cs.unachievable)
849         }
850         bal.logf("===")
851         bal.logf("%s total commitment (excluding unreferenced)", bal.stats.desired)
852         bal.logf("%s total usage", bal.stats.current)
853         bal.logf("===")
854         for _, srv := range bal.KeepServices {
855                 bal.logf("%s: %v\n", srv, srv.ChangeSet)
856         }
857         bal.logf("===")
858         bal.printHistogram(60)
859         bal.logf("===")
860 }
861
862 func (bal *Balancer) printHistogram(hashColumns int) {
863         bal.logf("Replication level distribution (counting N replicas on a single server as N):")
864         maxCount := 0
865         for _, count := range bal.stats.replHistogram {
866                 if maxCount < count {
867                         maxCount = count
868                 }
869         }
870         hashes := strings.Repeat("#", hashColumns)
871         countWidth := 1 + int(math.Log10(float64(maxCount+1)))
872         scaleCount := 10 * float64(hashColumns) / math.Floor(1+10*math.Log10(float64(maxCount+1)))
873         for repl, count := range bal.stats.replHistogram {
874                 nHashes := int(scaleCount * math.Log10(float64(count+1)))
875                 bal.logf("%2d: %*d %s", repl, countWidth, count, hashes[:nHashes])
876         }
877 }
878
879 // CheckSanityLate checks for configuration and runtime errors after
880 // GetCurrentState() and ComputeChangeSets() have finished.
881 //
882 // If it returns an error, it is dangerous to run any Commit methods.
883 func (bal *Balancer) CheckSanityLate() error {
884         if bal.errors != nil {
885                 for _, err := range bal.errors {
886                         bal.logf("deferred error: %v", err)
887                 }
888                 return fmt.Errorf("cannot proceed safely after deferred errors")
889         }
890
891         if bal.collScanned == 0 {
892                 return fmt.Errorf("received zero collections")
893         }
894
895         anyDesired := false
896         bal.BlockStateMap.Apply(func(_ arvados.SizedDigest, blk *BlockState) {
897                 for _, desired := range blk.Desired {
898                         if desired > 0 {
899                                 anyDesired = true
900                                 break
901                         }
902                 }
903         })
904         if !anyDesired {
905                 return fmt.Errorf("zero blocks have desired replication>0")
906         }
907
908         if dr := bal.DefaultReplication; dr < 1 {
909                 return fmt.Errorf("Default replication (%d) is less than 1", dr)
910         }
911
912         // TODO: no two services have identical indexes
913         // TODO: no collisions (same md5, different size)
914         return nil
915 }
916
917 // CommitPulls sends the computed lists of pull requests to the
918 // keepstore servers. This has the effect of increasing replication of
919 // existing blocks that are either underreplicated or poorly
920 // distributed according to rendezvous hashing.
921 func (bal *Balancer) CommitPulls(c *arvados.Client) error {
922         return bal.commitAsync(c, "send pull list",
923                 func(srv *KeepService) error {
924                         return srv.CommitPulls(c)
925                 })
926 }
927
928 // CommitTrash sends the computed lists of trash requests to the
929 // keepstore servers. This has the effect of deleting blocks that are
930 // overreplicated or unreferenced.
931 func (bal *Balancer) CommitTrash(c *arvados.Client) error {
932         return bal.commitAsync(c, "send trash list",
933                 func(srv *KeepService) error {
934                         return srv.CommitTrash(c)
935                 })
936 }
937
938 func (bal *Balancer) commitAsync(c *arvados.Client, label string, f func(srv *KeepService) error) error {
939         errs := make(chan error)
940         for _, srv := range bal.KeepServices {
941                 go func(srv *KeepService) {
942                         var err error
943                         defer func() { errs <- err }()
944                         label := fmt.Sprintf("%s: %v", srv, label)
945                         defer timeMe(bal.Logger, label)()
946                         err = f(srv)
947                         if err != nil {
948                                 err = fmt.Errorf("%s: %v", label, err)
949                         }
950                 }(srv)
951         }
952         var lastErr error
953         for range bal.KeepServices {
954                 if err := <-errs; err != nil {
955                         bal.logf("%v", err)
956                         lastErr = err
957                 }
958         }
959         close(errs)
960         return lastErr
961 }
962
963 func (bal *Balancer) logf(f string, args ...interface{}) {
964         if bal.Logger != nil {
965                 bal.Logger.Printf(f, args...)
966         }
967 }
968
969 // Rendezvous hash sort function. Less efficient than sorting on
970 // precomputed rendezvous hashes, but also rarely used.
971 func rendezvousLess(i, j string, blkid arvados.SizedDigest) bool {
972         a := md5.Sum([]byte(string(blkid[:32]) + i))
973         b := md5.Sum([]byte(string(blkid[:32]) + j))
974         return bytes.Compare(a[:], b[:]) < 0
975 }