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/prometheus/client_golang/prometheus"
  33         "github.com/sirupsen/logrus"
  34 )
  35
  36 const (
  37         defaultTimeout = arvados.Duration(2 * time.Second)
  38         maxClockSkew   = time.Minute
  39 )
  40
  41 // Aggregator implements service.Handler. It handles "GET /_health/all"
  42 // by checking the health of all configured services on the cluster
  43 // and responding 200 if everything is healthy.
  44 type Aggregator struct {
  45         setupOnce  sync.Once
  46         httpClient *http.Client
  47         timeout    arvados.Duration
  48
  49         Cluster *arvados.Cluster
  50
  51         // If non-nil, Log is called after handling each request.
  52         Log func(*http.Request, error)
  53
  54         // If non-nil, report clock skew on each health-check.
  55         MetricClockSkew prometheus.Gauge
  56 }
  57
  58 func (agg *Aggregator) setup() {
  59         agg.httpClient = &http.Client{
  60                 Transport: &http.Transport{
  61                         TLSClientConfig: &tls.Config{
  62                                 InsecureSkipVerify: agg.Cluster.TLS.Insecure,
  63                         },
  64                 },
  65         }
  66         if agg.timeout == 0 {
  67                 // this is always the case, except in the test suite
  68                 agg.timeout = defaultTimeout
  69         }
  70 }
  71
  72 func (agg *Aggregator) CheckHealth() error {
  73         return nil
  74 }
  75
  76 func (agg *Aggregator) Done() <-chan struct{} {
  77         return nil
  78 }
  79
  80 func (agg *Aggregator) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  81         agg.setupOnce.Do(agg.setup)
  82         sendErr := func(statusCode int, err error) {
  83                 resp.WriteHeader(statusCode)
  84                 json.NewEncoder(resp).Encode(map[string]string{"error": err.Error()})
  85                 if agg.Log != nil {
  86                         agg.Log(req, err)
  87                 }
  88         }
  89
  90         resp.Header().Set("Content-Type", "application/json")
  91
  92         if !agg.checkAuth(req) {
  93                 sendErr(http.StatusUnauthorized, errUnauthorized)
  94                 return
  95         }
  96         if req.URL.Path == "/_health/all" {
  97                 json.NewEncoder(resp).Encode(agg.ClusterHealth())
  98         } else if req.URL.Path == "/_health/ping" {
  99                 resp.Write(healthyBody)
 100         } else {
 101                 sendErr(http.StatusNotFound, errNotFound)
 102                 return
 103         }
 104         if agg.Log != nil {
 105                 agg.Log(req, nil)
 106         }
 107 }
 108
 109 type ClusterHealthResponse struct {
 110         // "OK" if all needed services are OK, otherwise "ERROR".
 111         Health string
 112
 113         // An entry for each known health check of each known instance
 114         // of each needed component: "instance of service S on node N
 115         // reports health-check C is OK."
 116         Checks map[string]CheckResult
 117
 118         // An entry for each service type: "service S is OK." This
 119         // exposes problems that can't be expressed in Checks, like
 120         // "service S is needed, but isn't configured to run
 121         // anywhere."
 122         Services map[arvados.ServiceName]ServiceHealth
 123
 124         // Difference between min/max timestamps in individual
 125         // health-check responses.
 126         ClockSkew arvados.Duration
 127
 128         Errors []string
 129 }
 130
 131 type CheckResult struct {
 132         Health         string
 133         Error          string                 `json:",omitempty"`
 134         HTTPStatusCode int                    `json:",omitempty"`
 135         Response       map[string]interface{} `json:",omitempty"`
 136         ResponseTime   json.Number
 137         ClockTime      time.Time
 138         Metrics
 139         respTime time.Duration
 140 }
 141
 142 type Metrics struct {
 143         ConfigSourceTimestamp time.Time
 144         ConfigSourceSHA256    string
 145         Version               string
 146 }
 147
 148 type ServiceHealth struct {
 149         Health string // "OK", "ERROR", or "SKIP"
 150         N      int
 151 }
 152
 153 func (agg *Aggregator) ClusterHealth() ClusterHealthResponse {
 154         agg.setupOnce.Do(agg.setup)
 155         resp := ClusterHealthResponse{
 156                 Health:   "OK",
 157                 Checks:   make(map[string]CheckResult),
 158                 Services: make(map[arvados.ServiceName]ServiceHealth),
 159         }
 160
 161         mtx := sync.Mutex{}
 162         wg := sync.WaitGroup{}
 163         for svcName, svc := range agg.Cluster.Services.Map() {
 164                 // Ensure svc is listed in resp.Services.
 165                 mtx.Lock()
 166                 if _, ok := resp.Services[svcName]; !ok {
 167                         resp.Services[svcName] = ServiceHealth{Health: "MISSING"}
 168                 }
 169                 mtx.Unlock()
 170
 171                 checkURLs := map[arvados.URL]bool{}
 172                 for addr := range svc.InternalURLs {
 173                         checkURLs[addr] = true
 174                 }
 175                 if len(checkURLs) == 0 && svc.ExternalURL.Host != "" {
 176                         checkURLs[svc.ExternalURL] = true
 177                 }
 178                 for addr := range checkURLs {
 179                         wg.Add(1)
 180                         go func(svcName arvados.ServiceName, addr arvados.URL) {
 181                                 defer wg.Done()
 182                                 var result CheckResult
 183                                 pingURL, err := agg.pingURL(addr)
 184                                 if err != nil {
 185                                         result = CheckResult{
 186                                                 Health: "ERROR",
 187                                                 Error:  err.Error(),
 188                                         }
 189                                 } else {
 190                                         result = agg.ping(pingURL)
 191                                         if result.Health != "SKIP" {
 192                                                 m, err := agg.metrics(pingURL)
 193                                                 if err != nil && result.Error == "" {
 194                                                         result.Error = "metrics: " + err.Error()
 195                                                 }
 196                                                 result.Metrics = m
 197                                         }
 198                                 }
 199
 200                                 mtx.Lock()
 201                                 defer mtx.Unlock()
 202                                 resp.Checks[fmt.Sprintf("%s+%s", svcName, pingURL)] = result
 203                                 if result.Health == "OK" || result.Health == "SKIP" {
 204                                         h := resp.Services[svcName]
 205                                         h.N++
 206                                         if result.Health == "OK" || h.N == 1 {
 207                                                 // "" => "SKIP" or "OK"
 208                                                 // "SKIP" => "OK"
 209                                                 h.Health = result.Health
 210                                         }
 211                                         resp.Services[svcName] = h
 212                                 } else {
 213                                         resp.Health = "ERROR"
 214                                         resp.Errors = append(resp.Errors, fmt.Sprintf("%s: %s: %s", svcName, result.Health, result.Error))
 215                                 }
 216                         }(svcName, addr)
 217                 }
 218         }
 219         wg.Wait()
 220
 221         // Report ERROR if a needed service didn't fail any checks
 222         // merely because it isn't configured to run anywhere.
 223         for svcName, sh := range resp.Services {
 224                 switch svcName {
 225                 case arvados.ServiceNameDispatchCloud,
 226                         arvados.ServiceNameDispatchLSF:
 227                         // ok to not run any given dispatcher
 228                 case arvados.ServiceNameHealth,
 229                         arvados.ServiceNameWorkbench1,
 230                         arvados.ServiceNameWorkbench2:
 231                         // typically doesn't have InternalURLs in config
 232                 default:
 233                         if sh.Health != "OK" && sh.Health != "SKIP" {
 234                                 resp.Health = "ERROR"
 235                                 resp.Errors = append(resp.Errors, fmt.Sprintf("%s: %s: no InternalURLs configured", svcName, sh.Health))
 236                                 continue
 237                         }
 238                 }
 239         }
 240
 241         // Check for clock skew between hosts
 242         var maxResponseTime time.Duration
 243         var clockMin, clockMax time.Time
 244         for _, result := range resp.Checks {
 245                 if result.ClockTime.IsZero() {
 246                         continue
 247                 }
 248                 if clockMin.IsZero() || result.ClockTime.Before(clockMin) {
 249                         clockMin = result.ClockTime
 250                 }
 251                 if result.ClockTime.After(clockMax) {
 252                         clockMax = result.ClockTime
 253                 }
 254                 if result.respTime > maxResponseTime {
 255                         maxResponseTime = result.respTime
 256                 }
 257         }
 258         skew := clockMax.Sub(clockMin)
 259         resp.ClockSkew = arvados.Duration(skew)
 260         if skew > maxClockSkew+maxResponseTime {
 261                 msg := fmt.Sprintf("clock skew detected: maximum timestamp spread is %s (exceeds warning threshold of %s)", resp.ClockSkew, arvados.Duration(maxClockSkew))
 262                 resp.Errors = append(resp.Errors, msg)
 263                 resp.Health = "ERROR"
 264         }
 265         if agg.MetricClockSkew != nil {
 266                 agg.MetricClockSkew.Set(skew.Seconds())
 267         }
 268
 269         // Check for mismatched config files
 270         var newest Metrics
 271         for _, result := range resp.Checks {
 272                 if result.Metrics.ConfigSourceTimestamp.After(newest.ConfigSourceTimestamp) {
 273                         newest = result.Metrics
 274                 }
 275         }
 276         var mismatches []string
 277         for target, result := range resp.Checks {
 278                 if hash := result.Metrics.ConfigSourceSHA256; hash != "" && hash != newest.ConfigSourceSHA256 {
 279                         mismatches = append(mismatches, target)
 280                 }
 281         }
 282         for _, target := range mismatches {
 283                 msg := fmt.Sprintf("outdated config: %s: config file (sha256 %s) does not match latest version with timestamp %s",
 284                         strings.TrimSuffix(target, "/_health/ping"),
 285                         resp.Checks[target].Metrics.ConfigSourceSHA256,
 286                         newest.ConfigSourceTimestamp.Format(time.RFC3339))
 287                 resp.Errors = append(resp.Errors, msg)
 288                 resp.Health = "ERROR"
 289         }
 290
 291         // Check for services running a different version than we are.
 292         for target, result := range resp.Checks {
 293                 if result.Metrics.Version != "" && !sameVersion(result.Metrics.Version, cmd.Version.String()) {
 294                         msg := fmt.Sprintf("version mismatch: %s is running %s -- expected %s",
 295                                 strings.TrimSuffix(target, "/_health/ping"),
 296                                 result.Metrics.Version,
 297                                 cmd.Version.String())
 298                         resp.Errors = append(resp.Errors, msg)
 299                         resp.Health = "ERROR"
 300                 }
 301         }
 302         return resp
 303 }
 304
 305 func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) {
 306         base := url.URL(svcURL)
 307         return base.Parse("/_health/ping")
 308 }
 309
 310 func (agg *Aggregator) ping(target *url.URL) (result CheckResult) {
 311         t0 := time.Now()
 312         defer func() {
 313                 result.respTime = time.Since(t0)
 314                 result.ResponseTime = json.Number(fmt.Sprintf("%.6f", result.respTime.Seconds()))
 315         }()
 316         result.Health = "ERROR"
 317
 318         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 319         defer cancel()
 320         req, err := http.NewRequestWithContext(ctx, "GET", target.String(), nil)
 321         if err != nil {
 322                 result.Error = err.Error()
 323                 return
 324         }
 325         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 326
 327         // Avoid workbench1's redirect-http-to-https feature
 328         req.Header.Set("X-Forwarded-Proto", "https")
 329
 330         resp, err := agg.httpClient.Do(req)
 331         if urlerr, ok := err.(*url.Error); ok {
 332                 if neterr, ok := urlerr.Err.(*net.OpError); ok && isLocalHost(target.Hostname()) {
 333                         result = CheckResult{
 334                                 Health: "SKIP",
 335                                 Error:  neterr.Error(),
 336                         }
 337                         err = nil
 338                         return
 339                 }
 340         }
 341         if err != nil {
 342                 result.Error = err.Error()
 343                 return
 344         }
 345         result.HTTPStatusCode = resp.StatusCode
 346         err = json.NewDecoder(resp.Body).Decode(&result.Response)
 347         if err != nil {
 348                 result.Error = fmt.Sprintf("cannot decode response: %s", err)
 349         } else if resp.StatusCode != http.StatusOK {
 350                 result.Error = fmt.Sprintf("HTTP %d %s", resp.StatusCode, resp.Status)
 351         } else if h, _ := result.Response["health"].(string); h != "OK" {
 352                 if e, ok := result.Response["error"].(string); ok && e != "" {
 353                         result.Error = e
 354                         return
 355                 } else {
 356                         result.Error = fmt.Sprintf("health=%q in ping response", h)
 357                         return
 358                 }
 359         }
 360         result.Health = "OK"
 361         result.ClockTime, _ = time.Parse(time.RFC1123, resp.Header.Get("Date"))
 362         return
 363 }
 364
 365 var (
 366         reConfigMetric  = regexp.MustCompile(`arvados_config_source_timestamp_seconds{sha256="([0-9a-f]+)"} (\d[\d\.e\+]+)`)
 367         reVersionMetric = regexp.MustCompile(`arvados_version_running{version="([^"]+)"} 1`)
 368 )
 369
 370 func (agg *Aggregator) metrics(pingURL *url.URL) (result Metrics, err error) {
 371         metricsURL, err := pingURL.Parse("/metrics")
 372         if err != nil {
 373                 return
 374         }
 375         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 376         defer cancel()
 377         req, err := http.NewRequestWithContext(ctx, "GET", metricsURL.String(), nil)
 378         if err != nil {
 379                 return
 380         }
 381         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 382
 383         // Avoid workbench1's redirect-http-to-https feature
 384         req.Header.Set("X-Forwarded-Proto", "https")
 385
 386         resp, err := agg.httpClient.Do(req)
 387         if err != nil {
 388                 return
 389         } else if resp.StatusCode != http.StatusOK {
 390                 err = fmt.Errorf("%s: HTTP %d %s", metricsURL.String(), resp.StatusCode, resp.Status)
 391                 return
 392         }
 393
 394         scanner := bufio.NewScanner(resp.Body)
 395         for scanner.Scan() {
 396                 if m := reConfigMetric.FindSubmatch(scanner.Bytes()); len(m) == 3 && len(m[1]) > 0 {
 397                         result.ConfigSourceSHA256 = string(m[1])
 398                         unixtime, _ := strconv.ParseFloat(string(m[2]), 64)
 399                         result.ConfigSourceTimestamp = time.UnixMicro(int64(unixtime * 1e6))
 400                 } else if m = reVersionMetric.FindSubmatch(scanner.Bytes()); len(m) == 2 && len(m[1]) > 0 {
 401                         result.Version = string(m[1])
 402                 }
 403         }
 404         if err = scanner.Err(); err != nil {
 405                 err = fmt.Errorf("error parsing response from %s: %w", metricsURL.String(), err)
 406                 return
 407         }
 408         return
 409 }
 410
 411 // Test whether host is an easily recognizable loopback address:
 412 // 0.0.0.0, 127.x.x.x, ::1, or localhost.
 413 func isLocalHost(host string) bool {
 414         ip := net.ParseIP(host)
 415         return ip.IsLoopback() || bytes.Equal(ip.To4(), []byte{0, 0, 0, 0}) || strings.EqualFold(host, "localhost")
 416 }
 417
 418 func (agg *Aggregator) checkAuth(req *http.Request) bool {
 419         creds := auth.CredentialsFromRequest(req)
 420         for _, token := range creds.Tokens {
 421                 if token != "" && token == agg.Cluster.ManagementToken {
 422                         return true
 423                 }
 424         }
 425         return false
 426 }
 427
 428 var errSilent = errors.New("")
 429
 430 var CheckCommand cmd.Handler = checkCommand{}
 431
 432 type checkCommand struct{}
 433
 434 func (ccmd checkCommand) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
 435         logger := ctxlog.New(stderr, "json", "info")
 436         ctx := ctxlog.Context(context.Background(), logger)
 437         err := ccmd.run(ctx, prog, args, stdin, stdout, stderr)
 438         if err != nil {
 439                 if err != errSilent {
 440                         fmt.Fprintln(stdout, err.Error())
 441                 }
 442                 return 1
 443         }
 444         return 0
 445 }
 446
 447 func (ccmd checkCommand) run(ctx context.Context, prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) error {
 448         flags := flag.NewFlagSet("", flag.ContinueOnError)
 449         flags.SetOutput(stderr)
 450         loader := config.NewLoader(stdin, ctxlog.New(stderr, "text", "info"))
 451         loader.SetupFlags(flags)
 452         versionFlag := flags.Bool("version", false, "Write version information to stdout and exit 0")
 453         timeout := flags.Duration("timeout", defaultTimeout.Duration(), "Maximum time to wait for health responses")
 454         outputYAML := flags.Bool("yaml", false, "Output full health report in YAML format (default mode shows errors as plain text, is silent on success)")
 455         if ok, _ := cmd.ParseFlags(flags, prog, args, "", stderr); !ok {
 456                 // cmd.ParseFlags already reported the error
 457                 return errSilent
 458         } else if *versionFlag {
 459                 cmd.Version.RunCommand(prog, args, stdin, stdout, stderr)
 460                 return nil
 461         }
 462         cfg, err := loader.Load()
 463         if err != nil {
 464                 return err
 465         }
 466         cluster, err := cfg.GetCluster("")
 467         if err != nil {
 468                 return err
 469         }
 470         logger := ctxlog.New(stderr, cluster.SystemLogs.Format, cluster.SystemLogs.LogLevel).WithFields(logrus.Fields{
 471                 "ClusterID": cluster.ClusterID,
 472         })
 473         ctx = ctxlog.Context(ctx, logger)
 474         agg := Aggregator{Cluster: cluster, timeout: arvados.Duration(*timeout)}
 475         resp := agg.ClusterHealth()
 476         if *outputYAML {
 477                 y, err := yaml.Marshal(resp)
 478                 if err != nil {
 479                         return err
 480                 }
 481                 stdout.Write(y)
 482                 if resp.Health != "OK" {
 483                         return errSilent
 484                 }
 485                 return nil
 486         }
 487         if resp.Health != "OK" {
 488                 for _, msg := range resp.Errors {
 489                         fmt.Fprintln(stdout, msg)
 490                 }
 491                 fmt.Fprintln(stderr, "health check failed")
 492                 return errSilent
 493         }
 494         return nil
 495 }
 496
 497 var reGoVersion = regexp.MustCompile(` \(go\d+([\d.])*\)$`)
 498
 499 // Return true if either a==b or the only difference is that one has a
 500 // " (go1.2.3)" suffix and the other does not.
 501 //
 502 // This allows us to recognize a non-Go (rails) service as the same
 503 // version as a Go service.
 504 func sameVersion(a, b string) bool {
 505         if a == b {
 506                 return true
 507         }
 508         anogo := reGoVersion.ReplaceAllLiteralString(a, "")
 509         bnogo := reGoVersion.ReplaceAllLiteralString(b, "")
 510         if (anogo == a) != (bnogo == b) {
 511                 // only one of a/b has a (go1.2.3) suffix, so compare
 512                 // without that part
 513                 return anogo == bnogo
 514         }
 515         // both or neither has a (go1.2.3) suffix, and we already know
 516         // a!=b
 517         return false
 518 }