[MongoDB Atlas] Add support of mongodb database datastream

packages/mongodb_atlas/_dev/build/docs/README.md

@@ -6,25 +6,27 @@
 
 Use the MongoDB Atlas integration to:
 
-- Collect MongoDB Audit logs and Process metrics for comprehensive monitoring and analysis.
+- Collect MongoDB audit, database logs and process metrics for comprehensive monitoring and analysis.
 - Create informative visualizations to track usage trends, measure key metrics, and derive actionable business insights.
 - Set up alerts to minimize Mean Time to Detect (MTTD) and Mean Time to Resolve (MTTR) by quickly referencing relevant logs during troubleshooting.
 
 ## Data streams
 
 The MongoDB Atlas integration collects logs and metrics.
 
-Logs help you keep a record of events that happen on your machine. The `Log` data stream collected by MongoDB Atlas integration is `mongod_audit`.
+Logs help you keep a record of events that happen on your machine. The `Log` data stream collected by MongoDB Atlas integration are `mongod_audit` and `mongod_database`.
 
 Metrics give you insight into the statistics of the MongoDB Atlas. The `Metric` data stream collected by the MongoDB Atlas integration is `process` so that the user can monitor and troubleshoot the performance of the MongoDB Atlas instance.
 
-Data streams:
+Data stream:
 - `mongod_audit`: The auditing facility allows administrators and users to track system activity for deployments with multiple users and applications. Mongod Audit logs capture events related to database operations such as insertions, updates, deletions, user authentication, etc., occurring within the mongod instances.
 
+- `mongod_database`: This datastream collects a running log of events, including entries such as incoming connections, commands run, and issues encountered. Generally, database log messages are useful for diagnosing issues, monitoring your deployment, and tuning performance.
+
 - `process` : This data stream collects host metrics per process for all the hosts of the specified group. Metrics like measurements for the host, such as CPU usage, number of I/O operations and memory are available on this data stream.
 
 Note:
-- Users can monitor and see the log inside the ingested documents for MongoDB Atlas in the `logs-*` index pattern from `Discover`.
+- Users can monitor and see the logs and metrics inside the ingested documents for MongoDB Atlas in the `logs-*` index pattern from `Discover`.
 
 ## Prerequisites
 
@@ -40,7 +42,7 @@ You can store and search your data using Elasticsearch and visualize and manage
 
 ### Steps to obtain Public Key, Private Key and GroupId
 
