X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/3147edd07223f5ecac0c12d4ec34cc4c4cb7d657..78e0599b0773976167ccb720fc1596236295337a:/services/keepproxy/keepproxy.go

diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go
index 888db7357d..581f7f4873 100644
--- a/services/keepproxy/keepproxy.go
+++ b/services/keepproxy/keepproxy.go
@@ -1,10 +1,10 @@
 package main
 
 import (
-	"git.curoverse.com/arvados.git/sdk/go/keepclient"
-	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
 	"flag"
 	"fmt"
+	"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
+	"git.curoverse.com/arvados.git/sdk/go/keepclient"
 	"github.com/gorilla/mux"
 	"io"
 	"io/ioutil"
@@ -30,6 +30,7 @@ func main() {
 		no_get           bool
 		no_put           bool
 		default_replicas int
+		timeout          int64
 		pidfile          string
 	)
 
@@ -61,6 +62,12 @@ func main() {
 		2,
 		"Default number of replicas to write if not specified by the client.")
 
+	flagset.Int64Var(
+		&timeout,
+		"timeout",
+		15,
+		"Timeout on requests to internal Keep services (default 15 seconds)")
+
 	flagset.StringVar(
 		&pidfile,
 		"pid",
@@ -91,6 +98,8 @@ func main() {
 
 	kc.Want_replicas = default_replicas
 
+	kc.Client.Timeout = time.Duration(timeout) * time.Second
+
 	listener, err = net.Listen("tcp", listen)
 	if err != nil {
 		log.Fatalf("Could not listen on %v", listen)
@@ -125,16 +134,27 @@ type ApiTokenCache struct {
 
 // Refresh the keep service list every five minutes.
 func RefreshServicesList(kc *keepclient.KeepClient) {
+	var sleeptime time.Duration
 	for {
-		time.Sleep(300 * time.Second)
 		oldservices := kc.ServiceRoots()
-		kc.DiscoverKeepServers()
-		newservices := kc.ServiceRoots()
-		s1 := fmt.Sprint(oldservices)
-		s2 := fmt.Sprint(newservices)
-		if s1 != s2 {
-			log.Printf("Updated server list to %v", s2)
+		newservices, err := kc.DiscoverKeepServers()
+		if err == nil && len(newservices) > 0 {
+			s1 := fmt.Sprint(oldservices)
+			s2 := fmt.Sprint(newservices)
+			if s1 != s2 {
+				log.Printf("Updated server list to %v", s2)
+			}
+			sleeptime = 300 * time.Second
+		} else {
+			// There was an error, or the list is empty, so wait 3 seconds and try again.
+			if err != nil {
+				log.Printf("Error retrieving server list: %v", err)
+			} else {
+				log.Printf("Retrieved an empty server list")
+			}
+			sleeptime = 3 * time.Second
 		}
+		time.Sleep(sleeptime)
 	}
 }
 
@@ -283,7 +303,7 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques
 
 	locator := keepclient.MakeLocator2(hash, hints)
 
-	log.Printf("%s: %s %s", GetRemoteAddress(req), req.Method, hash)
+	log.Printf("%s: %s %s begin", GetRemoteAddress(req), req.Method, hash)
 
 	var pass bool
 	var tok string
@@ -303,37 +323,49 @@ func (this GetBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques
 
 	if req.Method == "GET" {
 		reader, blocklen, _, err = kc.AuthorizedGet(hash, locator.Signature, locator.Timestamp)
-		defer reader.Close()
+		if reader != nil {
+			defer reader.Close()
+		}
 	} else if req.Method == "HEAD" {
 		blocklen, _, err = kc.AuthorizedAsk(hash, locator.Signature, locator.Timestamp)
 	}
 
-	if blocklen > 0 {
-		resp.Header().Set("Content-Length", fmt.Sprint(blocklen))
+	if blocklen == -1 {
+		log.Printf("%s: %s %s Keep server did not return Content-Length",
+			GetRemoteAddress(req), req.Method, hash)
 	}
 
+	var status = 0
 	switch err {
 	case nil:
+		status = http.StatusOK
+		resp.Header().Set("Content-Length", fmt.Sprint(blocklen))
 		if reader != nil {
 			n, err2 := io.Copy(resp, reader)
-			if n != blocklen {
-				log.Printf("%s: %s %s mismatched return %v with Content-Length %v error %v", GetRemoteAddress(req), req.Method, hash, n, blocklen, err2)
+			if blocklen > -1 && n != blocklen {
+				log.Printf("%s: %s %s %v %v mismatched copy size expected Content-Length: %v",
+					GetRemoteAddress(req), req.Method, hash, status, n, blocklen)
 			} else if err2 == nil {
-				log.Printf("%s: %s %s success returned %v bytes", GetRemoteAddress(req), req.Method, hash, n)
+				log.Printf("%s: %s %s %v %v",
+					GetRemoteAddress(req), req.Method, hash, status, n)
 			} else {
-				log.Printf("%s: %s %s returned %v bytes error %v", GetRemoteAddress(req), req.Method, hash, n, err.Error())
+				log.Printf("%s: %s %s %v %v copy error: %v",
+					GetRemoteAddress(req), req.Method, hash, status, n, err2.Error())
 			}
 		} else {
-			log.Printf("%s: %s %s success", GetRemoteAddress(req), req.Method, hash)
+			log.Printf("%s: %s %s %v 0", GetRemoteAddress(req), req.Method, hash, status)
 		}
 	case keepclient.BlockNotFound:
-		http.Error(resp, "Not found", http.StatusNotFound)
+		status = http.StatusNotFound
+		http.Error(resp, "Not Found", http.StatusNotFound)
 	default:
+		status = http.StatusBadGateway
 		http.Error(resp, err.Error(), http.StatusBadGateway)
 	}
 
 	if err != nil {
-		log.Printf("%s: %s %s error %s", GetRemoteAddress(req), req.Method, hash, err.Error())
+		log.Printf("%s: %s %s %v error: %v",
+			GetRemoteAddress(req), req.Method, hash, status, err.Error())
 	}
 }
 
@@ -358,7 +390,7 @@ func (this PutBlockHandler) ServeHTTP(resp http.ResponseWriter, req *http.Reques
 
 	log.Printf("%s: %s %s Content-Length %v", GetRemoteAddress(req), req.Method, hash, contentLength)
 
-	if contentLength < 1 {
+	if contentLength < 0 {
 		http.Error(resp, "Must include Content-Length header", http.StatusLengthRequired)
 		return
 	}