services/keep-balance/balance.go

   1 package main
   2
   3 import (
   4         "fmt"
   5         "log"
   6         "math"
   7         "os"
   8         "runtime"
   9         "sort"
  10         "strings"
  11         "sync"
  12         "time"
  13
  14         "git.curoverse.com/arvados.git/sdk/go/arvados"
  15         "git.curoverse.com/arvados.git/sdk/go/keepclient"
  16 )
  17
  18 // CheckConfig returns an error if anything is wrong with the given
  19 // config and runOptions.
  20 func CheckConfig(config Config, runOptions RunOptions) error {
  21         if len(config.KeepServiceList.Items) > 0 && config.KeepServiceTypes != nil {
  22                 return fmt.Errorf("cannot specify both KeepServiceList and KeepServiceTypes in config")
  23         }
  24         if !runOptions.Once && config.RunPeriod == arvados.Duration(0) {
  25                 return fmt.Errorf("you must either use the -once flag, or specify RunPeriod in config")
  26         }
  27         return nil
  28 }
  29
  30 // Balancer compares the contents of keepstore servers with the
  31 // collections stored in Arvados, and issues pull/trash requests
  32 // needed to get (closer to) the optimal data layout.
  33 //
  34 // In the optimal data layout: every data block referenced by a
  35 // collection is replicated at least as many times as desired by the
  36 // collection; there are no unreferenced data blocks older than
  37 // BlobSignatureTTL; and all N existing replicas of a given data block
  38 // are in the N best positions in rendezvous probe order.
  39 type Balancer struct {
  40         *BlockStateMap
  41         KeepServices       map[string]*KeepService
  42         DefaultReplication int
  43         Logger             *log.Logger
  44         Dumper             *log.Logger
  45         MinMtime           int64
  46
  47         collScanned  int
  48         serviceRoots map[string]string
  49         errors       []error
  50         mutex        sync.Mutex
  51 }
  52
  53 // Run performs a balance operation using the given config and
  54 // runOptions, and returns RunOptions suitable for passing to a
  55 // subsequent balance operation.
  56 //
  57 // Run should only be called once on a given Balancer object.
  58 //
  59 // Typical usage:
  60 //
  61 //   runOptions, err = (&Balancer{}).Run(config, runOptions)
  62 func (bal *Balancer) Run(config Config, runOptions RunOptions) (nextRunOptions RunOptions, err error) {
  63         nextRunOptions = runOptions
  64
  65         bal.Dumper = runOptions.Dumper
  66         bal.Logger = runOptions.Logger
  67         if bal.Logger == nil {
  68                 bal.Logger = log.New(os.Stderr, "", log.LstdFlags)
  69         }
  70
  71         defer timeMe(bal.Logger, "Run")()
  72
  73         if len(config.KeepServiceList.Items) > 0 {
  74                 err = bal.SetKeepServices(config.KeepServiceList)
  75         } else {
  76                 err = bal.DiscoverKeepServices(&config.Client, config.KeepServiceTypes)
  77         }
  78         if err != nil {
  79                 return
  80         }
  81
  82         if err = bal.CheckSanityEarly(&config.Client); err != nil {
  83                 return
  84         }
  85         rs := bal.rendezvousState()
  86         if runOptions.CommitTrash && rs != runOptions.SafeRendezvousState {
  87                 if runOptions.SafeRendezvousState != "" {
  88                         bal.logf("notice: KeepServices list has changed since last run")
  89                 }
  90                 bal.logf("clearing existing trash lists, in case the new rendezvous order differs from previous run")
  91                 if err = bal.ClearTrashLists(&config.Client); err != nil {
  92                         return
  93                 }
  94                 // The current rendezvous state becomes "safe" (i.e.,
  95                 // OK to compute changes for that state without
  96                 // clearing existing trash lists) only now, after we
  97                 // succeed in clearing existing trash lists.
  98                 nextRunOptions.SafeRendezvousState = rs
  99         }
 100         if err = bal.GetCurrentState(&config.Client, config.CollectionBatchSize, config.CollectionBuffers); err != nil {
 101                 return
 102         }
 103         bal.ComputeChangeSets()
 104         bal.PrintStatistics()
 105         if err = bal.CheckSanityLate(); err != nil {
 106                 return
 107         }
 108         if runOptions.CommitPulls {
 109                 err = bal.CommitPulls(&config.Client)
 110                 if err != nil {
 111                         // Skip trash if we can't pull. (Too cautious?)
 112                         return
 113                 }
 114         }
 115         if runOptions.CommitTrash {
 116                 err = bal.CommitTrash(&config.Client)
 117         }
 118         return
 119 }
 120
 121 // SetKeepServices sets the list of KeepServices to operate on.
 122 func (bal *Balancer) SetKeepServices(srvList arvados.KeepServiceList) error {
 123         bal.KeepServices = make(map[string]*KeepService)
 124         for _, srv := range srvList.Items {
 125                 bal.KeepServices[srv.UUID] = &KeepService{
 126                         KeepService: srv,
 127                         ChangeSet:   &ChangeSet{},
 128                 }
 129         }
 130         return nil
 131 }
 132
 133 // DiscoverKeepServices sets the list of KeepServices by calling the
 134 // API to get a list of all services, and selecting the ones whose
 135 // ServiceType is in okTypes.
 136 func (bal *Balancer) DiscoverKeepServices(c *arvados.Client, okTypes []string) error {
 137         bal.KeepServices = make(map[string]*KeepService)
 138         ok := make(map[string]bool)
 139         for _, t := range okTypes {
 140                 ok[t] = true
 141         }
 142         return c.EachKeepService(func(srv arvados.KeepService) error {
 143                 if ok[srv.ServiceType] {
 144                         bal.KeepServices[srv.UUID] = &KeepService{
 145                                 KeepService: srv,
 146                                 ChangeSet:   &ChangeSet{},
 147                         }
 148                 } else {
 149                         bal.logf("skipping %v with service type %q", srv.UUID, srv.ServiceType)
 150                 }
 151                 return nil
 152         })
 153 }
 154
 155 // CheckSanityEarly checks for configuration and runtime errors that
 156 // can be detected before GetCurrentState() and ComputeChangeSets()
 157 // are called.
 158 //
 159 // If it returns an error, it is pointless to run GetCurrentState or
 160 // ComputeChangeSets: after doing so, the statistics would be
 161 // meaningless and it would be dangerous to run any Commit methods.
 162 func (bal *Balancer) CheckSanityEarly(c *arvados.Client) error {
 163         u, err := c.CurrentUser()
 164         if err != nil {
 165                 return fmt.Errorf("CurrentUser(): %v", err)
 166         }
 167         if !u.IsActive || !u.IsAdmin {
 168                 return fmt.Errorf("current user (%s) is not an active admin user", u.UUID)
 169         }
 170         for _, srv := range bal.KeepServices {
 171                 if srv.ServiceType == "proxy" {
 172                         return fmt.Errorf("config error: %s: proxy servers cannot be balanced", srv)
 173                 }
 174         }
 175         return nil
 176 }
 177
 178 // rendezvousState returns a fingerprint (e.g., a sorted list of
 179 // UUID+host+port) of the current set of keep services.
 180 func (bal *Balancer) rendezvousState() string {
 181         srvs := make([]string, 0, len(bal.KeepServices))
 182         for _, srv := range bal.KeepServices {
 183                 srvs = append(srvs, srv.String())
 184         }
 185         sort.Strings(srvs)
 186         return strings.Join(srvs, "; ")
 187 }
 188
 189 // ClearTrashLists sends an empty trash list to each keep
 190 // service. Calling this before GetCurrentState avoids races.
 191 //
 192 // When a block appears in an index, we assume that replica will still
 193 // exist after we delete other replicas on other servers. However,
 194 // it's possible that a previous rebalancing operation made different
 195 // decisions (e.g., servers were added/removed, and rendezvous order
 196 // changed). In this case, the replica might already be on that
 197 // server's trash list, and it might be deleted before we send a
 198 // replacement trash list.
 199 //
 200 // We avoid this problem if we clear all trash lists before getting
 201 // indexes. (We also assume there is only one rebalancing process
 202 // running at a time.)
 203 func (bal *Balancer) ClearTrashLists(c *arvados.Client) error {
 204         for _, srv := range bal.KeepServices {
 205                 srv.ChangeSet = &ChangeSet{}
 206         }
 207         return bal.CommitTrash(c)
 208 }
 209
 210 // GetCurrentState determines the current replication state, and the
 211 // desired replication level, for every block that is either
 212 // retrievable or referenced.
 213 //
 214 // It determines the current replication state by reading the block index
 215 // from every known Keep service.
 216 //
 217 // It determines the desired replication level by retrieving all
 218 // collection manifests in the database (API server).
 219 //
 220 // It encodes the resulting information in BlockStateMap.
 221 func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) error {
 222         defer timeMe(bal.Logger, "GetCurrentState")()
 223         bal.BlockStateMap = NewBlockStateMap()
 224
 225         dd, err := c.DiscoveryDocument()
 226         if err != nil {
 227                 return err
 228         }
 229         bal.DefaultReplication = dd.DefaultCollectionReplication
 230         bal.MinMtime = time.Now().UnixNano() - dd.BlobSignatureTTL*1e9
 231
 232         errs := make(chan error, 2+len(bal.KeepServices))
 233         wg := sync.WaitGroup{}
 234
 235         // Start one goroutine for each KeepService: retrieve the
 236         // index, and add the returned blocks to BlockStateMap.
 237         for _, srv := range bal.KeepServices {
 238                 wg.Add(1)
 239                 go func(srv *KeepService) {
 240                         defer wg.Done()
 241                         bal.logf("%s: retrieve index", srv)
 242                         idx, err := srv.Index(c, "")
 243                         if err != nil {
 244                                 errs <- fmt.Errorf("%s: %v", srv, err)
 245                                 return
 246                         }
 247                         bal.logf("%s: add %d replicas to map", srv, len(idx))
 248                         bal.BlockStateMap.AddReplicas(srv, idx)
 249                         bal.logf("%s: done", srv)
 250                 }(srv)
 251         }
 252
 253         // collQ buffers incoming collections so we can start fetching
 254         // the next page without waiting for the current page to
 255         // finish processing.
 256         collQ := make(chan arvados.Collection, bufs)
 257
 258         // Start a goroutine to process collections. (We could use a
 259         // worker pool here, but even with a single worker we already
 260         // process collections much faster than we can retrieve them.)
 261         wg.Add(1)
 262         go func() {
 263                 defer wg.Done()
 264                 for coll := range collQ {
 265                         err := bal.addCollection(coll)
 266                         if err != nil {
 267                                 errs <- err
 268                                 for range collQ {
 269                                 }
 270                                 return
 271                         }
 272                         bal.collScanned++
 273                 }
 274         }()
 275
 276         // Start a goroutine to retrieve all collections from the
 277         // Arvados database and send them to collQ for processing.
 278         wg.Add(1)
 279         go func() {
 280                 defer wg.Done()
 281                 err = EachCollection(c, pageSize,
 282                         func(coll arvados.Collection) error {
 283                                 collQ <- coll
 284                                 if len(errs) > 0 {
 285                                         // some other GetCurrentState
 286                                         // error happened: no point
 287                                         // getting any more
 288                                         // collections.
 289                                         return fmt.Errorf("")
 290                                 }
 291                                 return nil
 292                         }, func(done, total int) {
 293                                 bal.logf("collections: %d/%d", done, total)
 294                         })
 295                 close(collQ)
 296                 if err != nil {
 297                         errs <- err
 298                 }
 299         }()
 300
 301         go func() {
 302                 // Send a nil error when all goroutines finish. If
 303                 // this is the first error sent to errs, then
 304                 // everything worked.
 305                 wg.Wait()
 306                 errs <- nil
 307         }()
 308         return <-errs
 309 }
 310
 311 func (bal *Balancer) addCollection(coll arvados.Collection) error {
 312         blkids, err := coll.SizedDigests()
 313         if err != nil {
 314                 bal.mutex.Lock()
 315                 bal.errors = append(bal.errors, fmt.Errorf("%v: %v", coll.UUID, err))
 316                 bal.mutex.Unlock()
 317                 return nil
 318         }
 319         repl := bal.DefaultReplication
 320         if coll.ReplicationDesired != nil {
 321                 repl = *coll.ReplicationDesired
 322         }
 323         debugf("%v: %d block x%d", coll.UUID, len(blkids), repl)
 324         bal.BlockStateMap.IncreaseDesired(repl, blkids)
 325         return nil
 326 }
 327
 328 // ComputeChangeSets compares, for each known block, the current and
 329 // desired replication states. If it is possible to get closer to the
 330 // desired state by copying or deleting blocks, it adds those changes
 331 // to the relevant KeepServices' ChangeSets.
 332 //
 333 // It does not actually apply any of the computed changes.
 334 func (bal *Balancer) ComputeChangeSets() {
 335         // This just calls balanceBlock() once for each block, using a
 336         // pool of worker goroutines.
 337         defer timeMe(bal.Logger, "ComputeChangeSets")()
 338         bal.setupServiceRoots()
 339
 340         type balanceTask struct {
 341                 blkid arvados.SizedDigest
 342                 blk   *BlockState
 343         }
 344         nWorkers := 1 + runtime.NumCPU()
 345         todo := make(chan balanceTask, nWorkers)
 346         var wg sync.WaitGroup
 347         for i := 0; i < nWorkers; i++ {
 348                 wg.Add(1)
 349                 go func() {
 350                         for work := range todo {
 351                                 bal.balanceBlock(work.blkid, work.blk)
 352                         }
 353                         wg.Done()
 354                 }()
 355         }
 356         bal.BlockStateMap.Apply(func(blkid arvados.SizedDigest, blk *BlockState) {
 357                 todo <- balanceTask{
 358                         blkid: blkid,
 359                         blk:   blk,
 360                 }
 361         })
 362         close(todo)
 363         wg.Wait()
 364 }
 365
 366 func (bal *Balancer) setupServiceRoots() {
 367         bal.serviceRoots = make(map[string]string)
 368         for _, srv := range bal.KeepServices {
 369                 bal.serviceRoots[srv.UUID] = srv.UUID
 370         }
 371 }
 372
 373 const (
 374         changeStay = iota
 375         changePull
 376         changeTrash
 377         changeNone
 378 )
 379
 380 var changeName = map[int]string{
 381         changeStay:  "stay",
 382         changePull:  "pull",
 383         changeTrash: "trash",
 384         changeNone:  "none",
 385 }
 386
 387 // balanceBlock compares current state to desired state for a single
 388 // block, and makes the appropriate ChangeSet calls.
 389 func (bal *Balancer) balanceBlock(blkid arvados.SizedDigest, blk *BlockState) {
 390         debugf("balanceBlock: %v %+v", blkid, blk)
 391         uuids := keepclient.NewRootSorter(bal.serviceRoots, string(blkid[:32])).GetSortedRoots()
 392         hasRepl := make(map[string]Replica, len(bal.serviceRoots))
 393         for _, repl := range blk.Replicas {
 394                 hasRepl[repl.UUID] = repl
 395                 // TODO: when multiple copies are on one server, use
 396                 // the oldest one that doesn't have a timestamp
 397                 // collision with other replicas.
 398         }
 399         // number of replicas already found in positions better than
 400         // the position we're contemplating now.
 401         reportedBestRepl := 0
 402         // To be safe we assume two replicas with the same Mtime are
 403         // in fact the same replica being reported more than
 404         // once. len(uniqueBestRepl) is the number of distinct
 405         // replicas in the best rendezvous positions we've considered
 406         // so far.
 407         uniqueBestRepl := make(map[int64]bool, len(bal.serviceRoots))
 408         // pulls is the number of Pull changes we have already
 409         // requested. (For purposes of deciding whether to Pull to
 410         // rendezvous position N, we should assume all pulls we have
 411         // requested on rendezvous positions M<N will be successful.)
 412         pulls := 0
 413         var changes []string
 414         for _, uuid := range uuids {
 415                 change := changeNone
 416                 srv := bal.KeepServices[uuid]
 417                 // TODO: request a Touch if Mtime is duplicated.
 418                 repl, ok := hasRepl[srv.UUID]
 419                 if ok {
 420                         // This service has a replica. We should
 421                         // delete it if [1] we already have enough
 422                         // distinct replicas in better rendezvous
 423                         // positions and [2] this replica's Mtime is
 424                         // distinct from all of the better replicas'
 425                         // Mtimes.
 426                         if !srv.ReadOnly &&
 427                                 repl.Mtime < bal.MinMtime &&
 428                                 len(uniqueBestRepl) >= blk.Desired &&
 429                                 !uniqueBestRepl[repl.Mtime] {
 430                                 srv.AddTrash(Trash{
 431                                         SizedDigest: blkid,
 432                                         Mtime:       repl.Mtime,
 433                                 })
 434                                 change = changeTrash
 435                         } else {
 436                                 change = changeStay
 437                         }
 438                         uniqueBestRepl[repl.Mtime] = true
 439                         reportedBestRepl++
 440                 } else if pulls+reportedBestRepl < blk.Desired &&
 441                         len(blk.Replicas) > 0 &&
 442                         !srv.ReadOnly {
 443                         // This service doesn't have a replica. We
 444                         // should pull one to this server if we don't
 445                         // already have enough (existing+requested)
 446                         // replicas in better rendezvous positions.
 447                         srv.AddPull(Pull{
 448                                 SizedDigest: blkid,
 449                                 Source:      blk.Replicas[0].KeepService,
 450                         })
 451                         pulls++
 452                         change = changePull
 453                 }
 454                 if bal.Dumper != nil {
 455                         changes = append(changes, fmt.Sprintf("%s:%d=%s,%d", srv.ServiceHost, srv.ServicePort, changeName[change], repl.Mtime))
 456                 }
 457         }
 458         if bal.Dumper != nil {
 459                 bal.Dumper.Printf("%s have=%d want=%d %s", blkid, len(blk.Replicas), blk.Desired, strings.Join(changes, " "))
 460         }
 461 }
 462
 463 type blocksNBytes struct {
 464         replicas int
 465         blocks   int
 466         bytes    int64
 467 }
 468
 469 func (bb blocksNBytes) String() string {
 470         return fmt.Sprintf("%d replicas (%d blocks, %d bytes)", bb.replicas, bb.blocks, bb.bytes)
 471 }
 472
 473 type balancerStats struct {
 474         lost, overrep, unref, garbage, underrep, justright blocksNBytes
 475         desired, current                                   blocksNBytes
 476         pulls, trashes                                     int
 477         replHistogram                                      []int
 478 }
 479
 480 func (bal *Balancer) getStatistics() (s balancerStats) {
 481         s.replHistogram = make([]int, 2)
 482         bal.BlockStateMap.Apply(func(blkid arvados.SizedDigest, blk *BlockState) {
 483                 surplus := len(blk.Replicas) - blk.Desired
 484                 bytes := blkid.Size()
 485                 switch {
 486                 case len(blk.Replicas) == 0 && blk.Desired > 0:
 487                         s.lost.replicas -= surplus
 488                         s.lost.blocks++
 489                         s.lost.bytes += bytes * int64(-surplus)
 490                 case len(blk.Replicas) < blk.Desired:
 491                         s.underrep.replicas -= surplus
 492                         s.underrep.blocks++
 493                         s.underrep.bytes += bytes * int64(-surplus)
 494                 case len(blk.Replicas) > 0 && blk.Desired == 0:
 495                         counter := &s.garbage
 496                         for _, r := range blk.Replicas {
 497                                 if r.Mtime >= bal.MinMtime {
 498                                         counter = &s.unref
 499                                         break
 500                                 }
 501                         }
 502                         counter.replicas += surplus
 503                         counter.blocks++
 504                         counter.bytes += bytes * int64(surplus)
 505                 case len(blk.Replicas) > blk.Desired:
 506                         s.overrep.replicas += surplus
 507                         s.overrep.blocks++
 508                         s.overrep.bytes += bytes * int64(len(blk.Replicas)-blk.Desired)
 509                 default:
 510                         s.justright.replicas += blk.Desired
 511                         s.justright.blocks++
 512                         s.justright.bytes += bytes * int64(blk.Desired)
 513                 }
 514
 515                 if blk.Desired > 0 {
 516                         s.desired.replicas += blk.Desired
 517                         s.desired.blocks++
 518                         s.desired.bytes += bytes * int64(blk.Desired)
 519                 }
 520                 if len(blk.Replicas) > 0 {
 521                         s.current.replicas += len(blk.Replicas)
 522                         s.current.blocks++
 523                         s.current.bytes += bytes * int64(len(blk.Replicas))
 524                 }
 525
 526                 for len(s.replHistogram) <= len(blk.Replicas) {
 527                         s.replHistogram = append(s.replHistogram, 0)
 528                 }
 529                 s.replHistogram[len(blk.Replicas)]++
 530         })
 531         for _, srv := range bal.KeepServices {
 532                 s.pulls += len(srv.ChangeSet.Pulls)
 533                 s.trashes += len(srv.ChangeSet.Trashes)
 534         }
 535         return
 536 }
 537
 538 // PrintStatistics writes statistics about the computed changes to
 539 // bal.Logger. It should not be called until ComputeChangeSets has
 540 // finished.
 541 func (bal *Balancer) PrintStatistics() {
 542         s := bal.getStatistics()
 543         bal.logf("===")
 544         bal.logf("%s lost (0=have<want)", s.lost)
 545         bal.logf("%s underreplicated (0<have<want)", s.underrep)
 546         bal.logf("%s just right (have=want)", s.justright)
 547         bal.logf("%s overreplicated (have>want>0)", s.overrep)
 548         bal.logf("%s unreferenced (have>want=0, new)", s.unref)
 549         bal.logf("%s garbage (have>want=0, old)", s.garbage)
 550         bal.logf("===")
 551         bal.logf("%s total commitment (excluding unreferenced)", s.desired)
 552         bal.logf("%s total usage", s.current)
 553         bal.logf("===")
 554         for _, srv := range bal.KeepServices {
 555                 bal.logf("%s: %v\n", srv, srv.ChangeSet)
 556         }
 557         bal.logf("===")
 558         bal.printHistogram(s, 60)
 559         bal.logf("===")
 560 }
 561
 562 func (bal *Balancer) printHistogram(s balancerStats, hashColumns int) {
 563         bal.logf("Replication level distribution (counting N replicas on a single server as N):")
 564         maxCount := 0
 565         for _, count := range s.replHistogram {
 566                 if maxCount < count {
 567                         maxCount = count
 568                 }
 569         }
 570         hashes := strings.Repeat("#", hashColumns)
 571         countWidth := 1 + int(math.Log10(float64(maxCount+1)))
 572         scaleCount := 10 * float64(hashColumns) / math.Floor(1+10*math.Log10(float64(maxCount+1)))
 573         for repl, count := range s.replHistogram {
 574                 nHashes := int(scaleCount * math.Log10(float64(count+1)))
 575                 bal.logf("%2d: %*d %s", repl, countWidth, count, hashes[:nHashes])
 576         }
 577 }
 578
 579 // CheckSanityLate checks for configuration and runtime errors after
 580 // GetCurrentState() and ComputeChangeSets() have finished.
 581 //
 582 // If it returns an error, it is dangerous to run any Commit methods.
 583 func (bal *Balancer) CheckSanityLate() error {
 584         if bal.errors != nil {
 585                 for _, err := range bal.errors {
 586                         bal.logf("deferred error: %v", err)
 587                 }
 588                 return fmt.Errorf("cannot proceed safely after deferred errors")
 589         }
 590
 591         if bal.collScanned == 0 {
 592                 return fmt.Errorf("received zero collections")
 593         }
 594
 595         anyDesired := false
 596         bal.BlockStateMap.Apply(func(_ arvados.SizedDigest, blk *BlockState) {
 597                 if blk.Desired > 0 {
 598                         anyDesired = true
 599                 }
 600         })
 601         if !anyDesired {
 602                 return fmt.Errorf("zero blocks have desired replication>0")
 603         }
 604
 605         if dr := bal.DefaultReplication; dr < 1 {
 606                 return fmt.Errorf("Default replication (%d) is less than 1", dr)
 607         }
 608
 609         // TODO: no two services have identical indexes
 610         // TODO: no collisions (same md5, different size)
 611         return nil
 612 }
 613
 614 // CommitPulls sends the computed lists of pull requests to the
 615 // keepstore servers. This has the effect of increasing replication of
 616 // existing blocks that are either underreplicated or poorly
 617 // distributed according to rendezvous hashing.
 618 func (bal *Balancer) CommitPulls(c *arvados.Client) error {
 619         return bal.commitAsync(c, "send pull list",
 620                 func(srv *KeepService) error {
 621                         return srv.CommitPulls(c)
 622                 })
 623 }
 624
 625 // CommitTrash sends the computed lists of trash requests to the
 626 // keepstore servers. This has the effect of deleting blocks that are
 627 // overreplicated or unreferenced.
 628 func (bal *Balancer) CommitTrash(c *arvados.Client) error {
 629         return bal.commitAsync(c, "send trash list",
 630                 func(srv *KeepService) error {
 631                         return srv.CommitTrash(c)
 632                 })
 633 }
 634
 635 func (bal *Balancer) commitAsync(c *arvados.Client, label string, f func(srv *KeepService) error) error {
 636         errs := make(chan error)
 637         for _, srv := range bal.KeepServices {
 638                 go func(srv *KeepService) {
 639                         var err error
 640                         defer func() { errs <- err }()
 641                         label := fmt.Sprintf("%s: %v", srv, label)
 642                         defer timeMe(bal.Logger, label)()
 643                         err = f(srv)
 644                         if err != nil {
 645                                 err = fmt.Errorf("%s: %v", label, err)
 646                         }
 647                 }(srv)
 648         }
 649         var lastErr error
 650         for range bal.KeepServices {
 651                 if err := <-errs; err != nil {
 652                         bal.logf("%v", err)
 653                         lastErr = err
 654                 }
 655         }
 656         close(errs)
 657         return lastErr
 658 }
 659
 660 func (bal *Balancer) logf(f string, args ...interface{}) {
 661         if bal.Logger != nil {
 662                 bal.Logger.Printf(f, args...)
 663         }
 664 }