Skip to content

chore!: Refactor OTel exporters to be signal-specific #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morgan-wowk merged 1 commit into from
Mar 5, 2026
Merged

chore!: Refactor OTel exporters to be signal-specific #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 42 additions & 25 deletions cloud_pipelines_backend/instrumentation/opentelemetry/_internal/configuration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,53 +15,70 @@ class ExporterProtocol(str, enum.Enum):


@dataclasses.dataclass(frozen=True, kw_only=True)
class OtelConfig:
class ExporterConfig:
endpoint: str
protocol: str


@dataclasses.dataclass(frozen=True, kw_only=True)
class OtelConfig:
service_name: str
service_version: str
trace_exporter: ExporterConfig | None = None


def _resolve_exporter(
endpoint_var: str,
protocol_var: str,
) -> ExporterConfig | None:
endpoint = os.environ.get(endpoint_var)
if not endpoint:
return None

protocol = os.environ.get(protocol_var, ExporterProtocol.GRPC)

if not endpoint.startswith(("http://", "https://")):
raise ValueError(
f"Invalid OTel endpoint format: {endpoint}. "
f"Expected format: http://<host>:<port> or https://<host>:<port>"
)
try:
ExporterProtocol(protocol)
except ValueError:
raise ValueError(
f"Invalid OTel protocol: {protocol}. "
f"Expected values: {', '.join(e.value for e in ExporterProtocol)}"
)

return ExporterConfig(endpoint=endpoint, protocol=protocol)


def resolve(
service_name: str | None = None,
service_version: str | None = None,
) -> OtelConfig | None:
"""Read and validate shared OTel configuration from environment variables.
"""Read and validate OTel configuration from environment variables.

Returns None if OTel is not configured (no exporter endpoint set).
Raises ValueError if the configuration is invalid.
Returns None if no signals are configured (no exporter endpoints set).
Raises ValueError if any configured exporter has invalid settings.
"""
otel_endpoint = os.environ.get("TANGLE_OTEL_EXPORTER_ENDPOINT")
if not otel_endpoint:
return None

otel_protocol = os.environ.get(
"TANGLE_OTEL_EXPORTER_PROTOCOL", ExporterProtocol.GRPC
trace_exporter = _resolve_exporter(
endpoint_var="TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT",
protocol_var="TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL",
)

if trace_exporter is None:
return None

if service_name is None:
app_env = os.environ.get("TANGLE_ENV", "unknown")
Copy link
Collaborator

@maxy-shpfy maxy-shpfy Feb 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What are possible values for TANGLE_ENV? Where it is being set?

Copy link
Collaborator Author

@morgan-wowk morgan-wowk Feb 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We improve the documentation for this upstack but happy to answer here:

This is used as part of the service name for your OTel exports. This is literally your env, so development, staging, etc.

Alexey and I landed on this name for it in an earlier merged PR.

service_name = f"tangle-{app_env}"

if service_version is None:
service_version = os.environ.get("TANGLE_SERVICE_VERSION", "unknown")
Copy link
Collaborator

@maxy-shpfy maxy-shpfy Feb 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we have this env set today? what the versioning strategy do we use?

Copy link
Collaborator Author

@morgan-wowk morgan-wowk Feb 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Technically, it's up to the OSS user of Tangle (someone who clones and deploys Tangle themselves).

They can use their commit shas, semantic versioning, whatever they feel is right. For our case, we use a revision (commit sha) provided by our CD platform.

As for recommendations, we could add some to our documentation of this variable which is upstack. Suitable values would be build ids, URLs, or commit shas.

Copy link
Collaborator Author

@morgan-wowk morgan-wowk Feb 26, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PS: I'm not fully happy with the docs yet even after this full stack as it is now. I'd love to collaborate and hear ideas on how in-depth we want to go on docs and where our source of truth should be for metric information.


if not otel_endpoint.startswith(("http://", "https://")):
raise ValueError(
f"Invalid OTel endpoint format: {otel_endpoint}. "
f"Expected format: http://<host>:<port> or https://<host>:<port>"
)
try:
ExporterProtocol(otel_protocol)
except ValueError:
raise ValueError(
f"Invalid OTel protocol: {otel_protocol}. "
f"Expected values: {', '.join(e.value for e in ExporterProtocol)}"
)

return OtelConfig(
endpoint=otel_endpoint,
protocol=otel_protocol,
service_name=service_name,
service_version=service_version,
trace_exporter=trace_exporter,
)
17 changes: 8 additions & 9 deletions cloud_pipelines_backend/instrumentation/opentelemetry/providers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ def setup(
"""
Configure global OpenTelemetry providers (traces, metrics).

No-op if TANGLE_OTEL_EXPORTER_ENDPOINT is not set.
No-op if no signal-specific exporter endpoints are set.

