v1.3.1 Design Plan: Placement Completion & Targeted Backend Follow-ups

Status: Draft for review Target version: v1.3.1 Scope: Manager and standalone modes. Worker mode is execution-only and does not own control-plane behavior.

Goal

v1.3.1 should finish the runtime work that v1.3.0 intentionally deferred, then land a small set of targeted backend follow-ups without mixing in the still-open metrics/stats and UDP broadcast design work.

This release contains:

workers.count: N runtime
workers.list: [...] runtime
Dynamic Kubernetes service provisioning
Scoped filename variables and partitioned file output
SNMP migration to pysnmp 7.x

This release does not contain:

standalone live stats parity
manager operational metrics
UDP broadcast for syslog / snmp_trap
UI revalidation

Those move to later releases because they need separate design work and cluster validation.

Design Decisions

1. Filename rendering needs scoped variables, not ad-hoc metadata tokens

This is broader than simply adding source_stem / source_suffix.

TRAM needs a clearer variable model for filename rendering:

pipeline/runtime scope
source scope
record-field scope
sink-file scope

Examples:

pipeline/runtime: pipeline, run_id, timestamp, part
source scope: source_filename, source_stem, source_suffix, source_path
record-field scope: field.nf_id, field.ne_id
sink-file scope: file-open timestamp, rolling part/index

Implementation still belongs primarily at the shared file-sink helper layer, but the semantics must be defined first. This is not just “whatever happens to be in meta”.

The important point is:

values may be metadata-derived at runtime
but the contract should be expressed as scoped filename variables, not implicit per-record metadata leakage

2. Dynamic K8s service provisioning should support both manager and standalone

This should not be manager-only.

The owner should be the control-plane process:

standalone mode: the daemon can create and manage a Service targeting itself
manager mode: the manager can create and manage a Service targeting the worker ingress path
worker mode: never owns service lifecycle

The behavior is mode-aware:

standalone: service targets the standalone daemon pod / service port
manager: service targets worker ingress (:8767) for push-source traffic

This makes the feature generally useful for creating a dedicated external endpoint where needed, while keeping lifecycle ownership in the control plane.

3. `workers.count: N` and `workers.list` are different invariants

These two features share some placement plumbing but must not be treated as one algorithm.

count: N

durable slot identity
mutable worker assignment
spare-worker replacement allowed

list: [...]

durable slot identity
worker identity pinned
spare-worker replacement forbidden

If these are blurred together, the reconciler will be wrong.

Current Code Baseline

The schema already supports forward-compatible workers config:

WorkersConfig exists in tram/models/pipeline.py
workers.list is accepted via alias to worker_ids
defaulting is already applied by source type

The current runtime still blocks the deferred modes:

WorkerPool.resolve() raises on workers.list
WorkerPool.resolve() raises on integer count > 1
PlacementReconciler is still effectively count: all logic

Relevant code surfaces:

