review fixes

Author: steveny91

argocd/assets/configuration/spec.yaml

@@ -60,7 +60,7 @@ files:
       hidden: true
       fleet_configurable: false
       description: |
-        Collect the Argo CD Prometheus/OpenMetrics endpoints, the check's
+        Collect the Argo CD Prometheus (OpenMetrics) endpoints, the check's
         default metrics. Enabled by default. Set to false to run only the
         generic resources collector (``collect_genresources``) without scraping
         metrics; in that mode the ``*_endpoint`` options are not required.
@@ -134,7 +134,10 @@ files:
       description: |
         Seconds between full Application scrapes, in both streaming and polling
         modes. A full scrape re-submits every Application to refresh its
-        time-to-live (TTL) and backfill anything the stream missed. Minimum of 1.
+        time-to-live (TTL) and backfill anything the stream missed. While
+        streaming, the dedup poll is disabled, so during a stream outage a newly
+        created Application is not collected until the next full scrape -- this
+        value is the worst-case discovery lag. Minimum of 1.
       value:
         type: integer
         example: 600
@@ -232,6 +235,19 @@ files:
         type: integer
         example: 30
         minimum: 1
+    - name: genresources_stream_read_timeout_seconds
+      hidden: true
+      fleet_configurable: false
+      description: |
+        Socket read timeout, in seconds, for the application watch stream. Argo CD
+        does not send keepalives on an idle stream, so this is effectively how long
+        an idle but healthy connection is held before it is treated as dead and
+        reconnected. Larger values reduce reconnect churn on low-traffic instances;
+        smaller values detect a silently dropped connection sooner. Minimum of 1.
+      value:
+        type: integer
+        example: 300
+        minimum: 1
     - template: instances/openmetrics
       overrides:
         openmetrics_endpoint.required: false

argocd/datadog_checks/argocd/config_models/defaults.py

@@ -92,6 +92,10 @@ def instance_genresources_stream_backoff_max_seconds():
     return 30
 
 
+def instance_genresources_stream_read_timeout_seconds():
+    return 300
+
+
 def instance_genresources_ttl_seconds():
     return 1800
 

argocd/datadog_checks/argocd/config_models/instance.py

@@ -126,6 +126,7 @@ class InstanceConfig(BaseModel):
     genresources_repository_scrape_interval_seconds: Optional[int] = Field(None, ge=1)
     genresources_stream_applications_enabled: Optional[bool] = None
     genresources_stream_backoff_max_seconds: Optional[int] = Field(None, ge=1)
+    genresources_stream_read_timeout_seconds: Optional[int] = Field(None, ge=1)
     genresources_ttl_seconds: Optional[int] = Field(None, ge=1)
     headers: Optional[MappingProxyType[str, Any]] = None
     histogram_buckets_as_distributions: Optional[bool] = None

argocd/datadog_checks/argocd/resources.py

@@ -225,6 +225,7 @@ def __init__(self, check: "ArgocdCheck") -> None:
         self._submitted: dict[str, str] = {}
         self._stream_enabled: bool = bool(config.genresources_stream_applications_enabled)
         self._backoff_max: int = config.genresources_stream_backoff_max_seconds
+        self._stream_read_timeout: int = config.genresources_stream_read_timeout_seconds
         self._app_poll_interval: int = config.genresources_application_poll_interval_seconds
         self._app_full_scrape_interval: int = config.genresources_application_full_scrape_interval_seconds
         self._cluster_scrape_interval: int = config.genresources_cluster_scrape_interval_seconds
@@ -266,8 +267,8 @@ def collect(self) -> None:
             return
         now = int(time.time())
         if self._stream_enabled:
-            self._ensure_listener()
-            connected = self._listener is not None and self._listener.is_connected()
+            listener = self._ensure_listener()
+            connected = listener is not None and listener.is_connected()
             self.check.gauge(GENRESOURCES_STREAM_UP_METRIC, 1 if connected else 0)
         self._collect_due_types(now)
 
@@ -299,17 +300,18 @@ def _collect_due_types(self, now: int) -> None:
             self._last_repository_scrape = now
             self._collect_type(REPOSITORY_SPEC, seen_at=now, expire_at=expire_at, force_full=True)
 
