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