diff --git a/cloudflare.go b/cloudflare.go index 68daced..46e61b3 100644 --- a/cloudflare.go +++ b/cloudflare.go @@ -2,13 +2,13 @@ package main import ( "context" + "errors" "strings" "time" cf "github.com/cloudflare/cloudflare-go/v4" 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" @@ -21,6 +21,12 @@ const ( apiPerPageLimit = 999 ) +// canceled reports whether err stems from context cancellation (e.g. graceful +// shutdown), which we expect and therefore do not log as an error. +func canceled(err error) bool { + return errors.Is(err, context.Canceled) +} + type cloudflareResponse struct { Viewer struct { Zones []zoneResp `json:"zones"` @@ -348,23 +354,28 @@ type lbResp struct { ZoneTag string `json:"zoneTag"` } -func fetchLoadblancerPools(account cfaccounts.Account) []cfload_balancers.Pool { - var cfPools []cfload_balancers.Pool - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) +func fetchLoadblancerPools(ctx context.Context, account cfaccounts.Account) []cfload_balancers.Pool { + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() + + var cfPools []cfload_balancers.Pool page := cfclient.LoadBalancers.Pools.ListAutoPaging(ctx, cfload_balancers.PoolListParams{ AccountID: cf.F(account.ID), }) if page.Err() != nil { - log.Errorf("error fetching loadbalancer pools, err:%v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error fetching loadbalancer pools, err:%v", page.Err()) + } return nil } seenIDs := make(map[string]struct{}) for page.Next() { if page.Err() != nil { - log.Errorf("error during paging pools: %v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error during paging pools: %v", page.Err()) + } break } pool := page.Current() @@ -379,10 +390,11 @@ func fetchLoadblancerPools(account cfaccounts.Account) []cfload_balancers.Pool { return cfPools } -func getAccountZoneList(accountID string) ([]cfzones.Zone, error) { - var zoneList []cfzones.Zone - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) +func getAccountZoneList(ctx context.Context, accountID string) ([]cfzones.Zone, error) { + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() + + var zoneList []cfzones.Zone page := cfclient.Zones.ListAutoPaging(ctx, cfzones.ZoneListParams{ Account: cf.F(cfzones.ZoneListParamsAccount{ID: cf.F(accountID)}), PerPage: cf.F(float64(apiPerPageLimit)), @@ -394,7 +406,9 @@ func getAccountZoneList(accountID string) ([]cfzones.Zone, error) { seenIDs := make(map[string]struct{}) for page.Next() { if page.Err() != nil { - log.Errorf("error during paging zoneList: %v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error during paging zoneList: %v", page.Err()) + } break } zone := page.Current() @@ -409,14 +423,16 @@ func getAccountZoneList(accountID string) ([]cfzones.Zone, error) { return zoneList, nil } -func fetchZones(accounts []cfaccounts.Account) []cfzones.Zone { +func fetchZones(ctx context.Context, accounts []cfaccounts.Account) []cfzones.Zone { var zones []cfzones.Zone for _, account := range accounts { - z, err := getAccountZoneList(account.ID) + z, err := getAccountZoneList(ctx, account.ID) if err != nil { - log.Errorf("error fetching zones: %v", err) + if !canceled(err) { + log.Errorf("error fetching zones: %v", err) + } continue } zones = append(zones, z...) @@ -424,13 +440,12 @@ func fetchZones(accounts []cfaccounts.Account) []cfzones.Zone { return zones } -func getRuleSetsList(params cfrulesets.RulesetListParams) ([]cfrulesets.RulesetListResponse, error) { - var ruleSetList []cfrulesets.RulesetListResponse - var page *cfpagination.CursorPagination[cfrulesets.RulesetListResponse] - var err error - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) +func getRuleSetsList(ctx context.Context, params cfrulesets.RulesetListParams) ([]cfrulesets.RulesetListResponse, error) { + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() - page, err = cfclient.Rulesets.List(ctx, params) + + var ruleSetList []cfrulesets.RulesetListResponse + page, err := cfclient.Rulesets.List(ctx, params) if err != nil { return nil, err } @@ -439,9 +454,7 @@ func getRuleSetsList(params cfrulesets.RulesetListParams) ([]cfrulesets.RulesetL for page.ResultInfo.Cursor != "" { params.Cursor = cf.F(page.ResultInfo.Cursor) - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) page, err = cfclient.Rulesets.List(ctx, params) - cancel() if err != nil { return nil, err } @@ -451,12 +464,17 @@ func getRuleSetsList(params cfrulesets.RulesetListParams) ([]cfrulesets.RulesetL return ruleSetList, nil } -func fetchFirewallRules(zoneID string) map[string]string { - listOfRulesets, err := getRuleSetsList(cfrulesets.RulesetListParams{ +func fetchFirewallRules(ctx context.Context, zoneID string) map[string]string { + ctx, cancel := context.WithTimeout(ctx, cftimeout) + defer cancel() + + listOfRulesets, err := getRuleSetsList(ctx, cfrulesets.RulesetListParams{ ZoneID: cf.F(zoneID), }) if err != nil { - log.Errorf("error fetching firewall rules, ZoneID:%s, Err:%v", zoneID, err) + if !canceled(err) { + log.Errorf("error fetching firewall rules, ZoneID:%s, Err:%v", zoneID, err) + } return nil } @@ -464,32 +482,30 @@ func fetchFirewallRules(zoneID string) map[string]string { for _, rulesetDesc := range listOfRulesets { if rulesetDesc.Phase == cfrulesets.PhaseHTTPRequestFirewallManaged { - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) ruleset, err := cfclient.Rulesets.Get(ctx, rulesetDesc.ID, cfrulesets.RulesetGetParams{ ZoneID: cf.F(zoneID), }) if err != nil { - log.Errorf("error fetching ruleset for managed firewall rules, ZoneID:%s, RulesetID:%s, Err:%v", zoneID, rulesetDesc.ID, err) - cancel() + if !canceled(err) { + log.Errorf("error fetching ruleset for managed firewall rules, ZoneID:%s, RulesetID:%s, Err:%v", zoneID, rulesetDesc.ID, err) + } continue } - cancel() for _, rule := range ruleset.Rules { firewallRulesMap[rule.ID] = rule.Description } } if rulesetDesc.Phase == cfrulesets.PhaseHTTPRequestFirewallCustom { - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) ruleset, err := cfclient.Rulesets.Get(ctx, rulesetDesc.ID, cfrulesets.RulesetGetParams{ ZoneID: cf.F(zoneID), }) if err != nil { - log.Errorf("error fetching ruleset for custom firewall rules, ZoneID:%s, RulesetID:%s, Err:%v", zoneID, rulesetDesc.ID, err) - cancel() + if !canceled(err) { + log.Errorf("error fetching ruleset for custom firewall rules, ZoneID:%s, RulesetID:%s, Err:%v", zoneID, rulesetDesc.ID, err) + } continue } - cancel() for _, rule := range ruleset.Rules { firewallRulesMap[rule.ID] = rule.Description } @@ -499,7 +515,10 @@ func fetchFirewallRules(zoneID string) map[string]string { return firewallRulesMap } -func fetchAccounts(targetAccountIDs []string) []cfaccounts.Account { +func fetchAccounts(ctx context.Context, targetAccountIDs []string) []cfaccounts.Account { + ctx, cancel := context.WithTimeout(ctx, cftimeout) + defer cancel() + var cfAccounts []cfaccounts.Account if len(targetAccountIDs) > 0 { @@ -510,14 +529,14 @@ func fetchAccounts(targetAccountIDs []string) []cfaccounts.Account { continue } - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) account, err := cfclient.Accounts.Get(ctx, cfaccounts.AccountGetParams{ AccountID: cf.F(accountID), }) - cancel() if err != nil { - log.Warnf("Account %s details unavailable, using ID only: %v", accountID, err) + if !canceled(err) { + log.Warnf("Account %s details unavailable, using ID only: %v", accountID, err) + } cfAccounts = append(cfAccounts, cfaccounts.Account{ ID: accountID, }) @@ -530,21 +549,23 @@ func fetchAccounts(targetAccountIDs []string) []cfaccounts.Account { } log.Info("Listing all accessible accounts (user-level token mode)") - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) - defer cancel() page := cfclient.Accounts.ListAutoPaging(ctx, cfaccounts.AccountListParams{ PerPage: cf.F(float64(apiPerPageLimit)), }) if page.Err() != nil { - log.Errorf("error fetching accounts:%v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error fetching accounts:%v", page.Err()) + } return nil } seenIDs := make(map[string]struct{}) for page.Next() { if page.Err() != nil { - log.Errorf("error during paging accounts: %v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error during paging accounts: %v", page.Err()) + } break } account := page.Current() @@ -558,7 +579,7 @@ func fetchAccounts(targetAccountIDs []string) []cfaccounts.Account { return cfAccounts } -func fetchZoneTotals(zoneIDs []string) (*cloudflareResponse, error) { +func fetchZoneTotals(ctx context.Context, zoneIDs []string) (*cloudflareResponse, error) { request := graphql.NewRequest(` query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -671,19 +692,21 @@ query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponse if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("failed to fetch zone totals, err:%v", err) + if !canceled(err) { + log.Errorf("failed to fetch zone totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchColoTotals(zoneIDs []string) (*cloudflareResponseColo, error) { +func fetchColoTotals(ctx context.Context, zoneIDs []string) (*cloudflareResponseColo, error) { request := graphql.NewRequest(` query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -721,19 +744,21 @@ func fetchColoTotals(zoneIDs []string) (*cloudflareResponseColo, error) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseColo if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("failed to fetch colocation totals, err:%v", err) + if !canceled(err) { + log.Errorf("failed to fetch colocation totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchWorkerTotals(accountID string) (*cloudflareResponseAccts, error) { +func fetchWorkerTotals(ctx context.Context, accountID string) (*cloudflareResponseAccts, error) { request := graphql.NewRequest(` query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -775,19 +800,21 @@ func fetchWorkerTotals(accountID string) (*cloudflareResponseAccts, error) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseAccts if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("error fetching worker totals, err:%v", err) + if !canceled(err) { + log.Errorf("error fetching worker totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchAccountHTTPDataTransfer(accountID string, from time.Time, to time.Time, requestSource string) (*cloudflareResponseAccountHTTPDataTransfer, error) { +func fetchAccountHTTPDataTransfer(ctx context.Context, accountID string, from time.Time, to time.Time, requestSource string) (*cloudflareResponseAccountHTTPDataTransfer, error) { request := graphql.NewRequest(` query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $requestSource: String!) { viewer { @@ -817,19 +844,21 @@ query ($accountID: String!, $mintime: Time!, $maxtime: Time!, $requestSource: St gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseAccountHTTPDataTransfer if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("failed to fetch account HTTP data transfer, err:%v", err) + if !canceled(err) { + log.Errorf("failed to fetch account HTTP data transfer, err:%v", err) + } return nil, err } return &resp, nil } -func fetchLoadBalancerTotals(zoneIDs []string) (*cloudflareResponseLb, error) { +func fetchLoadBalancerTotals(ctx context.Context, zoneIDs []string) (*cloudflareResponseLb, error) { request := graphql.NewRequest(` query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -889,18 +918,20 @@ func fetchLoadBalancerTotals(zoneIDs []string) (*cloudflareResponseLb, error) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseLb if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("error fetching load balancer totals, err:%v", err) + if !canceled(err) { + log.Errorf("error fetching load balancer totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchLogpushAccount(accountID string) (*cloudflareResponseLogpushAccount, error) { +func fetchLogpushAccount(ctx context.Context, accountID string) (*cloudflareResponseLogpushAccount, error) { request := graphql.NewRequest(`query($accountID: String!, $limit: Int!, $mintime: Time!, $maxtime: Time!) { viewer { accounts(filter: {accountTag : $accountID }) { @@ -934,18 +965,21 @@ func fetchLogpushAccount(accountID string) (*cloudflareResponseLogpushAccount, e gql.Mu.RLock() defer gql.Mu.RUnlock() - var resp cloudflareResponseLogpushAccount - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() + var resp cloudflareResponseLogpushAccount + if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("error fetching logpush account totals, err:%v", err) + if !canceled(err) { + log.Errorf("error fetching logpush account totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchLogpushZone(zoneIDs []string) (*cloudflareResponseLogpushZone, error) { +func fetchLogpushZone(ctx context.Context, zoneIDs []string) (*cloudflareResponseLogpushZone, error) { request := graphql.NewRequest(`query($zoneIDs: String!, $limit: Int!, $mintime: Time!, $maxtime: Time!) { viewer { zones(filter: {zoneTag_in : $zoneIDs }) { @@ -979,19 +1013,21 @@ func fetchLogpushZone(zoneIDs []string) (*cloudflareResponseLogpushZone, error) gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseLogpushZone if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("error fetching logpush zone totals, err:%v", err) + if !canceled(err) { + log.Errorf("error fetching logpush zone totals, err:%v", err) + } return nil, err } return &resp, nil } -func fetchEdgeErrorsByPath(zoneIDs []string) (*cloudflareResponseEdgeErrorsByPath, error) { +func fetchEdgeErrorsByPath(ctx context.Context, zoneIDs []string) (*cloudflareResponseEdgeErrorsByPath, error) { request := graphql.NewRequest(` query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -1026,19 +1062,21 @@ func fetchEdgeErrorsByPath(zoneIDs []string) (*cloudflareResponseEdgeErrorsByPat gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseEdgeErrorsByPath if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("failed to fetch edge errors by path, err:%v", err) + if !canceled(err) { + log.Errorf("failed to fetch edge errors by path, err:%v", err) + } return nil, err } return &resp, nil } -func fetchR2Account(accountID string) (*cloudflareResponseR2Account, error) { +func fetchR2Account(ctx context.Context, accountID string) (*cloudflareResponseR2Account, error) { request := graphql.NewRequest(`query($accountID: String!, $limit: Int!, $date: String!) { viewer { accounts(filter: {accountTag : $accountID }) { @@ -1078,21 +1116,24 @@ func fetchR2Account(accountID string) (*cloudflareResponseR2Account, error) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseR2Account if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("error fetching R2 account: %v", err) + if !canceled(err) { + log.Errorf("error fetching R2 account: %v", err) + } return nil, err } return &resp, nil } -func fetchCloudflareTunnels(account cfaccounts.Account) []cfzero_trust.TunnelListResponse { - var cfTunnels []cfzero_trust.TunnelListResponse - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) +func fetchCloudflareTunnels(ctx context.Context, account cfaccounts.Account) []cfzero_trust.TunnelListResponse { + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() + + var cfTunnels []cfzero_trust.TunnelListResponse page := cfclient.ZeroTrust.Tunnels.ListAutoPaging(ctx, cfzero_trust.TunnelListParams{ AccountID: cf.F(account.ID), @@ -1100,14 +1141,18 @@ func fetchCloudflareTunnels(account cfaccounts.Account) []cfzero_trust.TunnelLis IsDeleted: cf.F(false), }) if page.Err() != nil { - log.Errorf("error fetching tunnels, err:%v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error fetching tunnels, err:%v", page.Err()) + } return nil } seenIDs := make(map[string]struct{}) for page.Next() { if page.Err() != nil { - log.Errorf("error during paging tunnels: %v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error during paging tunnels: %v", page.Err()) + } break } tunnel := page.Current() @@ -1122,23 +1167,28 @@ func fetchCloudflareTunnels(account cfaccounts.Account) []cfzero_trust.TunnelLis return cfTunnels } -func fetchCloudflareTunnelConnectors(account cfaccounts.Account, tunnelID string) []cfzero_trust.Client { - var cfClients []cfzero_trust.Client - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) +func fetchCloudflareTunnelConnectors(ctx context.Context, account cfaccounts.Account, tunnelID string) []cfzero_trust.Client { + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() + + var cfClients []cfzero_trust.Client page := cfclient.ZeroTrust.Tunnels.Connections.GetAutoPaging(ctx, tunnelID, cfzero_trust.TunnelConnectionGetParams{ AccountID: cf.F(account.ID), }) if page.Err() != nil { - log.Errorf("error fetching tunnel connections, err:%v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error fetching tunnel connections, err:%v", page.Err()) + } return nil } for page.Next() { if page.Err() != nil { - log.Errorf("error during paging tunnel connections: %v", page.Err()) + if !canceled(page.Err()) { + log.Errorf("error during paging tunnel connections: %v", page.Err()) + } break } client := page.Current() @@ -1188,7 +1238,7 @@ func filterNonFreePlanZones(zones []cfzones.Zone) (filteredZones []cfzones.Zone) return } -func fetchASNTotals(zoneIDs []string) (*cloudflareResponseASN, error) { +func fetchASNTotals(ctx context.Context, zoneIDs []string) (*cloudflareResponseASN, error) { request := graphql.NewRequest(` query ($zoneIDs: [String!], $mintime: Time!, $maxtime: Time!, $limit: Int!) { viewer { @@ -1227,12 +1277,14 @@ func fetchASNTotals(zoneIDs []string) (*cloudflareResponseASN, error) { gql.Mu.RLock() defer gql.Mu.RUnlock() - ctx, cancel := context.WithTimeout(context.Background(), cftimeout) + ctx, cancel := context.WithTimeout(ctx, cftimeout) defer cancel() var resp cloudflareResponseASN if err := gql.Client.Run(ctx, request, &resp); err != nil { - log.Errorf("failed to fetch ASN totals, err:%v", err) + if !canceled(err) { + log.Errorf("failed to fetch ASN totals, err:%v", err) + } return nil, err } diff --git a/main.go b/main.go index 49d1e05..ce2519d 100644 --- a/main.go +++ b/main.go @@ -1,11 +1,15 @@ package main import ( + "context" + "errors" "net/http" _ "net/http/pprof" // #nosec G108 - pprof is controlled via enable_pprof flag + "os/signal" "runtime" "strings" "sync" + "syscall" "time" "github.com/nelkinda/health-go" @@ -112,32 +116,21 @@ func filterExcludedZones(all []cfzones.Zone, exclude []string) []cfzones.Zone { return filtered } -func fetchMetrics() { +func fetchMetrics(ctx context.Context) { var wg sync.WaitGroup targetAccounts := getTargetAccounts() - accounts := fetchAccounts(targetAccounts) + accounts := fetchAccounts(ctx, targetAccounts) for _, a := range accounts { - wg.Add(1) - go fetchWorkerAnalytics(a, &wg) - - wg.Add(1) - go fetchLogpushAnalyticsForAccount(a, &wg) - - wg.Add(1) - go fetchR2StorageForAccount(a, &wg) - - wg.Add(1) - go fetchLoadblancerPoolsHealth(a, &wg) - - wg.Add(1) - go fetchZeroTrustAnalyticsForAccount(a, &wg) - - wg.Add(1) - go fetchAccountHTTPDataTransferAnalytics(a, &wg) + wg.Go(func() { fetchWorkerAnalytics(ctx, a) }) + wg.Go(func() { fetchLogpushAnalyticsForAccount(ctx, a) }) + wg.Go(func() { fetchR2StorageForAccount(ctx, a) }) + wg.Go(func() { fetchLoadblancerPoolsHealth(ctx, a) }) + wg.Go(func() { fetchZeroTrustAnalyticsForAccount(ctx, a) }) + wg.Go(func() { fetchAccountHTTPDataTransferAnalytics(ctx, a) }) } - zones := fetchZones(accounts) + zones := fetchZones(ctx, accounts) tzones := getTargetZones() fzones := filterZones(zones, tzones) ezones := getExcludedZones() @@ -148,53 +141,37 @@ func fetchMetrics() { zoneCount := len(filteredZones) if zoneCount > 0 && zoneCount <= cfgraphqlreqlimit { - wg.Add(1) - go fetchZoneAnalytics(filteredZones, &wg) - - wg.Add(1) - go fetchZoneColocationAnalytics(filteredZones, &wg) - - wg.Add(1) - go fetchLoadBalancerAnalytics(filteredZones, &wg) - - wg.Add(1) - go fetchLogpushAnalyticsForZone(filteredZones, &wg) - - wg.Add(1) - go fetchZoneASNAnalytics(filteredZones, &wg) - - wg.Add(1) - go fetchEdgeErrorsByPathAnalytics(filteredZones, &wg) + fetchZoneMetrics(ctx, &wg, filteredZones) } else if zoneCount > cfgraphqlreqlimit { for s := 0; s < zoneCount; s += cfgraphqlreqlimit { e := s + cfgraphqlreqlimit if e > zoneCount { e = zoneCount } - wg.Add(1) - go fetchZoneAnalytics(filteredZones[s:e], &wg) - - wg.Add(1) - go fetchZoneColocationAnalytics(filteredZones[s:e], &wg) - - wg.Add(1) - go fetchLoadBalancerAnalytics(filteredZones[s:e], &wg) - - wg.Add(1) - go fetchLogpushAnalyticsForZone(filteredZones[s:e], &wg) - - wg.Add(1) - go fetchZoneASNAnalytics(filteredZones[s:e], &wg) - - wg.Add(1) - go fetchEdgeErrorsByPathAnalytics(filteredZones[s:e], &wg) + fetchZoneMetrics(ctx, &wg, filteredZones[s:e]) } } wg.Wait() } -func runExporter() { +// fetchZoneMetrics fans out every zone-scoped collector for a single batch of +// zones, each on its own goroutine tracked by wg. +func fetchZoneMetrics(ctx context.Context, wg *sync.WaitGroup, zones []cfzones.Zone) { + wg.Go(func() { fetchZoneAnalytics(ctx, zones) }) + wg.Go(func() { fetchZoneColocationAnalytics(ctx, zones) }) + wg.Go(func() { fetchLoadBalancerAnalytics(ctx, zones) }) + wg.Go(func() { fetchLogpushAnalyticsForZone(ctx, zones) }) + wg.Go(func() { fetchZoneASNAnalytics(ctx, zones) }) + wg.Go(func() { fetchEdgeErrorsByPathAnalytics(ctx, zones) }) +} + +func runExporter() error { + // The exporter owns the whole process lifecycle: derive a context that is + // cancelled on SIGINT/SIGTERM so we can drain connections on shutdown. + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + cfgMetricsPath := viper.GetString("metrics_path") // Handle pprof configuration @@ -221,8 +198,18 @@ func runExporter() { log.Info("Scrape interval set to ", scrapeInterval) go func() { - for ; true; <-time.NewTicker(scrapeInterval).C { - go fetchMetrics() + ticker := time.NewTicker(scrapeInterval) + defer ticker.Stop() + for { + // Run synchronously: the ticker coalesces ticks while a scrape is + // in flight, so a slow scrape can't pile up overlapping runs, and + // ctx cancellation unwinds the in-flight scrape before we return. + fetchMetrics(ctx) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } } }() @@ -243,16 +230,44 @@ func runExporter() { ReadHeaderTimeout: 3 * time.Second, } - log.Fatal(server.ListenAndServe()) + // Serve in the background so we can block on either a server error or a + // shutdown signal. http.ErrServerClosed is the expected, clean return + // value once Shutdown has been called. + serverErr := make(chan error, 1) + go func() { + if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + serverErr <- err + } + }() + + select { + case err := <-serverErr: + return err + case <-ctx.Done(): + log.Info("Shutdown signal received, draining connections...") + // Fresh context: ctx is already cancelled. Bound the drain so a stuck + // connection cannot hang shutdown forever. + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := server.Shutdown(shutdownCtx); err != nil { + return err + } + log.Info("Shutdown complete") + return nil + } } func main() { cmd := &cobra.Command{ Use: "cloudflare_exporter", Short: "Prometheus exporter exposing Cloudflare Analytics dashboard data on a per-zone basis, as well as Worker metrics", - Run: func(_ *cobra.Command, _ []string) { - runExporter() + RunE: func(_ *cobra.Command, _ []string) error { + return runExporter() }, + // runExporter errors are reported via log.Fatal below; don't let cobra + // also print the error or the usage text on a runtime failure. + SilenceErrors: true, + SilenceUsage: true, } viper.AutomaticEnv() @@ -383,5 +398,7 @@ func main() { log.Fatal("Please provide CF_API_KEY+CF_API_EMAIL or CF_API_TOKEN") } - cmd.Execute() + if err := cmd.Execute(); err != nil { + log.Fatal(err) + } } diff --git a/prometheus.go b/prometheus.go index c06b3dc..c90456f 100644 --- a/prometheus.go +++ b/prometheus.go @@ -1,10 +1,10 @@ package main import ( + "context" "fmt" "strconv" "strings" - "sync" "github.com/biter777/countries" cfaccounts "github.com/cloudflare/cloudflare-go/v4/accounts" @@ -598,10 +598,8 @@ func mustRegisterMetrics(deniedMetrics MetricsSet) { } } -func fetchLoadblancerPoolsHealth(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - - pools := fetchLoadblancerPools(account) +func fetchLoadblancerPoolsHealth(ctx context.Context, account cfaccounts.Account) { + pools := fetchLoadblancerPools(ctx, account) if pools == nil { return } @@ -632,12 +630,12 @@ func fetchLoadblancerPoolsHealth(account cfaccounts.Account, wg *sync.WaitGroup) } } -func fetchWorkerAnalytics(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - - r, err := fetchWorkerTotals(account.ID) +func fetchWorkerAnalytics(ctx context.Context, account cfaccounts.Account) { + r, err := fetchWorkerTotals(ctx, account.ID) if err != nil { - log.Error("failed to fetch worker analytics for account ", account.ID, ": ", err) + if !canceled(err) { + log.Error("failed to fetch worker analytics for account ", account.ID, ": ", err) + } return } @@ -660,17 +658,17 @@ func fetchWorkerAnalytics(account cfaccounts.Account, wg *sync.WaitGroup) { } } -func fetchLogpushAnalyticsForAccount(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchLogpushAnalyticsForAccount(ctx context.Context, account cfaccounts.Account) { if viper.GetBool("free_tier") { return } - r, err := fetchLogpushAccount(account.ID) + r, err := fetchLogpushAccount(ctx, account.ID) if err != nil { - log.Error("failed to fetch logpush analytics for account ", account.ID, ": ", err) + if !canceled(err) { + log.Error("failed to fetch logpush analytics for account ", account.ID, ": ", err) + } return } @@ -684,10 +682,8 @@ func fetchLogpushAnalyticsForAccount(account cfaccounts.Account, wg *sync.WaitGr } } -func fetchR2StorageForAccount(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - - r, err := fetchR2Account(account.ID) +func fetchR2StorageForAccount(ctx context.Context, account cfaccounts.Account) { + r, err := fetchR2Account(ctx, account.ID) if err != nil { return @@ -705,9 +701,7 @@ func fetchR2StorageForAccount(account cfaccounts.Account, wg *sync.WaitGroup) { } } -func fetchLogpushAnalyticsForZone(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchLogpushAnalyticsForZone(ctx context.Context, zones []cfzones.Zone) { if viper.GetBool("free_tier") { return } @@ -717,10 +711,12 @@ func fetchLogpushAnalyticsForZone(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - r, err := fetchLogpushZone(zoneIDs) + r, err := fetchLogpushZone(ctx, zoneIDs) if err != nil { - log.Error("failed to fetch logpush analytics for zones: ", err) + if !canceled(err) { + log.Error("failed to fetch logpush analytics for zones: ", err) + } return } @@ -733,9 +729,7 @@ func fetchLogpushAnalyticsForZone(zones []cfzones.Zone, wg *sync.WaitGroup) { } } -func fetchZoneColocationAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchZoneColocationAnalytics(ctx context.Context, zones []cfzones.Zone) { // Colocation metrics are not available in non-enterprise zones if viper.GetBool("free_tier") { return @@ -746,9 +740,11 @@ func fetchZoneColocationAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - r, err := fetchColoTotals(zoneIDs) + r, err := fetchColoTotals(ctx, zoneIDs) if err != nil { - log.Error("failed to fetch colocation analytics for zones: ", err) + if !canceled(err) { + log.Error("failed to fetch colocation analytics for zones: ", err) + } return } for _, z := range r.Viewer.Zones { @@ -762,9 +758,7 @@ func fetchZoneColocationAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { } } -func fetchZoneAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchZoneAnalytics(ctx context.Context, zones []cfzones.Zone) { // None of the below referenced metrics are available in the free tier if viper.GetBool("free_tier") { return @@ -775,9 +769,11 @@ func fetchZoneAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - r, err := fetchZoneTotals(zoneIDs) + r, err := fetchZoneTotals(ctx, zoneIDs) if err != nil { - log.Error("failed to fetch zone analytics: ", err) + if !canceled(err) { + log.Error("failed to fetch zone analytics: ", err) + } return } @@ -786,7 +782,7 @@ func fetchZoneAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { z := z addHTTPGroups(&z, name, account) - addFirewallGroups(&z, name, account) + addFirewallGroups(ctx, &z, name, account) addHealthCheckGroups(&z, name, account) addHTTPAdaptiveGroups(&z, name, account) } @@ -861,7 +857,7 @@ func addHTTPGroups(z *zoneResp, name string, account string) { zoneUniquesTotal.With(prometheus.Labels{"zone": name, "account": account}).Add(float64(zt.Unique.Uniques)) } -func addFirewallGroups(z *zoneResp, name string, account string) { +func addFirewallGroups(ctx context.Context, z *zoneResp, name string, account string) { // Nothing to do. if len(z.FirewallEventsAdaptiveGroups) == 0 { return @@ -871,7 +867,7 @@ func addFirewallGroups(z *zoneResp, name string, account string) { label := prometheus.Labels{"zone": name, "account": account} zoneFirewallEventsCount.DeletePartialMatch(label) - rulesMap := fetchFirewallRules(z.ZoneTag) + rulesMap := fetchFirewallRules(ctx, z.ZoneTag) for _, g := range z.FirewallEventsAdaptiveGroups { zoneFirewallEventsCount.With( prometheus.Labels{ @@ -976,9 +972,7 @@ func addHTTPAdaptiveGroups(z *zoneResp, name string, account string) { } } -func fetchEdgeErrorsByPathAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchEdgeErrorsByPathAnalytics(ctx context.Context, zones []cfzones.Zone) { if !viper.GetBool("enable_edge_errors_by_path") { return } @@ -992,9 +986,11 @@ func fetchEdgeErrorsByPathAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - r, err := fetchEdgeErrorsByPath(zoneIDs) + r, err := fetchEdgeErrorsByPath(ctx, zoneIDs) if err != nil { - log.Error("failed to fetch edge errors by path: ", err) + if !canceled(err) { + log.Error("failed to fetch edge errors by path: ", err) + } return } @@ -1024,9 +1020,7 @@ func addEdgeErrorsByPath(z *zoneRespEdgeErrorsByPath, name string, account strin } } -func fetchLoadBalancerAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchLoadBalancerAnalytics(ctx context.Context, zones []cfzones.Zone) { // None of the below referenced metrics are available in the free tier if viper.GetBool("free_tier") { return @@ -1037,9 +1031,11 @@ func fetchLoadBalancerAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - l, err := fetchLoadBalancerTotals(zoneIDs) + l, err := fetchLoadBalancerTotals(ctx, zoneIDs) if err != nil { - log.Error("failed to fetch load balancer analytics: ", err) + if !canceled(err) { + log.Error("failed to fetch load balancer analytics: ", err) + } return } for _, lb := range l.Viewer.Zones { @@ -1085,15 +1081,11 @@ func addLoadBalancingRequestsAdaptive(z *lbResp, name string, account string) { } } -func fetchZeroTrustAnalyticsForAccount(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - - addCloudflareTunnelStatus(account) +func fetchZeroTrustAnalyticsForAccount(ctx context.Context, account cfaccounts.Account) { + addCloudflareTunnelStatus(ctx, account) } -func fetchAccountHTTPDataTransferAnalytics(account cfaccounts.Account, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchAccountHTTPDataTransferAnalytics(ctx context.Context, account cfaccounts.Account) { if !viper.GetBool("enable_account_usage_metrics") { return } @@ -1101,9 +1093,11 @@ func fetchAccountHTTPDataTransferAnalytics(account cfaccounts.Account, wg *sync. now, monthStart, nextMonthStart := GetMonthRange() requestSource := viper.GetString("account_usage_request_source") - r, err := fetchAccountHTTPDataTransfer(account.ID, monthStart, now, requestSource) + r, err := fetchAccountHTTPDataTransfer(ctx, account.ID, monthStart, now, requestSource) if err != nil { - log.Error("failed to fetch account HTTP data transfer analytics for account ", account.ID, ": ", err) + if !canceled(err) { + log.Error("failed to fetch account HTTP data transfer analytics for account ", account.ID, ": ", err) + } return } @@ -1145,8 +1139,8 @@ func addAccountHTTPDataTransfer(r *cloudflareResponseAccountHTTPDataTransfer, ac accountHTTPDataTransferMonthTotal.With(labels).Set(totalSeconds) } -func addCloudflareTunnelStatus(account cfaccounts.Account) { - tunnels := fetchCloudflareTunnels(account) +func addCloudflareTunnelStatus(ctx context.Context, account cfaccounts.Account) { + tunnels := fetchCloudflareTunnels(ctx, account) for _, t := range tunnels { tunnelInfo.With( prometheus.Labels{ @@ -1165,7 +1159,7 @@ func addCloudflareTunnelStatus(account cfaccounts.Account) { // Each client/connector can open many connections to the Cloudflare edge, // we opt to not expose metrics for each individual connection. We do expose // an informational metric for each client/connector however. - clients := fetchCloudflareTunnelConnectors(account, t.ID) + clients := fetchCloudflareTunnelConnectors(ctx, account, t.ID) for _, c := range clients { originIP := "" if len(c.Conns) > 0 { @@ -1214,9 +1208,7 @@ func getCloudflareTunnelStatusValue(status string) uint8 { } } -func fetchZoneASNAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { - defer wg.Done() - +func fetchZoneASNAnalytics(ctx context.Context, zones []cfzones.Zone) { // ASN metrics are not available in free tier if viper.GetBool("free_tier") { return @@ -1227,7 +1219,7 @@ func fetchZoneASNAnalytics(zones []cfzones.Zone, wg *sync.WaitGroup) { return } - r, err := fetchASNTotals(zoneIDs) + r, err := fetchASNTotals(ctx, zoneIDs) if err != nil { return }