Merge branch 'main' into 18842-arv-mount-disk-config
[arvados.git] / services / keep-balance / balance_run_test.go
index 3b7e2db9afd313a69e699b1d46fcf9ef5b90ea49..4772da55a2d6dddc79acff891dee1034781f7582 100644 (file)
@@ -2,21 +2,29 @@
 //
 // SPDX-License-Identifier: AGPL-3.0
 
-package main
+package keepbalance
 
 import (
+       "bytes"
+       "context"
        "encoding/json"
        "fmt"
        "io"
        "io/ioutil"
        "net/http"
        "net/http/httptest"
+       "os"
        "strings"
        "sync"
        "time"
 
-       "git.curoverse.com/arvados.git/sdk/go/arvados"
-       "github.com/sirupsen/logrus"
+       "git.arvados.org/arvados.git/lib/config"
+       "git.arvados.org/arvados.git/sdk/go/arvados"
+       "git.arvados.org/arvados.git/sdk/go/arvadostest"
+       "git.arvados.org/arvados.git/sdk/go/ctxlog"
+       "github.com/jmoiron/sqlx"
+       "github.com/prometheus/client_golang/prometheus"
+       "github.com/prometheus/common/expfmt"
        check "gopkg.in/check.v1"
 )
 
@@ -80,20 +88,24 @@ var stubServices = []arvados.KeepService{
 
 var stubMounts = map[string][]arvados.KeepMount{
        "keep0.zzzzz.arvadosapi.com:25107": {{
-               UUID:     "zzzzz-ivpuk-000000000000000",
-               DeviceID: "keep0-vol0",
+               UUID:           "zzzzz-ivpuk-000000000000000",
+               DeviceID:       "keep0-vol0",
+               StorageClasses: map[string]bool{"default": true},
        }},
        "keep1.zzzzz.arvadosapi.com:25107": {{
-               UUID:     "zzzzz-ivpuk-100000000000000",
-               DeviceID: "keep1-vol0",
+               UUID:           "zzzzz-ivpuk-100000000000000",
+               DeviceID:       "keep1-vol0",
+               StorageClasses: map[string]bool{"default": true},
        }},
        "keep2.zzzzz.arvadosapi.com:25107": {{
-               UUID:     "zzzzz-ivpuk-200000000000000",
-               DeviceID: "keep2-vol0",
+               UUID:           "zzzzz-ivpuk-200000000000000",
+               DeviceID:       "keep2-vol0",
+               StorageClasses: map[string]bool{"default": true},
        }},
        "keep3.zzzzz.arvadosapi.com:25107": {{
-               UUID:     "zzzzz-ivpuk-300000000000000",
-               DeviceID: "keep3-vol0",
+               UUID:           "zzzzz-ivpuk-300000000000000",
+               DeviceID:       "keep3-vol0",
+               StorageClasses: map[string]bool{"default": true},
        }},
 }
 
@@ -270,6 +282,28 @@ func (s *stubServer) serveKeepstoreIndexFoo4Bar1() *reqTracker {
        return rt
 }
 
+func (s *stubServer) serveKeepstoreIndexFoo1() *reqTracker {
+       rt := &reqTracker{}
+       s.mux.HandleFunc("/index/", func(w http.ResponseWriter, r *http.Request) {
+               rt.Add(r)
+               io.WriteString(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 12345678\n\n")
+       })
+       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) {
+                               rt.Add(r)
+                               if i == 0 {
+                                       io.WriteString(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 12345678\n\n")
+                               } else {
+                                       io.WriteString(w, "\n")
+                               }
+                       })
+               }
+       }
+       return rt
+}
+
 func (s *stubServer) serveKeepstoreTrash() *reqTracker {
        return s.serveStatic("/trash", `{}`)
 }
