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