[ARVADOS] updated: b956a83421693e0dae56546d63e10dab6eb9cc3a
git at public.curoverse.com
git at public.curoverse.com
Sun Nov 8 15:58:23 EST 2015
Summary of changes:
sdk/go/arvadostest/run_servers.go | 53 +++++------
sdk/python/tests/run_test_server.py | 22 +++--
services/keep-web/handler.go | 60 +++++++++---
services/keep-web/handler_test.go | 103 +++++++++++++++++----
services/keepstore/pull_worker_integration_test.go | 4 +
5 files changed, 170 insertions(+), 72 deletions(-)
discards 78c21ef355567ce880c515715b8e977c6bb6328d (commit)
discards bfa9c38da625c0a327ace454fd85c082564dbf97 (commit)
discards 9b29da4c0b153ba99ae3f1586b5923de42cb544a (commit)
discards b244cb9f8bf3ae96da243e938c0147925101c3c5 (commit)
discards 5d0893cf0f42476b369a9046216d8ee056654a6d (commit)
via b956a83421693e0dae56546d63e10dab6eb9cc3a (commit)
via cf5734e08070805662bfcebab4bc8d3f90a3ae7e (commit)
via a7b140ac8c00ca5f9dfb1f3a46c78dd7b9a68730 (commit)
via b86c7d168ce73f830f01d204189ed0941bf8e469 (commit)
via 6fab045dba4714e9c748337ef974ef257df37353 (commit)
via a3dfcea5c3767936aa7bff6e03268d4ecadd0489 (commit)
via 1e10339657c1dee2b71b4d10eddffd0a35c949b3 (commit)
This update added new revisions after undoing existing revisions. That is
to say, the old revision is not a strict subset of the new revision. This
situation occurs when you --force push a change and generate a repository
containing something like this:
* -- * -- B -- O -- O -- O (78c21ef355567ce880c515715b8e977c6bb6328d)
\
N -- N -- N (b956a83421693e0dae56546d63e10dab6eb9cc3a)
When this happens we assume that you've already had alert emails for all
of the O revisions, and so we here report only the revisions in the N
branch from the common base, B.
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit b956a83421693e0dae56546d63e10dab6eb9cc3a
Author: Tom Clegg <tom at curoverse.com>
Date: Sun Nov 8 15:52:29 2015 -0500
5824: Propagate non-token parts of query string (notably ?attachment=disposition) when redirecting.
diff --git a/services/keep-web/handler.go b/services/keep-web/handler.go
index deef62d..5e772b7 100644
--- a/services/keep-web/handler.go
+++ b/services/keep-web/handler.go
@@ -183,7 +183,17 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
Path: "/",
HttpOnly: true,
})
- redir := (&url.URL{Host: r.Host, Path: r.URL.Path}).String()
+
+ // Propagate query parameters (except api_token) from
+ // the original request.
+ redirQuery := r.URL.Query()
+ redirQuery.Del("api_token")
+
+ redir := (&url.URL{
+ Host: r.Host,
+ Path: r.URL.Path,
+ RawQuery: redirQuery.Encode(),
+ }).String()
w.Header().Add("Location", redir)
statusCode, statusText = http.StatusSeeOther, redir
diff --git a/services/keep-web/handler_test.go b/services/keep-web/handler_test.go
index b4758e0..94a1cf7 100644
--- a/services/keep-web/handler_test.go
+++ b/services/keep-web/handler_test.go
@@ -32,9 +32,11 @@ func (s *IntegrationSuite) TestVhost404(c *check.C) {
arvadostest.NonexistentCollection + ".example.com/t=" + arvadostest.ActiveToken + "/theperthcountyconspiracy",
} {
resp := httptest.NewRecorder()
+ u := mustParseURL(testURL)
req := &http.Request{
- Method: "GET",
- URL: mustParseURL(testURL),
+ Method: "GET",
+ URL: u,
+ RequestURI: u.RequestURI(),
}
(&handler{}).ServeHTTP(resp, req)
c.Check(resp.Code, check.Equals, http.StatusNotFound)
@@ -120,10 +122,11 @@ func doVhostRequestsWithHostPath(c *check.C, authz authorizer, hostPath string)
} {
u := mustParseURL("http://" + hostPath)
req := &http.Request{
- Method: "GET",
- Host: u.Host,
- URL: u,
- Header: http.Header{},
+ Method: "GET",
+ Host: u.Host,
+ URL: u,
+ RequestURI: u.RequestURI(),
+ Header: http.Header{},
}
failCode := authz(req, tok)
resp := doReq(req)
@@ -157,10 +160,11 @@ func doReq(req *http.Request) *httptest.ResponseRecorder {
cookies := (&http.Response{Header: resp.Header()}).Cookies()
u, _ := req.URL.Parse(resp.Header().Get("Location"))
req = &http.Request{
- Method: "GET",
- Host: u.Host,
- URL: u,
- Header: http.Header{},
+ Method: "GET",
+ Host: u.Host,
+ URL: u,
+ RequestURI: u.RequestURI(),
+ Header: http.Header{},
}
for _, c := range cookies {
req.AddCookie(c)
@@ -228,6 +232,21 @@ func (s *IntegrationSuite) TestVhostRedirectQueryTokenSingleOriginError(c *check
)
}
+// If client requests an attachment by putting ?disposition=attachment
+// in the query string, and gets redirected, the redirect target
+// should respond with an attachment.
+func (s *IntegrationSuite) TestVhostRedirectQueryTokenRequestAttachment(c *check.C) {
+ resp := s.testVhostRedirectTokenToCookie(c, "GET",
+ arvadostest.FooCollection+".example.com/foo",
+ "?disposition=attachment&api_token="+arvadostest.ActiveToken,
+ "",
+ "",
+ http.StatusOK,
+ "foo",
+ )
+ c.Check(strings.Split(resp.Header().Get("Content-Disposition"), ";")[0], check.Equals, "attachment")
+}
+
func (s *IntegrationSuite) TestVhostRedirectQueryTokenTrustAllContent(c *check.C) {
defer func(orig bool) {
trustAllContent = orig
@@ -316,12 +335,13 @@ func (s *IntegrationSuite) TestAnonymousTokenError(c *check.C) {
}
func (s *IntegrationSuite) TestRange(c *check.C) {
- u, _ := url.Parse("http://example.com/c="+arvadostest.HelloWorldCollection+"/Hello%20world.txt")
+ u, _ := url.Parse("http://example.com/c=" + arvadostest.HelloWorldCollection + "/Hello%20world.txt")
req := &http.Request{
- Method: "GET",
- Host: u.Host,
- URL: u,
- Header: http.Header{"Range": {"bytes=0-4"}},
+ Method: "GET",
+ Host: u.Host,
+ URL: u,
+ RequestURI: u.RequestURI(),
+ Header: http.Header{"Range": {"bytes=0-4"}},
}
resp := httptest.NewRecorder()
(&handler{}).ServeHTTP(resp, req)
@@ -359,11 +379,12 @@ func (s *IntegrationSuite) TestRange(c *check.C) {
func (s *IntegrationSuite) testVhostRedirectTokenToCookie(c *check.C, method, hostPath, queryString, contentType, reqBody string, expectStatus int, expectRespBody string) *httptest.ResponseRecorder {
u, _ := url.Parse(`http://` + hostPath + queryString)
req := &http.Request{
- Method: method,
- Host: u.Host,
- URL: u,
- Header: http.Header{"Content-Type": {contentType}},
- Body: ioutil.NopCloser(strings.NewReader(reqBody)),
+ Method: method,
+ Host: u.Host,
+ URL: u,
+ RequestURI: u.RequestURI(),
+ Header: http.Header{"Content-Type": {contentType}},
+ Body: ioutil.NopCloser(strings.NewReader(reqBody)),
}
resp := httptest.NewRecorder()
@@ -376,15 +397,16 @@ func (s *IntegrationSuite) testVhostRedirectTokenToCookie(c *check.C, method, ho
if resp.Code != http.StatusSeeOther {
return resp
}
- c.Check(resp.Body.String(), check.Matches, `.*href="//`+regexp.QuoteMeta(html.EscapeString(hostPath))+`".*`)
+ c.Check(resp.Body.String(), check.Matches, `.*href="//`+regexp.QuoteMeta(html.EscapeString(hostPath))+`(\?[^"]*)?".*`)
cookies := (&http.Response{Header: resp.Header()}).Cookies()
u, _ = u.Parse(resp.Header().Get("Location"))
req = &http.Request{
- Method: "GET",
- Host: u.Host,
- URL: u,
- Header: http.Header{},
+ Method: "GET",
+ Host: u.Host,
+ URL: u,
+ RequestURI: u.RequestURI(),
+ Header: http.Header{},
}
for _, c := range cookies {
req.AddCookie(c)
commit cf5734e08070805662bfcebab4bc8d3f90a3ae7e
Author: Tom Clegg <tom at curoverse.com>
Date: Sun Nov 8 06:39:05 2015 -0500
5824: Support partial content with Range header (only if start==0).
diff --git a/services/keep-web/handler.go b/services/keep-web/handler.go
index e8678fa..deef62d 100644
--- a/services/keep-web/handler.go
+++ b/services/keep-web/handler.go
@@ -9,6 +9,7 @@ import (
"net/http"
"net/url"
"os"
+ "regexp"
"strconv"
"strings"
@@ -293,8 +294,10 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
}
defer rdr.Close()
- // One or both of these can be -1 if not found:
basenamePos := strings.LastIndex(filename, "/")
+ if basenamePos < 0 {
+ basenamePos = 0
+ }
extPos := strings.LastIndex(filename, ".")
if extPos > basenamePos {
// Now extPos is safely >= 0.
@@ -306,8 +309,38 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Length", fmt.Sprintf("%d", rdr.Len()))
}
+ applyContentDispositionHdr(w, r, filename[basenamePos:], attachment)
+ rangeRdr, statusCode := applyRangeHdr(w, r, rdr)
+
+ w.WriteHeader(statusCode)
+ _, err = io.Copy(w, rangeRdr)
+ if err != nil {
+ statusCode, statusText = http.StatusBadGateway, err.Error()
+ }
+}
+
+var rangeRe = regexp.MustCompile(`^bytes=0-([0-9]*)$`)
+
+func applyRangeHdr(w http.ResponseWriter, r *http.Request, rdr keepclient.ReadCloserWithLen) (io.Reader, int) {
+ w.Header().Set("Accept-Ranges", "bytes")
+ hdr := r.Header.Get("Range")
+ fields := rangeRe.FindStringSubmatch(hdr)
+ if fields == nil {
+ return rdr, http.StatusOK
+ }
+ rangeEnd, err := strconv.ParseInt(fields[1], 10, 64)
+ if err != nil {
+ // Empty or too big for int64 == send entire content
+ return rdr, http.StatusOK
+ }
+ w.Header().Set("Content-Length", fmt.Sprintf("%d", rangeEnd+1))
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", 0, rangeEnd, rdr.Len()))
+ return &io.LimitedReader{R: rdr, N: rangeEnd + 1}, http.StatusPartialContent
+}
+
+func applyContentDispositionHdr(w http.ResponseWriter, r *http.Request, filename string, isAttachment bool) {
disposition := "inline"
- if attachment {
+ if isAttachment {
disposition = "attachment"
}
if strings.ContainsRune(r.RequestURI, '?') {
@@ -316,18 +349,9 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
// "filename.txt?disposition=attachment".
//
// TODO(TC): Follow advice at RFC 6266 appendix D
- if basenamePos < 0 {
- basenamePos = 0
- }
- disposition += "; filename=" + strconv.QuoteToASCII(filename[basenamePos:])
+ disposition += "; filename=" + strconv.QuoteToASCII(filename)
}
if disposition != "inline" {
w.Header().Set("Content-Disposition", disposition)
}
-
- w.WriteHeader(http.StatusOK)
- _, err = io.Copy(w, rdr)
- if err != nil {
- statusCode, statusText = http.StatusBadGateway, err.Error()
- }
}
diff --git a/services/keep-web/handler_test.go b/services/keep-web/handler_test.go
index 392de94..b4758e0 100644
--- a/services/keep-web/handler_test.go
+++ b/services/keep-web/handler_test.go
@@ -315,6 +315,47 @@ func (s *IntegrationSuite) TestAnonymousTokenError(c *check.C) {
)
}
+func (s *IntegrationSuite) TestRange(c *check.C) {
+ u, _ := url.Parse("http://example.com/c="+arvadostest.HelloWorldCollection+"/Hello%20world.txt")
+ req := &http.Request{
+ Method: "GET",
+ Host: u.Host,
+ URL: u,
+ Header: http.Header{"Range": {"bytes=0-4"}},
+ }
+ resp := httptest.NewRecorder()
+ (&handler{}).ServeHTTP(resp, req)
+ c.Check(resp.Code, check.Equals, http.StatusPartialContent)
+ c.Check(resp.Body.String(), check.Equals, "Hello")
+ c.Check(resp.Header().Get("Content-Length"), check.Equals, "5")
+ c.Check(resp.Header().Get("Content-Range"), check.Equals, "bytes 0-4/12")
+
+ req.Header.Set("Range", "bytes=0-")
+ resp = httptest.NewRecorder()
+ (&handler{}).ServeHTTP(resp, req)
+ // 200 and 206 are both correct:
+ c.Check(resp.Code, check.Equals, http.StatusOK)
+ c.Check(resp.Body.String(), check.Equals, "Hello world\n")
+ c.Check(resp.Header().Get("Content-Length"), check.Equals, "12")
+
+ // Unsupported ranges are ignored
+ for _, hdr := range []string{
+ "bytes=5-5", // non-zero start byte
+ "bytes=-5", // last 5 bytes
+ "cubits=0-5", // unsupported unit
+ "bytes=0-340282366920938463463374607431768211456", // 2^128
+ } {
+ req.Header.Set("Range", hdr)
+ resp = httptest.NewRecorder()
+ (&handler{}).ServeHTTP(resp, req)
+ c.Check(resp.Code, check.Equals, http.StatusOK)
+ c.Check(resp.Body.String(), check.Equals, "Hello world\n")
+ c.Check(resp.Header().Get("Content-Length"), check.Equals, "12")
+ c.Check(resp.Header().Get("Content-Range"), check.Equals, "")
+ c.Check(resp.Header().Get("Accept-Ranges"), check.Equals, "bytes")
+ }
+}
+
func (s *IntegrationSuite) testVhostRedirectTokenToCookie(c *check.C, method, hostPath, queryString, contentType, reqBody string, expectStatus int, expectRespBody string) *httptest.ResponseRecorder {
u, _ := url.Parse(`http://` + hostPath + queryString)
req := &http.Request{
commit a7b140ac8c00ca5f9dfb1f3a46c78dd7b9a68730
Author: Tom Clegg <tom at curoverse.com>
Date: Sat Nov 7 04:36:01 2015 -0500
5824: Fix disposition=attachment handling.
Propagate disposition=attachment from Workbench to keep-web when
redirecting.
Include a filename in the Content-Disposition header if the request
URL contains "?", so UAs don't mistakenly include the query string as
part of the default filename.
diff --git a/apps/workbench/app/controllers/collections_controller.rb b/apps/workbench/app/controllers/collections_controller.rb
index 38b58a1..cadef38 100644
--- a/apps/workbench/app/controllers/collections_controller.rb
+++ b/apps/workbench/app/controllers/collections_controller.rb
@@ -148,6 +148,7 @@ class CollectionsController < ApplicationController
# to read the collection.
opts[:query_token] = usable_token
end
+ opts[:disposition] = params[:disposition]
return redirect_to keep_web_url(params[:uuid], params[:file], opts)
end
@@ -332,9 +333,18 @@ class CollectionsController < ApplicationController
end
uri.path += '_/'
uri.path += CGI::escape(file)
- if opts[:query_token]
- uri.query = 'api_token=' + CGI::escape(opts[:query_token])
+
+ query_params = []
+ { query_token: 'api_token',
+ disposition: 'disposition' }.each do |opt, param|
+ if opts[opt]
+ query_params << param + '=' + CGI::escape(opts[opt])
+ end
end
+ unless query_params.empty?
+ uri.query = query_params.join '&'
+ end
+
uri.to_s
end
diff --git a/apps/workbench/config/application.default.yml b/apps/workbench/config/application.default.yml
index 5504fd2..0a9ee9f 100644
--- a/apps/workbench/config/application.default.yml
+++ b/apps/workbench/config/application.default.yml
@@ -230,6 +230,13 @@ common:
# download facility.
#
# Examples:
- # keep_web_url: https://%{uuid_or_pdh}.dl.zzzzz.your.domain
- # keep_web_url: https://%{uuid_or_pdh}--dl.zzzzz.your.domain
+ # keep_web_url: https://%{uuid_or_pdh}.collections.uuid_prefix.arvadosapi.com
+ # keep_web_url: https://%{uuid_or_pdh}--collections.uuid_prefix.arvadosapi.com
+ #
+ # Example supporting only public data and collection-sharing links:
+ # keep_web_url: https://collections.uuid_prefix.arvadosapi.com/c=%{uuid_or_pdh}
+ #
+ # Example supporting only download/attachment: (using keep-web
+ # -attachment-only-host collections.uuid_prefix.arvadosapi.com):
+ # keep_web_url: https://collections.uuid_prefix.arvadosapi.com/c=%{uuid_or_pdh}
keep_web_url: false
diff --git a/apps/workbench/test/integration_helper.rb b/apps/workbench/test/integration_helper.rb
index 5750a1b..1cda127 100644
--- a/apps/workbench/test/integration_helper.rb
+++ b/apps/workbench/test/integration_helper.rb
@@ -27,6 +27,7 @@ Capybara.register_driver :selenium_with_download do |app|
profile['browser.download.folderList'] = 2 # "save to user-defined location"
profile['browser.download.manager.showWhenStarting'] = false
profile['browser.helperApps.alwaysAsk.force'] = false
+ profile['browser.helperApps.neverAsk.saveToDisk'] = 'text/plain,application/octet-stream'
Capybara::Selenium::Driver.new app, profile: profile
end
diff --git a/services/keep-web/handler.go b/services/keep-web/handler.go
index 3e38cc3..e8678fa 100644
--- a/services/keep-web/handler.go
+++ b/services/keep-web/handler.go
@@ -9,6 +9,7 @@ import (
"net/http"
"net/url"
"os"
+ "strconv"
"strings"
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
@@ -304,8 +305,24 @@ func (h *handler) ServeHTTP(wOrig http.ResponseWriter, r *http.Request) {
if rdr, ok := rdr.(keepclient.ReadCloserWithLen); ok {
w.Header().Set("Content-Length", fmt.Sprintf("%d", rdr.Len()))
}
+
+ disposition := "inline"
if attachment {
- w.Header().Set("Content-Disposition", "attachment")
+ disposition = "attachment"
+ }
+ if strings.ContainsRune(r.RequestURI, '?') {
+ // Help the UA realize that the filename is just
+ // "filename.txt", not
+ // "filename.txt?disposition=attachment".
+ //
+ // TODO(TC): Follow advice at RFC 6266 appendix D
+ if basenamePos < 0 {
+ basenamePos = 0
+ }
+ disposition += "; filename=" + strconv.QuoteToASCII(filename[basenamePos:])
+ }
+ if disposition != "inline" {
+ w.Header().Set("Content-Disposition", disposition)
}
w.WriteHeader(http.StatusOK)
commit b86c7d168ce73f830f01d204189ed0941bf8e469
Author: Tom Clegg <tom at curoverse.com>
Date: Sat Nov 7 04:06:47 2015 -0500
5824: Fixup new keepproxy tests to use simplified test setup.
See 813d35123538b00ab70719e247b6bb0881269460
diff --git a/services/keepproxy/keepproxy_test.go b/services/keepproxy/keepproxy_test.go
index 9c33948..7c5b362 100644
--- a/services/keepproxy/keepproxy_test.go
+++ b/services/keepproxy/keepproxy_test.go
@@ -73,6 +73,10 @@ func (s *ServerRequiredSuite) TearDownSuite(c *C) {
func (s *NoKeepServerSuite) SetUpSuite(c *C) {
arvadostest.StartAPI()
+ // We need API to have some keep services listed, but the
+ // services themselves should be unresponsive.
+ arvadostest.StartKeep(2, false)
+ arvadostest.StopKeep(2)
}
func (s *NoKeepServerSuite) SetUpTest(c *C) {
@@ -103,7 +107,7 @@ func runProxy(c *C, args []string, bogusClientToken bool) *keepclient.KeepClient
kc.Arvados.External = true
kc.Using_proxy = true
- return &kc
+ return kc
}
func (s *ServerRequiredSuite) TestPutAskGet(c *C) {
@@ -407,8 +411,7 @@ func (s *ServerRequiredSuite) TestGetIndex(c *C) {
}
func (s *ServerRequiredSuite) TestPutAskGetInvalidToken(c *C) {
- kc := runProxy(c, []string{"keepproxy"}, 28852, false)
- waitForListener()
+ kc := runProxy(c, nil, false)
defer closeListener()
// Put a test block
@@ -446,7 +449,7 @@ func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
// keepclient with no such keep server
kc := keepclient.New(&arv)
locals := map[string]string{
- "proxy": "http://localhost:12345",
+ TestProxyUUID: "http://localhost:12345",
}
kc.SetServiceRoots(locals, nil, nil)
@@ -467,22 +470,24 @@ func (s *ServerRequiredSuite) TestAskGetKeepProxyConnectionError(c *C) {
}
func (s *NoKeepServerSuite) TestAskGetNoKeepServerError(c *C) {
- kc := runProxy(c, []string{"keepproxy"}, 29999, false)
- waitForListener()
+ kc := runProxy(c, nil, false)
defer closeListener()
- // Ask should result in temporary connection refused error
hash := fmt.Sprintf("%x", md5.Sum([]byte("foo")))
- _, _, err := kc.Ask(hash)
- c.Check(err, NotNil)
- errNotFound, _ := err.(*keepclient.ErrNotFound)
- c.Check(errNotFound.Temporary(), Equals, true)
- c.Assert(strings.Contains(err.Error(), "HTTP 502"), Equals, true)
-
- // Get should result in temporary connection refused error
- _, _, _, err = kc.Get(hash)
- c.Check(err, NotNil)
- errNotFound, _ = err.(*keepclient.ErrNotFound)
- c.Check(errNotFound.Temporary(), Equals, true)
- c.Assert(strings.Contains(err.Error(), "HTTP 502"), Equals, true)
+ for _, f := range []func()error {
+ func() error {
+ _, _, err := kc.Ask(hash)
+ return err
+ },
+ func() error {
+ _, _, _, err := kc.Get(hash)
+ return err
+ },
+ } {
+ err := f()
+ c.Assert(err, NotNil)
+ errNotFound, _ := err.(*keepclient.ErrNotFound)
+ c.Check(errNotFound.Temporary(), Equals, true)
+ c.Check(err, ErrorMatches, `.*HTTP 502.*`)
+ }
}
commit 6fab045dba4714e9c748337ef974ef257df37353
Author: Tom Clegg <tom at curoverse.com>
Date: Sat Nov 7 04:03:27 2015 -0500
5824: Move "periodically refresh Keep services" func from keepproxy to SDK.
diff --git a/sdk/go/keepclient/discover.go b/sdk/go/keepclient/discover.go
new file mode 100644
index 0000000..650c2b5
--- /dev/null
+++ b/sdk/go/keepclient/discover.go
@@ -0,0 +1,130 @@
+package keepclient
+
+import (
+ "encoding/json"
+ "fmt"
+ "log"
+ "os"
+ "os/signal"
+ "reflect"
+ "strings"
+ "syscall"
+ "time"
+)
+
+// DiscoverKeepServers gets list of available keep services from api server
+func (this *KeepClient) DiscoverKeepServers() error {
+ var list svcList
+
+ // Get keep services from api server
+ err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &list)
+ if err != nil {
+ return err
+ }
+
+ return this.loadKeepServers(list)
+}
+
+// LoadKeepServicesFromJSON gets list of available keep services from given JSON
+func (this *KeepClient) LoadKeepServicesFromJSON(services string) error {
+ var list svcList
+
+ // Load keep services from given json
+ dec := json.NewDecoder(strings.NewReader(services))
+ if err := dec.Decode(&list); err != nil {
+ return err
+ }
+
+ return this.loadKeepServers(list)
+}
+
+// RefreshServices calls DiscoverKeepServers to refresh the keep
+// service list on SIGHUP; when the given interval has elapsed since
+// the last refresh; and (if the last refresh failed) the given
+// errInterval has elapsed.
+func (kc *KeepClient) RefreshServices(interval, errInterval time.Duration) {
+ var previousRoots = []map[string]string{}
+
+ timer := time.NewTimer(interval)
+ gotHUP := make(chan os.Signal, 1)
+ signal.Notify(gotHUP, syscall.SIGHUP)
+
+ for {
+ select {
+ case <-gotHUP:
+ case <-timer.C:
+ }
+ timer.Reset(interval)
+
+ if err := kc.DiscoverKeepServers(); err != nil {
+ log.Println("Error retrieving services list: %v (retrying in %v)", err, errInterval)
+ timer.Reset(errInterval)
+ continue
+ }
+ newRoots := []map[string]string{kc.LocalRoots(), kc.GatewayRoots()}
+
+ if !reflect.DeepEqual(previousRoots, newRoots) {
+ log.Printf("Updated services list: locals %v gateways %v", newRoots[0], newRoots[1])
+ previousRoots = newRoots
+ }
+
+ if len(newRoots[0]) == 0 {
+ log.Printf("WARNING: No local services (retrying in %v)", errInterval)
+ timer.Reset(errInterval)
+ }
+ }
+}
+
+// loadKeepServers
+func (this *KeepClient) loadKeepServers(list svcList) error {
+ listed := make(map[string]bool)
+ localRoots := make(map[string]string)
+ gatewayRoots := make(map[string]string)
+ writableLocalRoots := make(map[string]string)
+
+ // replicasPerService is 1 for disks; unknown or unlimited otherwise
+ this.replicasPerService = 1
+ this.Using_proxy = false
+
+ for _, service := range list.Items {
+ scheme := "http"
+ if service.SSL {
+ scheme = "https"
+ }
+ url := fmt.Sprintf("%s://%s:%d", scheme, service.Hostname, service.Port)
+
+ // Skip duplicates
+ if listed[url] {
+ continue
+ }
+ listed[url] = true
+
+ localRoots[service.Uuid] = url
+ if service.SvcType == "proxy" {
+ this.Using_proxy = true
+ }
+
+ if service.ReadOnly == false {
+ writableLocalRoots[service.Uuid] = url
+ if service.SvcType != "disk" {
+ this.replicasPerService = 0
+ }
+ }
+
+ // Gateway services are only used when specified by
+ // UUID, so there's nothing to gain by filtering them
+ // by service type. Including all accessible services
+ // (gateway and otherwise) merely accommodates more
+ // service configurations.
+ gatewayRoots[service.Uuid] = url
+ }
+
+ if this.Using_proxy {
+ this.setClientSettingsProxy()
+ } else {
+ this.setClientSettingsDisk()
+ }
+
+ this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots)
+ return nil
+}
diff --git a/sdk/go/keepclient/discover_test.go b/sdk/go/keepclient/discover_test.go
new file mode 100644
index 0000000..43a984e
--- /dev/null
+++ b/sdk/go/keepclient/discover_test.go
@@ -0,0 +1,21 @@
+package keepclient
+
+import (
+ "fmt"
+ "time"
+
+ "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
+)
+
+func ExampleRefreshServices() {
+ arv, err := arvadosclient.MakeArvadosClient()
+ if err != nil {
+ panic(err)
+ }
+ kc, err := MakeKeepClient(&arv)
+ if err != nil {
+ panic(err)
+ }
+ go kc.RefreshServices(5*time.Minute, 3*time.Second)
+ fmt.Printf("LocalRoots: %#v\n", kc.LocalRoots())
+}
diff --git a/sdk/go/keepclient/support.go b/sdk/go/keepclient/support.go
index 8414afa..90ac3bc 100644
--- a/sdk/go/keepclient/support.go
+++ b/sdk/go/keepclient/support.go
@@ -2,7 +2,6 @@ package keepclient
import (
"crypto/md5"
- "encoding/json"
"errors"
"fmt"
"git.curoverse.com/arvados.git/sdk/go/streamer"
@@ -87,86 +86,6 @@ type svcList struct {
Items []keepService `json:"items"`
}
-// DiscoverKeepServers gets list of available keep services from api server
-func (this *KeepClient) DiscoverKeepServers() error {
- var list svcList
-
- // Get keep services from api server
- err := this.Arvados.Call("GET", "keep_services", "", "accessible", nil, &list)
- if err != nil {
- return err
- }
-
- return this.loadKeepServers(list)
-}
-
-// LoadKeepServicesFromJSON gets list of available keep services from given JSON
-func (this *KeepClient) LoadKeepServicesFromJSON(services string) error {
- var list svcList
-
- // Load keep services from given json
- dec := json.NewDecoder(strings.NewReader(services))
- if err := dec.Decode(&list); err != nil {
- return err
- }
-
- return this.loadKeepServers(list)
-}
-
-// loadKeepServers
-func (this *KeepClient) loadKeepServers(list svcList) error {
- listed := make(map[string]bool)
- localRoots := make(map[string]string)
- gatewayRoots := make(map[string]string)
- writableLocalRoots := make(map[string]string)
-
- // replicasPerService is 1 for disks; unknown or unlimited otherwise
- this.replicasPerService = 1
- this.Using_proxy = false
-
- for _, service := range list.Items {
- scheme := "http"
- if service.SSL {
- scheme = "https"
- }
- url := fmt.Sprintf("%s://%s:%d", scheme, service.Hostname, service.Port)
-
- // Skip duplicates
- if listed[url] {
- continue
- }
- listed[url] = true
-
- localRoots[service.Uuid] = url
- if service.SvcType == "proxy" {
- this.Using_proxy = true
- }
-
- if service.ReadOnly == false {
- writableLocalRoots[service.Uuid] = url
- if service.SvcType != "disk" {
- this.replicasPerService = 0
- }
- }
-
- // Gateway services are only used when specified by
- // UUID, so there's nothing to gain by filtering them
- // by service type. Including all accessible services
- // (gateway and otherwise) merely accommodates more
- // service configurations.
- gatewayRoots[service.Uuid] = url
- }
-
- if this.Using_proxy {
- this.setClientSettingsProxy()
- } else {
- this.setClientSettingsDisk()
- }
-
- this.SetServiceRoots(localRoots, writableLocalRoots, gatewayRoots)
- return nil
-}
-
type uploadStatus struct {
err error
url string
diff --git a/services/keepproxy/keepproxy.go b/services/keepproxy/keepproxy.go
index 2efc9ce..7b5cd2b 100644
--- a/services/keepproxy/keepproxy.go
+++ b/services/keepproxy/keepproxy.go
@@ -14,7 +14,6 @@ import (
"net/http"
"os"
"os/signal"
- "reflect"
"regexp"
"sync"
"syscall"
@@ -104,7 +103,7 @@ func main() {
kc.Want_replicas = default_replicas
kc.Client.Timeout = time.Duration(timeout) * time.Second
- go RefreshServicesList(kc, 5*time.Minute, 3*time.Second)
+ go kc.RefreshServices(5*time.Minute, 3*time.Second)
listener, err = net.Listen("tcp", listen)
if err != nil {
@@ -135,42 +134,6 @@ type ApiTokenCache struct {
expireTime int64
}
-// Refresh the keep service list on SIGHUP; when the given interval
-// has elapsed since the last refresh; and (if the last refresh
-// failed) the given errInterval has elapsed.
-func RefreshServicesList(kc *keepclient.KeepClient, interval, errInterval time.Duration) {
- var previousRoots = []map[string]string{}
-
- timer := time.NewTimer(interval)
- gotHUP := make(chan os.Signal, 1)
- signal.Notify(gotHUP, syscall.SIGHUP)
-
- for {
- select {
- case <-gotHUP:
- case <-timer.C:
- }
- timer.Reset(interval)
-
- if err := kc.DiscoverKeepServers(); err != nil {
- log.Println("Error retrieving services list: %v (retrying in %v)", err, errInterval)
- timer.Reset(errInterval)
- continue
- }
- newRoots := []map[string]string{kc.LocalRoots(), kc.GatewayRoots()}
-
- if !reflect.DeepEqual(previousRoots, newRoots) {
- log.Printf("Updated services list: locals %v gateways %v", newRoots[0], newRoots[1])
- previousRoots = newRoots
- }
-
- if len(newRoots[0]) == 0 {
- log.Printf("WARNING: No local services (retrying in %v)", errInterval)
- timer.Reset(errInterval)
- }
- }
-}
-
// Cache the token and set an expire time. If we already have an expire time
// on the token, it is not updated.
func (this *ApiTokenCache) RememberToken(token string) {
commit a3dfcea5c3767936aa7bff6e03268d4ecadd0489
Author: Tom Clegg <tom at curoverse.com>
Date: Sat Nov 7 04:00:50 2015 -0500
5824: Fix server shutdown code.
* Pay attention to --num-keep-servers in stop_keep.
* Wait for processes to exit, to avoid start/stop races.
* Tighten exception handling in kill_server_pid() and warn instead of
crashing in various races.
* Log TERM signals.
* Log when a server does not shut down within the given deadline.
diff --git a/sdk/go/arvadosclient/arvadosclient_test.go b/sdk/go/arvadosclient/arvadosclient_test.go
index 75af3ca..d148723 100644
--- a/sdk/go/arvadosclient/arvadosclient_test.go
+++ b/sdk/go/arvadosclient/arvadosclient_test.go
@@ -25,6 +25,11 @@ func (s *ServerRequiredSuite) SetUpSuite(c *C) {
arvadostest.StartKeep(2, false)
}
+func (s *ServerRequiredSuite) TearDownSuite(c *C) {
+ arvadostest.StopKeep(2)
+ arvadostest.StopAPI()
+}
+
func (s *ServerRequiredSuite) SetUpTest(c *C) {
arvadostest.ResetEnv()
}
diff --git a/sdk/python/tests/run_test_server.py b/sdk/python/tests/run_test_server.py
index 8d1c202..a83566a 100644
--- a/sdk/python/tests/run_test_server.py
+++ b/sdk/python/tests/run_test_server.py
@@ -32,7 +32,6 @@ import arvados.config
ARVADOS_DIR = os.path.realpath(os.path.join(MY_DIRNAME, '../../..'))
SERVICES_SRC_DIR = os.path.join(ARVADOS_DIR, 'services')
-SERVER_PID_PATH = 'tmp/pids/test-server.pid'
if 'GOPATH' in os.environ:
gopaths = os.environ['GOPATH'].split(':')
gobins = [os.path.join(path, 'bin') for path in gopaths]
@@ -70,28 +69,62 @@ def kill_server_pid(pidfile, wait=10, passenger_root=False):
import signal
import subprocess
import time
- try:
- if passenger_root:
- # First try to shut down nicely
- restore_cwd = os.getcwd()
- os.chdir(passenger_root)
- subprocess.call([
- 'bundle', 'exec', 'passenger', 'stop', '--pid-file', pidfile])
- os.chdir(restore_cwd)
- now = time.time()
- timeout = now + wait
- with open(pidfile, 'r') as f:
- server_pid = int(f.read())
- while now <= timeout:
- if not passenger_root or timeout - now < wait / 2:
- # Half timeout has elapsed. Start sending SIGTERM
- os.kill(server_pid, signal.SIGTERM)
- # Raise OSError if process has disappeared
- os.getpgid(server_pid)
+
+ now = time.time()
+ startTERM = now
+ deadline = now + wait
+
+ if passenger_root:
+ # First try to shut down nicely
+ restore_cwd = os.getcwd()
+ os.chdir(passenger_root)
+ subprocess.call([
+ 'bundle', 'exec', 'passenger', 'stop', '--pid-file', pidfile])
+ os.chdir(restore_cwd)
+ # Use up to half of the +wait+ period waiting for "passenger
+ # stop" to work. If the process hasn't exited by then, start
+ # sending TERM signals.
+ startTERM += wait/2
+
+ server_pid = None
+ while now <= deadline and server_pid is None:
+ try:
+ with open(pidfile, 'r') as f:
+ server_pid = int(f.read())
+ except IOError:
+ # No pidfile = nothing to kill.
+ return
+ except ValueError as error:
+ # Pidfile exists, but we can't parse it. Perhaps the
+ # server has created the file but hasn't written its PID
+ # yet?
+ print("Parse error reading pidfile {}: {}".format(pidfile, error))
time.sleep(0.1)
now = time.time()
- except EnvironmentError:
- pass
+
+ while now <= deadline:
+ try:
+ exited, _ = os.waitpid(server_pid, os.WNOHANG)
+ if exited > 0:
+ return
+ except OSError:
+ # already exited, or isn't our child process
+ pass
+ try:
+ if now >= startTERM:
+ os.kill(server_pid, signal.SIGTERM)
+ print("Sent SIGTERM to {} ({})".format(server_pid, pidfile))
+ except OSError as error:
+ if error.errno == errno.ESRCH:
+ # Thrown by os.getpgid() or os.kill() if the process
+ # does not exist, i.e., our work here is done.
+ return
+ raise
+ time.sleep(0.1)
+ now = time.time()
+
+ print("Server PID {} ({}) did not exit, giving up after {}s".
+ format(server_pid, pidfile, wait))
def find_available_port():
"""Return an IPv4 port number that is not in use right now.
@@ -179,7 +212,7 @@ def run(leave_running_atexit=False):
# Delete cached discovery document.
shutil.rmtree(arvados.http_cache('discovery'))
- pid_file = os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH)
+ pid_file = _pidfile('api')
pid_file_ok = find_server_pid(pid_file, 0)
existing_api_host = os.environ.get('ARVADOS_TEST_API_HOST', my_api_host)
@@ -251,7 +284,7 @@ def run(leave_running_atexit=False):
start_msg = subprocess.check_output(
['bundle', 'exec',
'passenger', 'start', '-d', '-p{}'.format(port),
- '--pid-file', os.path.join(os.getcwd(), pid_file),
+ '--pid-file', pid_file,
'--log-file', os.path.join(os.getcwd(), 'log/test.log'),
'--ssl',
'--ssl-certificate', 'tmp/self-signed.pem',
@@ -313,7 +346,7 @@ def stop(force=False):
"""
global my_api_host
if force or my_api_host is not None:
- kill_server_pid(os.path.join(SERVICES_SRC_DIR, 'api', SERVER_PID_PATH))
+ kill_server_pid(_pidfile('api'))
my_api_host = None
def _start_keep(n, keep_args):
@@ -388,7 +421,7 @@ def run_keep(blob_signing_key=None, enforce_permissions=False, num_servers=2):
os.kill(int(open(proxypidfile).read()), signal.SIGHUP)
def _stop_keep(n):
- kill_server_pid(_pidfile('keep{}'.format(n)), 0)
+ kill_server_pid(_pidfile('keep{}'.format(n)))
if os.path.exists("{}/keep{}.volume".format(TEST_TMPDIR, n)):
with open("{}/keep{}.volume".format(TEST_TMPDIR, n), 'r') as r:
shutil.rmtree(r.read(), True)
@@ -436,7 +469,7 @@ def run_keep_proxy():
def stop_keep_proxy():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
return
- kill_server_pid(_pidfile('keepproxy'), wait=0)
+ kill_server_pid(_pidfile('keepproxy'))
def run_arv_git_httpd():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
@@ -461,7 +494,7 @@ def run_arv_git_httpd():
def stop_arv_git_httpd():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
return
- kill_server_pid(_pidfile('arv-git-httpd'), wait=0)
+ kill_server_pid(_pidfile('arv-git-httpd'))
def run_keep_web():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
@@ -486,7 +519,7 @@ def run_keep_web():
def stop_keep_web():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
return
- kill_server_pid(_pidfile('keep-web'), wait=0)
+ kill_server_pid(_pidfile('keep-web'))
def run_nginx():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
@@ -526,7 +559,7 @@ def run_nginx():
def stop_nginx():
if 'ARVADOS_TEST_PROXY_SERVICES' in os.environ:
return
- kill_server_pid(_pidfile('nginx'), wait=0)
+ kill_server_pid(_pidfile('nginx'))
def _pidfile(program):
return os.path.join(TEST_TMPDIR, program + '.pid')
@@ -678,7 +711,7 @@ if __name__ == "__main__":
elif args.action == 'start_keep':
run_keep(enforce_permissions=args.keep_enforce_permissions, num_servers=args.num_keep_servers)
elif args.action == 'stop_keep':
- stop_keep()
+ stop_keep(num_servers=args.num_keep_servers)
elif args.action == 'start_keep_proxy':
run_keep_proxy()
elif args.action == 'stop_keep_proxy':
commit 1e10339657c1dee2b71b4d10eddffd0a35c949b3
Author: Tom Clegg <tom at curoverse.com>
Date: Sat Nov 7 03:54:03 2015 -0500
5824: Fix Keep server shutdown, check errors, simplify stderr redirection.
(Oops, we forgot to actually Run() the python command for stop_keep.)
diff --git a/sdk/go/arvadostest/run_servers.go b/sdk/go/arvadostest/run_servers.go
index 27c552a..c61b68b 100644
--- a/sdk/go/arvadostest/run_servers.go
+++ b/sdk/go/arvadostest/run_servers.go
@@ -3,9 +3,6 @@ package arvadostest
import (
"bufio"
"bytes"
- "fmt"
- "io"
- "io/ioutil"
"log"
"os"
"os/exec"
@@ -68,24 +65,12 @@ func StartAPI() {
chdirToPythonTests()
cmd := exec.Command("python", "run_test_server.py", "start", "--auth", "admin")
- stderr, err := cmd.StderrPipe()
- if err != nil {
- log.Fatal(err)
- }
- go io.Copy(os.Stderr, stderr)
- stdout, err := cmd.StdoutPipe()
+ cmd.Stdin = nil
+ cmd.Stderr = os.Stderr
+
+ authScript, err := cmd.Output()
if err != nil {
- log.Fatal(err)
- }
- if err = cmd.Start(); err != nil {
- log.Fatal(err)
- }
- var authScript []byte
- if authScript, err = ioutil.ReadAll(stdout); err != nil {
- log.Fatal(err)
- }
- if err = cmd.Wait(); err != nil {
- log.Fatal(err)
+ log.Fatalf("%+v: %s", cmd.Args, err)
}
ParseAuthSettings(authScript)
ResetEnv()
@@ -96,7 +81,7 @@ func StopAPI() {
defer os.Chdir(cwd)
chdirToPythonTests()
- exec.Command("python", "run_test_server.py", "stop").Run()
+ bgRun(exec.Command("python", "run_test_server.py", "stop"))
}
// StartKeep starts the given number of keep servers,
@@ -112,16 +97,7 @@ func StartKeep(numKeepServers int, enforcePermissions bool) {
cmdArgs = append(cmdArgs, "--keep-enforce-permissions")
}
- cmd := exec.Command("python", cmdArgs...)
-
- stderr, err := cmd.StderrPipe()
- if err != nil {
- log.Fatalf("Setting up stderr pipe: %s", err)
- }
- go io.Copy(os.Stderr, stderr)
- if err := cmd.Run(); err != nil {
- panic(fmt.Sprintf("'python run_test_server.py start_keep' returned error %s", err))
- }
+ bgRun(exec.Command("python", cmdArgs...))
}
// StopKeep stops keep servers that were started with StartKeep.
@@ -132,5 +108,27 @@ func StopKeep(numKeepServers int) {
defer os.Chdir(cwd)
chdirToPythonTests()
- exec.Command("python", "run_test_server.py", "stop_keep", "--num-keep-servers", strconv.Itoa(numKeepServers))
+ cmd := exec.Command("python", "run_test_server.py", "stop_keep", "--num-keep-servers", strconv.Itoa(numKeepServers))
+ cmd.Stdin = nil
+ cmd.Stderr = os.Stderr
+ cmd.Stdout = os.Stderr
+ if err := cmd.Run(); err != nil {
+ log.Fatalf("%+v: %s", cmd.Args, err)
+ }
+}
+
+// Start cmd, with stderr and stdout redirected to our own
+// stderr. Return when the process exits, but do not wait for its
+// stderr and stdout to close: any grandchild processes will continue
+// writing to our stderr.
+func bgRun(cmd *exec.Cmd) {
+ cmd.Stdin = nil
+ cmd.Stderr = os.Stderr
+ cmd.Stdout = os.Stderr
+ if err := cmd.Start(); err != nil {
+ log.Fatalf("%+v: %s", cmd.Args, err)
+ }
+ if _, err := cmd.Process.Wait(); err != nil {
+ log.Fatalf("%+v: %s", cmd.Args, err)
+ }
}
diff --git a/services/keepstore/pull_worker_integration_test.go b/services/keepstore/pull_worker_integration_test.go
index 3a3069a..52b59ec 100644
--- a/services/keepstore/pull_worker_integration_test.go
+++ b/services/keepstore/pull_worker_integration_test.go
@@ -82,6 +82,8 @@ func TestPullWorkerIntegration_GetNonExistingLocator(t *testing.T) {
}
pullRequest := SetupPullWorkerIntegrationTest(t, testData, false)
+ defer arvadostest.StopAPI()
+ defer arvadostest.StopKeep(2)
performPullWorkerIntegrationTest(testData, pullRequest, t)
}
@@ -97,6 +99,8 @@ func TestPullWorkerIntegration_GetExistingLocator(t *testing.T) {
}
pullRequest := SetupPullWorkerIntegrationTest(t, testData, true)
+ defer arvadostest.StopAPI()
+ defer arvadostest.StopKeep(2)
performPullWorkerIntegrationTest(testData, pullRequest, t)
}
diff --git a/tools/keep-rsync/keep-rsync_test.go b/tools/keep-rsync/keep-rsync_test.go
index 9432a0d..94281fa 100644
--- a/tools/keep-rsync/keep-rsync_test.go
+++ b/tools/keep-rsync/keep-rsync_test.go
@@ -470,9 +470,11 @@ func (s *DoMainTestSuite) Test_doMainWithSrcAndDstConfig(c *C) {
args := []string{"-src", srcConfig.Name(), "-dst", dstConfig.Name()}
os.Args = append(os.Args, args...)
- // Start keepservers. Since we are not doing any tweaking as in setupRsync func,
- // kcSrc and kcDst will be the same and no actual copying to dst will happen, but that's ok.
+ // Start keepservers. Since we are not doing any tweaking as
+ // in setupRsync func, kcSrc and kcDst will be the same and no
+ // actual copying to dst will happen, but that's ok.
arvadostest.StartKeep(2, false)
+ defer arvadostest.StopKeep(2)
err := doMain()
c.Check(err, IsNil)
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list