tram/agent/worker_pool.py
tram/agent/reconciler.py
tram/pipeline/controller.py
tram/persistence/db.py
tram/connectors/file_sink_common.py
tram/connectors/*/sink.py
tram/connectors/snmp/*
tram/api/routers/mibs.py
tram/cli/main.py

Workstream A: `workers.count: N`

Objective

Support exact-N placement on healthy workers for valid multi-worker sources.

Runtime Model

count: N means:

create N logical slots
fill them with the least-loaded healthy workers at dispatch time
if one worker fails, reassign that slot to a healthy spare not already assigned
if the failed worker comes back after replacement, do nothing if target is already satisfied
if healthy capacity drops below N, placement becomes degraded
when healthy capacity recovers, fill missing slots until target is met

Invariants

slot index is durable identity
worker assignment is mutable
slot.current_run_id changes on re-dispatch
slot.worker_url and slot.worker_id may change during failover

Code Changes

tram/agent/worker_pool.py

extend resolve() to return top N healthy workers
extend multi_dispatch() to create slot data for N

tram/agent/reconciler.py

add spare-worker gap fill logic
avoid churn when target already satisfied
distinguish degraded caused by insufficient capacity from transient reconciling

tram/pipeline/controller.py

persist placement state with mutable worker assignment
ensure stop/restart behavior works for reassigned slots

tram/persistence/db.py

existing slots_json model already supports mutable current_run_id
verify updates are correct when worker_url changes for a slot

Tests

start N=3 on 5 workers
one worker fails and a spare fills the slot
original worker returns, no unnecessary rebalance
two slots fail simultaneously
pool shrinks below target
restart reconciliation with reassigned slots

Acceptance

no duplicate slot assignment
no oscillation across workers while target is satisfied
placement converges to running or degraded deterministically

Workstream B: `workers.list: [...]`

Objective

Support named-worker placement with pinned recovery semantics.

Runtime Model

workers.list means:

one logical slot per listed worker ID
slots may start degraded if some listed workers are unavailable
only the same listed worker may satisfy that slot later
unrelated healthy workers are never used as substitutes

Invariants

slot index is durable identity
listed worker identity is pinned to the slot
worker_url may change only if the same worker ID resolves to a new URL after recovery
no spare-worker substitution

Recovery rule

For workers.list, the pinned identity is the worker_id, not the last known URL.

Recovery logic must therefore re-resolve the current worker_url from the live _worker_ids map when a worker returns. The reconciler must not assume that the previously stored URL is still valid after pod restart, reschedule, or IP change.

Because _worker_ids is refreshed by health polling, dispatch/recovery must re-resolve from the live map at the moment of dispatch rather than blindly reusing slot.worker_url from persisted placement state.

Code Changes

tram/agent/worker_pool.py

map worker_ids to URLs via _worker_ids
validate listed IDs against discovered worker health map

tram/agent/reconciler.py

retry stale/dead slots only on their pinned worker
keep placement degraded while listed worker is unavailable

tram/pipeline/controller.py

persist pinned worker identity per slot

Tests

all listed workers healthy
one listed worker absent at startup
one listed worker goes stale
unrelated healthy worker is not chosen
restart reconciliation with partially restored listed workers

Acceptance

listed placements never drift to non-listed workers
degraded state remains visible until pinned workers return

Workstream C: Filename variables, classified routing, and dynamic partitioning

Objective

Add a proper filename-variable model for file-based sinks, separating the low-risk naming work from the heavier dynamic partitioning work.

Intended sinks

local
ftp
sftp
s3
gcs
azure blob

Variable scopes

Recommended initial filename-variable scopes:

runtime variables
- {pipeline}
- {timestamp}
- {epoch}
- {part}
source variables
- {source_filename}
- {source_stem}
- {source_suffix}
record-field variables
{field.nf_name}
{field.ne_id}
{field.site}

Supported record-field lookup syntax:

{field.nf_name}
{field.ne_name}

Only dotted field syntax is supported in v1.3.1. Do not support colon-based field syntax.

Slice C1: Runtime/source-context naming plus explicit classified routing

This is the lower-risk slice and should land first.

It includes:

runtime variables such as {pipeline}, {timestamp}, {epoch}, {part}
source-context variables such as {source_filename}, {source_stem}, {source_suffix}
bounded class routing using the existing per-sink condition feature

Examples:

filename_template: "STP_{epoch}.csv"

filename_template: "{source_stem}_{part}.csv"

sinks:
  - type: sftp
    condition: 'nf_name == "MSC"'
    filename_template: "MSC_{part}.csv"

  - type: sftp
    condition: 'nf_name == "SMSC"'
    filename_template: "SMSC_{part}.csv"

For bounded known classes, the existing per-sink condition path is the preferred approach. No new partitioning engine is required for that pattern.

Source-derived behavior

For file-based sources, derive source variables from source context:

file.csv -> source_stem=file, source_suffix=.csv
README -> source_stem=README, source_suffix=""

Fallback when source filename is absent:

source_filename="data"
source_stem="data"
source_suffix=""

Slice C2: Dynamic record-field partitioning

This is the heavier slice and depends on executor-side partitioning before serialization.

Record-field behavior

If a filename template references a record field such as {field.nf_id}, the sink should use partitioned output:

separate active files are maintained per distinct field value
rolling (max_records, max_time, max_bytes) applies per partition
missing field values use a stable placeholder partition key such as unknown

Example:

filename_template: "{field.nf_id}_{timestamp}_{part}.ndjson"
file_mode: append
max_records: 1000

If incoming records contain SMSC, MME, and UPF, the sink maintains separate rolling files for each partition key.

Default partitioning rule:

partition by referenced record field value only
source context does not implicitly participate in partitioning
source-derived variables only affect partitioning/file identity when they are explicitly present in the template

Examples:

"{field.nf_name}_{timestamp}_{part}.ndjson" partitions by nf_name only
"{field.nf_name}_{source_stem}_{timestamp}_{part}.ndjson" effectively partitions by nf_name + source_stem

Validation / lint behavior

Do not block expert usage of arbitrary fields.

But add lint warnings for suspicious filename fields that are likely to create runaway file counts or meaningless file naming, for example:

timestamp
ts
event_time
value
counter
bytes
latency

The warning should be advisory, not blocking.

Design implications

This feature is not only token rendering.

It also requires:

filename-variable scope definition
partition-aware file sink state management
lint guidance for likely high-cardinality choices

Chosen architectural direction

Record-field partitioned output does not fit cleanly into the current sink contract unless partitioning happens before serialization:

write(data: bytes, meta: dict)

Today the sink receives already-serialized bytes. Partitioned output by {field.nf_id} or {field.ne_id} therefore must split records by partition key before bytes are finalized.

Chosen direction for v1.3.1:

executor-side partitioning
split a batch by partition key before sink writes
serialize each partition separately
keep sinks byte-oriented with the existing write(data: bytes, meta: dict) contract

Sink-side partitioning is explicitly out of scope for v1.3.1.

Implementation status gate

Workstream C is unblocked only with the executor-side partitioning model above. Implementation should not redesign the sink interface in v1.3.1.

Implementation

Slice C1 implementation:

extend shared filename rendering to support runtime/source-context variables
keep behavior uniform across all file sinks
document bounded classified routing via existing per-sink condition
add tests for conditional sink routing to separate output files for known classes

Slice C2 executor work:

inspect structured records before serialization
compute partition keys from filename-variable references
group records by partition key
serialize each partition independently
call sink write() once per partition with partition-specific meta

Shared helper change in tram/connectors/file_sink_common.py:

extend token rendering helper to resolve scoped filename variables
derive source variables uniformly
support record-field lookup
support per-partition file state keys

Model updates in tram/models/pipeline.py:

documentation/comments for supported filename variable scopes

Per-sink work:

update append/rolling sinks to maintain separate state per partition key when C2 record-field variables are used
avoid sink-specific variable-derivation logic where possible

Tests

pure runtime naming, e.g. STP_{epoch}.csv
source filename with single suffix
source filename with no suffix
fallback when source filename missing
bounded classified routing using per-sink condition
record-field partitioning with multiple values
rolling behavior per partition
rendering in each affected sink family

Acceptance

scoped filename-variable behavior is defined and documented
source-derived naming is uniform across file sinks
bounded known-class routing is documented as an explicit supported pattern via per-sink condition
record-field naming produces partitioned output rather than ambiguous shared filenames
executor performs partitioning before serialization; sink interface remains byte-oriented
missing record fields fall back to a stable placeholder partition/file name
lint warnings exist for obviously risky field choices

Workstream D: SNMP migration to `pysnmp 7.x`

Objective

Migrate SNMP runtime and tooling to pysnmp 7.x while preserving poll, trap, and MIB-related functionality.

Areas affected

Connector runtime:

tram/connectors/snmp/source.py
tram/connectors/snmp/sink.py
tram/connectors/snmp/mib_utils.py

MIB tooling:

tram/api/routers/mibs.py
tram/cli/main.py

Tests:

tests/unit/test_snmp_connectors.py
tests/unit/test_mib_utils.py
tests/integration/test_mib_resolve.py

Implementation plan

inventory all imports and runtime assumptions tied to legacy PySNMP packaging
migrate connector runtime imports and API calls
migrate MIB view / utility code
migrate CLI and API MIB compile/download paths
update dependency pins and docs
update error messages that currently instruct users to install legacy PySNMP package names

Important rule

This migration is library/runtime work only.

It must not be mixed with UDP broadcast enablement in the same implementation slice.

Acceptance

poll source works
trap source/sink code imports and runs
MIB resolution still works
active docs/tests no longer reference deprecated PySNMP package names

Workstream E: Dynamic K8s Service Provisioning

Objective

Allow a pipeline to request a dedicated Kubernetes Service-managed endpoint when needed.

Ownership model

Control-plane owned:

standalone daemon owns service lifecycle in standalone
manager owns service lifecycle in manager
workers never create/delete services

Source scope

Recommended initial scope:

push HTTP sources only in v1.3.1
- webhook
- prometheus_rw

UDP sources stay out of scope until v1.3.2.

Mode-aware target model

Standalone:

service targets the standalone daemon
used to expose a dedicated external endpoint for the local /webhooks/* path

Manager:

service targets worker ingress port :8767
used to expose a dedicated endpoint for worker-side push traffic

Worker:

no control-plane ownership

Recommended initial feature boundary

To keep v1.3.1 contained, recommend starting with NodePort only.

This gives a usable endpoint-management feature without expanding immediately into:

LoadBalancer cloud-provider behavior
ingress-controller integration
UDP service rules

Required design points

Pipeline config block, likely:

kubernetes:
  enabled: true
  service_type: NodePort
  node_port: 30042
  service_name: ""

Resolved decisions for v1.3.1:

block name is kubernetes
service_type is the top-level selector for endpoint mode
supported values in v1.3.1:
- NodePort
- LoadBalancer
node_port is optional and only applies when service_type=NodePort
service creation happens only while the pipeline is active
custom service_name is supported
if service_name is not provided, use deterministic generated naming

Default naming rule:

compute pipeline-slug from pipeline.name:
- lowercase
- replace _ with -
- collapse repeated -
- trim leading/trailing -
sanitize to DNS-1123-compatible form
truncate to fit Kubernetes 63-character service-name limit
append a short deterministic hash suffix derived from the original pipeline name to avoid collisions for long names or truncation

Recommended default shape:

tram-p-<pipeline-slug>-<hash8>

This is safer than embedding the full pipeline name without truncation logic.

Implementation areas

tram/models/pipeline.py
new Kubernetes service manager helper/module
tram/pipeline/controller.py
Helm RBAC templates
Helm ServiceAccount wiring
Helm pod spec serviceAccountName
optional dependency wiring for Kubernetes client

Lifecycle rules

create on start / schedule activation when eligible
update on config change while active
do not create merely because YAML was registered
remove on delete
remove on stop/disable when the pipeline is no longer active
reconcile on startup from DB state
disable/no-op outside supported modes

Tests

config validation
mocked service create/update/delete
restart reconciliation
standalone vs manager branching
worker mode no-op
Helm render checks for RBAC

Acceptance

service lifecycle is deterministic and restart-safe
ownership is clear and mode-aware
no service management logic runs in worker mode
the control-plane pod spec explicitly sets serviceAccountName
that ServiceAccount is actually bound to the created Role / RoleBinding
long pipeline names do not breach Kubernetes service-name limits
optional custom service_name overrides generated naming

Recommended PR Split

workers.count: N
workers.list: [...]
filename variables + classified routing (C1)
dynamic field partitioning (C2)
SNMP migration
K8s service provisioning

This keeps the placement runtime work isolated first, then lands the independent follow-ups in separate slices.

Test Matrix

Unit

worker pool resolution for count: N
reconciler failover behavior
pinned-worker behavior for workers.list
filename token rendering
record-field partitioning behavior
SNMP import and runtime mocks
K8s lifecycle mocks

Integration

placement persistence and restart recovery
file sink token rendering in representative pipelines
MIB resolution with installed SNMP deps

Manual / cluster validation

count: N failover and recovery in kind
workers.list degraded recovery in kind
K8s service creation/update/delete in standalone and manager modes

Risks

count: N and workers.list share code but not invariants
SNMP migration may affect CLI/API MIB tooling as much as connector runtime
K8s provisioning can sprawl if service type and lifecycle semantics are left vague
filename-variable work can drift if implemented per sink instead of in the shared helper and partition model

Review Questions

These are the points to confirm before implementation:

Is the separation between count: N and workers.list invariants correct?
Is the filename-variable model correctly framed as scoped variables plus partitioned file output, rather than a narrow source_stem / source_suffix token addition?
Should K8s service provisioning explicitly support both standalone and manager, with worker mode excluded?
Is NodePort-only the right boundary for the first K8s provisioning slice?
Should SNMP migration remain strictly separate from UDP broadcast enablement?
Is executor-side partitioning before serialization acceptable as the fixed v1.3.1 direction, with sink-side partitioning explicitly deferred?

v1.3.1 Design Plan: Placement Completion & Targeted Backend Follow-ups

Goal

Design Decisions

1. Filename rendering needs scoped variables, not ad-hoc metadata tokens

2. Dynamic K8s service provisioning should support both manager and standalone

3. workers.count: N and workers.list are different invariants

Current Code Baseline

Workstream A: workers.count: N

Objective

Runtime Model

Invariants

Code Changes

Tests

Acceptance

Workstream B: workers.list: [...]

Objective

Runtime Model

Invariants

Recovery rule

Code Changes

Tests

Acceptance

Workstream C: Filename variables, classified routing, and dynamic partitioning

Objective

Intended sinks

Variable scopes

Slice C1: Runtime/source-context naming plus explicit classified routing

Source-derived behavior

Slice C2: Dynamic record-field partitioning

Record-field behavior

Validation / lint behavior

Design implications

Chosen architectural direction

Implementation status gate

Implementation

Tests

Acceptance

Workstream D: SNMP migration to pysnmp 7.x

Objective

Areas affected

Implementation plan

Important rule

Acceptance

Workstream E: Dynamic K8s Service Provisioning

Objective

Ownership model

Source scope

Mode-aware target model

Recommended initial feature boundary

Required design points

Implementation areas

Lifecycle rules

Tests

Acceptance

Recommended PR Split

Test Matrix

Unit

Integration

Manual / cluster validation

Risks

Review Questions

3. `workers.count: N` and `workers.list` are different invariants

Workstream A: `workers.count: N`

Workstream B: `workers.list: [...]`

Workstream D: SNMP migration to `pysnmp 7.x`