sdk/go/health/aggregator_test.go

   1 package health
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6         "net"
   7         "net/http"
   8         "net/http/httptest"
   9         "strings"
  10         "time"
  11
  12         "git.curoverse.com/arvados.git/sdk/go/arvados"
  13         "git.curoverse.com/arvados.git/sdk/go/arvadostest"
  14         "gopkg.in/check.v1"
  15 )
  16
  17 type AggregatorSuite struct {
  18         handler *Aggregator
  19         req     *http.Request
  20         resp    *httptest.ResponseRecorder
  21 }
  22
  23 // Gocheck boilerplate
  24 var _ = check.Suite(&AggregatorSuite{})
  25
  26 func (s *AggregatorSuite) TestInterface(c *check.C) {
  27         var _ http.Handler = &Aggregator{}
  28 }
  29
  30 func (s *AggregatorSuite) SetUpTest(c *check.C) {
  31         s.handler = &Aggregator{Config: &arvados.Config{
  32                 Clusters: map[string]arvados.Cluster{
  33                         "zzzzz": {
  34                                 ManagementToken: arvadostest.ManagementToken,
  35                                 SystemNodes:     map[string]arvados.SystemNode{},
  36                         },
  37                 },
  38         }}
  39         s.req = httptest.NewRequest("GET", "/_health/all", nil)
  40         s.req.Header.Set("Authorization", "Bearer "+arvadostest.ManagementToken)
  41         s.resp = httptest.NewRecorder()
  42 }
  43
  44 func (s *AggregatorSuite) TestNoAuth(c *check.C) {
  45         s.req.Header.Del("Authorization")
  46         s.handler.ServeHTTP(s.resp, s.req)
  47         s.checkError(c)
  48         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  49 }
  50
  51 func (s *AggregatorSuite) TestBadAuth(c *check.C) {
  52         s.req.Header.Set("Authorization", "xyzzy")
  53         s.handler.ServeHTTP(s.resp, s.req)
  54         s.checkError(c)
  55         c.Check(s.resp.Code, check.Equals, http.StatusUnauthorized)
  56 }
  57
  58 func (s *AggregatorSuite) TestEmptyConfig(c *check.C) {
  59         s.handler.ServeHTTP(s.resp, s.req)
  60         s.checkOK(c)
  61 }
  62
  63 func (s *AggregatorSuite) stubServer(handler http.Handler) (*httptest.Server, string) {
  64         srv := httptest.NewServer(handler)
  65         var port string
  66         if parts := strings.Split(srv.URL, ":"); len(parts) < 3 {
  67                 panic(srv.URL)
  68         } else {
  69                 port = parts[len(parts)-1]
  70         }
  71         return srv, ":" + port
  72 }
  73
  74 type unhealthyHandler struct{}
  75
  76 func (*unhealthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  77         if req.URL.Path == "/_health/ping" {
  78                 resp.Write([]byte(`{"health":"ERROR","error":"the bends"}`))
  79         } else {
  80                 http.Error(resp, "not found", http.StatusNotFound)
  81         }
  82 }
  83
  84 func (s *AggregatorSuite) TestUnhealthy(c *check.C) {
  85         srv, listen := s.stubServer(&unhealthyHandler{})
  86         defer srv.Close()
  87         s.handler.Config.Clusters["zzzzz"].SystemNodes["localhost"] = arvados.SystemNode{
  88                 Keepstore: arvados.Keepstore{Listen: listen},
  89         }
  90         s.handler.ServeHTTP(s.resp, s.req)
  91         s.checkUnhealthy(c)
  92 }
  93
  94 type healthyHandler struct{}
  95
  96 func (*healthyHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  97         if req.URL.Path == "/_health/ping" {
  98                 resp.Write([]byte(`{"health":"OK"}`))
  99         } else {
 100                 http.Error(resp, "not found", http.StatusNotFound)
 101         }
 102 }
 103
 104 func (s *AggregatorSuite) TestHealthy(c *check.C) {
 105         srv, listen := s.stubServer(&healthyHandler{})
 106         defer srv.Close()
 107         _, port, _ := net.SplitHostPort(listen)
 108         s.handler.Config.Clusters["zzzzz"].SystemNodes["localhost"] = arvados.SystemNode{
 109                 Keepstore: arvados.Keepstore{Listen: listen},
 110         }
 111         s.handler.ServeHTTP(s.resp, s.req)
 112         resp := s.checkOK(c)
 113         ep := resp.Checks[fmt.Sprintf("keepstore+http://localhost:%d/_health/ping", port)]
 114         c.Check(ep.Health, check.Equals, "OK")
 115         c.Check(ep.Status, check.Equals, 200)
 116 }
 117
 118 func (s *AggregatorSuite) TestHealthyAndUnhealthy(c *check.C) {
 119         srvH, listenH := s.stubServer(&healthyHandler{})
 120         defer srvH.Close()
 121         _, portH, _ := net.SplitHostPort(listenH)
 122         srvU, listenU := s.stubServer(&unhealthyHandler{})
 123         defer srvU.Close()
 124         _, portU, _ := net.SplitHostPort(listenU)
 125         s.handler.Config.Clusters["zzzzz"].SystemNodes["localhost"] = arvados.SystemNode{
 126                 Keepstore: arvados.Keepstore{Listen: listenH},
 127         }
 128         s.handler.Config.Clusters["zzzzz"].SystemNodes["127.0.0.1"] = arvados.SystemNode{
 129                 Keepstore: arvados.Keepstore{Listen: listenU},
 130         }
 131         s.handler.ServeHTTP(s.resp, s.req)
 132         resp := s.checkUnhealthy(c)
 133         ep := resp.Checks[fmt.Sprintf("keepstore+http://localhost:%d/_health/ping", portH)]
 134         c.Check(ep.Health, check.Equals, "OK")
 135         c.Check(ep.Status, check.Equals, 200)
 136         ep = resp.Checks[fmt.Sprintf("keepstore+http://127.0.0.1:%d/_health/ping", portU)]
 137         c.Check(ep.Health, check.Equals, "ERROR")
 138         c.Check(ep.Status, check.Equals, 200)
 139 }
 140
 141 func (s *AggregatorSuite) checkError(c *check.C) {
 142         c.Check(s.resp.Code, check.Not(check.Equals), http.StatusOK)
 143         var resp ClusterHealthResponse
 144         err := json.NewDecoder(s.resp.Body).Decode(&resp)
 145         c.Check(err, check.IsNil)
 146         c.Check(resp.Health, check.Not(check.Equals), "OK")
 147 }
 148
 149 func (s *AggregatorSuite) checkUnhealthy(c *check.C) ClusterHealthResponse {
 150         return s.checkResult(c, "ERROR")
 151 }
 152
 153 func (s *AggregatorSuite) checkOK(c *check.C) ClusterHealthResponse {
 154         return s.checkResult(c, "OK")
 155 }
 156
 157 func (s *AggregatorSuite) checkResult(c *check.C, health string) ClusterHealthResponse {
 158         c.Check(s.resp.Code, check.Equals, http.StatusOK)
 159         var resp ClusterHealthResponse
 160         err := json.NewDecoder(s.resp.Body).Decode(&resp)
 161         c.Check(err, check.IsNil)
 162         c.Check(resp.Health, check.Equals, health)
 163         return resp
 164 }
 165
 166 type slowHandler struct{}
 167
 168 func (*slowHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
 169         if req.URL.Path == "/_health/ping" {
 170                 time.Sleep(3 * time.Second)
 171                 resp.Write([]byte(`{"health":"OK"}`))
 172         } else {
 173                 http.Error(resp, "not found", http.StatusNotFound)
 174         }
 175 }
 176
 177 func (s *AggregatorSuite) TestPingTimeout(c *check.C) {
 178         s.handler.timeout = arvados.Duration(100 * time.Millisecond)
 179         srv, listen := s.stubServer(&slowHandler{})
 180         defer srv.Close()
 181         s.handler.Config.Clusters["zzzzz"].SystemNodes["localhost"] = arvados.SystemNode{
 182                 Keepstore: arvados.Keepstore{Listen: listen},
 183         }
 184         s.handler.ServeHTTP(s.resp, s.req)
 185         resp := s.checkUnhealthy(c)
 186         ep := resp.Checks["localhost/keepstore/_health/ping"]
 187         c.Check(ep.Health, check.Equals, "ERROR")
 188         c.Check(ep.Status, check.Equals, 0)
 189         rt, err := ep.ResponseTime.Float64()
 190         c.Check(err, check.IsNil)
 191         c.Check(rt > 0.005, check.Equals, true)
 192 }