feat(http): report per-run HTTP request and cache-hit counts as analytics

airbyte_cdk/entrypoint.py

@@ -35,11 +35,16 @@
 )
 from airbyte_cdk.sources import Source
 from airbyte_cdk.sources.connector_state_manager import HashableStreamDescriptor
+from airbyte_cdk.sources.streams.http.cache_stats import (
+    HTTP_CACHE_STATS,
+    HttpCacheStatsSnapshot,
+)
 from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit, split_config
 
 # from airbyte_cdk.utils import PrintBuffer, is_cloud_environment, message_utils  # add PrintBuffer back once fixed
 from airbyte_cdk.utils import is_cloud_environment, message_utils
 from airbyte_cdk.utils.airbyte_secrets_utils import get_secrets, update_secrets
+from airbyte_cdk.utils.analytics_message import create_analytics_message
 from airbyte_cdk.utils.constants import ENV_REQUEST_CACHE_PATH
 from airbyte_cdk.utils.memory_monitor import MemoryMonitor
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
@@ -168,6 +173,12 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
             self.logger.setLevel(logging.INFO)
 
         source_spec: ConnectorSpecification = self.source.spec(self.logger)
+        # The counters are process-wide, and a process can run more than one command
+        # (`entrypoint_wrapper` in connector tests, the manifest server). Reporting a
+        # delta against this baseline keeps every run's numbers its own; resetting
+        # instead would race whichever command the manifest server is serving next to it.
+        http_cache_stats_baseline = HTTP_CACHE_STATS.snapshot()
+        closing = False
         try:
             with tempfile.TemporaryDirectory(
                 # Cleanup can fail on Windows due to file locks. Ignore if so,
@@ -212,11 +223,51 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                         )
                     else:
                         raise Exception("Unexpected command " + cmd)
+        except GeneratorExit:
+            # The consumer is closing us -- yielding anything now would turn into
+            # `RuntimeError: generator ignored GeneratorExit`, which the interpreter
+            # reports on stderr on top of whatever really went wrong.
+            closing = True
+            raise
         finally:
             yield from [
                 self.airbyte_message_to_string(queued_message)
                 for queued_message in self._emit_queued_messages(self.source)
             ]
