Azure Bootstrap — v2 Tier 1 Primitives & Logging Transports (v2.1 + v3.0)¶

3. Python: v2 Tier 1 primitives (always-on, stdlib-only)¶

Everything here is importable from the top-level azure_bootstrap namespace (or its subpackage) with no extra installed.

Structured logging¶

from azure_bootstrap import configure_logging, JsonLogFormatter
from azure_bootstrap.logging import (
    ExtraFieldsFormatter, effective_log_level, env_flag, debug_logging_enabled,
)

configure_logging(
    format_string="%(asctime)s %(levelname)s %(name)s %(message)s",  # default
    silence_defaults=True,        # silence noisy third-party loggers (urllib3, azure.*, …)
    extra_noisy_loggers=(),       # add your own to silence
)

configure_logging() is idempotent (it force-replaces handlers), installs the ExtraFieldsFormatter + a CorrelationFilter, and sets the root level via effective_log_level(). The level honors LOG_LEVEL, but DEBUG requires a second gate: DEBUG_LOGGING_ENABLED must also be truthy, else it clamps to INFO (defends against a stray manifest leaking DEBUG into prod). In DEBUG, an extra={} key that collides with a reserved LogRecord attribute raises LoggingExtraConflictError.

JsonLogFormatter(*, ensure_ascii=False, mask_extras=True) — one JSON object per line. Fields: timestamp (ISO-8601 UTC), level, logger, message, exception (only with exc_info), plus every non-reserved extra={} field (correlation IDs included). Secret-keyed extras are redacted via mask_secrets_in_dict. Never raises — falls back to a minimal document. Use it for remote ingestion (this is what the Sumo Logic transport uses).

Correlation context¶

from azure_bootstrap import correlation_scope, get_correlation_id, set_correlation_id
from azure_bootstrap.logging import CorrelationFilter

with correlation_scope("req-123", user_id="u-456", email_id="e-789") as cid:
    logger.info("processing")   # log line auto-includes correlation_id, user_id, email_id

correlation_scope(correlation_id=None, **fields) pushes context for the with block (generates a 12-char hex id when None), and CorrelationFilter attaches every set context var as a record attribute. Built on contextvars, so it is async- and task-safe. get_correlation_id() / set_correlation_id(value) read/write outside a scope.

Masking & sanitization¶

from azure_bootstrap import (
    mask_api_key, mask_bearer_token, mask_email_address,
    mask_secrets_in_dict, safe_json_dumps, sanitize_for_log,
)
from azure_bootstrap.logging import register_secret_keys, content_preview

Function	Behavior
`mask_api_key(s)`	`*` if `None`/<4 chars, else `*{last4}`
`mask_bearer_token(t)`	`Bearer *` if it starts with `Bearer`, else `*`
`mask_email_address(e)`	`***{last2-local}@{domain}`
`mask_secrets_in_dict(d)`	shallow copy; redacts truthy values at ~20 secret-keyed names (`authorization`, `x-api-key`, `api_key`, `password`, `token`, `client_secret`, `connection_string`, …)
`register_secret_keys(*names)`	extend the secret-key allowlist at runtime
`sanitize_for_log(v, max_len=256)`	strip control chars → `?`, truncate
`safe_json_dumps(obj)`	JSON with `default=repr`; never raises
`content_preview(text, max_len=500)`	truncate a body for preview

Tracing & counters¶

from azure_bootstrap import traced, latency_snapshot, bump_counter, counter_snapshot

@traced(operation="reports.process", alert_on_error="error",
        sensitive_args=("api_key",), slow_threshold_seconds=2.0)
def process(report_id: str, api_key: str) -> dict: ...