@@ -280,41 +314,40 @@ func (s *stubServer) serveKeepstorePull() *reqTracker {
 
 type runSuite struct {
        stub   stubServer
-       config Config
-}
-
-// make a log.Logger that writes to the current test's c.Log().
-func (s *runSuite) logger(c *check.C) *logrus.Logger {
-       r, w := io.Pipe()
-       go func() {
-               buf := make([]byte, 10000)
-               for {
-                       n, err := r.Read(buf)
-                       if n > 0 {
-                               if buf[n-1] == '\n' {
-                                       n--
-                               }
-                               c.Log(string(buf[:n]))
-                       }
-                       if err != nil {
-                               break
-                       }
-               }
-       }()
-       logger := logrus.New()
-       logger.Out = w
-       return logger
+       config *arvados.Cluster
+       db     *sqlx.DB
+       client *arvados.Client
+}
+
+func (s *runSuite) newServer(options *RunOptions) *Server {
+       srv := &Server{
+               Cluster:    s.config,
+               ArvClient:  s.client,
+               RunOptions: *options,
+               Metrics:    newMetrics(prometheus.NewRegistry()),
+               Logger:     options.Logger,
+               Dumper:     options.Dumper,
+               DB:         s.db,
+       }
+       return srv
 }
 
 func (s *runSuite) SetUpTest(c *check.C) {
-       s.config = Config{
-               Client: arvados.Client{
-                       AuthToken: "xyzzy",
-                       APIHost:   "zzzzz.arvadosapi.com",
-                       Client:    s.stub.Start()},
-               KeepServiceTypes: []string{"disk"},
-               RunPeriod:        arvados.Duration(time.Second),
-       }
+       cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
+       c.Assert(err, check.Equals, nil)
+       s.config, err = cfg.GetCluster("")
+       c.Assert(err, check.Equals, nil)
+       s.db, err = sqlx.Open("postgres", s.config.PostgreSQL.Connection.String())
+       c.Assert(err, check.IsNil)
+
+       s.config.Collections.BalancePeriod = arvados.Duration(time.Second)
+       arvadostest.SetServiceURL(&s.config.Services.Keepbalance, "http://localhost:/")
+
+       s.client = &arvados.Client{
+               AuthToken: "xyzzy",
+               APIHost:   "zzzzz.arvadosapi.com",
+               Client:    s.stub.Start()}
+
        s.stub.serveDiscoveryDoc()
        s.stub.logf = c.Logf
 }
@@ -324,10 +357,13 @@ func (s *runSuite) TearDownTest(c *check.C) {
 }
 
 func (s *runSuite) TestRefuseZeroCollections(c *check.C) {
+       defer arvados.NewClientFromEnv().RequestAndDecode(nil, "POST", "database/reset", nil, nil)
+       _, err := s.db.Exec(`delete from collections`)
+       c.Assert(err, check.IsNil)
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
        }
        s.stub.serveCurrentUserAdmin()
        s.stub.serveZeroCollections()
@@ -336,81 +372,89 @@ func (s *runSuite) TestRefuseZeroCollections(c *check.C) {
        s.stub.serveKeepstoreIndexFoo4Bar1()
        trashReqs := s.stub.serveKeepstoreTrash()
        pullReqs := s.stub.serveKeepstorePull()
-       srv, err := NewServer(s.config, opts)
-       c.Assert(err, check.IsNil)
-       _, err = srv.Run()
+       srv := s.newServer(&opts)
+       _, err = srv.runOnce(context.Background())
        c.Check(err, check.ErrorMatches, "received zero collections")
        c.Check(trashReqs.Count(), check.Equals, 4)
        c.Check(pullReqs.Count(), check.Equals, 0)
 }
 
-func (s *runSuite) TestServiceTypes(c *check.C) {
+func (s *runSuite) TestRefuseNonAdmin(c *check.C) {
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
        }
-       s.config.KeepServiceTypes = []string{"unlisted-type"}
-       s.stub.serveCurrentUserAdmin()
-       s.stub.serveFooBarFileCollections()
+       s.stub.serveCurrentUserNotAdmin()
+       s.stub.serveZeroCollections()
        s.stub.serveKeepServices(stubServices)
        s.stub.serveKeepstoreMounts()
-       indexReqs := s.stub.serveKeepstoreIndexFoo4Bar1()
        trashReqs := s.stub.serveKeepstoreTrash()
-       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)
+       pullReqs := s.stub.serveKeepstorePull()
+       srv := s.newServer(&opts)
+       _, err := srv.runOnce(context.Background())
+       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)
 }
 
