使用逻辑应用将数据从 Log Analytics 工作区导出到存储帐户

项目
10/15/2024

本文介绍一种使用 Azure 逻辑应用从 Azure Monitor 中的 Log Analytics 工作区查询数据并将其发送到 Azure 存储的方法。如果需要导出 Azure Monitor 日志数据用于审核和合规性场景或允许其他服务检索此数据，请使用此过程。

其他导出方法

本文讨论的方法介绍了使用逻辑应用从日志查询进行计划导出。针对特定场景导出数据的其他选项包括：

若要将数据从 Log Analytics 工作区导出到存储帐户或 Azure 事件中心，请使用 Azure Monitor 日志的 Log Analytics 工作区数据导出功能。请参阅 Azure Monitor 中的 Log Analytics 工作区数据导出。
使用逻辑应用一次性导出。请参阅适用于逻辑应用的 Azure Monitor 日志连接器。
使用 PowerShell 脚本一次性导出到本地计算机。请参阅 Invoke-AzOperationalInsightsQueryExport。

概述

此过程使用 Azure Monitor 日志连接器，它可让你从逻辑应用运行日志查询，并将其输出用于工作流中的其他操作。此过程中使用 Azure Blob 存储连接器将查询输出发送到存储。

从 Log Analytics 工作区导出数据时，请限制逻辑应用工作流处理的数据量。筛选和聚合查询中的日志数据，以减少所需数据。例如，如需导出登录事件，应筛选所需事件并仅投影必填字段。例如：

SecurityEvent
| where EventID == 4624 or EventID == 4625
| project TimeGenerated , Account , AccountType , Computer

按计划导出数据时，请在查询中使用 ingestion_time() 函数，确保不会错过延迟到达的数据。如果数据由于网络或平台问题而延迟，使用引入时间可确保将数据包含在下一次逻辑应用执行中。有关示例，请参阅逻辑应用过程部分中的“添加 Azure Monitor 日志操作”步骤。

先决条件

在开始此过程之前，必须完成以下先决条件：

Log Analytics 工作区：创建逻辑应用的用户必须至少具有工作区的读取权限。
存储帐户：存储帐户不必与 Log Analytics 工作区位于同一订阅中。创建逻辑应用的用户必须具有存储帐户的写入权限。

连接器限制

Azure Monitor 中的 Log Analytics 工作区和日志查询均为多租户服务，其中具有可保护和隔离客户以及保持服务质量的限制。查询大量数据时，请考虑以下限制，这些限制可能会影响配置逻辑应用定期和日志查询的方式：

日志查询返回的行数不能超过 500,000。
日志查询返回的字节数不能超过 64,000,000。
日志查询的运行时间不能超过 10 分钟。
Log Analytics 连接器限制为每分钟 100 个调用。

逻辑应用过程

以下部分将引导你完成该过程。

在存储帐户中创建容器

使用创建容器中的过程将容器添加到存储帐户中，以保存导出的数据。本文中使用的容器名称是 loganalytics-data，但是你可以使用任何名称。

创建逻辑应用工作流

在 Azure 门户中转到“逻辑应用”，然后选择“添加”。请选择“订阅”、“资源组”和“区域”来存储新的逻辑应用。然后，为它指定一个唯一名称。可以打开“Log Analytics”设置来收集有关运行时数据和事件的信息，如设置 Azure Monitor 日志和收集 Azure 逻辑应用的诊断数据中所述。使用 Azure Monitor 日志连接器不需要此设置。
选择“查看 + 创建”，然后选择“创建”。部署完成后，请选择“转到资源”来打开“逻辑应用设计器”。

为工作流创建触发器

在“从通用触发器开始”下，选择“重复”。此设置会创建一个按固定间隔自动运行的逻辑应用工作流。在操作的“频率”框中，请选择“日”。在“间隔”框中，输入“1”即可每天运行一次工作流。

添加 Azure Monitor 日志操作

通过 Azure Monitor 日志操作，你可以指定要运行的查询。此示例中使用的日志查询针对每小时定期进行了优化。它收集在特定执行时间引入的数据。例如，如果工作流在 4:35 运行，则时间范围为 3:00-4:00。如果将逻辑应用更改为以其他频率运行，那么也需要更改查询。例如，如果将定期设置为每天运行，请将查询中的 startTime 设置为 startofday(make_datetime(year,month,day,0,0))。

系统将提示你选择一个租户，从而向工作流将用其来运行查询的帐户授予对 Log Analytics 工作区的访问权限。

选择“+ 新建步骤”，添加在完成重复操作后运行的操作。在“选择操作”下，输入“azure monitor”。然后，选择“Azure Monitor 日志”。
选择“Azure Log Analytics - 运行查询并列出结果”。
选择 Log Analytics 工作区的“订阅”和“资源组”。为“资源类型”选择“Log Analytics 工作区”。然后，在“资源名称”下选择工作区名称。

