SD-11577: add worker subrequest support (#12)

Author: rtalvarez

README.md

@@ -131,6 +131,8 @@ Note: `ZONE_<name>` configuration is not supported as flag.
 # HELP cloudflare_r2_storage_total_bytes Total storage used by R2
 # HELP cloudflare_kv_requests_count Number of KV operations by namespace and action type
 # 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)
 ```
 
 ## Helm chart repository

cloudflare.go

@@ -323,6 +323,29 @@ type kvAccountResp struct {
 	} `json:"kvOperationsAdaptiveGroups"`
 }
 
+type cloudflareResponseSubrequests struct {
+	Viewer struct {
+		Accounts []subrequestsAccountResp `json:"accounts"`
+	} `json:"viewer"`
+}
+
+type subrequestsAccountResp struct {
+	WorkersSubrequestsAdaptiveGroups []struct {
+		Dimensions struct {
+			ScriptName string `json:"scriptName"`
+		} `json:"dimensions"`
+		Sum struct {
+			Subrequests uint64 `json:"subrequests"`
+		} `json:"sum"`
+		Quantiles struct {
+			TimeToResponseUsP50  float32 `json:"timeToResponseUsP50"`
+			TimeToResponseUsP75  float32 `json:"timeToResponseUsP75"`
+			TimeToResponseUsP99  float32 `json:"timeToResponseUsP99"`
+			TimeToResponseUsP999 float32 `json:"timeToResponseUsP999"`
+		} `json:"quantiles"`
+	} `json:"workersSubrequestsAdaptiveGroups"`
+}
+
 type cloudflareResponseDNSFirewall struct {
 	Viewer struct {
 		Accounts []dnsFirewallAccountResp `json:"accounts"`
@@ -1154,6 +1177,50 @@ func fetchKVOperations(accountID string) (*cloudflareResponseKV, error) {
 	return &resp, nil
 }
 
+func fetchWorkerSubrequests(accountID string) (*cloudflareResponseSubrequests, error) {
+	request := graphql.NewRequest(`
+	query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
+		viewer {
+			accounts(filter: {accountTag: $accountID}) {
+				workersSubrequestsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
+					dimensions {
+						scriptName
+					}
+					sum {
+						subrequests
+					}
+					quantiles {
+						timeToResponseUsP50
+						timeToResponseUsP75
+						timeToResponseUsP99
+						timeToResponseUsP999
+					}
+				}
+			}
+		}
+	}`)
+
+	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 cloudflareResponseSubrequests
+	if err := gql.Client.Run(ctx, request, &resp); err != nil {
+		log.Errorf("error fetching worker subrequests, 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

@@ -115,6 +115,7 @@ func fetchMetrics(deniedMetricsSet MetricsSet) {
 	for _, a := range accounts {
 		go fetchWorkerAnalytics(a, &wg)
 		go fetchKVAnalytics(a, &wg, deniedMetricsSet)
+		go fetchWorkerSubrequestAnalytics(a, &wg, deniedMetricsSet)
 		go fetchLogpushAnalyticsForAccount(a, &wg)
 		go fetchR2StorageForAccount(a, &wg)
 		go fetchLoadblancerPoolsHealth(a, &wg)

prometheus.go

@@ -66,6 +66,8 @@ const (
 	dnsFirewallQueryCountMetricName              MetricName = "cloudflare_dns_firewall_query_count"
 	kvRequestsMetricName                         MetricName = "cloudflare_kv_requests_count"
 	kvLatencyMetricName                          MetricName = "cloudflare_kv_latency"
+	workerSubrequestsCountMetricName             MetricName = "cloudflare_worker_subrequests_count"
+	workerSubrequestTimeMetricName               MetricName = "cloudflare_worker_subrequest_time"
 )
 
 type MetricsSet map[MetricName]struct{}
@@ -348,6 +350,16 @@ var (
 		Help: "KV operation latency quantiles (milliseconds)",
 	}, []string{"namespace_id", "action_type", "account", "quantile"})
 
+	workerSubrequestsCount = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: workerSubrequestsCountMetricName.String(),
+		Help: "Number of subrequests by script name",
+	}, []string{"script_name", "account"})
+
+	workerSubrequestTime = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: workerSubrequestTimeMetricName.String(),
+		Help: "Subrequest response time quantiles (microseconds)",
+	}, []string{"script_name", "account", "quantile"})
+
 	dnsFirewallQueryCount = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 		Name: dnsFirewallQueryCountMetricName.String(),
 		Help: "DNS Firewall query count by query type and response code",
@@ -402,6 +414,8 @@ func buildAllMetricsSet() MetricsSet {
 	allMetricsSet.Add(dnsFirewallQueryCountMetricName)
 	allMetricsSet.Add(kvRequestsMetricName)
 	allMetricsSet.Add(kvLatencyMetricName)
+	allMetricsSet.Add(workerSubrequestsCountMetricName)
+	allMetricsSet.Add(workerSubrequestTimeMetricName)
 	return allMetricsSet
 }
 
@@ -557,6 +571,12 @@ func mustRegisterMetrics(deniedMetrics MetricsSet) {
 	if !deniedMetrics.Has(kvLatencyMetricName) {
 		prometheus.MustRegister(kvLatency)
 	}
+	if !deniedMetrics.Has(workerSubrequestsCountMetricName) {
+		prometheus.MustRegister(workerSubrequestsCount)
+	}
+	if !deniedMetrics.Has(workerSubrequestTimeMetricName) {
+		prometheus.MustRegister(workerSubrequestTime)
+	}
 }
 
 func fetchLoadblancerPoolsHealth(account cfaccounts.Account, wg *sync.WaitGroup) {
@@ -659,6 +679,33 @@ func fetchKVAnalytics(account cfaccounts.Account, wg *sync.WaitGroup, deniedMetr
 	}
 }
 
+func fetchWorkerSubrequestAnalytics(account cfaccounts.Account, wg *sync.WaitGroup, deniedMetricsSet MetricsSet) {
+	wg.Add(1)
+	defer wg.Done()
+
+	r, err := fetchWorkerSubrequests(account.ID)
+	if err != nil {
+		log.Error("failed to fetch worker subrequests for account ", account.ID, ": ", err)
+		return
+	}
+
+	accountName := strings.ToLower(strings.ReplaceAll(account.Name, " ", "-"))
+
+	for _, a := range r.Viewer.Accounts {
+		for _, sr := range a.WorkersSubrequestsAdaptiveGroups {
+			if !deniedMetricsSet.Has(workerSubrequestsCountMetricName) {
+				workerSubrequestsCount.With(prometheus.Labels{"script_name": sr.Dimensions.ScriptName, "account": accountName}).Add(float64(sr.Sum.Subrequests))
+			}
+			if !deniedMetricsSet.Has(workerSubrequestTimeMetricName) {
+				workerSubrequestTime.With(prometheus.Labels{"script_name": sr.Dimensions.ScriptName, "account": accountName, "quantile": "P50"}).Set(float64(sr.Quantiles.TimeToResponseUsP50))
+				workerSubrequestTime.With(prometheus.Labels{"script_name": sr.Dimensions.ScriptName, "account": accountName, "quantile": "P75"}).Set(float64(sr.Quantiles.TimeToResponseUsP75))
+				workerSubrequestTime.With(prometheus.Labels{"script_name": sr.Dimensions.ScriptName, "account": accountName, "quantile": "P99"}).Set(float64(sr.Quantiles.TimeToResponseUsP99))
+				workerSubrequestTime.With(prometheus.Labels{"script_name": sr.Dimensions.ScriptName, "account": accountName, "quantile": "P999"}).Set(float64(sr.Quantiles.TimeToResponseUsP999))
+			}
+		}
+	}
+}
+
 func fetchLogpushAnalyticsForAccount(account cfaccounts.Account, wg *sync.WaitGroup) {
 	wg.Add(1)
 	defer wg.Done()