Skip to content

Commit 316fd99

Browse files
davidbirdsongdavidbirdsong
committed
fix: expire boundary double-count and label sort order in counter chains
Two bugs that cause counter value spikes in pushed metrics: 1. When Lookback == WindowTTL, buckets at the exact boundary were expired then re-fetched, double-counting the gauge in the chain base. Changed expire check from >= to >. 2. Key.String() preserved label insertion order, so the same series could map to different counter chains when labels arrived in different order (e.g. seeding from VM). Labels are now sorted by name before serialization. Also: collapse WindowTTL into Lookback (they served the same purpose), add Prometheus counters for convergence tuning, add fixture capture tool and second spike incident test data.
1 parent 43232f6 commit 316fd99

18 files changed

Lines changed: 1366 additions & 97 deletions

‎cmd/capture-fixtures/main.go‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
// Command capture-fixtures fetches Cloudflare observations for a time range
2+
// and writes them as trimmed JSON fixtures for use in spike_test.go.
3+
//
4+
// Usage:
5+
//
6+
// go run ./cmd/capture-fixtures \
7+
// -zone 43c0ec06d9a970d6707dee37374c1b13 \
8+
// -start 2026-06-26T14:00:00Z \
9+
// -end 2026-06-26T14:22:00Z \
10+
// -lookback 25m \
11+
// -prefix spike_0626 \
12+
// -metric cloudflare_zone_requests_total \
13+
// -outdir converge/testdata
14+
package main
15+
16+
import (
17+
"bytes"
18+
"context"
19+
"encoding/json"
20+
"flag"
21+
"fmt"
22+
"log"
23+
"net/http"
24+
"os"
25+
"time"
26+
27+
cfzones "github.com/cloudflare/cloudflare-go/v4/zones"
28+
"github.com/lablabs/cloudflare-exporter/cfetch"
29+
"github.com/lablabs/cloudflare-exporter/cfgql"
30+
"github.com/lablabs/cloudflare-exporter/converge"
31+
)
32+
33+
type gqlClient struct{ token string }
34+
35+
type gqlResponse struct {
36+
Data any `json:"data"`
37+
Errors []gqlError `json:"errors"`
38+
}
39+
type gqlError struct {
40+
Message string `json:"message"`
41+
}
42+
43+
func (c *gqlClient) RunGQL(ctx context.Context, req *cfgql.GQLRequest, dest any) error {
44+
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
45+
defer cancel()
46+
47+
body, _ := json.Marshal(req)
48+
httpReq, _ := http.NewRequestWithContext(ctx, http.MethodPost,
49+
"https://api.cloudflare.com/client/v4/graphql/",
50+
bytes.NewReader(body))
51+
httpReq.Header.Set("Accept", "application/json; charset=utf-8")
52+
httpReq.Header.Set("Authorization", "Bearer "+c.token)
53+
54+
resp, err := http.DefaultClient.Do(httpReq)
55+
if err != nil {
56+
return err
57+
}
58+
defer resp.Body.Close()
59+
60+
var gResp gqlResponse
61+
gResp.Data = dest
62+
if err := json.NewDecoder(resp.Body).Decode(&gResp); err != nil {
63+
return err
64+
}
65+
for _, e := range gResp.Errors {
66+
return fmt.Errorf("graphql: %s", e.Message)
67+
}
68+
return nil
69+
}
70+
71+
type recordedFetch struct {
72+
Seq int `json:"seq"`
73+
Start time.Time `json:"start"`
74+
End time.Time `json:"end"`
75+
Observations []recordedObservation `json:"observations"`
76+
}
77+
78+
type recordedObservation struct {
79+
Name string `json:"name"`
80+
Labels map[string]string `json:"labels"`
81+
Value uint64 `json:"value"`
82+
Bucket time.Time `json:"bucket"`
83+
}
84+
85+
func main() {
86+
var (
87+
zoneID = flag.String("zone", "", "Cloudflare zone ID")
88+
startS = flag.String("start", "", "start time (RFC3339)")
89+
endS = flag.String("end", "", "end time (RFC3339)")
90+
lookb = flag.Duration("lookback", 25*time.Minute, "lookback duration for the live fetch")
91+
chunk = flag.Duration("chunk", 10*time.Minute, "backfill chunk size")
92+
prefix = flag.String("prefix", "spike", "filename prefix")
93+
metric = flag.String("metric", "cloudflare_zone_requests_total", "metric to keep (empty = all)")
94+
outdir = flag.String("outdir", "converge/testdata", "output directory")
95+
)
96+
flag.Parse()
97+
98+
token := os.Getenv("CF_API_TOKEN")
99+
if token == "" {
100+
log.Fatal("CF_API_TOKEN required")
101+
}
102+
if *zoneID == "" || *startS == "" || *endS == "" {
103+
log.Fatal("-zone, -start, -end are required")
104+
}
105+
106+
start, err := time.Parse(time.RFC3339, *startS)
107+
if err != nil {
108+
log.Fatalf("bad -start: %v", err)
109+
}
110+
end, err := time.Parse(time.RFC3339, *endS)
111+
if err != nil {
112+
log.Fatalf("bad -end: %v", err)
113+
}
114+
115+
client := &gqlClient{token: token}
116+
zones := []cfzones.Zone{{ID: *zoneID, Name: "captured"}}
117+
fetcher := cfetch.New(client, zones, nil)
118+
ctx := context.Background()
119+
120+
os.MkdirAll(*outdir, 0755)
121+
122+
seq := 0
123+
124+
// Live fetch: [end - lookback, end]
125+
liveStart := end.Add(-*lookb)
126+
writeFetch(ctx, fetcher, liveStart, end, seq, *prefix, *metric, *outdir)
127+
seq++
128+
129+
// Backfill: [start, end - lookback] in chunks
130+
cursor := start
131+
limit := end.Add(-*lookb)
132+
for cursor.Before(limit) {
133+
chunkEnd := cursor.Add(*chunk)
134+
if chunkEnd.After(limit) {
135+
chunkEnd = limit
136+
}
137+
writeFetch(ctx, fetcher, cursor, chunkEnd, seq, *prefix, *metric, *outdir)
138+
seq++
139+
cursor = chunkEnd
140+
}
141+
142+
log.Printf("wrote %d fixture files to %s", seq, *outdir)
143+
}
144+
145+
func writeFetch(ctx context.Context, f converge.Fetcher, start, end time.Time, seq int, prefix, metric, outdir string) {
146+
obs, err := f.Fetch(ctx, start, end)
147+
if err != nil {
148+
log.Printf("WARN: fetch [%s, %s] failed: %v", start, end, err)
149+
return
150+
}
151+
152+
rec := recordedFetch{
153+
Seq: seq,
154+
Start: start,
155+
End: end,
156+
}
157+
for _, o := range obs {
158+
if metric != "" && o.Key.Name != metric {
159+
continue
160+
}
161+
labels := make(map[string]string, len(o.Key.Labels))
162+
for _, l := range o.Key.Labels {
163+
labels[l.Name] = l.Value
164+
}
165+
rec.Observations = append(rec.Observations, recordedObservation{
166+
Name: o.Key.Name,
167+
Labels: labels,
168+
Value: o.Value,
169+
Bucket: o.Bucket,
170+
})
171+
}
172+
173+
path := fmt.Sprintf("%s/%s_%03d.json", outdir, prefix, seq)
174+
data, _ := json.MarshalIndent(rec, "", " ")
175+
os.WriteFile(path, data, 0600)
176+
log.Printf("%s: %d obs (filtered from %d) [%s, %s]",
177+
path, len(rec.Observations), len(obs),
178+
start.Format(time.RFC3339), end.Format(time.RFC3339))
179+
}

