TRAC-469: add support for wae metrics (#15)

Author: rtalvarez

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_<NAME>` |  `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_<name>` 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 `<WAE_DATASET_PREFIX><script_name_underscored>`. 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:

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")
 	}

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)

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
+}