将以下日志查询添加到“查询”窗口中：

let dt = now();
let year = datetime_part('year', dt);
let month = datetime_part('month', dt);
let day = datetime_part('day', dt);
 let hour = datetime_part('hour', dt);
let startTime = make_datetime(year,month,day,hour,0)-1h;
let endTime = startTime + 1h - 1tick;
AzureActivity
| where ingestion_time() between(startTime .. endTime)
| project 
    TimeGenerated,
    BlobTime = startTime, 
    OperationName ,
    OperationNameValue ,
    Level ,
    ActivityStatus ,
    ResourceGroup ,
    SubscriptionId ,
    Category ,
    EventSubmissionTimestamp ,
    ClientIpAddress = parse_json(HTTPRequest).clientIpAddress ,
    ResourceId = _ResourceId

时间范围根据 TimeGenerated 列来指定将包含在查询中的记录。值应大于在查询中选择的时间范围。由于此查询未使用 TimeGenerated 列，因此“在查询中设置”选项不可用。有关时间范围的详细信息，请参阅查询范围。对于“时间范围”，选择“过去 4 小时” 。此设置可确保引入时间大于 TimeGenerated 的任何记录都将包含在结果中。

（可选）添加分析 JSON 操作

“运行查询并列出结果”操作的输出采用 JSON 格式。可在准备进行“撰写”操作的过程中分析此数据并对其进行操作。

可以提供一个 JSON 架构用于描述预期要接收的有效负载。设计器使用此架构分析 JSON 内容，并生成用户友好的标记来表示 JSON 内容中的属性。然后，你可以在整个逻辑应用工作流中轻松引用和使用这些属性。

可以使用“运行查询并列出结果”步骤中的示例输出。

在逻辑应用功能区中选择“运行触发器”。然后，选择“运行”，下载并保存输出记录。对于上一步中的示例查询，可以使用以下示例输出：

{
    "TimeGenerated": "2020-09-29T23:11:02.578Z",
    "BlobTime": "2020-09-29T23:00:00Z",
    "OperationName": "Returns Storage Account SAS Token",
    "OperationNameValue": "MICROSOFT.RESOURCES/DEPLOYMENTS/WRITE",
    "Level": "Informational",
    "ActivityStatus": "Started",
    "ResourceGroup": "monitoring",
    "SubscriptionId": "00000000-0000-0000-0000-000000000000",
    "Category": "Administrative",
    "EventSubmissionTimestamp": "2020-09-29T23:11:02Z",
    "ClientIpAddress": "192.168.1.100",
    "ResourceId": "/subscriptions/00000000-0000-0000-0000-000000000000/resourcegroups/monitoring/providers/microsoft.storage/storageaccounts/my-storage-account"
}

选择“+ 新建步骤”，然后选择“+ 添加操作”。在“选择操作”下，输入“json”，然后选择“分析 JSON”。
选择“内容”框即可显示以前活动的值的列表。从“运行查询并列出结果”操作中选择“正文” 。此输出来自日志查询。
复制之前保存的示例记录。选择“使用示例有效负载生成架构”并粘贴它。

添加“撰写”操作

“撰写”操作获取已分析的 JSON 输出，并创建需要存储在 blob 中的对象。

选择“+ 新建步骤”，然后选择“+ 添加操作”。在“选择操作”下，输入“撰写”。然后，选择“撰写”操作。
选择“输入”框，显示以前活动的值的列表。从“分析 JSON”操作中选择“正文” 。这个已分析的输出来自日志查询。

添加“创建 Blob”操作

“创建 Blob”操作会将撰写的 JSON 写入存储。

选择“+ 新建步骤”，然后选择“+ 添加操作”。在“选择操作”下，输入“Blob”。然后，选择“创建 Blob”操作。
在“连接名称”中输入连接到存储帐户的名称。然后，选择“文件夹路径”框中的文件夹图标，从而选择存储帐户中的容器。选择“Blob 名称”，查看以前活动的值的列表。选择“表达式”，然后输入与你的时间间隔匹配的表达式。对于每小时运行一次的该查询，以下表达式将设置每个前一小时的 Blob 名称：
```
subtractFromTime(formatDateTime(utcNow(),'yyyy-MM-ddTHH:00:00'), 1,'Hour')
```
选择“Blob 内容”框可显示以前活动的值的列表。然后，在“撰写”部分中选择“输出”。

测试工作流

