influxdb(seed): chunk per year + pipefail; drop bucket-data idempotency
Last run completed with exit 0 but the actual stream got cut off: Error: failed to execute query: Post ".../api/v2/query?org=renner": EOF [21:50:31] Seed complete. Two problems: 1. Pivoting the full bucket in one Flux query materialises the entire result set in legacy's memory before flushing. A multi-year HA- metrics bucket exceeds what legacy can hold; the connection drops mid-stream with EOF. Subsequent verification showed *some* data landed but coverage was unclear (count() queries against legacy were timing out at 60s afterwards — server slow under load). 2. Script had `set -eu` but no `pipefail`, so `influx query`'s non- zero exit didn't propagate through the pipe. `influx write` saw the truncated stream, returned 0, script logged "Seed complete." K8s marked the Job Complete; Flux saw success and stopped trying. A genuinely broken transfer was reported as a successful seed. Fixes: - Loop the pivot per calendar year (2018 → current). Each year's pivot produces a much smaller result set; legacy holds it without OOM-ing. Year boundaries don't need to align with anything in InfluxDB — points retain their original timestamps end-to-end. - `set -euo pipefail` so `influx query` EOF on any year aborts the loop with non-zero exit, the Job goes to Failed, the K8s backoff retries. - Drop the script-side `query|grep ^,_result` idempotency check. Two reasons: (a) InfluxDB writes are upserts on (timestamp,measurement,tagset,field), so a partial run leaves the bucket in a state where re-running just re-asserts the same points — no corruption, no duplicates; (b) the previous check would short- circuit after a partial-success leaving later years unprocessed. The K8s Job's `Complete` state is the actual idempotency guard: Flux reconcile sees Complete and won't recreate. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.7
@@ -90,43 +90,57 @@ spec:
|
||||
- sh
|
||||
- -c
|
||||
- |
|
||||
set -eu
|
||||
set -euo pipefail
|
||||
log() { printf '[%s] %s\n' "$(date -u +%H:%M:%S)" "$*"; }
|
||||
|
||||
# Idempotency: if target bucket already has data, skip.
|
||||
ROWS=$(influx query --raw \
|
||||
'from(bucket: "'"$INFLUX_BUCKET"'") |> range(start: -100y) |> limit(n: 1)' 2>/dev/null \
|
||||
| grep -c '^,_result' || true)
|
||||
if [ "$ROWS" -gt 0 ]; then
|
||||
log "$INFLUX_BUCKET already has data — seed skipped."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Stream legacy bucket → local bucket through the v2 HTTP
|
||||
# API. Annotated-CSV is the wire format on both ends, but
|
||||
# `influx query` emits LONG format (one row per
|
||||
# _time/_measurement/_field/_value tuple) while `influx
|
||||
# write --format csv` requires WIDE format (one row per
|
||||
# _time/_measurement, each field as its own column). The
|
||||
# `pivot()` step in the Flux query collapses _field+_value
|
||||
# into wide columns. `drop()` removes the _start/_stop
|
||||
# range columns that `influx write` would otherwise
|
||||
# complain about.
|
||||
log "Streaming $LEGACY_HOST/$LEGACY_BUCKET → $INFLUX_HOST/$INFLUX_BUCKET (annotated CSV, wide format) ..."
|
||||
influx query --raw \
|
||||
--host "$LEGACY_HOST" \
|
||||
--token "$LEGACY_TOKEN" \
|
||||
--org "$LEGACY_ORG" \
|
||||
'from(bucket: "'"$LEGACY_BUCKET"'")
|
||||
|> range(start: 1970-01-01T00:00:00Z)
|
||||
|> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
|
||||
|> drop(columns: ["_start", "_stop"])' \
|
||||
| influx write \
|
||||
--host "$INFLUX_HOST" \
|
||||
--token "$INFLUX_TOKEN" \
|
||||
--org "$INFLUX_ORG" \
|
||||
--bucket "$INFLUX_BUCKET" \
|
||||
--format csv
|
||||
# API in per-year chunks. Annotated-CSV is the wire format
|
||||
# on both ends, but `influx query` emits LONG format (one
|
||||
# row per _time/_measurement/_field/_value tuple) while
|
||||
# `influx write --format csv` requires WIDE format (one row
|
||||
# per _time/_measurement, each field as its own column).
|
||||
# The `pivot()` step collapses _field+_value into wide
|
||||
# columns; `drop()` strips the _start/_stop range columns
|
||||
# that `influx write` rejects.
|
||||
#
|
||||
# Why chunked: pivoting the entire bucket at once
|
||||
# materialises the full result set in legacy's memory
|
||||
# before flushing — for a multi-year HA-metrics bucket
|
||||
# that exceeds available memory and the connection drops
|
||||
# mid-stream (`Post .../query: EOF`). One year at a time
|
||||
# keeps each pivot under a few hundred MB.
|
||||
#
|
||||
# No script-side idempotency check: InfluxDB writes are
|
||||
# upserts on (timestamp,measurement,tagset,field), so
|
||||
# re-running this Job re-applies the same data without
|
||||
# corrupting it. The K8s Job's `Complete` state is the
|
||||
# post-success guard against re-runs — Flux reconcile
|
||||
# sees Complete and leaves the Job alone.
|
||||
#
|
||||
# `pipefail` is on so an `influx query` EOF mid-pipe
|
||||
# propagates a non-zero exit code (without it, `influx
|
||||
# write` swallows the empty stream and returns 0,
|
||||
# producing a silent partial-success).
|
||||
START_YEAR=2018
|
||||
END_YEAR=$(date -u +%Y)
|
||||
for YEAR in $(seq "$START_YEAR" "$END_YEAR"); do
|
||||
NEXT=$((YEAR + 1))
|
||||
log "Streaming $YEAR-01-01 → ${NEXT}-01-01 ..."
|
||||
influx query --raw \
|
||||
--host "$LEGACY_HOST" \
|
||||
--token "$LEGACY_TOKEN" \
|
||||
--org "$LEGACY_ORG" \
|
||||
"from(bucket: \"$LEGACY_BUCKET\")
|
||||
|> range(start: ${YEAR}-01-01T00:00:00Z, stop: ${NEXT}-01-01T00:00:00Z)
|
||||
|> pivot(rowKey: [\"_time\"], columnKey: [\"_field\"], valueColumn: \"_value\")
|
||||
|> drop(columns: [\"_start\", \"_stop\"])" \
|
||||
| influx write \
|
||||
--host "$INFLUX_HOST" \
|
||||
--token "$INFLUX_TOKEN" \
|
||||
--org "$INFLUX_ORG" \
|
||||
--bucket "$INFLUX_BUCKET" \
|
||||
--format csv
|
||||
done
|
||||
|
||||
log "Seed complete."
|
||||
volumeMounts:
|
||||
|
||||
Reference in New Issue
Block a user