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                         arvados.ServiceNameDispatchSLURM:
 228                         // ok to not run any given dispatcher
 229                 case arvados.ServiceNameHealth,
 230                         arvados.ServiceNameWorkbench1,
 231                         arvados.ServiceNameWorkbench2:
 232                         // typically doesn't have InternalURLs in config
 233                 default:
 234                         if sh.Health != "OK" && sh.Health != "SKIP" {
 235                                 resp.Health = "ERROR"
 236                                 resp.Errors = append(resp.Errors, fmt.Sprintf("%s: %s: no InternalURLs configured", svcName, sh.Health))
 237                                 continue
 238                         }
 239                 }
 240         }
 241
 242         // Check for clock skew between hosts
 243         var maxResponseTime time.Duration
 244         var clockMin, clockMax time.Time
 245         for _, result := range resp.Checks {
 246                 if result.ClockTime.IsZero() {
 247                         continue
 248                 }
 249                 if clockMin.IsZero() || result.ClockTime.Before(clockMin) {
 250                         clockMin = result.ClockTime
 251                 }
 252                 if result.ClockTime.After(clockMax) {
 253                         clockMax = result.ClockTime
 254                 }
 255                 if result.respTime > maxResponseTime {
 256                         maxResponseTime = result.respTime
 257                 }
 258         }
 259         skew := clockMax.Sub(clockMin)
 260         resp.ClockSkew = arvados.Duration(skew)
 261         if skew > maxClockSkew+maxResponseTime {
 262                 msg := fmt.Sprintf("clock skew detected: maximum timestamp spread is %s (exceeds warning threshold of %s)", resp.ClockSkew, arvados.Duration(maxClockSkew))
 263                 resp.Errors = append(resp.Errors, msg)
 264                 resp.Health = "ERROR"
 265         }
 266         if agg.MetricClockSkew != nil {
 267                 agg.MetricClockSkew.Set(skew.Seconds())
 268         }
 269
 270         // Check for mismatched config files
 271         var newest Metrics
 272         for _, result := range resp.Checks {
 273                 if result.Metrics.ConfigSourceTimestamp.After(newest.ConfigSourceTimestamp) {
 274                         newest = result.Metrics
 275                 }
 276         }
 277         var mismatches []string
 278         for target, result := range resp.Checks {
 279                 if hash := result.Metrics.ConfigSourceSHA256; hash != "" && hash != newest.ConfigSourceSHA256 {
 280                         mismatches = append(mismatches, target)
 281                 }
 282         }
 283         for _, target := range mismatches {
 284                 msg := fmt.Sprintf("outdated config: %s: config file (sha256 %s) does not match latest version with timestamp %s",
 285                         strings.TrimSuffix(target, "/_health/ping"),
 286                         resp.Checks[target].Metrics.ConfigSourceSHA256,
 287                         newest.ConfigSourceTimestamp.Format(time.RFC3339))
 288                 resp.Errors = append(resp.Errors, msg)
 289                 resp.Health = "ERROR"
 290         }
 291
 292         // Check for services running a different version than we are.
 293         for target, result := range resp.Checks {
 294                 if result.Metrics.Version != "" && !sameVersion(result.Metrics.Version, cmd.Version.String()) {
 295                         msg := fmt.Sprintf("version mismatch: %s is running %s -- expected %s",
 296                                 strings.TrimSuffix(target, "/_health/ping"),
 297                                 result.Metrics.Version,
 298                                 cmd.Version.String())
 299                         resp.Errors = append(resp.Errors, msg)
 300                         resp.Health = "ERROR"
 301                 }
 302         }
 303         return resp
 304 }
 305
 306 func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) {
 307         base := url.URL(svcURL)
 308         return base.Parse("/_health/ping")
 309 }
 310
 311 func (agg *Aggregator) ping(target *url.URL) (result CheckResult) {
 312         t0 := time.Now()
 313         defer func() {
 314                 result.respTime = time.Since(t0)
 315                 result.ResponseTime = json.Number(fmt.Sprintf("%.6f", result.respTime.Seconds()))
 316         }()
 317         result.Health = "ERROR"
 318
 319         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 320         defer cancel()
 321         req, err := http.NewRequestWithContext(ctx, "GET", target.String(), nil)
 322         if err != nil {
 323                 result.Error = err.Error()
 324                 return
 325         }
 326         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 327
 328         // Avoid workbench1's redirect-http-to-https feature
 329         req.Header.Set("X-Forwarded-Proto", "https")
 330
 331         resp, err := agg.httpClient.Do(req)
 332         if urlerr, ok := err.(*url.Error); ok {
 333                 if neterr, ok := urlerr.Err.(*net.OpError); ok && isLocalHost(target.Hostname()) {
 334                         result = CheckResult{
 335                                 Health: "SKIP",
 336                                 Error:  neterr.Error(),
 337                         }
 338                         err = nil
 339                         return
 340                 }
 341         }
 342         if err != nil {
 343                 result.Error = err.Error()
 344                 return
 345         }
 346         result.HTTPStatusCode = resp.StatusCode
 347         err = json.NewDecoder(resp.Body).Decode(&result.Response)
 348         if err != nil {
 349                 result.Error = fmt.Sprintf("cannot decode response: %s", err)
 350         } else if resp.StatusCode != http.StatusOK {
 351                 result.Error = fmt.Sprintf("HTTP %d %s", resp.StatusCode, resp.Status)
 352         } else if h, _ := result.Response["health"].(string); h != "OK" {
 353                 if e, ok := result.Response["error"].(string); ok && e != "" {
 354                         result.Error = e
 355                         return
 356                 } else {
 357                         result.Error = fmt.Sprintf("health=%q in ping response", h)
 358                         return
 359                 }
 360         }
 361         result.Health = "OK"
 362         result.ClockTime, _ = time.Parse(time.RFC1123, resp.Header.Get("Date"))
 363         return
 364 }
 365
 366 var (
 367         reConfigMetric  = regexp.MustCompile(`arvados_config_source_timestamp_seconds{sha256="([0-9a-f]+)"} (\d[\d\.e\+]+)`)
 368         reVersionMetric = regexp.MustCompile(`arvados_version_running{version="([^"]+)"} 1`)
 369 )
 370
 371 func (agg *Aggregator) metrics(pingURL *url.URL) (result Metrics, err error) {
 372         metricsURL, err := pingURL.Parse("/metrics")
 373         if err != nil {
 374                 return
 375         }
 376         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout))
 377         defer cancel()
 378         req, err := http.NewRequestWithContext(ctx, "GET", metricsURL.String(), nil)
 379         if err != nil {
 380                 return
 381         }
 382         req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
 383
 384         // Avoid workbench1's redirect-http-to-https feature
 385         req.Header.Set("X-Forwarded-Proto", "https")
 386
 387         resp, err := agg.httpClient.Do(req)
 388         if err != nil {
 389                 return
 390         } else if resp.StatusCode != http.StatusOK {
 391                 err = fmt.Errorf("%s: HTTP %d %s", metricsURL.String(), resp.StatusCode, resp.Status)
 392                 return
 393         }
 394
 395         scanner := bufio.NewScanner(resp.Body)
 396         for scanner.Scan() {
 397                 if m := reConfigMetric.FindSubmatch(scanner.Bytes()); len(m) == 3 && len(m[1]) > 0 {
 398                         result.ConfigSourceSHA256 = string(m[1])
 399                         unixtime, _ := strconv.ParseFloat(string(m[2]), 64)
 400                         result.ConfigSourceTimestamp = time.UnixMicro(int64(unixtime * 1e6))
 401                 } else if m = reVersionMetric.FindSubmatch(scanner.Bytes()); len(m) == 2 && len(m[1]) > 0 {
 402                         result.Version = string(m[1])
 403                 }
 404         }
 405         if err = scanner.Err(); err != nil {
 406                 err = fmt.Errorf("error parsing response from %s: %w", metricsURL.String(), err)
 407                 return
 408         }
 409         return
 410 }
 411
 412 // Test whether host is an easily recognizable loopback address:
 413 // 0.0.0.0, 127.x.x.x, ::1, or localhost.
 414 func isLocalHost(host string) bool {
 415         ip := net.ParseIP(host)
 416         return ip.IsLoopback() || bytes.Equal(ip.To4(), []byte{0, 0, 0, 0}) || strings.EqualFold(host, "localhost")
 417 }
 418
 419 func (agg *Aggregator) checkAuth(req *http.Request) bool {
 420         creds := auth.CredentialsFromRequest(req)
 421         for _, token := range creds.Tokens {
 422                 if token != "" && token == agg.Cluster.ManagementToken {
 423                         return true
 424                 }
 425         }
 426         return false
 427 }
 428
 429 var errSilent = errors.New("")
 430
 431 var CheckCommand cmd.Handler = checkCommand{}
 432
 433 type checkCommand struct{}
 434
 435 func (ccmd checkCommand) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
 436         logger := ctxlog.New(stderr, "json", "info")
 437         ctx := ctxlog.Context(context.Background(), logger)
 438         err := ccmd.run(ctx, prog, args, stdin, stdout, stderr)
 439         if err != nil {
 440                 if err != errSilent {
 441                         fmt.Fprintln(stderr, err.Error())
 442                 }
 443                 return 1
 444         }
 445         return 0
 446 }
 447
 448 func (ccmd checkCommand) run(ctx context.Context, prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) error {
 449         flags := flag.NewFlagSet("", flag.ContinueOnError)
 450         flags.SetOutput(stderr)
 451         loader := config.NewLoader(stdin, ctxlog.New(stderr, "text", "info"))
 452         loader.SetupFlags(flags)
 453         versionFlag := flags.Bool("version", false, "Write version information to stdout and exit 0")
 454         timeout := flags.Duration("timeout", defaultTimeout.Duration(), "Maximum time to wait for health responses")
 455         quiet := flags.Bool("quiet", false, "Silent on success (suppress 'health check OK' message on stderr)")
 456         outputYAML := flags.Bool("yaml", false, "Output full health report in YAML format (default mode shows errors as plain text, is silent on success)")
 457         if ok, _ := cmd.ParseFlags(flags, prog, args, "", stderr); !ok {
 458                 // cmd.ParseFlags already reported the error
 459                 return errSilent
 460         } else if *versionFlag {
 461                 cmd.Version.RunCommand(prog, args, stdin, stdout, stderr)
 462                 return nil
 463         }
 464         cfg, err := loader.Load()
 465         if err != nil {
 466                 return err
 467         }
 468         cluster, err := cfg.GetCluster("")
 469         if err != nil {
 470                 return err
 471         }
 472         logger := ctxlog.New(stderr, cluster.SystemLogs.Format, cluster.SystemLogs.LogLevel).WithFields(logrus.Fields{
 473                 "ClusterID": cluster.ClusterID,
 474         })
 475         ctx = ctxlog.Context(ctx, logger)
 476         agg := Aggregator{Cluster: cluster, timeout: arvados.Duration(*timeout)}
 477         resp := agg.ClusterHealth()
 478         if *outputYAML {
 479                 y, err := yaml.Marshal(resp)
 480                 if err != nil {
 481                         return err
 482                 }
 483                 stdout.Write(y)
 484                 if resp.Health != "OK" {
 485                         return errSilent
 486                 }
 487                 return nil
 488         }
 489         if resp.Health != "OK" {
 490                 for _, msg := range resp.Errors {
 491                         fmt.Fprintln(stderr, msg)
 492                 }
 493                 fmt.Fprintln(stderr, "health check failed")
 494                 return errSilent
 495         }
 496         if !*quiet {
 497                 fmt.Fprintln(stderr, "health check OK")
 498         }
 499         return nil
 500 }
 501
 502 var reGoVersion = regexp.MustCompile(` \(go\d+([\d.])*\)$`)
 503
 504 // Return true if either a==b or the only difference is that one has a
 505 // " (go1.2.3)" suffix and the other does not.
 506 //
 507 // This allows us to recognize a non-Go (rails) service as the same
 508 // version as a Go service.
 509 func sameVersion(a, b string) bool {
 510         if a == b {
 511                 return true
 512         }
 513         anogo := reGoVersion.ReplaceAllLiteralString(a, "")
 514         bnogo := reGoVersion.ReplaceAllLiteralString(b, "")
 515         if (anogo == a) != (bnogo == b) {
 516                 // only one of a/b has a (go1.2.3) suffix, so compare
 517                 // without that part
 518                 return anogo == bnogo
 519         }
 520         // both or neither has a (go1.2.3) suffix, and we already know
 521         // a!=b
 522         return false
 523 }