@traced works on sync or async functions (auto-detected). It records latency on every call (success and exception), logs entry/exit only at DEBUG (the hot path skips inspect.signature when DEBUG is off), masks sensitive_args (and anything that "looks" sensitive) in logs, and — if the alerts subpackage is importable — fires a slow-budget WARN and/or an alert_on_error alert. With no alerts extra it silently degrades to log-only. latency_snapshot() returns {operation: {count, errors, slow, p50, p95, p99, max, last_seen}}. bump_counter(name, n=1) is thread-safe and never raises; counter_snapshot() returns a copy.

Bootstrap helpers¶

from azure_bootstrap import (
    ensure_bootstrap, bootstrap_initialized, load_local_settings, refresh_setting,
)

Function	Behavior
`ensure_bootstrap()`	lazy, idempotent wrapper over `initialize_application()`; short-circuits when `USE_MOCK_BOOTSTRAP` is truthy; re-raises on failure after logging
`bootstrap_initialized() -> bool`	process-local flag — wire into `/health/ready`
`load_local_settings(path="local.settings.json") -> int`	load Azure-Functions-style settings; skips `_`-prefixed keys; never overwrites existing env; returns count
`refresh_setting(*names)`	re-read named keys from the cached App Config repo into `os.environ`; best-effort, no-op before `initialize_application()`

Error vocabulary¶

from azure_bootstrap import (
    PipelineError, UnrecoverableError, TransientError,
    InvalidMessageError, RateLimitError, NetworkError, is_unrecoverable,
)

PipelineError is the base. UnrecoverableError (→ InvalidMessageError, OversizedAttachmentError, MalformedAttachmentError, ZipBombError, UpstreamResourceMissing) means "dead-letter it." TransientError (→ RateLimitError, NetworkError, AuthenticationError) means "retry/back off." is_unrecoverable(exc) is a single classifier the retry/soft-fail/consumer helpers all consult.

Resilience — soft-fail & phases¶

from azure_bootstrap import (
    soft_fail, soft_fail_with, SoftFailResult, run_phase, run_phases, PhaseResult,
)

# Degrade gracefully with a fallback value:
res = soft_fail_with(fetch_thumbnail, blob_id, fallback=None,
                     operation="thumbnail.fetch", counter_name="thumbnail.failed")
if res.degraded:
    logger.warning("thumbnail unavailable", extra={"reason": res.reason})

# Context-manager form:
with soft_fail(operation="enrich") as state:
    record["extra"] = enrich(record)
if state["degraded"]:
    ...

# Sequential pipeline that never aborts mid-way:
results = run_phases([("download", download), ("parse", parse), ("index", index)])

soft_fail_with(...) re-raises UnrecoverableError by default (re_raise_unrecoverable=True) — set False to swallow everything in catch. run_phase/run_phases never re-raise; each bumps {namespace}.{name}.ok / .failed counters and returns PhaseResult(name, ok, value, exception, elapsed_seconds).

Validation¶

from azure_bootstrap import validate_message, queue_message_schema, MessageSchema

schema = queue_message_schema(
    required_fields=("correlation_id",),
    path_field="blob_path", path_required_prefix="reports/",
)
data = validate_message(payload, schema)   # raises InvalidMessageError on violation

