// Copyright (C) The Arvados Authors. All rights reserved. // // SPDX-License-Identifier: Apache-2.0 package health import ( "bufio" "bytes" "context" "crypto/tls" "encoding/json" "errors" "flag" "fmt" "io" "net" "net/http" "net/url" "regexp" "strconv" "strings" "sync" "time" "git.arvados.org/arvados.git/lib/cmd" "git.arvados.org/arvados.git/lib/config" "git.arvados.org/arvados.git/sdk/go/arvados" "git.arvados.org/arvados.git/sdk/go/auth" "git.arvados.org/arvados.git/sdk/go/ctxlog" "github.com/ghodss/yaml" "github.com/prometheus/client_golang/prometheus" "github.com/sirupsen/logrus" ) const ( defaultTimeout = arvados.Duration(2 * time.Second) maxClockSkew = time.Minute ) // Aggregator implements service.Handler. It handles "GET /_health/all" // by checking the health of all configured services on the cluster // and responding 200 if everything is healthy. type Aggregator struct { setupOnce sync.Once httpClient *http.Client timeout arvados.Duration Cluster *arvados.Cluster // If non-nil, Log is called after handling each request. Log func(*http.Request, error) // If non-nil, report clock skew on each health-check. MetricClockSkew prometheus.Gauge } func (agg *Aggregator) setup() { agg.httpClient = &http.Client{ Transport: &http.Transport{ TLSClientConfig: &tls.Config{ InsecureSkipVerify: agg.Cluster.TLS.Insecure, }, }, } if agg.timeout == 0 { // this is always the case, except in the test suite agg.timeout = defaultTimeout } } func (agg *Aggregator) CheckHealth() error { return nil } func (agg *Aggregator) Done() <-chan struct{} { return nil } func (agg *Aggregator) ServeHTTP(resp http.ResponseWriter, req *http.Request) { agg.setupOnce.Do(agg.setup) sendErr := func(statusCode int, err error) { resp.WriteHeader(statusCode) json.NewEncoder(resp).Encode(map[string]string{"error": err.Error()}) if agg.Log != nil { agg.Log(req, err) } } resp.Header().Set("Content-Type", "application/json") if !agg.checkAuth(req) { sendErr(http.StatusUnauthorized, errUnauthorized) return } if req.URL.Path == "/_health/all" { json.NewEncoder(resp).Encode(agg.ClusterHealth()) } else if req.URL.Path == "/_health/ping" { resp.Write(healthyBody) } else { sendErr(http.StatusNotFound, errNotFound) return } if agg.Log != nil { agg.Log(req, nil) } } type ClusterHealthResponse struct { // "OK" if all needed services are OK, otherwise "ERROR". Health string // An entry for each known health check of each known instance // of each needed component: "instance of service S on node N // reports health-check C is OK." Checks map[string]CheckResult // An entry for each service type: "service S is OK." This // exposes problems that can't be expressed in Checks, like // "service S is needed, but isn't configured to run // anywhere." Services map[arvados.ServiceName]ServiceHealth // Difference between min/max timestamps in individual // health-check responses. ClockSkew arvados.Duration Errors []string } type CheckResult struct { Health string Error string `json:",omitempty"` HTTPStatusCode int `json:",omitempty"` Response map[string]interface{} `json:",omitempty"` ResponseTime json.Number ClockTime time.Time Server string // "Server" header in http response Metrics respTime time.Duration } type Metrics struct { ConfigSourceTimestamp time.Time ConfigSourceSHA256 string Version string } type ServiceHealth struct { Health string // "OK", "ERROR", or "SKIP" N int } func (agg *Aggregator) ClusterHealth() ClusterHealthResponse { agg.setupOnce.Do(agg.setup) resp := ClusterHealthResponse{ Health: "OK", Checks: make(map[string]CheckResult), Services: make(map[arvados.ServiceName]ServiceHealth), } mtx := sync.Mutex{} wg := sync.WaitGroup{} for svcName, svc := range agg.Cluster.Services.Map() { // Ensure svc is listed in resp.Services. mtx.Lock() if _, ok := resp.Services[svcName]; !ok { resp.Services[svcName] = ServiceHealth{Health: "MISSING"} } mtx.Unlock() checkURLs := map[arvados.URL]bool{} for addr := range svc.InternalURLs { checkURLs[addr] = true } if len(checkURLs) == 0 && svc.ExternalURL.Host != "" { checkURLs[svc.ExternalURL] = true } for addr := range checkURLs { wg.Add(1) go func(svcName arvados.ServiceName, addr arvados.URL) { defer wg.Done() var result CheckResult pingURL, err := agg.pingURL(addr) if err != nil { result = CheckResult{ Health: "ERROR", Error: err.Error(), } } else { result = agg.ping(pingURL) if result.Health != "SKIP" { m, err := agg.metrics(pingURL) if err != nil && result.Error == "" { result.Error = "metrics: " + err.Error() } result.Metrics = m } } mtx.Lock() defer mtx.Unlock() resp.Checks[fmt.Sprintf("%s+%s", svcName, pingURL)] = result if result.Health == "OK" || result.Health == "SKIP" { h := resp.Services[svcName] h.N++ if result.Health == "OK" || h.N == 1 { // "" => "SKIP" or "OK" // "SKIP" => "OK" h.Health = result.Health } resp.Services[svcName] = h } else { resp.Health = "ERROR" resp.Errors = append(resp.Errors, fmt.Sprintf("%s: %s: %s", svcName, result.Health, result.Error)) } }(svcName, addr) } } wg.Wait() // Report ERROR if a needed service didn't fail any checks // merely because it isn't configured to run anywhere. for svcName, sh := range resp.Services { switch svcName { case arvados.ServiceNameDispatchCloud, arvados.ServiceNameDispatchLSF, arvados.ServiceNameDispatchSLURM: // ok to not run any given dispatcher case arvados.ServiceNameHealth, arvados.ServiceNameWorkbench1, arvados.ServiceNameWorkbench2: // typically doesn't have InternalURLs in config default: if sh.Health != "OK" && sh.Health != "SKIP" { resp.Health = "ERROR" resp.Errors = append(resp.Errors, fmt.Sprintf("%s: %s: no InternalURLs configured", svcName, sh.Health)) continue } } } // Check for clock skew between hosts var maxResponseTime time.Duration var clockMin, clockMax time.Time for _, result := range resp.Checks { if result.ClockTime.IsZero() { continue } if clockMin.IsZero() || result.ClockTime.Before(clockMin) { clockMin = result.ClockTime } if result.ClockTime.After(clockMax) { clockMax = result.ClockTime } if result.respTime > maxResponseTime { maxResponseTime = result.respTime } } skew := clockMax.Sub(clockMin) resp.ClockSkew = arvados.Duration(skew) if skew > maxClockSkew+maxResponseTime { msg := fmt.Sprintf("clock skew detected: maximum timestamp spread is %s (exceeds warning threshold of %s)", resp.ClockSkew, arvados.Duration(maxClockSkew)) resp.Errors = append(resp.Errors, msg) resp.Health = "ERROR" } if agg.MetricClockSkew != nil { agg.MetricClockSkew.Set(skew.Seconds()) } // Check for mismatched config files var newest Metrics for _, result := range resp.Checks { if result.Metrics.ConfigSourceTimestamp.After(newest.ConfigSourceTimestamp) { newest = result.Metrics } } var mismatches []string for target, result := range resp.Checks { if hash := result.Metrics.ConfigSourceSHA256; hash != "" && hash != newest.ConfigSourceSHA256 { mismatches = append(mismatches, target) } } for _, target := range mismatches { msg := fmt.Sprintf("outdated config: %s: config file (sha256 %s) does not match latest version with timestamp %s", strings.TrimSuffix(target, "/_health/ping"), resp.Checks[target].Metrics.ConfigSourceSHA256, newest.ConfigSourceTimestamp.Format(time.RFC3339)) resp.Errors = append(resp.Errors, msg) resp.Health = "ERROR" } // Check for services running a different version than we are. for target, result := range resp.Checks { if result.Metrics.Version != "" && !sameVersion(result.Metrics.Version, cmd.Version.String()) { msg := fmt.Sprintf("version mismatch: %s is running %s -- expected %s", strings.TrimSuffix(target, "/_health/ping"), result.Metrics.Version, cmd.Version.String()) resp.Errors = append(resp.Errors, msg) resp.Health = "ERROR" } } return resp } func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) { base := url.URL(svcURL) return base.Parse("/_health/ping") } func (agg *Aggregator) ping(target *url.URL) (result CheckResult) { t0 := time.Now() defer func() { result.respTime = time.Since(t0) result.ResponseTime = json.Number(fmt.Sprintf("%.6f", result.respTime.Seconds())) }() result.Health = "ERROR" ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout)) defer cancel() req, err := http.NewRequestWithContext(ctx, "GET", target.String(), nil) if err != nil { result.Error = err.Error() return } req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken) // Avoid workbench1's redirect-http-to-https feature req.Header.Set("X-Forwarded-Proto", "https") resp, err := agg.httpClient.Do(req) if urlerr, ok := err.(*url.Error); ok { if neterr, ok := urlerr.Err.(*net.OpError); ok && isLocalHost(target.Hostname()) { result = CheckResult{ Health: "SKIP", Error: neterr.Error(), } err = nil return } } if err != nil { result.Error = err.Error() return } result.HTTPStatusCode = resp.StatusCode err = json.NewDecoder(resp.Body).Decode(&result.Response) if err != nil { result.Error = fmt.Sprintf("cannot decode response: %s", err) } else if resp.StatusCode != http.StatusOK { result.Error = fmt.Sprintf("HTTP %d %s", resp.StatusCode, resp.Status) } else if h, _ := result.Response["health"].(string); h != "OK" { if e, ok := result.Response["error"].(string); ok && e != "" { result.Error = e return } else { result.Error = fmt.Sprintf("health=%q in ping response", h) return } } result.Health = "OK" result.ClockTime, _ = time.Parse(time.RFC1123, resp.Header.Get("Date")) result.Server = resp.Header.Get("Server") return } var ( reConfigMetric = regexp.MustCompile(`arvados_config_source_timestamp_seconds{sha256="([0-9a-f]+)"} (\d[\d\.e\+]+)`) reVersionMetric = regexp.MustCompile(`arvados_version_running{version="([^"]+)"} 1`) ) func (agg *Aggregator) metrics(pingURL *url.URL) (result Metrics, err error) { metricsURL, err := pingURL.Parse("/metrics") if err != nil { return } ctx, cancel := context.WithTimeout(context.Background(), time.Duration(agg.timeout)) defer cancel() req, err := http.NewRequestWithContext(ctx, "GET", metricsURL.String(), nil) if err != nil { return } req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken) // Avoid workbench1's redirect-http-to-https feature req.Header.Set("X-Forwarded-Proto", "https") resp, err := agg.httpClient.Do(req) if err != nil { return } else if resp.StatusCode != http.StatusOK { err = fmt.Errorf("%s: HTTP %d %s", metricsURL.String(), resp.StatusCode, resp.Status) return } scanner := bufio.NewScanner(resp.Body) for scanner.Scan() { if m := reConfigMetric.FindSubmatch(scanner.Bytes()); len(m) == 3 && len(m[1]) > 0 { result.ConfigSourceSHA256 = string(m[1]) unixtime, _ := strconv.ParseFloat(string(m[2]), 64) result.ConfigSourceTimestamp = time.UnixMicro(int64(unixtime * 1e6)) } else if m = reVersionMetric.FindSubmatch(scanner.Bytes()); len(m) == 2 && len(m[1]) > 0 { result.Version = string(m[1]) } } if err = scanner.Err(); err != nil { err = fmt.Errorf("error parsing response from %s: %w", metricsURL.String(), err) return } return } // Test whether host is an easily recognizable loopback address: // 0.0.0.0, 127.x.x.x, ::1, or localhost. func isLocalHost(host string) bool { ip := net.ParseIP(host) return ip.IsLoopback() || bytes.Equal(ip.To4(), []byte{0, 0, 0, 0}) || strings.EqualFold(host, "localhost") } func (agg *Aggregator) checkAuth(req *http.Request) bool { creds := auth.CredentialsFromRequest(req) for _, token := range creds.Tokens { if token != "" && token == agg.Cluster.ManagementToken { return true } } return false } var errSilent = errors.New("") var CheckCommand cmd.Handler = checkCommand{} type checkCommand struct{} func (ccmd checkCommand) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int { logger := ctxlog.New(stderr, "json", "info") ctx := ctxlog.Context(context.Background(), logger) err := ccmd.run(ctx, prog, args, stdin, stdout, stderr) if err != nil { if err != errSilent { fmt.Fprintln(stderr, err.Error()) } return 1 } return 0 } func (ccmd checkCommand) run(ctx context.Context, prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) error { flags := flag.NewFlagSet("", flag.ContinueOnError) flags.SetOutput(stderr) loader := config.NewLoader(stdin, ctxlog.New(stderr, "text", "info")) loader.SetupFlags(flags) versionFlag := flags.Bool("version", false, "Write version information to stdout and exit 0") timeout := flags.Duration("timeout", defaultTimeout.Duration(), "Maximum time to wait for health responses") quiet := flags.Bool("quiet", false, "Silent on success (suppress 'health check OK' message on stderr)") outputYAML := flags.Bool("yaml", false, "Output full health report in YAML format (default mode shows errors as plain text, is silent on success)") if ok, _ := cmd.ParseFlags(flags, prog, args, "", stderr); !ok { // cmd.ParseFlags already reported the error return errSilent } else if *versionFlag { cmd.Version.RunCommand(prog, args, stdin, stdout, stderr) return nil } cfg, err := loader.Load() if err != nil { return err } cluster, err := cfg.GetCluster("") if err != nil { return err } logger := ctxlog.New(stderr, cluster.SystemLogs.Format, cluster.SystemLogs.LogLevel).WithFields(logrus.Fields{ "ClusterID": cluster.ClusterID, }) ctx = ctxlog.Context(ctx, logger) agg := Aggregator{Cluster: cluster, timeout: arvados.Duration(*timeout)} resp := agg.ClusterHealth() if *outputYAML { y, err := yaml.Marshal(resp) if err != nil { return err } stdout.Write(y) if resp.Health != "OK" { return errSilent } return nil } if resp.Health != "OK" { for _, msg := range resp.Errors { fmt.Fprintln(stderr, msg) } fmt.Fprintln(stderr, "health check failed") return errSilent } if !*quiet { fmt.Fprintln(stderr, "health check OK") } return nil } var reGoVersion = regexp.MustCompile(` \(go\d+([\d.])*\)$`) // Return true if either a==b or the only difference is that one has a // " (go1.2.3)" suffix and the other does not. // // This allows us to recognize a non-Go (rails) service as the same // version as a Go service. func sameVersion(a, b string) bool { if a == b { return true } anogo := reGoVersion.ReplaceAllLiteralString(a, "") bnogo := reGoVersion.ReplaceAllLiteralString(b, "") if (anogo == a) != (bnogo == b) { // only one of a/b has a (go1.2.3) suffix, so compare // without that part return anogo == bnogo } // both or neither has a (go1.2.3) suffix, and we already know // a!=b return false }