rfgen.log_sinks

Structured log destinations for the framework. Every framework log line is a LogRecord instance handed to a BaseLogSink implementation, which delivers the record to its final home (stdout, a local file, or a cloud log service). Cloud sinks live behind their cloud extras and lazy-import their SDKs.

For nearly all users StdoutSink is the correct choice. Cloud Logging on GCP, CloudWatch Logs on AWS, and Application Insights on Azure all auto-ingest structured JSON from stdout when running on their managed compute. Direct-push sinks (CloudLoggingSink, CloudWatchSink, AzureMonitorSink) exist for users running outside the managed-compute path or for those who want to bypass the log agent entirely.

Warning

Pre-implementation. The cloud sinks listed below are stubs. Class signatures and parameter lists for the stub classes are proposals; once code lands, this page will be regenerated from docstrings via sphinx.ext.autodoc. The Layer 2 core surface (BaseLogSink, StdoutSink, FileSink, LogRouter, LogRecord, the config models, and the drop-counter getter) is implemented; once docstrings are wired into the build, those sections will be regenerated from source.

Module summary

from rfgen.log_sinks import (
    LogRecord,
    StdoutSink,
    FileSink,
    LogRouter,
    get_dropped_records_total,
)

stdout_sink = StdoutSink()
file_sink = FileSink(path="/var/log/rfgen/run.jsonl")

router = LogRouter(
    sinks=[stdout_sink, file_sink],
    min_level="INFO",
    per_sink_min_level={"file": "DEBUG"},
)

record = LogRecord.now(
    level="INFO",
    event="shard.complete",
    attributes={"samples": 1000, "elapsed_ms": 47230},
    run_id="5e2c1f...",
    shard_id="abc123",
)
router.emit(record)
router.flush()

dropped = get_dropped_records_total()  # monotonically non-decreasing int

Class index

Class	Kind	Notes
BaseLogSink	ABC	Subclass to add a custom sink; override emit_once
LogRecord	frozen dataclass	The structured record handed to every sink
LogRouter	concrete (@final)	Fans out one record to every configured sink by level
StdoutSink	concrete (core)	Default. Writes JSON lines to sys.stdout
FileSink	concrete (core)	Append-mode local file
BaseLogSinkConfig	Pydantic v2 model	Base config block; one entry per configured sink
StdoutSinkConfig	Pydantic v2 model	Config for StdoutSink
FileSinkConfig	Pydantic v2 model	Config for FileSink
CloudLoggingSink	concrete (rfgen[gcp])	GCP Cloud Logging direct push. (stub)
CloudWatchSink	concrete (rfgen[aws])	AWS CloudWatch Logs direct push. (stub)
AzureMonitorSink	concrete (rfgen[azure])	Azure Monitor direct push. (stub)

For the conceptual framing and the structured-field contract every record carries, see Reference / Logging & Observability.

---

Abstract base class

class rfgen.log_sinks.BaseLogSink

class BaseLogSink(ABC):
    sink_name: ClassVar[str]

    def __init_subclass__(cls, **kwargs) -> None: ...

    @abstractmethod
    def emit_once(self, record: LogRecord) -> None: ...

    def emit(self, record: LogRecord) -> None: ...   # concrete; drives retry

    def flush(self) -> None: ...                     # concrete no-op default

The framework drives delivery through emit, which is the concrete retry-with-jitter driver: it wraps emit_once in a bounded loop, retries with full-jitter exponential backoff, and after the retry budget is exhausted drops the record and increments the module-level drop counter. The framework never crashes because a log sink failed.

emit_once is the public extension hook subclasses override. It must raise on any delivery failure; the retry loop in emit either retries or drops.

The sink_name ClassVar[str] is enforced at class-definition time via __init_subclass__: a concrete subclass that omits the attribute (or sets it to an empty string or non-string value) raises TypeError when the class body executes.

Retry policy

Pinned constants (do not hand-tune without a decision-log entry):

Constant	Value
max_retries	5
base_delay_s	0.1
cap_s	10.0

The schedule is full-jitter exponential backoff per the AWS “Exponential Backoff and Jitter” recommendation:

delay = min(cap_s, base_delay_s * 2**attempt) * random.uniform(0, 1)

