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         "bufio"
   9         "bytes"
  10         "context"
  11         "crypto/tls"
  12         "encoding/json"
  13         "errors"
  14         "flag"
  15         "fmt"
  16         "io"
  17         "net"
  18         "net/http"
  19         "net/url"
  20         "regexp"
  21         "strconv"
  22         "strings"
  23         "sync"
  24         "time"
  25
  26         "git.arvados.org/arvados.git/lib/cmd"
  27         "git.arvados.org/arvados.git/lib/config"
  28         "git.arvados.org/arvados.git/sdk/go/arvados"
  29         "git.arvados.org/arvados.git/sdk/go/auth"
  30         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  31         "github.com/ghodss/yaml"
  32         "github.com/sirupsen/logrus"
  33 )
  34
  35 const defaultTimeout = arvados.Duration(2 * time.Second)
  36
  37 // Aggregator implements service.Handler. It handles "GET /_health/all"
  38 // by checking the health of all configured services on the cluster
  39 // and responding 200 if everything is healthy.
  40 type Aggregator struct {
  41         setupOnce  sync.Once
  42         httpClient *http.Client
  43         timeout    arvados.Duration
  44
  45         Cluster *arvados.Cluster
  46
  47         // If non-nil, Log is called after handling each request.
  48         Log func(*http.Request, error)
  49 }
  50
  51 func (agg *Aggregator) setup() {
  52         agg.httpClient = &http.Client{
  53                 Transport: &http.Transport{
  54                         TLSClientConfig: &tls.Config{
  55                                 InsecureSkipVerify: agg.Cluster.TLS.Insecure,
  56                         },
  57                 },
  58         }
  59         if agg.timeout == 0 {
  60                 // this is always the case, except in the test suite
  61                 agg.timeout = defaultTimeout
  62         }
  63 }
  64
  65 func (agg *Aggregator) CheckHealth() error {
  66         return nil
  67 }
  68
  69 func (agg *Aggregator) Done() <-chan struct{} {
  70         return nil
  71 }
  72
  73 func (agg *Aggregator) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  74         agg.setupOnce.Do(agg.setup)
  75         sendErr := func(statusCode int, err error) {
  76                 resp.WriteHeader(statusCode)
  77                 json.NewEncoder(resp).Encode(map[string]string{"error": err.Error()})
  78                 if agg.Log != nil {
  79                         agg.Log(req, err)
  80                 }
  81         }
  82
  83         resp.Header().Set("Content-Type", "application/json")
  84
  85         if !agg.checkAuth(req) {
  86                 sendErr(http.StatusUnauthorized, errUnauthorized)
  87                 return
  88         }
  89         if req.URL.Path == "/_health/all" {
  90                 json.NewEncoder(resp).Encode(agg.ClusterHealth())
  91         } else if req.URL.Path == "/_health/ping" {
  92                 resp.Write(healthyBody)
  93         } else {
  94                 sendErr(http.StatusNotFound, errNotFound)
  95                 return
  96         }
  97         if agg.Log != nil {
  98                 agg.Log(req, nil)
  99         }
 100 }
 101
 102 type ClusterHealthResponse struct {
 103         // "OK" if all needed services are OK, otherwise "ERROR".
 104         Health string `json:"health"`
 105
 106         // An entry for each known health check of each known instance
 107         // of each needed component: "instance of service S on node N
 108         // reports health-check C is OK."
 109         Checks map[string]CheckResult `json:"checks"`
 110
 111         // An entry for each service type: "service S is OK." This
 112         // exposes problems that can't be expressed in Checks, like
 113         // "service S is needed, but isn't configured to run
 114         // anywhere."
 115         Services map[arvados.ServiceName]ServiceHealth `json:"services"`
 116
 117         Errors []string `json:"errors"`
 118 }
 119
 120 type CheckResult struct {
 121         Health         string                 `json:"health"`
 122         Error          string                 `json:"error,omitempty"`
 123         HTTPStatusCode int                    `json:",omitempty"`
 124         HTTPStatusText string                 `json:",omitempty"`
 125         Response       map[string]interface{} `json:"response"`
 126         ResponseTime   json.Number            `json:"responseTime"`
 127         Metrics        Metrics                `json:"-"`
 128 }
 129
 130 type Metrics struct {
 131         ConfigSourceTimestamp time.Time
 132         ConfigSourceSHA256    string
 133 }
 134
 135 type ServiceHealth struct {
 136         Health string `json:"health"` // "OK", "ERROR", or "SKIP"
 137         N      int    `json:"n"`
 138 }
 139
 140 func (agg *Aggregator) ClusterHealth() ClusterHealthResponse {
 141         agg.setupOnce.Do(agg.setup)
 142         resp := ClusterHealthResponse{
 143                 Health:   "OK",
 144                 Checks:   make(map[string]CheckResult),
 145                 Services: make(map[arvados.ServiceName]ServiceHealth),
 146         }
 147
 148         mtx := sync.Mutex{}
 149         wg := sync.WaitGroup{}
 150         for svcName, svc := range agg.Cluster.Services.Map() {
 151                 // Ensure svc is listed in resp.Services.
 152                 mtx.Lock()
 153                 if _, ok := resp.Services[svcName]; !ok {
 154                         resp.Services[svcName] = ServiceHealth{Health: "NONE"}
 155                 }
 156                 mtx.Unlock()
 157
 158                 checkURLs := map[arvados.URL]bool{}
 159                 for addr := range svc.InternalURLs {
 160                         checkURLs[addr] = true
 161                 }
 162                 if len(checkURLs) == 0 && svc.ExternalURL.Host != "" {
 163                         checkURLs[svc.ExternalURL] = true
 164                 }
 165                 for addr := range checkURLs {
 166                         wg.Add(1)
 167                         go func(svcName arvados.ServiceName, addr arvados.URL) {
 168                                 defer wg.Done()
 169                                 var result CheckResult
 170                                 pingURL, err := agg.pingURL(addr)
 171                                 if err != nil {
 172                                         result = CheckResult{
 173                                                 Health: "ERROR",
 174                                                 Error:  err.Error(),
 175                                         }
 176                                 } else {
 177                                         result = agg.ping(pingURL)
 178                                         if result.Health != "SKIP" {
 179                                                 m, err := agg.metrics(pingURL)
 180                                                 if err != nil && result.Error == "" {
 181                                                         result.Error = "metrics: " + err.Error()
 182                                                 }
 183                                                 result.Metrics = m
 184                                         }
 185                                 }
 186
 187                                 mtx.Lock()
 188                                 defer mtx.Unlock()
 189                                 resp.Checks[fmt.Sprintf("%s+%s", svcName, pingURL)] = result
 190                                 if result.Health == "OK" {
 191                                         h := resp.Services[svcName]
 192                                         h.N++
 193                                         h.Health = "OK"
 194                                         resp.Services[svcName] = h
 195                                 } else if result.Health != "SKIP" {
 196                                         resp.Health = "ERROR"
 197                                 }
 198                         }(svcName, addr)
 199                 }
 200         }
 201         wg.Wait()
 202
 203         // Report ERROR if a needed service didn't fail any checks
 204         // merely because it isn't configured to run anywhere.
 205         for svcName, sh := range resp.Services {
 206                 switch svcName {
 207                 case arvados.ServiceNameDispatchCloud,
 208                         arvados.ServiceNameDispatchLSF:
 209                         // ok to not run any given dispatcher
 210                 case arvados.ServiceNameHealth,
 211                         arvados.ServiceNameWorkbench1,
 212                         arvados.ServiceNameWorkbench2:
 213                         // typically doesn't have InternalURLs in config
 214                 default:
 215                         if sh.Health != "OK" && sh.Health != "SKIP" {
 216                                 resp.Health = "ERROR"
 217                                 continue
 218                         }
 219                 }
 220         }
 221
 222         var newest Metrics
 223         for _, result := range resp.Checks {
 224                 if result.Metrics.ConfigSourceTimestamp.After(newest.ConfigSourceTimestamp) {
 225                         newest = result.Metrics
 226                 }
 227         }
 228         var mismatches []string
 229         for target, result := range resp.Checks {
 230                 if hash := result.Metrics.ConfigSourceSHA256; hash != "" && hash != newest.ConfigSourceSHA256 {
 231                         mismatches = append(mismatches, target)
 232                 }
 233         }
 234         for _, target := range mismatches {
 235                 msg := fmt.Sprintf("outdated config: %s: config file (sha256 %s) does not match latest version with timestamp %s",
 236                         strings.TrimSuffix(target, "/_health/ping"),
 237                         resp.Checks[target].Metrics.ConfigSourceSHA256,
 238                         newest.ConfigSourceTimestamp.Format(time.RFC3339))
 239                 resp.Errors = append(resp.Errors, msg)
 240                 resp.Health = "ERROR"
 241         }
 242         return resp
 243 }
 244
 245 func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) {
 246         base := url.URL(svcURL)
 247         return base.Parse("/_health/ping")
 248 }
 249
 250 func (agg *Aggregator) ping(target *url.URL) (result CheckResult) {
 251         t0 := time.Now()
 252         defer func() {
 253                 result.ResponseTime = json.Number(fmt.Sprintf("%.6f", time.Since(t0).Seconds()))
 254         }()
 255         result.Health = "ERROR"
 256
 257         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 258         defer cancel()
 259         req, err := http.NewRequestWithContext(ctx, "GET", target.String(), nil)
 260         if err != nil {
 261                 result.Error = err.Error()
 262                 return
 263         }
 264         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 265
 266         // Avoid workbench1's redirect-http-to-https feature
 267         req.Header.Set("X-Forwarded-Proto", "https")
 268
 269         resp, err := agg.httpClient.Do(req)
 270         if urlerr, ok := err.(*url.Error); ok {
 271                 if neterr, ok := urlerr.Err.(*net.OpError); ok && isLocalHost(target.Hostname()) {
 272                         result = CheckResult{
 273                                 Health: "SKIP",
 274                                 Error:  neterr.Error(),
 275                         }
 276                         err = nil
 277                         return
 278                 }
 279         }
 280         if err != nil {
 281                 result.Error = err.Error()
 282                 return
 283         }
 284         result.HTTPStatusCode = resp.StatusCode
 285         result.HTTPStatusText = resp.Status
 286         err = json.NewDecoder(resp.Body).Decode(&result.Response)
 287         if err != nil {
 288                 result.Error = fmt.Sprintf("cannot decode response: %s", err)
 289         } else if resp.StatusCode != http.StatusOK {
 290                 result.Error = fmt.Sprintf("HTTP %d %s", resp.StatusCode, resp.Status)
 291         } else if h, _ := result.Response["health"].(string); h != "OK" {
 292                 if e, ok := result.Response["error"].(string); ok && e != "" {
 293                         result.Error = e
 294                         return
 295                 } else {
 296                         result.Error = fmt.Sprintf("health=%q in ping response", h)
 297                         return
 298                 }
 299         }
 300         result.Health = "OK"
 301         return
 302 }
 303
 304 var reMetric = regexp.MustCompile(`([a-z_]+){sha256="([0-9a-f]+)"} (\d[\d\.e\+]+)`)
 305
 306 func (agg *Aggregator) metrics(pingURL *url.URL) (result Metrics, err error) {
 307         metricsURL, err := pingURL.Parse("/metrics")
 308         if err != nil {
 309                 return
 310         }
 311         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 312         defer cancel()
 313         req, err := http.NewRequestWithContext(ctx, "GET", metricsURL.String(), nil)
 314         if err != nil {
 315                 return
 316         }
 317         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 318
 319         // Avoid workbench1's redirect-http-to-https feature
 320         req.Header.Set("X-Forwarded-Proto", "https")
 321
 322         resp, err := agg.httpClient.Do(req)
 323         if err != nil {
 324                 return
 325         } else if resp.StatusCode != http.StatusOK {
 326                 err = fmt.Errorf("%s: HTTP %d %s", metricsURL.String(), resp.StatusCode, resp.Status)
 327                 return
 328         }
 329
 330         scanner := bufio.NewScanner(resp.Body)
 331         for scanner.Scan() {
 332                 m := reMetric.FindSubmatch(scanner.Bytes())
 333                 if len(m) != 4 || string(m[1]) != "arvados_config_source_timestamp_seconds" {
 334                         continue
 335                 }
 336                 result.ConfigSourceSHA256 = string(m[2])
 337                 unixtime, _ := strconv.ParseFloat(string(m[3]), 64)
 338                 result.ConfigSourceTimestamp = time.UnixMicro(int64(unixtime * 1e6))
 339         }
 340         if err = scanner.Err(); err != nil {
 341                 err = fmt.Errorf("error parsing response from %s: %w", metricsURL.String(), err)
 342                 return
 343         }
 344         return
 345 }
 346
 347 // Test whether host is an easily recognizable loopback address:
 348 // 0.0.0.0, 127.x.x.x, ::1, or localhost.
 349 func isLocalHost(host string) bool {
 350         ip := net.ParseIP(host)
 351         return ip.IsLoopback() || bytes.Equal(ip.To4(), []byte{0, 0, 0, 0}) || strings.EqualFold(host, "localhost")
 352 }
 353
 354 func (agg *Aggregator) checkAuth(req *http.Request) bool {
 355         creds := auth.CredentialsFromRequest(req)
 356         for _, token := range creds.Tokens {
 357                 if token != "" && token == agg.Cluster.ManagementToken {
 358                         return true
 359                 }
 360         }
 361         return false
 362 }
 363
 364 var errSilent = errors.New("")
 365
 366 var CheckCommand cmd.Handler = checkCommand{}
 367
 368 type checkCommand struct{}
 369
 370 func (ccmd checkCommand) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
 371         logger := ctxlog.New(stderr, "json", "info")
 372         ctx := ctxlog.Context(context.Background(), logger)
 373         err := ccmd.run(ctx, prog, args, stdin, stdout, stderr)
 374         if err != nil {
 375                 if err != errSilent {
 376                         fmt.Fprintln(stdout, err.Error())
 377                 }
 378                 return 1
 379         }
 380         return 0
 381 }
 382
 383 func (ccmd checkCommand) run(ctx context.Context, prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) error {
 384         flags := flag.NewFlagSet("", flag.ContinueOnError)
 385         flags.SetOutput(stderr)
 386         loader := config.NewLoader(stdin, ctxlog.New(stderr, "text", "info"))
 387         loader.SetupFlags(flags)
 388         versionFlag := flags.Bool("version", false, "Write version information to stdout and exit 0")
 389         timeout := flags.Duration("timeout", defaultTimeout.Duration(), "Maximum time to wait for health responses")
 390         if ok, _ := cmd.ParseFlags(flags, prog, args, "", stderr); !ok {
 391                 // cmd.ParseFlags already reported the error
 392                 return errSilent
 393         } else if *versionFlag {
 394                 cmd.Version.RunCommand(prog, args, stdin, stdout, stderr)
 395                 return nil
 396         }
 397         cfg, err := loader.Load()
 398         if err != nil {
 399                 return err
 400         }
 401         cluster, err := cfg.GetCluster("")
 402         if err != nil {
 403                 return err
 404         }
 405         logger := ctxlog.New(stderr, cluster.SystemLogs.Format, cluster.SystemLogs.LogLevel).WithFields(logrus.Fields{
 406                 "ClusterID": cluster.ClusterID,
 407         })
 408         ctx = ctxlog.Context(ctx, logger)
 409         agg := Aggregator{Cluster: cluster, timeout: arvados.Duration(*timeout)}
 410         resp := agg.ClusterHealth()
 411         buf, err := yaml.Marshal(resp)
 412         if err != nil {
 413                 return err
 414         }
 415         stdout.Write(buf)
 416         if resp.Health != "OK" {
 417                 return fmt.Errorf("health check failed")
 418         }
 419         return nil
 420 }