X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/c07314d641177cfcecf4321a7b7e6771702f5916..49754950aa80cd1163a053de0f37d975c277e012:/services/keep-balance/balance_run_test.go diff --git a/services/keep-balance/balance_run_test.go b/services/keep-balance/balance_run_test.go index edc88aa262..7e2adcfedd 100644 --- a/services/keep-balance/balance_run_test.go +++ b/services/keep-balance/balance_run_test.go @@ -1,11 +1,14 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( - _ "encoding/json" + "encoding/json" "fmt" "io" "io/ioutil" - "log" "net/http" "net/http/httptest" "strings" @@ -13,7 +16,7 @@ import ( "time" "git.curoverse.com/arvados.git/sdk/go/arvados" - + "github.com/sirupsen/logrus" check "gopkg.in/check.v1" ) @@ -37,6 +40,63 @@ func (rt *reqTracker) Add(req *http.Request) int { return len(rt.reqs) } +var stubServices = []arvados.KeepService{ + { + UUID: "zzzzz-bi6l4-000000000000000", + ServiceHost: "keep0.zzzzz.arvadosapi.com", + ServicePort: 25107, + ServiceSSLFlag: false, + ServiceType: "disk", + }, + { + UUID: "zzzzz-bi6l4-000000000000001", + ServiceHost: "keep1.zzzzz.arvadosapi.com", + ServicePort: 25107, + ServiceSSLFlag: false, + ServiceType: "disk", + }, + { + UUID: "zzzzz-bi6l4-000000000000002", + ServiceHost: "keep2.zzzzz.arvadosapi.com", + ServicePort: 25107, + ServiceSSLFlag: false, + ServiceType: "disk", + }, + { + UUID: "zzzzz-bi6l4-000000000000003", + ServiceHost: "keep3.zzzzz.arvadosapi.com", + ServicePort: 25107, + ServiceSSLFlag: false, + ServiceType: "disk", + }, + { + UUID: "zzzzz-bi6l4-h0a0xwut9qa6g3a", + ServiceHost: "keep.zzzzz.arvadosapi.com", + ServicePort: 25333, + ServiceSSLFlag: true, + ServiceType: "proxy", + }, +} + +var stubMounts = map[string][]arvados.KeepMount{ + "keep0.zzzzz.arvadosapi.com:25107": {{ + UUID: "zzzzz-ivpuk-000000000000000", + DeviceID: "keep0-vol0", + }}, + "keep1.zzzzz.arvadosapi.com:25107": {{ + UUID: "zzzzz-ivpuk-100000000000000", + DeviceID: "keep1-vol0", + }}, + "keep2.zzzzz.arvadosapi.com:25107": {{ + UUID: "zzzzz-ivpuk-200000000000000", + DeviceID: "keep2-vol0", + }}, + "keep3.zzzzz.arvadosapi.com:25107": {{ + UUID: "zzzzz-ivpuk-300000000000000", + DeviceID: "keep3-vol0", + }}, +} + // stubServer is an HTTP transport that intercepts and processes all // requests using its own handlers. type stubServer struct { @@ -123,7 +183,8 @@ func (s *stubServer) serveFooBarFileCollections() *reqTracker { if strings.Contains(r.Form.Get("filters"), `modified_at`) { io.WriteString(w, `{"items_available":0,"items":[]}`) } else { - io.WriteString(w, `{"items_available":2,"items":[ + io.WriteString(w, `{"items_available":3,"items":[ + {"uuid":"zzzzz-4zz18-aaaaaaaaaaaaaaa","portable_data_hash":"fa7aeb5140e2848d39b416daeef4ffc5+45","manifest_text":". 37b51d194a7513e45b56f6524f2d51f2+3 0:3:bar\n","modified_at":"2014-02-03T17:22:54Z"}, {"uuid":"zzzzz-4zz18-ehbhgtheo8909or","portable_data_hash":"fa7aeb5140e2848d39b416daeef4ffc5+45","manifest_text":". 37b51d194a7513e45b56f6524f2d51f2+3 0:3:bar\n","modified_at":"2014-02-03T17:22:54Z"}, {"uuid":"zzzzz-4zz18-znfnqtbbv4spc3w","portable_data_hash":"1f4b0bc7583c2a7f9102c395f4ffc5e3+45","manifest_text":". acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:foo\n","modified_at":"2014-02-03T17:22:54Z"}]}`) } @@ -138,7 +199,9 @@ func (s *stubServer) serveCollectionsButSkipOne() *reqTracker { rt.Add(r) if strings.Contains(r.Form.Get("filters"), `"modified_at","\u003c="`) { io.WriteString(w, `{"items_available":3,"items":[]}`) - } else if strings.Contains(r.Form.Get("filters"), `"modified_at","\u003e="`) { + } else if strings.Contains(r.Form.Get("filters"), `"modified_at","\u003e`) { + io.WriteString(w, `{"items_available":0,"items":[]}`) + } else if strings.Contains(r.Form.Get("filters"), `"modified_at","="`) && strings.Contains(r.Form.Get("filters"), `"uuid","\u003e"`) { io.WriteString(w, `{"items_available":0,"items":[]}`) } else { io.WriteString(w, `{"items_available":2,"items":[ @@ -150,17 +213,32 @@ func (s *stubServer) serveCollectionsButSkipOne() *reqTracker { } func (s *stubServer) serveZeroKeepServices() *reqTracker { - return s.serveStatic("/arvados/v1/keep_services", - `{"items":[],"items_available":0}`) + return s.serveJSON("/arvados/v1/keep_services", arvados.KeepServiceList{}) +} + +func (s *stubServer) serveKeepServices(svcs []arvados.KeepService) *reqTracker { + return s.serveJSON("/arvados/v1/keep_services", arvados.KeepServiceList{ + ItemsAvailable: len(svcs), + Items: svcs, + }) +} + +func (s *stubServer) serveJSON(path string, resp interface{}) *reqTracker { + rt := &reqTracker{} + s.mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) { + rt.Add(r) + json.NewEncoder(w).Encode(resp) + }) + return rt } -func (s *stubServer) serveFourDiskKeepServices() *reqTracker { - return s.serveStatic("/arvados/v1/keep_services", `{"items_available":5,"items":[ - {"uuid":"zzzzz-bi6l4-000000000000000","service_host":"keep0.zzzzz.arvadosapi.com","service_port":25107,"service_ssl_flag":false,"service_type":"disk"}, - {"uuid":"zzzzz-bi6l4-000000000000001","service_host":"keep1.zzzzz.arvadosapi.com","service_port":25107,"service_ssl_flag":false,"service_type":"disk"}, - {"uuid":"zzzzz-bi6l4-000000000000002","service_host":"keep2.zzzzz.arvadosapi.com","service_port":25107,"service_ssl_flag":false,"service_type":"disk"}, - {"uuid":"zzzzz-bi6l4-000000000000003","service_host":"keep3.zzzzz.arvadosapi.com","service_port":25107,"service_ssl_flag":false,"service_type":"disk"}, - {"uuid":"zzzzz-bi6l4-h0a0xwut9qa6g3a","service_host":"keep.zzzzz.arvadosapi.com","service_port":25333,"service_ssl_flag":true,"service_type":"proxy"}]}`) +func (s *stubServer) serveKeepstoreMounts() *reqTracker { + rt := &reqTracker{} + s.mux.HandleFunc("/mounts", func(w http.ResponseWriter, r *http.Request) { + rt.Add(r) + json.NewEncoder(w).Encode(stubMounts[r.Host]) + }) + return rt } func (s *stubServer) serveKeepstoreIndexFoo4Bar1() *reqTracker { @@ -172,6 +250,21 @@ func (s *stubServer) serveKeepstoreIndexFoo4Bar1() *reqTracker { } fmt.Fprintf(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 %d\n\n", 12345678+count) }) + for _, mounts := range stubMounts { + for i, mnt := range mounts { + i := i + s.mux.HandleFunc(fmt.Sprintf("/mounts/%s/blocks", mnt.UUID), func(w http.ResponseWriter, r *http.Request) { + count := rt.Add(r) + if i == 0 && r.Host == "keep0.zzzzz.arvadosapi.com:25107" { + io.WriteString(w, "37b51d194a7513e45b56f6524f2d51f2+3 12345678\n") + } + if i == 0 { + fmt.Fprintf(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 %d\n", 12345678+count) + } + fmt.Fprintf(w, "\n") + }) + } + } return rt } @@ -189,7 +282,7 @@ type runSuite struct { } // make a log.Logger that writes to the current test's c.Log(). -func (s *runSuite) logger(c *check.C) *log.Logger { +func (s *runSuite) logger(c *check.C) *logrus.Logger { r, w := io.Pipe() go func() { buf := make([]byte, 10000) @@ -206,7 +299,9 @@ func (s *runSuite) logger(c *check.C) *log.Logger { } } }() - return log.New(w, "", log.LstdFlags) + logger := logrus.New() + logger.Out = w + return logger } func (s *runSuite) SetUpTest(c *check.C) { @@ -215,7 +310,9 @@ func (s *runSuite) SetUpTest(c *check.C) { AuthToken: "xyzzy", APIHost: "zzzzz.arvadosapi.com", Client: s.stub.Start()}, - KeepServiceTypes: []string{"disk"}} + KeepServiceTypes: []string{"disk"}, + RunPeriod: arvados.Duration(time.Second), + } s.stub.serveDiscoveryDoc() s.stub.logf = c.Logf } @@ -232,11 +329,14 @@ func (s *runSuite) TestRefuseZeroCollections(c *check.C) { } s.stub.serveCurrentUserAdmin() s.stub.serveZeroCollections() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() - _, err := (&Balancer{}).Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + _, err = srv.Run() c.Check(err, check.ErrorMatches, "received zero collections") c.Check(trashReqs.Count(), check.Equals, 4) c.Check(pullReqs.Count(), check.Equals, 0) @@ -251,10 +351,13 @@ func (s *runSuite) TestServiceTypes(c *check.C) { s.config.KeepServiceTypes = []string{"unlisted-type"} s.stub.serveCurrentUserAdmin() s.stub.serveFooBarFileCollections() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() indexReqs := s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() - _, err := (&Balancer{}).Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + _, err = srv.Run() c.Check(err, check.IsNil) c.Check(indexReqs.Count(), check.Equals, 0) c.Check(trashReqs.Count(), check.Equals, 0) @@ -268,10 +371,13 @@ func (s *runSuite) TestRefuseNonAdmin(c *check.C) { } s.stub.serveCurrentUserNotAdmin() s.stub.serveZeroCollections() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() - _, err := (&Balancer{}).Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + _, err = srv.Run() c.Check(err, check.ErrorMatches, "current user .* is not .* admin user") c.Check(trashReqs.Count(), check.Equals, 0) c.Check(pullReqs.Count(), check.Equals, 0) @@ -285,11 +391,14 @@ func (s *runSuite) TestDetectSkippedCollections(c *check.C) { } s.stub.serveCurrentUserAdmin() s.stub.serveCollectionsButSkipOne() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() - _, err := (&Balancer{}).Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + _, err = srv.Run() c.Check(err, check.ErrorMatches, `Retrieved 2 collections with modtime <= .* but server now reports there are 3 collections.*`) c.Check(trashReqs.Count(), check.Equals, 4) c.Check(pullReqs.Count(), check.Equals, 0) @@ -302,23 +411,30 @@ func (s *runSuite) TestDryRun(c *check.C) { Logger: s.logger(c), } s.stub.serveCurrentUserAdmin() - s.stub.serveFooBarFileCollections() - s.stub.serveFourDiskKeepServices() + collReqs := s.stub.serveFooBarFileCollections() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() - var bal Balancer - _, err := bal.Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + bal, err := srv.Run() c.Check(err, check.IsNil) + for _, req := range collReqs.reqs { + c.Check(req.Form.Get("include_trash"), check.Equals, "true") + c.Check(req.Form.Get("include_old_versions"), check.Equals, "true") + } c.Check(trashReqs.Count(), check.Equals, 0) c.Check(pullReqs.Count(), check.Equals, 0) - stats := bal.getStatistics() - c.Check(stats.pulls, check.Not(check.Equals), 0) - c.Check(stats.underrep.replicas, check.Not(check.Equals), 0) - c.Check(stats.overrep.replicas, check.Not(check.Equals), 0) + c.Check(bal.stats.pulls, check.Not(check.Equals), 0) + c.Check(bal.stats.underrep.replicas, check.Not(check.Equals), 0) + c.Check(bal.stats.overrep.replicas, check.Not(check.Equals), 0) } func (s *runSuite) TestCommit(c *check.C) { + s.config.Listen = ":" + s.config.ManagementToken = "xyzzy" opts := RunOptions{ CommitPulls: true, CommitTrash: true, @@ -327,24 +443,34 @@ func (s *runSuite) TestCommit(c *check.C) { } s.stub.serveCurrentUserAdmin() s.stub.serveFooBarFileCollections() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() - var bal Balancer - _, err := bal.Run(s.config, opts) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + bal, err := srv.Run() c.Check(err, check.IsNil) c.Check(trashReqs.Count(), check.Equals, 8) c.Check(pullReqs.Count(), check.Equals, 4) - stats := bal.getStatistics() // "foo" block is overreplicated by 2 - c.Check(stats.trashes, check.Equals, 2) + c.Check(bal.stats.trashes, check.Equals, 2) // "bar" block is underreplicated by 1, and its only copy is // in a poor rendezvous position - c.Check(stats.pulls, check.Equals, 2) + c.Check(bal.stats.pulls, check.Equals, 2) + + metrics := s.getMetrics(c, srv) + c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_total_bytes 15\n.*`) + c.Check(metrics, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_sum [0-9\.]+\n.*`) + c.Check(metrics, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count 1\n.*`) + c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_dedup_byte_ratio 1\.5\n.*`) + c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_dedup_block_ratio 1\.5\n.*`) } func (s *runSuite) TestRunForever(c *check.C) { + s.config.Listen = ":" + s.config.ManagementToken = "xyzzy" opts := RunOptions{ CommitPulls: true, CommitTrash: true, @@ -353,14 +479,22 @@ func (s *runSuite) TestRunForever(c *check.C) { } s.stub.serveCurrentUserAdmin() s.stub.serveFooBarFileCollections() - s.stub.serveFourDiskKeepServices() + s.stub.serveKeepServices(stubServices) + s.stub.serveKeepstoreMounts() s.stub.serveKeepstoreIndexFoo4Bar1() trashReqs := s.stub.serveKeepstoreTrash() pullReqs := s.stub.serveKeepstorePull() stop := make(chan interface{}) s.config.RunPeriod = arvados.Duration(time.Millisecond) - go RunForever(s.config, opts, stop) + srv, err := NewServer(s.config, opts) + c.Assert(err, check.IsNil) + + done := make(chan bool) + go func() { + srv.RunForever(stop) + close(done) + }() // Each run should send 4 pull lists + 4 trash lists. The // first run should also send 4 empty trash lists at @@ -370,6 +504,21 @@ func (s *runSuite) TestRunForever(c *check.C) { time.Sleep(time.Millisecond) } stop <- true + <-done c.Check(pullReqs.Count() >= 16, check.Equals, true) - c.Check(trashReqs.Count(), check.Equals, pullReqs.Count() + 4) + c.Check(trashReqs.Count(), check.Equals, pullReqs.Count()+4) + c.Check(s.getMetrics(c, srv), check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count `+fmt.Sprintf("%d", pullReqs.Count()/4)+`\n.*`) +} + +func (s *runSuite) getMetrics(c *check.C, srv *Server) string { + resp, err := http.Get("http://" + srv.listening + "/metrics") + c.Assert(err, check.IsNil) + c.Check(resp.StatusCode, check.Equals, http.StatusUnauthorized) + + resp, err = http.Get("http://" + srv.listening + "/metrics?api_token=xyzzy") + c.Assert(err, check.IsNil) + c.Check(resp.StatusCode, check.Equals, http.StatusOK) + buf, err := ioutil.ReadAll(resp.Body) + c.Check(err, check.IsNil) + return string(buf) }