sdk/go/health/aggregator.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         "context"
   9         "encoding/json"
  10         "errors"
  11         "fmt"
  12         "net/http"
  13         "net/url"
  14         "sync"
  15         "time"
  16
  17         "git.arvados.org/arvados.git/sdk/go/arvados"
  18         "git.arvados.org/arvados.git/sdk/go/auth"
  19 )
  20
  21 const defaultTimeout = arvados.Duration(2 * time.Second)
  22
  23 // Aggregator implements http.Handler. It handles "GET /_health/all"
  24 // by checking the health of all configured services on the cluster
  25 // and responding 200 if everything is healthy.
  26 type Aggregator struct {
  27         setupOnce  sync.Once
  28         httpClient *http.Client
  29         timeout    arvados.Duration
  30
  31         Cluster *arvados.Cluster
  32
  33         // If non-nil, Log is called after handling each request.
  34         Log func(*http.Request, error)
  35 }
  36
  37 func (agg *Aggregator) setup() {
  38         agg.httpClient = http.DefaultClient
  39         if agg.timeout == 0 {
  40                 // this is always the case, except in the test suite
  41                 agg.timeout = defaultTimeout
  42         }
  43 }
  44
  45 func (agg *Aggregator) CheckHealth() error {
  46         return nil
  47 }
  48
  49 func (agg *Aggregator) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  50         agg.setupOnce.Do(agg.setup)
  51         sendErr := func(statusCode int, err error) {
  52                 resp.WriteHeader(statusCode)
  53                 json.NewEncoder(resp).Encode(map[string]string{"error": err.Error()})
  54                 if agg.Log != nil {
  55                         agg.Log(req, err)
  56                 }
  57         }
  58
  59         resp.Header().Set("Content-Type", "application/json")
  60
  61         if !agg.checkAuth(req) {
  62                 sendErr(http.StatusUnauthorized, errUnauthorized)
  63                 return
  64         }
  65         if req.URL.Path == "/_health/all" {
  66                 json.NewEncoder(resp).Encode(agg.ClusterHealth())
  67         } else if req.URL.Path == "/_health/ping" {
  68                 resp.Write(healthyBody)
  69         } else {
  70                 sendErr(http.StatusNotFound, errNotFound)
  71                 return
  72         }
  73         if agg.Log != nil {
  74                 agg.Log(req, nil)
  75         }
  76 }
  77
  78 type ClusterHealthResponse struct {
  79         // "OK" if all needed services are OK, otherwise "ERROR".
  80         Health string `json:"health"`
  81
  82         // An entry for each known health check of each known instance
  83         // of each needed component: "instance of service S on node N
  84         // reports health-check C is OK."
  85         Checks map[string]CheckResult `json:"checks"`
  86
  87         // An entry for each service type: "service S is OK." This
  88         // exposes problems that can't be expressed in Checks, like
  89         // "service S is needed, but isn't configured to run
  90         // anywhere."
  91         Services map[arvados.ServiceName]ServiceHealth `json:"services"`
  92 }
  93
  94 type CheckResult struct {
  95         Health         string                 `json:"health"`
  96         Error          string                 `json:"error,omitempty"`
  97         HTTPStatusCode int                    `json:",omitempty"`
  98         HTTPStatusText string                 `json:",omitempty"`
  99         Response       map[string]interface{} `json:"response"`
 100         ResponseTime   json.Number            `json:"responseTime"`
 101 }
 102
 103 type ServiceHealth struct {
 104         Health string `json:"health"`
 105         N      int    `json:"n"`
 106 }
 107
 108 func (agg *Aggregator) ClusterHealth() ClusterHealthResponse {
 109         agg.setupOnce.Do(agg.setup)
 110         resp := ClusterHealthResponse{
 111                 Health:   "OK",
 112                 Checks:   make(map[string]CheckResult),
 113                 Services: make(map[arvados.ServiceName]ServiceHealth),
 114         }
 115
 116         mtx := sync.Mutex{}
 117         wg := sync.WaitGroup{}
 118         for svcName, svc := range agg.Cluster.Services.Map() {
 119                 // Ensure svc is listed in resp.Services.
 120                 mtx.Lock()
 121                 if _, ok := resp.Services[svcName]; !ok {
 122                         resp.Services[svcName] = ServiceHealth{Health: "ERROR"}
 123                 }
 124                 mtx.Unlock()
 125
 126                 for addr := range svc.InternalURLs {
 127                         wg.Add(1)
 128                         go func(svcName arvados.ServiceName, addr arvados.URL) {
 129                                 defer wg.Done()
 130                                 var result CheckResult
 131                                 pingURL, err := agg.pingURL(addr)
 132                                 if err != nil {
 133                                         result = CheckResult{
 134                                                 Health: "ERROR",
 135                                                 Error:  err.Error(),
 136                                         }
 137                                 } else {
 138                                         result = agg.ping(pingURL)
 139                                 }
 140
 141                                 mtx.Lock()
 142                                 defer mtx.Unlock()
 143                                 resp.Checks[fmt.Sprintf("%s+%s", svcName, pingURL)] = result
 144                                 if result.Health == "OK" {
 145                                         h := resp.Services[svcName]
 146                                         h.N++
 147                                         h.Health = "OK"
 148                                         resp.Services[svcName] = h
 149                                 } else {
 150                                         resp.Health = "ERROR"
 151                                 }
 152                         }(svcName, addr)
 153                 }
 154         }
 155         wg.Wait()
 156
 157         // Report ERROR if a needed service didn't fail any checks
 158         // merely because it isn't configured to run anywhere.
 159         for _, sh := range resp.Services {
 160                 if sh.Health != "OK" {
 161                         resp.Health = "ERROR"
 162                         break
 163                 }
 164         }
 165         return resp
 166 }
 167
 168 func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) {
 169         base := url.URL(svcURL)
 170         return base.Parse("/_health/ping")
 171 }
 172
 173 func (agg *Aggregator) ping(target *url.URL) (result CheckResult) {
 174         t0 := time.Now()
 175
 176         var err error
 177         defer func() {
 178                 result.ResponseTime = json.Number(fmt.Sprintf("%.6f", time.Since(t0).Seconds()))
 179                 if err != nil {
 180                         result.Health, result.Error = "ERROR", err.Error()
 181                 } else {
 182                         result.Health = "OK"
 183                 }
 184         }()
 185
 186         req, err := http.NewRequest("GET", target.String(), nil)
 187         if err != nil {
 188                 return
 189         }
 190         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 191
 192         ctx, cancel := context.WithTimeout(req.Context(), time.Duration(agg.timeout))
 193         defer cancel()
 194         req = req.WithContext(ctx)
 195         resp, err := agg.httpClient.Do(req)
 196         if err != nil {
 197                 return
 198         }
 199         result.HTTPStatusCode = resp.StatusCode
 200         result.HTTPStatusText = resp.Status
 201         err = json.NewDecoder(resp.Body).Decode(&result.Response)
 202         if err != nil {
 203                 err = fmt.Errorf("cannot decode response: %s", err)
 204         } else if resp.StatusCode != http.StatusOK {
 205                 err = fmt.Errorf("HTTP %d %s", resp.StatusCode, resp.Status)
 206         } else if h, _ := result.Response["health"].(string); h != "OK" {
 207                 if e, ok := result.Response["error"].(string); ok && e != "" {
 208                         err = errors.New(e)
 209                 } else {
 210                         err = fmt.Errorf("health=%q in ping response", h)
 211                 }
 212         }
 213         return
 214 }
 215
 216 func (agg *Aggregator) checkAuth(req *http.Request) bool {
 217         creds := auth.CredentialsFromRequest(req)
 218         for _, token := range creds.Tokens {
 219                 if token != "" && token == agg.Cluster.ManagementToken {
 220                         return true
 221                 }
 222         }
 223         return false
 224 }