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