1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: Apache-2.0
17 "git.arvados.org/arvados.git/sdk/go/arvados"
18 "git.arvados.org/arvados.git/sdk/go/auth"
21 const defaultTimeout = arvados.Duration(2 * time.Second)
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 {
28 httpClient *http.Client
29 timeout arvados.Duration
31 Cluster *arvados.Cluster
33 // If non-nil, Log is called after handling each request.
34 Log func(*http.Request, error)
37 func (agg *Aggregator) setup() {
38 agg.httpClient = http.DefaultClient
40 // this is always the case, except in the test suite
41 agg.timeout = defaultTimeout
45 func (agg *Aggregator) CheckHealth() error {
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()})
59 resp.Header().Set("Content-Type", "application/json")
61 if !agg.checkAuth(req) {
62 sendErr(http.StatusUnauthorized, errUnauthorized)
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)
70 sendErr(http.StatusNotFound, errNotFound)
78 type ClusterHealthResponse struct {
79 // "OK" if all needed services are OK, otherwise "ERROR".
80 Health string `json:"health"`
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"`
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
91 Services map[arvados.ServiceName]ServiceHealth `json:"services"`
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"`
103 type ServiceHealth struct {
104 Health string `json:"health"`
108 func (agg *Aggregator) ClusterHealth() ClusterHealthResponse {
109 agg.setupOnce.Do(agg.setup)
110 resp := ClusterHealthResponse{
112 Checks: make(map[string]CheckResult),
113 Services: make(map[arvados.ServiceName]ServiceHealth),
117 wg := sync.WaitGroup{}
118 for svcName, svc := range agg.Cluster.Services.Map() {
119 // Ensure svc is listed in resp.Services.
121 if _, ok := resp.Services[svcName]; !ok {
122 resp.Services[svcName] = ServiceHealth{Health: "ERROR"}
126 for addr := range svc.InternalURLs {
128 go func(svcName arvados.ServiceName, addr arvados.URL) {
130 var result CheckResult
131 pingURL, err := agg.pingURL(addr)
133 result = CheckResult{
138 result = agg.ping(pingURL)
143 resp.Checks[fmt.Sprintf("%s+%s", svcName, pingURL)] = result
144 if result.Health == "OK" {
145 h := resp.Services[svcName]
148 resp.Services[svcName] = h
150 resp.Health = "ERROR"
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"
168 func (agg *Aggregator) pingURL(svcURL arvados.URL) (*url.URL, error) {
169 base := url.URL(svcURL)
170 return base.Parse("/_health/ping")
173 func (agg *Aggregator) ping(target *url.URL) (result CheckResult) {
178 result.ResponseTime = json.Number(fmt.Sprintf("%.6f", time.Since(t0).Seconds()))
180 result.Health, result.Error = "ERROR", err.Error()
186 req, err := http.NewRequest("GET", target.String(), nil)
190 req.Header.Set("Authorization", "Bearer "+agg.Cluster.ManagementToken)
192 ctx, cancel := context.WithTimeout(req.Context(), time.Duration(agg.timeout))
194 req = req.WithContext(ctx)
195 resp, err := agg.httpClient.Do(req)
199 result.HTTPStatusCode = resp.StatusCode
200 result.HTTPStatusText = resp.Status
201 err = json.NewDecoder(resp.Body).Decode(&result.Response)
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 != "" {
210 err = fmt.Errorf("health=%q in ping response", h)
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 {