azure_monitor_logs_ingestion_client_upload_custom_logs.py

1import os
2from azure.identity import DefaultAzureCredential
3from azure.monitor.ingestion import LogsIngestionClient
4
5# The Data Collection Endpoint (DCE) URI
6endpoint = os.environ["DATA_COLLECTION_ENDPOINT"]
7# The Immutable ID of the Data Collection Rule (DCR)
8dcr_id = os.environ["DATA_COLLECTION_RULE_ID"]
9# The name of the stream in the DCR that routes data to the target table
10stream_name = os.environ["STREAM_NAME"]
11
12credential = DefaultAzureCredential()
13client = LogsIngestionClient(endpoint, credential)
14
15# Data must match the structure defined in your DCR/Table
16body = [
17      {
18        "Time": "2023-01-01T00:00:00.000Z",
19        "Computer": "Computer1",
20        "AdditionalContext": "Sample log message"
21      },
22      {
23        "Time": "2023-01-01T00:00:01.000Z",
24        "Computer": "Computer2",
25        "AdditionalContext": "Another sample log message"
26      }
27    ]
28
29try:
30    client.upload(rule_id=dcr_id, stream_name=stream_name, logs=body)
31    print("Logs uploaded successfully.")
32except Exception as e:
33    print(f"Upload failed: {e}")