azure_monitor_ingestion_custom_logs_upload_quickstart.py

1import os
2from azure.identity import DefaultAzureCredential
3from azure.monitor.ingestion import LogsIngestionClient
4from azure.core.exceptions import HttpResponseError
5
6def main():
7    # To use the client, you need the endpoint of your Data Collection Endpoint (DCE),
8    # the Immutable ID of your Data Collection Rule (DCR), and the stream name.
9    endpoint = os.environ["DATA_COLLECTION_ENDPOINT"]
10    rule_id = os.environ["DATA_COLLECTION_RULE_ID"]
11    stream_name = os.environ["STREAM_NAME"]
12
13    # Use DefaultAzureCredential for authentication. 
14    # Ensure your application/user has 'Monitoring Metrics Publisher' role on the DCR.
15    credential = DefaultAzureCredential()
16    client = LogsIngestionClient(endpoint, credential)
17
18    # The logs should be a list of dictionaries matching the structure 
19    # defined in your Data Collection Rule.
20    body = [
21        {
22            "Time": "2023-01-01T00:00:00.000Z",
23            "Computer": "Computer1",
24            "AdditionalContext": "Sample log message"
25        },
26        {
27            "Time": "2023-01-01T00:00:01.000Z",
28            "Computer": "Computer2",
29            "AdditionalContext": "Another sample log message"
30        }
31    ]
32
33    try:
34        client.upload(rule_id=rule_id, stream_name=stream_name, logs=body)
35        print("Logs uploaded successfully.")
36    except HttpResponseError as e:
37        print(f"Upload failed: {e}")
38
39if __name__ == "__main__":
40    main()