-1. Generate programmatic API Keys with project owner permissions using the instructions in the Atlas [documentation](https://www.mongodb.com/docs/atlas/configure-api-access/#create-an-api-key-for-a-project). Then, copy the public key and private key. These serve the same function as a username and API Key respectively.
+1. Generate programmatic API keys with project owner permissions by following the instructions in the Atlas [documentation](https://www.mongodb.com/docs/atlas/configure-api-access/#create-an-api-key-for-a-project). Then, copy the public and private keys which function as a username and API key respectively.
 2. Enable Database Auditing for the Atlas project you want to monitor logs. You can follow the instructions provided in this Atlas [document](https://www.mongodb.com/docs/atlas/database-auditing/#procedure).
 3. You can find your Project ID (Group ID) in the Atlas UI. To do this, navigate to your project, click on Settings, and copy the Project ID (Group ID). You can also programmatically find it using the Atlas Admin API or Atlas CLI as described in this Atlas [document](https://www.mongodb.com/docs/atlas/app-services/apps/metadata/#find-a-project-id).
 
@@ -61,7 +63,7 @@ Note: Both of above attributes can be set by using `period` in configuration par
 6. Finally, save the integration.
 
 Note:
-- The `mongod_audit` data stream gathers historical data spanning the previous 30 minutes.
+- The `mongod_audit` and `mongod_database` data streams gathers historical data spanning the previous 30 minutes.
 - Mongod: Mongod is the primary daemon method for the MongoDB system. It helps in handling the data requests, managing the data access, performing background management operations, and other core database operations.
 
 ## Troubleshooting
@@ -85,6 +87,14 @@ This is the `mongod_audit` data stream. This data stream allows administrators a
 
 {{fields "mongod_audit"}}
 
+### Mongod Database
+
+This is the `mongod_database` data stream. This datastream collects a running log of events, including entries such as incoming connections, commands run, monitoring deployment, tuning performance and issues encountered.
+
+{{event "mongod_database"}}
+
+{{fields "mongod_database"}}
+
 ## Metrics reference
 
 ### Process

packages/mongodb_atlas/_dev/deploy/docker/mongodb_atlas/mongod_database_data.log

@@ -0,0 +1,5 @@
+{ "t": { "$date": "2024-03-20T19:17:06.188+00:00" }, "s": "W", "c": "CONTROL", "id": 22120, "ctx": "initandlisten", "msg": "Access control is not enabled for the database. Read and write access to data and configuration is unrestricted", "tags": [ "startupWarnings" ] }
+{ "t": { "$date": "2024-02-18T14:45:23.512+00:00" }, "s": "I", "c": "NETWORK", "id": 67890, "ctx": "conn123", "msg": "Client connection accepted", "tags": [ "connection" ] }
+{ "t": { "$date": "2024-02-22T10:20:05.933+00:00" }, "s": "E", "c": "STORAGE", "id": 13579, "ctx": "journal", "msg": "Journal file not found", "tags": [ "journalError" ] }
+{ "t": { "$date": "2024-02-25T16:55:36.124+00:00" }, "s": "I", "c": "NETWORK", "id": 24680, "ctx": "conn456", "msg": "Client disconnected", "tags": [ "connection" ] }
+{ "t": { "$date": "2024-02-28T09:12:50.007+00:00" }, "s": "E", "c": "QUERY", "id": 98765, "ctx": "queryExecutor", "msg": "Query execution failed", "tags": [ "queryError" ] }

packages/mongodb_atlas/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "0.0.3"
+  changes:
+    - description: MongoDB Atlas integration package with "mongod_database" data stream.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/9539
 - version: "0.0.2"
   changes:
     - description: MongoDB Atlas integration package with "mongod_audit" data stream.

packages/mongodb_atlas/data_stream/mongod_database/_dev/test/pipeline/test-common-config.yml

@@ -0,0 +1,5 @@
+dynamic_fields:
+  "event.ingested": ".*"
+fields:
+  tags:
+    - preserve_original_event

packages/mongodb_atlas/data_stream/mongod_database/_dev/test/pipeline/test-mongod-database.json

@@ -0,0 +1,20 @@
+{
+    "events": [
+        {
+            "@timestamp": "2024-02-08T06:20:49.729Z",
+            "message": "No data for given time period or host is unreachable"
+        },
+        {
+            "@timestamp": "2024-02-08T06:20:56.621Z",
+            "message": "{ \"t\": { \"$date\": \"2020-05-20T19:17:06.188+00:00\" }, \"s\": \"W\", \"c\": \"CONTROL\", \"id\": 22120, \"ctx\": \"initandlisten\", \"msg\": \"Access control is not enabled for the database. Read and write access to data and configuration is unrestricted\", \"tags\": [ \"startupWarnings\" ] }"
+        },
+        {
+            "@timestamp": "2024-02-08T06:21:56.001Z",
+            "message": "{ \"t\": { \"$date\": \"2024-02-25T16:55:36.124+00:00\" }, \"s\": \"I\", \"c\": \"NETWORK\", \"id\": 24680, \"ctx\": \"conn456\", \"msg\": \"Client disconnected\", \"tags\": [ \"connection\" ]}"
+        },
+        {
+            "@timestamp": "2024-02-08T06:22:52.121Z",
+            "message": "{ \"t\": { \"$date\": \"2024-02-28T09:12:50.007+00:00\" }, \"s\": \"E\", \"c\": \"QUERY\", \"id\": 98765, \"ctx\": \"queryExecutor\", \"msg\": \"Query execution failed\", \"tags\": [ \"queryError\" ]}"
+        }
+    ]
+}

packages/mongodb_atlas/data_stream/mongod_database/_dev/test/pipeline/test-mongod-database.json-expected.json

@@ -0,0 +1,119 @@
+{
+    "expected": [
+        null,
+        {
+            "@timestamp": "2020-05-20T19:17:06.188Z",
+            "ecs": {
+                "version": "8.11.0"
+            },
+            "event": {
+                "category": [
+                    "network",
+                    "database"
+                ],
+                "kind": "event",
+                "module": "mongodb_atlas",
+                "original": "{ \"t\": { \"$date\": \"2020-05-20T19:17:06.188+00:00\" }, \"s\": \"W\", \"c\": \"CONTROL\", \"id\": 22120, \"ctx\": \"initandlisten\", \"msg\": \"Access control is not enabled for the database. Read and write access to data and configuration is unrestricted\", \"tags\": [ \"startupWarnings\" ] }",
+                "type": [
+                    "access",
+                    "info"
+                ]
+            },
+            "log": {
+                "level": "warning"
+            },
+            "mongodb_atlas": {
+                "mongod_database": {
+                    "component": "CONTROL",
+                    "id": 22120,
+                    "message": "Access control is not enabled for the database. Read and write access to data and configuration is unrestricted",
+                    "tags": [
+                        "startupWarnings"
+                    ],
+                    "thread": {
+                        "name": "initandlisten"
+                    }
+                }
+            },
+            "tags": [
+                "preserve_original_event"
+            ]
+        },
+        {
+            "@timestamp": "2024-02-25T16:55:36.124Z",
+            "ecs": {
+                "version": "8.11.0"
+            },
+            "event": {
+                "category": [
+                    "network",
+                    "database"
+                ],
+                "kind": "event",
+                "module": "mongodb_atlas",
+                "original": "{ \"t\": { \"$date\": \"2024-02-25T16:55:36.124+00:00\" }, \"s\": \"I\", \"c\": \"NETWORK\", \"id\": 24680, \"ctx\": \"conn456\", \"msg\": \"Client disconnected\", \"tags\": [ \"connection\" ]}",
+                "type": [
+                    "access",
+                    "info"
+                ]
+            },
+            "log": {
+                "level": "informational"
+            },
+            "mongodb_atlas": {
+                "mongod_database": {
+                    "component": "NETWORK",
+                    "id": 24680,
+                    "message": "Client disconnected",
+                    "tags": [
+                        "connection"
+                    ],
+                    "thread": {
+                        "name": "conn456"
+                    }
+                }
+            },
+            "tags": [
+                "preserve_original_event"
+            ]
+        },
+        {
+            "@timestamp": "2024-02-28T09:12:50.007Z",
+            "ecs": {
+                "version": "8.11.0"
+            },
+            "event": {
+                "category": [
+                    "network",
+                    "database"
+                ],
+                "kind": "event",
+                "module": "mongodb_atlas",
+                "original": "{ \"t\": { \"$date\": \"2024-02-28T09:12:50.007+00:00\" }, \"s\": \"E\", \"c\": \"QUERY\", \"id\": 98765, \"ctx\": \"queryExecutor\", \"msg\": \"Query execution failed\", \"tags\": [ \"queryError\" ]}",
+                "type": [
+                    "access",
+                    "info"
+                ]
+            },
+            "log": {
+                "level": "error"
+            },
+            "mongodb_atlas": {
+                "mongod_database": {
+                    "component": "QUERY",
+                    "id": 98765,
+                    "message": "Query execution failed",
+                    "tags": [
+                        "queryError"
+                    ],
+                    "thread": {
+                        "name": "queryExecutor"
+                    }
+                }
+            },
+            "tags": [
+                "preserve_original_event"
+            ]
+        }
+    ]
+}

packages/mongodb_atlas/data_stream/mongod_database/_dev/test/system/test-default-config.yml

@@ -0,0 +1,13 @@
+vars:
+  url:
+    - http://{{Hostname}}:{{Port}}
+  public_key:
+    - admin
+  private_key:
+    - MongoDB@123
+data_stream:
+  vars:
+    groupId:
+      - mongodb-group1
+input: cel
+service: mongodbatlas

packages/mongodb_atlas/data_stream/mongod_database/agent/stream/input.yml.hbs

@@ -0,0 +1,108 @@
+config_version: 2
+interval: {{interval}}
+{{#if enable_request_tracer}}
+resource.tracer.filename: "../../logs/cel/http-request-trace-*.ndjson"
+{{/if}}
+{{#if ssl}}
+resource.ssl: {{ssl}}
+{{/if}}
+{{#if http_client_timeout}}
+resource.timeout: {{http_client_timeout}}
+{{/if}}
+tags:
+{{#if preserve_original_event}}
+  - preserve_original_event
+{{/if}}
+{{#each tags as |tag|}}
+  - {{tag}}
+{{/each}}
+{{#contains "forwarded" tags}}
+publisher_pipeline.disable_host: true
+{{/contains}}
+{{#if processors}}
+processors:
+{{processors}}
+{{/if}}
+auth.digest:
+  user: {{public_key}}
+  password: {{private_key}}
+resource.url: {{url}}
+state:
+  group_id: {{groupId}}
+  want_more: false
+  page_num: 1
+redact:
+  fields:
+    - password
+program: |
+  (
+    (
+      has(state.hostlist) && size(state.hostlist) > 0
+      ?
+        state
+      : 
+        (
+          state.page_num != 1 
+        ? 
+          state 
+        :
+          state.with({
+            "startDate": state.?cursor.last_timestamp.orValue(int(now - duration("30m"))),
+            "endDate": int(now)
+        })).as(state, state.with(request("GET", state.url + "/api/atlas/v2/groups/" + state.group_id + "/processes?pageNum=" + string(state.page_num) + "&itemsPerPage=100").with({
+            "Header": {
+              "Accept": ["application/vnd.atlas." + string(now.getFullYear()) + "-01-01+gzip"]
+            }
+        }).do_request().as(resp, 
+            bytes(resp.Body).decode_json().as(body, {
+              "hostlist": body.results.map(e, e.hostname),
+              "next": 0,
+              "page_num": body.links.exists_one(res,res.rel=="next") ? (int(state.page_num)+1) : 1
+            })
+        )))
+    ).as(state, state.next < size(state.hostlist)
+      ?
+        (
+          request("GET", state.url + "/api/atlas/v2/groups/" + state.group_id + "/clusters/" + state.hostlist[state.next] + "/logs/mongodb.gz?startDate=" + string(int(state.startDate)) + "&endDate=" + string(int(state.endDate))
+          ).with({
+            "Header": {
+              "Accept": ["application/vnd.atlas." + string(now.getFullYear()) + "-01-01+gzip"]
+            }
+          }).do_request().as(resp, (
+            resp.StatusCode == 200 && resp.ContentLength != 0
+            ?
+              string(bytes(resp.Body).mime('application/gzip')).as(body,{
+                "events": body.trim_space().split("\n").map(value,{"message":value, "host_name": state.hostlist[state.next]}),
+                "cursor": {
+                  "last_timestamp": state.endDate
+                },
+                "hostlist": (int(state.next)+1) < size(state.hostlist) ? state.hostlist : [],
+                "next": (int(state.next)+1) < size(state.hostlist) ? (int(state.next)+1) : 0,
+                "want_more": ((int(state.next)+1) < size(state.hostlist) || state.page_num != 1),
+                "page_num": state.page_num,
+                "startDate": state.startDate,
+                "endDate": state.endDate,
+                "group_id": state.group_id,
+              })
+            :
+              {
+                // If data is not available during this period, or if the host is unreachable, 
+                // an event will be generated with the following message, and this event will be dropped during pipeline processing.
+                // keep this message in a sync with pipeline as it is being used there.
+                "events": [{"message":"No data for given time period or host is unreachable"}],
+                "cursor": {
+                  "last_timestamp": state.endDate
+                },
+                "hostlist": (int(state.next)+1) < size(state.hostlist) ? state.hostlist : [],
+                "next": (int(state.next)+1) < size(state.hostlist) ? (int(state.next)+1) : 0,
+                "want_more": ((int(state.next)+1) < size(state.hostlist) || state.page_num != 1),
+                "page_num": state.page_num,
+                "startDate": state.startDate,
+                "endDate": state.endDate,
+                "group_id": state.group_id,
+              }
+        )))
+      :
+        {}
+    )
+  )