From 8ab6a726a8461bab2e35a1bf65e5f95f0e4172ff Mon Sep 17 00:00:00 2001 From: Roberto Alvarez Date: Mon, 18 May 2026 09:26:57 -0500 Subject: [PATCH] TRAC-469: add support for wae metrics --- README.md | 33 +++++++++++++++++- main.go | 8 +++++ prometheus.go | 94 +++++++++++++++++++++++++++++++++++++++++++++++++++ wae.go | 71 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 205 insertions(+), 1 deletion(-) create mode 100644 wae.go diff --git a/README.md b/README.md index acc5aa9..8f45f3a 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ dashboard. Required authentication scopes: - `Zone/Analytics:Read` is required for zone-level metrics -- `Account/Account Analytics:Read` is required for Worker metrics +- `Account/Account Analytics:Read` is required for Worker metrics and Workers Analytics Engine (WAE) queries - `Account/Account Settings:Read` is required for Worker metrics (for listing accessible accounts, scraping all available Workers included in authentication scope) - `Zone/Firewall Services:Read` is required to fetch zone rule name for `cloudflare_zone_firewall_events_count` metric @@ -67,6 +67,7 @@ The exporter can be configured using env variables or command flags. | `SCRAPE_INTERVAL` | scrape interval in seconds (will query cloudflare every SCRAPE_INTERVAL seconds), default `60` | | `METRICS_DENYLIST` | (Optional) cloudflare-exporter metrics to not export, comma delimited list of cloudflare-exporter metrics. If not set, all metrics are exported | | `KV_NAMESPACE_IDS` | (Optional) KV namespace IDs to track individually, comma delimited. Unlisted namespaces are aggregated as `other` | +| `WAE_DATASET_PREFIX` | (Optional) Prefix used to discover Workers Analytics Engine (WAE) datasets via `SHOW TABLES`. Datasets matching this prefix are auto-discovered and queried; the suffix (with `_` → `-`) becomes the `script_name` label. Empty (default) disables WAE discovery. See [WAE integration](#workers-analytics-engine-wae-integration) below. | | `ENABLE_PPROF` | (Optional) enable pprof profiling endpoints at `/debug/pprof/`. Accepts `true` or `false`, default `false`. **Warning**: Only enable in development/debugging environments | | `ZONE_` | `DEPRECATED since 0.0.5` (optional) Zone ID. Add zones you want to scrape by adding env vars in this format. You can find the zone ids in Cloudflare dashboards. | | `LOG_LEVEL` | Set loglevel. Options are error, warn, info, debug. default `error` | @@ -86,6 +87,7 @@ Corresponding flags: -scrape_delay=300: scrape delay in seconds, defaults to 300 -scrape_interval=60: scrape interval in seconds, defaults to 60 -kv_namespace_ids="": KV namespace IDs to track individually, comma delimited + -wae_dataset_prefix="": prefix for auto-discovered WAE datasets (empty disables discovery) -metrics_denylist="": cloudflare-exporter metrics to not export, comma delimited list -enable_pprof=false: enable pprof profiling endpoints at /debug/pprof/ -log_level="error": log level(error,warn,info,debug) @@ -140,8 +142,37 @@ Note: `ZONE_` configuration is not supported as flag. # 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 +# HELP cloudflare_worker_operation_duration_seconds Average duration of a worker operation in seconds, sliced by metric_type (worker-defined event type) and label (worker-defined slice) ``` +## Workers Analytics Engine (WAE) integration + +The exporter can scrape Cloudflare Workers Analytics Engine datasets and expose them as Prometheus metrics. WAE datasets are auto-discovered by name prefix — no per-worker configuration in the exporter is required. + +Discovery is **disabled by default**. Set `WAE_DATASET_PREFIX` (e.g. `my_worker_metrics_`) to enable. + +### Onboarding a new worker + +1. In your worker's wrangler config, define a WAE binding whose dataset name follows the convention ``. For example, with a configured prefix of `my_worker_metrics_`, a worker called `my-service-prod` names its dataset `my_worker_metrics_my_service_prod`. WAE dataset names must be valid SQL identifiers, so dashes are forbidden; the exporter converts the suffix back to dashes when emitting the `script_name` label. + +2. Emit data points with the following schema: + + | Field | Meaning | + |---|---| + | `index1` | metric_type string — categorizes the operation (e.g. `origin_fetch`, `api_call`) | + | `blob1` | label — secondary slicing dimension (e.g. status, endpoint) | + | `double1` | duration in milliseconds | + + Additional `blobN` / `doubleN` fields are ignored by the exporter; they're for your own diagnostic queries against WAE. + +3. Deploy. On the next scrape cycle the exporter will discover the dataset and expose `cloudflare_worker_operation_duration_seconds` series labeled with your `script_name`, `metric_type` (from `index1`), and `label` (from `blob1`). + +### Notes + +- The WAE aggregation window is 5 minutes. If your worker is low-traffic, expect series to disappear when traffic doesn't land in the window. +- Series for a `script_name` whose dataset no longer exists (or has had no traffic in the window) are automatically removed from `/metrics` — they don't go stale at their last value. +- Discovery uses `SHOW TABLES` against the WAE SQL API and requires `Account/Account Analytics:Read`. + ## Helm chart repository To deploy the exporter into Kubernetes, we recommend using our manager Helm repository: diff --git a/main.go b/main.go index 0c7530b..f0de3d3 100644 --- a/main.go +++ b/main.go @@ -24,6 +24,7 @@ var ( cfclient *cf.Client cftimeout time.Duration gql *GraphQL + waeClient *WAEClient log = logrus.New() // kvTrackedNamespaces is the set of KV namespace IDs that get their own @@ -122,6 +123,7 @@ func fetchMetrics(deniedMetricsSet MetricsSet) { go fetchLoadblancerPoolsHealth(a, &wg) go fetchZeroTrustAnalyticsForAccount(a, &wg) go fetchDNSFirewallAnalytics(a, &wg, deniedMetricsSet) + go fetchWorkerWAEAnalytics(a, &wg, deniedMetricsSet) } zones := fetchZones(accounts) @@ -283,6 +285,10 @@ func main() { viper.BindEnv("metrics_denylist") viper.SetDefault("metrics_denylist", "") + flags.String("wae_dataset_prefix", "", "Prefix used to discover Cloudflare Workers Analytics Engine datasets via SHOW TABLES. The remainder of each dataset name (with underscores converted to dashes) is emitted as the script_name label. Empty (default) disables WAE discovery.") + viper.BindEnv("wae_dataset_prefix") + viper.SetDefault("wae_dataset_prefix", "") + flags.String("log_level", "info", "log level") viper.BindEnv("log_level") viper.SetDefault("log_level", "info") @@ -330,6 +336,7 @@ func main() { Transport: middlewares, } gql = NewGraphQLClient(gqlHTTPClient) + waeClient = NewWAEClient(gqlHTTPClient) } else if len(viper.GetString("cf_api_email")) > 0 && len(viper.GetString("cf_api_key")) > 0 { cfclient = cf.NewClient( cfoption.WithAPIKey(viper.GetString("cf_api_key")), @@ -343,6 +350,7 @@ func main() { Transport: middlewares, } gql = NewGraphQLClient(gqlHTTPClient) + waeClient = NewWAEClient(gqlHTTPClient) } else { log.Fatal("Please provide CF_API_KEY+CF_API_EMAIL or CF_API_TOKEN") } diff --git a/prometheus.go b/prometheus.go index 7693602..3b9cad0 100644 --- a/prometheus.go +++ b/prometheus.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "strconv" "strings" @@ -75,6 +76,7 @@ const ( queueOperationBytesMetricName MetricName = "cloudflare_queue_operations_bytes" queueOperationLagTimeMetricName MetricName = "cloudflare_queue_operations_lag_time" queueOperationRetryCountMetricName MetricName = "cloudflare_queue_operations_retry_count" + workerOperationDurationMetricName MetricName = "cloudflare_worker_operation_duration_seconds" ) type MetricsSet map[MetricName]struct{} @@ -407,6 +409,11 @@ var ( Help: "DNS Firewall query count by query type and response code", }, []string{"account_id", "account_name", "dns_firewall_id", "query_type", "response_code"}, ) + + workerOperationDuration = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: workerOperationDurationMetricName.String(), + Help: "Average duration of a worker operation in seconds, sliced by metric_type (worker-defined event type) and label (worker-defined slice)", + }, []string{"script_name", "account", "metric_type", "label"}) ) func buildAllMetricsSet() MetricsSet { @@ -465,6 +472,7 @@ func buildAllMetricsSet() MetricsSet { allMetricsSet.Add(queueOperationBytesMetricName) allMetricsSet.Add(queueOperationLagTimeMetricName) allMetricsSet.Add(queueOperationRetryCountMetricName) + allMetricsSet.Add(workerOperationDurationMetricName) return allMetricsSet } @@ -647,6 +655,9 @@ func mustRegisterMetrics(deniedMetrics MetricsSet) { if !deniedMetrics.Has(queueOperationRetryCountMetricName) { prometheus.MustRegister(queueOperationRetryCount) } + if !deniedMetrics.Has(workerOperationDurationMetricName) { + prometheus.MustRegister(workerOperationDuration) + } } func fetchLoadblancerPoolsHealth(account cfaccounts.Account, wg *sync.WaitGroup) { @@ -1348,6 +1359,89 @@ func addCloudflareTunnelStatus(account cfaccounts.Account) { } } +// fetchWorkerWAEAnalytics discovers all WAE datasets in the account that match +// the configured prefix and queries each one for worker metrics. The dataset +// name's suffix (with underscores converted to dashes) is emitted as the +// script_name label so the series are joinable with cloudflare_worker_* metrics. +func fetchWorkerWAEAnalytics(account cfaccounts.Account, wg *sync.WaitGroup, deniedMetricsSet MetricsSet) { + wg.Add(1) + defer wg.Done() + + prefix := viper.GetString("wae_dataset_prefix") + if prefix == "" { + return + } + + ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + defer cancel() + + listResp, err := waeClient.Query(ctx, account.ID, "SHOW TABLES") + if err != nil { + log.Error("failed to list WAE datasets for account ", account.ID, ": ", err) + return + } + + var datasets []string + for _, row := range listResp.Data { + name, _ := row["dataset"].(string) + if strings.HasPrefix(name, prefix) { + datasets = append(datasets, name) + } + } + if len(datasets) == 0 { + log.Debugf("no WAE datasets matched prefix %q in account %s", prefix, account.ID) + return + } + + accountName := strings.ToLower(strings.ReplaceAll(account.Name, " ", "-")) + + // Drop this account's previous label combinations so labels that no longer + // see traffic (or scripts that no longer exist) disappear instead of going + // stale at their last value. + workerOperationDuration.DeletePartialMatch(prometheus.Labels{"account": accountName}) + + for _, dataset := range datasets { + scriptName := strings.ReplaceAll(strings.TrimPrefix(dataset, prefix), "_", "-") + + query := fmt.Sprintf(` + SELECT + index1 AS metric_type, + blob1 AS label, + SUM(_sample_interval * double1) / SUM(_sample_interval) AS avg_duration_ms, + SUM(_sample_interval) AS total_requests + FROM %s + WHERE timestamp > NOW() - INTERVAL '5' MINUTE + GROUP BY index1, blob1 + FORMAT JSON`, dataset) + + resp, err := waeClient.Query(ctx, account.ID, query) + if err != nil { + log.Errorf("failed to fetch WAE analytics from %s (account %s): %v", dataset, account.ID, err) + continue + } + + if deniedMetricsSet.Has(workerOperationDurationMetricName) { + continue + } + + for _, row := range resp.Data { + metricType, _ := row["metric_type"].(string) + label, _ := row["label"].(string) + avgDurationMs, _ := row["avg_duration_ms"].(float64) + if metricType == "" { + continue + } + + workerOperationDuration.With(prometheus.Labels{ + "script_name": scriptName, + "account": accountName, + "metric_type": metricType, + "label": label, + }).Set(avgDurationMs / 1000.0) + } + } +} + // The status of the tunnel. // Valid values are: // - inactive (tunnel has never been run) diff --git a/wae.go b/wae.go new file mode 100644 index 0000000..399a9e3 --- /dev/null +++ b/wae.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +const waeAPIBase = "https://api.cloudflare.com/client/v4/accounts" + +// WAEClient queries the Workers Analytics Engine SQL API. +type WAEClient struct { + httpClient *http.Client +} + +// WAEResponse is the JSON response from the WAE SQL API. +// This is NOT the standard Cloudflare v4 API envelope. +type WAEResponse struct { + Meta []WAEColumnMeta `json:"meta"` + Data []map[string]interface{} `json:"data"` + Rows int `json:"rows"` +} + +type WAEColumnMeta struct { + Name string `json:"name"` + Type string `json:"type"` +} + +func NewWAEClient(httpClient *http.Client) *WAEClient { + if httpClient == nil { + httpClient = http.DefaultClient + } + return &WAEClient{httpClient: httpClient} +} + +// Query executes a SQL query against the WAE API for the given account. +func (w *WAEClient) Query(ctx context.Context, accountID, query string) (*WAEResponse, error) { + url := fmt.Sprintf("%s/%s/analytics_engine/sql", waeAPIBase, accountID) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, strings.NewReader(query)) + if err != nil { + return nil, fmt.Errorf("creating WAE request: %w", err) + } + + log.Debugf("WAE query (account=%s): %s", accountID, query) + + resp, err := w.httpClient.Do(req) + if err != nil { + return nil, fmt.Errorf("executing WAE query: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading WAE response: %w", err) + } + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("WAE query failed (HTTP %d): %s", resp.StatusCode, body) + } + + var result WAEResponse + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("parsing WAE response: %w", err) + } + + return &result, nil +}