Building a schema with a path_field automatically adds path-traversal defense (forbidden substrings .. and ://). On failure validate_message bumps a counter and raises InvalidMessageError (an UnrecoverableError — so consumers dead-letter).

Path safety¶

from azure_bootstrap import sanitize_path_segment, confine_to_root

safe = sanitize_path_segment(user_filename)            # strips bidi/zero-width, caps 64 chars
path = confine_to_root(raw, allowed_root="/data/work") # raises ValueError on escape

confine_to_root canonicalizes both sides (expanduser().resolve()) before comparison, defeating .. traversal and symlink escape.

Security & fail-close¶

from azure_bootstrap import compare_secrets
from azure_bootstrap.failclose import require_env, optional_env, fail_open_env

compare_secrets(a, b) -> bool — constant-time (hmac.compare_digest); False on any None/empty input.
require_env(name, message=None) — return value or raise ConfigurationError (auth-critical settings → fail closed).
optional_env(name, default="") — stripped value or default (URLs with sane defaults).
fail_open_env(name) — value when truthy, else None ("feature disabled" semantics).

Identity & audit¶

from azure_bootstrap.identity import build_credential, credential_kind, CredentialKind
from azure_bootstrap.audit import build_audit_extra

cred = build_credential()   # ClientSecret (if secret set) → WorkloadIdentity → DefaultAzureCredential
logger.info("EMAIL_AUDIT", extra=build_audit_extra("send", sender=addr, subject=subj))

build_credential(*, tenant_id=None, client_id=None, client_secret=None, prefer=None, token_file_path=...) codifies the credential preference (Workload Identity first in a cluster — no client secrets in pod env). credential_kind() previews the choice without constructing anything (handy for health probes). build_audit_extra(operation, **fields) always injects operation + a UTC ISO-8601 timestamp, masks email/secret fields, and truncates long fields (subject, error, traceback, …).

Tier 1 environment variables¶

Variable	Default	Purpose
`LOG_LEVEL`	`INFO`	base log level
`DEBUG_LOGGING_ENABLED`	off	second gate required for DEBUG output
`USE_MOCK_BOOTSTRAP`	off	short-circuit `ensure_bootstrap()` (local/dev/tests)

4. Python: logging transports (v2.1 + v3.0 — ten sinks)¶

A transport is a named factory Callable[[], logging.Handler | None]. Enabling attaches its handler to the root logger; disabling detaches and closes it. This decouples where logs go from how they're formatted.

from azure_bootstrap import (
    configure_transports, register_transport,
    enable_transport, disable_transport, list_transports,
)

# One call to wire built-ins (explicit bool wins; else env flag):
configure_transports(console=True, app_insights=False, sumo_logic=True)

# v3.0 — enable additional sinks in one call:
configure_transports(console=True, panther=True, file=True, blob=True, sql=True,
                     nosql=True, adx=True, event_hubs=True)

list_transports()   # {'console': {'registered': True, 'enabled': True}, ...}

configure_transports() also sets the root logger to effective_log_level() so enabled transports actually receive records. It is idempotent and re-runnable.

Built-in transports (ten total)¶

Name	Handler	Extra	Env flag	Default
`console`	`StreamHandler` + `ExtraFieldsFormatter` + `CorrelationFilter`	stdlib	`CONSOLE_LOGGING_ENABLED`	on
`app_insights`	OpenTelemetry handler (delegates to v1 `TelemetryManager`)	stdlib	`APP_INSIGHTS_LOGGING_ENABLED`	off
`sumo_logic`	`SumoLogicHandler` (buffered async POST)	`[sumologic]`	`SUMO_LOGIC_LOGGING_ENABLED`	off
`panther`	Panther SIEM HTTP Source	`[panther]`	`PANTHER_LOGGING_ENABLED`	off
`file`	Rotating local log file (`JsonLogFormatter`)	stdlib	`FILE_LOGGING_ENABLED`	off
`blob`	Azure Blob Storage NDJSON	`[bloblog]`	`BLOB_LOGGING_ENABLED`	off
`sql`	SQL table append	`[sqllog]`	`SQL_LOGGING_ENABLED`	off
`nosql`	Mongo/Cosmos collection append	`[nosqllog]`	`NOSQL_LOGGING_ENABLED`	off
`adx`	Azure Data Explorer ingest	`[adxlog]`	`ADX_LOGGING_ENABLED`	off
`event_hubs`	Azure Event Hubs producer	`[eventhubslog]`	`EVENTHUBS_LOGGING_ENABLED`	off

Two caveats: (1) the console transport installs the same StreamHandler stack as configure_logging() — enabling both produces duplicate console lines, so pick one; (2) if you also call configure_logging() (which does basicConfig(force=True) and replaces root handlers), call it before configure_transports(), which reconciles against the live root handlers on each run. Disabling app_insights only detaches the OTel handler (the exporter is not torn down). A custom sink is one call away:

register_transport("my_syslog", lambda: logging.handlers.SysLogHandler())
enable_transport("my_syslog")

`SumoLogicHandler` deep-dive¶

Requires the [sumologic] extra (requests). make_sumo_logic_handler() returns None (a soft no-op) when SUMO_LOGIC_COLLECTOR_URL is unset or when requests isn't installed — so enabling the transport without the extra never errors.

Behavior: emit() only appends to an in-memory bounded deque (a daemon thread does the network I/O — never blocks, never raises). It ships NDJSON (via JsonLogFormatter), gzips bodies at/above the threshold, batches by count and byte size (Sumo's 100 KB–1 MB sweet spot), and uses a urllib3 Retry adapter that retries 408/429/5xx with backoff+jitter, honors Retry-After, and never retries 401/other 4xx. Flushes on interval, on batch_size, and at atexit. Counters: sumologic.transport.{posts,ok,error,throttled,dropped,records}.

import os
os.environ["SUMO_LOGIC_COLLECTOR_URL"] = "https://collectors.sumologic.com/receiver/v1/http/XXXX"
os.environ["SUMO_LOGIC_LOGGING_ENABLED"] = "true"
configure_transports(sumo_logic=True)   # or rely on the env flag alone

Env var	Default	Purpose
`SUMO_LOGIC_COLLECTOR_URL`	— (required)	HTTP Source endpoint (unset → transport stays off)
`SUMO_LOGIC_COLLECTOR_TOKEN`	—	`x-sumo-token` auth header
`SUMO_LOGIC_SOURCE_CATEGORY`	—	`X-Sumo-Category`
`SUMO_LOGIC_SOURCE_HOST`	—	`X-Sumo-Host`
`SUMO_LOGIC_FIELDS`	—	`X-Sumo-Fields` (`k=v,k2=v2`)
`SUMO_LOGIC_BATCH_SIZE`	`100`	records per POST
`SUMO_LOGIC_MAX_BATCH_BYTES`	`1000000`	byte cap per POST
`SUMO_LOGIC_GZIP_THRESHOLD`	`1024`	gzip bodies ≥ this
`SUMO_LOGIC_FLUSH_INTERVAL`	`5.0`	timer flush (s)
`SUMO_LOGIC_MAX_BUFFER`	`10000`	buffer cap (oldest dropped on overflow)
`SUMO_LOGIC_TIMEOUT`	`5.0`	POST timeout (s)

v3.0.0 transports — shared `_BufferedShipper` contract¶

All v3 network/storage transports subclass _BufferedShipper — the same guarantees as Sumo Logic: never block the caller, never raise, bounded buffer with drop counting, background flush thread, batch by count and bytes, flush at atexit. Factories return None (soft no-op) when required env vars are unset or the pip extra is missing.

Name	Key env vars (factory soft no-op if unset)
`panther`	`PANTHER_API_HOST`, `PANTHER_LOG_SOURCE_ID` / `PANTHER_LOG_SOURCE_TOKEN` (or `PANTHER_API_KEY`)
`file`	`FILE_LOG_PATH` (+ optional `FILE_LOG_ROOT`, `FILE_LOG_ROTATION`, `FILE_LOG_MAX_BYTES`, …)
`blob`	`BLOB_*` connection/container settings
`sql`	`SQL_LOG_DSN`, `SQL_LOG_TABLE`
`nosql`	`NOSQL_LOG_URI`, `NOSQL_LOG_DATABASE`, `NOSQL_LOG_COLLECTION`
`adx`	`ADX_CLUSTER_URI`, `ADX_DATABASE`
`event_hubs`	`EVENTHUB_FQNS`, `EVENTHUB_NAME`

Install all transport deps with pip install 'azure-bootstrap[logging-all]'. See examples/39_v3_transports.py.

Plugin: azure-bootstrap · View SKILL.md on GitHub