Add new metrics for jobs in flight (#859)

metrics/metrics.go

@@ -30,6 +30,9 @@ type CatalystAPIMetrics struct {
 	CDNRedirectCount            *prometheus.CounterVec
 	CDNRedirectWebRTC406        *prometheus.CounterVec
 
+	JobsInFlight         prometheus.Gauge
+	HTTPRequestsInFlight prometheus.Gauge
+
 	TranscodingStatusUpdate ClientMetrics
 	BroadcasterClient       ClientMetrics
 	MistClient              ClientMetrics
@@ -48,6 +51,16 @@ func NewMetrics() *CatalystAPIMetrics {
 			Help: "Current Git SHA / Tag that's running. Incremented once on app startup.",
 		}, []string{"app", "version"}),
 
+		// Node-level Capacity Metrics
+		JobsInFlight: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "jobs_in_flight",
+			Help: "A count of the jobs in flight",
+		}),
+		HTTPRequestsInFlight: promauto.NewGauge(prometheus.GaugeOpts{
+			Name: "http_requests_in_flight",
+			Help: "A count of the http requests in flight",
+		}),
+
 		// /api/vod request metrics
 		UploadVODRequestCount: promauto.NewCounter(prometheus.CounterOpts{
 			Name: "upload_vod_request_count",

middleware/capacity.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/julienschmidt/httprouter"
 	"github.com/livepeer/catalyst-api/config"
+	"github.com/livepeer/catalyst-api/metrics"
 	"github.com/livepeer/catalyst-api/pipeline"
 )
 
@@ -15,6 +16,10 @@ type CapacityMiddleware struct {
 
 func (c *CapacityMiddleware) HasCapacity(vodEngine *pipeline.Coordinator, next httprouter.Handle) httprouter.Handle {
 	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
+		// Keep a gauge of HTTP requests in flight
+		metrics.Metrics.HTTPRequestsInFlight.Add(1)
+		defer metrics.Metrics.HTTPRequestsInFlight.Add(-1)
+
 		requestsInFlight := c.requestsInFlight.Add(1)
 		defer c.requestsInFlight.Add(-1)
 

pipeline/coordinator.go

@@ -254,6 +254,7 @@ func (c *Coordinator) StartUploadJob(p UploadJobPayload) {
 	si.ReportProgress(clients.TranscodeStatusPreparing, 0)
 	c.Jobs.Store(streamName, si)
 	log.Log(si.RequestID, "Wrote to jobs cache")
+	metrics.Metrics.JobsInFlight.Set(float64(len(c.Jobs.GetKeys())))
 
 	c.runHandlerAsync(si, func() (*HandlerOutput, error) {
 		sourceURL, err := url.Parse(p.SourceFile)
@@ -467,6 +468,7 @@ func (c *Coordinator) startOneUploadJob(si *JobInfo, handler Handler, hasFallbac
 	si.ReportProgress(clients.TranscodeStatusPreparing, 0)
 	c.Jobs.Store(si.StreamName, si)
 	log.Log(si.RequestID, "Wrote to jobs cache")
+	metrics.Metrics.JobsInFlight.Set(float64(len(c.Jobs.GetKeys())))
 
 	c.runHandlerAsync(si, func() (*HandlerOutput, error) {
 		return si.handler.HandleStartUploadJob(si)
@@ -517,8 +519,8 @@ func (c *Coordinator) finishJob(job *JobInfo, out *HandlerOutput, err error) {
 	// Automatically delete jobs after an error or result
 	success := err == nil && err2 == nil
 	c.Jobs.Remove(job.StreamName)
-
 	log.Log(job.RequestID, "Finished job and deleted from job cache", "success", success)
+	metrics.Metrics.JobsInFlight.Set(float64(len(c.Jobs.GetKeys())))
 
 	var labels = []string{
 		job.sourceCodecVideo,

test/cucumber_test.go

@@ -69,6 +69,7 @@ func InitializeScenario(ctx *godog.ScenarioContext) {
 	ctx.Step(`^I get an HTTP response with code "([^"]*)" and the following body "([^"]*)"$`, stepContext.CheckHTTPResponseCodeAndBody)
 	ctx.Step(`^Mist calls the "([^"]*)" trigger with "([^"]*)" and ID "([^"]*)"$`, stepContext.CreateTriggerRequest)
 	ctx.Step(`^my "(failed|successful)" (vod|playback) request metrics get recorded$`, stepContext.CheckRecordedMetrics)
+	ctx.Step(`^a "([^"]*)" metric is recorded with a value of "([^"]*)"$`, stepContext.CheckMetricEqual)
 	ctx.Step(`^the body matches file "([^"]*)"$`, stepContext.CheckHTTPResponseBodyFromFile)
 	ctx.Step(`^the gate API will (allow|deny) playback$`, stepContext.SetGateAPIResponse)
 	ctx.Step(`^the gate API will be called (\d+) times$`, stepContext.CheckGateAPICallCount)

test/features/vod.feature

@@ -36,6 +36,7 @@ Feature: VOD Streaming
     Then I get an HTTP response with code "200"
     And I receive a Request ID in the response body
     And the source playback manifest is written to storage within "5" seconds
+    And a "jobs_in_flight" metric is recorded with a value of "1"
     And my "successful" vod request metrics get recorded
     And "4" source segments are written to storage within "5" seconds
     And the source manifest is written to storage within "3" seconds and contains "4" segments

test/steps/http.go

@@ -330,6 +330,30 @@ func (s *StepContext) CheckGateAPICallCount(expectedCount int) error {
 	return nil
 }
 
+func (s *StepContext) CheckMetricEqual(metricName, value string) error {
+	var url = s.BaseInternalURL + "/metrics"
+
+	res, err := http.Get(url)
+	if err != nil {
+		return err
+	}
+
+	body, err := io.ReadAll(res.Body)
+	if err != nil {
+		return err
+	}
+
+	r := regexp.MustCompile(`\n` + metricName + ` ` + value + `\n`)
+	if !r.Match(body) {
+		// Try to build a more useful error message than "not found" in the case where the value is wrong
+		r2 := regexp.MustCompile(`\n` + metricName + ` \d\n`)
+		found := r2.Find(body)
+
+		return fmt.Errorf("could not find metric %s equal to %s. Got: %s", metricName, value, strings.TrimSpace(string(found)))
+	}
+	return nil
+}
+
 func (s *StepContext) CheckRecordedMetrics(metricsType, requestType string) error {
 	var url = s.BaseInternalURL + "/metrics"