X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/624c5c8d13d6c2e21a80379928387944bedae2a3..35c8b0a58073ab45912d794c90a4c40c17a371be:/services/keepstore/handlers.go?ds=sidebyside

diff --git a/services/keepstore/handlers.go b/services/keepstore/handlers.go
index f462b0aae1..16c39c2c88 100644
--- a/services/keepstore/handlers.go
+++ b/services/keepstore/handlers.go
@@ -40,35 +40,19 @@ func MakeRESTRouter() *mux.Router {
 
 	rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, PutBlockHandler).Methods("PUT")
 	rest.HandleFunc(`/{hash:[0-9a-f]{32}}`, DeleteHandler).Methods("DELETE")
-
-	// For IndexHandler we support:
-	//   /index           - returns all locators
-	//   /index/{prefix}  - returns all locators that begin with {prefix}
-	//      {prefix} is a string of hexadecimal digits between 0 and 32 digits.
-	//      If {prefix} is the empty string, return an index of all locators
-	//      (so /index and /index/ behave identically)
-	//      A client may supply a full 32-digit locator string, in which
-	//      case the server will return an index with either zero or one
-	//      entries. This usage allows a client to check whether a block is
-	//      present, and its size and upload time, without retrieving the
-	//      entire block.
-	//
+	// List all blocks stored here. Privileged client only.
 	rest.HandleFunc(`/index`, IndexHandler).Methods("GET", "HEAD")
