分析 Azure Monitor 日志中的文本数据

Azure Monitor 收集的某些日志数据会在单个属性中包括多条信息。 将此数据分析为多个属性可以更轻松地在查询中进行使用。 一个常见示例是收集在单个属性中包含多个值的整个日志项目的自定义日志。 通过为不同值创建单独的属性,可以对每个值进行搜索和聚合。

本文介绍了用于在引入数据时以及在查询中检索时分析 Azure Monitor 中的日志数据的不同选项,比较了每个选项的相对优点。

所需的权限

分析方法

在收集数据的引入时间或是在使用查询分析数据的查询时间,可以分析数据。 每种策略都具有独特的优点。

在收集时分析数据

使用转换在收集时分析数据,并定义要向其发送分析数据的列。

优点:

  • 更易于查询收集的数据,因为无需在查询中包含分析命令。
  • 查询性能更好,因为查询无需执行分析。

缺点:

  • 必须提前定义。 不能包括已收集的数据。
  • 如果更改分析逻辑,则它仅应用于新数据。
  • 增加收集数据的延迟时间。
  • 错误可能难以处理。

在查询时分析数据

如果在查询时分析数据,则在查询中包含用于将数据分析为多个字段的逻辑。 实际表本身不进行修改。

优点:

  • 应用于任何数据,包括已收集的数据。
  • 逻辑中的更改可以立即应用于所有数据。
  • 灵活的分析选项,包括用于特定数据结构的预定义逻辑。

缺点:

  • 需要更复杂的查询。 此缺点可以通过使用函数模拟表来进行缓解。
  • 必须在多个查询中复制分析逻辑。 可以通过函数共享某些逻辑。
  • 在对非常大的记录集(数十亿个记录)运行复杂逻辑时可能会形成开销。

在收集时分析数据

有关在收集数据时分析数据的详细信息,请参阅 Azure Monitor 中的转换结构。 此方法会在表中创建可以由查询使用的自定义属性,就跟任何其他属性一样。

使用模式在查询中分析数据

当要分析的数据可以通过在记录间重复的某种模式进行标识时,可以使用 Kusto 查询语言中的不同运算符将特定数据段提取到一个或多个新属性中。

简单文本模式

在查询中使用分析运算符创建可以从字符串表达式中提取的一个或多个自定义属性。 指定要标识的模式以及要创建的属性的名称。 对于具有形式类似于 key=value 的键/值字符串的数据,此方法很有用。

请考虑具有以下格式的数据的自定义日志:

Time=2018-03-10 01:34:36 Event Code=207 Status=Success Message=Client 05a26a97-272a-4bc9-8f64-269d154b0e39 connected
Time=2018-03-10 01:33:33 Event Code=208 Status=Warning Message=Client ec53d95c-1c88-41ae-8174-92104212de5d disconnected
Time=2018-03-10 01:35:44 Event Code=209 Status=Success Message=Transaction 10d65890-b003-48f8-9cfc-9c74b51189c8 succeeded
Time=2018-03-10 01:38:22 Event Code=302 Status=Error Message=Application could not connect to database
Time=2018-03-10 01:31:34 Event Code=303 Status=Error Message=Application lost connection to database

以下查询会将此数据分析为各个单独属性。 添加了包含 project 的行以便仅返回计算的属性,而不是 RawData,这是从自定义日志保存整个条目的单个属性。

MyCustomLog_CL
| parse RawData with * "Time=" EventTime " Event Code=" Code " Status=" Status " Message=" Message
| project EventTime, Code, Status, Message

本例分解了 AzureActivity 表中的 UPN 的用户名。

AzureActivity
| parse  Caller with UPNUserPart "@" * 
| where UPNUserPart != "" //Remove non UPN callers (apps, SPNs, etc)
| distinct UPNUserPart, Caller

正则表达式

如果可以使用正则表达式标识数据,则可以通过使用正则表达式的函数提取各个值。 下面的示例使用提取分解 AzureActivity 记录中的 UPN 字段,然后返回非重复用户。

AzureActivity
| extend UPNUserPart = extract("([a-z.]*)@", 1, Caller) 
| distinct UPNUserPart, Caller

