X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/dd1157d0a3492621fcc8f2c7696bbc166c604677..0210746ac1fef5a588279b208313ed06e51b19de:/services/keep-balance/balance.go

diff --git a/services/keep-balance/balance.go b/services/keep-balance/balance.go
index 7683835dd9..fd39ee693e 100644
--- a/services/keep-balance/balance.go
+++ b/services/keep-balance/balance.go
@@ -18,7 +18,7 @@ import (
 
 	"git.curoverse.com/arvados.git/sdk/go/arvados"
 	"git.curoverse.com/arvados.git/sdk/go/keepclient"
-	"github.com/Sirupsen/logrus"
+	"github.com/sirupsen/logrus"
 )
 
 // Balancer compares the contents of keepstore servers with the
@@ -31,8 +31,8 @@ import (
 // BlobSignatureTTL; and all N existing replicas of a given data block
 // are in the N best positions in rendezvous probe order.
 type Balancer struct {
-	Logger  *logrus.Logger
-	Dumper  *logrus.Logger
+	Logger  logrus.FieldLogger
+	Dumper  logrus.FieldLogger
 	Metrics *metrics
 
 	*BlockStateMap
@@ -206,6 +206,24 @@ func (bal *Balancer) CheckSanityEarly(c *arvados.Client) error {
 			return fmt.Errorf("config error: %s: proxy servers cannot be balanced", srv)
 		}
 	}
+
+	var checkPage arvados.CollectionList
+	if err = c.RequestAndDecode(&checkPage, "GET", "arvados/v1/collections", nil, arvados.ResourceListParams{
+		Limit:              new(int),
+		Count:              "exact",
+		IncludeTrash:       true,
+		IncludeOldVersions: true,
+		Filters: []arvados.Filter{{
+			Attr:     "modified_at",
+			Operator: "=",
+			Operand:  nil,
+		}},
+	}); err != nil {
+		return err
+	} else if n := checkPage.ItemsAvailable; n > 0 {
+		return fmt.Errorf("%d collections exist with null modified_at; cannot fetch reliably", n)
+	}
+
 	return nil
 }
 
@@ -263,7 +281,7 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) erro
 	bal.DefaultReplication = dd.DefaultCollectionReplication
 	bal.MinMtime = time.Now().UnixNano() - dd.BlobSignatureTTL*1e9
 
-	errs := make(chan error, 2+len(bal.KeepServices))
+	errs := make(chan error, 1)
 	wg := sync.WaitGroup{}
 
 	// When a device is mounted more than once, we will get its
@@ -298,7 +316,10 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) erro
 			bal.logf("mount %s: retrieve index from %s", mounts[0], mounts[0].KeepService)
 			idx, err := mounts[0].KeepService.IndexMount(c, mounts[0].UUID, "")
 			if err != nil {
-				errs <- fmt.Errorf("%s: retrieve index: %v", mounts[0], err)
+				select {
+				case errs <- fmt.Errorf("%s: retrieve index: %v", mounts[0], err):
+				default:
+				}
 				return
 			}
 			if len(errs) > 0 {
@@ -329,8 +350,11 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) erro
 		defer wg.Done()
 		for coll := range collQ {
 			err := bal.addCollection(coll)
-			if err != nil {
-				errs <- err
+			if err != nil || len(errs) > 0 {
+				select {
+				case errs <- err:
+				default:
+				}
 				for range collQ {
 				}
 				return
@@ -360,7 +384,10 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) erro
 			})
 		close(collQ)
 		if err != nil {
-			errs <- err
+			select {
+			case errs <- err:
+			default:
+			}
 		}
 	}()
 
@@ -374,10 +401,7 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) erro
 func (bal *Balancer) addCollection(coll arvados.Collection) error {
 	blkids, err := coll.SizedDigests()
 	if err != nil {
-		bal.mutex.Lock()
-		bal.errors = append(bal.errors, fmt.Errorf("%v: %v", coll.UUID, err))
-		bal.mutex.Unlock()
-		return nil
+		return fmt.Errorf("%v: %v", coll.UUID, err)
 	}
 	repl := bal.DefaultReplication
 	if coll.ReplicationDesired != nil {
@@ -435,7 +459,7 @@ func (bal *Balancer) ComputeChangeSets() {
 
 func (bal *Balancer) setupLookupTables() {
 	bal.serviceRoots = make(map[string]string)
-	bal.classes = []string{"default"}
+	bal.classes = defaultClasses
 	bal.mountsByClass = map[string]map[*KeepMount]bool{"default": {}}
 	bal.mounts = 0
 	for _, srv := range bal.KeepServices {
@@ -723,17 +747,17 @@ func (bal *Balancer) balanceBlock(blkid arvados.SizedDigest, blk *BlockState) ba
 				From:        slot.mnt,
 			})
 			change = changeTrash
-		case len(blk.Replicas) == 0:
-			change = changeNone
-		case slot.repl == nil && slot.want && !slot.mnt.ReadOnly:
+		case len(blk.Replicas) > 0 && slot.repl == nil && slot.want && !slot.mnt.ReadOnly:
 			slot.mnt.KeepService.AddPull(Pull{
 				SizedDigest: blkid,
 				From:        blk.Replicas[0].KeepMount.KeepService,
 				To:          slot.mnt,
 			})
 			change = changePull
-		default:
+		case slot.repl != nil:
 			change = changeStay
+		default:
+			change = changeNone
 		}
 		if bal.Dumper != nil {
 			var mtime int64
@@ -745,7 +769,7 @@ func (bal *Balancer) balanceBlock(blkid arvados.SizedDigest, blk *BlockState) ba
 		}
 	}
 	if bal.Dumper != nil {
-		bal.Dumper.Printf("%s have=%d want=%v %s", blkid, have, want, strings.Join(changes, " "))
+		bal.Dumper.Printf("%s refs=%d have=%d want=%v %v %v", blkid, blk.RefCount, have, want, blk.Desired, changes)
 	}
 	return balanceResult{
 		blk:        blk,