1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
20 "git.curoverse.com/arvados.git/sdk/go/arvados"
21 "git.curoverse.com/arvados.git/sdk/go/keepclient"
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")
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")
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.
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 {
47 KeepServices map[string]*KeepService
48 DefaultReplication int
55 mountsByClass map[string]map[*KeepMount]bool
57 serviceRoots map[string]string
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.
67 // Run should only be called once on a given Balancer object.
71 // runOptions, err = (&Balancer{}).Run(config, runOptions)
72 func (bal *Balancer) Run(config Config, runOptions RunOptions) (nextRunOptions RunOptions, err error) {
73 nextRunOptions = runOptions
75 bal.Dumper = runOptions.Dumper
76 bal.Logger = runOptions.Logger
77 if bal.Logger == nil {
78 bal.Logger = log.New(os.Stderr, "", log.LstdFlags)
81 defer timeMe(bal.Logger, "Run")()
83 if len(config.KeepServiceList.Items) > 0 {
84 err = bal.SetKeepServices(config.KeepServiceList)
86 err = bal.DiscoverKeepServices(&config.Client, config.KeepServiceTypes)
92 for _, srv := range bal.KeepServices {
93 err = srv.discoverMounts(&config.Client)
100 if err = bal.CheckSanityEarly(&config.Client); err != nil {
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")
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 {
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
118 if err = bal.GetCurrentState(&config.Client, config.CollectionBatchSize, config.CollectionBuffers); err != nil {
121 bal.ComputeChangeSets()
122 bal.PrintStatistics()
123 if err = bal.CheckSanityLate(); err != nil {
126 if runOptions.CommitPulls {
127 err = bal.CommitPulls(&config.Client)
129 // Skip trash if we can't pull. (Too cautious?)
133 if runOptions.CommitTrash {
134 err = bal.CommitTrash(&config.Client)
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{
145 ChangeSet: &ChangeSet{},
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 {
160 return c.EachKeepService(func(srv arvados.KeepService) error {
161 if ok[srv.ServiceType] {
162 bal.KeepServices[srv.UUID] = &KeepService{
164 ChangeSet: &ChangeSet{},
167 bal.logf("skipping %v with service type %q", srv.UUID, srv.ServiceType)
173 func (bal *Balancer) cleanupMounts() {
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
182 // Drop the readonly mounts whose device is mounted RW
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])
190 dedup = append(dedup, mnt)
195 for _, srv := range bal.KeepServices {
196 for _, mnt := range srv.mounts {
197 if mnt.Replication <= 0 {
198 log.Printf("%s: mount %s reports replication=%d, using replication=1", srv, mnt.UUID, mnt.Replication)
205 // CheckSanityEarly checks for configuration and runtime errors that
206 // can be detected before GetCurrentState() and ComputeChangeSets()
209 // If it returns an error, it is pointless to run GetCurrentState or
210 // ComputeChangeSets: after doing so, the statistics would be
211 // meaningless and it would be dangerous to run any Commit methods.
212 func (bal *Balancer) CheckSanityEarly(c *arvados.Client) error {
213 u, err := c.CurrentUser()
215 return fmt.Errorf("CurrentUser(): %v", err)
217 if !u.IsActive || !u.IsAdmin {
218 return fmt.Errorf("current user (%s) is not an active admin user", u.UUID)
220 for _, srv := range bal.KeepServices {
221 if srv.ServiceType == "proxy" {
222 return fmt.Errorf("config error: %s: proxy servers cannot be balanced", srv)
228 // rendezvousState returns a fingerprint (e.g., a sorted list of
229 // UUID+host+port) of the current set of keep services.
230 func (bal *Balancer) rendezvousState() string {
231 srvs := make([]string, 0, len(bal.KeepServices))
232 for _, srv := range bal.KeepServices {
233 srvs = append(srvs, srv.String())
236 return strings.Join(srvs, "; ")
239 // ClearTrashLists sends an empty trash list to each keep
240 // service. Calling this before GetCurrentState avoids races.
242 // When a block appears in an index, we assume that replica will still
243 // exist after we delete other replicas on other servers. However,
244 // it's possible that a previous rebalancing operation made different
245 // decisions (e.g., servers were added/removed, and rendezvous order
246 // changed). In this case, the replica might already be on that
247 // server's trash list, and it might be deleted before we send a
248 // replacement trash list.
250 // We avoid this problem if we clear all trash lists before getting
251 // indexes. (We also assume there is only one rebalancing process
252 // running at a time.)
253 func (bal *Balancer) ClearTrashLists(c *arvados.Client) error {
254 for _, srv := range bal.KeepServices {
255 srv.ChangeSet = &ChangeSet{}
257 return bal.CommitTrash(c)
260 // GetCurrentState determines the current replication state, and the
261 // desired replication level, for every block that is either
262 // retrievable or referenced.
264 // It determines the current replication state by reading the block index
265 // from every known Keep service.
267 // It determines the desired replication level by retrieving all
268 // collection manifests in the database (API server).
270 // It encodes the resulting information in BlockStateMap.
271 func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) error {
272 defer timeMe(bal.Logger, "GetCurrentState")()
273 bal.BlockStateMap = NewBlockStateMap()
275 dd, err := c.DiscoveryDocument()
279 bal.DefaultReplication = dd.DefaultCollectionReplication
280 bal.MinMtime = time.Now().UnixNano() - dd.BlobSignatureTTL*1e9
282 errs := make(chan error, 2+len(bal.KeepServices))
283 wg := sync.WaitGroup{}
285 // When a device is mounted more than once, we will get its
286 // index only once, and call AddReplicas on all of the mounts.
287 // equivMount keys are the mounts that will be indexed, and
288 // each value is a list of mounts to apply the received index
290 equivMount := map[*KeepMount][]*KeepMount{}
291 // deviceMount maps each device ID to the one mount that will
292 // be indexed for that device.
293 deviceMount := map[string]*KeepMount{}
294 for _, srv := range bal.KeepServices {
295 for _, mnt := range srv.mounts {
296 equiv := deviceMount[mnt.DeviceID]
299 if mnt.DeviceID != "" {
300 deviceMount[mnt.DeviceID] = equiv
303 equivMount[equiv] = append(equivMount[equiv], mnt)
307 // Start one goroutine for each (non-redundant) mount:
308 // retrieve the index, and add the returned blocks to
310 for _, mounts := range equivMount {
312 go func(mounts []*KeepMount) {
314 bal.logf("mount %s: retrieve index from %s", mounts[0], mounts[0].KeepService)
315 idx, err := mounts[0].KeepService.IndexMount(c, mounts[0].UUID, "")
317 errs <- fmt.Errorf("%s: retrieve index: %v", mounts[0], err)
321 // Some other goroutine encountered an
322 // error -- any further effort here
326 for _, mount := range mounts {
327 bal.logf("%s: add %d replicas to map", mount, len(idx))
328 bal.BlockStateMap.AddReplicas(mount, idx)
329 bal.logf("%s: added %d replicas", mount, len(idx))
331 bal.logf("mount %s: index done", mounts[0])
335 // collQ buffers incoming collections so we can start fetching
336 // the next page without waiting for the current page to
337 // finish processing.
338 collQ := make(chan arvados.Collection, bufs)
340 // Start a goroutine to process collections. (We could use a
341 // worker pool here, but even with a single worker we already
342 // process collections much faster than we can retrieve them.)
346 for coll := range collQ {
347 err := bal.addCollection(coll)
358 // Start a goroutine to retrieve all collections from the
359 // Arvados database and send them to collQ for processing.
363 err = EachCollection(c, pageSize,
364 func(coll arvados.Collection) error {
367 // some other GetCurrentState
368 // error happened: no point
371 return fmt.Errorf("")
374 }, func(done, total int) {
375 bal.logf("collections: %d/%d", done, total)
390 func (bal *Balancer) addCollection(coll arvados.Collection) error {
391 blkids, err := coll.SizedDigests()
394 bal.errors = append(bal.errors, fmt.Errorf("%v: %v", coll.UUID, err))
398 repl := bal.DefaultReplication
399 if coll.ReplicationDesired != nil {
400 repl = *coll.ReplicationDesired
402 debugf("%v: %d block x%d", coll.UUID, len(blkids), repl)
403 bal.BlockStateMap.IncreaseDesired(coll.StorageClassesDesired, repl, blkids)
407 // ComputeChangeSets compares, for each known block, the current and
408 // desired replication states. If it is possible to get closer to the
409 // desired state by copying or deleting blocks, it adds those changes
410 // to the relevant KeepServices' ChangeSets.
412 // It does not actually apply any of the computed changes.
413 func (bal *Balancer) ComputeChangeSets() {
414 // This just calls balanceBlock() once for each block, using a
415 // pool of worker goroutines.
416 defer timeMe(bal.Logger, "ComputeChangeSets")()
417 bal.setupLookupTables()
419 type balanceTask struct {
420 blkid arvados.SizedDigest
423 workers := runtime.GOMAXPROCS(-1)
424 todo := make(chan balanceTask, workers)
426 bal.BlockStateMap.Apply(func(blkid arvados.SizedDigest, blk *BlockState) {
434 results := make(chan balanceResult, workers)
436 var wg sync.WaitGroup
437 for i := 0; i < workers; i++ {
440 for work := range todo {
441 results <- bal.balanceBlock(work.blkid, work.blk)
449 bal.collectStatistics(results)
452 func (bal *Balancer) setupLookupTables() {
453 bal.serviceRoots = make(map[string]string)
454 bal.classes = []string{"default"}
455 bal.mountsByClass = map[string]map[*KeepMount]bool{"default": {}}
457 for _, srv := range bal.KeepServices {
458 bal.serviceRoots[srv.UUID] = srv.UUID
459 for _, mnt := range srv.mounts {
462 // All mounts on a read-only service are
463 // effectively read-only.
464 mnt.ReadOnly = mnt.ReadOnly || srv.ReadOnly
466 if len(mnt.StorageClasses) == 0 {
467 bal.mountsByClass["default"][mnt] = true
470 for _, class := range mnt.StorageClasses {
471 if mbc := bal.mountsByClass[class]; mbc == nil {
472 bal.classes = append(bal.classes, class)
473 bal.mountsByClass[class] = map[*KeepMount]bool{mnt: true}
480 // Consider classes in lexicographic order to avoid flapping
481 // between balancing runs. The outcome of the "prefer a mount
482 // we're already planning to use for a different storage
483 // class" case in balanceBlock depends on the order classes
485 sort.Strings(bal.classes)
495 var changeName = map[int]string{
498 changeTrash: "trash",
502 type balanceResult struct {
504 blkid arvados.SizedDigest
507 classState map[string]balancedBlockState
510 // balanceBlock compares current state to desired state for a single
511 // block, and makes the appropriate ChangeSet calls.
512 func (bal *Balancer) balanceBlock(blkid arvados.SizedDigest, blk *BlockState) balanceResult {
513 debugf("balanceBlock: %v %+v", blkid, blk)
516 mnt *KeepMount // never nil
517 repl *Replica // replica already stored here (or nil)
518 want bool // we should pull/leave a replica here
521 // Build a list of all slots (one per mounted volume).
522 slots := make([]slot, 0, bal.mounts)
523 for _, srv := range bal.KeepServices {
524 for _, mnt := range srv.mounts {
526 for r := range blk.Replicas {
527 if blk.Replicas[r].KeepMount == mnt {
528 repl = &blk.Replicas[r]
531 // Initial value of "want" is "have, and can't
532 // delete". These untrashable replicas get
533 // prioritized when sorting slots: otherwise,
534 // non-optimal readonly copies would cause us
536 slots = append(slots, slot{
539 want: repl != nil && (mnt.ReadOnly || repl.Mtime >= bal.MinMtime),
544 uuids := keepclient.NewRootSorter(bal.serviceRoots, string(blkid[:32])).GetSortedRoots()
545 srvRendezvous := make(map[*KeepService]int, len(uuids))
546 for i, uuid := range uuids {
547 srv := bal.KeepServices[uuid]
548 srvRendezvous[srv] = i
551 // Below we set underreplicated=true if we find any storage
552 // class that's currently underreplicated -- in that case we
553 // won't want to trash any replicas.
554 underreplicated := false
556 classState := make(map[string]balancedBlockState, len(bal.classes))
557 unsafeToDelete := make(map[int64]bool, len(slots))
558 for _, class := range bal.classes {
559 desired := blk.Desired[class]
561 countedDev := map[string]bool{}
563 for _, slot := range slots {
564 if slot.repl != nil && bal.mountsByClass[class][slot.mnt] && !countedDev[slot.mnt.DeviceID] {
565 have += slot.mnt.Replication
566 if slot.mnt.DeviceID != "" {
567 countedDev[slot.mnt.DeviceID] = true
571 classState[class] = balancedBlockState{
573 surplus: have - desired,
580 // Sort the slots by desirability.
581 sort.Slice(slots, func(i, j int) bool {
582 si, sj := slots[i], slots[j]
583 if classi, classj := bal.mountsByClass[class][si.mnt], bal.mountsByClass[class][sj.mnt]; classi != classj {
584 // Prefer a mount that satisfies the
586 return bal.mountsByClass[class][si.mnt]
587 } else if wanti, wantj := si.want, si.want; wanti != wantj {
588 // Prefer a mount that will have a
589 // replica no matter what we do here
590 // -- either because it already has an
591 // untrashable replica, or because we
592 // already need it to satisfy a
593 // different storage class.
595 } else if orderi, orderj := srvRendezvous[si.mnt.KeepService], srvRendezvous[sj.mnt.KeepService]; orderi != orderj {
596 // Prefer a better rendezvous
598 return orderi < orderj
599 } else if repli, replj := si.repl != nil, sj.repl != nil; repli != replj {
600 // Prefer a mount that already has a
604 // If pull/trash turns out to be
605 // needed, distribute the
606 // new/remaining replicas uniformly
607 // across qualifying mounts on a given
609 return rendezvousLess(si.mnt.DeviceID, sj.mnt.DeviceID, blkid)
613 // Servers/mounts/devices (with or without existing
614 // replicas) that are part of the best achievable
615 // layout for this storage class.
616 wantSrv := map[*KeepService]bool{}
617 wantMnt := map[*KeepMount]bool{}
618 wantDev := map[string]bool{}
619 // Positions (with existing replicas) that have been
620 // protected (via unsafeToDelete) to ensure we don't
621 // reduce replication below desired level when
622 // trashing replicas that aren't optimal positions for
623 // any storage class.
624 protMnt := map[*KeepMount]bool{}
625 // Replication planned so far (corresponds to wantMnt).
627 // Protected replication (corresponds to protMnt).
630 // trySlot tries using a slot to meet requirements,
631 // and returns true if all requirements are met.
632 trySlot := func(i int) bool {
634 if wantMnt[slot.mnt] || wantDev[slot.mnt.DeviceID] {
635 // Already allocated a replica to this
636 // backend device, possibly on a
640 if replProt < desired && slot.repl != nil && !protMnt[slot.mnt] {
641 unsafeToDelete[slot.repl.Mtime] = true
642 protMnt[slot.mnt] = true
643 replProt += slot.mnt.Replication
645 if replWant < desired && (slot.repl != nil || !slot.mnt.ReadOnly) {
647 wantSrv[slot.mnt.KeepService] = true
648 wantMnt[slot.mnt] = true
649 if slot.mnt.DeviceID != "" {
650 wantDev[slot.mnt.DeviceID] = true
652 replWant += slot.mnt.Replication
654 return replProt >= desired && replWant >= desired
657 // First try to achieve desired replication without
658 // using the same server twice.
660 for i := 0; i < len(slots) && !done; i++ {
661 if !wantSrv[slots[i].mnt.KeepService] {
666 // If that didn't suffice, do another pass without the
667 // "distinct services" restriction. (Achieving the
668 // desired volume replication on fewer than the
669 // desired number of services is better than
670 // underreplicating.)
671 for i := 0; i < len(slots) && !done; i++ {
675 if !underreplicated {
677 for _, slot := range slots {
678 if slot.repl == nil || !bal.mountsByClass[class][slot.mnt] {
681 if safe += slot.mnt.Replication; safe >= desired {
685 underreplicated = safe < desired
688 // set the unachievable flag if there aren't enough
689 // slots offering the relevant storage class. (This is
690 // as easy as checking slots[desired] because we
691 // already sorted the qualifying slots to the front.)
692 if desired >= len(slots) || !bal.mountsByClass[class][slots[desired].mnt] {
693 cs := classState[class]
694 cs.unachievable = true
695 classState[class] = cs
698 // Avoid deleting wanted replicas from devices that
699 // are mounted on multiple servers -- even if they
700 // haven't already been added to unsafeToDelete
701 // because the servers report different Mtimes.
702 for _, slot := range slots {
703 if slot.repl != nil && wantDev[slot.mnt.DeviceID] {
704 unsafeToDelete[slot.repl.Mtime] = true
709 // TODO: If multiple replicas are trashable, prefer the oldest
710 // replica that doesn't have a timestamp collision with
713 countedDev := map[string]bool{}
715 for _, slot := range slots {
716 if countedDev[slot.mnt.DeviceID] {
720 want += slot.mnt.Replication
722 if slot.repl != nil {
723 have += slot.mnt.Replication
725 if slot.mnt.DeviceID != "" {
726 countedDev[slot.mnt.DeviceID] = true
731 for _, slot := range slots {
732 // TODO: request a Touch if Mtime is duplicated.
735 case !underreplicated && slot.repl != nil && !slot.want && !unsafeToDelete[slot.repl.Mtime]:
736 slot.mnt.KeepService.AddTrash(Trash{
738 Mtime: slot.repl.Mtime,
742 case len(blk.Replicas) == 0:
744 case slot.repl == nil && slot.want && !slot.mnt.ReadOnly:
745 slot.mnt.KeepService.AddPull(Pull{
747 From: blk.Replicas[0].KeepMount.KeepService,
754 if bal.Dumper != nil {
756 if slot.repl != nil {
757 mtime = slot.repl.Mtime
759 srv := slot.mnt.KeepService
760 changes = append(changes, fmt.Sprintf("%s:%d/%s=%s,%d", srv.ServiceHost, srv.ServicePort, slot.mnt.UUID, changeName[change], mtime))
763 if bal.Dumper != nil {
764 bal.Dumper.Printf("%s have=%d want=%v %s", blkid, have, want, strings.Join(changes, " "))
766 return balanceResult{
771 classState: classState,
775 type blocksNBytes struct {
781 func (bb blocksNBytes) String() string {
782 return fmt.Sprintf("%d replicas (%d blocks, %d bytes)", bb.replicas, bb.blocks, bb.bytes)
785 type balancerStats struct {
790 underrep blocksNBytes
791 unachievable blocksNBytes
792 justright blocksNBytes
798 classStats map[string]replicationStats
801 type replicationStats struct {
805 unachievable blocksNBytes
808 type balancedBlockState struct {
814 func (bal *Balancer) collectStatistics(results <-chan balanceResult) {
816 s.replHistogram = make([]int, 2)
817 s.classStats = make(map[string]replicationStats, len(bal.classes))
818 for result := range results {
819 surplus := result.have - result.want
820 bytes := result.blkid.Size()
822 for class, state := range result.classState {
823 cs := s.classStats[class]
824 if state.unachievable {
825 cs.unachievable.blocks++
826 cs.unachievable.bytes += bytes
828 if state.desired > 0 {
829 cs.desired.replicas += state.desired
831 cs.desired.bytes += bytes * int64(state.desired)
833 if state.surplus > 0 {
834 cs.surplus.replicas += state.surplus
836 cs.surplus.bytes += bytes * int64(state.surplus)
837 } else if state.surplus < 0 {
838 cs.short.replicas += -state.surplus
840 cs.short.bytes += bytes * int64(-state.surplus)
842 s.classStats[class] = cs
846 case result.have == 0 && result.want > 0:
847 s.lost.replicas -= surplus
849 s.lost.bytes += bytes * int64(-surplus)
851 s.underrep.replicas -= surplus
853 s.underrep.bytes += bytes * int64(-surplus)
854 case surplus > 0 && result.want == 0:
855 counter := &s.garbage
856 for _, r := range result.blk.Replicas {
857 if r.Mtime >= bal.MinMtime {
862 counter.replicas += surplus
864 counter.bytes += bytes * int64(surplus)
866 s.overrep.replicas += surplus
868 s.overrep.bytes += bytes * int64(result.have-result.want)
870 s.justright.replicas += result.want
872 s.justright.bytes += bytes * int64(result.want)
876 s.desired.replicas += result.want
878 s.desired.bytes += bytes * int64(result.want)
881 s.current.replicas += result.have
883 s.current.bytes += bytes * int64(result.have)
886 for len(s.replHistogram) <= result.have {
887 s.replHistogram = append(s.replHistogram, 0)
889 s.replHistogram[result.have]++
891 for _, srv := range bal.KeepServices {
892 s.pulls += len(srv.ChangeSet.Pulls)
893 s.trashes += len(srv.ChangeSet.Trashes)
898 // PrintStatistics writes statistics about the computed changes to
899 // bal.Logger. It should not be called until ComputeChangeSets has
901 func (bal *Balancer) PrintStatistics() {
903 bal.logf("%s lost (0=have<want)", bal.stats.lost)
904 bal.logf("%s underreplicated (0<have<want)", bal.stats.underrep)
905 bal.logf("%s just right (have=want)", bal.stats.justright)
906 bal.logf("%s overreplicated (have>want>0)", bal.stats.overrep)
907 bal.logf("%s unreferenced (have>want=0, new)", bal.stats.unref)
908 bal.logf("%s garbage (have>want=0, old)", bal.stats.garbage)
909 for _, class := range bal.classes {
910 cs := bal.stats.classStats[class]
912 bal.logf("storage class %q: %s desired", class, cs.desired)
913 bal.logf("storage class %q: %s short", class, cs.short)
914 bal.logf("storage class %q: %s surplus", class, cs.surplus)
915 bal.logf("storage class %q: %s unachievable", class, cs.unachievable)
918 bal.logf("%s total commitment (excluding unreferenced)", bal.stats.desired)
919 bal.logf("%s total usage", bal.stats.current)
921 for _, srv := range bal.KeepServices {
922 bal.logf("%s: %v\n", srv, srv.ChangeSet)
925 bal.printHistogram(60)
929 func (bal *Balancer) printHistogram(hashColumns int) {
930 bal.logf("Replication level distribution (counting N replicas on a single server as N):")
932 for _, count := range bal.stats.replHistogram {
933 if maxCount < count {
937 hashes := strings.Repeat("#", hashColumns)
938 countWidth := 1 + int(math.Log10(float64(maxCount+1)))
939 scaleCount := 10 * float64(hashColumns) / math.Floor(1+10*math.Log10(float64(maxCount+1)))
940 for repl, count := range bal.stats.replHistogram {
941 nHashes := int(scaleCount * math.Log10(float64(count+1)))
942 bal.logf("%2d: %*d %s", repl, countWidth, count, hashes[:nHashes])
946 // CheckSanityLate checks for configuration and runtime errors after
947 // GetCurrentState() and ComputeChangeSets() have finished.
949 // If it returns an error, it is dangerous to run any Commit methods.
950 func (bal *Balancer) CheckSanityLate() error {
951 if bal.errors != nil {
952 for _, err := range bal.errors {
953 bal.logf("deferred error: %v", err)
955 return fmt.Errorf("cannot proceed safely after deferred errors")
958 if bal.collScanned == 0 {
959 return fmt.Errorf("received zero collections")
963 bal.BlockStateMap.Apply(func(_ arvados.SizedDigest, blk *BlockState) {
964 for _, desired := range blk.Desired {
972 return fmt.Errorf("zero blocks have desired replication>0")
975 if dr := bal.DefaultReplication; dr < 1 {
976 return fmt.Errorf("Default replication (%d) is less than 1", dr)
979 // TODO: no two services have identical indexes
980 // TODO: no collisions (same md5, different size)
984 // CommitPulls sends the computed lists of pull requests to the
985 // keepstore servers. This has the effect of increasing replication of
986 // existing blocks that are either underreplicated or poorly
987 // distributed according to rendezvous hashing.
988 func (bal *Balancer) CommitPulls(c *arvados.Client) error {
989 return bal.commitAsync(c, "send pull list",
990 func(srv *KeepService) error {
991 return srv.CommitPulls(c)
995 // CommitTrash sends the computed lists of trash requests to the
996 // keepstore servers. This has the effect of deleting blocks that are
997 // overreplicated or unreferenced.
998 func (bal *Balancer) CommitTrash(c *arvados.Client) error {
999 return bal.commitAsync(c, "send trash list",
1000 func(srv *KeepService) error {
1001 return srv.CommitTrash(c)
1005 func (bal *Balancer) commitAsync(c *arvados.Client, label string, f func(srv *KeepService) error) error {
1006 errs := make(chan error)
1007 for _, srv := range bal.KeepServices {
1008 go func(srv *KeepService) {
1010 defer func() { errs <- err }()
1011 label := fmt.Sprintf("%s: %v", srv, label)
1012 defer timeMe(bal.Logger, label)()
1015 err = fmt.Errorf("%s: %v", label, err)
1020 for range bal.KeepServices {
1021 if err := <-errs; err != nil {
1030 func (bal *Balancer) logf(f string, args ...interface{}) {
1031 if bal.Logger != nil {
1032 bal.Logger.Printf(f, args...)
1036 // Rendezvous hash sort function. Less efficient than sorting on
1037 // precomputed rendezvous hashes, but also rarely used.
1038 func rendezvousLess(i, j string, blkid arvados.SizedDigest) bool {
1039 a := md5.Sum([]byte(string(blkid[:32]) + i))
1040 b := md5.Sum([]byte(string(blkid[:32]) + j))
1041 return bytes.Compare(a[:], b[:]) < 0