-    def _ensure_listener(self) -> None:
+    def _ensure_listener(self) -> ArgocdApplicationStreamListener | None:
         with self._listener_lock:
             if self._stopped:
-                return
+                return None
             if self._listener is None:
                 self._listener = ArgocdApplicationStreamListener(
                     self.check,
                     self,
                     endpoint=self._endpoint,
                     auth_token=self._auth_token,
                     backoff_max_seconds=self._backoff_max,
+                    read_timeout_seconds=self._stream_read_timeout,
                     http=RequestsWrapper(
                         self.check.instance,
                         self.check.init_config,
@@ -319,6 +321,7 @@ def _ensure_listener(self) -> None:
                 )
             if not self._listener.is_alive():
                 self._listener.start()
+            return self._listener
 
     def stop(self) -> None:
         """Signal the stream listener to stop; must not block, per AgentCheck.cancel()'s contract."""
@@ -329,7 +332,10 @@ def stop(self) -> None:
             listener.cancel()
 
     def emit_stream_application(self, application: dict) -> None:
-        """Emit a single application received from the stream (ADDED/MODIFIED) through the shared pipeline."""
+        """Emit a single application received from the stream (ADDED/MODIFIED) through the shared pipeline.
+
+        Mutates ``application`` in place (credential sanitization); the caller must not reuse the dict after.
+        """
         seen_at = int(time.time())
         self._emit_item(
             application, APPLICATION_SPEC, seen_at=seen_at, expire_at=seen_at + self._ttl_seconds, force_full=False
@@ -382,13 +388,18 @@ def _collect_type(self, spec: ResourceTypeSpec, *, seen_at: int, expire_at: int,
             cache_key = self._emit_item(item, spec, seen_at=seen_at, expire_at=expire_at, force_full=force_full)
             if cache_key is not None:
                 seen.add(cache_key)
+        # A stream frame that lands between _fetch above and this prune isn't in `seen`, so its fresh cache
+        # entry is dropped here. Harmless: the app was already submitted; it just gets one redundant re-submit
+        # on its next stream frame. Holding the lock across the fetch would stall real-time emits instead.
         namespace = f"{spec.resource_type}{KEY_SEPARATOR}"
         with self._submitted_lock:
             self._submitted = {k: v for k, v in self._submitted.items() if not k.startswith(namespace) or k in seen}
 
     def _fetch(self, api_path: str) -> list[dict]:
         url = self._endpoint.rstrip("/") + api_path
-        response = self.check.http.get(url, headers=auth_headers(self._auth_token))
+        # extra_headers (not headers) so a dedicated token is *merged* into the wrapper's configured headers;
+        # passing headers= would shadow them and drop inherited auth when genresources_auth_token is unset.
+        response = self.check.http.get(url, extra_headers=auth_headers(self._auth_token))
         response.raise_for_status()
         payload = response.json()
         return list(payload.get("items") or [])

argocd/datadog_checks/argocd/stream_listener.py

@@ -26,7 +26,6 @@
 
 STREAM_PATH = "/api/v1/stream/applications"
 CONNECT_TIMEOUT_SECONDS = 10
-READ_TIMEOUT_SECONDS = 60
 INITIAL_BACKOFF_SECONDS = 1
 
 
@@ -39,6 +38,10 @@ class ArgocdApplicationStreamListener:
     shutdown path is ``cancel()`` (signal the loop and close the socket to unblock ``iter_lines``); it does not
     block, and callers may ``join()`` afterwards to wait. ``daemon=True`` is only a backstop for a hard
     interpreter exit that never calls ``cancel()``.
+
+    Metrics (``count``) and generic resources (``submit_generic_resource`` via the collector) are submitted
+    directly from this thread; the Datadog aggregator tolerates concurrent submission (same pattern as
+    ``DBMAsyncJob``), and submitting inline is what makes stream updates near-real-time.
     """
 
     def __init__(
@@ -49,6 +52,7 @@ def __init__(
         endpoint: str,
         auth_token: str | None,
         backoff_max_seconds: int,
+        read_timeout_seconds: int,
         http: "RequestsWrapper",
     ) -> None:
         self.check = check
@@ -57,6 +61,7 @@ def __init__(
         self._endpoint = endpoint.rstrip("/")
         self._auth_token = auth_token
         self._backoff_max = max(1, backoff_max_seconds)
+        self._read_timeout = max(1, read_timeout_seconds)
         self._stop = threading.Event()
         self._connected = threading.Event()
         self._thread: threading.Thread | None = None
@@ -79,6 +84,9 @@ def is_connected(self) -> bool:
     def cancel(self) -> None:
         """Signal the loop to stop and close the active connection to unblock ``iter_lines``. Does not block."""
         self._stop.set()
+        # If cancel lands in the narrow window between _stream_once's get() returning and its
+        # ``self._response = response``, we see None and skip close(); the thread still exits on its next
+        # _stop check, at worst after the read timeout. Bounded shutdown latency on a daemon thread, not a leak.
         response = self._response
         if response is not None:
             try:
@@ -119,8 +127,10 @@ def _stream_once(self) -> bool:
         url = self._endpoint + STREAM_PATH
         kwargs: dict = {
             "stream": True,
-            "timeout": (CONNECT_TIMEOUT_SECONDS, READ_TIMEOUT_SECONDS),
-            "headers": auth_headers(self._auth_token),
+            "timeout": (CONNECT_TIMEOUT_SECONDS, self._read_timeout),
+            # extra_headers (not headers): merge the token into the wrapper's configured headers rather than
+            # shadowing them, so inherited auth survives when genresources_auth_token is unset.
+            "extra_headers": auth_headers(self._auth_token),
         }
         got_data = False
         response = None

argocd/tests/test_argocd.py

@@ -105,8 +105,8 @@ def test_app_controller_service_check(dd_run_check, aggregator, mock_http_respon
 
 
 def test_both_collectors_disabled_raises(dd_run_check):
+    check = ArgocdCheck('argocd', {}, [{"collect_openmetrics": False, "collect_genresources": False}])
     with pytest.raises(Exception, match="Enable at least one of"):
-        check = ArgocdCheck('argocd', {}, [{"collect_openmetrics": False, "collect_genresources": False}])
         dd_run_check(check)
 
 

argocd/tests/test_resources.py

@@ -14,6 +14,7 @@
     GENRESOURCES_API_UP_METRIC,
     REPOSITORY_INCLUDE,
 )
+from datadog_checks.base.utils.http import RequestsWrapper
 from datadog_checks.dev.http import MockResponse
 
 from .common import _check as _make_check
@@ -636,3 +637,23 @@ def test_collector_warns_when_exclude_empties_a_type(caplog):
 def test_collector_warns_when_ttl_shorter_than_longest_scrape_interval(caplog):
     _check(genresources_ttl_seconds=100, genresources_application_full_scrape_interval_seconds=600)
     assert any("shorter than the longest scrape interval" in rec.message for rec in caplog.records)
+
+
+def test_fetch_adds_bearer_via_extra_headers_so_configured_auth_is_preserved():
+    check = _check(genresources_auth_token="tok")
+    with patch.object(RequestsWrapper, "get", return_value=MockResponse(json_data={"items": []})) as get:
+        check._resource_collector._fetch("/api/v1/applications")
+
+    kwargs = get.call_args.kwargs
+    assert kwargs.get("extra_headers") == {"Authorization": "Bearer tok"}  # merged into the wrapper's headers
+    assert "headers" not in kwargs  # never pass `headers`; it would shadow the wrapper's configured auth
+
+
+def test_fetch_inherits_wrapper_auth_when_no_genresources_token():
+    check = _check(genresources_auth_token=None)  # rely on the instance's HTTP auth (headers / auth_token)
+    with patch.object(RequestsWrapper, "get", return_value=MockResponse(json_data={"items": []})) as get:
+        check._resource_collector._fetch("/api/v1/applications")
+
+    kwargs = get.call_args.kwargs
+    assert "headers" not in kwargs  # must not clobber the wrapper's configured headers
+    assert not kwargs.get("extra_headers")  # nothing added -> the inherited auth headers survive

argocd/tests/test_resources_stream.py

@@ -40,6 +40,7 @@ def _listener(collector, *, check=None) -> ArgocdApplicationStreamListener:
         endpoint=GENRESOURCES_ENDPOINT,
         auth_token=None,
         backoff_max_seconds=10,
+        read_timeout_seconds=60,
         http=Mock(),
     )
 
@@ -174,6 +175,17 @@ def test_ensure_listener_gives_the_listener_its_own_http_session():
     assert http is not check.http  # a dedicated RequestsWrapper, not the check's shared session
 
 
+def test_ensure_listener_passes_the_configured_read_timeout():
+    check = _check(genresources_stream_applications_enabled=True, genresources_stream_read_timeout_seconds=123)
+    collector = check._resource_collector
+
+    with patch("datadog_checks.argocd.resources.ArgocdApplicationStreamListener") as listener_cls:
+        listener_cls.return_value.is_alive.return_value = False
+        collector._ensure_listener()
+
+    assert listener_cls.call_args.kwargs["read_timeout_seconds"] == 123
+
+
 def test_collect_with_stream_emits_stream_up_one_when_the_listener_is_connected(aggregator):
     check = _check(genresources_stream_applications_enabled=True)
     collector = check._resource_collector
@@ -263,7 +275,13 @@ def close(self):
     http = Mock()
     http.get.return_value = _Resp()
     listener = ArgocdApplicationStreamListener(
-        check, collector, endpoint=GENRESOURCES_ENDPOINT, auth_token=None, backoff_max_seconds=10, http=http
+        check,
+        collector,
+        endpoint=GENRESOURCES_ENDPOINT,
+        auth_token=None,
+        backoff_max_seconds=10,
+        read_timeout_seconds=60,
+        http=http,
     )
 
     with patch.object(check, "submit_generic_resource") as submit:
@@ -293,7 +311,13 @@ def close(self):
     http = Mock()
     http.get.return_value = _Resp()
     listener = ArgocdApplicationStreamListener(
-        check, collector, endpoint=GENRESOURCES_ENDPOINT, auth_token=None, backoff_max_seconds=10, http=http
+        check,
+        collector,
+        endpoint=GENRESOURCES_ENDPOINT,
+        auth_token=None,
+        backoff_max_seconds=10,
+        read_timeout_seconds=60,
+        http=http,
     )
 
     with patch.object(check, "submit_generic_resource"):
@@ -302,6 +326,39 @@ def close(self):
     assert got_data is True  # data arrived before the drop; _stream_once caught the error and reported it
 
 
+def test_stream_once_inherits_wrapper_auth_when_no_token():
+    check = _check(genresources_stream_applications_enabled=True, genresources_auth_token=None)
+    collector = check._resource_collector
+
+    class _Resp:
+        def raise_for_status(self):
+            pass
+
+        def iter_lines(self):
+            return iter(())
+
+        def close(self):
+            pass
+
+    http = Mock()
+    http.get.return_value = _Resp()
+    listener = ArgocdApplicationStreamListener(
+        check,
+        collector,
+        endpoint=GENRESOURCES_ENDPOINT,
+        auth_token=None,
+        backoff_max_seconds=10,
+        read_timeout_seconds=60,
+        http=http,
+    )
+
+    listener._stream_once()
+
+    kwargs = http.get.call_args.kwargs
+    assert "headers" not in kwargs  # must not clobber the wrapper's configured auth
+    assert not kwargs.get("extra_headers")  # no token -> nothing added, inherited auth survives
+
+
 def test_handle_line_emits_events_received_metric_matching_metadata(aggregator):
     check = _check(genresources_stream_applications_enabled=True)
     collector = check._resource_collector
@@ -357,7 +414,13 @@ def close(self):
     http = Mock()
     http.get.return_value = _Resp()
     listener = ArgocdApplicationStreamListener(
-        check, collector, endpoint=GENRESOURCES_ENDPOINT, auth_token=None, backoff_max_seconds=1, http=http
+        check,
+        collector,
+        endpoint=GENRESOURCES_ENDPOINT,
+        auth_token=None,
+        backoff_max_seconds=1,
+        read_timeout_seconds=60,
+        http=http,
     )
 
     listener.start()