Skip to content

Commit e423b75

Browse files
rtalvarezrtalvarez
committed
SD-11577: augment worker metrics
1 parent 96a2ee9 commit e423b75

3 files changed

Lines changed: 429 additions & 0 deletions

File tree

‎cloudflare.go‎

Lines changed: 251 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import (
99
cfaccounts "github.com/cloudflare/cloudflare-go/v4/accounts"
1010
cfload_balancers "github.com/cloudflare/cloudflare-go/v4/load_balancers"
1111
cfpagination "github.com/cloudflare/cloudflare-go/v4/packages/pagination"
12+
cfqueues "github.com/cloudflare/cloudflare-go/v4/queues"
1213
cfrulesets "github.com/cloudflare/cloudflare-go/v4/rulesets"
1314
cfzero_trust "github.com/cloudflare/cloudflare-go/v4/zero_trust"
1415
cfzones "github.com/cloudflare/cloudflare-go/v4/zones"
@@ -122,10 +123,87 @@ type accountResp struct {
122123
DurationP75 float32 `json:"durationP75"`
123124
DurationP99 float32 `json:"durationP99"`
124125
DurationP999 float32 `json:"durationP999"`
126+
WallTimeP50 float32 `json:"wallTimeP50"`
127+
WallTimeP75 float32 `json:"wallTimeP75"`
128+
WallTimeP99 float32 `json:"wallTimeP99"`
129+
WallTimeP999 float32 `json:"wallTimeP999"`
125130
} `json:"quantiles"`
126131
} `json:"workersInvocationsAdaptive"`
127132
}
128133

134+
type cloudflareResponseKV struct {
135+
Viewer struct {
136+
Accounts []kvAccountResp `json:"accounts"`
137+
} `json:"viewer"`
138+
}
139+
140+
type kvAccountResp struct {
141+
KvOperationsAdaptiveGroups []struct {
142+
Dimensions struct {
143+
NamespaceID string `json:"namespaceId"`
144+
ActionType string `json:"actionType"`
145+
} `json:"dimensions"`
146+
Sum struct {
147+
Requests uint64 `json:"requests"`
148+
} `json:"sum"`
149+
Quantiles struct {
150+
LatencyMsP50 float32 `json:"latencyMsP50"`
151+
LatencyMsP75 float32 `json:"latencyMsP75"`
152+
LatencyMsP99 float32 `json:"latencyMsP99"`
153+
LatencyMsP999 float32 `json:"latencyMsP999"`
154+
} `json:"quantiles"`
155+
} `json:"kvOperationsAdaptiveGroups"`
156+
}
157+
158+
type cloudflareResponseSubrequests struct {
159+
Viewer struct {
160+
Accounts []subrequestsAccountResp `json:"accounts"`
161+
} `json:"viewer"`
162+
}
163+
164+
type subrequestsAccountResp struct {
165+
WorkersSubrequestsAdaptiveGroups []struct {
166+
Dimensions struct {
167+
ScriptName string `json:"scriptName"`
168+
} `json:"dimensions"`
169+
Sum struct {
170+
Subrequests uint64 `json:"subrequests"`
171+
} `json:"sum"`
172+
Quantiles struct {
173+
TimeToResponseUsP50 float32 `json:"timeToResponseUsP50"`
174+
TimeToResponseUsP75 float32 `json:"timeToResponseUsP75"`
175+
TimeToResponseUsP99 float32 `json:"timeToResponseUsP99"`
176+
TimeToResponseUsP999 float32 `json:"timeToResponseUsP999"`
177+
} `json:"quantiles"`
178+
} `json:"workersSubrequestsAdaptiveGroups"`
179+
}
180+
181+
type cloudflareResponseQueues struct {
182+
Viewer struct {
183+
Accounts []queueAccountResp `json:"accounts"`
184+
} `json:"viewer"`
185+
}
186+
187+
type queueAccountResp struct {
188+
QueueBacklogAdaptiveGroups []struct {
189+
Dimensions struct {
190+
QueueID string `json:"queueId"`
191+
} `json:"dimensions"`
192+
Avg struct {
193+
Messages float64 `json:"messages"`
194+
Bytes float64 `json:"bytes"`
195+
} `json:"avg"`
196+
} `json:"queueBacklogAdaptiveGroups"`
197+
QueueConsumerMetricsAdaptiveGroups []struct {
198+
Dimensions struct {
199+
QueueID string `json:"queueId"`
200+
} `json:"dimensions"`
201+
Avg struct {
202+
Concurrency float64 `json:"concurrency"`
203+
} `json:"avg"`
204+
} `json:"queueConsumerMetricsAdaptiveGroups"`
205+
}
206+
129207
type zoneRespColo struct {
130208
ColoGroups []struct {
131209
Dimensions struct {
@@ -737,6 +815,10 @@ func fetchWorkerTotals(accountID string) (*cloudflareResponseAccts, error) {
737815
durationP75
738816
durationP99
739817
durationP999
818+
wallTimeP50
819+
wallTimeP75
820+
wallTimeP99
821+
wallTimeP999
740822
}
741823
}
742824
}
@@ -1077,6 +1159,175 @@ func filterNonFreePlanZones(zones []cfzones.Zone) (filteredZones []cfzones.Zone)
10771159
return
10781160
}
10791161

