feat: metrics for the tasks of background workers

The existing worker metrics apply to background workers as they are:
busy_workers counts a thread holding a task, from pickup to the close of
the task's stream, and worker_queue_depth counts the tasks waiting for a
thread, which is the only queue a background worker has. Two new ones
break tasks down: worker_task_count{worker,outcome} with completed,
aborted (the script ended with the task open), abandoned (the sender
closed its stream first) or timeout (no thread picked the task up in
time), settled by whichever side closes first so every task counts once,
and worker_task_time, the seconds spent on tasks from pickup to close.

The threads endpoint follows: a background thread is busy while it holds
a task and waiting otherwise, counted per thread since a script may hold
several.

Author: nicolas-grekas

caddy/caddy_test.go

@@ -866,7 +866,7 @@ func TestWorkerMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 2
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{server="",worker="` + workerName + `"} 0
 
@@ -1023,7 +1023,7 @@ func TestNamedWorkerMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 2
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
         # TYPE frankenphp_busy_workers gauge
         frankenphp_busy_workers{server="",worker="my_app"} 0
 
@@ -1119,7 +1119,7 @@ func TestAutoWorkerConfig(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads ` + workers + `
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{server="",worker="` + workerName + `"} 0
 
@@ -1386,6 +1386,7 @@ func TestMaxWaitTimeWorker(t *testing.T) {
 	require.NoError(t, err)
 
 	expectedMetrics := `
+	# HELP frankenphp_worker_queue_depth Number of queued requests for this worker, or of tasks waiting for a thread of a background worker
 	# TYPE frankenphp_worker_queue_depth gauge
 	frankenphp_worker_queue_depth{server="",worker="service"} 0
 	`
@@ -1486,7 +1487,7 @@ func TestMultiWorkersMetrics(t *testing.T) {
 	# TYPE frankenphp_busy_threads gauge
 	frankenphp_busy_threads 5
 
-	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+	# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 	# TYPE frankenphp_busy_workers gauge
 	frankenphp_busy_workers{server="",worker="service1"} 0
 

docs/metrics.md

@@ -16,13 +16,15 @@ When [Caddy metrics](https://caddyserver.com/docs/metrics) are enabled, FrankenP
 - `frankenphp_busy_threads`: The number of PHP threads currently processing a request (running workers always consume a thread).
 - `frankenphp_queue_depth`: The number of regular queued requests.
 - `frankenphp_total_workers{worker="[worker_name]",server="[server_name]"}`: The total number of workers.
-- `frankenphp_busy_workers{worker="[worker_name]",server="[server_name]"}`: The number of workers currently processing a request.
+- `frankenphp_busy_workers{worker="[worker_name]",server="[server_name]"}`: The number of workers currently processing a request, or a task for a background worker.
 - `frankenphp_worker_request_time{worker="[worker_name]",server="[server_name]"}`: The time spent processing requests by all workers.
 - `frankenphp_worker_request_count{worker="[worker_name]",server="[server_name]"}`: The number of requests processed by all workers.
 - `frankenphp_ready_workers{worker="[worker_name]",server="[server_name]"}`: The number of workers that have reached their ready point at least once: `frankenphp_handle_request()` for HTTP workers, `WorkerHandle::tick()` for background workers.
 - `frankenphp_worker_crashes{worker="[worker_name]",server="[server_name]"}`: The number of times a worker has unexpectedly terminated.
 - `frankenphp_worker_restarts{worker="[worker_name]",server="[server_name]"}`: The number of times a worker has been deliberately restarted.
-- `frankenphp_worker_queue_depth{worker="[worker_name]",server="[server_name]"}`: The number of queued requests.
+- `frankenphp_worker_queue_depth{worker="[worker_name]",server="[server_name]"}`: The number of queued requests, or of tasks waiting for a thread of a background worker.
+- `frankenphp_worker_task_count{worker="[worker_name]",server="[server_name]",outcome="[outcome]"}`: The number of tasks sent to a background worker, by outcome: `completed`, `aborted` (the script ended with the task open), `abandoned` (the sender closed its stream first) or `timeout` (no thread picked the task up in time).
+- `frankenphp_worker_task_time{worker="[worker_name]",server="[server_name]"}`: The time spent on tasks by all threads of a background worker, from pickup to the close of the task's stream.
 
 `[worker_name]` is the worker name from the Caddyfile, or the absolute path of the worker file when it has none, with a number appended when several workers share a script. `[server_name]` is the name of the `php_server` block the worker belongs to, and is empty for a worker declared in the global `frankenphp` block. The two stay apart, so a query on the worker name alone still selects that worker in every server.
 
@@ -73,8 +75,8 @@ Each entry in `ThreadDebugStates` contains:
 | `Index` | integer | The index of the thread. |
 | `Name` | string | The name of the thread (e.g., the worker file path). |
 | `State` | string | The internal state of the thread (e.g., `ready`, `shutting down`). |
-| `IsWaiting` | boolean | Whether the thread is waiting for a request. |
-| `IsBusy` | boolean | Whether the thread is currently processing a request. |
+| `IsWaiting` | boolean | Whether the thread is waiting for a request, or for a task in a background worker. |
+| `IsBusy` | boolean | Whether the thread is currently processing a request, or a task in a background worker. |
 | `WaitingSinceMilliseconds` | integer | How long the thread has been idle, in milliseconds. `0` if the thread is busy. |
 | `CurrentURI` | string | The URI currently being processed. Empty if the thread is idle. |
 | `CurrentMethod` | string | The HTTP method of the current request (e.g., `GET`, `POST`). Empty if the thread is idle. |

metrics.go

@@ -16,6 +16,16 @@ const (
 
 type StopReason int
 
+// TaskOutcome is how a task sent to a background worker ended
+type TaskOutcome string
+
+const (
+	TaskOutcomeCompleted TaskOutcome = "completed" // the worker closed the task's stream
+	TaskOutcomeAborted   TaskOutcome = "aborted"   // the worker's script ended with the task open
+	TaskOutcomeAbandoned TaskOutcome = "abandoned" // the sender closed its stream first
+	TaskOutcomeTimeout   TaskOutcome = "timeout"   // no thread picked the task up in time
+)
+
 // Metrics reports what the workers and the threads of a FrankenPHP instance
 // are doing. A worker is identified by its name alone, where a worker scoped
 // to a server is reported as "<server name>:<name>". An implementation that
@@ -61,6 +71,14 @@ type ServerMetrics interface {
 	StartWorkerRequestOnServer(name, server string)
 	QueuedWorkerRequestOnServer(name, server string)
 	DequeuedWorkerRequestOnServer(name, server string)
+	// the tasks of background workers have no packed counterpart in
+	// Metrics, so they take the pair under their own name
+	// StartWorkerTask collects tasks picked up by a thread of a background worker
+	StartWorkerTask(name, server string)
+	// StopWorkerTask collects tasks a thread of a background worker is done with
+	StopWorkerTask(name, server string, duration time.Duration)
+	// WorkerTaskOutcome collects how tasks sent to a background worker ended
+	WorkerTaskOutcome(name, server string, outcome TaskOutcome)
 }
 
 // workerMetrics is what the runtime reports on: Metrics with the worker
@@ -80,6 +98,9 @@ type workerMetrics interface {
 	DequeuedWorkerRequest(name, server string)
 	QueuedRequest()
 	DequeuedRequest()
+	StartWorkerTask(name, server string)
+	StopWorkerTask(name, server string, duration time.Duration)
+	WorkerTaskOutcome(name, server string, outcome TaskOutcome)
 }
 
 // metricsAdapter routes the worker methods to ServerMetrics when the
@@ -104,6 +125,26 @@ func packedWorkerName(name, server string) string {
 	return server + ":" + name
 }
 
+// the task methods have no packed form: an implementation that does not
+// satisfy ServerMetrics simply does not collect them
+func (a metricsAdapter) StartWorkerTask(name, server string) {
+	if a.server != nil {
+		a.server.StartWorkerTask(name, server)
+	}
+}
+
+func (a metricsAdapter) StopWorkerTask(name, server string, duration time.Duration) {
+	if a.server != nil {
+		a.server.StopWorkerTask(name, server, duration)
+	}
+}
+
+func (a metricsAdapter) WorkerTaskOutcome(name, server string, outcome TaskOutcome) {
+	if a.server != nil {
+		a.server.WorkerTaskOutcome(name, server, outcome)
+	}
+}
+
 func (a metricsAdapter) StartWorker(name, server string) {
 	if a.server != nil {
 		a.server.StartWorkerOnServer(name, server)
@@ -207,6 +248,12 @@ func (n nullMetrics) DequeuedWorkerRequest(string, string) {}
 func (n nullMetrics) QueuedRequest()   {}
 func (n nullMetrics) DequeuedRequest() {}
 
+func (n nullMetrics) StartWorkerTask(string, string) {}
+
+func (n nullMetrics) StopWorkerTask(string, string, time.Duration) {}
+
+func (n nullMetrics) WorkerTaskOutcome(string, string, TaskOutcome) {}
+
 type PrometheusMetrics struct {
 	registry           prometheus.Registerer
 	totalThreads       prometheus.Gauge
@@ -219,6 +266,8 @@ type PrometheusMetrics struct {
 	workerRequestTime  *prometheus.CounterVec
 	workerRequestCount *prometheus.CounterVec
 	workerQueueDepth   *prometheus.GaugeVec
+	workerTaskCount    *prometheus.CounterVec
+	workerTaskTime     *prometheus.CounterVec
 	queueDepth         prometheus.Gauge
 	mu                 sync.RWMutex
 }
@@ -341,7 +390,7 @@ func (m *PrometheusMetrics) TotalWorkersOnServer(string, string, int) {
 		m.busyWorkers = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Namespace: ns,
 			Name:      "busy_workers",
-			Help:      "Number of busy PHP workers for this worker",
+			Help:      "Number of busy PHP workers for this worker: processing a request, or a task for a background worker",
 		}, basicLabels)
 		m.mustRegister(m.busyWorkers)
 	}
@@ -389,9 +438,30 @@ func (m *PrometheusMetrics) TotalWorkersOnServer(string, string, int) {
 			Namespace: "frankenphp",
 			Subsystem: sub,
 			Name:      "queue_depth",
+			Help:      "Number of queued requests for this worker, or of tasks waiting for a thread of a background worker",
 		}, basicLabels)
 		m.mustRegister(m.workerQueueDepth)
 	}
+
+	if m.workerTaskCount == nil {
+		m.workerTaskCount = prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: ns,
+			Subsystem: sub,
+			Name:      "task_count",
+			Help:      "Number of tasks sent to this background worker, by outcome: completed, aborted (the script ended with the task open), abandoned (the sender closed its stream first) or timeout (no thread picked the task up in time)",
+		}, []string{"worker", "server", "outcome"})
+		m.mustRegister(m.workerTaskCount)
+	}
+
+	if m.workerTaskTime == nil {
+		m.workerTaskTime = prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: ns,
+			Subsystem: sub,
+			Name:      "task_time",
+			Help:      "Time spent on tasks by all threads of this background worker, from pickup to the close of the task's stream",
+		}, basicLabels)
+		m.mustRegister(m.workerTaskTime)
+	}
 }
 
 func (m *PrometheusMetrics) TotalThreads(num int) {
@@ -472,6 +542,37 @@ func (m *PrometheusMetrics) DequeuedRequest() {
 	m.queueDepth.Dec()
 }
 
+func (m *PrometheusMetrics) StartWorkerTask(name, server string) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	if m.busyWorkers == nil {
+		return
+	}
+	m.busyWorkers.WithLabelValues(name, server).Inc()
+}
+
+func (m *PrometheusMetrics) StopWorkerTask(name, server string, duration time.Duration) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	if m.workerTaskTime == nil {
+		return
+	}
+	m.busyWorkers.WithLabelValues(name, server).Dec()
+	m.workerTaskTime.WithLabelValues(name, server).Add(duration.Seconds())
+}
+
+func (m *PrometheusMetrics) WorkerTaskOutcome(name, server string, outcome TaskOutcome) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	if m.workerTaskCount == nil {
+		return
+	}
+	m.workerTaskCount.WithLabelValues(name, server, string(outcome)).Inc()
+}
+
 func (m *PrometheusMetrics) Shutdown() {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -511,6 +612,14 @@ func (m *PrometheusMetrics) Shutdown() {
 	if m.workerQueueDepth != nil {
 		m.registry.Unregister(m.workerQueueDepth)
 	}
+
+	if m.workerTaskCount != nil {
+		m.registry.Unregister(m.workerTaskCount)
+	}
+
+	if m.workerTaskTime != nil {
+		m.registry.Unregister(m.workerTaskTime)
+	}
 }
 
 func NewPrometheusMetrics(registry prometheus.Registerer) *PrometheusMetrics {
@@ -540,6 +649,8 @@ func NewPrometheusMetrics(registry prometheus.Registerer) *PrometheusMetrics {
 		workerCrashes:      nil,
 		readyWorkers:       nil,
 		workerQueueDepth:   nil,
+		workerTaskCount:    nil,
+		workerTaskTime:     nil,
 	}
 
 	m.mustRegister(m.totalThreads)

metrics_test.go

@@ -51,6 +51,8 @@ func TestPrometheusMetrics_TotalWorkers(t *testing.T) {
 	require.Nil(t, m.workerRestarts)
 	require.Nil(t, m.workerRequestTime)
 	require.Nil(t, m.workerRequestCount)
+	require.Nil(t, m.workerTaskCount)
+	require.Nil(t, m.workerTaskTime)
 
 	m.TotalWorkersOnServer("test_worker", "test_server", 2)
 
@@ -61,6 +63,65 @@ func TestPrometheusMetrics_TotalWorkers(t *testing.T) {
 	require.NotNil(t, m.workerRestarts)
 	require.NotNil(t, m.workerRequestTime)
 	require.NotNil(t, m.workerRequestCount)
+	require.NotNil(t, m.workerTaskCount)
+	require.NotNil(t, m.workerTaskTime)
+}
+
+func TestPrometheusMetrics_WorkerTask(t *testing.T) {
+	m := createPrometheusMetrics()
+	m.TotalWorkersOnServer("bg_worker", "bg_server", 1)
+	m.StartWorkerTask("bg_worker", "bg_server")
+	m.StopWorkerTask("bg_worker", "bg_server", 3*time.Second)
+	m.WorkerTaskOutcome("bg_worker", "bg_server", TaskOutcomeCompleted)
+	m.WorkerTaskOutcome("bg_worker", "bg_server", TaskOutcomeTimeout)
+
+	inputs := []struct {
+		name     string
+		c        prometheus.Collector
+		metadata string
+		expect   string
+	}{
+		{
+			name: "Testing BusyWorkers",
+			c:    m.busyWorkers,
+			metadata: `
+				# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
+				# TYPE frankenphp_busy_workers gauge
+			`,
+			expect: `
+				frankenphp_busy_workers{server="bg_server",worker="bg_worker"} 0
+			`,
+		},
+		{
+			name: "Testing WorkerTaskTime",
+			c:    m.workerTaskTime,
+			metadata: `
+				# HELP frankenphp_worker_task_time Time spent on tasks by all threads of this background worker, from pickup to the close of the task's stream
+				# TYPE frankenphp_worker_task_time counter
+			`,
+			expect: `
+				frankenphp_worker_task_time{server="bg_server",worker="bg_worker"} 3
+			`,
+		},
+		{
+			name: "Testing WorkerTaskCount",
+			c:    m.workerTaskCount,
+			metadata: `
+				# HELP frankenphp_worker_task_count Number of tasks sent to this background worker, by outcome: completed, aborted (the script ended with the task open), abandoned (the sender closed its stream first) or timeout (no thread picked the task up in time)
+				# TYPE frankenphp_worker_task_count counter
+			`,
+			expect: `
+				frankenphp_worker_task_count{outcome="completed",server="bg_server",worker="bg_worker"} 1
+				frankenphp_worker_task_count{outcome="timeout",server="bg_server",worker="bg_worker"} 1
+			`,
+		},
+	}
+
+	for _, input := range inputs {
+		t.Run(input.name, func(t *testing.T) {
+			require.NoError(t, testutil.CollectAndCompare(input.c, strings.NewReader(input.metadata+input.expect)))
+		})
+	}
 }
 
 func TestPrometheusMetrics_StopWorkerRequest(t *testing.T) {
@@ -89,7 +150,7 @@ func TestPrometheusMetrics_StopWorkerRequest(t *testing.T) {
 			name: "Testing BusyWorkers",
 			c:    m.busyWorkers,
 			metadata: `
-				# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+				# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 				# TYPE frankenphp_busy_workers gauge
 			`,
 			expect: `
@@ -132,7 +193,7 @@ func TestPrometheusMetrics_StartWorkerRequest(t *testing.T) {
 			name: "Testing BusyWorkers",
 			c:    m.busyWorkers,
 			metadata: `
-				# HELP frankenphp_busy_workers Number of busy PHP workers for this worker
+				# HELP frankenphp_busy_workers Number of busy PHP workers for this worker: processing a request, or a task for a background worker
 				# TYPE frankenphp_busy_workers gauge
 			`,
 			expect: `
@@ -253,6 +314,13 @@ func (m *splitMetrics) StopWorkerRequestOnServer(name, server string, _ time.Dur
 func (m *splitMetrics) StartWorkerRequestOnServer(name, server string)    { m.record(name, server) }
 func (m *splitMetrics) QueuedWorkerRequestOnServer(name, server string)   { m.record(name, server) }
 func (m *splitMetrics) DequeuedWorkerRequestOnServer(name, server string) { m.record(name, server) }
+func (m *splitMetrics) StartWorkerTask(name, server string)               { m.record(name, server) }
+func (m *splitMetrics) StopWorkerTask(name, server string, _ time.Duration) {
+	m.record(name, server)
+}
+func (m *splitMetrics) WorkerTaskOutcome(name, server string, _ TaskOutcome) {
+	m.record(name, server)
+}
 
 func (m *splitMetrics) StartWorker(string)            { m.t.Fatal("packed StartWorker called") }
 func (m *splitMetrics) ReadyWorker(string)            { m.t.Fatal("packed ReadyWorker called") }

threadbackgroundworker.go

@@ -50,6 +50,11 @@ type backgroundWorkerThread struct {
 	// after backgroundBootWarnDelay; only touched on the PHP thread
 	bootTimer *time.Timer
 
+	// openTasks counts the tasks picked up and not closed yet: the thread
+	// is busy rather than waiting on the threads endpoint meanwhile. Only
+	// touched on the PHP thread, pickup and close both happen there.
+	openTasks int
+
 	// stopSock holds the Go side's end of this thread's stop socket pair
 	// (per thread so pool workers drain independently); the other end is
 	// exposed to the script via WorkerHandle::getStream(). Wide enough
@@ -178,6 +183,7 @@ func (handler *backgroundWorkerThread) setupScript() error {
 
 	handler.isBootingScript = true
 	handler.runStartedAt = time.Now()
+	handler.openTasks = 0
 	metrics.StartWorker(handler.worker.name, handler.worker.server.name)
 	// the run's logger and context, not the globals: Stop() does not wait
 	// for a callback that already started, and a shutdown finishing