Observability

craft-easy-file-import instruments the full file import pipeline with OpenTelemetry distributed tracing and Prometheus metrics. Both are opt-in at startup.

---

OpenTelemetry tracing

Initialization

Call init_telemetry() once at application startup. It configures a TracerProvider with a BatchSpanProcessor and an OTLP gRPC exporter.

from craft_easy_file_import.telemetry import init_telemetry

init_telemetry(
    service_name="craft-easy-file-import",  # default
    otlp_endpoint="http://otel-collector:4317",  # default: localhost:4317
)

init_telemetry() is idempotent — safe to call multiple times; subsequent calls are no-ops.

If opentelemetry-exporter-otlp-proto-grpc is not installed, initialization is silently skipped and all spans become no-ops. This means tracing is completely optional: the code path is identical with or without the exporter installed.

pip install opentelemetry-exporter-otlp-proto-grpc

Spans

The ImportRunner._process_file() method creates the following spans for each file:

Span name	Attributes
file_import.process_file	file.name, file.path, file.size_bytes, template.id, template.name, source.type, source.name
pre_flight.filter	—
pre_flight.encoding	sets file.encoding on parent
pre_flight.hash	sets file.hash (first 16 chars) on parent
run_creation	sets run.id on parent
pre_flight.signature	—
transforms.crypto	—
parse	file.format
commit	—

Result attributes on file_import.process_file after processing:

• result.status — success, completed_with_errors, or failed
• result.total_records — total rows in the file
• result.error_count — number of rejected rows
• offload.queued — true if the file was handed off to craft-easy-jobs
• file.skipped + file.skip_reason — when the file is skipped before processing

Trace context propagation

When a file is offloaded to craft-easy-jobs for async processing, the current trace context is propagated as traceparent metadata on the queued job so the worker can continue the same trace:

from craft_easy_file_import.telemetry import inject_trace_context, restore_trace_context

# In the runner (producer side) — injected automatically:
carrier = inject_trace_context()  # {"traceparent": "00-abc123...-xyz-01"}

# In the job worker (consumer side):
token = restore_trace_context(carrier)
try:
    # ... process file
finally:
    from opentelemetry.context import detach
    detach(token)

Accessing the tracer

The module-level tracer is available for custom instrumentation:

from craft_easy_file_import.telemetry import tracer

with tracer.start_as_current_span("my_custom_step") as span:
    span.set_attribute("rows_processed", 1234)
    # ...

Getting the current trace ID

from craft_easy_file_import.telemetry import get_current_trace_id

trace_id = get_current_trace_id()  # "4bf92f3577b34da6a3ce929d0e0e4736" or None

---

Prometheus metrics

All metrics use the cef_ prefix (Craft Easy File-import). Labels are kept low-cardinality: source_type (e.g. s3, azure_blob, sftp) and template_category (the template name).

Counters

Metric	Labels	Description
cef_files_processed_total	source_type, template_category, result	Total files processed. result: success, completed_with_errors, failed
cef_rows_processed_total	source_type, template_category, result	Total rows across all processed files
cef_rejected_records_total	source_type, template_category, layer	Rejected records by validation layer (schema, row, business, plugin, cross_row)

Histograms

Metric	Labels	Buckets	Description
cef_file_processing_duration_seconds	source_type, template_category	1, 5, 15, 30, 60, 300, 600, 1800, 3600 s	End-to-end file processing time, including transforms and commits
cef_row_processing_duration_seconds	stage	1, 5, 10, 50, 100, 500 ms, 1 s	Per-row processing time by pipeline stage

Gauges

Metric	Labels	Description
cef_queue_depth	priority	Current job queue depth for offloaded imports
cef_active_runs	source_type	Number of imports currently in progress
cef_sla_violations_open	severity	Open SLA violation count by severity

Exposing the /metrics endpoint

The metrics module uses prometheus_client — expose the endpoint through your application or a standalone HTTP server:

from prometheus_client import start_http_server

start_http_server(9090)  # exposes /metrics on port 9090

Or in a FastAPI application:

from prometheus_client import make_asgi_app

metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)

Importing metrics for custom instrumentation

from craft_easy_file_import.telemetry.metrics import (
    files_processed_total,
    rows_processed_total,
    rejected_records_total,
    file_processing_duration_seconds,
    row_processing_duration_seconds,
    queue_depth,
    active_runs,
    sla_violations_open,
)

# Example: increment the rejected records counter from a custom pipeline
rejected_records_total.labels(
    source_type="sftp",
    template_category="supplier-payments",
    layer="row",
).inc()

---

Recommended alerts

Alert	Expression	Threshold
High file failure rate	rate(cef_files_processed_total{result="failed"}[5m]) / rate(cef_files_processed_total[5m])	> 0.1
High rejection rate	rate(cef_rejected_records_total[5m]) / rate(cef_rows_processed_total[5m])	> 0.05
Long file processing	histogram_quantile(0.95, cef_file_processing_duration_seconds)	> 600 s
Queue buildup	cef_queue_depth	> 50
Open SLA violations	cef_sla_violations_open{severity="high"}	> 0