diff --git a/README.md b/README.md index 5b923a0..3c5d248 100644 --- a/README.md +++ b/README.md @@ -33,7 +33,7 @@ A modern, high-performance ETW (Event Tracing for Windows) toolkit for Python, p - **Live Dashboard**: Browser-based real-time visualization with Gradio - **Event Correlation Engine**: Auto-correlate events by PID/TID/Handle - **Recording & Replay**: Capture and replay ETW sessions (.etwpack format) -- **OpenTelemetry span mapping**: Map ETW events to OTLP spans and write them to a file. Sending to a collector is [not implemented yet](https://github.com/m96-chan/PyETWkit/issues/88) +- **OpenTelemetry Exporter**: Send events to an OTLP collector over HTTP (Jaeger, Grafana, Datadog), or write spans to a file. No extra dependency ### Export Formats - CSV, JSON, JSONL, Parquet, Arrow @@ -168,16 +168,13 @@ for event in player.events(): print(f"Event {event['event_id']}") ``` -### OpenTelemetry Span Export +### OpenTelemetry Export -> **Sending to a collector is not implemented.** `OtlpExporter` maps events to -> spans but has no HTTP or gRPC transport, and its `flush()` raises -> `NotImplementedError` rather than reporting a delivery it cannot make. Use -> `OtlpFileExporter` to get spans out today; see -> [#88](https://github.com/m96-chan/PyETWkit/issues/88) for the transport. +Spans are sent as OTLP/HTTP with JSON encoding, so no extra dependency is +needed. Note **4318** — 4317 is the gRPC port and will not answer HTTP. ```python -from pyetwkit import OtlpFileExporter, SpanMapper +from pyetwkit import OtlpExporter, SpanMapper # Map ETW events to spans mapper = SpanMapper() @@ -185,10 +182,29 @@ mapper.add_rule( provider="Microsoft-Windows-Kernel-Process", event_id=1, span_name="process.start", - attributes=["ProcessId", "ImageFileName"], + attributes=["ProcessID", "ImageName"], ) -# Write spans to a file, ready to be shipped by a collector agent +exporter = OtlpExporter( + endpoint="http://collector:4318", + service_name="my-service", + resource_attributes={"deployment.environment": "production"}, + span_mapper=mapper, +) + +for event in events: + exporter.export(event) + +# False means nothing was delivered; the batch is kept so it can be retried. +if not exporter.flush(): + log.warning("OTLP export failed; see the log for the reason") +``` + +To write spans to a file instead, for a collector agent to pick up: + +```python +from pyetwkit import OtlpFileExporter + exporter = OtlpFileExporter("traces.json", service_name="my-service") for event in events: exporter.export(event) @@ -296,7 +312,7 @@ is the first release since v3.0.1 that PyPI users will see._ - **Live Dashboard**: Gradio-based real-time UI (`pyetwkit dashboard` CLI) - **Event Correlation Engine**: Link events by PID/TID/Handle with timeline export - **Recording & Replay**: Capture sessions to `.etwpack` format with compression -- **OpenTelemetry Exporter**: Export to OTLP endpoints (Jaeger, Grafana, etc.) — _announced, but the transport was never implemented; see [#88](https://github.com/m96-chan/PyETWkit/issues/88)_ +- **OpenTelemetry Exporter**: Export to OTLP endpoints (Jaeger, Grafana, etc.) — _announced here, but the transport was not actually implemented until v3.2.0 (#88, #90)_ ### v2.0.0 (2024-12) - **Multi-session support**: Run multiple ETW sessions simultaneously diff --git a/src/pyetwkit/exporters/otlp.py b/src/pyetwkit/exporters/otlp.py index e17e4c5..4b0c61d 100644 --- a/src/pyetwkit/exporters/otlp.py +++ b/src/pyetwkit/exporters/otlp.py @@ -8,18 +8,69 @@ import json import logging +import re import time import uuid from dataclasses import dataclass, field +from datetime import datetime from enum import Enum from pathlib import Path from typing import TYPE_CHECKING, Any +from urllib.error import HTTPError, URLError +from urllib.parse import urlsplit, urlunsplit +from urllib.request import Request, urlopen if TYPE_CHECKING: pass logger = logging.getLogger(__name__) +# OTLP/HTTP puts traces here. The default port is 4318; 4317 is the gRPC one. +OTLP_TRACES_PATH = "/v1/traces" + +# OTLP's JSON encoding follows the protobuf JSON mapping with one relevant +# deviation: "only integer enum values are allowed in OTLP JSON Protobuf +# Encoding; the enum name strings MUST NOT be used". These used to be sent as +# "INTERNAL" and "OK", which a collector rejects. +# https://opentelemetry.io/docs/specs/otlp/ +SPAN_KIND_INTERNAL = 1 +STATUS_CODE_OK = 1 + + +def _timestamp_seconds(raw: Any) -> float: + """Seconds since the epoch, from whatever an event carries. + + `EtwEvent.timestamp` is an RFC 3339 string with nanosecond precision, e.g. + "2026-09-05T13:09:24.061643500+00:00". `float()` on that raises, which meant + every real event blew up here -- the existing tests all used mocks or plain + numbers. `datetime.fromisoformat` cannot take nine fractional digits either + on the Python versions this supports, so the fraction is trimmed to six. + """ + if hasattr(raw, "timestamp"): # datetime + return float(raw.timestamp()) + + if isinstance(raw, str): + text = re.sub(r"(\.\d{6})\d+", r"\1", raw) + try: + return datetime.fromisoformat(text).timestamp() + except ValueError: + logger.warning("Unparseable event timestamp %r; using now()", raw) + return time.time() + + return float(raw) + + +def _package_version() -> str: + """The package version, imported lazily. + + `pyetwkit/__init__.py` imports this module, so importing it back at module + scope would be a cycle that happens to work only because `__version__` is + defined before that import runs. + """ + from pyetwkit import __version__ + + return __version__ + class ExportMode(Enum): """Export modes for ETW events.""" @@ -145,27 +196,31 @@ def extract_attributes(self, event: Any) -> dict[str, Any]: class OtlpExporter: - """Maps ETW events to OTLP spans. **Sending is not implemented.** + """Exports ETW events to an OTLP collector over HTTP. + + Spans are sent as OTLP/HTTP with JSON encoding, POSTed to ``/v1/traces`` on + the given endpoint. That encoding needs no dependencies beyond the standard + library, which is why it is used in preference to gRPC or protobuf. + + The endpoint is used as given; only a missing path is filled in. **OTLP/HTTP + is normally port 4318** -- 4317 is the gRPC port and will not answer an HTTP + request. - Nothing in this class transmits anything: there is no HTTP or gRPC client - behind `endpoint`, and :meth:`flush` raises :class:`NotImplementedError` - rather than reporting a success it cannot deliver. Event to span mapping, - sampling and batching all work, so this remains useful for building the - payload, but it will not reach a collector. + :meth:`flush` returns False and logs the reason when a send fails, keeping + the batch so it can be retried. It never raises: exporters are driven from + event callbacks, and a collector being down should not stop a trace session. - Use :class:`OtlpFileExporter` to write spans out today, and see #88 for the - transport. + See :class:`OtlpFileExporter` to write spans to a file instead. Example: >>> exporter = OtlpExporter( - ... endpoint="http://collector:4317", + ... endpoint="http://collector:4318", ... service_name="windows-etw" ... ) - >>> exporter.export(event) # buffers the span + >>> exporter.export(event) True - >>> exporter.flush() # raises NotImplementedError - Traceback (most recent call last): - NotImplementedError: ... + >>> if not exporter.flush(): + ... log.warning("OTLP export failed; see the log for the reason") """ def __init__( @@ -239,17 +294,14 @@ def sample_rate(self) -> float: return self._sample_rate def export(self, event: Any) -> bool: - """Buffer one event as a span. + """Buffer one event as a span, sending once the batch is full. Args: event: ETW event to export. Returns: - True once the span is buffered. - - Raises: - NotImplementedError: if this fills the batch, since that triggers a - :meth:`flush` and there is no transport to flush to. + True once the span is buffered, or the result of the :meth:`flush` + this triggers if the event fills the batch. """ # Apply sampling if self._sample_rate < 1.0: @@ -283,38 +335,119 @@ def export_batch(self, events: list[Any]) -> bool: self.export(event) return self.flush() + def _traces_url(self) -> str: + """The URL to POST to. + + The endpoint is used as given. Only the path is filled in, and only when + there is none: rewriting what the caller passed -- including "helpfully" + turning the gRPC port 4317 into the HTTP one -- would break anyone + serving OTLP/HTTP somewhere else, and quietly. + """ + parts = urlsplit(self._endpoint) + if parts.path in ("", "/"): + return urlunsplit(parts._replace(path=OTLP_TRACES_PATH)) + return self._endpoint + def flush(self) -> bool: - """Send pending spans to the collector. **Not implemented.** + """Send pending spans to the collector. - Raises: - NotImplementedError: always, when there is anything to send. + Returns: + True if there was nothing to send or the collector accepted it, + False if the send failed. Failures are logged with the reason. - This used to clear the batch and return True, so every event was - discarded and the caller was told it had been delivered. Failing is the - only honest answer until there is a transport: a monitoring pipeline - that reports success while sending nothing is worse than one that stops. + The batch is kept when a send fails, so the events can be retried rather + than lost. Nothing is raised: this is called from event callbacks, and a + collector being down should not take the trace session with it. """ if not self._batch: self._last_export = time.time() return True - pending = len(self._batch) - raise NotImplementedError( - f"OtlpExporter cannot send: no OTLP transport is implemented, so " - f"{pending} span(s) would be silently discarded. Use OtlpFileExporter " - f"to write spans to a file, or follow " - f"https://github.com/m96-chan/PyETWkit/issues/88 for the transport." - ) + url = self._traces_url() + payload = json.dumps(self._build_request(self._batch)).encode("utf-8") + + request = Request(url, data=payload, method="POST") + request.add_header("Content-Type", "application/json") + for name, value in self._headers.items(): + request.add_header(name, value) + + timeout = self._config.timeout_ms / 1000.0 + try: + with urlopen(request, timeout=timeout) as response: # noqa: S310 - caller's URL + status = response.status + except HTTPError as e: + logger.error( + "OTLP export to %s failed: HTTP %s %s. %d span(s) kept for retry.", + url, + e.code, + e.reason, + len(self._batch), + ) + return False + except URLError as e: + hint = "" + if urlsplit(self._endpoint).port == 4317: + hint = " (port 4317 is the OTLP/gRPC port; OTLP/HTTP is usually 4318)" + logger.error( + "OTLP export to %s failed: %s%s. %d span(s) kept for retry.", + url, + e.reason, + hint, + len(self._batch), + ) + return False + except OSError as e: + logger.error( + "OTLP export to %s failed: %s. %d span(s) kept for retry.", + url, + e, + len(self._batch), + ) + return False + + if not 200 <= status < 300: + logger.error( + "OTLP export to %s failed: HTTP %s. %d span(s) kept for retry.", + url, + status, + len(self._batch), + ) + return False + + self._batch.clear() + self._last_export = time.time() + return True + + def _build_request(self, spans: list[dict[str, Any]]) -> dict[str, Any]: + """Wrap spans in the OTLP ExportTraceServiceRequest envelope.""" + attributes = [ + {"key": "service.name", "value": {"stringValue": self._service_name}}, + *( + {"key": key, "value": _attribute_value(value)} + for key, value in self._resource_attributes.items() + ), + ] + return { + "resourceSpans": [ + { + "resource": {"attributes": attributes}, + "scopeSpans": [ + { + "scope": {"name": "pyetwkit", "version": _package_version()}, + "spans": spans, + } + ], + } + ] + } def shutdown(self) -> None: - """Shutdown the exporter. + """Flush what is left and stop. - Discards anything still buffered rather than raising from a teardown - path, since callers reach this from ``finally`` blocks. `flush` is where - the missing transport is reported. + Callers reach this from ``finally``, so a failure is logged by `flush` + and otherwise ignored rather than raised from a teardown path. """ - self._batch.clear() - self._last_export = time.time() + self.flush() def attach_to_session(self, session: Any) -> None: """Attach the exporter to an ETW session. @@ -424,18 +557,13 @@ def event_to_span( thread_id = getattr(event, "thread_id", 0) properties = getattr(event, "properties", {}) - # Convert timestamp to float (seconds since epoch) - if hasattr(raw_timestamp, "timestamp"): - # datetime object - timestamp = raw_timestamp.timestamp() - else: - timestamp = float(raw_timestamp) + timestamp = _timestamp_seconds(raw_timestamp) return { "traceId": uuid.uuid4().hex, "spanId": uuid.uuid4().hex[:16], "name": span_name or f"{provider_name}.{event_id}", - "kind": "INTERNAL", + "kind": SPAN_KIND_INTERNAL, "startTimeUnixNano": int(timestamp * 1e9), "endTimeUnixNano": int(timestamp * 1e9), "attributes": [ @@ -446,7 +574,7 @@ def event_to_span( {"key": "thread.id", "value": {"intValue": thread_id}}, *[{"key": f"etw.{k}", "value": _attribute_value(v)} for k, v in properties.items()], ], - "status": {"code": "OK"}, + "status": {"code": STATUS_CODE_OK}, } @@ -469,12 +597,7 @@ def event_to_log( process_id = getattr(event, "process_id", 0) properties = getattr(event, "properties", {}) - # Convert timestamp to float (seconds since epoch) - if hasattr(raw_timestamp, "timestamp"): - # datetime object - timestamp = raw_timestamp.timestamp() - else: - timestamp = float(raw_timestamp) + timestamp = _timestamp_seconds(raw_timestamp) return { "timeUnixNano": int(timestamp * 1e9), diff --git a/tests/test_otlp_exporter.py b/tests/test_otlp_exporter.py index 43ba874..43e6676 100644 --- a/tests/test_otlp_exporter.py +++ b/tests/test_otlp_exporter.py @@ -238,19 +238,22 @@ def test_exporter_shutdown(self) -> None: assert hasattr(exporter, "shutdown") def test_flush_does_not_report_success_without_sending(self) -> None: - """There is no transport, so flush must not claim delivery. + """Nothing delivered means no success, however that comes about. - It used to clear the batch and return True, which meant every event was - discarded while the caller was told it had been sent. Port 1 is closed, - so a real transport could not succeed here either. + This once returned True while discarding everything (#88), then raised + NotImplementedError while there was no transport, and now reports a + failed send. The contract each time is the same: do not claim delivery. + Port 1 is closed, so nothing can have arrived. + + The successful path lives in test_otlp_transport.py, which runs a real + collector. """ from pyetwkit.exporters import OtlpExporter exporter = OtlpExporter(endpoint="http://127.0.0.1:1") exporter.export({"provider_name": "P", "event_id": 1, "process_id": 1, "properties": {}}) - with pytest.raises(NotImplementedError, match="no OTLP transport"): - exporter.flush() + assert exporter.flush() is False def test_export_batch_does_not_report_success_either(self) -> None: """export_batch() ends in a flush, so it cannot claim delivery either.""" @@ -258,10 +261,12 @@ def test_export_batch_does_not_report_success_either(self) -> None: exporter = OtlpExporter(endpoint="http://127.0.0.1:1") - with pytest.raises(NotImplementedError): + assert ( exporter.export_batch( [{"provider_name": "P", "event_id": 1, "process_id": 1, "properties": {}}] ) + is False + ) def test_flush_with_nothing_buffered_is_not_an_error(self) -> None: """Nothing to send is not a failure to send.""" diff --git a/tests/test_otlp_transport.py b/tests/test_otlp_transport.py new file mode 100644 index 0000000..f36e7f1 --- /dev/null +++ b/tests/test_otlp_transport.py @@ -0,0 +1,299 @@ +"""Tests for the OTLP/HTTP transport (#90). + +These run a real HTTP server from the standard library and assert on what was +actually received, rather than mocking the send. The exporter shipped for months +returning success without a transport at all (#88), which no amount of mocking +would have caught. +""" + +from __future__ import annotations + +import json +import logging +import threading +from http.server import BaseHTTPRequestHandler, HTTPServer +from types import SimpleNamespace +from typing import Any + +import pytest + +from pyetwkit import OtlpExporter + + +# A stand-in for EtwEvent. The exporter reads events by attribute, so a plain +# dict would silently produce a span of defaults -- which is what the old tests +# in this repo were unknowingly asserting against. +def make_event(**overrides: Any) -> SimpleNamespace: + fields: dict[str, Any] = { + "provider_name": "Microsoft-Windows-Kernel-Process", + "event_id": 1, + "process_id": 4104, + "thread_id": 512, + "timestamp": 1788613764.0, + "properties": {"ImageName": "cmd.exe"}, + } + fields.update(overrides) + return SimpleNamespace(**fields) + + +SAMPLE = make_event() + + +class _Collector(BaseHTTPRequestHandler): + """Minimal OTLP collector that records what it was sent.""" + + received: list[dict[str, Any]] = [] + status = 200 + + def do_POST(self) -> None: # noqa: N802 - name fixed by BaseHTTPRequestHandler + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) + type(self).received.append( + { + "path": self.path, + "content_type": self.headers.get("Content-Type"), + "headers": dict(self.headers), + "body": json.loads(body) if body else None, + } + ) + self.send_response(type(self).status) + self.end_headers() + + def log_message(self, *args: Any) -> None: + """Keep the handler's own logging out of the test output.""" + + +@pytest.fixture +def collector(): + _Collector.received = [] + _Collector.status = 200 + server = HTTPServer(("127.0.0.1", 0), _Collector) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + host, port = server.server_address + yield f"http://{host}:{port}", _Collector + finally: + server.shutdown() + thread.join(timeout=5) + server.server_close() + + +def _spans(request: dict[str, Any]) -> list[dict[str, Any]]: + return request["body"]["resourceSpans"][0]["scopeSpans"][0]["spans"] + + +class TestOtlpHttpTransport: + def test_flush_posts_otlp_json_to_v1_traces(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint, service_name="svc") + exporter.export(SAMPLE) + + assert exporter.flush() is True + assert len(sink.received) == 1 + + request = sink.received[0] + assert request["path"] == "/v1/traces" + assert request["content_type"] == "application/json" + assert len(_spans(request)) == 1 + + def test_resource_carries_service_name_and_attributes(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter( + endpoint=endpoint, + service_name="windows-etw", + resource_attributes={"deployment.environment": "production"}, + ) + exporter.export(SAMPLE) + exporter.flush() + + resource = sink.received[0]["body"]["resourceSpans"][0]["resource"] + attributes = {a["key"]: a["value"] for a in resource["attributes"]} + assert attributes["service.name"]["stringValue"] == "windows-etw" + assert attributes["deployment.environment"]["stringValue"] == "production" + + def test_enums_are_integers_not_names(self, collector) -> None: + """OTLP JSON forbids enum name strings. + + "only integer enum values are allowed in OTLP JSON Protobuf Encoding; + the enum name strings MUST NOT be used." -- the exporter used to send + "INTERNAL" and "OK", which a collector rejects. + """ + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(SAMPLE) + exporter.flush() + + span = _spans(sink.received[0])[0] + assert span["kind"] == 1 + assert span["status"]["code"] == 1 + + def test_trace_and_span_ids_are_hex_of_the_right_length(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(SAMPLE) + exporter.flush() + + span = _spans(sink.received[0])[0] + assert len(span["traceId"]) == 32 + assert len(span["spanId"]) == 16 + int(span["traceId"], 16) + int(span["spanId"], 16) + + def test_custom_headers_are_sent(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint, headers={"X-Api-Key": "secret"}) + exporter.export(SAMPLE) + exporter.flush() + + assert sink.received[0]["headers"]["X-Api-Key"] == "secret" + + def test_endpoint_with_an_explicit_path_is_left_alone(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=f"{endpoint}/custom/v1/traces") + exporter.export(SAMPLE) + exporter.flush() + + assert sink.received[0]["path"] == "/custom/v1/traces" + + def test_batch_is_cleared_after_a_successful_send(self, collector) -> None: + endpoint, _ = collector + + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(SAMPLE) + assert exporter.flush() is True + + # Nothing left, so a second flush sends nothing and still succeeds. + assert exporter.flush() is True + + def test_export_batch_sends_everything(self, collector) -> None: + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint) + assert exporter.export_batch([SAMPLE, SAMPLE, SAMPLE]) is True + assert len(_spans(sink.received[0])) == 3 + + def test_reaching_batch_size_sends_without_an_explicit_flush(self, collector) -> None: + from pyetwkit.exporters import OtlpExporterConfig + + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint, config=OtlpExporterConfig(batch_size=2)) + exporter.export(SAMPLE) + assert sink.received == [] + + exporter.export(SAMPLE) + assert len(sink.received) == 1 + assert len(_spans(sink.received[0])) == 2 + + +class TestRealEventShapes: + """The shapes an actual `EtwEvent` has, as opposed to a mock's.""" + + def test_iso8601_timestamp_with_nanoseconds(self, collector) -> None: + """`EtwEvent.timestamp` is an RFC 3339 string, not a number. + + `float()` on it raises, so every real event used to blow up here. The + existing tests all passed mocks or plain floats, which is why nobody + noticed. Nine fractional digits also defeat `datetime.fromisoformat`. + """ + endpoint, sink = collector + + event = make_event(timestamp="2026-09-05T13:09:24.061643500+00:00") + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(event) + + assert exporter.flush() is True + + span = _spans(sink.received[0])[0] + # 2026-09-05T13:09:24Z in nanoseconds, to the second. + assert span["startTimeUnixNano"] // 1_000_000_000 == 1788613764 + + def test_datetime_timestamp_still_works(self, collector) -> None: + from datetime import datetime, timezone + + endpoint, sink = collector + + when = datetime(2026, 9, 5, 13, 9, 24, tzinfo=timezone.utc) + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(make_event(timestamp=when)) + exporter.flush() + + span = _spans(sink.received[0])[0] + assert span["startTimeUnixNano"] // 1_000_000_000 == int(when.timestamp()) + + def test_an_unparseable_timestamp_does_not_lose_the_span(self, collector) -> None: + """A bad timestamp is not worth dropping the event over.""" + endpoint, sink = collector + + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(make_event(timestamp="not a timestamp")) + + assert exporter.flush() is True + assert len(_spans(sink.received[0])) == 1 + + def test_events_read_from_an_etl_file_export(self, collector) -> None: + """End to end, with events from the committed capture.""" + from pathlib import Path + + from pyetwkit._core import EtlReader + + fixture = Path(__file__).parent / "fixtures" / "sample.etl" + if not fixture.exists(): + pytest.skip("Requires sample ETL file") + + endpoint, sink = collector + events = EtlReader(str(fixture)).read_all() + assert events, "fixture produced no events" + + exporter = OtlpExporter(endpoint=endpoint, service_name="windows-etw") + for event in events: + exporter.export(event) + + assert exporter.flush() is True + assert len(_spans(sink.received[0])) == len(events) + + +class TestOtlpHttpFailures: + """A failure must be visible and must not cost the events.""" + + def test_server_error_returns_false_and_keeps_the_batch(self, collector, caplog) -> None: + endpoint, sink = collector + sink.status = 503 + + exporter = OtlpExporter(endpoint=endpoint) + exporter.export(SAMPLE) + + with caplog.at_level(logging.ERROR): + assert exporter.flush() is False + + assert "503" in caplog.text + # Kept, so the caller can retry rather than losing the events. + assert exporter.flush() is False + assert len(sink.received) == 2 + + def test_unreachable_collector_returns_false_without_raising(self, caplog) -> None: + """Port 1 is closed. The exception must not reach the caller.""" + exporter = OtlpExporter(endpoint="http://127.0.0.1:1") + exporter.export(SAMPLE) + + with caplog.at_level(logging.ERROR): + assert exporter.flush() is False + + assert caplog.text + + def test_flush_with_nothing_buffered_is_not_a_failure(self) -> None: + exporter = OtlpExporter(endpoint="http://127.0.0.1:1") + assert exporter.flush() is True + + def test_shutdown_does_not_raise_when_the_collector_is_gone(self) -> None: + """Callers reach shutdown from `finally`.""" + exporter = OtlpExporter(endpoint="http://127.0.0.1:1") + exporter.export(SAMPLE) + exporter.shutdown()