X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/244159419c42341baeb388236ad29cc546b7eca1..57b96147e575d7630ddf81cbb2c554d3adc9c7e2:/services/keepstore/keepstore.go

diff --git a/services/keepstore/keepstore.go b/services/keepstore/keepstore.go
index 9556185fc0..1e8c3d1e0f 100644
--- a/services/keepstore/keepstore.go
+++ b/services/keepstore/keepstore.go
@@ -13,6 +13,8 @@ import (
 	"strings"
 	"syscall"
 	"time"
+	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
+	"git.curoverse.com/arvados.git/sdk/go/keepclient"
 )
 
 // ======================
@@ -34,10 +36,6 @@ const MIN_FREE_KILOBYTES = BLOCKSIZE / 1024
 
 var PROC_MOUNTS = "/proc/mounts"
 
-// The Keep VolumeManager maintains a list of available volumes.
-// Initialized by the --volumes flag (or by FindKeepVolumes).
-var KeepVM VolumeManager
-
 // enforce_permissions controls whether permission signatures
 // should be enforced (affecting GET and DELETE requests).
 // Initialized by the --enforce-permissions flag.
@@ -67,6 +65,7 @@ type KeepError struct {
 
 var (
 	BadRequestError     = &KeepError{400, "Bad Request"}
+	UnauthorizedError   = &KeepError{401, "Unauthorized"}
 	CollisionError      = &KeepError{500, "Collision"}
 	RequestHashError    = &KeepError{422, "Hash mismatch in request"}
 	PermissionError     = &KeepError{403, "Forbidden"}
@@ -83,6 +82,28 @@ func (e *KeepError) Error() string {
 	return e.ErrMsg
 }
 
+// ========================
+// Internal data structures
+//
+// These global variables are used by multiple parts of the
+// program. They are good candidates for moving into their own
+// packages.
+
+// The Keep VolumeManager maintains a list of available volumes.
+// Initialized by the --volumes flag (or by FindKeepVolumes).
+var KeepVM VolumeManager
+
+// The pull list manager and trash queue are threadsafe queues which
+// support atomic update operations. The PullHandler and TrashHandler
+// store results from Data Manager /pull and /trash requests here.
+//
+// See the Keep and Data Manager design documents for more details:
+// https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
+// https://arvados.org/projects/arvados/wiki/Data_Manager_Design_Doc
+//
+var pullq *WorkQueue
+var trashq *WorkQueue
+
 // TODO(twp): continue moving as much code as possible out of main
 // so it can be effectively tested. Esp. handling and postprocessing
 // of command line flags (identifying Keep volumes and initializing
@@ -243,9 +264,11 @@ func main() {
 	// Start a round-robin VolumeManager with the volumes we have found.
 	KeepVM = MakeRRVolumeManager(goodvols)
 
-	// Tell the built-in HTTP server to direct all requests to the REST
-	// router.
-	http.Handle("/", MakeRESTRouter())
+	// Tell the built-in HTTP server to direct all requests to the REST router.
+	loggingRouter := MakeLoggingRESTRouter()
+	http.HandleFunc("/", func(resp http.ResponseWriter, req *http.Request) {
+		loggingRouter.ServeHTTP(resp, req)
+	})
 
 	// Set up a TCP listener.
 	listener, err := net.Listen("tcp", listen)
@@ -253,6 +276,20 @@ func main() {
 		log.Fatal(err)
 	}
 
+	// Initialize Pull queue and worker
+	arv, err := arvadosclient.MakeArvadosClient()
+	if err != nil {
+		log.Fatalf("Error setting up arvados client %s", err.Error())
+	}
+
+	keepClient, err := keepclient.MakeKeepClient(&arv)
+	if err != nil {
+		log.Fatalf("Error setting up keep client %s", err.Error())
+	}
+
+	pullq = NewWorkQueue()
+	go RunPullWorker(pullq, keepClient)
+
 	// Shut down the server gracefully (by closing the listener)
 	// if SIGTERM is received.
 	term := make(chan os.Signal, 1)