Send collector data through the CA file and proxy

.changesets/apply-ca-file-path-and-http-proxy-to-collector-data.md

@@ -0,0 +1,6 @@
+---
+bump: patch
+type: fix
+---
+
+Apply the `ca_file_path` and `http_proxy` configuration options to the data sent to a collector. Before this change both options were only applied to the data sent by the agent, so a custom certificate authority file or a proxy had no effect when a collector was used.

src/appsignal/opentelemetry.py

@@ -3,6 +3,7 @@
 import os
 from typing import TYPE_CHECKING, Callable, List, Mapping, Union, cast
 
+import requests
 from opentelemetry import _logs as logs
 from opentelemetry import metrics, trace
 from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
@@ -243,9 +244,28 @@ def stop() -> None:
     _providers.clear()
 
 
+# Build the session an exporter sends its requests through, so that the
+# `http_proxy` option applies to them. Returns `None` when no proxy is
+# configured, which leaves the exporter to build its own session.
+#
+# Each exporter needs its own session, because they each send from their own
+# thread and a `requests` session is not thread safe.
+def _exporter_session(config: Config) -> requests.Session | None:
+    http_proxy = config.option("http_proxy")
+
+    if not http_proxy:
+        return None
+
+    session = requests.Session()
+    session.proxies = {"http": http_proxy, "https": http_proxy}
+    return session
+
+
 def _otlp_span_processor(config: Config) -> BatchSpanProcessor:
     otlp_exporter = OTLPSpanExporter(
         endpoint=f"{_opentelemetry_endpoint(config)}/v1/traces",
+        certificate_file=config.option("ca_file_path"),
+        session=_exporter_session(config),
     )
     return BatchSpanProcessor(otlp_exporter)
 
@@ -286,6 +306,8 @@ def _start_tracer(config: Config) -> None:
 def _start_metrics(config: Config) -> None:
     metric_exporter = OTLPMetricExporter(
         endpoint=f"{_opentelemetry_endpoint(config)}/v1/metrics",
+        certificate_file=config.option("ca_file_path"),
+        session=_exporter_session(config),
         preferred_temporality=METRICS_PREFERRED_TEMPORALITY,
     )
     metric_reader = PeriodicExportingMetricReader(
@@ -300,6 +322,8 @@ def _start_metrics(config: Config) -> None:
 def _start_logging(config: Config) -> None:
     log_exporter = OTLPLogExporter(
         endpoint=f"{_opentelemetry_endpoint(config)}/v1/logs",
+        certificate_file=config.option("ca_file_path"),
+        session=_exporter_session(config),
     )
     provider = LoggerProvider(resource=_resource(config))
     provider.add_log_record_processor(BatchLogRecordProcessor(log_exporter))

tests/test_opentelemetry.py

@@ -4,7 +4,15 @@
 from unittest.mock import Mock
 
 from appsignal.config import Config, Options
-from appsignal.opentelemetry import _providers, add_instrumentations, stop
+from appsignal.opentelemetry import (
+    _exporter_session,
+    _providers,
+    _start_logging,
+    _start_metrics,
+    _start_tracer,
+    add_instrumentations,
+    stop,
+)
 
 
 def raise_module_not_found_error(_config: Config) -> None:
@@ -93,3 +101,53 @@ def test_stop_without_started_providers():
     stop()
 
     assert _providers == []
+
+
+def test_exporter_session_without_a_proxy():
+    assert _exporter_session(Config()) is None
+
+
+def test_exporter_session_with_a_proxy():
+    config = Config(Options(http_proxy="http://proxy.example:3128"))
+
+    session = _exporter_session(config)
+
+    assert session is not None
+    assert session.proxies == {
+        "http": "http://proxy.example:3128",
+        "https": "http://proxy.example:3128",
+    }
+
+
+def test_exporter_sessions_are_not_shared():
+    # Each exporter sends from its own thread, and a session is not thread
+    # safe, so they must not share one.
+    config = Config(Options(http_proxy="http://proxy.example:3128"))
+
+    assert _exporter_session(config) is not _exporter_session(config)
+
+
+def test_exporters_are_given_the_ca_file_and_the_proxy(mocker):
+    span_exporter = mocker.patch("appsignal.opentelemetry.OTLPSpanExporter")
+    metric_exporter = mocker.patch("appsignal.opentelemetry.OTLPMetricExporter")
+    log_exporter = mocker.patch("appsignal.opentelemetry.OTLPLogExporter")
+
+    config = Config(
+        Options(
+            ca_file_path="/path/to/cacert.pem",
+            http_proxy="http://proxy.example:3128",
+            collector_endpoint="https://collector.example",
+        )
+    )
+
+    _start_tracer(config)
+    _start_metrics(config)
+    _start_logging(config)
+
+    for exporter in [span_exporter, metric_exporter, log_exporter]:
+        kwargs = exporter.call_args.kwargs
+        assert kwargs["certificate_file"] == "/path/to/cacert.pem"
+        assert kwargs["session"].proxies == {
+            "http": "http://proxy.example:3128",
+            "https": "http://proxy.example:3128",
+        }