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 (per config)
+* `tidepool_task_workers_available` - (gauge) - number of available task queue workers, sorted by queue
+* `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 claim-token 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/client/external.go

@@ -3,7 +3,6 @@ package client
 import (
 	"context"
 	"net/http"
-	"strconv"
 	"sync"
 	"time"
 
@@ -16,6 +15,7 @@ import (
 	"github.com/tidepool-org/platform/auth"
 	"github.com/tidepool-org/platform/client"
 	"github.com/tidepool-org/platform/config"
+	"github.com/tidepool-org/platform/duration"
 	"github.com/tidepool-org/platform/errors"
 	"github.com/tidepool-org/platform/log"
 	"github.com/tidepool-org/platform/permission"
@@ -380,13 +380,10 @@ func (l *externalConfigReporterLoader) Load(cfg *ExternalConfig) error {
 		return err
 	}
 	cfg.ServerSessionTokenSecret = l.Reporter.GetWithDefault("server_session_token_secret", "")
-	if serverSessionTokenTimeoutString, err := l.Reporter.Get("server_session_token_timeout"); err == nil {
-		var serverSessionTokenTimeoutInteger int64
-		serverSessionTokenTimeoutInteger, err = strconv.ParseInt(serverSessionTokenTimeoutString, 10, 0)
-		if err != nil {
-			return errors.New("server session token timeout is invalid")
-		}
-		cfg.ServerSessionTokenTimeout = time.Duration(serverSessionTokenTimeoutInteger) * time.Second
+	if serverSessionTokenTimeout, err := duration.Parse(l.Reporter.GetWithDefault("server_session_token_timeout", cfg.ServerSessionTokenTimeout.String()), time.Second); err != nil {
+		return errors.New("server session token timeout is invalid")
+	} else {
+		cfg.ServerSessionTokenTimeout = serverSessionTokenTimeout
 	}
 
 	return nil

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(),

auth/service/service/client.go

@@ -88,8 +88,8 @@ func (c *Client) CreateProviderSession(ctx context.Context, create *auth.Provide
 
 	if err = prvdr.OnCreate(ctx, providerSession); err != nil {
 		log.LoggerFromContext(ctx).WithError(err).Error("Unable to finalize creation of provider session")
-		if err := c.deleteProviderSession(ctx, repository, providerSession); err != nil {
-			log.LoggerFromContext(ctx).WithError(err).Warn("Unable to delete provider session")
+		if deleteErr := c.deleteProviderSession(ctx, repository, providerSession); deleteErr != nil {
+			log.LoggerFromContext(ctx).WithError(deleteErr).Warn("Unable to delete provider session")
 		}
 		return nil, err
 	}

client/client.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"reflect"
 	"strings"
+	"time"
 	"unicode/utf8"
 
 	"github.com/tidepool-org/platform/errors"
@@ -31,8 +32,7 @@ type ErrorResponseParser interface {
 }
 
 type Client struct {
-	address             string
-	userAgent           string
+	config              Config
 	errorResponseParser ErrorResponseParser
 }
 
@@ -48,14 +48,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 {
@@ -78,25 +77,54 @@ func (c *Client) AppendURLQuery(urlString string, query map[string]string) strin
 }
 
 func (c *Client) RequestStreamWithHTTPClient(ctx context.Context, method string, url string, mutators []request.RequestMutator, requestBody interface{}, inspectors []request.ResponseInspector, httpClient *http.Client) (io.ReadCloser, error) {
+	if ctx == nil {
+		return nil, errors.New("context is missing")
+	}
 	if httpClient == nil {
 		return nil, errors.New("http client is missing")
 	}
 
+	// The request must carry a cancelable context for the timeout to reach the transport. A deadline on that context
+	// cannot be used, though, as it would also abort any read of the returned response body, so instead cancel via a
+	// timer that is stopped once the response headers arrive. The cause preserves the deadline exceeded error.
+	ctx, cancel := context.WithCancelCause(ctx)
+
 	req, err := c.createRequest(ctx, method, url, mutators, requestBody)
 	if err != nil {
+		cancel(nil)
 		return nil, err
 	}
 
+	var timer *time.Timer
+	if c.config.Timeout > 0 {
+		timer = time.AfterFunc(c.config.Timeout, func() { cancel(context.DeadlineExceeded) })
+	}
+
 	res, err := httpClient.Do(req)
+
+	if timer != nil {
+		timer.Stop()
+	}
+
 	if err != nil {
+		cancel(nil)
 		return nil, errors.Wrapf(err, "unable to perform request to %s %s", method, url)
 	}
 
 	for _, inspector := range inspectors {
 		inspector.InspectResponse(res)
 	}
 
-	return c.handleResponse(ctx, res, req)
+	body, err := c.handleResponse(ctx, res, req)
+	if body == nil {
+		cancel(nil)
+		return nil, err
+	}
+
+	return &ReadCloserWithCancelCause{
+		ReadCloser:      body,
+		CancelCauseFunc: cancel,
+	}, err
 }
 
 func (c *Client) RequestDataWithHTTPClient(ctx context.Context, method string, url string, mutators []request.RequestMutator, requestBody interface{}, responseBody interface{}, inspectors []request.ResponseInspector, httpClient *http.Client) error {
@@ -127,8 +155,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
@@ -247,6 +275,16 @@ func drainAndClose(reader io.ReadCloser) {
 	reader.Close()
 }
 
+type ReadCloserWithCancelCause struct {
+	io.ReadCloser
+	context.CancelCauseFunc
+}
+
+func (r *ReadCloserWithCancelCause) Close() error {
+	defer r.CancelCauseFunc(nil)
+	return r.ReadCloser.Close()
+}
+
 func NewSerializableErrorResponseParser() *SerializableErrorResponseParser {
 	return &SerializableErrorResponseParser{}
 }

client/client_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -773,6 +774,40 @@ var _ = Describe("Client", func() {
 						Expect(server.ReceivedRequests()).To(HaveLen(1))
 					})
 				})
+
+				Context("with a timeout", func() {
+					var timeout time.Duration
+
+					BeforeEach(func() {
+						timeout = 100 * time.Millisecond
+						config.Timeout = timeout
+					})
+
+					It("returns an error if the response is not received within the timeout", func() {
+						server.AppendHandlers(func(res http.ResponseWriter, req *http.Request) {
+							time.Sleep(3 * timeout)
+						})
+
+						reader, err = clnt.RequestStreamWithHTTPClient(ctx, method, url, mutators, requestBody, inspectors, httpClient)
+						Expect(errors.Is(errors.Cause(err), context.DeadlineExceeded)).To(BeTrue())
+						Expect(reader).To(BeNil())
+					})
+
+					It("does not apply the timeout to reading the response body", func() {
+						server.AppendHandlers(func(res http.ResponseWriter, req *http.Request) {
+							res.WriteHeader(http.StatusOK)
+							res.(http.Flusher).Flush()
+							time.Sleep(3 * timeout)
+							res.Write([]byte(responseString))
+						})
+
+						reader, err = clnt.RequestStreamWithHTTPClient(ctx, method, url, mutators, requestBody, inspectors, httpClient)
+						Expect(err).ToNot(HaveOccurred())
+						Expect(reader).ToNot(BeNil())
+						Expect(io.ReadAll(reader)).To(Equal([]byte(responseString)))
+						Expect(server.ReceivedRequests()).To(HaveLen(1))
+					})
+				})
 			})
 
 			Context("RequestDataWithHTTPClient", func() {

client/config.go

@@ -2,10 +2,12 @@ package client
 
 import (
 	"net/url"
+	"time"
 
 	"github.com/kelseyhightower/envconfig"
 
 	"github.com/tidepool-org/platform/config"
+	"github.com/tidepool-org/platform/duration"
 	"github.com/tidepool-org/platform/errors"
 )
 
@@ -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 `envconfig:"TIDEPOOL_CLIENT_TIMEOUT"`
 }
 
 func NewConfig() *Config {
@@ -36,6 +41,11 @@ 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 timeout, parseErr := duration.Parse(reporter.GetWithDefault("timeout", c.Timeout.String()), time.Second); parseErr != nil {
+		return errors.New("timeout is invalid")
+	} else {
+		c.Timeout = timeout
+	}
 	return nil
 }
 
@@ -45,6 +55,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
 }
@@ -70,6 +83,11 @@ func NewConfigReporterLoader(reporter config.Reporter) *configReporterLoader {
 func (l *configReporterLoader) Load(cfg *Config) error {
 	cfg.Address = l.Reporter.GetWithDefault("address", cfg.Address)
 	cfg.UserAgent = l.Reporter.GetWithDefault("user_agent", cfg.UserAgent)
+	if timeout, parseErr := duration.Parse(l.Reporter.GetWithDefault("timeout", cfg.Timeout.String()), time.Second); parseErr != nil {
+		return errors.New("timeout is invalid")
+	} else {
+		cfg.Timeout = timeout
+	}
 	return nil
 }