適用于 Python 的 Azure 監視器查詢用戶端程式庫 - 1.2.0 版

發行項
05/10/2023

Azure 監視器查詢用戶端程式庫可用來對 Azure 監視器的兩個數據平臺執行唯讀查詢：

記錄 - 從受監視的資源收集並組織記錄和效能資料。來自不同來源的資料，例如來自 Azure 服務的平臺記錄、虛擬機器代理程式的記錄和效能資料，以及來自應用程式的使用量和效能資料，都可以合併成單一 Azure Log Analytics 工作區。您可以使用Kusto 查詢語言一起分析各種資料類型。
計量 - 從受監視的資源收集數值資料到時間序列資料庫。計量是定期收集的數值，並且會在特定時間描述資源的某些層面。計量是輕量型且能夠支援近乎即時的案例，因此有助於警示和快速偵測問題。

資源：

開始使用

Prerequisites

Python 3.7 或更新版本
Azure 訂用帳戶
TokenCredential 實作，例如Azure 身分識別程式庫認證類型。
若要查詢記錄，您需要 Azure Log Analytics 工作區。
若要查詢計量，您需要任何類型的 Azure 資源 (儲存體帳戶、金鑰保存庫、Cosmos DB 等) 。

安裝套件

使用 pip安裝適用于 Python 的 Azure 監視器查詢用戶端程式庫：

pip install azure-monitor-query

建立用戶端

需要經過驗證的用戶端，才能查詢記錄或計量。程式庫同時包含用戶端的同步和非同步形式。若要進行驗證，請建立權杖認證的實例。建立 LogsQueryClient 或 MetricsQueryClient 時，請使用該實例。下列範例會從azure 身分識別套件使用 DefaultAzureCredential 。

同步用戶端

請考慮下列範例，此範例會針對記錄和計量查詢建立同步用戶端：

from azure.identity import DefaultAzureCredential
from azure.monitor.query import LogsQueryClient, MetricsQueryClient

credential = DefaultAzureCredential()
logs_client = LogsQueryClient(credential)
metrics_client = MetricsQueryClient(credential)

非同步用戶端

查詢用戶端 API 的非同步形式可在 -suffixed 命名空間中找到 .aio 。例如：

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

credential = DefaultAzureCredential()
async_logs_client = LogsQueryClient(credential)
async_metrics_client = MetricsQueryClient(credential)

設定非公用 Azure 雲端的用戶端

根據預設， LogsQueryClient 和 MetricsQueryClient 會設定為連線到公用 Azure 雲端。您可以藉由傳入正確的 endpoint 引數，將這些設定為連線到非公用 Azure 雲端：例如：

logs_client = LogsQueryClient(credential, endpoint="https://api.loganalytics.azure.cn/v1")
metrics_client = MetricsQueryClient(credential, endpoint="https://management.chinacloudapi.cn")

注意：目前， MetricsQueryClient 使用 Azure Resource Manager (ARM) 端點來查詢計量，因此使用此用戶端時，您需要雲端的對應管理端點。未來可能會有所變更。

執行查詢

For examples of Logs and Metrics queries, see the Examples section.

重要概念

記錄查詢速率限制和節流

當要求速率太高時，Log Analytics 服務會套用節流。 Kusto 查詢上也會套用限制，例如傳回的資料列數目上限。如需詳細資訊，請參閱查詢 API。

如果您要執行批次記錄查詢，節流要求會傳回 LogsQueryError 物件。該物件的 code 值將會是 ThrottledError 。

計量資料結構

每組計量值都是具有下列特性的時間序列：

收集到值的時間
與值相關聯的資源
作用類似計量類別的命名空間
計量名稱
值本身
某些計量可能會有多個維度，如多維度計量中所述。自訂計量最多可以有 10 個維度。

記錄查詢

此範例示範如何查詢 Log Analytics 工作區。若要處理回應，並以表格式形式檢視回應，會使用 pandas 程式庫。如果您選擇不使用 pandas，請參閱範例。

指定時間範圍

參數 timespan 會指定要查詢資料的時間持續時間。這個值可以是下列其中一個值：

timedelta
和 timedelta 開始日期時間
開始日期時間/結束日期時間

例如：

import os
import pandas as pd
from datetime import datetime, timezone
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.identity import DefaultAzureCredential
from azure.core.exceptions import HttpResponseError

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)

