Refactor task queue to resolve task-related failures

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", help),
+			},
+			PrometheusLabelNames(),
+		),
+		requestDurationHistogramVec: promauto.NewHistogramVec(
+			prometheus.HistogramOpts{
+				Name:    fmt.Sprintf("%s_request_duration_seconds", name),
+				Help:    fmt.Sprintf("%s request duration (seconds)", 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
+}