X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/aae2802ab7e3ecb35ff86998e79927fe7de26e49..282d7a0c9d8279fa0a1293573e4313c8223be1b5:/services/keep-balance/balance.go

diff --git a/services/keep-balance/balance.go b/services/keep-balance/balance.go
index 346de30b6f..08a6c5881c 100644
--- a/services/keep-balance/balance.go
+++ b/services/keep-balance/balance.go
@@ -8,12 +8,16 @@ import (
 	"bytes"
 	"crypto/md5"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"log"
 	"math"
+	"os"
 	"runtime"
 	"sort"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
 	"git.curoverse.com/arvados.git/sdk/go/arvados"
@@ -35,6 +39,8 @@ type Balancer struct {
 	Dumper  logrus.FieldLogger
 	Metrics *metrics
 
+	LostBlocksFile string
+
 	*BlockStateMap
 	KeepServices       map[string]*KeepService
 	DefaultReplication int
@@ -48,6 +54,7 @@ type Balancer struct {
 	errors        []error
 	stats         balancerStats
 	mutex         sync.Mutex
+	lostBlocks    io.Writer
 }
 
 // Run performs a balance operation using the given config and
@@ -64,6 +71,30 @@ func (bal *Balancer) Run(config Config, runOptions RunOptions) (nextRunOptions R
 
 	defer bal.time("sweep", "wall clock time to run one full sweep")()
 
+	var lbFile *os.File
+	if bal.LostBlocksFile != "" {
+		tmpfn := bal.LostBlocksFile + ".tmp"
+		lbFile, err = os.OpenFile(tmpfn, os.O_CREATE|os.O_WRONLY, 0777)
+		if err != nil {
+			return
+		}
+		defer lbFile.Close()
+		err = syscall.Flock(int(lbFile.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+		if err != nil {
+			return
+		}
+		defer func() {
+			// Remove the tempfile only if we didn't get
+			// as far as successfully renaming it.
+			if lbFile != nil {
+				os.Remove(tmpfn)
+			}
+		}()
+		bal.lostBlocks = lbFile
+	} else {
+		bal.lostBlocks = ioutil.Discard
+	}
+
 	if len(config.KeepServiceList.Items) > 0 {
 		err = bal.SetKeepServices(config.KeepServiceList)
 	} else {
@@ -107,6 +138,17 @@ func (bal *Balancer) Run(config Config, runOptions RunOptions) (nextRunOptions R
 	if err = bal.CheckSanityLate(); err != nil {
 		return
 	}
+	if lbFile != nil {
+		err = lbFile.Sync()
+		if err != nil {
+			return
+		}
+		err = os.Rename(bal.LostBlocksFile+".tmp", bal.LostBlocksFile)
+		if err != nil {
+			return
+		}
+		lbFile = nil
+	}
 	if runOptions.CommitPulls {
 		err = bal.CommitPulls(&config.Client)
 		if err != nil {
@@ -206,6 +248,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
 }
 
@@ -332,7 +392,7 @@ 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 {
+			if err != nil || len(errs) > 0 {
 				select {
 				case errs <- err:
 				default:
@@ -383,17 +443,20 @@ 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 {
 		repl = *coll.ReplicationDesired
 	}
 	debugf("%v: %d block x%d", coll.UUID, len(blkids), repl)
-	bal.BlockStateMap.IncreaseDesired(coll.StorageClassesDesired, repl, blkids)
+	// Pass pdh to IncreaseDesired only if LostBlocksFile is being
+	// written -- otherwise it's just a waste of memory.
+	pdh := ""
+	if bal.LostBlocksFile != "" {
+		pdh = coll.PortableDataHash
+	}
+	bal.BlockStateMap.IncreaseDesired(pdh, coll.StorageClassesDesired, repl, blkids)
 	return nil
 }
 
@@ -444,7 +507,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 {
@@ -867,6 +930,11 @@ func (bal *Balancer) collectStatistics(results <-chan balanceResult) {
 			s.lost.replicas -= surplus
 			s.lost.blocks++
 			s.lost.bytes += bytes * int64(-surplus)
+			fmt.Fprintf(bal.lostBlocks, "%s", strings.SplitN(string(result.blkid), "+", 2)[0])
+			for pdh := range result.blk.Refs {
+				fmt.Fprintf(bal.lostBlocks, " %s", pdh)
+			}
+			fmt.Fprint(bal.lostBlocks, "\n")
 		case surplus < 0:
 			s.underrep.replicas -= surplus
 			s.underrep.blocks++