After max_retries consecutive failures on a single record the record is dropped, the module-level integer _dropped_records_total is incremented, and the optional Prometheus mirror (when installed) is incremented to match. emit never raises.

Drop counter

Symbol	Type	Description
_dropped_records_total	module-level int	Canonical counter; monotonically non-decreasing; never reset by the framework
get_dropped_records_total() -> int	function	Public getter; the only supported way to read the counter
rfgen_log_sinks_dropped_records_total	prometheus_client.Counter	Optional telemetry mirror, gated behind rfgen[prometheus]; readers should treat the integer as canonical

A contract test forces N drops, asserts get_dropped_records_total() increments by exactly N, and (when the extra is installed) asserts the Prometheus counter agrees.

Method: emit_once (abstract)

@abstractmethod
def emit_once(self, record: LogRecord) -> None

Deliver one record. Subclasses MUST override. Any exception raised here is caught by the retry loop in emit.

Method: emit (concrete)

def emit(self, record: LogRecord) -> None

Concrete on the ABC. Drives the retry-with-jitter loop around emit_once. Subclasses MUST NOT override.

Method: flush

def flush(self) -> None

Concrete no-op default. Subclasses that buffer override; idempotent.

Extension contract

Every BaseLogSink subclass MUST satisfy:

1. emit_once raises on failure. Do not swallow delivery errors silently; the retry loop in emit is the single place that decides retry vs. drop.

2. sink_name is a ClassVar[str] with a non-empty value. Enforced by __init_subclass__; appears in manifest.json["log_sinks"] and in registry keys, so it MUST be stable across process restarts of the same configuration.

3. emit_once is not guaranteed thread-safe by default. The framework serialises calls through LogRouter; sinks shared across threads must add their own locking.

4. flush is idempotent. Calling it when nothing is buffered is a no-op.

5. No credential storage. Sinks that require credentials MUST obtain them via BaseCredentialsProvider, not by reading environment variables directly.

Extension points

Symbol	Status	Notes
emit_once(record)	Must override	Raw delivery; raise on failure
flush()	May override	No-op default; override if the sink buffers
sink_name	Must declare	ClassVar[str]; enforced by __init_subclass__
emit(record)	Do not override	Concrete retry-with-jitter driver

---

class rfgen.log_sinks.LogRecord

@dataclass(frozen=True, slots=True)
class LogRecord:
    timestamp: str
    level: str
    event: str
    attributes: dict[str, object] = field(default_factory=dict)
    run_id: str = ""
    shard_id: str | None = None

Frozen dataclass passed to every BaseLogSink.emit_once call.

Fields

Name	Type	Required	Default	Description
timestamp	str	yes	-	ISO-8601 UTC timestamp of the event
level	str	yes	-	One of "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"
event	str	yes	-	Short event identifier (e.g. "shard.complete")
attributes	dict[str, object]	no	{}	Free-form structured attachments; must be JSON-serialisable. Non-serialisable values are coerced via json.dumps(default=str) by the core sinks
run_id	str	no	""	Stable run identifier; correlates events across shards and processes within one run
shard_id	str | None	no	None	Stable shard identifier; correlates events across transformations within one shard. None for events not tied to a specific shard

The run_id and shard_id defaults exist so downstream log aggregators can correlate per-shard events without every call site having to thread the IDs.

Class method: now

@classmethod
def now(
    cls,
    *,
    level: str,
    event: str,
    attributes: Mapping[str, object] | None = None,
    run_id: str = "",
    shard_id: str | None = None,
) -> LogRecord

Construct a record with a fresh UTC timestamp. The keyword-only signature matches the field order above.

---

class rfgen.log_sinks.LogRouter

@final
class LogRouter:
    def __init__(
        self,
        sinks: Iterable[BaseLogSink],
        *,
        min_level: str = "INFO",
        per_sink_min_level: Mapping[str, str] | None = None,
    ) -> None: ...

    def emit(self, record: LogRecord) -> None: ...
    def flush(self) -> None: ...

Fans one LogRecord out to every configured sink whose minimum-level threshold passes.

The class is decorated @typing.final: subclassing is not a supported extension point. Callers extend the system by writing new BaseLogSink subclasses, not by subclassing the router.

