From 4526ec96866032b07eea21c9ba74b1023d856c67 Mon Sep 17 00:00:00 2001 From: Roberto Alvarez Date: Thu, 19 Mar 2026 12:46:31 -0500 Subject: [PATCH] SD-11577: add queue health metrics --- README.md | 7 +++ cloudflare.go | 134 ++++++++++++++++++++++++++++++++++++++++++++++++++ main.go | 1 + prometheus.go | 132 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 274 insertions(+) diff --git a/README.md b/README.md index 90ce12f..acc5aa9 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,13 @@ Note: `ZONE_` 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 diff --git a/cloudflare.go b/cloudflare.go index 696506b..cb3fce0 100644 --- a/cloudflare.go +++ b/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!) { diff --git a/main.go b/main.go index 09c79af..0c7530b 100644 --- a/main.go +++ b/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) diff --git a/prometheus.go b/prometheus.go index 54beae9..7693602 100644 --- a/prometheus.go +++ b/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).Set(float64(op.Sum.BillableOperations)) + } + if !deniedMetricsSet.Has(queueOperationBytesMetricName) { + queueOperationBytes.With(labels).Set(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()