Sending Custom Logs to Azure Monitor with Python SDK

Centralized logging is essential for understanding distributed system behavior, and Azure Monitor provides powerful log analytics capabilities for querying and alerting. The Azure Monitor Ingestion SDK for Python enables AI agents to send custom logs directly to Log Analytics workspaces using the Logs Ingestion API, supporting scenarios from application-specific telemetry to security event forwarding where standard instrumentation doesn't apply.

What This Skill Does

The azure-monitor-ingestion-py skill provides Python client libraries for sending custom logs to Azure Monitor through Data Collection Endpoints and Data Collection Rules. It handles automatic log batching and compression for efficient transmission, parallel chunk uploads for high throughput, partial failure handling with custom error callbacks, asynchronous upload support for high-concurrency applications, and integration with Data Collection Rules that define schema transformation and routing.

This skill enables agents to send structured log data to custom Log Analytics tables, batch large log collections automatically without manual chunking, implement retry logic for failed uploads through error callbacks, use async patterns for non-blocking log submission, and integrate with sovereign Azure clouds like Azure Government.

The SDK automatically splits large log collections into 1MB chunks, compresses each chunk with gzip, and uploads chunks in parallel for optimal performance. This automatic batching eliminates the need for manual log management while ensuring efficient network utilization.

Getting Started

Install the Azure Monitor ingestion library and Azure Identity:

pip install azure-monitor-ingestion
pip install azure-identity

Configure environment variables for your Data Collection infrastructure:

export AZURE_DCE_ENDPOINT="https://my-dce.eastus.ingest.monitor.azure.com"
export AZURE_DCR_RULE_ID="dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
export AZURE_DCR_STREAM_NAME="Custom-MyAppLogs_CL"

Create a client with DefaultAzureCredential:

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential
import os

client = LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
)

Before sending logs, ensure you have a Data Collection Rule configured with appropriate stream definitions and a target Log Analytics table created.

Key Features

Automatic Batching: The SDK handles all batching complexity automatically. Send large collections of logs without manual chunking. The SDK splits logs into optimal 1MB chunks, ensuring efficient transmission without exceeding API limits.

Compression: Each log chunk is automatically compressed with gzip before transmission, reducing network bandwidth usage and improving upload speeds for large log volumes.

Parallel Uploads: Multiple log chunks upload concurrently, maximizing throughput for large datasets. This parallelization happens automatically without configuration.

Error Handling: Configure custom error callbacks that receive failed log entries and exception details. Implement retry logic, alternative storage, or alerting for upload failures without losing visibility into partial failures.

Async Support: The SDK provides an async client using Python's async/await patterns. Async operations enable non-blocking log uploads in applications with high concurrency requirements or event-driven architectures.

Schema Flexibility: Logs are Python dictionaries that serialize to JSON matching your Data Collection Rule schema. The DCR defines transformations mapping incoming fields to table columns, enabling schema evolution without code changes.

Usage Examples

Sending structured application logs with standard fields:

from azure.monitor.ingestion import LogsIngestionClient
from azure.identity import DefaultAzureCredential
import os
from datetime import datetime

client = LogsIngestionClient(
    endpoint=os.environ["AZURE_DCE_ENDPOINT"],
    credential=DefaultAzureCredential()
)

rule_id = os.environ["AZURE_DCR_RULE_ID"]
stream_name = os.environ["AZURE_DCR_STREAM_NAME"]

logs = [
    {
        "TimeGenerated": datetime.utcnow().isoformat() + "Z",
        "Level": "INFO",
        "Component": "authentication",
        "Message": "User login successful",
        "UserId": "user123"
    },
    {
        "TimeGenerated": datetime.utcnow().isoformat() + "Z",
        "Level": "ERROR",
        "Component": "database",
        "Message": "Connection timeout",
        "RetryCount": 3
    }
]

client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Implementing robust error handling with retry logic:

failed_logs = []

def handle_upload_error(error):
    print(f"Upload failed: {error.error}")
    print(f"Failed log count: {len(error.failed_logs)}")
    failed_logs.extend(error.failed_logs)

client.upload(
    rule_id=rule_id,
    stream_name=stream_name,
    logs=logs,
    on_error=handle_upload_error
)

