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         "encoding/json"
  10         "io/ioutil"
  11         "net/http"
  12         "net/http/httptest"
  13         "regexp"
  14         "strings"
  15         "time"
  16
  17         "git.arvados.org/arvados.git/lib/config"
  18         "git.arvados.org/arvados.git/sdk/go/arvados"
  19         "git.arvados.org/arvados.git/sdk/go/arvadostest"
  20         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  21         "github.com/ghodss/yaml"
  22         "gopkg.in/check.v1"
  23 )
  24
  25 type AggregatorSuite struct {
  26         handler *Aggregator
  27         req     *http.Request
  28         resp    *httptest.ResponseRecorder
  29 }
  30
  31 // Gocheck boilerplate
  32 var _ = check.Suite(&AggregatorSuite{})
  33
  34 func (s *AggregatorSuite) TestInterface(c *check.C) {
  35         var _ http.Handler = &Aggregator{}
  36 }
  37
  38 func (s *AggregatorSuite) SetUpTest(c *check.C) {
  39         ldr := config.NewLoader(bytes.NewBufferString(`Clusters: {zzzzz: {}}`), ctxlog.TestLogger(c))
  40         ldr.Path = "-"
  41         cfg, err := ldr.Load()
  42         c.Assert(err, check.IsNil)
  43         cluster, err := cfg.GetCluster("")
  44         c.Assert(err, check.IsNil)
  45         cluster.ManagementToken = arvadostest.ManagementToken
  46         cluster.SystemRootToken = arvadostest.SystemRootToken
  47         cluster.Collections.BlobSigningKey = arvadostest.BlobSigningKey
  48         cluster.Volumes["z"] = arvados.Volume{StorageClasses: map[string]bool{"default": true}}
  49         s.handler = &Aggregator{Cluster: cluster}
  50         s.req = httptest.NewRequest("GET", "/_health/all", nil)
  51         s.req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
  52         s.resp = httptest.NewRecorder()
  53 }
  54
  55 func (s *AggregatorSuite) TestNoAuth(c *check.C) {
  56         s.req.Header.Del("Authorization")
  57         s.handler.ServeHTTP(s.resp, s.req)
  58         s.checkError(c)
  59         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  60 }
  61
  62 func (s *AggregatorSuite) TestBadAuth(c *check.C) {
  63         s.req.Header.Set("Authorization", "xyzzy")
  64         s.handler.ServeHTTP(s.resp, s.req)
  65         s.checkError(c)
  66         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  67 }
  68
  69 func (s *AggregatorSuite) TestNoServicesConfigured(c *check.C) {
  70         s.handler.ServeHTTP(s.resp, s.req)
  71         s.checkUnhealthy(c)
  72 }
  73
  74 func (s *AggregatorSuite) stubServer(handler http.Handler) (*httptest.Server, string) {
  75         srv := httptest.NewServer(handler)
  76         var port string
  77         if parts := strings.Split(srv.URL, ":"); len(parts) < 3 {
  78                 panic(srv.URL)
  79         } else {
  80                 port = parts[len(parts)-1]
  81         }
  82         return srv, ":" + port
  83 }
  84
  85 func (s *AggregatorSuite) TestUnhealthy(c *check.C) {
  86         srv, listen := s.stubServer(&unhealthyHandler{})
  87         defer srv.Close()
  88         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listen+"/")
  89         s.handler.ServeHTTP(s.resp, s.req)
  90         s.checkUnhealthy(c)
  91 }
  92
  93 func (s *AggregatorSuite) TestHealthy(c *check.C) {
  94         srv, listen := s.stubServer(&healthyHandler{})
  95         defer srv.Close()
  96         s.setAllServiceURLs(listen)
  97         s.handler.ServeHTTP(s.resp, s.req)
  98         resp := s.checkOK(c)
  99         svc := "keepstore+http://localhost" + listen + "/_health/ping"
 100         c.Logf("%#v", resp)
 101         ep := resp.Checks[svc]
 102         c.Check(ep.Health, check.Equals, "OK")
 103         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 104 }
 105
 106 func (s *AggregatorSuite) TestHealthyAndUnhealthy(c *check.C) {
 107         srvH, listenH := s.stubServer(&healthyHandler{})
 108         defer srvH.Close()
 109         srvU, listenU := s.stubServer(&unhealthyHandler{})
 110         defer srvU.Close()
 111         s.setAllServiceURLs(listenH)
 112         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listenH+"/", "http://127.0.0.1"+listenU+"/")
 113         s.handler.ServeHTTP(s.resp, s.req)
 114         resp := s.checkUnhealthy(c)
 115         ep := resp.Checks["keepstore+http://localhost"+listenH+"/_health/ping"]
 116         c.Check(ep.Health, check.Equals, "OK")
 117         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 118         ep = resp.Checks["keepstore+http://127.0.0.1"+listenU+"/_health/ping"]
 119         c.Check(ep.Health, check.Equals, "ERROR")
 120         c.Check(ep.HTTPStatusCode, check.Equals, 200)
 121         c.Logf("%#v", ep)
 122 }
 123
 124 func (s *AggregatorSuite) TestPingTimeout(c *check.C) {
 125         s.handler.timeout = arvados.Duration(100 * time.Millisecond)
 126         srv, listen := s.stubServer(&slowHandler{})
 127         defer srv.Close()
 128         arvadostest.SetServiceURL(&s.handler.Cluster.Services.Keepstore, "http://localhost"+listen+"/")
 129         s.handler.ServeHTTP(s.resp, s.req)
 130         resp := s.checkUnhealthy(c)
 131         ep := resp.Checks["keepstore+http://localhost"+listen+"/_health/ping"]
 132         c.Check(ep.Health, check.Equals, "ERROR")
 133         c.Check(ep.HTTPStatusCode, check.Equals, 0)
 134         rt, err := ep.ResponseTime.Float64()
 135         c.Check(err, check.IsNil)
 136         c.Check(rt > 0.005, check.Equals, true)
 137 }
 138
 139 func (s *AggregatorSuite) TestCheckCommand(c *check.C) {
 140         srv, listen := s.stubServer(&healthyHandler{})
 141         defer srv.Close()
 142         s.setAllServiceURLs(listen)
 143         tmpdir := c.MkDir()
 144         confdata, err := yaml.Marshal(arvados.Config{Clusters: map[string]arvados.Cluster{s.handler.Cluster.ClusterID: *s.handler.Cluster}})
 145         c.Assert(err, check.IsNil)
 146         confdata = regexp.MustCompile(`SourceTimestamp: [^\n]+\n`).ReplaceAll(confdata, []byte{})
 147         err = ioutil.WriteFile(tmpdir+"/config.yml", confdata, 0777)
 148         c.Assert(err, check.IsNil)
 149         var stdout, stderr bytes.Buffer
 150         exitcode := CheckCommand.RunCommand("check", []string{"-config=" + tmpdir + "/config.yml"}, &bytes.Buffer{}, &stdout, &stderr)
 151         c.Check(exitcode, check.Equals, 0)
 152         c.Check(stderr.String(), check.Equals, "")
 153         c.Check(stdout.String(), check.Matches, `(?ms).*(\n|^)health: OK\n.*`)
 154 }
 155
 156 func (s *AggregatorSuite) checkError(c *check.C) {
 157         c.Check(s.resp.Code, check.Not(check.Equals), http.StatusOK)
 158         var resp ClusterHealthResponse
 159         err := json.Unmarshal(s.resp.Body.Bytes(), &resp)
 160         c.Check(err, check.IsNil)
 161         c.Check(resp.Health, check.Not(check.Equals), "OK")
 162 }
 163
 164 func (s *AggregatorSuite) checkUnhealthy(c *check.C) ClusterHealthResponse {
 165         return s.checkResult(c, "ERROR")
 166 }
 167
 168 func (s *AggregatorSuite) checkOK(c *check.C) ClusterHealthResponse {
 169         return s.checkResult(c, "OK")
 170 }
 171
 172 func (s *AggregatorSuite) checkResult(c *check.C, health string) ClusterHealthResponse {
 173         c.Check(s.resp.Code, check.Equals, http.StatusOK)
 174         var resp ClusterHealthResponse
 175         c.Log(s.resp.Body.String())
 176         err := json.Unmarshal(s.resp.Body.Bytes(), &resp)
 177         c.Check(err, check.IsNil)
 178         c.Check(resp.Health, check.Equals, health)
 179         return resp
 180 }
 181
 182 func (s *AggregatorSuite) setAllServiceURLs(listen string) {
 183         svcs := &s.handler.Cluster.Services
 184         for _, svc := range []*arvados.Service{
 185                 &svcs.Controller,
 186                 &svcs.DispatchCloud,
 187                 &svcs.DispatchLSF,
 188                 &svcs.Keepbalance,
 189                 &svcs.Keepproxy,
 190                 &svcs.Keepstore,
 191                 &svcs.Health,
 192                 &svcs.RailsAPI,
 193                 &svcs.WebDAV,
 194                 &svcs.Websocket,
 195                 &svcs.Workbench1,
 196                 &svcs.Workbench2,
 197         } {
 198                 arvadostest.SetServiceURL(svc, "http://localhost"+listen+"/")
 199         }
 200 }
 201
 202 type unhealthyHandler struct{}
 203
 204 func (*unhealthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 205         if req.URL.Path == "/_health/ping" {
 206                 resp.Write([]byte(`{"health":"ERROR","error":"the bends"}`))
 207         } else {
 208                 http.Error(resp, "not found", http.StatusNotFound)
 209         }
 210 }
 211
 212 type healthyHandler struct{}
 213
 214 func (*healthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 215         if req.URL.Path == "/_health/ping" {
 216                 resp.Write([]byte(`{"health":"OK"}`))
 217         } else {
 218                 http.Error(resp, "not found", http.StatusNotFound)
 219         }
 220 }
 221
 222 type slowHandler struct{}
 223
 224 func (*slowHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 225         if req.URL.Path == "/_health/ping" {
 226                 time.Sleep(3 * time.Second)
 227                 resp.Write([]byte(`{"health":"OK"}`))
 228         } else {
 229                 http.Error(resp, "not found", http.StatusNotFound)
 230         }
 231 }