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         "git.curoverse.com/arvados.git/sdk/go/ctxlog"
  23         "github.com/prometheus/client_golang/prometheus"
  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(prometheus.NewRegistry()),
 319                 Logger:     options.Logger,
 320                 Dumper:     options.Dumper,
 321         }
 322         srv.setup()
 323         return srv
 324 }
 325
 326 func (s *runSuite) SetUpTest(c *check.C) {
 327         cfg, err := config.NewLoader(nil, ctxlog.TestLogger(c)).Load()
 328         c.Assert(err, check.Equals, nil)
 329         s.config, err = cfg.GetCluster("")
 330         c.Assert(err, check.Equals, nil)
 331
 332         s.config.Collections.BalancePeriod = arvados.Duration(time.Second)
 333         arvadostest.SetServiceURL(&s.config.Services.Keepbalance, "http://localhost:/")
 334
 335         s.client = &arvados.Client{
 336                 AuthToken: "xyzzy",
 337                 APIHost:   "zzzzz.arvadosapi.com",
 338                 Client:    s.stub.Start()}
 339
 340         s.stub.serveDiscoveryDoc()
 341         s.stub.logf = c.Logf
 342 }
 343
 344 func (s *runSuite) TearDownTest(c *check.C) {
 345         s.stub.Close()
 346 }
 347
 348 func (s *runSuite) TestRefuseZeroCollections(c *check.C) {
 349         opts := RunOptions{
 350                 CommitPulls: true,
 351                 CommitTrash: true,
 352                 Logger:      ctxlog.TestLogger(c),
 353         }
 354         s.stub.serveCurrentUserAdmin()
 355         s.stub.serveZeroCollections()
 356         s.stub.serveKeepServices(stubServices)
 357         s.stub.serveKeepstoreMounts()
 358         s.stub.serveKeepstoreIndexFoo4Bar1()
 359         trashReqs := s.stub.serveKeepstoreTrash()
 360         pullReqs := s.stub.serveKeepstorePull()
 361         srv := s.newServer(&opts)
 362         _, err := srv.runOnce()
 363         c.Check(err, check.ErrorMatches, "received zero collections")
 364         c.Check(trashReqs.Count(), check.Equals, 4)
 365         c.Check(pullReqs.Count(), check.Equals, 0)
 366 }
 367
 368 func (s *runSuite) TestRefuseNonAdmin(c *check.C) {
 369         opts := RunOptions{
 370                 CommitPulls: true,
 371                 CommitTrash: true,
 372                 Logger:      ctxlog.TestLogger(c),
 373         }
 374         s.stub.serveCurrentUserNotAdmin()
 375         s.stub.serveZeroCollections()
 376         s.stub.serveKeepServices(stubServices)
 377         s.stub.serveKeepstoreMounts()
 378         trashReqs := s.stub.serveKeepstoreTrash()
 379         pullReqs := s.stub.serveKeepstorePull()
 380         srv := s.newServer(&opts)
 381         _, err := srv.runOnce()
 382         c.Check(err, check.ErrorMatches, "current user .* is not .* admin user")
 383         c.Check(trashReqs.Count(), check.Equals, 0)
 384         c.Check(pullReqs.Count(), check.Equals, 0)
 385 }
 386
 387 func (s *runSuite) TestDetectSkippedCollections(c *check.C) {
 388         opts := RunOptions{
 389                 CommitPulls: true,
 390                 CommitTrash: true,
 391                 Logger:      ctxlog.TestLogger(c),
 392         }
 393         s.stub.serveCurrentUserAdmin()
 394         s.stub.serveCollectionsButSkipOne()
 395         s.stub.serveKeepServices(stubServices)
 396         s.stub.serveKeepstoreMounts()
 397         s.stub.serveKeepstoreIndexFoo4Bar1()
 398         trashReqs := s.stub.serveKeepstoreTrash()
 399         pullReqs := s.stub.serveKeepstorePull()
 400         srv := s.newServer(&opts)
 401         _, err := srv.runOnce()
 402         c.Check(err, check.ErrorMatches, `Retrieved 2 collections with modtime <= .* but server now reports there are 3 collections.*`)
 403         c.Check(trashReqs.Count(), check.Equals, 4)
 404         c.Check(pullReqs.Count(), check.Equals, 0)
 405 }
 406
 407 func (s *runSuite) TestWriteLostBlocks(c *check.C) {
 408         lostf, err := ioutil.TempFile("", "keep-balance-lost-blocks-test-")
 409         c.Assert(err, check.IsNil)
 410         s.config.Collections.BlobMissingReport = lostf.Name()
 411         defer os.Remove(lostf.Name())
 412         opts := RunOptions{
 413                 CommitPulls: true,
 414                 CommitTrash: true,
 415                 Logger:      ctxlog.TestLogger(c),
 416         }
 417         s.stub.serveCurrentUserAdmin()
 418         s.stub.serveFooBarFileCollections()
 419         s.stub.serveKeepServices(stubServices)
 420         s.stub.serveKeepstoreMounts()
 421         s.stub.serveKeepstoreIndexFoo1()
 422         s.stub.serveKeepstoreTrash()
 423         s.stub.serveKeepstorePull()
 424         srv := s.newServer(&opts)
 425         c.Assert(err, check.IsNil)
 426         _, err = srv.runOnce()
 427         c.Check(err, check.IsNil)
 428         lost, err := ioutil.ReadFile(lostf.Name())
 429         c.Assert(err, check.IsNil)
 430         c.Check(string(lost), check.Equals, "37b51d194a7513e45b56f6524f2d51f2 fa7aeb5140e2848d39b416daeef4ffc5+45\n")
 431 }
 432
 433 func (s *runSuite) TestDryRun(c *check.C) {
 434         opts := RunOptions{
 435                 CommitPulls: false,
 436                 CommitTrash: false,
 437                 Logger:      ctxlog.TestLogger(c),
 438         }
 439         s.stub.serveCurrentUserAdmin()
 440         collReqs := s.stub.serveFooBarFileCollections()
 441         s.stub.serveKeepServices(stubServices)
 442         s.stub.serveKeepstoreMounts()
 443         s.stub.serveKeepstoreIndexFoo4Bar1()
 444         trashReqs := s.stub.serveKeepstoreTrash()
 445         pullReqs := s.stub.serveKeepstorePull()
 446         srv := s.newServer(&opts)
 447         bal, err := srv.runOnce()
 448         c.Check(err, check.IsNil)
 449         for _, req := range collReqs.reqs {
 450                 c.Check(req.Form.Get("include_trash"), check.Equals, "true")
 451                 c.Check(req.Form.Get("include_old_versions"), check.Equals, "true")
 452         }
 453         c.Check(trashReqs.Count(), check.Equals, 0)
 454         c.Check(pullReqs.Count(), check.Equals, 0)
 455         c.Check(bal.stats.pulls, check.Not(check.Equals), 0)
 456         c.Check(bal.stats.underrep.replicas, check.Not(check.Equals), 0)
 457         c.Check(bal.stats.overrep.replicas, check.Not(check.Equals), 0)
 458 }
 459
 460 func (s *runSuite) TestCommit(c *check.C) {
 461         lostf, err := ioutil.TempFile("", "keep-balance-lost-blocks-test-")
 462         c.Assert(err, check.IsNil)
 463         s.config.Collections.BlobMissingReport = lostf.Name()
 464         defer os.Remove(lostf.Name())
 465
 466         s.config.ManagementToken = "xyzzy"
 467         opts := RunOptions{
 468                 CommitPulls: true,
 469                 CommitTrash: true,
 470                 Logger:      ctxlog.TestLogger(c),
 471                 Dumper:      ctxlog.TestLogger(c),
 472         }
 473         s.stub.serveCurrentUserAdmin()
 474         s.stub.serveFooBarFileCollections()
 475         s.stub.serveKeepServices(stubServices)
 476         s.stub.serveKeepstoreMounts()
 477         s.stub.serveKeepstoreIndexFoo4Bar1()
 478         trashReqs := s.stub.serveKeepstoreTrash()
 479         pullReqs := s.stub.serveKeepstorePull()
 480         srv := s.newServer(&opts)
 481         bal, err := srv.runOnce()
 482         c.Check(err, check.IsNil)
 483         c.Check(trashReqs.Count(), check.Equals, 8)
 484         c.Check(pullReqs.Count(), check.Equals, 4)
 485         // "foo" block is overreplicated by 2
 486         c.Check(bal.stats.trashes, check.Equals, 2)
 487         // "bar" block is underreplicated by 1, and its only copy is
 488         // in a poor rendezvous position
 489         c.Check(bal.stats.pulls, check.Equals, 2)
 490
 491         lost, err := ioutil.ReadFile(lostf.Name())
 492         c.Assert(err, check.IsNil)
 493         c.Check(string(lost), check.Equals, "")
 494
 495         metrics := s.getMetrics(c, srv)
 496         c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_total_bytes 15\n.*`)
 497         c.Check(metrics, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_sum [0-9\.]+\n.*`)
 498         c.Check(metrics, check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count 1\n.*`)
 499         c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_dedup_byte_ratio 1\.5\n.*`)
 500         c.Check(metrics, check.Matches, `(?ms).*\narvados_keep_dedup_block_ratio 1\.5\n.*`)
 501 }
 502
 503 func (s *runSuite) TestRunForever(c *check.C) {
 504         s.config.ManagementToken = "xyzzy"
 505         opts := RunOptions{
 506                 CommitPulls: true,
 507                 CommitTrash: true,
 508                 Logger:      ctxlog.TestLogger(c),
 509                 Dumper:      ctxlog.TestLogger(c),
 510         }
 511         s.stub.serveCurrentUserAdmin()
 512         s.stub.serveFooBarFileCollections()
 513         s.stub.serveKeepServices(stubServices)
 514         s.stub.serveKeepstoreMounts()
 515         s.stub.serveKeepstoreIndexFoo4Bar1()
 516         trashReqs := s.stub.serveKeepstoreTrash()
 517         pullReqs := s.stub.serveKeepstorePull()
 518
 519         stop := make(chan interface{})
 520         s.config.Collections.BalancePeriod = arvados.Duration(time.Millisecond)
 521         srv := s.newServer(&opts)
 522
 523         done := make(chan bool)
 524         go func() {
 525                 srv.runForever(stop)
 526                 close(done)
 527         }()
 528
 529         // Each run should send 4 pull lists + 4 trash lists. The
 530         // first run should also send 4 empty trash lists at
 531         // startup. We should complete all four runs in much less than
 532         // a second.
 533         for t0 := time.Now(); pullReqs.Count() < 16 && time.Since(t0) < 10*time.Second; {
 534                 time.Sleep(time.Millisecond)
 535         }
 536         stop <- true
 537         <-done
 538         c.Check(pullReqs.Count() >= 16, check.Equals, true)
 539         c.Check(trashReqs.Count(), check.Equals, pullReqs.Count()+4)
 540         c.Check(s.getMetrics(c, srv), check.Matches, `(?ms).*\narvados_keepbalance_changeset_compute_seconds_count `+fmt.Sprintf("%d", pullReqs.Count()/4)+`\n.*`)
 541 }
 542
 543 func (s *runSuite) getMetrics(c *check.C, srv *Server) string {
 544         req := httptest.NewRequest("GET", "/metrics", nil)
 545         resp := httptest.NewRecorder()
 546         srv.ServeHTTP(resp, req)
 547         c.Check(resp.Code, check.Equals, http.StatusUnauthorized)
 548
 549         req = httptest.NewRequest("GET", "/metrics?api_token=xyzzy", nil)
 550         resp = httptest.NewRecorder()
 551         srv.ServeHTTP(resp, req)
 552         c.Check(resp.Code, check.Equals, http.StatusOK)
 553
 554         buf, err := ioutil.ReadAll(resp.Body)
 555         c.Check(err, check.IsNil)
 556         return string(buf)
 557 }