chore: migrate from filesystem to object store

@caterryan is already working on this.

Since Jul 22, 2026.

# 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:

### 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/-.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.