X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/74dba22bf46c375d6d8daa8e5afff3e55df993ae..ac7939b70866c8ed8bcdd9f0854c6c845534ba78:/services/datamanager/datamanager.go

diff --git a/services/datamanager/datamanager.go b/services/datamanager/datamanager.go
index 64073afd71..a8e506eacb 100644
--- a/services/datamanager/datamanager.go
+++ b/services/datamanager/datamanager.go
@@ -5,15 +5,52 @@ package main
 import (
 	"flag"
 	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
+	"git.curoverse.com/arvados.git/sdk/go/logger"
 	"git.curoverse.com/arvados.git/sdk/go/util"
 	"git.curoverse.com/arvados.git/services/datamanager/collection"
 	"git.curoverse.com/arvados.git/services/datamanager/keep"
+	"git.curoverse.com/arvados.git/services/datamanager/loggerutil"
 	"log"
+	"time"
 )
 
+var (
+	logEventTypePrefix  string
+	logFrequencySeconds int
+	minutesBetweenRuns  int
+)
+
+func init() {
+	flag.StringVar(&logEventTypePrefix,
+		"log-event-type-prefix",
+		"experimental-data-manager",
+		"Prefix to use in the event_type of our arvados log entries. Set to empty to turn off logging")
+	flag.IntVar(&logFrequencySeconds,
+		"log-frequency-seconds",
+		20,
+		"How frequently we'll write log entries in seconds.")
+	flag.IntVar(&minutesBetweenRuns,
+		"minutes-between-runs",
+		0,
+		"How many minutes we wait betwen data manager runs. 0 means run once and exit.")
+}
+
 func main() {
 	flag.Parse()
+	if minutesBetweenRuns == 0 {
+		singlerun()
+	} else {
+		waitTime := time.Minute * time.Duration(minutesBetweenRuns)
+		for {
+			log.Println("Beginning Run")
+			singlerun()
+			log.Printf("Sleeping for %d minutes", minutesBetweenRuns)
+			time.Sleep(waitTime)
+		}
+	}
+}
 
+func singlerun() {
 	arv, err := arvadosclient.MakeArvadosClient()
 	if err != nil {
 		log.Fatalf("Error setting up arvados client %s", err.Error())
@@ -25,24 +62,40 @@ func main() {
 		log.Fatalf("Current user is not an admin. Datamanager can only be run by admins.")
 	}
 
-	readCollections := collection.GetCollections(
-		collection.GetCollectionsParams{
-			Client: arv, Limit: 50, LogEveryNthCollectionProcessed: 10})
+	var arvLogger *logger.Logger
+	if logEventTypePrefix != "" {
+		arvLogger = logger.NewLogger(logger.LoggerParams{Client: arv,
+			EventTypePrefix: logEventTypePrefix,
+			WriteInterval:   time.Second * time.Duration(logFrequencySeconds)})
+	}
+
+	loggerutil.LogRunInfo(arvLogger)
+	if arvLogger != nil {
+		arvLogger.AddWriteHook(loggerutil.LogMemoryAlloc)
+	}
 
-	log.Printf("Read Collections: %v", readCollections)
+	collectionChannel := make(chan collection.ReadCollections)
 
-	// TODO(misha): Add a "readonly" flag. If we're in readonly mode,
-	// lots of behaviors can become warnings (and obviously we can't
-	// write anything).
-	// if !readCollections.ReadAllCollections {
-	// 	log.Fatalf("Did not read all collections")
-	// }
+	go func() {
+		collectionChannel <- collection.GetCollectionsAndSummarize(
+			collection.GetCollectionsParams{
+				Client: arv, Logger: arvLogger, BatchSize: 50})
+	}()
 
-	log.Printf("Read and processed %d collections",
-		len(readCollections.UuidToCollection))
+	keepServerInfo := keep.GetKeepServersAndSummarize(
+		keep.GetKeepServersParams{Client: arv, Logger: arvLogger, Limit: 1000})
 
-	readServers := keep.GetKeepServers(
-		keep.GetKeepServersParams{Client: arv, Limit: 1000})
+	readCollections := <-collectionChannel
 
-	log.Printf("Returned %d keep disks", len(readServers.AddressToContents))
+	// TODO(misha): Use these together to verify replication.
+	_ = readCollections
+	_ = keepServerInfo
+
+	// Log that we're finished. We force the recording, since go will
+	// not wait for the timer before exiting.
+	if arvLogger != nil {
+		arvLogger.FinalUpdate(func(p map[string]interface{}, e map[string]interface{}) {
+			p["run_info"].(map[string]interface{})["finished_at"] = time.Now()
+		})
+	}
 }