Refactor task queue to resolve task-related failures

.travis.yml

@@ -29,7 +29,7 @@ before_install:
   - sudo apt-get install --allow-downgrades -y docker-buildx-plugin mongodb-org=${MONGODB} mongodb-org-database=${MONGODB} mongodb-org-server=${MONGODB} mongodb-mongosh=${MONGOSH} mongodb-org-mongos=${MONGODB} mongodb-org-tools
   - mkdir -p /var/ramfs/mongodb/data
   - /usr/bin/mongod --dbpath /var/ramfs/mongodb/data --bind_ip 127.0.0.1 --replSet rs0 --logpath /var/ramfs/mongodb/mongod.log &> /dev/null &
-  - until nc -z localhost 27017; do echo Waiting for MongoDB; sleep 1; done
+  - until nc -z 127.0.0.1 27017; do echo Waiting for MongoDB; sleep 1; done
   - /usr/bin/mongosh --eval 'rs.initiate(); while (rs.status().startupStatus || (rs.status().hasOwnProperty("myState") && rs.status().myState != 1)) { printjson( rs.status() ); sleep(1000); }; printjson( rs.status() );'
   - echo -e "machine github.com\n  login $GITHUB_TOKEN" > ~/.netrc
 

Makefile

@@ -240,7 +240,7 @@ vet: tmp
 	@echo "go vet ./..."
 	@cd $(ROOT_DIRECTORY) && \
 		{ [ -z `go env GOWORK` ] || GOWORK_FLAGS=-mod=readonly; } && \
-		go vet $${GOWORK_FLAGS:-} ./... > _tmp/govet.out 2>&1 || \
+		$(TIMING_CMD) go vet $${GOWORK_FLAGS:-} ./... > _tmp/govet.out 2>&1 || \
 		(diff .govetignore _tmp/govet.out && exit 1)
 
 vet-ignore:

README.md

@@ -159,3 +159,47 @@ Review all pending changes to all dependencies. If any changes could have a nega
 Ensure the `ci-build` and `ci-test` Makefile targets pass using the target Golang version.
 
 If you previously noted any changes or issues of concern, perform any explicit tests necessary.
+
+## Prometheus Metrics
+
+See source files for further details about and usage of each metric.
+
+### Summary Store
+
+* `tidepool_summary_queue_lag` - (histogram) - the current summary queue lag, in minutes
+* `tidepool_summary_queue_length` - (gauge) - the current summary queue length, in number of summaries
+
+### C2C
+
+#### Abbott
+
+* `tidepool_abbott_api_request_count` - (counter) - Abbott API request count, sorted by method, path, and status
+* `tidepool_abbott_api_request_duration_seconds` - (histogram) - Abbott API duration of each request, in seconds, sorted by method, path, and status
+
+#### Dexcom
+
+* `tidepool_dexcom_api_request_count` - (counter) - Dexcom API request count, sorted by method, path, and status
+* `tidepool_dexcom_api_request_duration_seconds` - (histogram) - Dexcom API duration of each request, in seconds, sorted by method, path, and status
+* `tidepool_dexcom_api_request_time_seconds` - (histogram) - Dexcom API duration of each request, as reported in the "request-time" response header from Dexcom, in seconds, sorted by method, path, and status
+
+#### Oura
+
+* `tidepool_oura_api_request_count` - (counter) - Oura API request count, sorted by method, path, and status
+* `tidepool_oura_api_request_duration_seconds` - (histogram) - Oura API duration of each request, in seconds, sorted by method, path, and status
+
+### Task
+
+#### Queue
+
+* `tidepool_task_workers_total` - (gauge) - configured number of task queue workers, sorted by queue (5, per config)
+* `tidepool_task_workers_available` - (gauge) - number of available task queue workers, sorted by queue (5, per config)
+* `tidepool_task_runner_not_found_total` - (counter) - total number of task runs with no registered runner for the task type, sorted by type (ideally zero)
+* `tidepool_task_run_duration_seconds` - (histogram) - duration of task runs in seconds, sorted by type
+* `tidepool_task_runner_timeout_exceeded_total` - (counter) - total number of task runs that exceeded the runner timeout, sorted by type and disposition ("blocked", "recovered") (ideally zero)
+* `tidepool_task_run_panic_total` - (counter) - total number of task runs that panicked, sorted by type (ideally 0)
+
+#### Store
+
+* `tidepool_task_type_state_total` - (counter) - total number of tasks run, sorted by type and state
+* `tidepool_task_type_lost_completion_total` - (counter) - total number of task completions dropped because the state-lock compare-and-swap missed, sorted by type (ideally low-ish)
+* `tidepool_task_type_revision_mismatch_total` - (counter) - total number of task revisions that do not match the task revision in the database, sorted by type (ideally zero)

auth/service/api/v1/metrics.go