Use this for non-FastAPI entrypoints (e.g. orchestrators, workers) that
need telemetry but have no ASGI app to auto-instrument.
Expand All @@ -42,11 +42,10 @@ def setup(
if otel_config is None:
return

tracing.setup(
endpoint=otel_config.endpoint,
protocol=otel_config.protocol,
service_name=otel_config.service_name,
service_version=otel_config.service_version,
)

# TODO: Setup metrics provider once it's available
if otel_config.trace_exporter:
tracing.setup(
endpoint=otel_config.trace_exporter.endpoint,
protocol=otel_config.trace_exporter.protocol,
service_name=otel_config.service_name,
service_version=otel_config.service_version,
)
86 changes: 57 additions & 29 deletions tests/instrumentation/opentelemetry/_internal/test_configuration.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,117 +24,145 @@ def test_invalid_value_raises(self):
class TestResolve:
"""Tests for configuration.resolve()."""

def test_returns_none_when_endpoint_not_set(self, monkeypatch):
monkeypatch.delenv("TANGLE_OTEL_EXPORTER_ENDPOINT", raising=False)
def test_returns_none_when_no_exporters_configured(self, monkeypatch):
monkeypatch.delenv("TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", raising=False)

result = configuration.resolve()

assert result is None

def test_returns_config_with_defaults(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.delenv("TANGLE_OTEL_EXPORTER_PROTOCOL", raising=False)
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.delenv("TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL", raising=False)
monkeypatch.delenv("TANGLE_ENV", raising=False)
monkeypatch.delenv("TANGLE_SERVICE_VERSION", raising=False)

result = configuration.resolve()

assert result is not None
assert result.endpoint == "http://localhost:4317"
assert result.protocol == configuration.ExporterProtocol.GRPC
assert result.trace_exporter.endpoint == "http://localhost:4317"
assert result.trace_exporter.protocol == configuration.ExporterProtocol.GRPC
assert result.service_name == "tangle-unknown"
assert result.service_version == "unknown"

def test_uses_custom_service_name(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

result = configuration.resolve(service_name="my-api")

assert result.service_name == "my-api"

def test_service_name_includes_tangle_env(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.setenv("TANGLE_ENV", "production")

result = configuration.resolve()

assert result.service_name == "tangle-production"

def test_respects_http_protocol(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4318")
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_PROTOCOL", "http")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4318"
)
monkeypatch.setenv("TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL", "http")

result = configuration.resolve()

assert result.protocol == configuration.ExporterProtocol.HTTP
assert result.trace_exporter.protocol == configuration.ExporterProtocol.HTTP

def test_respects_grpc_protocol(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_PROTOCOL", "grpc")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.setenv("TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL", "grpc")

result = configuration.resolve()

assert result.protocol == configuration.ExporterProtocol.GRPC
assert result.trace_exporter.protocol == configuration.ExporterProtocol.GRPC

def test_raises_on_invalid_endpoint_format(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "localhost:4317")
monkeypatch.setenv("TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "localhost:4317")

with pytest.raises(ValueError, match="Invalid OTel endpoint format"):
configuration.resolve()

def test_raises_on_invalid_protocol(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_PROTOCOL", "websocket")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.setenv("TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL", "websocket")

with pytest.raises(ValueError, match="Invalid OTel protocol"):
configuration.resolve()

def test_accepts_https_endpoint(self, monkeypatch):
monkeypatch.setenv(
"TANGLE_OTEL_EXPORTER_ENDPOINT", "https://collector.example.com:4317"
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "https://collector.example.com:4317"
)

result = configuration.resolve()

assert result.endpoint == "https://collector.example.com:4317"
assert result.trace_exporter.endpoint == "https://collector.example.com:4317"

def test_uses_custom_service_version(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

result = configuration.resolve(service_version="abc123")

assert result.service_version == "abc123"

def test_service_version_from_env(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.setenv("TANGLE_SERVICE_VERSION", "def456")

result = configuration.resolve()

assert result.service_version == "def456"

def test_service_version_defaults_to_unknown(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.delenv("TANGLE_SERVICE_VERSION", raising=False)

result = configuration.resolve()

assert result.service_version == "unknown"

def test_config_is_frozen(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

result = configuration.resolve()

with pytest.raises(AttributeError):
result.endpoint = "http://other:4317"
result.service_name = "other"

def test_config_requires_keyword_arguments(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
def test_exporter_config_is_frozen(self, monkeypatch):
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

result = configuration.resolve()

with pytest.raises(AttributeError):
result.trace_exporter.endpoint = "http://other:4317"

def test_config_requires_keyword_arguments(self):
with pytest.raises(TypeError):
configuration.OtelConfig("my-service", "unknown")

def test_exporter_config_requires_keyword_arguments(self):
with pytest.raises(TypeError):
configuration.OtelConfig(
result.endpoint, result.protocol, result.service_name
)
configuration.ExporterConfig("http://localhost:4317", "grpc")
20 changes: 13 additions & 7 deletions tests/instrumentation/opentelemetry/test_providers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@
class TestProvidersSetup:
"""Tests for providers.setup()."""

def test_noop_when_endpoint_not_set(self, monkeypatch):
monkeypatch.delenv("TANGLE_OTEL_EXPORTER_ENDPOINT", raising=False)
def test_noop_when_no_exporters_configured(self, monkeypatch):
monkeypatch.delenv("TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", raising=False)

providers.setup()

Expand All @@ -21,31 +21,37 @@ def test_noop_when_endpoint_not_set(self, monkeypatch):
)

def test_configures_tracer_provider(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.delenv("TANGLE_OTEL_EXPORTER_PROTOCOL", raising=False)
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)
monkeypatch.delenv("TANGLE_OTEL_TRACE_EXPORTER_PROTOCOL", raising=False)

providers.setup()

assert isinstance(trace.get_tracer_provider(), otel_sdk_trace.TracerProvider)

def test_passes_custom_service_name(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

providers.setup(service_name="my-orchestrator")

provider = trace.get_tracer_provider()
assert provider.resource.attributes["service.name"] == "my-orchestrator"

def test_passes_custom_service_version(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "http://localhost:4317")
monkeypatch.setenv(
"TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "http://localhost:4317"
)

providers.setup(service_name="test-service", service_version="abc123")

provider = trace.get_tracer_provider()
assert provider.resource.attributes["service.version"] == "abc123"

def test_catches_validation_errors(self, monkeypatch):
monkeypatch.setenv("TANGLE_OTEL_EXPORTER_ENDPOINT", "bad-endpoint")
monkeypatch.setenv("TANGLE_OTEL_TRACE_EXPORTER_ENDPOINT", "bad-endpoint")

providers.setup()

Expand Down