SD-11577: add queue health metrics

Author: rtalvarez

README.md

@@ -133,6 +133,13 @@ Note: `ZONE_<name>` configuration is not supported as flag.
 # HELP cloudflare_kv_latency KV operation latency quantiles (milliseconds)
 # HELP cloudflare_worker_subrequests_count Number of subrequests by script name
 # HELP cloudflare_worker_subrequest_time Subrequest response time quantiles (microseconds)
+# HELP cloudflare_queue_backlog_messages Average number of messages in queue backlog
+# HELP cloudflare_queue_backlog_bytes Average backlog size in bytes
+# HELP cloudflare_queue_consumer_concurrency Average number of concurrent queue consumers
+# HELP cloudflare_queue_operations_count Number of queue message operations
+# HELP cloudflare_queue_operations_bytes Total bytes processed by queue message operations
+# HELP cloudflare_queue_operations_lag_time Average lag time between write and read/delete (milliseconds)
+# HELP cloudflare_queue_operations_retry_count Average retry count for queue message operations
 ```
 
 ## Helm chart repository

cloudflare.go

@@ -9,6 +9,7 @@ import (
 	cfaccounts "github.com/cloudflare/cloudflare-go/v4/accounts"
 	cfload_balancers "github.com/cloudflare/cloudflare-go/v4/load_balancers"
 	cfpagination "github.com/cloudflare/cloudflare-go/v4/packages/pagination"
+	cfqueues "github.com/cloudflare/cloudflare-go/v4/queues"
 	cfrulesets "github.com/cloudflare/cloudflare-go/v4/rulesets"
 	cfzero_trust "github.com/cloudflare/cloudflare-go/v4/zero_trust"
 	cfzones "github.com/cloudflare/cloudflare-go/v4/zones"
@@ -346,6 +347,47 @@ type subrequestsAccountResp struct {
 	} `json:"workersSubrequestsAdaptiveGroups"`
 }
 
+type cloudflareResponseQueues struct {
+	Viewer struct {
+		Accounts []queueAccountResp `json:"accounts"`
+	} `json:"viewer"`
+}
+
+type queueAccountResp struct {
+	QueueBacklogAdaptiveGroups []struct {
+		Dimensions struct {
+			QueueID string `json:"queueId"`
+		} `json:"dimensions"`
+		Avg struct {
+			Messages float64 `json:"messages"`
+			Bytes    float64 `json:"bytes"`
+		} `json:"avg"`
+	} `json:"queueBacklogAdaptiveGroups"`
+	QueueConsumerMetricsAdaptiveGroups []struct {
+		Dimensions struct {
+			QueueID string `json:"queueId"`
+		} `json:"dimensions"`
+		Avg struct {
+			Concurrency float64 `json:"concurrency"`
+		} `json:"avg"`
+	} `json:"queueConsumerMetricsAdaptiveGroups"`
+	QueueMessageOperationsAdaptiveGroups []struct {
+		Dimensions struct {
+			QueueID    string `json:"queueId"`
+			ActionType string `json:"actionType"`
+			Outcome    string `json:"outcome"`
+		} `json:"dimensions"`
+		Sum struct {
+			BillableOperations uint64 `json:"billableOperations"`
+			Bytes              uint64 `json:"bytes"`
+		} `json:"sum"`
+		Avg struct {
+			LagTime    float64 `json:"lagTime"`
+			RetryCount float64 `json:"retryCount"`
+		} `json:"avg"`
+	} `json:"queueMessageOperationsAdaptiveGroups"`
+}
+
 type cloudflareResponseDNSFirewall struct {
 	Viewer struct {
 		Accounts []dnsFirewallAccountResp `json:"accounts"`
@@ -1221,6 +1263,98 @@ func fetchWorkerSubrequests(accountID string) (*cloudflareResponseSubrequests, e
 	return &resp, nil
 }
 
+func fetchQueueNames(accountID string) (map[string]string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
+	defer cancel()
+	page := cfclient.Queues.ListAutoPaging(ctx,
+		cfqueues.QueueListParams{
+			AccountID: cf.F(accountID),
+		})
+	if page.Err() != nil {
+		return nil, page.Err()
+	}
+
+	names := make(map[string]string)
+	seenIDs := make(map[string]struct{})
+	for page.Next() {
+		if page.Err() != nil {
+			log.Errorf("error during paging queues: %v", page.Err())
+			break
+		}
+		q := page.Current()
+		if _, exists := seenIDs[q.QueueID]; exists {
+			log.Errorf("fetchQueueNames: duplicate queue ID detected (%s), breaking loop", q.QueueID)
+			break
+		}
+		seenIDs[q.QueueID] = struct{}{}
+		names[q.QueueID] = q.QueueName
+	}
+
+	return names, nil
+}
+
+func fetchQueueMetrics(accountID string) (*cloudflareResponseQueues, error) {
+	request := graphql.NewRequest(`
+	query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
+		viewer {
+			accounts(filter: {accountTag: $accountID}) {
+				queueBacklogAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
+					dimensions {
+						queueId
+					}
+					avg {
+						messages
+						bytes
+					}
+				}
+				queueConsumerMetricsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
+					dimensions {
+						queueId
+					}
+					avg {
+						concurrency
+					}
+				}
+				queueMessageOperationsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
+					dimensions {
+						queueId
+						actionType
+						outcome
+					}
+					sum {
+						billableOperations
+						bytes
+					}
+					avg {
+						lagTime
+						retryCount
+					}
+				}
+			}
+		}
+	}`)
+
+	now, now1mAgo := GetTimeRange()
+	request.Var("limit", gqlQueryLimit)
+	request.Var("maxtime", now)
+	request.Var("mintime", now1mAgo)
+	request.Var("accountID", accountID)
+
+	gql.Mu.RLock()
+	defer gql.Mu.RUnlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
+	defer cancel()
+
+	var resp cloudflareResponseQueues
+	if err := gql.Client.Run(ctx, request, &resp); err != nil {
+		log.Errorf("error fetching queue metrics, err:%v", err)
+		return nil, err
+	}
+
+	return &resp, nil
+}
+
 func fetchDNSFirewallTotals(accountID string) (*cloudflareResponseDNSFirewall, error) {
 	request := graphql.NewRequest(`
 query ($accountID: string, $mintime: Time!, $maxtime: Time!, $limit: Int!) {

main.go

@@ -116,6 +116,7 @@ func fetchMetrics(deniedMetricsSet MetricsSet) {
 		go fetchWorkerAnalytics(a, &wg)
 		go fetchKVAnalytics(a, &wg, deniedMetricsSet)
 		go fetchWorkerSubrequestAnalytics(a, &wg, deniedMetricsSet)
+		go fetchQueueAnalytics(a, &wg, deniedMetricsSet)
 		go fetchLogpushAnalyticsForAccount(a, &wg)
 		go fetchR2StorageForAccount(a, &wg)
 		go fetchLoadblancerPoolsHealth(a, &wg)

prometheus.go

@@ -68,6 +68,13 @@ const (
 	kvLatencyMetricName                          MetricName = "cloudflare_kv_latency"
 	workerSubrequestsCountMetricName             MetricName = "cloudflare_worker_subrequests_count"
 	workerSubrequestTimeMetricName               MetricName = "cloudflare_worker_subrequest_time"
+	queueBacklogMessagesMetricName               MetricName = "cloudflare_queue_backlog_messages"
+	queueBacklogBytesMetricName                  MetricName = "cloudflare_queue_backlog_bytes"
+	queueConsumerConcurrencyMetricName           MetricName = "cloudflare_queue_consumer_concurrency"
+	queueOperationsMetricName                    MetricName = "cloudflare_queue_operations_count"
+	queueOperationBytesMetricName                MetricName = "cloudflare_queue_operations_bytes"
+	queueOperationLagTimeMetricName              MetricName = "cloudflare_queue_operations_lag_time"
+	queueOperationRetryCountMetricName           MetricName = "cloudflare_queue_operations_retry_count"
 )
 
 type MetricsSet map[MetricName]struct{}
@@ -360,6 +367,41 @@ var (
 		Help: "Subrequest response time quantiles (microseconds)",
 	}, []string{"script_name", "account", "quantile"})
 
+	queueBacklogMessages = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueBacklogMessagesMetricName.String(),
+		Help: "Average number of messages in queue backlog",
+	}, []string{"queue_name", "account"})
+
+	queueBacklogBytes = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueBacklogBytesMetricName.String(),
+		Help: "Average backlog size in bytes",
+	}, []string{"queue_name", "account"})
+
+	queueConsumerConcurrency = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueConsumerConcurrencyMetricName.String(),
+		Help: "Average number of concurrent queue consumers",
+	}, []string{"queue_name", "account"})
+
+	queueOperations = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueOperationsMetricName.String(),
+		Help: "Number of queue message operations",
+	}, []string{"queue_name", "account", "action_type", "outcome"})
+
+	queueOperationBytes = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueOperationBytesMetricName.String(),
+		Help: "Total bytes processed by queue message operations",
+	}, []string{"queue_name", "account", "action_type", "outcome"})
+
+	queueOperationLagTime = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueOperationLagTimeMetricName.String(),
+		Help: "Average lag time between write and read/delete (milliseconds)",
+	}, []string{"queue_name", "account", "action_type", "outcome"})
+
+	queueOperationRetryCount = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: queueOperationRetryCountMetricName.String(),
+		Help: "Average retry count for queue message operations",
+	}, []string{"queue_name", "account", "action_type", "outcome"})
+
 	dnsFirewallQueryCount = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Name: dnsFirewallQueryCountMetricName.String(),
 		Help: "DNS Firewall query count by query type and response code",