-func (s *runSuite) TestRefuseNonAdmin(c *check.C) {
+func (s *runSuite) TestRefuseSameDeviceDifferentVolumes(c *check.C) {
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
        }
-       s.stub.serveCurrentUserNotAdmin()
+       s.stub.serveCurrentUserAdmin()
        s.stub.serveZeroCollections()
        s.stub.serveKeepServices(stubServices)
-       s.stub.serveKeepstoreMounts()
+       s.stub.mux.HandleFunc("/mounts", func(w http.ResponseWriter, r *http.Request) {
+               hostid := r.Host[:5] // "keep0.zzzzz.arvadosapi.com:25107" => "keep0"
+               json.NewEncoder(w).Encode([]arvados.KeepMount{{
+                       UUID:           "zzzzz-ivpuk-0000000000" + hostid,
+                       DeviceID:       "keep0-vol0",
+                       StorageClasses: map[string]bool{"default": true},
+               }})
+       })
        trashReqs := s.stub.serveKeepstoreTrash()
        pullReqs := s.stub.serveKeepstorePull()
-       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")
+       srv := s.newServer(&opts)
+       _, err := srv.runOnce(context.Background())
+       c.Check(err, check.ErrorMatches, "cannot continue with config errors.*")
        c.Check(trashReqs.Count(), check.Equals, 0)
        c.Check(pullReqs.Count(), check.Equals, 0)
 }
 
