chore: migrate from filesystem to object store

[caterryan]

Object store migration plan

Tracking document for migrating influxdata/ plugins to the horizontally distributed
processing engine. Audit date: 2026-07-21 (all 31 plugins in influxdata/ reviewed).

Target environment assumptions

• Horizontally distributed workers. Any invocation of a trigger may run on any
  worker. In-process state (module globals, influxdb3_local.cache) is per-worker,
  not shared.
• Ephemeral local filesystem. The filesystem exists and may be used as scratch
  space within a single invocation. It must NOT be used for persistence between
  invocations — contents may vanish at any time (worker recycle, rescheduling).
• Object store API (obstore-based). Shared between all workers, isolated between
  triggers. Provides get/put/list/delete, range reads, file-like
  open_reader, and conditional puts (mode="create", etag-based CAS).
  Reference: https://github.com/developmentseed/obstore

The allowed pattern (for docs / CONTRIBUTING)

Local filesystem = scratch within one invocation.
Object store = anything that must outlive the invocation.
Fetch bytes from the object store, write to a tempfile if a library needs a path,
clean up before returning.

Migration inventory

Status legend: unchanged = no code change required · mechanical = straightforward
port · port = real but bounded work · redesign = architectural change.

Plugin	Filesystem usage today	Action	Risk
simple_data_replicator	gzip queue files: append/read/rewrite/rotate (simple_data_replicator.py:273-294) + TOML	redesign	High
prophet_forecasting	Model save/load across invocations (prophet_forecasting.py:1197-1209, 1592-1606) + TOML	port	Medium
chronos_forecasting	HF model cache /tmp/hf_cache (chronos_forecasting.py:155-158) + TOML	unchanged (perf work optional)	Medium (perf/cost)
kafka_subscriber	Within-invocation protoc tempdirs (kafka_subscriber.py:2031-2123); reads TOML, CA certs, .avsc/.proto schemas	unchanged compile; asset reads depend on config-distribution decision	Low-Medium
influxdb_to_iceberg	TOML; user-supplied pyiceberg catalog may target local fs	mechanical + docs	Low
opcua	TOML + client cert/private key files (opcua.py:122-128)	mechanical (tempfile pattern)	Low
mqtt_subscriber	TOML + CA cert file	mechanical (tempfile pattern)	Low
amqp_subscriber	TOML + CA cert file	mechanical (tempfile pattern)	Low
schema_validator	TOML + JSON schema file (schema_validator.py:148)	mechanical	Low
system_metrics	TOML; psutil reads /proc, statvfs	unchanged + docs note	Low (semantic)
basic_transformation	TOML config only	mechanical	Low
downsampler	TOML config only	mechanical	Low
forecast_error_evaluator	TOML config only	mechanical	Low
import	TOML config only	mechanical	Low
mad_check	TOML config only	mechanical	Low
resampler	TOML via influxdata-plugin-utils	mechanical (via shared loader)	Low
river_anomaly_detector	TOML only (state checkpoints already in DB)	mechanical	Low
river_auto_profiler	TOML only (state checkpoints already in DB)	mechanical	Low
river_forecaster	TOML only (state checkpoints already in DB)	mechanical	Low
sagemaker	TOML config only	mechanical	Low
signal_filter	TOML via influxdata-plugin-utils	mechanical (via shared loader)	Low
state_change	TOML config only	mechanical	Low
stateless_adtk_detector	TOML config only	mechanical	Low
stock_plugin	TOML config only (also latent __file__-style plugin-dir default)	mechanical	Low
threshold_deadman_checks	TOML config only	mechanical	Low
valuecounter	TOML config only	mechanical	Low
bird_data_simulator	None	unchanged	— (see cache caveat)
notifier	None	unchanged	—
nws_weather	None	unchanged	—
signal_generator	None	unchanged	— (see cache caveat)
synthefy_forecasting	None	unchanged	—