query = """AppRequests | take 5"""

start_time=datetime(2021, 7, 2, tzinfo=timezone.utc)
end_time=datetime(2021, 7, 4, tzinfo=timezone.utc)

try:
    response = client.query_workspace(
        workspace_id=os.environ['LOG_WORKSPACE_ID'],
        query=query,
        timespan=(start_time, end_time)
        )
    if response.status == LogsQueryStatus.PARTIAL:
        error = response.partial_error
        data = response.partial_data
        print(error)
    elif response.status == LogsQueryStatus.SUCCESS:
        data = response.tables
    for table in data:
        df = pd.DataFrame(data=table.rows, columns=table.columns)
        print(df)
except HttpResponseError as err:
    print("something fatal happened")
    print(err)

處理記錄查詢回應

API 會 query_workspace 傳 LogsQueryResult 回或 LogsQueryPartialResult 物件。 API 會 batch_query 傳回可能包含 LogsQueryResult 、 LogsQueryPartialResult 和 LogsQueryError 物件的清單。以下是回應的階層：

LogsQueryResult
|---statistics
|---visualization
|---tables (list of `LogsTable` objects)
    |---name
    |---rows
    |---columns
    |---columns_types

LogsQueryPartialResult
|---statistics
|---visualization
|---partial_error (a `LogsQueryError` object)
    |---code
    |---message
    |---details
    |---status
|---partial_data (list of `LogsTable` objects)
    |---name
    |---rows
    |---columns
    |---columns_types

直接 LogsQueryResult 逐一查看資料表以方便起見。例如，若要使用資料表處理記錄查詢回應，並使用 pandas 加以顯示：

response = client.query(...)
for table in response:
    df = pd.DataFrame(table.rows, columns=[col.name for col in table.columns])

您可以在這裡找到完整的範例。

以類似的方式處理批次記錄查詢回應：

for result in response:
    if result.status == LogsQueryStatus.SUCCESS:
        for table in result:
            df = pd.DataFrame(table.rows, columns=table.columns)
            print(df)

您可以在這裡找到完整的範例。

批次記錄查詢

下列範例示範如何使用批次查詢 API 同時傳送多個查詢。查詢可以表示為物件清單 LogsBatchQuery 或字典。此範例使用先前的方法。

import os
from datetime import timedelta, datetime, timezone
import pandas as pd
from azure.monitor.query import LogsQueryClient, LogsBatchQuery, LogsQueryStatus
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)
requests = [
    LogsBatchQuery(
        query="AzureActivity | summarize count()",
        timespan=timedelta(hours=1),
        workspace_id=os.environ['LOG_WORKSPACE_ID']
    ),
    LogsBatchQuery(
        query= """bad query""",
        timespan=timedelta(days=1),
        workspace_id=os.environ['LOG_WORKSPACE_ID']
    ),
    LogsBatchQuery(
        query= """let Weight = 92233720368547758;
        range x from 1 to 3 step 1
        | summarize percentilesw(x, Weight * 100, 50)""",
        workspace_id=os.environ['LOG_WORKSPACE_ID'],
        timespan=(datetime(2021, 6, 2, tzinfo=timezone.utc), datetime(2021, 6, 5, tzinfo=timezone.utc)), # (start, end)
        include_statistics=True
    ),
]
results = client.query_batch(requests)

for res in results:
    if res.status == LogsQueryStatus.FAILURE:
        # this will be a LogsQueryError
        print(res.message)
    elif res.status == LogsQueryStatus.PARTIAL:
        ## this will be a LogsQueryPartialResult
        print(res.partial_error)
        for table in res.partial_data:
            df = pd.DataFrame(table.rows, columns=table.columns)
            print(df)
    elif res.status == LogsQueryStatus.SUCCESS:
        ## this will be a LogsQueryResult
        table = res.tables[0]
        df = pd.DataFrame(table.rows, columns=table.columns)
        print(df)

資源記錄查詢

下列範例示範如何在不使用 Log Analytics 工作區的情況下，直接從 Azure 資源查詢記錄。在這裡，會 query_resource 使用方法， query_workspace 而不是使用，而不是工作區識別碼，Azure 資源識別碼會傳入 (例如 /subscriptions/{subscription-id}/resourceGroups/{resource-group-name}/providers/{resource-provider}/{resource-type}/{resource-name}) 。

