Azure Monitor Ingestion Py

1---
2name: azure-monitor-ingestion-py
3description: |
4  Azure Monitor Ingestion SDK for Python. Use for sending custom logs to Log Analytics workspace via Logs Ingestion API.
5  Triggers: "azure-monitor-ingestion", "LogsIngestionClient", "custom logs", "DCR", "data collection rule", "Log Analytics".
6package: azure-monitor-ingestion
7---
8 
9# Azure Monitor Ingestion SDK for Python
10 
11Send custom logs to Azure Monitor Log Analytics workspace using the Logs Ingestion API.
12 
13## Installation
14 
15```bash
16pip install azure-monitor-ingestion
17pip install azure-identity
18```
19 
20## Environment Variables
21 
22```bash
23# Data Collection Endpoint (DCE)
24AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com
25 
26# Data Collection Rule (DCR) immutable ID
27AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
28 
29# Stream name from DCR
30AZURE_DCR_STREAM_NAME=Custom-MyTable_CL
31```
32 
33## Prerequisites
34 
35Before using this SDK, you need:
36 
371. **Log Analytics Workspace** — Target for your logs
382. **Data Collection Endpoint (DCE)** — Ingestion endpoint
393. **Data Collection Rule (DCR)** — Defines schema and destination
404. **Custom Table** — In Log Analytics (created via DCR or manually)
41 
42## Authentication
43 
44```python
45from azure.monitor.ingestion import LogsIngestionClient
46from azure.identity import DefaultAzureCredential
47import os
48 
49client = LogsIngestionClient(
50    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
51    credential=DefaultAzureCredential()
52)
53```
54 
55## Upload Custom Logs
56 
57```python
58from azure.monitor.ingestion import LogsIngestionClient
59from azure.identity import DefaultAzureCredential
60import os
61 
62client = LogsIngestionClient(
63    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
64    credential=DefaultAzureCredential()
65)
66 
67rule_id = os.environ["AZURE_DCR_RULE_ID"]
68stream_name = os.environ["AZURE_DCR_STREAM_NAME"]
69 
70logs = [
71    {"TimeGenerated": "2024-01-15T10:00:00Z", "Computer": "server1", "Message": "Application started"},
72    {"TimeGenerated": "2024-01-15T10:01:00Z", "Computer": "server1", "Message": "Processing request"},
73    {"TimeGenerated": "2024-01-15T10:02:00Z", "Computer": "server2", "Message": "Connection established"}
74]
75 
76client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)
77```
78 
79## Upload from JSON File
80 
81```python
82import json
83 
84with open("logs.json", "r") as f:
85    logs = json.load(f)
86 
87client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)
88```
89 
90## Custom Error Handling
91 
92Handle partial failures with a callback:
93 
94```python
95failed_logs = []
96 
97def on_error(error):
98    print(f"Upload failed: {error.error}")
99    failed_logs.extend(error.failed_logs)
100 
101client.upload(
102    rule_id=rule_id,
103    stream_name=stream_name,
104    logs=logs,
105    on_error=on_error
106)
107 
108# Retry failed logs
109if failed_logs:
110    print(f"Retrying {len(failed_logs)} failed logs...")
111    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)
112```
113 
114## Ignore Errors
115 
116```python
117def ignore_errors(error):
118    pass  # Silently ignore upload failures
119 
120client.upload(
121    rule_id=rule_id,
122    stream_name=stream_name,
123    logs=logs,
124    on_error=ignore_errors
125)
126```
127 
128## Async Client
129 
130```python
131import asyncio
132from azure.monitor.ingestion.aio import LogsIngestionClient
133from azure.identity.aio import DefaultAzureCredential
134 
135async def upload_logs():
136    async with LogsIngestionClient(
137        endpoint=endpoint,
138        credential=DefaultAzureCredential()
139    ) as client:
140        await client.upload(
141            rule_id=rule_id,
142            stream_name=stream_name,
143            logs=logs
144        )
145 
146asyncio.run(upload_logs())
147```
148 
149## Sovereign Clouds
150 
151```python
152from azure.identity import AzureAuthorityHosts, DefaultAzureCredential
153from azure.monitor.ingestion import LogsIngestionClient
154 
155# Azure Government
156credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
157client = LogsIngestionClient(
158    endpoint="https://example.ingest.monitor.azure.us",
159    credential=credential,
160    credential_scopes=["https://monitor.azure.us/.default"]
161)
162```
163 
164## Batching Behavior
165 
166The SDK automatically:
167- Splits logs into chunks of 1MB or less
168- Compresses each chunk with gzip
169- Uploads chunks in parallel
170 
171No manual batching needed for large log sets.
172 
173## Client Types
174 
175| Client | Purpose |
176|--------|---------|
177| `LogsIngestionClient` | Sync client for uploading logs |
178| `LogsIngestionClient` (aio) | Async client for uploading logs |
179 
180## Key Concepts
181 
182| Concept | Description |
183|---------|-------------|
184| **DCE** | Data Collection Endpoint — ingestion URL |
185| **DCR** | Data Collection Rule — defines schema, transformations, destination |
186| **Stream** | Named data flow within a DCR |
187| **Custom Table** | Target table in Log Analytics (ends with `_CL`) |
188 
189## DCR Stream Name Format
190 
191Stream names follow patterns:
192- `Custom-<TableName>_CL` — For custom tables
193- `Microsoft-<TableName>` — For built-in tables
194 
195## Best Practices
196 
1971. **Use DefaultAzureCredential** for authentication
1982. **Handle errors gracefully** — use `on_error` callback for partial failures
1993. **Include TimeGenerated** — Required field for all logs
2004. **Match DCR schema** — Log fields must match DCR column definitions
2015. **Use async client** for high-throughput scenarios
2026. **Batch uploads** — SDK handles batching, but send reasonable chunks
2037. **Monitor ingestion** — Check Log Analytics for ingestion status
2048. **Use context manager** — Ensures proper client cleanup
205