+ if expectedBody != "" && response.Body.String() != expectedBody {
+ c.Errorf("%s: expected response body '%s', got %+v",
+ testname, expectedBody, response)
+ }
+}
+
+// See #7121
+func (s *HandlerSuite) TestPutNeedsOnlyOneBuffer(c *check.C) {
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ defer func(orig *bufferPool) {
+ bufs = orig
+ }(bufs)
+ bufs = newBufferPool(ctxlog.TestLogger(c), 1, BlockSize)
+
+ ok := make(chan struct{})
+ go func() {
+ for i := 0; i < 2; i++ {
+ response := IssueRequest(s.handler,
+ &RequestTester{
+ method: "PUT",
+ uri: "/" + TestHash,
+ requestBody: TestBlock,
+ })
+ ExpectStatusCode(c,
+ "TestPutNeedsOnlyOneBuffer", http.StatusOK, response)
+ }
+ ok <- struct{}{}
+ }()
+
+ select {
+ case <-ok:
+ case <-time.After(time.Second):
+ c.Fatal("PUT deadlocks with MaxKeepBlobBuffers==1")
+ }
+}
+
+// Invoke the PutBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func (s *HandlerSuite) TestPutHandlerNoBufferleak(c *check.C) {
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ ok := make(chan bool)
+ go func() {
+ for i := 0; i < s.cluster.API.MaxKeepBlobBuffers+1; i++ {
+ // Unauthenticated request, no server key
+ // => OK (unsigned response)
+ unsignedLocator := "/" + TestHash
+ response := IssueRequest(s.handler,
+ &RequestTester{
+ method: "PUT",
+ uri: unsignedLocator,
+ requestBody: TestBlock,
+ })
+ ExpectStatusCode(c,
+ "TestPutHandlerBufferleak", http.StatusOK, response)
+ ExpectBody(c,
+ "TestPutHandlerBufferleak",
+ TestHashPutResp, response)
+ }
+ ok <- true
+ }()
+ select {
+ case <-time.After(20 * time.Second):
+ // If the buffer pool leaks, the test goroutine hangs.
+ c.Fatal("test did not finish, assuming pool leaked")
+ case <-ok:
+ }
+}
+
+type notifyingResponseRecorder struct {
+ *httptest.ResponseRecorder
+ closer chan bool
+}
+
+func (r *notifyingResponseRecorder) CloseNotify() <-chan bool {
+ return r.closer
+}
+
+func (s *HandlerSuite) TestGetHandlerClientDisconnect(c *check.C) {
+ s.cluster.Collections.BlobSigning = false
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ defer func(orig *bufferPool) {
+ bufs = orig
+ }(bufs)
+ bufs = newBufferPool(ctxlog.TestLogger(c), 1, BlockSize)
+ defer bufs.Put(bufs.Get(BlockSize))
+
+ if err := s.handler.volmgr.AllWritable()[0].Put(context.Background(), TestHash, TestBlock); err != nil {
+ c.Error(err)
+ }
+
+ resp := ¬ifyingResponseRecorder{
+ ResponseRecorder: httptest.NewRecorder(),
+ closer: make(chan bool, 1),
+ }
+ if _, ok := http.ResponseWriter(resp).(http.CloseNotifier); !ok {
+ c.Fatal("notifyingResponseRecorder is broken")
+ }
+ // If anyone asks, the client has disconnected.
+ resp.closer <- true
+
+ ok := make(chan struct{})
+ go func() {
+ req, _ := http.NewRequest("GET", fmt.Sprintf("/%s+%d", TestHash, len(TestBlock)), nil)
+ s.handler.ServeHTTP(resp, req)
+ ok <- struct{}{}
+ }()
+
+ select {
+ case <-time.After(20 * time.Second):
+ c.Fatal("request took >20s, close notifier must be broken")
+ case <-ok:
+ }
+
+ ExpectStatusCode(c, "client disconnect", http.StatusServiceUnavailable, resp.ResponseRecorder)
+ for i, v := range s.handler.volmgr.AllWritable() {
+ if calls := v.Volume.(*MockVolume).called["GET"]; calls != 0 {
+ c.Errorf("volume %d got %d calls, expected 0", i, calls)
+ }
+ }
+}
+
+// Invoke the GetBlockHandler a bunch of times to test for bufferpool resource
+// leak.
+func (s *HandlerSuite) TestGetHandlerNoBufferLeak(c *check.C) {
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ vols := s.handler.volmgr.AllWritable()
+ if err := vols[0].Put(context.Background(), TestHash, TestBlock); err != nil {
+ c.Error(err)
+ }
+
+ ok := make(chan bool)
+ go func() {
+ for i := 0; i < s.cluster.API.MaxKeepBlobBuffers+1; i++ {
+ // Unauthenticated request, unsigned locator
+ // => OK
+ unsignedLocator := "/" + TestHash
+ response := IssueRequest(s.handler,
+ &RequestTester{
+ method: "GET",
+ uri: unsignedLocator,
+ })
+ ExpectStatusCode(c,
+ "Unauthenticated request, unsigned locator", http.StatusOK, response)
+ ExpectBody(c,
+ "Unauthenticated request, unsigned locator",
+ string(TestBlock),
+ response)
+ }
+ ok <- true
+ }()
+ select {
+ case <-time.After(20 * time.Second):
+ // If the buffer pool leaks, the test goroutine hangs.
+ c.Fatal("test did not finish, assuming pool leaked")
+ case <-ok:
+ }
+}
+
+func (s *HandlerSuite) TestPutReplicationHeader(c *check.C) {
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ resp := IssueRequest(s.handler, &RequestTester{
+ method: "PUT",
+ uri: "/" + TestHash,
+ requestBody: TestBlock,
+ })
+ if r := resp.Header().Get("X-Keep-Replicas-Stored"); r != "1" {
+ c.Logf("%#v", resp)
+ c.Errorf("Got X-Keep-Replicas-Stored: %q, expected %q", r, "1")
+ }
+}
+
+func (s *HandlerSuite) TestUntrashHandler(c *check.C) {
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ // Set up Keep volumes
+ vols := s.handler.volmgr.AllWritable()
+ vols[0].Put(context.Background(), TestHash, TestBlock)
+
+ s.cluster.SystemRootToken = "DATA MANAGER TOKEN"
+
+ // unauthenticatedReq => UnauthorizedError
+ unauthenticatedReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/" + TestHash,
+ }
+ response := IssueRequest(s.handler, unauthenticatedReq)
+ ExpectStatusCode(c,
+ "Unauthenticated request",
+ UnauthorizedError.HTTPCode,
+ response)
+
+ // notDataManagerReq => UnauthorizedError
+ notDataManagerReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/" + TestHash,
+ apiToken: knownToken,
+ }
+
+ response = IssueRequest(s.handler, notDataManagerReq)
+ ExpectStatusCode(c,
+ "Non-datamanager token",
+ UnauthorizedError.HTTPCode,
+ response)
+
+ // datamanagerWithBadHashReq => StatusBadRequest
+ datamanagerWithBadHashReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/thisisnotalocator",
+ apiToken: s.cluster.SystemRootToken,
+ }
+ response = IssueRequest(s.handler, datamanagerWithBadHashReq)
+ ExpectStatusCode(c,
+ "Bad locator in untrash request",
+ http.StatusBadRequest,
+ response)
+
+ // datamanagerWrongMethodReq => StatusBadRequest
+ datamanagerWrongMethodReq := &RequestTester{
+ method: "GET",
+ uri: "/untrash/" + TestHash,
+ apiToken: s.cluster.SystemRootToken,
+ }
+ response = IssueRequest(s.handler, datamanagerWrongMethodReq)
+ ExpectStatusCode(c,
+ "Only PUT method is supported for untrash",
+ http.StatusMethodNotAllowed,
+ response)
+
+ // datamanagerReq => StatusOK
+ datamanagerReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/" + TestHash,
+ apiToken: s.cluster.SystemRootToken,
+ }
+ response = IssueRequest(s.handler, datamanagerReq)
+ ExpectStatusCode(c,
+ "",
+ http.StatusOK,
+ response)
+ expected := "Successfully untrashed on: [MockVolume],[MockVolume]"
+ if response.Body.String() != expected {
+ c.Errorf(
+ "Untrash response mismatched: expected %s, got:\n%s",
+ expected, response.Body.String())
+ }
+}
+
+func (s *HandlerSuite) TestUntrashHandlerWithNoWritableVolumes(c *check.C) {
+ // Change all volumes to read-only
+ for uuid, v := range s.cluster.Volumes {
+ v.ReadOnly = true
+ s.cluster.Volumes[uuid] = v
+ }
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+
+ // datamanagerReq => StatusOK
+ datamanagerReq := &RequestTester{
+ method: "PUT",
+ uri: "/untrash/" + TestHash,
+ apiToken: s.cluster.SystemRootToken,
+ }
+ response := IssueRequest(s.handler, datamanagerReq)
+ ExpectStatusCode(c,
+ "No writable volumes",
+ http.StatusNotFound,
+ response)
+}
+
+func (s *HandlerSuite) TestHealthCheckPing(c *check.C) {
+ s.cluster.ManagementToken = arvadostest.ManagementToken
+ c.Assert(s.handler.setup(context.Background(), s.cluster, "", prometheus.NewRegistry(), testServiceURL), check.IsNil)
+ pingReq := &RequestTester{
+ method: "GET",
+ uri: "/_health/ping",
+ apiToken: arvadostest.ManagementToken,
+ }
+ response := IssueHealthCheckRequest(s.handler, pingReq)
+ ExpectStatusCode(c,
+ "",
+ http.StatusOK,
+ response)
+ want := `{"health":"OK"}`
+ if !strings.Contains(response.Body.String(), want) {
+ c.Errorf("expected response to include %s: got %s", want, response.Body.String())