services/keep-balance/balance_run_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bytes"
   9         "encoding/json"
  10         "fmt"
  11         "io"
  12         "io/ioutil"
  13         "net/http"
  14         "net/http/httptest"
  15         "os"
  16         "strings"
  17         "sync"
  18         "time"
  19
  20         "git.arvados.org/arvados.git/lib/config"
  21         "git.arvados.org/arvados.git/sdk/go/arvados"
  22         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  23         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  24         "github.com/prometheus/client_golang/prometheus"
  25         "github.com/prometheus/common/expfmt"
  26         check "gopkg.in/check.v1"
  27 )
  28
  29 var _ = check.Suite(&runSuite{})
  30
  31 type reqTracker struct {
  32         reqs []http.Request
  33         sync.Mutex
  34 }
  35
  36 func (rt *reqTracker) Count() int {
  37         rt.Lock()
  38         defer rt.Unlock()
  39         return len(rt.reqs)
  40 }
  41
  42 func (rt *reqTracker) Add(req *http.Request) int {
  43         rt.Lock()
  44         defer rt.Unlock()
  45         rt.reqs = append(rt.reqs, *req)
  46         return len(rt.reqs)
  47 }
  48
  49 var stubServices = []arvados.KeepService{
  50         {
  51                 UUID:           "zzzzz-bi6l4-000000000000000",
  52                 ServiceHost:    "keep0.zzzzz.arvadosapi.com",
  53                 ServicePort:    25107,
  54                 ServiceSSLFlag: false,
  55                 ServiceType:    "disk",
  56         },
  57         {
  58                 UUID:           "zzzzz-bi6l4-000000000000001",
  59                 ServiceHost:    "keep1.zzzzz.arvadosapi.com",
  60                 ServicePort:    25107,
  61                 ServiceSSLFlag: false,
  62                 ServiceType:    "disk",
  63         },
  64         {
  65                 UUID:           "zzzzz-bi6l4-000000000000002",
  66                 ServiceHost:    "keep2.zzzzz.arvadosapi.com",
  67                 ServicePort:    25107,
  68                 ServiceSSLFlag: false,
  69                 ServiceType:    "disk",
  70         },
  71         {
  72                 UUID:           "zzzzz-bi6l4-000000000000003",
  73                 ServiceHost:    "keep3.zzzzz.arvadosapi.com",
  74                 ServicePort:    25107,
  75                 ServiceSSLFlag: false,
  76                 ServiceType:    "disk",
  77         },
  78         {
  79                 UUID:           "zzzzz-bi6l4-h0a0xwut9qa6g3a",
  80                 ServiceHost:    "keep.zzzzz.arvadosapi.com",
  81                 ServicePort:    25333,
  82                 ServiceSSLFlag: true,
  83                 ServiceType:    "proxy",
  84         },
  85 }
  86
  87 var stubMounts = map[string][]arvados.KeepMount{
  88         "keep0.zzzzz.arvadosapi.com:25107": {{
  89                 UUID:     "zzzzz-ivpuk-000000000000000",
  90                 DeviceID: "keep0-vol0",
  91         }},
  92         "keep1.zzzzz.arvadosapi.com:25107": {{
  93                 UUID:     "zzzzz-ivpuk-100000000000000",
  94                 DeviceID: "keep1-vol0",
  95         }},
  96         "keep2.zzzzz.arvadosapi.com:25107": {{
  97                 UUID:     "zzzzz-ivpuk-200000000000000",
  98                 DeviceID: "keep2-vol0",
  99         }},
 100         "keep3.zzzzz.arvadosapi.com:25107": {{
 101                 UUID:     "zzzzz-ivpuk-300000000000000",
 102                 DeviceID: "keep3-vol0",
 103         }},
 104 }
 105
 106 // stubServer is an HTTP transport that intercepts and processes all
 107 // requests using its own handlers.
 108 type stubServer struct {
 109         mux      *http.ServeMux
 110         srv      *httptest.Server
 111         mutex    sync.Mutex
 112         Requests reqTracker
 113         logf     func(string, ...interface{})
 114 }
 115
 116 // Start initializes the stub server and returns an *http.Client that
 117 // uses the stub server to handle all requests.
 118 //
 119 // A stubServer that has been started should eventually be shut down
 120 // with Close().
 121 func (s *stubServer) Start() *http.Client {
 122         // Set up a config.Client that forwards all requests to s.mux
 123         // via s.srv. Test cases will attach handlers to s.mux to get
 124         // the desired responses.
 125         s.mux = http.NewServeMux()
 126         s.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 127                 s.mutex.Lock()
 128                 s.Requests.Add(r)
 129                 s.mutex.Unlock()
 130                 w.Header().Set("Content-Type", "application/json")
 131                 s.mux.ServeHTTP(w, r)
 132         }))
 133         return &http.Client{Transport: s}
 134 }
 135
 136 func (s *stubServer) RoundTrip(req *http.Request) (*http.Response, error) {
 137         w := httptest.NewRecorder()
 138         s.mux.ServeHTTP(w, req)
 139         return &http.Response{
 140                 StatusCode: w.Code,
 141                 Status:     fmt.Sprintf("%d %s", w.Code, http.StatusText(w.Code)),
 142                 Header:     w.HeaderMap,
 143                 Body:       ioutil.NopCloser(w.Body)}, nil
 144 }
 145
 146 // Close releases resources used by the server.
 147 func (s *stubServer) Close() {
 148         s.srv.Close()
 149 }
 150
 151 func (s *stubServer) serveStatic(path, data string) *reqTracker {
 152         rt := &reqTracker{}
 153         s.mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
 154                 rt.Add(r)
 155                 if r.Body != nil {
 156                         ioutil.ReadAll(r.Body)
 157                         r.Body.Close()
 158                 }
 159                 io.WriteString(w, data)
 160         })
 161         return rt
 162 }
 163
 164 func (s *stubServer) serveCurrentUserAdmin() *reqTracker {
 165         return s.serveStatic("/arvados/v1/users/current",
 166                 `{"uuid":"zzzzz-tpzed-000000000000000","is_admin":true,"is_active":true}`)
 167 }
 168
 169 func (s *stubServer) serveCurrentUserNotAdmin() *reqTracker {
 170         return s.serveStatic("/arvados/v1/users/current",
 171                 `{"uuid":"zzzzz-tpzed-000000000000000","is_admin":false,"is_active":true}`)
 172 }
 173
 174 func (s *stubServer) serveDiscoveryDoc() *reqTracker {
 175         return s.serveStatic("/discovery/v1/apis/arvados/v1/rest",
 176                 `{"defaultCollectionReplication":2}`)
 177 }
 178
 179 func (s *stubServer) serveZeroCollections() *reqTracker {
 180         return s.serveStatic("/arvados/v1/collections",
 181                 `{"items":[],"items_available":0}`)
 182 }
 183
 184 func (s *stubServer) serveFooBarFileCollections() *reqTracker {
 185         rt := &reqTracker{}
 186         s.mux.HandleFunc("/arvados/v1/collections", func(w http.ResponseWriter, r *http.Request) {
 187                 r.ParseForm()
 188                 rt.Add(r)
 189                 if strings.Contains(r.Form.Get("filters"), `modified_at`) {
 190                         io.WriteString(w, `{"items_available":0,"items":[]}`)
 191                 } else {
 192                         io.WriteString(w, `{"items_available":3,"items":[
 193                                 {"uuid":"zzzzz-4zz18-aaaaaaaaaaaaaaa","portable_data_hash":"fa7aeb5140e2848d39b416daeef4ffc5+45","manifest_text":". 37b51d194a7513e45b56f6524f2d51f2+3 0:3:bar\n","modified_at":"2014-02-03T17:22:54Z"},
 194                                 {"uuid":"zzzzz-4zz18-ehbhgtheo8909or","portable_data_hash":"fa7aeb5140e2848d39b416daeef4ffc5+45","manifest_text":". 37b51d194a7513e45b56f6524f2d51f2+3 0:3:bar\n","modified_at":"2014-02-03T17:22:54Z"},
 195                                 {"uuid":"zzzzz-4zz18-znfnqtbbv4spc3w","portable_data_hash":"1f4b0bc7583c2a7f9102c395f4ffc5e3+45","manifest_text":". acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:foo\n","modified_at":"2014-02-03T17:22:54Z"}]}`)
 196                 }
 197         })
 198         return rt
 199 }
 200
 201 func (s *stubServer) serveCollectionsButSkipOne() *reqTracker {
 202         rt := &reqTracker{}
 203         s.mux.HandleFunc("/arvados/v1/collections", func(w http.ResponseWriter, r *http.Request) {
 204                 r.ParseForm()
 205                 rt.Add(r)
 206                 if strings.Contains(r.Form.Get("filters"), `"modified_at","\u003c="`) {
 207                         io.WriteString(w, `{"items_available":3,"items":[]}`)
 208                 } else if strings.Contains(r.Form.Get("filters"), `"modified_at","\u003e`) {
 209                         io.WriteString(w, `{"items_available":0,"items":[]}`)
 210                 } else if strings.Contains(r.Form.Get("filters"), `"modified_at","="`) && strings.Contains(r.Form.Get("filters"), `"uuid","\u003e"`) {
 211                         io.WriteString(w, `{"items_available":0,"items":[]}`)
 212                 } else if strings.Contains(r.Form.Get("filters"), `"modified_at","=",null`) {
 213                         io.WriteString(w, `{"items_available":0,"items":[]}`)
 214                 } else {
 215                         io.WriteString(w, `{"items_available":2,"items":[
 216                                 {"uuid":"zzzzz-4zz18-ehbhgtheo8909or","portable_data_hash":"fa7aeb5140e2848d39b416daeef4ffc5+45","manifest_text":". 37b51d194a7513e45b56f6524f2d51f2+3 0:3:bar\n","modified_at":"2014-02-03T17:22:54Z"},
 217                                 {"uuid":"zzzzz-4zz18-znfnqtbbv4spc3w","portable_data_hash":"1f4b0bc7583c2a7f9102c395f4ffc5e3+45","manifest_text":". acbd18db4cc2f85cedef654fccc4a4d8+3 0:3:foo\n","modified_at":"2014-02-03T17:22:54Z"}]}`)
 218                 }
 219         })
 220         return rt
 221 }
 222
 223 func (s *stubServer) serveZeroKeepServices() *reqTracker {
 224         return s.serveJSON("/arvados/v1/keep_services", arvados.KeepServiceList{})
 225 }
 226
 227 func (s *stubServer) serveKeepServices(svcs []arvados.KeepService) *reqTracker {
 228         return s.serveJSON("/arvados/v1/keep_services", arvados.KeepServiceList{
 229                 ItemsAvailable: len(svcs),
 230                 Items:          svcs,
 231         })
 232 }
 233
 234 func (s *stubServer) serveJSON(path string, resp interface{}) *reqTracker {
 235         rt := &reqTracker{}
 236         s.mux.HandleFunc(path, func(w http.ResponseWriter, r *http.Request) {
 237                 rt.Add(r)
 238                 json.NewEncoder(w).Encode(resp)
 239         })
 240         return rt
 241 }
 242
 243 func (s *stubServer) serveKeepstoreMounts() *reqTracker {
 244         rt := &reqTracker{}
 245         s.mux.HandleFunc("/mounts", func(w http.ResponseWriter, r *http.Request) {
 246                 rt.Add(r)
 247                 json.NewEncoder(w).Encode(stubMounts[r.Host])
 248         })
 249         return rt
 250 }
 251
 252 func (s *stubServer) serveKeepstoreIndexFoo4Bar1() *reqTracker {
 253         rt := &reqTracker{}
 254         s.mux.HandleFunc("/index/", func(w http.ResponseWriter, r *http.Request) {
 255                 count := rt.Add(r)
 256                 if r.Host == "keep0.zzzzz.arvadosapi.com:25107" {
 257                         io.WriteString(w, "37b51d194a7513e45b56f6524f2d51f2+3 12345678\n")
 258                 }
 259                 fmt.Fprintf(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 %d\n\n", 12345678+count)
 260         })
 261         for _, mounts := range stubMounts {
 262                 for i, mnt := range mounts {
 263                         i := i
 264                         s.mux.HandleFunc(fmt.Sprintf("/mounts/%s/blocks", mnt.UUID), func(w http.ResponseWriter, r *http.Request) {
 265                                 count := rt.Add(r)
 266                                 if i == 0 && r.Host == "keep0.zzzzz.arvadosapi.com:25107" {
 267                                         io.WriteString(w, "37b51d194a7513e45b56f6524f2d51f2+3 12345678\n")
 268                                 }
 269                                 if i == 0 {
 270                                         fmt.Fprintf(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 %d\n", 12345678+count)
 271                                 }
 272                                 fmt.Fprintf(w, "\n")
 273                         })
 274                 }
 275         }
 276         return rt
 277 }
 278
 279 func (s *stubServer) serveKeepstoreIndexFoo1() *reqTracker {
 280         rt := &reqTracker{}
 281         s.mux.HandleFunc("/index/", func(w http.ResponseWriter, r *http.Request) {
 282                 rt.Add(r)
 283                 io.WriteString(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 12345678\n\n")
 284         })
 285         for _, mounts := range stubMounts {
 286                 for i, mnt := range mounts {
 287                         i := i
 288                         s.mux.HandleFunc(fmt.Sprintf("/mounts/%s/blocks", mnt.UUID), func(w http.ResponseWriter, r *http.Request) {
 289                                 rt.Add(r)
 290                                 if i == 0 {
 291                                         io.WriteString(w, "acbd18db4cc2f85cedef654fccc4a4d8+3 12345678\n\n")
 292                                 } else {
 293                                         io.WriteString(w, "\n")
 294                                 }
 295                         })
 296                 }
 297         }
 298         return rt
 299 }
 300
 301 func (s *stubServer) serveKeepstoreTrash() *reqTracker {
 302         return s.serveStatic("/trash", `{}`)
 303 }
 304
 305 func (s *stubServer) serveKeepstorePull() *reqTracker {
 306         return s.serveStatic("/pull", `{}`)
 307 }
 308
 309 type runSuite struct {
 310         stub   stubServer
 311         config *arvados.Cluster
 312         client *arvados.Client
 313 }
 314
 315 func (s *runSuite) newServer(options *RunOptions) *Server {
 316         srv := &Server{
 317                 Cluster:    s.config,
 318                 ArvClient:  s.client,
 319                 RunOptions: *options,
 320                 Metrics:    newMetrics(prometheus.NewRegistry()),
 321                 Logger:     options.Logger,
 322                 Dumper:     options.Dumper,
 323         }
 324         return srv
 325 }
 326
 327 func (s *runSuite) SetUpTest(c *check.C) {
 328         cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
 329         c.Assert(err, check.Equals, nil)
 330         s.config, err = cfg.GetCluster("")
 331         c.Assert(err, check.Equals, nil)
 332
 333         s.config.Collections.BalancePeriod = arvados.Duration(time.Second)
 334         arvadostest.SetServiceURL(&s.config.Services.Keepbalance, "http://localhost:/")
 335
 336         s.client = &arvados.Client{
 337                 AuthToken: "xyzzy",
 338                 APIHost:   "zzzzz.arvadosapi.com",
 339                 Client:    s.stub.Start()}
 340
 341         s.stub.serveDiscoveryDoc()
 342         s.stub.logf = c.Logf
 343 }
 344
 345 func (s *runSuite) TearDownTest(c *check.C) {
 346         s.stub.Close()
 347 }
 348
 349 func (s *runSuite) TestRefuseZeroCollections(c *check.C) {
 350         opts := RunOptions{
 351                 CommitPulls: true,
 352                 CommitTrash: true,
 353                 Logger:      ctxlog.TestLogger(c),
 354         }
 355         s.stub.serveCurrentUserAdmin()
 356         s.stub.serveZeroCollections()
 357         s.stub.serveKeepServices(stubServices)
 358         s.stub.serveKeepstoreMounts()
 359         s.stub.serveKeepstoreIndexFoo4Bar1()
 360         trashReqs := s.stub.serveKeepstoreTrash()
 361         pullReqs := s.stub.serveKeepstorePull()
 362         srv := s.newServer(&opts)
 363         _, err := srv.runOnce()
 364         c.Check(err, check.ErrorMatches, "received zero collections")
 365         c.Check(trashReqs.Count(), check.Equals, 4)
 366         c.Check(pullReqs.Count(), check.Equals, 0)
 367 }
 368
 369 func (s *runSuite) TestRefuseNonAdmin(c *check.C) {
 370         opts := RunOptions{
 371                 CommitPulls: true,
 372                 CommitTrash: true,
 373                 Logger:      ctxlog.TestLogger(c),
 374         }
 375         s.stub.serveCurrentUserNotAdmin()
 376         s.stub.serveZeroCollections()
 377         s.stub.serveKeepServices(stubServices)
 378         s.stub.serveKeepstoreMounts()
 379         trashReqs := s.stub.serveKeepstoreTrash()
 380         pullReqs := s.stub.serveKeepstorePull()
 381         srv := s.newServer(&opts)
 382         _, err := srv.runOnce()
 383         c.Check(err, check.ErrorMatches, "current user .* is not .* admin user")
 384         c.Check(trashReqs.Count(), check.Equals, 0)
 385         c.Check(pullReqs.Count(), check.Equals, 0)
 386 }
 387
 388 func (s *runSuite) TestDetectSkippedCollections(c *check.C) {
 389         opts := RunOptions{
 390                 CommitPulls: true,
 391                 CommitTrash: true,
 392                 Logger:      ctxlog.TestLogger(c),
 393         }
 394         s.stub.serveCurrentUserAdmin()
 395         s.stub.serveCollectionsButSkipOne()
 396         s.stub.serveKeepServices(stubServices)
 397         s.stub.serveKeepstoreMounts()
 398         s.stub.serveKeepstoreIndexFoo4Bar1()
 399         trashReqs := s.stub.serveKeepstoreTrash()
 400         pullReqs := s.stub.serveKeepstorePull()
 401         srv := s.newServer(&opts)
 402         _, err := srv.runOnce()
 403         c.Check(err, check.ErrorMatches, `Retrieved 2 collections with modtime <= .* but server now reports there are 3 collections.*`)
 404         c.Check(trashReqs.Count(), check.Equals, 4)
 405         c.Check(pullReqs.Count(), check.Equals, 0)
 406 }
 407
 408 func (s *runSuite) TestWriteLostBlocks(c *check.C) {
 409         lostf, err := ioutil.TempFile("", "keep-balance-lost-blocks-test-")
 410         c.Assert(err, check.IsNil)
 411         s.config.Collections.BlobMissingReport = lostf.Name()
 412         defer os.Remove(lostf.Name())
 413         opts := RunOptions{
 414                 CommitPulls: true,
 415                 CommitTrash: true,
 416                 Logger:      ctxlog.TestLogger(c),
 417         }
 418         s.stub.serveCurrentUserAdmin()
 419         s.stub.serveFooBarFileCollections()
 420         s.stub.serveKeepServices(stubServices)
 421         s.stub.serveKeepstoreMounts()
 422         s.stub.serveKeepstoreIndexFoo1()
 423         s.stub.serveKeepstoreTrash()
 424         s.stub.serveKeepstorePull()
 425         srv := s.newServer(&opts)
 426         c.Assert(err, check.IsNil)
 427         _, err = srv.runOnce()
 428         c.Check(err, check.IsNil)
 429         lost, err := ioutil.ReadFile(lostf.Name())
 430         c.Assert(err, check.IsNil)
 431         c.Check(string(lost), check.Equals, "37b51d194a7513e45b56f6524f2d51f2 fa7aeb5140e2848d39b416daeef4ffc5+45\n")
 432 }
 433
 434 func (s *runSuite) TestDryRun(c *check.C) {
 435         opts := RunOptions{
 436                 CommitPulls: false,
 437                 CommitTrash: false,
 438                 Logger:      ctxlog.TestLogger(c),
 439         }
 440         s.stub.serveCurrentUserAdmin()
 441         collReqs := s.stub.serveFooBarFileCollections()
 442         s.stub.serveKeepServices(stubServices)
 443         s.stub.serveKeepstoreMounts()
 444         s.stub.serveKeepstoreIndexFoo4Bar1()
 445         trashReqs := s.stub.serveKeepstoreTrash()
 446         pullReqs := s.stub.serveKeepstorePull()
 447         srv := s.newServer(&opts)
 448         bal, err := srv.runOnce()
 449         c.Check(err, check.IsNil)
 450         for _, req := range collReqs.reqs {
 451                 c.Check(req.Form.Get("include_trash"), check.Equals, "true")
 452                 c.Check(req.Form.Get("include_old_versions"), check.Equals, "true")
 453         }
 454         c.Check(trashReqs.Count(), check.Equals, 0)
 455         c.Check(pullReqs.Count(), check.Equals, 0)
 456         c.Check(bal.stats.pulls, check.Not(check.Equals), 0)
 457         c.Check(bal.stats.underrep.replicas, check.Not(check.Equals), 0)
 458         c.Check(bal.stats.overrep.replicas, check.Not(check.Equals), 0)
 459 }
 460
 461 func (s *runSuite) TestCommit(c *check.C) {
 462         lostf, err := ioutil.TempFile("", "keep-balance-lost-blocks-test-")
 463         c.Assert(err, check.IsNil)
 464         s.config.Collections.BlobMissingReport = lostf.Name()
 465         defer os.Remove(lostf.Name())
 466
 467         s.config.ManagementToken = "xyzzy"
 468         opts := RunOptions{
 469                 CommitPulls: true,
 470                 CommitTrash: true,
 471                 Logger:      ctxlog.TestLogger(c),
 472                 Dumper:      ctxlog.TestLogger(c),
 473         }
 474         s.stub.serveCurrentUserAdmin()
 475         s.stub.serveFooBarFileCollections()
 476         s.stub.serveKeepServices(stubServices)
 477         s.stub.serveKeepstoreMounts()
 478         s.stub.serveKeepstoreIndexFoo4Bar1()
 479         trashReqs := s.stub.serveKeepstoreTrash()
 480         pullReqs := s.stub.serveKeepstorePull()
 481         srv := s.newServer(&opts)
 482         bal, err := srv.runOnce()
 483         c.Check(err, check.IsNil)
 484         c.Check(trashReqs.Count(), check.Equals, 8)
 485         c.Check(pullReqs.Count(), check.Equals, 4)
 486         // "foo" block is overreplicated by 2
 487         c.Check(bal.stats.trashes, check.Equals, 2)
 488         // "bar" block is underreplicated by 1, and its only copy is
 489         // in a poor rendezvous position
 490         c.Check(bal.stats.pulls, check.Equals, 2)
 491
 492         lost, err := ioutil.ReadFile(lostf.Name())
 493         c.Assert(err, check.IsNil)
 494         c.Check(string(lost), check.Equals, "")
 495
 496         buf, err := s.getMetrics(c, srv)
 497         c.Check(err, check.IsNil)
 498         c.Check(buf, check.Matches, `(?ms).*\narvados_keep_total_bytes 15\n.*`)
 499         c.Check(buf, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_sum [0-9\.]+\n.*`)
 500         c.Check(buf, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count 1\n.*`)
 501         c.Check(buf, check.Matches, `(?ms).*\narvados_keep_dedup_byte_ratio 1\.5\n.*`)
 502         c.Check(buf, check.Matches, `(?ms).*\narvados_keep_dedup_block_ratio 1\.5\n.*`)
 503 }
 504
 505 func (s *runSuite) TestRunForever(c *check.C) {
 506         s.config.ManagementToken = "xyzzy"
 507         opts := RunOptions{
 508                 CommitPulls: true,
 509                 CommitTrash: true,
 510                 Logger:      ctxlog.TestLogger(c),
 511                 Dumper:      ctxlog.TestLogger(c),
 512         }
 513         s.stub.serveCurrentUserAdmin()
 514         s.stub.serveFooBarFileCollections()
 515         s.stub.serveKeepServices(stubServices)
 516         s.stub.serveKeepstoreMounts()
 517         s.stub.serveKeepstoreIndexFoo4Bar1()
 518         trashReqs := s.stub.serveKeepstoreTrash()
 519         pullReqs := s.stub.serveKeepstorePull()
 520
 521         stop := make(chan interface{})
 522         s.config.Collections.BalancePeriod = arvados.Duration(time.Millisecond)
 523         srv := s.newServer(&opts)
 524
 525         done := make(chan bool)
 526         go func() {
 527                 srv.runForever(stop)
 528                 close(done)
 529         }()
 530
 531         // Each run should send 4 pull lists + 4 trash lists. The
 532         // first run should also send 4 empty trash lists at
 533         // startup. We should complete all four runs in much less than
 534         // a second.
 535         for t0 := time.Now(); pullReqs.Count() < 16 && time.Since(t0) < 10*time.Second; {
 536                 time.Sleep(time.Millisecond)
 537         }
 538         stop <- true
 539         <-done
 540         c.Check(pullReqs.Count() >= 16, check.Equals, true)
 541         c.Check(trashReqs.Count(), check.Equals, pullReqs.Count()+4)
 542
 543         buf, err := s.getMetrics(c, srv)
 544         c.Check(err, check.IsNil)
 545         c.Check(buf, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count `+fmt.Sprintf("%d", pullReqs.Count()/4)+`\n.*`)
 546 }
 547
 548 func (s *runSuite) getMetrics(c *check.C, srv *Server) (*bytes.Buffer, error) {
 549         mfs, err := srv.Metrics.reg.Gather()
 550         if err != nil {
 551                 return nil, err
 552         }
 553
 554         var buf bytes.Buffer
 555         for _, mf := range mfs {
 556                 if _, err := expfmt.MetricFamilyToText(&buf, mf); err != nil {
 557                         return nil, err
 558                 }
 559         }
 560
 561         return &buf, nil
 562 }