-	rest.HandleFunc(
-		`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+	// List blocks stored here whose hash has the given prefix.
+	// Privileged client only.
+	rest.HandleFunc(`/index/{prefix:[0-9a-f]{0,32}}`, IndexHandler).Methods("GET", "HEAD")
+
+	// List volumes: path, device number, bytes used/avail.
 	rest.HandleFunc(`/status.json`, StatusHandler).Methods("GET", "HEAD")
 
-	// The PullHandler and TrashHandler process "PUT /pull" and "PUT
-	// /trash" requests from Data Manager.  These requests instruct
-	// Keep to replicate or delete blocks; see
-	// https://arvados.org/projects/arvados/wiki/Keep_Design_Doc
-	// for more details.
-	//
-	// Each handler parses the JSON list of block management requests
-	// in the message body, and replaces any existing pull queue or
-	// trash queue with their contentes.
-	//
+	// Replace the current pull queue.
 	rest.HandleFunc(`/pull`, PullHandler).Methods("PUT")
+
+	// Replace the current trash queue.
 	rest.HandleFunc(`/trash`, TrashHandler).Methods("PUT")
 
 	// Any request which does not match any of these routes gets
@@ -128,25 +112,17 @@ func GetBlockHandler(resp http.ResponseWriter, req *http.Request) {
 	}
 
 	block, err := GetBlock(hash, false)
-
-	// Garbage collect after each GET. Fixes #2865.
-	// TODO(twp): review Keep memory usage and see if there's
-	// a better way to do this than blindly garbage collecting
-	// after every block.
-	defer runtime.GC()
-
 	if err != nil {
 		// This type assertion is safe because the only errors
 		// GetBlock can return are DiskHashError or NotFoundError.
 		http.Error(resp, err.Error(), err.(*KeepError).HTTPCode)
 		return
 	}
+	defer bufs.Put(block)
 
-	resp.Header().Set("Content-Length", fmt.Sprintf("%d", len(block)))
-
-	_, err = resp.Write(block)
-
-	return
+	resp.Header().Set("Content-Length", strconv.Itoa(len(block)))
+	resp.Header().Set("Content-Type", "application/octet-stream")
+	resp.Write(block)
 }
 
 func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
@@ -160,6 +136,11 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
 	// the body: avoid transmitting data that will not end up
 	// being written anyway.
 
+	if req.ContentLength == -1 {
+		http.Error(resp, SizeRequiredError.Error(), SizeRequiredError.HTTPCode)
+		return
+	}
+
 	if req.ContentLength > BLOCKSIZE {
 		http.Error(resp, TooLongError.Error(), TooLongError.HTTPCode)
 		return
@@ -170,17 +151,17 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
 		return
 	}
 
-	buf := make([]byte, req.ContentLength)
-	nread, err := io.ReadFull(req.Body, buf)
+	buf := bufs.Get(int(req.ContentLength))
+	_, err := io.ReadFull(req.Body, buf)
 	if err != nil {
 		http.Error(resp, err.Error(), 500)
-		return
-	} else if int64(nread) < req.ContentLength {
-		http.Error(resp, "request truncated", 500)
+		bufs.Put(buf)
 		return
 	}
 
 	err = PutBlock(buf, hash)
+	bufs.Put(buf)
+
 	if err != nil {
 		ke := err.(*KeepError)
 		http.Error(resp, ke.Error(), ke.HTTPCode)
@@ -189,10 +170,10 @@ func PutBlockHandler(resp http.ResponseWriter, req *http.Request) {
 
 	// Success; add a size hint, sign the locator if possible, and
 	// return it to the client.
-	return_hash := fmt.Sprintf("%s+%d", hash, len(buf))
+	return_hash := fmt.Sprintf("%s+%d", hash, req.ContentLength)
 	api_token := GetApiToken(req)
 	if PermissionSecret != nil && api_token != "" {
-		expiry := time.Now().Add(permission_ttl)
+		expiry := time.Now().Add(blob_signature_ttl)
 		return_hash = SignLocator(return_hash, api_token, expiry)
 	}
 	resp.Write([]byte(return_hash + "\n"))
@@ -210,11 +191,18 @@ func IndexHandler(resp http.ResponseWriter, req *http.Request) {
 
 	prefix := mux.Vars(req)["prefix"]
 
-	var index string
 	for _, vol := range KeepVM.AllReadable() {
-		index = index + vol.Index(prefix)
+		if err := vol.IndexTo(prefix, resp); err != nil {
+			// The only errors returned by IndexTo are
+			// write errors returned by resp.Write(),
+			// which probably means the client has
+			// disconnected and this error will never be
+			// reported to the client -- but it will
+			// appear in our own error log.
+			http.Error(resp, err.Error(), http.StatusInternalServerError)
+			return
+		}
 	}
-	resp.Write([]byte(index))
 }
 
 // StatusHandler
@@ -421,7 +409,7 @@ func PullHandler(resp http.ResponseWriter, req *http.Request) {
 	var pr []PullRequest
 	r := json.NewDecoder(req.Body)
 	if err := r.Decode(&pr); err != nil {
-		http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+		http.Error(resp, err.Error(), BadRequestError.HTTPCode)
 		return
 	}
 
@@ -455,7 +443,7 @@ func TrashHandler(resp http.ResponseWriter, req *http.Request) {
 	var trash []TrashRequest
 	r := json.NewDecoder(req.Body)
 	if err := r.Decode(&trash); err != nil {
-		http.Error(resp, BadRequestError.Error(), BadRequestError.HTTPCode)
+		http.Error(resp, err.Error(), BadRequestError.HTTPCode)
 		return
 	}
 
@@ -537,6 +525,7 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
 			log.Printf("%s: checksum mismatch for request %s (actual %s)\n",
 				vol, hash, filehash)
 			error_to_caller = DiskHashError
+			bufs.Put(buf)
 			continue
 		}
 		if error_to_caller == DiskHashError {
@@ -548,6 +537,7 @@ func GetBlock(hash string, update_timestamp bool) ([]byte, error) {
 				error_to_caller = GenericError
 				log.Printf("%s: Touch %s failed: %s",
 					vol, hash, error_to_caller)
+				bufs.Put(buf)
 				continue
 			}
 		}
@@ -598,6 +588,7 @@ func PutBlock(block []byte, hash string) error {
 	// so there is nothing special to do if err != nil.
 	//
 	if oldblock, err := GetBlock(hash, true); err == nil {
+		defer bufs.Put(oldblock)
 		if bytes.Compare(block, oldblock) == 0 {
 			// The block already exists; return success.
 			return nil