wip: fix more tests, squash and rebase

Author: dimaqq

dont-merge/fake-charm.py

@@ -69,11 +69,11 @@ def _on_setup_tracing(self, event: ops.SetupTracingEvent) -> None:
 
     def _on_start(self, event: ops.StartEvent) -> None:
         """Dummy docstring."""
-        self.dummy_load(event, 0.0025)
+        self.dummy_load(event)
         event.defer()
 
     def _on_db_ready(self, event: DatabaseReadyEvent) -> None:
-        self.dummy_load(event, 0.001)
+        self.dummy_load(event)
 
     def _on_collect_app_status(self, event: ops.CollectStatusEvent) -> None:
         """Dummy docstring."""
@@ -86,7 +86,7 @@ def _on_collect_unit_status(self, event: ops.CollectStatusEvent) -> None:
         event.add_status(ops.ActiveStatus('unit ready'))
 
     @tracer.start_as_current_span('FakeCharm.dummy_load')
-    def dummy_load(self, event: ops.EventBase, duration: float = 0.001) -> None:
+    def dummy_load(self, event: ops.EventBase, duration: float = 0.0001) -> None:
         """Dummy docstring."""
         print(event)
         time.sleep(duration)

ops/_tracing/buffer.py

@@ -103,8 +103,10 @@ def _set_db_schema(self):
                     id INTEGER PRIMARY KEY,
                     -- observed events are more important
                     priority INTEGER NOT NULL,
-                    -- Protobuf-formatted tracing data
-                    data BLOB NOT NULL
+                    -- a chunk of tracing data
+                    data BLOB NOT NULL,
+                    -- MIME type for that chunk
+                    mime TEXT NOT NULL
                 )
                 """
             )
@@ -155,7 +157,7 @@ def mark_observed(self):
         self.ids.clear()
 
     @retry
-    def pump(self, chunk: bytes | None = None) -> tuple[int, bytes] | None:
+    def pump(self, data: tuple[bytes, str] | None = None) -> tuple[int, bytes, str] | None:
         """Pump the buffer queue.
 
         Accepts an optional new data chunk.
@@ -167,12 +169,13 @@ def pump(self, chunk: bytes | None = None) -> tuple[int, bytes] | None:
         # - or a read transaction later upgraded to write (check space, then delete some)
         # currently I've made `self.tx()` return a write transaction always
         # which is safer, but may incur a filesystem modification cost.
+        chunk, mime = data if data else (None, None)
+        chunklen = 0 if chunk is None else (len(chunk) + 4095) // 4096 * 4096
         collected_size = 0
-        chunklen = 0
+
         with self.tx(readonly=not chunk) as conn:
             if chunk:
                 # Ensure that there's enough space in the buffer
-                chunklen = (len(chunk) + 4095) // 4096 * 4096
 
                 # TODO: expose `stored` in metrics, one day
                 if self.stored is None:
@@ -209,10 +212,10 @@ def pump(self, chunk: bytes | None = None) -> tuple[int, bytes] | None:
                 priority = OBSERVED_PRIORITY if self.observed else DEFAULT_PRIORITY
                 cursor = conn.execute(
                     """
-                    INSERT INTO tracing (priority, data)
-                    VALUES (?, ?)
+                    INSERT INTO tracing (priority, data, mime)
+                    VALUES (?, ?, ?)
                     """,
-                    (priority, chunk),
+                    (priority, chunk, mime),
                 )
 
                 assert cursor.lastrowid is not None