Constructor parameters

Name	Type	Required	Default	Description
sinks	Iterable[BaseLogSink]	yes	-	Sinks in fan-out order; preserved when records are emitted
min_level	str	no	"INFO"	Default minimum level applied to sinks not listed in per_sink_min_level
per_sink_min_level	Mapping[str, str] | None	no	None	Optional per-sink_name override of the minimum level

__init__ raises ValueError if min_level or any per-sink override is not one of "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL".

Method: emit

Fans the record out to every sink whose threshold passes. Sink failures are absorbed by the per-sink retry loop in BaseLogSink.emit; the router never raises and never lets one slow sink block the others. Concurrent fan-out is intentionally out of scope at Layer 2: the orchestrator runs sinks serially, but call order is deterministic.

Method: flush

Flushes every sink in registration order. Per-sink flush failures are absorbed (non-fatal).

---

class rfgen.log_sinks.StdoutSink

class StdoutSink(BaseLogSink):
    sink_name: ClassVar[str] = "stdout"

    def __init__(
        self,
        config: StdoutSinkConfig | None = None,
        *,
        stream: IO[str] | None = None,
        flush_each: bool | None = None,
    ) -> None: ...

    def emit_once(self, record: LogRecord) -> None: ...
    def flush(self) -> None: ...

The default sink. Each call to emit_once writes one JSON object followed by a newline, so the stream is parseable line-by-line by Cloud Logging, CloudWatch Logs agent, Fluent Bit, jq, and any other JSON-lines-aware consumer.

Constructor parameters

Name	Type	Required	Default	Description
config	StdoutSinkConfig | None	no	None	Validated config block. Use this on the config-driven path
stream	IO[str] | None	no	None	Destination stream. None means sys.stdout is resolved at emit time, so reassignments (e.g. by pytest’s capture) are honoured
flush_each	bool | None	no	None	When True, the underlying stream is flushed after every record. When None and config is provided, the config value is used; otherwise the default is True

The constructor accepts either a validated StdoutSinkConfig instance or raw kwargs. When both are supplied, the kwargs win so in-process callers (notably tests) can target a specific stream.

Notes

• Default sink. The Layer 9 orchestrator installs a StdoutSink automatically unless overridden by config.

• Cloud-agent ingestion. Spark serverless and Vertex Workbench ship with a Cloud Logging agent that scrapes stdout and parses JSON payloads into indexed labels. No extra wiring is required.

• Secret redaction runs inside JsonFormatter before the record reaches any sink, so StdoutSink never sees raw secrets even if a logger is misused. See Logging & Observability § What does NOT get logged.

• Determinism. The sink does not add fields of its own; everything in the emitted line came from the formatter, so two runs with identical input produce byte-identical log lines (modulo timestamp).

---

class rfgen.log_sinks.FileSink

class FileSink(BaseLogSink):
    sink_name: ClassVar[str] = "file"

    def __init__(
        self,
        config: FileSinkConfig | None = None,
        *,
        path: str | Path | None = None,
        encoding: str | None = None,
    ) -> None: ...

    def emit_once(self, record: LogRecord) -> None: ...
    def flush(self) -> None: ...
    def close(self) -> None: ...

Writes JSON-line records to a file on local disk. The file is opened in append mode at construction time and held open for the lifetime of the sink.

Constructor parameters

Name	Type	Required	Default	Description
config	FileSinkConfig | None	no	None	Validated config block. When supplied without path, config.path is used
path	str | Path | None	no	None	Destination file path. Parent directory must exist. Exactly one of config.path or path must resolve to a non-None value
encoding	str | None	no	None	Text encoding. Falls back to config.encoding when set, otherwise "utf-8"

__init__ raises TypeError when neither config nor path is provided.

Rotation is intentionally out of scope at Layer 2; callers that need rotation install Python’s logging.handlers.RotatingFileHandler upstream.

Method: close

Closes the underlying file handle. Idempotent; safe to call multiple times.

---

Config models

class rfgen.log_sinks.BaseLogSinkConfig

