Kit SDK Azure Monitor Ingestion pour Python

Envoyez des logs personnalisés vers l'espace de travail Azure Monitor Log Analytics en utilisant l'API Logs Ingestion.

Installation

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

Variables d'environnement

# Data Collection Endpoint (DCE)
AZURE_DCE_ENDPOINT=https://<dce-name>.<region>.ingest.monitor.azure.com  # Obligatoire pour toutes les méthodes d'authentification

# Identifiant immuable Data Collection Rule (DCR)
AZURE_DCR_RULE_ID=dcr-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx  # Obligatoire pour toutes les méthodes d'authentification

# Nom du stream depuis DCR
AZURE_DCR_STREAM_NAME=Custom-MyTable_CL  # Obligatoire pour toutes les méthodes d'authentification
AZURE_TOKEN_CREDENTIALS=prod # Obligatoire seulement si DefaultAzureCredential est utilisé en production

Prérequis

Avant d'utiliser ce SDK, vous devez disposer de :

1. Espace de travail Log Analytics — Destination de vos logs
2. Data Collection Endpoint (DCE) — Point de terminaison d'ingestion
3. Data Collection Rule (DCR) — Définit le schéma et la destination
4. Tableau personnalisé — Dans Log Analytics (créé via DCR ou manuellement)

Authentification

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

# Dev local : DefaultAzureCredential. Production : définissez AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Ou utilisez une credential spécifique directement en production :
# Voir https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()

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

Télécharger des logs personnalisés

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

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": "2024-01-15T10:00:00Z", "Computer": "server1", "Message": "Application started"},
    {"TimeGenerated": "2024-01-15T10:01:00Z", "Computer": "server1", "Message": "Processing request"},
    {"TimeGenerated": "2024-01-15T10:02:00Z", "Computer": "server2", "Message": "Connection established"}
]

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

Télécharger depuis un fichier JSON

import json

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

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

Gestion personnalisée des erreurs

Gérez les défaillances partielles avec un callback :

failed_logs = []

def on_error(error):
    print(f"Upload failed: {error.error}")
    failed_logs.extend(error.failed_logs)

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

# Réessayez les logs échoués
if failed_logs:
    print(f"Retrying {len(failed_logs)} failed logs...")
    client.upload(rule_id=rule_id, stream_name=stream_name, logs=failed_logs)

Ignorer les erreurs

def ignore_errors(error):
    pass  # Ignorer silencieusement les défaillances de téléchargement

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

Client asynchrone

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

async def upload_logs():
    async with LogsIngestionClient(
        endpoint=endpoint,
        credential=DefaultAzureCredential()
    ) as client:
        await client.upload(
            rule_id=rule_id,
            stream_name=stream_name,
            logs=logs
        )

asyncio.run(upload_logs())

Clouds souverains

from azure.identity import AzureAuthorityHosts, DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

# Azure Government
credential = DefaultAzureCredential(authority=AzureAuthorityHosts.AZURE_GOVERNMENT)
client = LogsIngestionClient(
    endpoint="https://example.ingest.monitor.azure.us",
    credential=credential,
    credential_scopes=["https://monitor.azure.us/.default"]
)

Comportement de traitement par lots

Le SDK effectue automatiquement :

• Division des logs en chunks de 1 MB ou moins
• Compression de chaque chunk avec gzip
• Téléchargement des chunks en parallèle

Aucun traitement manuel par lots n'est nécessaire pour les grands ensembles de logs.

Types de client

Concepts clés

Format du nom de stream DCR

Les noms de stream suivent les motifs :

• Custom-<TableName>_CL — Pour les tableaux personnalisés
• Microsoft-<TableName> — Pour les tableaux intégrés

Bonnes pratiques

1. Utilisez DefaultAzureCredential pour l'authentification
2. Gérez les erreurs correctement — utilisez le callback on_error pour les défaillances partielles
3. Incluez TimeGenerated — Champ obligatoire pour tous les logs
4. Correspondez au schéma DCR — Les champs des logs doivent correspondre aux définitions de colonnes DCR
5. Utilisez le client asynchrone pour les scénarios haute throughput
6. Traitez par lots — Le SDK gère le traitement par lots, mais envoyez des chunks raisonnables
7. Supervisez l'ingestion — Vérifiez Log Analytics pour le statut d'ingestion
8. Utilisez le gestionnaire de contexte — Assure un nettoyage correct du client