1162+
func fetchKVOperations(accountID string) (*cloudflareResponseKV, error) {
1163+
request := graphql.NewRequest(`
1164+
query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
1165+
viewer {
1166+
accounts(filter: {accountTag: $accountID}) {
1167+
kvOperationsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
1168+
dimensions {
1169+
namespaceId
1170+
actionType
1171+
}
1172+
sum {
1173+
requests
1174+
}
1175+
quantiles {
1176+
latencyMsP50
1177+
latencyMsP75
1178+
latencyMsP99
1179+
latencyMsP999
1180+
}
1181+
}
1182+
}
1183+
}
1184+
}
1185+
`)
1186+
1187+
now, now1mAgo := GetTimeRange()
1188+
request.Var("limit", gqlQueryLimit)
1189+
request.Var("maxtime", now)
1190+
request.Var("mintime", now1mAgo)
1191+
request.Var("accountID", accountID)
1192+
1193+
gql.Mu.RLock()
1194+
defer gql.Mu.RUnlock()
1195+
1196+
ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
1197+
defer cancel()
1198+
1199+
var resp cloudflareResponseKV
1200+
if err := gql.Client.Run(ctx, request, &resp); err != nil {
1201+
log.Errorf("error fetching KV operations, err:%v", err)
1202+
return nil, err
1203+
}
1204+
1205+
return &resp, nil
1206+
}
1207+
1208+
func fetchWorkerSubrequests(accountID string) (*cloudflareResponseSubrequests, error) {
1209+
request := graphql.NewRequest(`
1210+
query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
1211+
viewer {
1212+
accounts(filter: {accountTag: $accountID}) {
1213+
workersSubrequestsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
1214+
dimensions {
1215+
scriptName
1216+
}
1217+
sum {
1218+
subrequests
1219+
}
1220+
quantiles {
1221+
timeToResponseUsP50
1222+
timeToResponseUsP75
1223+
timeToResponseUsP99
1224+
timeToResponseUsP999
1225+
}
1226+
}
1227+
}
1228+
}
1229+
}
1230+
`)
1231+
1232+
now, now1mAgo := GetTimeRange()
1233+
request.Var("limit", gqlQueryLimit)
1234+
request.Var("maxtime", now)
1235+
request.Var("mintime", now1mAgo)
1236+
request.Var("accountID", accountID)
1237+
1238+
gql.Mu.RLock()
1239+
defer gql.Mu.RUnlock()
1240+
1241+
ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
1242+
defer cancel()
1243+
1244+
var resp cloudflareResponseSubrequests
1245+
if err := gql.Client.Run(ctx, request, &resp); err != nil {
1246+
log.Errorf("error fetching worker subrequests, err:%v", err)
1247+
return nil, err
1248+
}
1249+
1250+
return &resp, nil
1251+
}
1252+
1253+
func fetchQueueNames(accountID string) (map[string]string, error) {
1254+
ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
1255+
defer cancel()
1256+
page := cfclient.Queues.ListAutoPaging(ctx,
1257+
cfqueues.QueueListParams{
1258+
AccountID: cf.F(accountID),
1259+
})
1260+
if page.Err() != nil {
1261+
return nil, page.Err()
1262+
}
1263+
1264+
queueNames := make(map[string]string)
1265+
seenIDs := make(map[string]struct{})
1266+
for page.Next() {
1267+
if page.Err() != nil {
1268+
log.Errorf("error during paging queues: %v", page.Err())
1269+
break
1270+
}
1271+
q := page.Current()
1272+
if _, exists := seenIDs[q.QueueID]; exists {
1273+
log.Errorf("fetchQueueNames: duplicate queue ID detected (%s), breaking loop", q.QueueID)
1274+
break
1275+
}
1276+
seenIDs[q.QueueID] = struct{}{}
1277+
queueNames[q.QueueID] = q.QueueName
1278+
}
1279+
1280+
return queueNames, nil
1281+
}
1282+
1283+
func fetchQueueMetrics(accountID string) (*cloudflareResponseQueues, error) {
1284+
request := graphql.NewRequest(`
1285+
query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) {
1286+
viewer {
1287+
accounts(filter: {accountTag: $accountID}) {
1288+
queueBacklogAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
1289+
dimensions {
1290+
queueId
1291+
}
1292+
avg {
1293+
messages
1294+
bytes
1295+
}
1296+
}
1297+
queueConsumerMetricsAdaptiveGroups(limit: $limit, filter: {datetime_geq: $mintime, datetime_lt: $maxtime}) {
1298+
dimensions {
1299+
queueId
1300+
}
1301+
avg {
1302+
concurrency
1303+
}
1304+
}
1305+
}
1306+
}
1307+
}
1308+
`)
1309+
1310+
now, now1mAgo := GetTimeRange()
1311+
request.Var("limit", gqlQueryLimit)
1312+
request.Var("maxtime", now)
1313+
request.Var("mintime", now1mAgo)
1314+
request.Var("accountID", accountID)
1315+
1316+
gql.Mu.RLock()
1317+
defer gql.Mu.RUnlock()
1318+
1319+
ctx, cancel := context.WithTimeout(context.Background(), cftimeout)
1320+
defer cancel()
1321+
1322+
var resp cloudflareResponseQueues
1323+
if err := gql.Client.Run(ctx, request, &resp); err != nil {
1324+
log.Errorf("error fetching queue metrics, err:%v", err)
1325+
return nil, err
1326+
}
1327+
1328+
return &resp, nil
1329+
}
1330+
10801331
func fetchDNSFirewallTotals(accountID string) (*cloudflareResponseDNSFirewall, error) {
10811332
request := graphql.NewRequest(`
10821333
query ($accountID: string, $mintime: Time!, $maxtime: Time!, $limit: Int!) {

‎main.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,9 @@ func fetchMetrics(deniedMetricsSet MetricsSet) {
115115
go fetchLoadblancerPoolsHealth(a, &wg)
116116
go fetchZeroTrustAnalyticsForAccount(a, &wg)
117117
go fetchDNSFirewallAnalytics(a, &wg, deniedMetricsSet)
118+
go fetchKVAnalytics(a, &wg)
119+
go fetchWorkerSubrequestAnalytics(a, &wg)
120+
go fetchQueueAnalytics(a, &wg)
118121
}
119122

120123
zones := fetchZones(accounts)

0 commit comments

Comments
 (0)