若要测试工作流，请选择“运行”。如果工作流存在错误，会在有问题的步骤中进行指示。可以查看执行并钻取到每个步骤，通过查看输入和输出来进行故障调查。如有必要，请参阅在 Azure 逻辑应用中排查和诊断工作流故障。

查看存储中的日志

转到 Azure 门户中的“存储帐户”菜单，然后选择你的存储帐户。选择“Blob”磁贴。然后，选择在“创建 Blob”操作中指定的容器。选择一个 Blob，然后选择“编辑 Blob”。

逻辑应用模板

模板中不包含可选的分析 JSON 步骤

{
    "definition": {
        "$schema": "https://schema.management.azure.com/providers/Microsoft.Logic/schemas/2016-06-01/workflowdefinition.json#",
        "actions": {
            "Compose": {
                "inputs": "@body('Run_query_and_list_results')",
                "runAfter": {
                    "Run_query_and_list_results": [
                        "Succeeded"
                    ]
                },
                "type": "Compose"
            },
            "Create_blob_(V2)": {
                "inputs": {
                    "body": "@outputs('Compose')",
                    "headers": {
                        "ReadFileMetadataFromServer": true
                    },
                    "host": {
                        "connection": {
                            "name": "@parameters('$connections')['azureblob']['connectionId']"
                        }
                    },
                    "method": "post",
                    "path": "/v2/datasets/@{encodeURIComponent(encodeURIComponent('AccountNameFromSettings'))}/files",
                    "queries": {
                        "folderPath": "/logicappexport",
                        "name": "@{utcNow()}",
                        "queryParametersSingleEncoded": true
                    }
                },
                "runAfter": {
                    "Compose": [
                        "Succeeded"
                    ]
                },
                "runtimeConfiguration": {
                    "contentTransfer": {
                        "transferMode": "Chunked"
                    }
                },
                "type": "ApiConnection"
            },
            "Run_query_and_list_results": {
                "inputs": {
                    "body": "let dt = now();\nlet year = datetime_part('year', dt);\nlet month = datetime_part('month', dt);\nlet day = datetime_part('day', dt);\n let hour = datetime_part('hour', dt);\nlet startTime = make_datetime(year,month,day,hour,0)-1h;\nlet endTime = startTime + 1h - 1tick;\nAzureActivity\n| where ingestion_time() between(startTime .. endTime)\n| project \n    TimeGenerated,\n    BlobTime = startTime, \n    OperationName ,\n    OperationNameValue ,\n    Level ,\n    ActivityStatus ,\n    ResourceGroup ,\n    SubscriptionId ,\n    Category ,\n    EventSubmissionTimestamp ,\n    ClientIpAddress = parse_json(HTTPRequest).clientIpAddress ,\n    ResourceId = _ResourceId ",
                    "host": {
                        "connection": {
                            "name": "@parameters('$connections')['azuremonitorlogs']['connectionId']"
                        }
                    },
                    "method": "post",
                    "path": "/queryData",
                    "queries": {
                        "resourcegroups": "resource-group-name",
                        "resourcename": "workspace-name",
                        "resourcetype": "Log Analytics Workspace",
                        "subscriptions": "workspace-subscription-id",
                        "timerange": "Set in query"
                    }
                },
                "runAfter": {},
                "type": "ApiConnection"
            }
        },
        "contentVersion": "1.0.0.0",
        "outputs": {},
        "parameters": {
            "$connections": {
                "defaultValue": {},
                "type": "Object"
            }
        },
        "triggers": {
            "Recurrence": {
                "evaluatedRecurrence": {
                    "frequency": "Day",
                    "interval": 1
                },
                "recurrence": {
                    "frequency": "Day",
                    "interval": 1
                },
                "type": "Recurrence"
            }
        }
    },
    "parameters": {
        "$connections": {
            "value": {
                "azureblob": {
                    "connectionId": "/subscriptions/logic-app-subscription-id/resourceGroups/logic-app-resource-group-name/providers/Microsoft.Web/connections/blob-connection-name",
                    "connectionName": "blob-connection-name",
                    "id": "/subscriptions/logic-app-subscription-id/providers/Microsoft.Web/locations/canadacentral/managedApis/azureblob"
                },
                "azuremonitorlogs": {
                    "connectionId": "/subscriptions/blob-connection-name/resourceGroups/logic-app-resource-group-name/providers/Microsoft.Web/connections/azure-monitor-logs-connection-name",
                    "connectionName": "azure-monitor-logs-connection-name",
                    "id": "/subscriptions/blob-connection-name/providers/Microsoft.Web/locations/canadacentral/managedApis/azuremonitorlogs"
                }
            }
        }
    }
}

后续步骤

详细了解 Azure Monitor 中的日志查询。
了解有关逻辑应用的详细信息。
详细了解 Power Automate。

通过