# Retry failed logs after delay
if failed_logs:
    import time
    time.sleep(5)
    print(f"Retrying {len(failed_logs)} failed logs...")
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)

Using async client for high-throughput scenarios:

import asyncio
from azure.monitor.ingestion.aio import LogsIngestionClient
from azure.identity.aio import DefaultAzureCredential

async def upload_logs_async():
    async with LogsIngestionClient(
        endpoint=os.environ["AZURE_DCE_ENDPOINT"],
        credential=DefaultAzureCredential()
    ) as client:
        await client.upload(
            rule_id=rule_id,
            stream_name=stream_name,
            logs=logs
        )
        print("Async upload completed")

asyncio.run(upload_logs_async())

Loading and sending logs from JSON files:

import json

with open("application_logs.json", "r") as f:
    logs = json.load(f)

# SDK handles large batches automatically
client.upload(rule_id=rule_id, stream_name=stream_name, logs=logs)

Best Practices

Use DefaultAzureCredential for authentication to support multiple authentication methods. In development, it uses Azure CLI credentials. In production on Azure, it automatically uses managed identity. For CI/CD pipelines, it discovers service principal credentials from environment variables.

Always include TimeGenerated fields in log entries. Most Azure Monitor tables require timestamp fields for time-series analysis and querying. Use ISO 8601 format with UTC timezone (e.g., "2024-01-15T10:30:00Z") for consistency.

Match log entry fields to your Data Collection Rule schema. DCR transformations map incoming JSON fields to table columns. Mismatched field names or incompatible types cause ingestion failures. Test schema compatibility in development before production deployment.

Implement error callbacks for production systems. Don't let partial failures silently lose logs. Log failed entries to a file, queue them for retry with exponential backoff, or trigger alerts for investigation.

Use async clients in event-driven applications. If your application uses asyncio for web servers, message consumers, or other async patterns, the async client provides better integration and resource utilization.

Batch logs over time windows rather than sending individual entries. Collect logs over 30-60 second windows and upload in batches. This reduces API calls, improves throughput, and lowers costs compared to real-time individual log submission.

Monitor ingestion success through Log Analytics queries. After uploading, query your target table to verify logs arrived. Check for ingestion delays or schema errors that might prevent data from appearing in queries.

When to Use This Skill

Use this skill when sending custom application logs that don't fit standard Application Insights categories. Business events, workflow state transitions, or domain-specific telemetry often require custom table structures and schemas.

It's ideal for aggregating logs from distributed systems into centralized storage. Microservices, serverless functions, background workers, and batch jobs can all send logs to the same workspace for unified querying and analysis.

The skill is valuable for security event ingestion from custom sources. Security tools, network devices, or custom security scanners can forward events to Azure Monitor for SIEM correlation, alerting, and compliance reporting.

Use it for implementing custom operational metrics beyond standard performance counters. Track business KPIs, SLA metrics, or operational health indicators in Log Analytics for dashboard visualization and alerting.

When Not to Use This Skill

Don't use this skill for standard application telemetry covered by Application Insights. If you're tracking HTTP requests, dependencies, exceptions, or traces, use the Application Insights SDK designed for those scenarios with better performance and richer features.

If you're working with Azure services that support diagnostic settings, use those instead. Many Azure services send logs directly to Log Analytics through diagnostic settings without custom code.

Avoid it for high-frequency time-series metrics with millisecond precision. Azure Monitor Metrics is optimized for numeric metric storage. Logs Ingestion is designed for structured log events, not high-frequency metric samples.

Don't use it for real-time alerting with sub-second latency requirements. Log ingestion and indexing has latency (typically 1-5 minutes). For real-time alerting, use Azure Event Hubs or Stream Analytics instead.

Related Skills

• azure-monitor-ingestion-java - Ingest custom logs from Java applications
• azure-monitor-query-py - Query ingested logs using KQL
• azure-monitor-opentelemetry-exporter-py - Export OpenTelemetry data to Azure Monitor

Source

This skill is provided by Microsoft as part of the Azure SDK for Python. Learn more at the PyPI package page, explore the GitHub source code, and review the Azure Monitor Logs Ingestion API documentation for comprehensive guidance.