+            if not closing:
+                yield from map(
+                    AirbyteEntrypoint.airbyte_message_to_string,
+                    self._http_cache_stats_messages(http_cache_stats_baseline),
+                )
+
+    @staticmethod
+    def _http_cache_stats_messages(
+        baseline: HttpCacheStatsSnapshot,
+    ) -> Iterable[AirbyteMessage]:
+        """Report how many requests the run made and how many its cache served.
+
+        A `requests_cache` hit never reaches the wire, so this is the only place
+        it can be observed from outside the process. Emitted as analytics, which
+        ride the protocol as TRACE messages on stdout: no `LOG_LEVEL=DEBUG` to
+        turn on, and one message pair per run rather than per-request log spam.
+
+        Reported as a delta against `baseline`, the snapshot taken when this run
+        started, so that a process running several commands attributes each
+        request to the run that made it rather than to every run after it.
+
+        Silent when the run made no requests, so `spec` -- and every non-HTTP
+        connector -- does not report a meaningless `0`. That silence is load
+        bearing for readers: absent means *not measured*, which is also what a
+        connector on an older CDK looks like, and is not the same as `0%`.
+        """
+        stats = HTTP_CACHE_STATS.snapshot()
+        requests_made = stats.requests - baseline.requests
+        if requests_made <= 0:
+            return
+        yield create_analytics_message("http-request-count", requests_made)
+        yield create_analytics_message(
+            "http-cache-hit-count", stats.cache_hits - baseline.cache_hits
+        )
 
     def check(
         self, source_spec: ConnectorSpecification, config: TConfig

airbyte_cdk/sources/streams/http/cache_stats.py

@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+"""Process-wide counters for HTTP requests and `requests_cache` hits.
+
+A `requests_cache` hit is served inside `Session.send()` and never reaches the
+wire, so nothing outside the connector process can observe it -- not a proxy, not
+the platform. The connector therefore has to count its own hits and report them,
+which is what these counters exist for; `AirbyteEntrypoint.run` turns the final
+snapshot into analytics messages at the end of every command.
+
+The counters are module-level because the thing being measured is, too: one
+`requests_cache` backend is shared by every stream of a run, and the question the
+numbers answer ("does this connector's caching work, and did this version
+regress it") is about the run rather than about any one stream or client. Being
+process-wide, they are cumulative: a reader wanting one run's numbers takes a
+snapshot when it starts and subtracts, which is what `AirbyteEntrypoint.run` does.
+
+Scope: every `HttpClient` in the process records here, but only
+`AirbyteEntrypoint.run` reports. Destinations (`Destination.run_cmd`), the
+manifest server, and the Connector Builder drive their commands without going
+through it, so they accumulate counts nothing reads. "No counts reported" means
+"a source connector run through the entrypoint, or nothing".
+
+`http-request-count` includes cache hits, so it is responses handled rather than
+wire flows; subtract `http-cache-hit-count` to get the number a proxy would see.
+"""
+
+from __future__ import annotations
+
+import threading
+from dataclasses import dataclass
+
+import requests
+
+
+@dataclass(frozen=True)
+class HttpCacheStatsSnapshot:
+    """The counters at one instant, detached from the lock that guards them."""
+
+    requests: int
+    cache_hits: int
+
+
+class HttpCacheStats:
+    """Requests made and requests served from the connector's own cache.
+
+    Guarded by a lock because concurrent sources read streams on a thread pool,
+    so `record_response` is called from several threads at once and `+= 1` on a
+    plain attribute would drop counts.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._requests = 0
+        self._cache_hits = 0
+
+    def record_response(self, response: requests.Response) -> None:
+        """Count one request, and one cache hit when the response was cached.
+
+        `from_cache` is set by `requests_cache.CacheMixin`, so it is present
+        exactly when caching is in play and absent -- counted as a live request
+        -- when it is not.
+        """
+        from_cache = bool(getattr(response, "from_cache", False))
+        with self._lock:
+            self._requests += 1
+            if from_cache:
+                self._cache_hits += 1
+
+    def snapshot(self) -> HttpCacheStatsSnapshot:
+        with self._lock:
+            return HttpCacheStatsSnapshot(requests=self._requests, cache_hits=self._cache_hits)
+
+    def reset(self) -> None:
+        """Zero the counters. For tests, which share one process across cases."""
+        with self._lock:
+            self._requests = 0
+            self._cache_hits = 0
+
+
+HTTP_CACHE_STATS = HttpCacheStats()

airbyte_cdk/sources/streams/http/http_client.py

@@ -25,6 +25,7 @@
 from airbyte_cdk.sources.http_config import MAX_CONNECTION_POOL_SIZE
 from airbyte_cdk.sources.message import MessageRepository
 from airbyte_cdk.sources.streams.call_rate import APIBudget, CachedLimiterSession, LimiterSession
+from airbyte_cdk.sources.streams.http.cache_stats import HTTP_CACHE_STATS
 from airbyte_cdk.sources.streams.http.error_handlers import (
     BackoffStrategy,
     DefaultBackoffStrategy,
@@ -349,6 +350,13 @@ def _send(
         except requests.RequestException as e:
             exc = e
 
+        if response is not None:
+            # Counted per `_send` call, so a retried request counts once per
+            # attempt. Cache hits are counted here too, so this is responses
+            # handled rather than wire flows: the figure comparable to what a
+            # proxy sees is `requests - cache_hits`, not `requests`.
+            HTTP_CACHE_STATS.record_response(response)
+
         error_resolution: ErrorResolution = self._error_handler.interpret_response(
             response if response is not None else exc
         )

cdk-migrations.md

@@ -1,5 +1,23 @@
 # CDK Migration Guide
 
+## Upgrading to the CDK that reports HTTP cache stats
+
+Every source that makes at least one HTTP request through `HttpClient` now ends each
+`spec`/`check`/`discover`/`read` with two extra `TRACE`/`ANALYTICS` messages:
+`http-request-count` and `http-cache-hit-count`. A run that made no requests emits
+nothing, so absent still means *not measured* rather than `0`.
+
+Migration steps: connector tests that assert an exact protocol message count need
+updating. `EntrypointOutput.trace_messages` grows by two for any run that made a
+request, including runs under `HttpMocker`, which leaves `Session.send` in the call
+path. Assertions of the shape `assert len(output.trace_messages) == N` are the ones
+that break; `> 0` and stream-status filters are unaffected. Nothing else changes --
+no records, state, schemas, or exit codes.
+
+Note on the numbers themselves: `http-request-count` counts responses handled, cache
+hits included, so the figure comparable to a proxy's wire-flow count is
+`http-request-count - http-cache-hit-count`.
+
 ## Upgrading to 7.0.0
 
 [Version 7.0.0](https://github.com/airbytehq/airbyte-python-cdk/releases/tag/v7.0.0) of the CDK migrates the CDK to the Concurrent CDK by removing some of the Declarative CDK concepts that are better expressed in the Concurrent CDK or that are outright incompatible with it. This changes mostly impact the Python implementations although the concept of CustomIncrementalSync has been removed from the declarative language as well.

unit_tests/sources/streams/http/test_cache_stats.py

@@ -0,0 +1,79 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import threading
+
+import requests
+
+from airbyte_cdk.sources.streams.http.cache_stats import HttpCacheStats, HttpCacheStatsSnapshot
+
+
+def _response(*, from_cache: bool | None = None) -> requests.Response:
+    """A response with, or deliberately without, the `requests_cache` marker.
+
+    `from_cache` absent is the case that matters most: it is what an uncached
+    session produces, and counting it as a hit would report every connector that
+    does no caching as caching perfectly.
+    """
+    response = requests.Response()
+    response.status_code = 200
+    if from_cache is not None:
+        response.from_cache = from_cache  # type: ignore[attr-defined]
+    return response
+
+
+def test_a_fresh_counter_reports_nothing() -> None:
+    assert HttpCacheStats().snapshot() == HttpCacheStatsSnapshot(requests=0, cache_hits=0)
+
+
+def test_only_a_from_cache_response_counts_as_a_hit() -> None:
+    stats = HttpCacheStats()
+
+    stats.record_response(_response())
+    stats.record_response(_response(from_cache=False))
+    stats.record_response(_response(from_cache=True))
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=3, cache_hits=1)
+
+
+def test_a_snapshot_does_not_move_under_the_reader() -> None:
+    """The snapshot is a detached value, so a later request cannot backdate it."""
+    stats = HttpCacheStats()
+    stats.record_response(_response(from_cache=True))
+
+    taken = stats.snapshot()
+    stats.record_response(_response(from_cache=True))
+
+    assert taken == HttpCacheStatsSnapshot(requests=1, cache_hits=1)
+
+
+def test_concurrent_recording_loses_no_counts() -> None:
+    """Concurrent sources read streams on a thread pool, so this is the real shape.
+
+    `+= 1` on a plain attribute is not atomic under free-threaded CPython and is
+    only accidentally so under the GIL, which is why the counters take a lock.
+    """
+    stats = HttpCacheStats()
+    threads = [
+        threading.Thread(
+            target=lambda: [stats.record_response(_response(from_cache=True)) for _ in range(200)]
+        )
+        for _ in range(8)
+    ]
+
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=1600, cache_hits=1600)
+
+
+def test_reset_zeroes_the_counters() -> None:
+    stats = HttpCacheStats()
+    stats.record_response(_response(from_cache=True))
+
+    stats.reset()
+
+    assert stats.snapshot() == HttpCacheStatsSnapshot(requests=0, cache_hits=0)