Not plugins: library/ (metadata JSON). Test files (test_valuecounter.py,
test_signal_filter.py) use tempfile at dev time only — no action.

Cross-cutting work items (do these first)

1. Decide config/asset distribution

~22 plugins read operator-placed files (TOML configs, JSON schema, CA certs,
client keys, .avsc/.proto schemas) from PLUGIN_DIR. Plugins never write these.
The decision is deployment, not code:

• Plugin dir ships to every worker as a deployment artifact → all TOML-only
  plugins run unchanged.
• Only code ships → config/asset delivery moves to the object store and every
  reader must be ported (mechanical, via item 2).

This single decision determines whether ~22 plugins change at all.

2. Centralize the loader in influxdata-plugin-utils

Do not port path resolution 20+ times. influxdata-plugin-utils/src/influxdata_plugin_utils/config.py
already owns TOML loading for resampler and signal_filter
(load_plugin_config, resolution chain PLUGIN_DIR → INFLUXDB3_PLUGIN_DIR →
VIRTUAL_ENV parent). Plan:

1. Add an object-store backend to load_plugin_config (config key instead of path).
2. Add a shared asset-fetch helper: object store bytes → tempfile → return path
   (for libraries that require filesystem paths), cleaned up per invocation.
3. Migrate the remaining ~20 plugins off their copy-pasted resolution blocks onto
   the shared loader. Turns the migration into 1 real change + N mechanical ones.
4. Add per-worker in-memory TTL caching for config reads — object-store GET per
   trigger invocation adds an RTT to every write batch otherwise.

3. Add a shared-asset namespace to the object store design

Per-trigger isolation fits mutable state (queues, models) but fights shared
assets: configs, CA certs, proto schemas, and model weights are shared across
many triggers today. Per-trigger-only storage means duplicating every cert per
trigger and re-uploading on rotation × N triggers, and blocks model-cache seeding
for chronos_forecasting. Recommendation: two namespaces —
shared read-only assets + per-trigger read-write state.

Per-plugin details (plugins requiring work)

simple_data_replicator — redesign (highest risk)

The plugin's core purpose is cross-invocation persistence: buffer failed writes on
disk, replay in later invocations. This is exactly what ephemeral fs forbids —
failure mode is silent data loss on worker recycle, the worst outcome for a
replication plugin.

Current mechanics (simple_data_replicator.py:273-294): gzip.open(queue_file, "at")
append, size-based rotation via os.path.getsize, read-then-rewrite truncation.
Object stores have no append and no truncate; a single growing object is racy and
O(size) per write.

Redesign sketch:

• One object per failed batch: put under a unique key
  (e.g. queue/<timestamp>-<uuid>.lp.gz), list + replay + delete on success.
  This maps to object-store semantics better than the current append log.
• Concurrency: overlapping executions of the same trigger on different workers can
  double-replay. Use conditional put (mode="create") to take a claim/lease per
  batch, or etag CAS. Open question: does the engine serialize executions per
  trigger? Answer determines locking depth.
• Preserve max-queue-size semantics via object count/size budget during list.

prophet_forecasting — port (silent behavior change if skipped)

Models are written in one invocation and read in later ones
(prophet_forecasting.py:1197-1209, 1592-1606; dir setup at 612-617). On
ephemeral fs this does not error — the plugin quietly retrains from scratch after
every worker recycle: different forecasts, wasted compute, no log signal.

• Port model save/load to object store get/put (models are small serialized
  JSON — straightforward).
• Consider etag CAS if concurrent trainers on two workers matter; otherwise
  last-write-wins.
• Bonus fix: prophet_forecasting.py:612 uses Path(__file__), which is undefined
  under the server's exec()-based runtime (see comment in
  stock_plugin/stock_plugin.py:960). The port removes this latent bug.

chronos_forecasting — unchanged functionally; perf/cost work optional

/tmp/hf_cache + HF_HOME (chronos_forecasting.py:155-158) is cache, not
persistence: when missing, from_pretrained re-downloads. torch mmap requires a
local file and gets one. Already compliant with the ephemeral-fs contract.

