[pull] master from DataDog:master

datadog_checks_base/changelog.d/22880.fixed

@@ -0,0 +1 @@
+Fix schema collection silently dropping all collected metadata when the last discovered database has no tables.

datadog_checks_base/datadog_checks/base/utils/db/schemas.py

@@ -75,17 +75,16 @@ def collect_schemas(self) -> bool:
                     continue
                 with self._get_cursor(database_name) as cursor:
                     # Get the next row from the cursor
-                    # We need to know when we've reached the last row so we can efficiently flush the last payload
-                    # without an empty final payload
                     next_row = self._get_next(cursor)
                     while next_row:
                         self._queued_rows.append(self._map_row(database, next_row))
                         self._total_rows_count += 1
                         # Because we're iterating over a cursor we need to try to get
                         # the next row to see if we've reached the last row
                         next_row = self._get_next(cursor)
-                        is_last_payload = database == databases[-1] and next_row is None
-                        self.maybe_flush(is_last_payload)
+                        self.maybe_flush(is_last_payload=False)
+                is_last_payload = database == databases[-1]
+                self.maybe_flush(is_last_payload)
                 self._log.debug("Completed collection of schemas for database %s", database_name)
         except Exception as e:
             status = "error"

datadog_checks_base/tests/base/utils/db/test_schemas.py

@@ -81,6 +81,36 @@ def kind(self):
         return "test_databases"
 
 
+class TestSchemaCollectorEmptyLastDb(TestSchemaCollector):
+    """Simulates multiple databases where the last one has no tables."""
+
+    __test__ = False
+
+    def __init__(self, check, config):
+        super().__init__(check, config)
+        self._db_rows = {
+            'db_with_tables': [{'table_name': 'users'}, {'table_name': 'orders'}],
+            'db_empty_last': [],
+        }
+        self._current_rows = []
+
+    def _get_databases(self):
+        return [{'name': 'db_with_tables'}, {'name': 'db_empty_last'}]
+
+    @contextmanager
+    def _get_cursor(self, database: str):
+        self._current_rows = list(self._db_rows.get(database, []))
+        self._row_index = 0
+        yield {}
+
+    def _get_next(self, _cursor):
+        if self._row_index < len(self._current_rows):
+            row = self._current_rows[self._row_index]
+            self._row_index += 1
+            return row
+        return None
+
+
 @pytest.mark.unit
 def test_schema_collector(aggregator):
     check = TestDatabaseCheck()
@@ -100,3 +130,37 @@ def test_schema_collector(aggregator):
     assert event['cloud_metadata'] == check.cloud_metadata
     assert event['metadata'][0]['name'] == 'test_database'
     assert event['metadata'][0]['tables'][0]['table_name'] == 'test_table'
+
+
+@pytest.mark.unit
+def test_schema_collector_empty_last_database(aggregator):
+    """Verify that queued rows are flushed even when the last database returns 0 rows."""
+    check = TestDatabaseCheck()
+    collector = TestSchemaCollectorEmptyLastDb(check, SchemaCollectorConfig())
+    collector.collect_schemas()
+
+    events = aggregator.get_event_platform_events("dbm-metadata")
+    assert len(events) == 1, "Expected 1 payload but got {}".format(len(events))
+    event = events[0]
+    assert len(event['metadata']) == 2
+    assert event['metadata'][0]['name'] == 'db_with_tables'
+    assert event['metadata'][0]['tables'][0]['table_name'] == 'users'
+    assert event['metadata'][1]['name'] == 'db_with_tables'
+    assert event['metadata'][1]['tables'][0]['table_name'] == 'orders'
+    assert event['collection_payloads_count'] == 1
+
+
+@pytest.mark.unit
+def test_schema_collector_chunk_size_flush(aggregator):
+    """Verify that collection_payloads_count is set even when all rows are flushed by chunk size."""
+    check = TestDatabaseCheck()
+    config = SchemaCollectorConfig()
+    config.payload_chunk_size = 1
+    collector = TestSchemaCollector(check, config)
+    collector.collect_schemas()
+
+    events = aggregator.get_event_platform_events("dbm-metadata")
+    # chunk_size=1 flushes the single row mid-loop, then a final payload marks the snapshot
+    assert len(events) == 2
+    assert 'collection_payloads_count' not in events[0]
+    assert events[-1]['collection_payloads_count'] == len(events)

disk/README.md

@@ -17,7 +17,7 @@ The disk check is included in the [Datadog Agent][1] package, so you don't need
 The Disk check is enabled by default, and the Agent collects metrics on all local partitions. To configure the check with custom options, edit the `disk.d/conf.yaml` file, in the `conf.d/` folder at the root of your [Agent's configuration directory][2]. See the [sample disk.d/conf.yaml][3] for all available configuration options.
 
 #### Note for Windows hosts
-There are three scenarios where the Disk check can be used:
+See the following scenarios where the Disk check can be used:
 
 1. Monitoring physical drives
 

spark/manifest.json

@@ -51,7 +51,7 @@
         "spec": "assets/configuration/spec.yaml"
       },
       "events": {
-        "creates_events": false
+        "creates_events": true
       },
       "metrics": {
         "prefix": "spark.",