Merge branch '4253-gitolite-migration-wip'

[arvados.git] / services / keepproxy / keepproxy.go

diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go
index 888db7357d47bc8035301501b8d201a3dc8c4c36..581f7f48739fd2fef826ef6b6dfae6c9a8baefec 100644 (file)
--- 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
        }