Risks are economic: every fresh worker re-downloads model weights (hundreds of MB
to ~GB, 30s+ cold start), multiplied by thundering herd on scale-out, and repeated
per trigger if the store stays per-trigger-isolated.

Mitigations (pick as needed):

• Shared-asset namespace holding model blobs; seed the local HF cache from it.
• Bake common models into the worker image.
• Worker affinity for model-heavy triggers.

kafka_subscriber — compile flow unchanged; asset reads follow item 1

The protobuf compile flow (kafka_subscriber.py:2031-2123) — tempfile.mkdtemp →
write .proto files → grpc_tools.protoc → read descriptor set →
shutil.rmtree — is entirely within-invocation scratch. Allowed pattern; no change.

• Local schema files (.avsc, .proto) and CA certs move per the config/asset
  distribution decision (tempfile pattern for protoc inputs and librdkafka certs).
• Optional perf: cache the compiled FileDescriptorSet bytes in the object store
  (compile once, other workers load in-memory via descriptor_pb2) to avoid
  recompiling on every cold worker.

TLS/cert readers — mqtt_subscriber, amqp_subscriber, opcua (+ kafka)

Native libs (paho, pika, librdkafka, asyncua) want filesystem paths for cert/key
material. With ephemeral fs allowed, the universal pattern is: fetch PEM bytes from
object store → write tempfile → hand path to the library unchanged → clean up.
No per-library SSLContext/cadata surgery needed. Use the shared asset-fetch
helper from cross-cutting item 2.

Security note: private keys (opcua.py:128) move from operator-controlled local
disk into the shared object store — confirm encryption-at-rest / access-control
posture as part of the API design.

influxdb_to_iceberg — docs only

pyiceberg has native S3/GCS/Azure FileIO. Document that file:// warehouses and
sqlite-file catalogs are unsupported (both are filesystem persistence smuggled in
via user config); REST/Glue/object-store catalogs are fine.

system_metrics — docs only

psutil reads /proc and statvfs — always available, but metrics now describe
whichever worker ran the invocation. Add a docs note; consider making the hostname
tag more prominent, or node-pinning guidance.

Non-filesystem distribution caveats (same migration, different API)

Horizontal distribution breaks in-process state regardless of the object store.
Audit influxdb3_local.cache semantics in the distributed engine; if it remains
per-worker:

• signal_generator — last-emitted-time and gap seed in cache
  (signal_generator.py:635-657): trigger hopping workers produces gaps/overlaps
  in the generated signal.
• bird_data_simulator — simulation state in cache (bird_data_simulator.py:32,
  366-390): state resets on worker change.
• river_anomaly_detector / river_auto_profiler / river_forecaster — models
  live in process memory between DB checkpoints (_system.model_checkpoints);
  two workers can train divergent copies between checkpoint/restore cycles.
• sagemaker — config/columns cache (sagemaker.py:146-148): per-worker misses
  only; cosmetic.

Either back influxdb3_local.cache with a shared store in the distributed engine,
or document these plugins as best-effort under distribution.

Suggested sequencing

1. Land the object store API (+ shared-asset namespace decision, cross-cutting item 3).
2. Decide config/asset distribution (cross-cutting item 1).
3. Add object-store backend + asset-fetch helper + TTL cache to
   influxdata-plugin-utils (cross-cutting item 2).
4. Redesign simple_data_replicator (only architectural change).
5. Port prophet_forecasting model persistence.
6. Mechanical passes: cert readers (opcua, mqtt_subscriber, amqp_subscriber,
   kafka_subscriber assets), schema_validator, then TOML-only plugins onto the
   shared loader (skip if config ships as deployment artifact).
7. Optional perf: kafka descriptor-set caching, chronos model seeding.
8. Docs: CONTRIBUTING "allowed pattern" paragraph, influxdb_to_iceberg catalog
   constraints, system_metrics semantics, cache-API caveats.