@@ -0,0 +1,21 @@
+package v1
+
+import (
+	"net/http"
+
+	"github.com/ant0ine/go-json-rest/rest"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+func (r *Router) MetricsRoutes() []*rest.Route {
+	return []*rest.Route{
+		rest.Get("/v1/metrics", r.PrometheusMetrics),
+	}
+}
+
+func (r *Router) PrometheusMetrics(res rest.ResponseWriter, req *rest.Request) {
+	// The default go-json-rest middleware gzips the content
+	promhttp.HandlerFor(prometheus.DefaultGatherer, promhttp.HandlerOpts{DisableCompression: true}).
+		ServeHTTP(res.(http.ResponseWriter), req.Request)
+}

auth/service/api/v1/router.go

@@ -23,6 +23,7 @@ func NewRouter(svc service.Service) (*Router, error) {
 
 func (r *Router) Routes() []*rest.Route {
 	routes := [][]*rest.Route{
+		r.MetricsRoutes(),
 		r.OAuthRoutes(),
 		r.ProviderSessionsRoutes(),
 		r.RestrictedTokensRoutes(),

client/client.go

@@ -31,8 +31,7 @@ type ErrorResponseParser interface {
 }
 
 type Client struct {
-	address             string
-	userAgent           string
+	config              Config
 	errorResponseParser ErrorResponseParser
 }
 
@@ -48,14 +47,13 @@ func NewWithErrorParser(cfg *Config, errorResponseParser ErrorResponseParser) (*
 	}
 
 	return &Client{
-		address:             cfg.Address,
-		userAgent:           cfg.UserAgent,
+		config:              *cfg,
 		errorResponseParser: errorResponseParser,
 	}, nil
 }
 
 func (c *Client) ConstructURL(paths ...string) string {
-	return ConstructURL(c.address, paths...)
+	return ConstructURL(c.config.Address, paths...)
 }
 
 func (c *Client) AppendURLQuery(urlString string, query map[string]string) string {
@@ -87,6 +85,12 @@ func (c *Client) RequestStreamWithHTTPClient(ctx context.Context, method string,
 		return nil, err
 	}
 
+	if c.config.Timeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, c.config.Timeout)
+		defer cancel()
+	}
+
 	res, err := httpClient.Do(req)
 	if err != nil {
 		return nil, errors.Wrapf(err, "unable to perform request to %s %s", method, url)
@@ -127,8 +131,8 @@ func (c *Client) createRequest(ctx context.Context, method string, url string, m
 		return nil, errors.New("url is missing")
 	}
 
-	if c.userAgent != "" {
-		mutators = append(mutators, request.NewHeaderMutator("User-Agent", c.userAgent))
+	if c.config.UserAgent != "" {
+		mutators = append(mutators, request.NewHeaderMutator("User-Agent", c.config.UserAgent))
 	}
 
 	var body io.Reader

client/config.go

@@ -2,6 +2,8 @@ package client
 
 import (
 	"net/url"
+	"strconv"
+	"time"
 
 	"github.com/kelseyhightower/envconfig"
 
@@ -23,6 +25,9 @@ type Config struct {
 	//
 	// More info: https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent
 	UserAgent string `envconfig:"TIDEPOOL_USER_AGENT"`
+
+	// Timeout specifies the maximum amount of time a request can take. Zero means no timeout.
+	Timeout time.Duration
 }
 
 func NewConfig() *Config {
@@ -36,6 +41,13 @@ func (c *Config) Load(loader ConfigLoader) error {
 func (c *Config) LoadFromConfigReporter(reporter config.Reporter) error {
 	c.Address = reporter.GetWithDefault("address", c.Address)
 	c.UserAgent = reporter.GetWithDefault("user_agent", c.UserAgent)
+	if timeoutString, err := reporter.Get("timeout"); err == nil {
+		if timeout, parseErr := strconv.ParseInt(timeoutString, 10, 0); parseErr != nil {
+			return errors.New("timeout is invalid")
+		} else {
+			c.Timeout = time.Duration(timeout) * time.Second
+		}
+	}
 	return nil
 }
 
@@ -45,6 +57,9 @@ func (c *Config) Validate() error {
 	} else if _, err := url.Parse(c.Address); err != nil {
 		return errors.New("address is invalid")
 	}
+	if c.Timeout < 0 {
+		return errors.New("timeout is invalid")
+	}
 
 	return nil
 }

client/prometheus.go

@@ -0,0 +1,147 @@
+package client
+
+import (
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/tidepool-org/platform/pointer"
+)
+
+const PathPatternAny = "/"
+
+var (
+	DurationBucketsDefault = []float64{0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 15, 20, 30, 60}
+
+	PrometheusLabelNameMethod = "method"
+	PrometheusLabelNamePath   = "path"
+	PrometheusLabelNameStatus = "status"
+
+	PrometheusLabelValueError = "ERROR"
+)
+
+func PrometheusLabelNames() []string {
+	return []string{
+		PrometheusLabelNameMethod,
+		PrometheusLabelNamePath,
+		PrometheusLabelNameStatus,
+	}
+}
+
+// Where there are numerous discrete paths possible, then must simplify via patterns to prevent overwhelming Prometheus.
+// If no patterns are specified, then all paths are recorded as-is. If one or more patterns are specified, but a path
+// does not match one of the patterns, then the path is NOT captured by Prometheus. If you wish to match "all other" paths
+// and records those paths as-is, then add the pattern PathPatternAny at the end of your patterns.
+//
+// Uses standard Go HTTP pattern matching. See https://go.dev/src/net/http/pattern.go.
+//
+// For example: /one/{id}, /two/{id}
+type PrometheusRequestURLPathMatcher struct {
+	pathPatternMux *http.ServeMux
+}
+
+func NewPrometheusRequestURLPathPatternMatcher(pathPatterns ...string) *PrometheusRequestURLPathMatcher {
+	var pathPatternMux *http.ServeMux
+
+	if len(pathPatterns) > 0 {
+		pathPatternMux = http.NewServeMux()
+		for _, pattern := range pathPatterns {
+			pathPatternMux.HandleFunc(pattern, func(http.ResponseWriter, *http.Request) {})
+		}
+	}
+
+	return &PrometheusRequestURLPathMatcher{
+		pathPatternMux: pathPatternMux,
+	}
+}
+
+func (p *PrometheusRequestURLPathMatcher) MatchPath(req *http.Request) *string {
+	path := req.URL.Path
+	if p.pathPatternMux != nil {
+		if _, pattern := p.pathPatternMux.Handler(req); pattern == "" {
+			return nil
+		} else if pattern != PathPatternAny {
+			path = pattern
+		}
+	}
+	return &path
+}
+
+type PrometheusRequestRoundTripper struct {
+	*RoundTripper
+	*PrometheusRequestURLPathMatcher
+}
+
+func NewPrometheusRequestRoundTripper(pathPatterns ...string) *PrometheusRequestRoundTripper {
+	return &PrometheusRequestRoundTripper{
+		RoundTripper:                    NewRoundTripper(nil),
+		PrometheusRequestURLPathMatcher: NewPrometheusRequestURLPathPatternMatcher(pathPatterns...),
+	}
+}
+
+type PrometheusRequestMetricsRoundTripper struct {
+	*PrometheusRequestRoundTripper
+	requestCountCounterVec      *prometheus.CounterVec
+	requestDurationHistogramVec *prometheus.HistogramVec
+}
+
+func NewPrometheusRequestMetricsRoundTripper(name string, help string) *PrometheusRequestMetricsRoundTripper {
+	return NewPrometheusRequestMetricsRoundTripperWithPathPatternsAndDurationBuckets(name, help, nil, nil)
+}
+
+func NewPrometheusRequestMetricsRoundTripperWithPathPatternsAndDurationBuckets(name string, help string, pathPatterns []string, durationBuckets []float64) *PrometheusRequestMetricsRoundTripper {
+	return &PrometheusRequestMetricsRoundTripper{
+		PrometheusRequestRoundTripper: NewPrometheusRequestRoundTripper(pathPatterns...),
+		requestCountCounterVec: promauto.NewCounterVec(
+			prometheus.CounterOpts{
+				Name: fmt.Sprintf("%s_request_count", name),
+				Help: fmt.Sprintf("%s request count, sorted by method, path, and status", help),
+			},
+			PrometheusLabelNames(),
+		),
+		requestDurationHistogramVec: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name:    fmt.Sprintf("%s_request_duration_seconds", name),
+				Help:    fmt.Sprintf("%s request duration, in seconds, sorted by method, path, and status", help),
+				Buckets: pointer.DefaultArray(durationBuckets, DurationBucketsDefault),
+			},
+			PrometheusLabelNames(),
+		),
+	}
+}
+
+func (p *PrometheusRequestMetricsRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	start := time.Now()
+	res, err := p.PrometheusRequestRoundTripper.RoundTrip(req)
+	duration := time.Since(start)
+
+	if labels := p.Labels(req, res); labels != nil {
+		p.requestCountCounterVec.With(*labels).Inc()
+		p.requestDurationHistogramVec.With(*labels).Observe(duration.Seconds())
+	}
+
+	return res, err
+}
+
+func (p *PrometheusRequestMetricsRoundTripper) Labels(req *http.Request, res *http.Response) *prometheus.Labels {
+	path := p.MatchPath(req)
+	if path == nil {
+		return nil
+	}
+
+	labels := prometheus.Labels{
+		PrometheusLabelNameMethod: req.Method,
+		PrometheusLabelNamePath:   *path,
+	}
+	if res != nil {
+		labels[PrometheusLabelNameStatus] = strconv.Itoa(res.StatusCode)
+	} else {
+		labels[PrometheusLabelNameStatus] = PrometheusLabelValueError
+	}
+
+	return &labels
+}