为了显示大规模高效分析,Azure Monitor 使用 re2 版本的正则表达式,这与某些其他正则表达式变体相似,但并不完全相同。 有关详细信息,请参阅 re2 表达式语法

在查询中分析带分隔符的数据

带分隔符的数据使用常见字符(例如 CSV 文件中的逗号)来分隔字段。 通过拆分函数可使用指定分隔符来分析带分隔符的数据。 可以将此方法与扩展运算符结合使用,以返回数据中的所有字段,或指定要包括在输出中的各个字段。

注意

由于拆分返回动态对象,因此结果可能需要显式强制转换为数据类型(如字符串)以便在运算符和筛选器中使用。

请考虑具有以下 CSV 格式的数据的自定义日志:

2018-03-10 01:34:36, 207,Success,Client 05a26a97-272a-4bc9-8f64-269d154b0e39 connected
2018-03-10 01:33:33, 208,Warning,Client ec53d95c-1c88-41ae-8174-92104212de5d disconnected
2018-03-10 01:35:44, 209,Success,Transaction 10d65890-b003-48f8-9cfc-9c74b51189c8 succeeded
2018-03-10 01:38:22, 302,Error,Application could not connect to database
2018-03-10 01:31:34, 303,Error,Application lost connection to database

以下查询会分析此数据和按两个计算的属性进行汇总。 第一行将 RawData 属性拆分为字符串数组。 接下来各行会向单独的属性提供名称,并使用将它们转换为相应数据类型的函数将它们添加到输出。

MyCustomCSVLog_CL
| extend CSVFields  = split(RawData, ',')
| extend EventTime  = todatetime(CSVFields[0])
| extend Code       = toint(CSVFields[1]) 
| extend Status     = tostring(CSVFields[2]) 
| extend Message    = tostring(CSVFields[3]) 
| where getyear(EventTime) == 2018
| summarize count() by Status,Code

在查询中分析预定义结构

如果数据采用已知结构设置格式,则可能能够使用 Kusto 查询语言中的一个函数来分析预定义结构:

下面的示例查询分析 AzureActivity 表(采用 JSON 结构)的 Properties 字段。 它将结果保存到一个名为 parsedProp 的动态属性,其中包含采用 JSON 的各个命名值。 这些值用于筛选和汇总查询结果。

AzureActivity
| extend parsedProp = parse_json(Properties) 
| where parsedProp.isComplianceCheck == "True" 
| summarize count() by ResourceGroup, tostring(parsedProp.tags.businessowner)

这些分析函数可能处理器密集型函数。 仅当查询使用经过格式设置的数据中的多个属性时,才使用这些函数。 否则,简单模式匹配处理速度更快。

下面的示例演示域控制器 TGT Preauth 类型的分解。 该类型仅存在于 EventData 字段中(这是一个 XML 字符串)。 此字段中的任何其他数据都不需要。 在这种情况下,分析用于挑选出所需数据段。

SecurityEvent
| where EventID == 4768
| parse EventData with * 'PreAuthType">' PreAuthType '</Data>' * 
| summarize count() by PreAuthType

使用函数模拟表

你可能有多个查询都对特定表执行相同分析。 在这种情况下,创建一个函数以返回经过分析的数据,而不是在每个查询中复制分析逻辑。 随后可以在其他查询中使用函数别名来代替原始表。

请考虑上述以逗号分隔的自定义日志示例。 若要在多个查询中使用经过分析的数据,请使用以下查询创建函数,并使用别名 MyCustomCSVLog 保存该函数。

MyCustomCSVLog_CL
| extend CSVFields = split(RawData, ',')
| extend DateTime  = tostring(CSVFields[0])
| extend Code      = toint(CSVFields[1]) 
| extend Status    = tostring(CSVFields[2]) 
| extend Message   = tostring(CSVFields[3]) 

现在可在查询中使用别名 MyCustomCSVLog 代替实际表名,如下所示:

MyCustomCSVLog
| summarize count() by Status,Code

后续步骤

了解日志查询以便分析从数据源和解决方案中收集的数据。