import os
import pandas as pd
from datetime import timedelta
from azure.monitor.query import LogsQueryClient, LogsQueryStatus
from azure.core.exceptions import HttpResponseError
from azure.identity import DefaultAzureCredential

credential  = DefaultAzureCredential()
client = LogsQueryClient(credential)

query = """AzureActivity | take 5"""

try:
    response = client.query_resource(os.environ['LOGS_RESOURCE_ID'], query, timespan=timedelta(days=1))
    if response.status == LogsQueryStatus.PARTIAL:
        error = response.partial_error
        data = response.partial_data
        print(error)
    elif response.status == LogsQueryStatus.SUCCESS:
        data = response.tables
    for table in data:
        df = pd.DataFrame(data=table.rows, columns=table.columns)
        print(df)
except HttpResponseError as err:
    print("something fatal happened")
    print(err)

進階記錄查詢案例

設定記錄查詢逾時

下列範例顯示以秒為單位設定伺服器逾時。如果查詢所花費的時間超過所述的逾時，則會引發閘道逾時。預設值為 180 秒，而且可以設定為 10 分鐘， (600 秒) 。

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

credential = DefaultAzureCredential()
client = LogsQueryClient(credential)

response = client.query_workspace(
    os.environ['LOG_WORKSPACE_ID'],
    "range x from 1 to 10000000000 step 1 | count",
    timespan=timedelta(days=1),
    server_timeout=600 # sets the timeout to 10 minutes
    )

查詢多個工作區

相同的記錄查詢可以跨多個 Log Analytics 工作區執行。除了 Kusto 查詢之外，還需要下列參數：

workspace_id - 第一個 (主要) 工作區識別碼。
additional_workspaces - 工作區清單，不包括參數中 workspace_id 提供的工作區。參數的清單專案可能包含下列識別碼格式：
- 限定工作區名稱
- 工作區識別碼
- Azure 資源識別碼

例如，下列查詢會在三個工作區中執行：

client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    additional_workspaces=['<workspace 2>', '<workspace 3>']
    )

您可以在這裡找到完整的範例。

包含統計資料

若要取得記錄查詢執行統計資料，例如 CPU 和記憶體耗用量：

將 include_statistics 參數設定為 True。
statistics存取物件內的 LogsQueryResult 欄位。

下列範例會列印查詢執行時間：

query = "AzureActivity | top 10 by TimeGenerated"
result = client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    include_statistics=True
    )

execution_time = result.statistics.get("query", {}).get("executionTime")
print(f"Query execution time: {execution_time}")

欄位 statistics 是 dict 對應至原始 JSON 回應的，其結構可能會因查詢而異。統計資料可在屬性中找到 query 。例如：

{
  "query": {
    "executionTime": 0.0156478,
    "resourceUsage": {...},
    "inputDatasetStatistics": {...},
    "datasetStatistics": [{...}]
  }
}

包含視覺效果

若要使用轉譯運算子取得記錄查詢的視覺效果資料：

將 include_visualization 屬性設定為 True。
visualization存取物件內的 LogsQueryResult 欄位。

例如：

query = (
    "StormEvents"
    "| summarize event_count = count() by State"
    "| where event_count > 10"
    "| project State, event_count"
    "| render columnchart"
)
result = client.query_workspace(
    <workspace_id>,
    query,
    timespan=timedelta(days=1),
    include_visualization=True
    )

print(f"Visualization result: {result.visualization}")

欄位 visualization 是 dict 對應至原始 JSON 回應的，其結構可能會因查詢而異。例如：

{
  "visualization": "columnchart",
  "title": "the chart title",
  "accumulate": False,
  "isQuerySorted": False,
  "kind": None,
  "legend": None,
  "series": None,
  "yMin": "NaN",
  "yMax": "NaN",
  "xAxis": None,
  "xColumn": None,
  "xTitle": "x axis title",
  "yAxis": None,
  "yColumns": None,
  "ySplit": None,
  "yTitle": None,
  "anomalyColumns": None
}

計量查詢

下列範例會取得事件方格訂用帳戶的計量。資源 URI 是事件方格主題的 URI。

資源 URI 必須是正在查詢計量的資源 URI。其格式通常為 /subscriptions/<id>/resourceGroups/<rg-name>/providers/<source>/topics/<resource-name> 。

