sdk/go/health/aggregator_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package health
   6
   7 import (
   8         "bytes"
   9         "crypto/sha256"
  10         "encoding/json"
  11         "fmt"
  12         "io/ioutil"
  13         "net/http"
  14         "net/http/httptest"
  15         "regexp"
  16         "runtime"
  17         "strings"
  18         "time"
  19
  20         "git.arvados.org/arvados.git/lib/cmd"
  21         "git.arvados.org/arvados.git/lib/config"
  22         "git.arvados.org/arvados.git/sdk/go/arvados"
  23         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  24         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  25         "github.com/ghodss/yaml"
  26         "gopkg.in/check.v1"
  27 )
  28
  29 type AggregatorSuite struct {
  30         handler *Aggregator
  31         req     *http.Request
  32         resp    *httptest.ResponseRecorder
  33 }
  34
  35 // Gocheck boilerplate
  36 var _ = check.Suite(&AggregatorSuite{})
  37
  38 func (s *AggregatorSuite) TestInterface(c *check.C) {
  39         var _ http.Handler = &Aggregator{}
  40 }
  41
  42 func (s *AggregatorSuite) SetUpTest(c *check.C) {
  43         ldr := config.NewLoader(bytes.NewBufferString(`Clusters: {zzzzz: {}}`), ctxlog.TestLogger(c))
  44         ldr.Path = "-"
  45         cfg, err := ldr.Load()
  46         c.Assert(err, check.IsNil)
  47         cluster, err := cfg.GetCluster("")
  48         c.Assert(err, check.IsNil)
  49         cluster.ManagementToken = arvadostest.ManagementToken
  50         cluster.SystemRootToken = arvadostest.SystemRootToken
  51         cluster.Collections.BlobSigningKey = arvadostest.BlobSigningKey
  52         cluster.Volumes["z"] = arvados.Volume{StorageClasses: map[string]bool{"default": true}}
  53         s.handler = &Aggregator{Cluster: cluster}
  54         s.req = httptest.NewRequest("GET", "/_health/all", nil)
  55         s.req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
  56         s.resp = httptest.NewRecorder()
  57 }
  58
  59 func (s *AggregatorSuite) TestNoAuth(c *check.C) {
  60         s.req.Header.Del("Authorization")
  61         s.handler.ServeHTTP(s.resp, s.req)
  62         s.checkError(c)
  63         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  64 }
  65
  66 func (s *AggregatorSuite) TestBadAuth(c *check.C) {
  67         s.req.Header.Set("Authorization", "xyzzy")
  68         s.handler.ServeHTTP(s.resp, s.req)
  69         s.checkError(c)
  70         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  71 }
  72
  73 func (s *AggregatorSuite) TestNoServicesConfigured(c *check.C) {
  74         s.handler.ServeHTTP(s.resp, s.req)
  75         s.checkUnhealthy(c)
  76 }
  77
  78 func (s *AggregatorSuite) stubServer(handler http.Handler) (*httptest.Server, string) {
  79         srv := httptest.NewServer(handler)
  80         var port string
  81         if parts := strings.Split(srv.URL, ":"); len(parts) < 3 {
  82                 panic(srv.URL)
  83         } else {
  84                 port = parts[len(parts)-1]
  85         }
  86         return srv, ":" + port
  87 }
  88
  89 func (s *AggregatorSuite) TestUnhealthy(c *check.C) {
  90         srv, listen := s.stubServer(&unhealthyHandler{})
  91         defer srv.Close()
  92         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listen+"/")
  93         s.handler.ServeHTTP(s.resp, s.req)
  94         s.checkUnhealthy(c)
  95 }
  96
  97 func (s *AggregatorSuite) TestHealthy(c *check.C) {
  98         srv, listen := s.stubServer(&healthyHandler{})
  99         defer srv.Close()
 100         s.setAllServiceURLs(listen)
 101         s.handler.ServeHTTP(s.resp, s.req)
 102         resp := s.checkOK(c)
 103         svc := "keepstore+http://localhost" + listen + "/_health/ping"
 104         c.Logf("%#v", resp)
 105         ep := resp.Checks[svc]
 106         c.Check(ep.Health, check.Equals, "OK")
 107         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 108 }
 109
 110 func (s *AggregatorSuite) TestHealthyAndUnhealthy(c *check.C) {
 111         srvH, listenH := s.stubServer(&healthyHandler{})
 112         defer srvH.Close()
 113         srvU, listenU := s.stubServer(&unhealthyHandler{})
 114         defer srvU.Close()
 115         s.setAllServiceURLs(listenH)
 116         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listenH+"/", "http://127.0.0.1"+listenU+"/")
 117         s.handler.ServeHTTP(s.resp, s.req)
 118         resp := s.checkUnhealthy(c)
 119         ep := resp.Checks["keepstore+http://localhost"+listenH+"/_health/ping"]
 120         c.Check(ep.Health, check.Equals, "OK")
 121         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 122         ep = resp.Checks["keepstore+http://127.0.0.1"+listenU+"/_health/ping"]
 123         c.Check(ep.Health, check.Equals, "ERROR")
 124         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 125         c.Logf("%#v", ep)
 126 }
 127
 128 // If an InternalURL host is 0.0.0.0, localhost, 127/8, or ::1 and
 129 // nothing is listening there, don't fail the health check -- instead,
 130 // assume the relevant component just isn't installed/enabled on this
 131 // node, but does work when contacted through ExternalURL.
 132 func (s *AggregatorSuite) TestUnreachableLoopbackPort(c *check.C) {
 133         srvH, listenH := s.stubServer(&healthyHandler{})
 134         defer srvH.Close()
 135         s.setAllServiceURLs(listenH)
 136         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepproxy, "http://localhost:9/")
 137         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Workbench1, "http://0.0.0.0:9/")
 138         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepbalance, "http://127.0.0.127:9/")
 139         arvadostest.SetServiceURL(&s.handler.Cluster.Services.WebDAV, "http://[::1]:9/")
 140         s.handler.ServeHTTP(s.resp, s.req)
 141         s.checkOK(c)
 142
 143         // If a non-loopback address is unreachable, that's still a
 144         // fail.
 145         s.resp = httptest.NewRecorder()
 146         arvadostest.SetServiceURL(&s.handler.Cluster.Services.WebDAV, "http://172.31.255.254:9/")
 147         s.handler.ServeHTTP(s.resp, s.req)
 148         s.checkUnhealthy(c)
 149 }
 150
 151 func (s *AggregatorSuite) TestIsLocalHost(c *check.C) {
 152         c.Check(isLocalHost("Localhost"), check.Equals, true)
 153         c.Check(isLocalHost("localhost"), check.Equals, true)
 154         c.Check(isLocalHost("127.0.0.1"), check.Equals, true)
 155         c.Check(isLocalHost("127.0.0.127"), check.Equals, true)
 156         c.Check(isLocalHost("127.1.2.7"), check.Equals, true)
 157         c.Check(isLocalHost("0.0.0.0"), check.Equals, true)
 158         c.Check(isLocalHost("::1"), check.Equals, true)
 159         c.Check(isLocalHost("1.2.3.4"), check.Equals, false)
 160         c.Check(isLocalHost("1::1"), check.Equals, false)
 161         c.Check(isLocalHost("example.com"), check.Equals, false)
 162         c.Check(isLocalHost("127.0.0"), check.Equals, false)
 163         c.Check(isLocalHost(""), check.Equals, false)
 164 }
 165
 166 func (s *AggregatorSuite) TestConfigMismatch(c *check.C) {
 167         // time1/hash1: current config
 168         time1 := time.Now().Add(time.Second - time.Minute - time.Hour)
 169         hash1 := fmt.Sprintf("%x", sha256.Sum256([]byte(`Clusters: {zzzzz: {SystemRootToken: xyzzy}}`)))
 170         // time2/hash2: old config
 171         time2 := time1.Add(-time.Hour)
 172         hash2 := fmt.Sprintf("%x", sha256.Sum256([]byte(`Clusters: {zzzzz: {SystemRootToken: old-token}}`)))
 173
 174         // srv1: current file
 175         handler1 := healthyHandler{configHash: hash1, configTime: time1}
 176         srv1, listen1 := s.stubServer(&handler1)
 177         defer srv1.Close()
 178         // srv2: old file, current content
 179         handler2 := healthyHandler{configHash: hash1, configTime: time2}
 180         srv2, listen2 := s.stubServer(&handler2)
 181         defer srv2.Close()
 182         // srv3: old file, old content
 183         handler3 := healthyHandler{configHash: hash2, configTime: time2}
 184         srv3, listen3 := s.stubServer(&handler3)
 185         defer srv3.Close()
 186         // srv4: no metrics handler
 187         handler4 := healthyHandler{}
 188         srv4, listen4 := s.stubServer(&handler4)
 189         defer srv4.Close()
 190
 191         s.setAllServiceURLs(listen1)
 192
 193         // listen2 => old timestamp, same content => no problem
 194         s.resp = httptest.NewRecorder()
 195         arvadostest.SetServiceURL(&s.handler.Cluster.Services.DispatchCloud,
 196                 "http://localhost"+listen2+"/")
 197         s.handler.ServeHTTP(s.resp, s.req)
 198         resp := s.checkOK(c)
 199
 200         // listen4 => no metrics on some services => no problem
 201         s.resp = httptest.NewRecorder()
 202         arvadostest.SetServiceURL(&s.handler.Cluster.Services.WebDAV,
 203                 "http://localhost"+listen4+"/")
 204         s.handler.ServeHTTP(s.resp, s.req)
 205         resp = s.checkOK(c)
 206
 207         // listen3 => old timestamp, old content => report discrepancy
 208         s.resp = httptest.NewRecorder()
 209         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore,
 210                 "http://localhost"+listen1+"/",
 211                 "http://localhost"+listen3+"/")
 212         s.handler.ServeHTTP(s.resp, s.req)
 213         resp = s.checkUnhealthy(c)
 214         if c.Check(len(resp.Errors) > 0, check.Equals, true) {
 215                 c.Check(resp.Errors[0], check.Matches, `outdated config: \Qkeepstore+http://localhost`+listen3+`\E: config file \(sha256 .*\) does not match latest version with timestamp .*`)
 216         }
 217
 218         // no services report config time (migrating to current version) => no problem
 219         s.resp = httptest.NewRecorder()
 220         s.setAllServiceURLs(listen4)
 221         s.handler.ServeHTTP(s.resp, s.req)
 222         s.checkOK(c)
 223 }
 224
 225 func (s *AggregatorSuite) TestClockSkew(c *check.C) {
 226         // srv1: report real wall clock time
 227         handler1 := healthyHandler{}
 228         srv1, listen1 := s.stubServer(&handler1)
 229         defer srv1.Close()
 230         // srv2: report near-future time
 231         handler2 := healthyHandler{headerDate: time.Now().Add(3 * time.Second)}
 232         srv2, listen2 := s.stubServer(&handler2)
 233         defer srv2.Close()
 234         // srv3: report far-future time
 235         handler3 := healthyHandler{headerDate: time.Now().Add(3*time.Minute + 3*time.Second)}
 236         srv3, listen3 := s.stubServer(&handler3)
 237         defer srv3.Close()
 238
 239         s.setAllServiceURLs(listen1)
 240
 241         // near-future time => OK
 242         s.resp = httptest.NewRecorder()
 243         arvadostest.SetServiceURL(&s.handler.Cluster.Services.DispatchCloud,
 244                 "http://localhost"+listen2+"/")
 245         s.handler.ServeHTTP(s.resp, s.req)
 246         s.checkOK(c)
 247
 248         // far-future time => error
 249         s.resp = httptest.NewRecorder()
 250         arvadostest.SetServiceURL(&s.handler.Cluster.Services.WebDAV,
 251                 "http://localhost"+listen3+"/")
 252         s.handler.ServeHTTP(s.resp, s.req)
 253         resp := s.checkUnhealthy(c)
 254         if c.Check(len(resp.Errors) > 0, check.Equals, true) {
 255                 c.Check(resp.Errors[0], check.Matches, `clock skew detected: maximum timestamp spread is 3m.* \(exceeds warning threshold of 1m\)`)
 256         }
 257 }
 258
 259 func (s *AggregatorSuite) TestVersionSkew(c *check.C) {
 260         // srv1: report same version
 261         handler1 := healthyHandler{version: cmd.Version.String()}
 262         srv1, listen1 := s.stubServer(&handler1)
 263         defer srv1.Close()
 264         // srv2: report same version but without " (go1.2.3)" part
 265         handler2 := healthyHandler{version: strings.Fields(cmd.Version.String())[0]}
 266         srv2, listen2 := s.stubServer(&handler2)
 267         defer srv2.Close()
 268         // srv3: report different version
 269         handler3 := healthyHandler{version: "1.2.3~4 (" + runtime.Version() + ")"}
 270         srv3, listen3 := s.stubServer(&handler3)
 271         defer srv3.Close()
 272
 273         s.setAllServiceURLs(listen1)
 274
 275         // same version but without go1.2.3 part => OK
 276         s.resp = httptest.NewRecorder()
 277         arvadostest.SetServiceURL(&s.handler.Cluster.Services.RailsAPI,
 278                 "http://localhost"+listen2+"/")
 279         s.handler.ServeHTTP(s.resp, s.req)
 280         s.checkOK(c)
 281
 282         // different version => error
 283         s.resp = httptest.NewRecorder()
 284         arvadostest.SetServiceURL(&s.handler.Cluster.Services.WebDAV,
 285                 "http://localhost"+listen3+"/")
 286         s.handler.ServeHTTP(s.resp, s.req)
 287         resp := s.checkUnhealthy(c)
 288         if c.Check(len(resp.Errors) > 0, check.Equals, true) {
 289                 c.Check(resp.Errors[0], check.Matches, `version mismatch: \Qkeep-web+http://localhost`+listen3+`\E is running 1.2.3~4 (.*) -- expected \Q`+cmd.Version.String()+`\E`)
 290         }
 291 }
 292
 293 func (s *AggregatorSuite) TestPingTimeout(c *check.C) {
 294         s.handler.timeout = arvados.Duration(100 * time.Millisecond)
 295         srv, listen := s.stubServer(&slowHandler{})
 296         defer srv.Close()
 297         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listen+"/")
 298         s.handler.ServeHTTP(s.resp, s.req)
 299         resp := s.checkUnhealthy(c)
 300         ep := resp.Checks["keepstore+http://localhost"+listen+"/_health/ping"]
 301         c.Check(ep.Health, check.Equals, "ERROR")
 302         c.Check(ep.HTTPStatusCode, check.Equals, 0)
 303         rt, err := ep.ResponseTime.Float64()
 304         c.Check(err, check.IsNil)
 305         c.Check(rt > 0.005, check.Equals, true)
 306 }
 307
 308 func (s *AggregatorSuite) TestCheckCommand(c *check.C) {
 309         srv, listen := s.stubServer(&healthyHandler{})
 310         defer srv.Close()
 311         s.setAllServiceURLs(listen)
 312         tmpdir := c.MkDir()
 313         confdata, err := yaml.Marshal(arvados.Config{Clusters: map[string]arvados.Cluster{s.handler.Cluster.ClusterID: *s.handler.Cluster}})
 314         c.Assert(err, check.IsNil)
 315         confdata = regexp.MustCompile(`Source(Timestamp|SHA256): [^\n]+\n`).ReplaceAll(confdata, []byte{})
 316         err = ioutil.WriteFile(tmpdir+"/config.yml", confdata, 0777)
 317         c.Assert(err, check.IsNil)
 318
 319         var stdout, stderr bytes.Buffer
 320
 321         exitcode := CheckCommand.RunCommand("check", []string{"-config=" + tmpdir + "/config.yml"}, &bytes.Buffer{}, &stdout, &stderr)
 322         c.Check(exitcode, check.Equals, 0)
 323         c.Check(stderr.String(), check.Equals, "")
 324         c.Check(stdout.String(), check.Equals, "")
 325
 326         stdout.Reset()
 327         stderr.Reset()
 328         exitcode = CheckCommand.RunCommand("check", []string{"-config=" + tmpdir + "/config.yml", "-yaml"}, &bytes.Buffer{}, &stdout, &stderr)
 329         c.Check(exitcode, check.Equals, 0)
 330         c.Check(stderr.String(), check.Equals, "")
 331         c.Check(stdout.String(), check.Matches, `(?ms).*(\n|^)Health: OK\n.*`)
 332 }
 333
 334 func (s *AggregatorSuite) checkError(c *check.C) {
 335         c.Check(s.resp.Code, check.Not(check.Equals), http.StatusOK)
 336         var resp ClusterHealthResponse
 337         err := json.Unmarshal(s.resp.Body.Bytes(), &resp)
 338         c.Check(err, check.IsNil)
 339         c.Check(resp.Health, check.Not(check.Equals), "OK")
 340 }
 341
 342 func (s *AggregatorSuite) checkUnhealthy(c *check.C) ClusterHealthResponse {
 343         return s.checkResult(c, "ERROR")
 344 }
 345
 346 func (s *AggregatorSuite) checkOK(c *check.C) ClusterHealthResponse {
 347         return s.checkResult(c, "OK")
 348 }
 349
 350 func (s *AggregatorSuite) checkResult(c *check.C, health string) ClusterHealthResponse {
 351         c.Check(s.resp.Code, check.Equals, http.StatusOK)
 352         var resp ClusterHealthResponse
 353         c.Log(s.resp.Body.String())
 354         err := json.Unmarshal(s.resp.Body.Bytes(), &resp)
 355         c.Check(err, check.IsNil)
 356         c.Check(resp.Health, check.Equals, health)
 357         return resp
 358 }
 359
 360 func (s *AggregatorSuite) setAllServiceURLs(listen string) {
 361         svcs := &s.handler.Cluster.Services
 362         for _, svc := range []*arvados.Service{
 363                 &svcs.Controller,
 364                 &svcs.DispatchCloud,
 365                 &svcs.DispatchLSF,
 366                 &svcs.DispatchSLURM,
 367                 &svcs.GitHTTP,
 368                 &svcs.Keepbalance,
 369                 &svcs.Keepproxy,
 370                 &svcs.Keepstore,
 371                 &svcs.Health,
 372                 &svcs.RailsAPI,
 373                 &svcs.WebDAV,
 374                 &svcs.Websocket,
 375                 &svcs.Workbench1,
 376                 &svcs.Workbench2,
 377         } {
 378                 arvadostest.SetServiceURL(svc, "http://localhost"+listen+"/")
 379         }
 380 }
 381
 382 type unhealthyHandler struct{}
 383
 384 func (*unhealthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 385         if req.URL.Path == "/_health/ping" {
 386                 resp.Write([]byte(`{"health":"ERROR","error":"the bends"}`))
 387         } else {
 388                 http.Error(resp, "not found", http.StatusNotFound)
 389         }
 390 }
 391
 392 type healthyHandler struct {
 393         version    string
 394         configHash string
 395         configTime time.Time
 396         headerDate time.Time
 397 }
 398
 399 func (h *healthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 400         if !h.headerDate.IsZero() {
 401                 resp.Header().Set("Date", h.headerDate.Format(time.RFC1123))
 402         }
 403         authOK := req.Header.Get("Authorization") == "Bearer "+arvadostest.ManagementToken
 404         if req.URL.Path == "/_health/ping" {
 405                 if !authOK {
 406                         http.Error(resp, "unauthorized", http.StatusUnauthorized)
 407                         return
 408                 }
 409                 resp.Write([]byte(`{"health":"OK"}`))
 410         } else if req.URL.Path == "/metrics" {
 411                 if !authOK {
 412                         http.Error(resp, "unauthorized", http.StatusUnauthorized)
 413                         return
 414                 }
 415                 t := h.configTime
 416                 if t.IsZero() {
 417                         t = time.Now()
 418                 }
 419                 fmt.Fprintf(resp, `# HELP arvados_config_load_timestamp_seconds Time when config file was loaded.
 420 # TYPE arvados_config_load_timestamp_seconds gauge
 421 arvados_config_load_timestamp_seconds{sha256="%s"} %g
 422 # HELP arvados_config_source_timestamp_seconds Timestamp of config file when it was loaded.
 423 # TYPE arvados_config_source_timestamp_seconds gauge
 424 arvados_config_source_timestamp_seconds{sha256="%s"} %g
 425 # HELP arvados_version_running Indicated version is running.
 426 # TYPE arvados_version_running gauge
 427 arvados_version_running{version="%s"} 1
 428 `,
 429                         h.configHash, float64(time.Now().UnixNano())/1e9,
 430                         h.configHash, float64(t.UnixNano())/1e9,
 431                         h.version)
 432         } else {
 433                 http.Error(resp, "not found", http.StatusNotFound)
 434         }
 435 }
 436
 437 type slowHandler struct{}
 438
 439 func (*slowHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 440         if req.URL.Path == "/_health/ping" {
 441                 time.Sleep(3 * time.Second)
 442                 resp.Write([]byte(`{"health":"OK"}`))
 443         } else {
 444                 http.Error(resp, "not found", http.StatusNotFound)
 445         }
 446 }