class BaseLogSinkConfig(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    sink_name: str = Field(min_length=1)
    params: dict[str, object] = Field(default_factory=dict)

Pydantic v2 BaseModel describing one configured sink. Concrete sinks subclass this with their own option fields; the base keeps sink_name plus a free-form params mapping so YAML configs can declare sink-specific options without changing the base schema.

Field	Type	Required	Default	Description
sink_name	str (min length 1)	yes	-	Stable identifier matching the concrete sink’s sink_name ClassVar
params	dict[str, object]	no	{}	Free-form sink-specific options forwarded to the concrete sink constructor by the Layer 9 orchestrator

class rfgen.log_sinks.StdoutSinkConfig

class StdoutSinkConfig(BaseLogSinkConfig):
    sink_name: str = "stdout"
    flush_each: bool = True

Field	Type	Required	Default	Description
sink_name	str	no	"stdout"	Pinned to "stdout"
flush_each	bool	no	True	Flush the stream after every record

class rfgen.log_sinks.FileSinkConfig

class FileSinkConfig(BaseLogSinkConfig):
    sink_name: str = "file"
    path: str = Field(min_length=1)
    encoding: str = "utf-8"

Field	Type	Required	Default	Description
sink_name	str	no	"file"	Pinned to "file"
path	str (min length 1)	yes	-	Destination file path. Parent directory must exist
encoding	str	no	"utf-8"	Text encoding

---

Cloud sinks (stubs)

The cloud sinks below ship behind their cloud extras and lazy-import their SDKs. They subclass BaseLogSink and override emit_once; the retry-with-jitter loop in emit is inherited.

class rfgen.log_sinks.CloudLoggingSink

class CloudLoggingSink(BaseLogSink):
    sink_name: ClassVar[str] = "cloud_logging"

Kind: concrete, requires rfgen[gcp]. (stub)

Pushes records directly to GCP Cloud Logging via google-cloud-logging, bypassing the stdout-scraping path. Useful when running outside Spark serverless / Vertex Workbench (e.g. on bare GCE) where no log agent is present. On Spark serverless and Vertex Workbench prefer StdoutSink.

class rfgen.log_sinks.CloudWatchSink

class CloudWatchSink(BaseLogSink):
    sink_name: ClassVar[str] = "cloudwatch"

Kind: concrete, requires rfgen[aws]. (stub)

Pushes records directly to AWS CloudWatch Logs via boto3. Used when running on AWS compute (EMR, Batch, ECS) outside the managed log-agent path, or when log group / stream names need to be controlled from the framework rather than from the host.

class rfgen.log_sinks.AzureMonitorSink

class AzureMonitorSink(BaseLogSink):
    sink_name: ClassVar[str] = "azure_monitor"

Kind: concrete, requires rfgen[azure]. (stub)

Pushes records directly to Azure Monitor / Application Insights via the opencensus-ext-azure exporter. Used when running on Azure compute (Synapse, AML) outside the managed log-collection path.

---

Stub class anchors

The formatter referenced in the prose above (JsonFormatter) does not yet have a full class signature block. Its proposed surface lives in Reference / Logging & Observability § Logging strategy (the canonical pseudocode); the anchor label below gives the stub a stable cross-reference target so the operations / observability page and downstream how-to guides can use {ref} links per the STYLE.md code-span linking rule. The anchor resolves to this section; once a full class block lands, the anchor moves next to the corresponding class rfgen.log_sinks.JsonFormatter heading without changing inbound links.

Per-stub note:

• JsonFormatter: subclass of logging.Formatter that converts a stdlib logging.LogRecord to a LogRecord instance and runs the secret-redaction hook before the record reaches any BaseLogSink. Installed by the Layer 9 orchestrator’s setup_logging(). The framework relies on the redaction hook firing here (not in the sink) so all sinks see pre-redacted records uniformly. See Logging & Observability § What does NOT get logged for the redaction policy.

---

See Also

• Reference / Logging & Observability: strategy, log levels, structured fields, retry policy.

• BaseLogSink: ABC contract and the retry-with-jitter driver.

• rfgen.errors: error classes that drive ERROR-level log lines.

• Reference / Plugin Metadata: entry-point group rfgen.log_sinks.

• BaseCredentialsProvider: how cloud sinks must obtain credentials.