若要尋找資源 URI：

流覽至Azure 入口網站中的資源頁面。
從 [ 概觀 ] 刀鋒視窗中，選取 [JSON 檢視] 連結。
在產生的 JSON 中，複製屬性的值 id 。

注意：計量會以傳送metric_names的順序傳回。

import os
from datetime import timedelta, datetime
from azure.monitor.query import MetricsQueryClient
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = MetricsQueryClient(credential)
start_time = datetime(2021, 5, 25)
duration = timedelta(days=1)
metrics_uri = os.environ['METRICS_RESOURCE_URI']
response = client.query_resource(
    metrics_uri,
    metric_names=["PublishSuccessCount"],
    timespan=(start_time, duration)
    )

for metric in response.metrics:
    print(metric.name)
    for time_series_element in metric.timeseries:
        for metric_value in time_series_element.data:
            print(metric_value.time_stamp)

處理計量查詢回應

計量查詢 API 會 MetricsQueryResult 傳回物件。物件 MetricsQueryResult 包含屬性，例如類型物件清單 Metric 、 granularity 、 namespace 和 timespan 。您可以使用 Metric 參數來存取 metrics 物件清單。此清單中的每個 Metric 物件都包含物件清單 TimeSeriesElement 。每個 TimeSeriesElement 物件都包含 data 和 metadata_values 屬性。在視覺化形式中，回應的物件階層類似下列結構：

MetricsQueryResult
|---granularity
|---timespan
|---cost
|---namespace
|---resource_region
|---metrics (list of `Metric` objects)
    |---id
    |---type
    |---name
    |---unit
    |---timeseries (list of `TimeSeriesElement` objects)
        |---metadata_values
        |---data (list of data points represented by `MetricValue` objects)

處理回應的範例

import os
from azure.monitor.query import MetricsQueryClient, MetricAggregationType
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()
client = MetricsQueryClient(credential)

metrics_uri = os.environ['METRICS_RESOURCE_URI']
response = client.query_resource(
    metrics_uri,
    metric_names=["MatchedEventCount"],
    aggregations=[MetricAggregationType.COUNT]
    )

for metric in response.metrics:
    print(metric.name)
    for time_series_element in metric.timeseries:
        for metric_value in time_series_element.data:
            if metric_value.count != 0:
                print(
                    "There are {} matched events at {}".format(
                        metric_value.count,
                        metric_value.time_stamp
                    )
                )

疑難排解

如需如何診斷各種失敗案例的詳細資料，請參閱我們的疑難排解指南。

下一步

若要深入瞭解 Azure 監視器，請參閱 Azure 監視器服務檔。

範例

下列程式碼範例顯示 Azure 監視器查詢用戶端程式庫的常見案例。

記錄查詢範例

計量查詢範例

參與

此專案歡迎參與和提供建議。大部分的參與都要求您同意「參與者授權合約 (CLA)」，宣告您有權且確實授與我們使用投稿的權利。如需詳細資訊，請造訪 cla.microsoft.com。

當您提交提取要求時，CLA Bot 會自動判斷您是否需要提供 CLA，並適當地裝飾 PR (例如標籤、註解)。請遵循 bot 提供的指示。您只需要使用 CLA 在所有存放庫上執行此動作一次。

此專案採用 Microsoft Open Source Code of Conduct (Microsoft 開放原始碼管理辦法)。如需詳細資訊，請參閱管理辦法常見問題集，如有任何其他問題或意見請連絡 opencode@microsoft.com。

Share via

適用于 Python 的 Azure 監視器查詢用戶端程式庫 - 1.2.0 版

開始使用

Prerequisites

安裝套件

建立用戶端

同步用戶端

非同步用戶端

設定非公用 Azure 雲端的用戶端

執行查詢

重要概念

記錄查詢速率限制和節流

計量資料結構

範例

記錄查詢

指定時間範圍

處理記錄查詢回應

批次記錄查詢

資源記錄查詢

進階記錄查詢案例

設定記錄查詢逾時

查詢多個工作區

包含統計資料

包含視覺效果

計量查詢

處理計量查詢回應

處理回應的範例

疑難排解

下一步

範例

記錄查詢範例

計量查詢範例

參與

意見反應

其他資源