azure-monitor-query-py

Azure Monitor Query SDK for Python

Azure Monitor および Log Analytics ワークスペースからログとメトリックをクエリします。

インストール

pip install azure-monitor-query azure-identity

環境変数

# Log Analytics
AZURE_LOG_ANALYTICS_WORKSPACE_ID=<workspace-id>

# Metrics
AZURE_METRICS_RESOURCE_URI=/subscriptions/<sub>/resourceGroups/<rg>/providers/<provider>/<type>/<name>

認証

import os
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()

ログクエリクライアント

基本的なクエリ

from azure.monitor.query import LogsQueryClient
from datetime import timedelta

client = LogsQueryClient(credential)

query = """
AppRequests
| where TimeGenerated > ago(1h)
| summarize count() by bin(TimeGenerated, 5m), ResultCode
| order by TimeGenerated desc
"""

response = client.query_workspace(
    workspace_id=os.environ["AZURE_LOG_ANALYTICS_WORKSPACE_ID"],
    query=query,
    timespan=timedelta(hours=1)
)

for table in response.tables:
    for row in table.rows:
        print(row)

時間範囲を指定したクエリ

from datetime import datetime, timezone

response = client.query_workspace(
    workspace_id=workspace_id,
    query="AppRequests | take 10",
    timespan=(
        datetime(2024, 1, 1, tzinfo=timezone.utc),
        datetime(2024, 1, 2, tzinfo=timezone.utc)
    )
)

DataFrame への変換

import pandas as pd

response = client.query_workspace(workspace_id, query, timespan=timedelta(hours=1))

if response.tables:
    table = response.tables[0]
    df = pd.DataFrame(data=table.rows, columns=table.columns)
    print(df.head())

バッチクエリ

from azure.monitor.query import LogsBatchQuery

queries = [
    LogsBatchQuery(workspace_id=workspace_id, query="AppRequests | take 5", timespan=timedelta(hours=1)),
    LogsBatchQuery(workspace_id=workspace_id, query="AppExceptions | take 5", timespan=timedelta(hours=1))
]

responses = client.query_batch(queries)

for response in responses:
    if response.tables:
        print(f"Rows: {len(response.tables[0].rows)}")

部分的な結果の処理

from azure.monitor.query import LogsQueryStatus

response = client.query_workspace(workspace_id, query, timespan=timedelta(hours=24))

if response.status == LogsQueryStatus.PARTIAL:
    print(f"Partial results: {response.partial_error}")
elif response.status == LogsQueryStatus.FAILURE:
    print(f"Query failed: {response.partial_error}")

メトリッククエリクライアント（旧 azure-monitor-query 1.x のみ）

バージョン 2.0 では MetricsQueryClient が削除されています。上記のバージョンを固定していない現行インストールでは、以下の旧形式のコードを実行しないでください。新しい開発では、ドキュメントに記載されたスキーマを使用して azure-monitor-querymetrics または azure-mgmt-monitor を使用してください。公式の移行に関するお知らせを参照してください。

リソースメトリックのクエリ

from azure.monitor.query import MetricsQueryClient
from datetime import timedelta

metrics_client = MetricsQueryClient(credential)

response = metrics_client.query_resource(
    resource_uri=os.environ["AZURE_METRICS_RESOURCE_URI"],
    metric_names=["Percentage CPU", "Network In Total"],
    timespan=timedelta(hours=1),
    granularity=timedelta(minutes=5)
)

for metric in response.metrics:
    print(f"{metric.name}:")
    for time_series in metric.timeseries:
        for data in time_series.data:
            print(f"  {data.timestamp}: {data.average}")

集計

from azure.monitor.query import MetricAggregationType

response = metrics_client.query_resource(
    resource_uri=resource_uri,
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    aggregations=[
        MetricAggregationType.AVERAGE,
        MetricAggregationType.MAXIMUM,
        MetricAggregationType.MINIMUM,
        MetricAggregationType.COUNT
    ]
)

ディメンションによるフィルタリング

response = metrics_client.query_resource(
    resource_uri=resource_uri,
    metric_names=["Requests"],
    timespan=timedelta(hours=1),
    filter="ApiName eq 'GetBlob'"
)

メトリック定義の一覧表示

definitions = metrics_client.list_metric_definitions(resource_uri)
for definition in definitions:
    print(f"{definition.name}: {definition.unit}")

メトリック名前空間の一覧表示

namespaces = metrics_client.list_metric_namespaces(resource_uri)
for ns in namespaces:
    print(ns.fully_qualified_namespace)

非同期クライアント

from azure.monitor.query.aio import LogsQueryClient
from azure.identity.aio import DefaultAzureCredential

async def query_logs():
    credential = DefaultAzureCredential()
    client = LogsQueryClient(credential)
    
    response = await client.query_workspace(
        workspace_id=workspace_id,
        query="AppRequests | take 10",
        timespan=timedelta(hours=1)
    )
    
    await client.close()
    await credential.close()
    return response

よく使用する Kusto クエリ

// Requests by status code
AppRequests
| summarize count() by ResultCode
| order by count_ desc

// Exceptions over time
AppExceptions
| summarize count() by bin(TimeGenerated, 1h)

// Slow requests
AppRequests
| where DurationMs > 1000
| project TimeGenerated, Name, DurationMs
| order by DurationMs desc

// Top errors
AppExceptions
| summarize count() by ExceptionType
| top 10 by count_

クライアントの種類

クライアント	用途
LogsQueryClient	Log Analytics ワークスペースにクエリを実行
MetricsQueryClient	Azure Monitor メトリックにクエリを実行

ベストプラクティス

1. 相対的な時間範囲には timedelta を使用します
2. 大規模なクエリでは 部分的な結果を処理します
3. 複数のクエリを実行するときは バッチクエリを使用します
4. データポイントを減らすため、メトリックには 適切な粒度を設定します
5. データ分析を容易にするため、DataFrame に変換します
6. メトリックデータを要約するには 集計を使用します
7. メトリックの結果を絞り込むには ディメンションでフィルタリングします

使用する場面

このスキルは、概要で説明されているワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の説明範囲に明確に該当する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中止して確認してください。