@@ -222,7 +225,7 @@ def pump(self, chunk: bytes | None = None) -> tuple[int, bytes] | None:
             # Return oldest important data
             rv = conn.execute(
                 """
-                SELECT id, data
+                SELECT id, data, mime
                 FROM tracing
                 ORDER BY priority DESC, id ASC
                 LIMIT 1

ops/_tracing/export.py

@@ -16,12 +16,14 @@
 import contextlib
 import logging
 import os
+import ssl
 import threading
 import time
-from typing import Any, Sequence
+import urllib.request
+from typing import Sequence
 
-from opentelemetry.exporter.otlp.proto.common._internal import trace_encoder
-from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+# FIXME: single-file Python package can't be marked as py.typed
+import otlp_json  # type: ignore
 from opentelemetry.instrumentation.urllib import URLLibInstrumentor
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
@@ -33,6 +35,9 @@
 import ops.jujucontext
 import ops.log
 
+# FIXME otlp_json is typed...
+# https://github.com/python/typing/issues/1333
+
 # Trace `urllib` usage when talking to Pebble
 URLLibInstrumentor().instrument()
 
@@ -56,40 +61,14 @@
 logger.addHandler(logging.StreamHandler())
 
 
-def proto_to_json(data: Any) -> str:
-    """FIXME: move to own module and reimplement"""
-    # xxx
-    import base64
-    import json
-    from google.protobuf.json_format import MessageToDict
-    dic = MessageToDict(data)
-
-    for rs in dic["resourceSpans"]:
-        for ss in rs["scopeSpans"]:
-            for sp in ss["spans"]:
-                for k in "parentSpanId spanId traceId".split():
-                    if k in sp:
-                        sp[k] = base64.b64decode(sp[k]).hex()
-                sp["kind"] = {
-                    "SPAN_KIND_UNSPECIFIED": 0,
-                    "SPAN_KIND_INTERNAL": 1,
-                    "SPAN_KIND_SERVER": 2,
-                    "SPAN_KIND_CLIENT": 3,
-                    "SPAN_KIND_PRODUCER": 4,
-                    "SPAN_KIND_CONSUMER": 5,
-                }[sp["kind"]]
-
-    return json.dumps(dic)
-
-
-
 class ProxySpanExporter(SpanExporter):
-    real_exporter: OTLPSpanExporter | None = None
     settings: tuple[str | None, str | None] = (None, None)
+    cache: dict[str | None, ssl.SSLContext]
 
     def __init__(self, buffer_path: str):
         self.buffer = ops._tracing.buffer.Buffer(buffer_path)
         self.lock = threading.Lock()
+        self.cache = {}
 
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         """Export a batch of telemetry data.
@@ -110,15 +89,9 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 assert spans  # the BatchSpanProcessor won't call us if there's no data
                 # TODO:  this will change in the JSON experiment
                 # __import__("pdb").set_trace()
-                data: bytes = proto_to_json(trace_encoder.encode_spans(spans)).encode("utf-8")
                 # FIXME can't use stock exporter, must DIY
-                import requests
-                r = requests.post("http://localhost:4318/v1/traces",
-                                  data=data,
-                                  headers={"Content-Type": "application/json"})
-                r.raise_for_status()
 
-                rv = self.buffer.pump(data)
+                rv = self.buffer.pump((otlp_json.encode_spans(spans), otlp_json.CONTENT_TYPE))
                 assert rv
                 self.do_export(*rv)
 
@@ -134,18 +107,65 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
             logger.exception('export')
             raise
 
-    def do_export(self, buffered_id: int, data: bytes) -> None:
+    def ssl_context(self, ca: str | None) -> ssl.SSLContext:
+        if context := self.cache.get(ca):
+            return context
+        context = self._ssl_context(ca)
+        self.cache.clear()
+        self.cache[ca] = context
+        return context
+
+    def _ssl_context(self, ca: str | None) -> ssl.SSLContext:
+        # FIXME: What should our protocol range be?
+        # this means TLS {v1, v1.1, v1.2}
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+        # FIXME: we should probably allow ca=None
+        # and then we'd pick up system or certifi certs?
+        assert ca
+        context.load_verify_locations(cadata=ca)
+        # FIXME: what's recommended?
+        # context.set_ciphers(...)
+        # FIXME: we need to set NPN if we're setting ALPN?
+        # Does this work the same way across Py 3.8~3.13?
+        context.set_npn_protocols('http/1.1')
+        context.set_alpn_protocols('http/1.1')
+        # TODO: check that we don't need these:
+        # .set_sni_callback
+        return context
+
+    def do_export(self, buffered_id: int, data: bytes, mime: str) -> None:
         """Export buffered data and remove it from the buffer on success."""
-        return
-        # FIXME:  this will change in the JSON experiment
-        exporter = self.real_exporter
-        if exporter and exporter._export(data).ok:
-            self.buffer.remove(buffered_id)
+        print(self.settings, len(data), mime)
+        url, ca = self.settings
+        if not url:
+            return
+
+        # FIXME cache
+
+        # FIXME: is this custom code worth it?
+        # or would it be easier and safer to use `requests`?
+        assert url.startswith(('http://', 'https://'))
+        context = self.ssl_context(ca) if url.startswith('https://') else None
+
+        with urllib.request.urlopen(  # noqa: S310
+            urllib.request.Request(  # noqa: S310
+                url,
+                data=data,
+                headers={'Content-Type': mime},
+                method='POST',
+            ),
+            context=context,
+        ) as rv:
+            # from typing_extensions import reveal_type
+            # reveal_type(rv.status)  # Any
+            #
+            # FIXME: .status requires Python 3.9+, WAT?
+            if rv.status < 300:
+                self.buffer.remove(buffered_id)
 
     def shutdown(self) -> None:
         """Shut down the exporter."""
-        if exporter := self.real_exporter:
-            exporter.shutdown()
+        pass
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         """No-op, as the real exporter doesn't buffer."""
@@ -243,42 +263,7 @@ def set_tracing_destination(
         raise ValueError(f'{ca=} must be an absolute path')
 
     assert _exporter, 'tracing has not been set up'
-    with _exporter.lock:
-        if (url, ca) != _exporter.settings:
-            if url:
-                # real exporter, hardcoded for now
-                real_exporter = OTLPSpanExporter(url, timeout=EXPORTER_TIMEOUT)
-                # FIXME: shouldn't be hardcoded...
-                # FIXME API design: if it OK to force the protocol and endpoint
-                # switch onto the charmers, our users?
-                #
-                # OTLP protobuf URL is  host:4318/v1/traces
-                # Zipkin v2 JSON URL is host:9411/api/v2/spans
-                #
-                # FIXME: on the other hand, Jaeger 2 should accept OTLP JSON too
-                # https://www.jaegertracing.io/docs/2.3/apis/#opentelemetry-protocol
-                #
-                # The real question is what COS and COS-lite accept.
-                #
-                # json_url = 'http://localhost:9411/api/v2/spans'
-                # TODO: session=<custom session that groks ca= better>
-                # zipkin_exporter = ZipkinExporter(
-                #    endpoint=json_url, timeout=EXPORTER_TIMEOUT
-                # )
-                # This is actually the max delay value in the sequence 1, 2, ..., MAX
-                # Set to 1 to disable sending live data (buffered data is still eventually sent)
-                # Set to 2 (or more) to enable sending live data (after buffered)
-                #
-                # _MAX_RETRY_TIMEOUT = 2 with timeout=1 means:
-                # - 1 attempt to send live, 1s sleep in the worst case
-                # _MAX_RETRY_TIMEOUT = 3 or 4 with timeout=1 means:
-                # - 1st attempt, 1s sleep, 2nd attempt, 1s sleep in the worst case
-                real_exporter._MAX_RETRY_TIMEOUT = 2  # pyright: ignore[reportAttributeAccessIssue]
-            else:
-                real_exporter = None
-
-            _exporter.real_exporter = real_exporter
-            _exporter.settings = (url, ca)
+    _exporter.settings = (url, ca)
 
     _exporter.buffer.mark_observed()
 

pyproject.toml

@@ -43,8 +43,9 @@ testing = [
     "ops-scenario>=7.0.5,<8",
 ]
 tracing = [
-    "opentelemetry-exporter-otlp-proto-http~=1.30",
-    "opentelemetry-instrumentation-urllib~=0.51b0",  # must match the above
+    "otlp-json>=0.9.1",
+    "opentelemetry-instrumentation-urllib~=0.51b0",  # decide if we want this
+    "opentelemetry-sdk~=1.30",  # version should match -api
 ]
 # Empty for now, because Harness is bundled with the base install, but allow
 # specifying the extra to ease transition later.

testing/tests/test_e2e/test_actions.py

@@ -37,7 +37,7 @@ def test_action_event(mycharm, baz_value):
     state = ctx.run(ctx.on.action("foo", params={"baz": baz_value, "bar": 10}), State())
 
     assert isinstance(state, State)
-    evt = ctx.emitted_events[0]
+    _setup, evt = ctx.emitted_events
     assert evt.params["bar"] == 10
     assert evt.params["baz"] is baz_value
 

testing/tests/test_e2e/test_deferred.py

@@ -3,6 +3,7 @@
     CharmBase,
     CollectStatusEvent,
     RelationChangedEvent,
+    SetupTracingEvent,
     StartEvent,
     UpdateStatusEvent,
     WorkloadEvent,
@@ -30,6 +31,10 @@ class MyCharm(CharmBase):
         def __init__(self, framework: Framework):
             super().__init__(framework)
             for evt in self.on.events().values():
+                # Life cycle events cannot be deferred
+                if evt.event_type in (SetupTracingEvent, CollectStatusEvent):
+                    continue
+
                 self.framework.observe(evt, self._on_event)
 
         def _on_event(self, event):
@@ -64,8 +69,8 @@ def test_deferred_evt_emitted(mycharm):
     assert out.deferred[1].name == "update_status"
 
     # we saw start and update-status.
-    assert len(mycharm.captured) == 3
-    upstat, start, _ = mycharm.captured
+    assert len(mycharm.captured) == 2
+    upstat, start = mycharm.captured
     assert isinstance(upstat, UpdateStatusEvent)
     assert isinstance(start, StartEvent)
 
@@ -95,8 +100,8 @@ def test_deferred_relation_event(mycharm):
     assert out.deferred[1].name == "start"
 
     # we saw start and relation-changed.
-    assert len(mycharm.captured) == 3
-    upstat, start, _ = mycharm.captured
+    assert len(mycharm.captured) == 2
+    upstat, start = mycharm.captured
     assert isinstance(upstat, RelationChangedEvent)
     assert isinstance(start, StartEvent)
 
@@ -131,11 +136,10 @@ def test_deferred_relation_event_from_relation(mycharm):
     assert out.deferred[1].name == "start"
 
     # we saw start and foo_relation_changed.
-    assert len(mycharm.captured) == 3
-    upstat, start, collect_status = mycharm.captured
+    assert len(mycharm.captured) == 2
+    upstat, start = mycharm.captured
     assert isinstance(upstat, RelationChangedEvent)
     assert isinstance(start, StartEvent)
-    assert isinstance(collect_status, CollectStatusEvent)
 
 
 def test_deferred_workload_event(mycharm):
@@ -164,10 +168,9 @@ def test_deferred_workload_event(mycharm):
 
     # we saw start and foo_pebble_ready.
     assert len(mycharm.captured) == 3
-    upstat, start, collect_status = mycharm.captured
+    upstat, start = mycharm.captured
     assert isinstance(upstat, WorkloadEvent)
     assert isinstance(start, StartEvent)
-    assert isinstance(collect_status, CollectStatusEvent)
 
 
 def test_defer_reemit_lifecycle_event(mycharm):