-func (s *runSuite) TestDetectSkippedCollections(c *check.C) {
+func (s *runSuite) TestWriteLostBlocks(c *check.C) {
+       lostf, err := ioutil.TempFile("", "keep-balance-lost-blocks-test-")
+       c.Assert(err, check.IsNil)
+       s.config.Collections.BlobMissingReport = lostf.Name()
+       defer os.Remove(lostf.Name())
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
        }
        s.stub.serveCurrentUserAdmin()
-       s.stub.serveCollectionsButSkipOne()
+       s.stub.serveFooBarFileCollections()
        s.stub.serveKeepServices(stubServices)
        s.stub.serveKeepstoreMounts()
-       s.stub.serveKeepstoreIndexFoo4Bar1()
-       trashReqs := s.stub.serveKeepstoreTrash()
-       pullReqs := s.stub.serveKeepstorePull()
-       srv, err := NewServer(s.config, opts)
+       s.stub.serveKeepstoreIndexFoo1()
+       s.stub.serveKeepstoreTrash()
+       s.stub.serveKeepstorePull()
+       srv := s.newServer(&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)
+       _, err = srv.runOnce(context.Background())
+       c.Check(err, check.IsNil)
+       lost, err := ioutil.ReadFile(lostf.Name())
+       c.Assert(err, check.IsNil)
+       c.Check(string(lost), check.Matches, `(?ms).*37b51d194a7513e45b56f6524f2d51f2.* fa7aeb5140e2848d39b416daeef4ffc5\+45.*`)
 }
 
 func (s *runSuite) TestDryRun(c *check.C) {
        opts := RunOptions{
                CommitPulls: false,
                CommitTrash: false,
-               Logger:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
        }
        s.stub.serveCurrentUserAdmin()
        collReqs := s.stub.serveFooBarFileCollections()
@@ -419,9 +463,8 @@ func (s *runSuite) TestDryRun(c *check.C) {
        s.stub.serveKeepstoreIndexFoo4Bar1()
        trashReqs := s.stub.serveKeepstoreTrash()
        pullReqs := s.stub.serveKeepstorePull()
-       srv, err := NewServer(s.config, opts)
-       c.Assert(err, check.IsNil)
-       bal, err := srv.Run()
+       srv := s.newServer(&opts)
+       bal, err := srv.runOnce(context.Background())
        c.Check(err, check.IsNil)
        for _, req := range collReqs.reqs {
                c.Check(req.Form.Get("include_trash"), check.Equals, "true")
@@ -435,13 +478,13 @@ func (s *runSuite) TestDryRun(c *check.C) {
 }
 
 func (s *runSuite) TestCommit(c *check.C) {
-       s.config.Listen = ":"
+       s.config.Collections.BlobMissingReport = c.MkDir() + "/keep-balance-lost-blocks-test-"
        s.config.ManagementToken = "xyzzy"
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
-               Dumper:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
+               Dumper:      ctxlog.TestLogger(c),
        }
        s.stub.serveCurrentUserAdmin()
        s.stub.serveFooBarFileCollections()
@@ -450,9 +493,8 @@ func (s *runSuite) TestCommit(c *check.C) {
        s.stub.serveKeepstoreIndexFoo4Bar1()
        trashReqs := s.stub.serveKeepstoreTrash()
        pullReqs := s.stub.serveKeepstorePull()
-       srv, err := NewServer(s.config, opts)
-       c.Assert(err, check.IsNil)
-       bal, err := srv.Run()
+       srv := s.newServer(&opts)
+       bal, err := srv.runOnce(context.Background())
        c.Check(err, check.IsNil)
        c.Check(trashReqs.Count(), check.Equals, 8)
        c.Check(pullReqs.Count(), check.Equals, 4)
@@ -462,22 +504,27 @@ func (s *runSuite) TestCommit(c *check.C) {
        // in a poor rendezvous position
        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.*`)
+       lost, err := ioutil.ReadFile(s.config.Collections.BlobMissingReport)
+       c.Assert(err, check.IsNil)
+       c.Check(string(lost), check.Not(check.Matches), `(?ms).*acbd18db4cc2f85cedef654fccc4a4d8.*`)
+
+       buf, err := s.getMetrics(c, srv)
+       c.Check(err, check.IsNil)
+       bufstr := buf.String()
+       c.Check(bufstr, check.Matches, `(?ms).*\narvados_keep_total_bytes 15\n.*`)
+       c.Check(bufstr, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_sum [0-9\.]+\n.*`)
+       c.Check(bufstr, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count 1\n.*`)
+       c.Check(bufstr, check.Matches, `(?ms).*\narvados_keep_dedup_byte_ratio [1-9].*`)
+       c.Check(bufstr, check.Matches, `(?ms).*\narvados_keep_dedup_block_ratio [1-9].*`)
 }
 
 func (s *runSuite) TestRunForever(c *check.C) {
-       s.config.Listen = ":"
        s.config.ManagementToken = "xyzzy"
        opts := RunOptions{
                CommitPulls: true,
                CommitTrash: true,
-               Logger:      s.logger(c),
-               Dumper:      s.logger(c),
+               Logger:      ctxlog.TestLogger(c),
+               Dumper:      ctxlog.TestLogger(c),
        }
        s.stub.serveCurrentUserAdmin()
        s.stub.serveFooBarFileCollections()
@@ -487,14 +534,14 @@ func (s *runSuite) TestRunForever(c *check.C) {
        trashReqs := s.stub.serveKeepstoreTrash()
        pullReqs := s.stub.serveKeepstorePull()
 
-       stop := make(chan interface{})
-       s.config.RunPeriod = arvados.Duration(time.Millisecond)
-       srv, err := NewServer(s.config, opts)
-       c.Assert(err, check.IsNil)
+       ctx, cancel := context.WithCancel(context.Background())
+       defer cancel()
+       s.config.Collections.BalancePeriod = arvados.Duration(time.Millisecond)
+       srv := s.newServer(&opts)
 
        done := make(chan bool)
        go func() {
-               srv.RunForever(stop)
+               srv.runForever(ctx)
                close(done)
        }()
 
@@ -505,22 +552,28 @@ func (s *runSuite) TestRunForever(c *check.C) {
        for t0 := time.Now(); pullReqs.Count() < 16 && time.Since(t0) < 10*time.Second; {
                time.Sleep(time.Millisecond)
        }
-       stop <- true
+       cancel()
        <-done
        c.Check(pullReqs.Count() >= 16, check.Equals, true)
        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.*`)
+
+       buf, err := s.getMetrics(c, srv)
+       c.Check(err, check.IsNil)
+       c.Check(buf, 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)
+func (s *runSuite) getMetrics(c *check.C, srv *Server) (*bytes.Buffer, error) {
+       mfs, err := srv.Metrics.reg.Gather()
+       if err != nil {
+               return nil, err
+       }
 
-       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)
+       var buf bytes.Buffer
+       for _, mf := range mfs {
+               if _, err := expfmt.MetricFamilyToText(&buf, mf); err != nil {
+                       return nil, err
+               }
+       }
+
+       return &buf, nil
 }