[pull] master from DataDog:master

cisco_duo/README.md

@@ -9,6 +9,8 @@ This integration ingests the following logs:
 - Telephony
 - Offline Enrollment
 
+**Note**: The Administrator and Offline Enrollment log endpoints and dashboards are being deprecated. Update any custom assets using these endpoints to use the Activity Logs endpoint instead to avoid service disruption.
+
 The Cisco Duo integration seamlessly collects multi-factor authentication (MFA) and secure access logs, channeling them into Datadog for analysis. Leveraging the built-in logs pipeline, these logs are parsed and enriched, enabling effortless search and analysis. The integration provides insight into fraud authentication events, authentication activity timeline, locations of accessed, authentication devices, and many more through the out-of-the-box dashboards.
 
 ## Setup

ddev/changelog.d/24315.fixed

@@ -0,0 +1 @@
+Fix the EventBusOrchestrator overwriting a task's timeout cancellation reason with a bare cancel during shutdown cleanup.

ddev/src/ddev/event_bus/orchestrator.py

@@ -28,6 +28,15 @@
 DEFAULT_ORCHESTRATOR_MAX_TIMEOUT = 300.0
 
 
+class OrchestratorTimeout(Exception):
+    """Internal signal raised when ``max_timeout`` is exceeded.
+
+    Caught by :meth:`EventBusOrchestrator.process_messages` so the timeout reason can be
+    handed to the single cancellation loop in its ``finally`` block, instead of cancelling
+    tasks from two places.
+    """
+
+
 @dataclass
 class BaseMessage:
     """
@@ -306,6 +315,7 @@ async def process_messages(self):
         get_task = asyncio.create_task(self._queue.get())
 
         start_time = asyncio.get_running_loop().time()
+        cancel_reason: str | None = None
 
         try:
             while not await self.__should_stop(start_time, running_tasks, get_task):
@@ -321,13 +331,17 @@ async def process_messages(self):
                         break
 
                 self.__process_finished_tasks(done, current_get_task, running_tasks)
+        except OrchestratorTimeout as timeout:
+            cancel_reason = str(timeout)
         finally:
             # If we exit the loop and tasks are still running (e.g. timeout or forced break),
             # we must clean them up before returning to ensure finalize() runs in a safe state.
+            # This is the single place that cancels ``running_tasks``, so every remaining task
+            # is cancelled exactly once, with the reason (if any) attached.
             if running_tasks:
                 self._logger.info("Cancelling %s remaining tasks...", len(running_tasks))
                 for task in running_tasks:
-                    task.cancel()
+                    task.cancel(cancel_reason)
 
                 # Wait for them to actually finish cancelling
                 await asyncio.wait(running_tasks)
@@ -352,7 +366,7 @@ async def __should_stop(self, start_time: float, running_tasks: set[asyncio.Task
                 len(running_tasks),
             )
             get_task.cancel()
-            return True
+            raise OrchestratorTimeout(f"Orchestrator exceeded max_timeout of {self._max_timeout}s")
 
         # Check exit condition: empty queue (implied by get_task not done) and no running processors
         if not running_tasks and not get_task.done() and self._queue.empty():

ddev/tests/event_bus/test_event_bus.py

@@ -8,7 +8,7 @@
 import math
 import time
 from collections.abc import Generator
-from contextlib import AbstractContextManager, contextmanager
+from contextlib import AbstractContextManager, contextmanager, suppress
 from contextlib import nullcontext as does_not_raise
 from dataclasses import dataclass
 
@@ -633,6 +633,84 @@ async def process_message(self, message: Memo):
     assert slow_processor.cancelled
 
 
+def test_max_timeout_interruption_preserves_cancellation_reason(orchestrator: MockOrchestrator):
+    # A timed-out task's CancelledError should carry the timeout reason, not be
+    # silently re-cancelled without a message by the generic shutdown cleanup.
+    orchestrator._max_timeout = 0.5
+    orchestrator._grace_period = 5.0
+
+    class SlowProcessor(AsyncProcessor[Memo]):
+        def __init__(self, name: str):
+            super().__init__(name)
+            self.cancellation_message: str | None = None
+
+        async def process_message(self, message: Memo):
+            try:
+                await asyncio.sleep(2.0)
+            except asyncio.CancelledError as e:
+                self.cancellation_message = str(e)
+                raise
+
+    slow_processor = SlowProcessor("slow_processor")
+    orchestrator.register_processor(slow_processor, [Memo])
+
+    orchestrator.submit_message(Memo("slow_memo"))
+
+    with assert_time(0.5, 1.5):
+        orchestrator.run()
+
+    assert slow_processor.cancellation_message
+    assert "max_timeout" in slow_processor.cancellation_message
+
+
+def test_fatal_processing_error_cancels_task_that_already_swallowed_a_cancellation(
+    orchestrator: MockOrchestrator,
+):
+    # A task that previously caught and suppressed a CancelledError (without calling
+    # Task.uncancel()) still has a pending cancellation count. The shutdown cleanup
+    # must not treat that as "already being cancelled" and skip it: it should still be
+    # cancelled (and stopped) when a sibling processor's FatalProcessingError shuts
+    # the bus down.
+    class StubbornAnalyst(AsyncProcessor[TaskAssignment]):
+        def __init__(self, name: str):
+            super().__init__(name)
+            self.cancelled_again = False
+            self.finished = False
+
+        async def process_message(self, message: TaskAssignment):
+            asyncio.current_task().cancel()  # simulate an earlier, unrelated cancellation
+            with suppress(asyncio.CancelledError):
+                await asyncio.sleep(0)
+
+            try:
+                await asyncio.sleep(10)
+            except asyncio.CancelledError:
+                self.cancelled_again = True
+                raise
+            self.finished = True
+
+    stubborn = StubbornAnalyst("stubborn")
+    orchestrator.register_processor(stubborn, [TaskAssignment])
+
+    original_on_message = orchestrator.on_message_received
+
+    async def on_message_fatal(message: BaseMessage):
+        await original_on_message(message)
+        if message.id == "fatal_msg":
+            raise FatalProcessingError("Fatal error triggered")
+
+    orchestrator.on_message_received = on_message_fatal  # type: ignore[method-assign]
+
+    orchestrator.submit_message(TaskAssignment("stubborn_msg", task_type="slow"))
+    orchestrator.submit_message(Memo("fatal_msg"))
+
+    with assert_time(0, 2.0), pytest.raises(FatalProcessingError, match="Fatal error triggered"):
+        orchestrator.run()
+
+    assert stubborn.cancelled_again
+    assert not stubborn.finished
+
+
 def test_sync_processor_thread_execution(orchestrator: MockOrchestrator, secretary: Secretary):
     import threading
 

snmp/changelog.d/24107.fixed

@@ -0,0 +1 @@
+Fix Cisco ASAv memory coverage in cisco-firepower-asa SNMP profile by replacing hardcoded single-pool scalar OIDs with a full table walk of cempMemPoolTable tagged by pool name.

snmp/datadog_checks/snmp/data/default_profiles/cisco-firepower-asa.yaml

@@ -39,14 +39,21 @@ metrics:
         tag: cpu
   - MIB: CISCO-ENHANCED-MEMPOOL-MIB
     metric_type: gauge
-    symbol:
-      name: memory.used
-      OID: 1.3.6.1.4.1.9.9.221.1.1.1.1.18.1.1  # cempMemPoolHCUsed.1.1
-  - MIB: CISCO-ENHANCED-MEMPOOL-MIB
-    metric_type: gauge
-    symbol:
-      name: memory.free
-      OID: 1.3.6.1.4.1.9.9.221.1.1.1.1.20.1.1  # cempMemPoolHCFree.1.1
+    table:
+      OID: 1.3.6.1.4.1.9.9.221.1.1.1
+      name: cempMemPoolTable
+    symbols:
+      - OID: 1.3.6.1.4.1.9.9.221.1.1.1.1.18
+        name: memory.used
+      - OID: 1.3.6.1.4.1.9.9.221.1.1.1.1.20
+        name: memory.free
+    metric_tags:
+      - tag: mem_pool_name
+        symbol:
+          OID: 1.3.6.1.4.1.9.9.221.1.1.1.1.3
+          name: cempMemPoolName
+      - tag: mem_pool_index
+        index: 1
   - MIB: CISCO-REMOTE-ACCESS-MONITOR-MIB
     symbol:
       OID: 1.3.6.1.4.1.9.9.392.1.4.1.2.0

snmp/tests/compose/data/cisco-firepower-asa.snmprec

@@ -244,7 +244,7 @@
 1.3.6.1.4.1.9.9.109.1.4.1.1.13.33481.29654.61594|70|4963720931156138597
 1.3.6.1.4.1.9.9.221.1.1.1.1.1.1.1|2|22
 1.3.6.1.4.1.9.9.221.1.1.1.1.2.1.1|2|12
-1.3.6.1.4.1.9.9.221.1.1.1.1.3.1.1|4x|717561696e746c79206275742062757420717561696e746c79207a6f6d626965732074686569722074686569722064726976696e67
+1.3.6.1.4.1.9.9.221.1.1.1.1.3.1.1|4x|53797374656d206d656d6f7279
 1.3.6.1.4.1.9.9.221.1.1.1.1.4.1.1|6|1.3.6.1.3.101.206.92
 1.3.6.1.4.1.9.9.221.1.1.1.1.5.1.1|2|7
 1.3.6.1.4.1.9.9.221.1.1.1.1.6.1.1|2|1

snmp/tests/test_e2e_core_profiles/test_profile_cisco_firepower_asa.py

@@ -48,9 +48,13 @@ def test_e2e_profile_cisco_firepower_asa(dd_agent_check):
     aggregator.assert_metric('snmp.crasNumSessions', metric_type=aggregator.GAUGE, tags=common_tags)
     aggregator.assert_metric('snmp.crasNumSetupFailInsufResources', metric_type=aggregator.COUNT, tags=common_tags)
     aggregator.assert_metric('snmp.crasNumUsers', metric_type=aggregator.GAUGE, tags=common_tags)
-    aggregator.assert_metric('snmp.memory.free', metric_type=aggregator.GAUGE, tags=common_tags)
-    aggregator.assert_metric('snmp.memory.usage', metric_type=aggregator.GAUGE, tags=common_tags)
-    aggregator.assert_metric('snmp.memory.used', metric_type=aggregator.GAUGE, tags=common_tags)
+    mem_tag_rows = [
+        ['mem_pool_index:1', 'mem_pool_name:System memory'],
+    ]
+    for tag_row in mem_tag_rows:
+        aggregator.assert_metric('snmp.memory.free', metric_type=aggregator.GAUGE, tags=common_tags + tag_row)
+        aggregator.assert_metric('snmp.memory.used', metric_type=aggregator.GAUGE, tags=common_tags + tag_row)
+        aggregator.assert_metric('snmp.memory.usage', metric_type=aggregator.GAUGE, tags=common_tags + tag_row)
     tag_rows = [
         ['cpu:34646'],
         ['cpu:7885'],