Skip to content

SD-11577: augment worker metrics #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rtalvarez wants to merge 3 commits into from
Closed

SD-11577: augment worker metrics #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e423b75
SD-11577: augment worker metrics
rtalvarez Mar 13, 2026
364711e
SD-11577: add debug logs
rtalvarez Mar 13, 2026
5b168a5
SD-11577: remove some functionality
rtalvarez Mar 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
221 changes: 221 additions & 0 deletions cloudflare.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"

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"
Expand Down Expand Up @@ -122,10 +123,87 @@ type accountResp struct {
DurationP75 float32 `json:"durationP75"`
DurationP99 float32 `json:"durationP99"`
DurationP999 float32 `json:"durationP999"`
WallTimeP50 float32 `json:"wallTimeP50"`
WallTimeP75 float32 `json:"wallTimeP75"`
WallTimeP99 float32 `json:"wallTimeP99"`
WallTimeP999 float32 `json:"wallTimeP999"`
} `json:"quantiles"`
} `json:"workersInvocationsAdaptive"`
}

type cloudflareResponseKV struct {
Viewer struct {
Accounts []kvAccountResp `json:"accounts"`
} `json:"viewer"`
}

type kvAccountResp struct {
KvOperationsAdaptiveGroups []struct {
Dimensions struct {
NamespaceID string `json:"namespaceId"`
ActionType string `json:"actionType"`
} `json:"dimensions"`
Sum struct {
Requests uint64 `json:"requests"`
} `json:"sum"`
Quantiles struct {
LatencyMsP50 float32 `json:"latencyMsP50"`
LatencyMsP75 float32 `json:"latencyMsP75"`
LatencyMsP99 float32 `json:"latencyMsP99"`
LatencyMsP999 float32 `json:"latencyMsP999"`
} `json:"quantiles"`
} `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 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"`
}

type zoneRespColo struct {
ColoGroups []struct {
Dimensions struct {
Expand Down Expand Up @@ -737,6 +815,10 @@ func fetchWorkerTotals(accountID string) (*cloudflareResponseAccts, error) {
durationP75
durationP99
durationP999
wallTimeP50
wallTimeP75
wallTimeP99
wallTimeP999
}
}
}
Expand Down Expand Up @@ -1077,6 +1159,145 @@ func filterNonFreePlanZones(zones []cfzones.Zone) (filteredZones []cfzones.Zone)
return
}

func fetchKVOperations(accountID string) (*cloudflareResponseKV, error) {
request := graphql.NewRequest(`
query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
viewer {
accounts(filter: {accountTag: $accountID}) {
kvOperationsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
dimensions {
namespaceId
actionType
}
sum {
requests
}
quantiles {
latencyMsP50
latencyMsP75
latencyMsP99
latencyMsP999
}
}
}
}
}
`)

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 cloudflareResponseKV
if err := gql.Client.Run(ctx, request, &resp); err != nil {
log.Errorf("error fetching KV operations, err:%v", err)
return nil, err
}

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

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!) {
Expand Down
6 changes: 5 additions & 1 deletion main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,9 @@ func fetchMetrics(deniedMetricsSet MetricsSet) {
go fetchLoadblancerPoolsHealth(a, &wg)
go fetchZeroTrustAnalyticsForAccount(a, &wg)
go fetchDNSFirewallAnalytics(a, &wg, deniedMetricsSet)
go fetchKVAnalytics(a, &wg)
go fetchWorkerSubrequestAnalytics(a, &wg)
go fetchQueueAnalytics(a, &wg)
}

zones := fetchZones(accounts)
Expand Down Expand Up @@ -151,6 +154,7 @@ func fetchMetrics(deniedMetricsSet MetricsSet) {
}

func runExporter() {
log.Info("Starting cloudflare_exporter with KV, subrequests, and queue metrics enabled")
cfgMetricsPath := viper.GetString("metrics_path")

// Handle pprof configuration
Expand Down Expand Up @@ -192,7 +196,7 @@ func runExporter() {
h := health.New(health.Health{})
http.HandleFunc("/health", h.Handler)

log.Info("Beginning to serve metrics on ", viper.GetString("listen"), cfgMetricsPath)
log.Info("(live change v1) Beginning to serve metrics on ", viper.GetString("listen"), cfgMetricsPath)

server := &http.Server{
Addr: viper.GetString("listen"),
Expand Down
Loading
Loading