@@ -416,6 +458,13 @@ func buildAllMetricsSet() MetricsSet {
 	allMetricsSet.Add(kvLatencyMetricName)
 	allMetricsSet.Add(workerSubrequestsCountMetricName)
 	allMetricsSet.Add(workerSubrequestTimeMetricName)
+	allMetricsSet.Add(queueBacklogMessagesMetricName)
+	allMetricsSet.Add(queueBacklogBytesMetricName)
+	allMetricsSet.Add(queueConsumerConcurrencyMetricName)
+	allMetricsSet.Add(queueOperationsMetricName)
+	allMetricsSet.Add(queueOperationBytesMetricName)
+	allMetricsSet.Add(queueOperationLagTimeMetricName)
+	allMetricsSet.Add(queueOperationRetryCountMetricName)
 	return allMetricsSet
 }
 
@@ -577,6 +626,27 @@ func mustRegisterMetrics(deniedMetrics MetricsSet) {
 	if !deniedMetrics.Has(workerSubrequestTimeMetricName) {
 		prometheus.MustRegister(workerSubrequestTime)
 	}
+	if !deniedMetrics.Has(queueBacklogMessagesMetricName) {
+		prometheus.MustRegister(queueBacklogMessages)
+	}
+	if !deniedMetrics.Has(queueBacklogBytesMetricName) {
+		prometheus.MustRegister(queueBacklogBytes)
+	}
+	if !deniedMetrics.Has(queueConsumerConcurrencyMetricName) {
+		prometheus.MustRegister(queueConsumerConcurrency)
+	}
+	if !deniedMetrics.Has(queueOperationsMetricName) {
+		prometheus.MustRegister(queueOperations)
+	}
+	if !deniedMetrics.Has(queueOperationBytesMetricName) {
+		prometheus.MustRegister(queueOperationBytes)
+	}
+	if !deniedMetrics.Has(queueOperationLagTimeMetricName) {
+		prometheus.MustRegister(queueOperationLagTime)
+	}
+	if !deniedMetrics.Has(queueOperationRetryCountMetricName) {
+		prometheus.MustRegister(queueOperationRetryCount)
+	}
 }
 
 func fetchLoadblancerPoolsHealth(account cfaccounts.Account, wg *sync.WaitGroup) {
@@ -706,6 +776,68 @@ func fetchWorkerSubrequestAnalytics(account cfaccounts.Account, wg *sync.WaitGro
 	}
 }
 
