X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/e250e8435dd357e6a90e13c9e0f8db60d31d022c..4d7567b7b3577b561dd064d397c10a9331c3ee16:/services/keep-balance/balance.go

diff --git a/services/keep-balance/balance.go b/services/keep-balance/balance.go
index fd39ee693e..9f814a20d3 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 {
@@ -408,7 +450,13 @@ func (bal *Balancer) addCollection(coll arvados.Collection) error {
 		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
 }
 
@@ -475,7 +523,7 @@ func (bal *Balancer) setupLookupTables() {
 				bal.mountsByClass["default"][mnt] = true
 				continue
 			}
-			for _, class := range mnt.StorageClasses {
+			for class := range mnt.StorageClasses {
 				if mbc := bal.mountsByClass[class]; mbc == nil {
 					bal.classes = append(bal.classes, class)
 					bal.mountsByClass[class] = map[*KeepMount]bool{mnt: true}
@@ -882,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++