‎converge.go‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,6 @@ import (
1515

1616
const (
1717
argConvergeThreshold = "converge_threshold"
18-
argConvergeWindowTTL = "converge_window_ttl"
1918
argConvergePollInterval = "converge_poll_interval"
2019
argConvergeLookback = "converge_lookback"
2120
argConvergeMaxBackfill = "converge_max_backfill"
@@ -123,7 +122,6 @@ func cfetchdnsEnabledSet() map[string]bool {
123122
func convergeConfig() converge.Config {
124123
cfg := converge.DefaultConfig()
125124
cfg.Threshold = viper.GetInt(argConvergeThreshold)
126-
cfg.WindowTTL = viper.GetDuration(argConvergeWindowTTL)
127125
cfg.PollInterval = viper.GetDuration(argConvergePollInterval)
128126
cfg.Lookback = viper.GetDuration(argConvergeLookback)
129127
cfg.MaxBackfill = viper.GetDuration(argConvergeMaxBackfill)
@@ -142,6 +140,20 @@ func setupConvergerWithFetcher(ctx context.Context, component string, fetcher co
142140
return nil, err
143141
}
144142
cfg := convergeConfig()
143+
cfg.OnTick = func(ts converge.TickStats) {
144+
convergeIngestSamples.WithLabelValues(component).Add(float64(ts.IngestSamples))
145+
convergePostStabilizeUpdates.WithLabelValues(component).Add(float64(ts.PostStabilizeUpdates))
146+
convergeExpireFlushes.WithLabelValues(component).Add(float64(ts.ExpireFlushes))
147+
convergeExpireSamples.WithLabelValues(component).Add(float64(ts.ExpireSamples))
148+
convergeLiveFetchObservations.WithLabelValues(component).Add(float64(ts.LiveObservations))
149+
convergeBackfillFetchObservations.WithLabelValues(component).Add(float64(ts.BackfillObservations))
150+
convergeSnapshotSamples.WithLabelValues(component).Add(float64(ts.SnapshotSamples))
151+
convergeOpenWindows.WithLabelValues(component).Set(float64(ts.OpenWindows))
152+
convergeTrackerCount.WithLabelValues(component).Set(float64(ts.TrackerCount))
153+
convergeGaugeDownRevisions.WithLabelValues(component).Add(float64(ts.GaugeDownRevisions))
154+
convergeCounterRegressions.WithLabelValues(component).Add(float64(ts.CounterRegressions))
155+
convergePushErrors.WithLabelValues(component).Add(float64(ts.PushErrors))
156+
}
145157
return func(ctx context.Context) error {
146158
return converge.Run(
147159
converge.ContextWithLogger(ctx, log.WithField("component", component)),

‎converge/README.md‎

Lines changed: 94 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -50,9 +50,8 @@ eng := converge.NewEngine(cfg)
5050
// Feed observations. Returns samples for any tracker that just stabilized.
5151
samples := eng.Ingest(observations)
5252

53-
// Check TTLs. Returns forced-push samples from expiring windows.
54-
// Removes closed windows.
55-
samples = eng.Expire(time.Now())
53+
// Expire windows older than Lookback. Returns forced-push samples.
54+
samples = eng.Expire(time.Now(), true)
5655

5756
// Graceful shutdown: flush best-known values for all open windows.
5857
samples = eng.Flush()
@@ -141,9 +140,8 @@ Ingest/Expire/Flush yourself.
141140
```go
142141
converge.Config{
143142
Threshold: 3, // identical observations to stabilize
144-
WindowTTL: 15 * time.Minute, // window lifespan
145143
PollInterval: 30 * time.Second, // live lane tick
146-
Lookback: 10 * time.Minute, // live query range
144+
Lookback: 10 * time.Minute, // live query range and window TTL
147145
MaxBackfill: 2 * time.Hour, // startup backfill cap
148146
BackfillChunk: 10 * time.Minute, // time range per backfill call
149147
BackfillCallsPerTick: 1, // API budget for backfill per tick
@@ -157,6 +155,31 @@ Eager: Threshold=1 -- push on first sight, correct later
157155
Conservative: Threshold=3 -- wait for convergence, push once
158156
```
159157

158+
## Prometheus metrics
159+
160+
The root package registers per-tick counters and gauges on `/metrics`,
161+
labeled by `component` (`converge` or `converge-dns`). These help tune
162+
Threshold, Lookback, and MaxBackfill by making the tradeoffs observable.
163+
164+
| Metric | Type | What it tells you |
165+
|--------|------|-------------------|
166+
| `converge_ingest_samples_total` | counter | Tracker stabilizations. Responds to threshold tuning. |
167+
| `converge_post_stabilize_updates_total` | counter | Values that changed after push. High = threshold too low. |
168+
| `converge_expire_flushes_total` | counter | Windows force-flushed without stabilizing. High = threshold too high. |
169+
| `converge_expire_samples_total` | counter | Samples from TTL expiry (corrections on eviction). |
170+
| `converge_live_fetch_observations_total` | counter | Raw observation volume from CF live fetches. |
171+
| `converge_backfill_fetch_observations_total` | counter | Raw observation volume from backfill. |
172+
| `converge_snapshot_samples_total` | counter | Samples in the post-backfill snapshot. |
173+
| `converge_open_windows` | gauge | Current active windows. |
174+
| `converge_tracker_count` | gauge | Current active trackers. |
175+
| `converge_push_errors_total` | counter | Failed pushes to sink. |
176+
177+
Tuning signals:
178+
179+
- `post_stabilize_updates` climbing: threshold is too low, pushing before CF data settles.
180+
- `expire_flushes` climbing: threshold is too high, values never converge within the lookback window.
181+
- `open_windows` growing without bound: expire boundary issue or backfill stuck.
182+
160183
## Integration
161184

162185
The engine defines two interfaces:
@@ -243,6 +266,72 @@ the VM query endpoint, derived from the write endpoint.
243266

244267
Reference: [VictoriaMetrics Storage, Retention, Merging, and Deduplication](https://victoriametrics.com/blog/vmstorage-retention-merging-deduplication/)
245268

269+
## Why gauge-to-counter conversion is fundamentally hard
270+
271+
Cloudflare's API returns **gauges**: independent per-minute counts. Each
272+
bucket stands alone. "742 requests in the 14:10 minute" has no relationship
273+
to any other bucket.
274+
275+
Our dashboards expect **counters**: a monotonically increasing value where
276+
`rate()` computes the derivative. Counter at 14:10 = sum of all gauges from
277+
the beginning of time.
278+
279+
The conversion (prefix sum) creates three problems that don't exist with
280+
raw gauges:
281+
282+
**1. State that outlives a process.** A gauge is stateless. A counter
283+
carries accumulated history. When the exporter restarts, the accumulated
284+
base is lost. The counter resets to near zero, and `rate()` interprets the
285+
drop as either a reset (handled) or a massive negative spike (not handled
286+
by VM's "highest value wins" dedup). Gauges don't have this problem because
287+
each value is self-contained.
288+
289+
**2. Ordering dependencies between writes.** With gauges, you can push
290+
bucket 14:10 before 14:05 and nothing goes wrong. With counters, the value
291+
at 14:10 depends on every bucket before it. If a backfill chunk arrives and
292+
inserts earlier data, all later counter values shift (the cascade). You must
293+
either serialize all writes or suppress intermediate states. Gauges have no
294+
such dependency.
295+
296+
**3. Corrections amplify instead of replacing.** When CF revises a gauge
297+
(700 to 742), it's a simple overwrite. When we convert to a counter, that
298+
+42 correction cascades forward: every subsequent counter shifts by +42 and
299+
gets re-pushed. Combined with VM's "highest value wins" dedup, the old
300+
higher values from before the correction persist forever. Gauges just
301+
overwrite in place.
302+
303+
### The core tension
304+
305+
We're trying to maintain a **cumulative running total** from a source that
306+
provides **independent snapshots**, push it to a store that **doesn't
307+
support last-write-wins**, and do it across **process restarts** without
308+
losing the accumulated state.
309+
310+
Each of those properties in isolation is manageable. Together they create a
311+
combinatorial surface of edge cases: restart + backfill + correction + dedup
312+
= spike artifacts that are permanent in VM.
313+
314+
### What we've built to manage it
315+
316+
- Suppress + snapshot (don't push during backfill, push final state once)
317+
- Chain base seeding (read back last value from VM on restart)
318+
- Expire boundary fix (prevent double-counting at the lookback edge)
319+
- Label sort normalization (prevent duplicate chains)
320+
- Prometheus observability counters (make tuning data-driven)
321+
- Integration tests + fixture capture (reproduce and regression-test spikes)
322+
323+
Each fix addresses a real incident, but the root cause is the same:
324+
counters have global state, gauges don't.
325+
326+
### The alternative
327+
328+
Push gauges to VM and let the query layer compute `sum(increase(...))`
329+
instead of `rate()`. This eliminates all the complexity above: no chain
330+
state, no base seeding, no cascade, no dedup sensitivity, no restart
331+
artifacts. The tradeoff is that existing dashboards using `rate()` would
332+
need to change, and `increase()` over gauges is slightly less ergonomic
333+
than `rate()` over counters.
334+
246335
## File layout
247336

248337
```

‎converge/counter_chain.go‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,13 +22,15 @@ type counterEmission struct {
2222
// and the chain reports which values need re-pushing.
2323
//
2424
// This is a prefix sum array with point-update support. For the small number
25-
// of active buckets (bounded by WindowTTL / bucket_interval, roughly 15), a
25+
// of active buckets (bounded by Lookback / bucket_interval, roughly 15), a
2626
// sorted slice with linear recomputation is optimal. A Fenwick tree (Binary
2727
// Indexed Tree) solves the same problem class in O(log n) per operation but
2828
// adds complexity that isn't justified at this scale.
2929
type counterChain struct {
30-
base uint64 // accumulated sum of evicted (expired) buckets
31-
entries []chainEntry // sorted ascending by bucket time
30+
base uint64 // accumulated sum of evicted (expired) buckets
31+
entries []chainEntry // sorted ascending by bucket time
32+
gaugeDownRevisions uint64 // CF revised a gauge downward
33+
counterRegressions uint64 // computed counter was lower than previous emission
3234
}
3335

3436
type chainEntry struct {
@@ -52,6 +54,9 @@ func (c *counterChain) Set(bucket time.Time, gauge uint64) []counterEmission {
5254
if c.entries[idx].gauge == gauge {
5355
return nil
5456
}
57+
if gauge < c.entries[idx].gauge {
58+
c.gaugeDownRevisions++
59+
}
5560
c.entries[idx].gauge = gauge
5661
} else {
5762
// New entry: insert at sorted position.
@@ -73,6 +78,9 @@ func (c *counterChain) Set(bucket time.Time, gauge uint64) []counterEmission {
7378
newCounter := prev + c.entries[i].gauge
7479

7580
if newCounter != c.entries[i].counter || i == idx {
81+
if c.entries[i].counter > 0 && newCounter < c.entries[i].counter {
82+
c.counterRegressions++
83+
}
7684
c.entries[i].counter = newCounter
7785
emissions = append(emissions, counterEmission{
7886
Bucket: c.entries[i].bucket,

0 commit comments

Comments
 (0)