+func fetchQueueAnalytics(account cfaccounts.Account, wg *sync.WaitGroup, deniedMetricsSet MetricsSet) {
+	wg.Add(1)
+	defer wg.Done()
+
+	names, err := fetchQueueNames(account.ID)
+	if err != nil {
+		log.Error("failed to fetch queue names for account ", account.ID, ": ", err)
+		return
+	}
+
+	r, err := fetchQueueMetrics(account.ID)
+	if err != nil {
+		log.Error("failed to fetch queue metrics for account ", account.ID, ": ", err)
+		return
+	}
+
+	accountName := strings.ToLower(strings.ReplaceAll(account.Name, " ", "-"))
+
+	resolveQueueName := func(queueID string) string {
+		if name, ok := names[queueID]; ok {
+			return name
+		}
+		return queueID
+	}
+
+	for _, a := range r.Viewer.Accounts {
+		for _, b := range a.QueueBacklogAdaptiveGroups {
+			qName := resolveQueueName(b.Dimensions.QueueID)
+			if !deniedMetricsSet.Has(queueBacklogMessagesMetricName) {
+				queueBacklogMessages.With(prometheus.Labels{"queue_name": qName, "account": accountName}).Set(b.Avg.Messages)
+			}
+			if !deniedMetricsSet.Has(queueBacklogBytesMetricName) {
+				queueBacklogBytes.With(prometheus.Labels{"queue_name": qName, "account": accountName}).Set(b.Avg.Bytes)
+			}
+		}
+
+		for _, c := range a.QueueConsumerMetricsAdaptiveGroups {
+			qName := resolveQueueName(c.Dimensions.QueueID)
+			if !deniedMetricsSet.Has(queueConsumerConcurrencyMetricName) {
+				queueConsumerConcurrency.With(prometheus.Labels{"queue_name": qName, "account": accountName}).Set(c.Avg.Concurrency)
+			}
+		}
+
+		for _, op := range a.QueueMessageOperationsAdaptiveGroups {
+			qName := resolveQueueName(op.Dimensions.QueueID)
+			labels := prometheus.Labels{"queue_name": qName, "account": accountName, "action_type": op.Dimensions.ActionType, "outcome": op.Dimensions.Outcome}
+			if !deniedMetricsSet.Has(queueOperationsMetricName) {
+				queueOperations.With(labels).Add(float64(op.Sum.BillableOperations))
+			}
+			if !deniedMetricsSet.Has(queueOperationBytesMetricName) {
+				queueOperationBytes.With(labels).Add(float64(op.Sum.Bytes))
+			}
+			if !deniedMetricsSet.Has(queueOperationLagTimeMetricName) {
+				queueOperationLagTime.With(labels).Set(op.Avg.LagTime)
+			}
+			if !deniedMetricsSet.Has(queueOperationRetryCountMetricName) {
+				queueOperationRetryCount.With(labels).Set(op.Avg.RetryCount)
+			}
+		}
+	}
+}
+
 func fetchLogpushAnalyticsForAccount(account cfaccounts.Account, wg *sync.WaitGroup) {
 	wg.Add(1)
 	defer wg.Done()