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