如何使用 Grafana Loki 分析 SQL Server Audit 日誌

上一篇文章已经教大家如何针对特定事件分析 Windows 登入日誌，该技巧几乎可以运用在任何类型的日誌。今天就是要来教大家高阶一点的应用，如何针对 SQL Server Audit 日誌进行分析与仪表板的製作。

首先我们要使用 SQL Server Audit 功能，没用过的朋友请参考此篇文章。

将使用资料库或者资料表执行 CREATE、ALTER 或 DROP 作为演示範例

建立稽核
从物件总管中点选执行个体 > 安全性 > 稽核，新增稽核。

这个稽核是用于定义蒐集到的稽核纪录如何储存，选择 Application Log。

刚被建立的稽核会处于停用状态，必须要启用纪录才能够被保存。

新增伺服器稽核规格
从物件总管中点选执行个体 > 安全性 > 伺服器稽核规格，新增伺服器稽核规格。

输入伺服器稽核规格的名称并选择上面所建立的稽核

我们想要稽核资料库或者资料表执行 CREATE、ALTER 或 DROP 陈述式时，选择以下稽核动作群组名称。

DATABASE_OBJECT_CHANGE_GROUPSCHEMA_OBJECT_CHANGE_GROUP

关于稽核动作群组名称的详细描述，请参考此篇文章。

完成伺服器稽核规格设定后，一样需要启用才会开始进行稽核。
需要注意的是伺服器稽核规格与稽核都必须要同时启动，才能够正确纪录。

检视稽核纪录
确认建立资料库与资料表的动作是否会被稽核记录

利用下列的 T-SQL 指令码建立资料库

USE [master]GOCREATE DATABASE [Database_1] CONTAINMENT = NONE ON  PRIMARY ( NAME = N'Database_1', FILENAME = N'C:\Program Files\Microsoft SQL Server\MSSQL13.MSSQLSERVER\MSSQL\DATA\Database_1.mdf' , SIZE = 8192KB , MAXSIZE = UNLIMITED, FILEGROWTH = 65536KB ) LOG ON ( NAME = N'Database_1_log', FILENAME = N'C:\Program Files\Microsoft SQL Server\MSSQL13.MSSQLSERVER\MSSQL\DATA\Database_1_log.ldf' , SIZE = 8192KB , MAXSIZE = 2048GB , FILEGROWTH = 65536KB )GO

利用下列的 T-SQL 指令码建立资料表

USE [Database_1]GOCREATE TABLE [dbo].[Table_1]( [Column_1] [nchar](10) NULL) ON [PRIMARY]GO

确认建立资料库与资料表的动作都被稽核记录下来了

利用下列的 T-SQL 指令码新增一些测试资料，等一下建立仪表板会使用到。

USE [Database_1]GOCREATE TABLE [dbo].[Table_2]( [Column_1] [nchar](10) NULL) ON [PRIMARY]GOCREATE VIEW [dbo].[View_1] AS  SELECT *  FROM [dbo].[Table_1]GOCREATE VIEW [dbo].[View_2] AS  SELECT *  FROM [dbo].[Table_2]GOALTER VIEW [dbo].[View_2] AS  SELECT *  FROM [dbo].[Table_1]UNION ALL SELECT *  FROM [dbo].[Table_2]GODROP VIEW [dbo].[View_2] GO

Promtail
使用到的 promtail-local-config.yaml 範例如下，主要是将 Application 日誌推送到 Loki 并将 source、event_id 与 leveltext 栏位贴上标籤。

server:  http_listen_port: 9080  grpc_listen_port: 0positions:  filename: "./positions.yaml"clients:  - url: http://your_loki_ip:3100/loki/api/v1/push    scrape_configs:- job_name: windows  windows_events:    eventlog_name: "Application"    use_incoming_timestamp: true    xpath_query: '*'    bookmark_path: "./bookmark-application.xml"    exclude_event_data: true    exclude_user_data: true    labels:      logsource: windows-eventlog  pipeline_stages:  - json:      expressions:        source: source        eventID: event_id        level: levelText  - labels:      source:      eventID:      level:

一定有朋友好奇，那我能不能把稽核纪录如何储存到 File。

再透过 static_configs 读取 AuditLog 底下的日誌呢？

server:  http_listen_port: 9080  grpc_listen_port: 0positions:  filename: "./positions.yaml"clients:  - url: http://your_loki_ip:3100/loki/api/v1/push    scrape_configs:- job_name: mssql  static_configs:  - targets:      - localhost    labels:      host: your_hostname      job: auditlog      __path__: D:\AuditLog\*.sqlaudit

答案是不行的，该日誌档已经被 SQL Server 加密过了。

建立一个 SQL Server Audit 事件日誌仪表板，点选 Setting 中的 Variables。

我们先建立两个变数 computer 与 search 之后会过滤会使用到。

变数 computer
General

Name 输入 computer 当作变数名称使用Label 输入电脑做为显示名称

Query options

Data Source 选择 LokiQuery Type 选择 Label vaules，Label 选择 computer

Regex

就依自己的命名规则，看如何筛选出 SQL Server 啰

Selection options

勾选 Multi-value勾选 Include All option

若有配置正确，Preview of vaules 会出现您想要的变数数值，按下 Apply。

变数 search
回到 Setting 中的 Variables，这次我们建立一个 Text box 类型的变数。

General

Name 输入 search 当作变数名称使用Label 输入搜寻做为显示名称

事件纪录 Panel
新增一个 Panel，选择 Table 可视化。资料来源选择 Loki，顺便填入 Title 为事件纪录。

Label browser 输入 LogQL 语法如下

日誌流选择器带入变数 computer 过滤日誌源日誌管道带入变数 search 过滤日誌内容透过 pattern 将解析到的栏位存成标籤透过 label_format 将 action_id 与 class_type 转成名称过滤 action_id 为 CREATE、ALTER、DROP

可以用 (?i) 做为正规表达式的前缀，切换为不区分大小写。

{computer=~"$computer", source="MSSQLSERVER", eventID="33205"}|~ "(?i)$search" | pattern `<_>action_id:<action_id>\n<_>` | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`| action_id =~ "(CREATE|ALTER|DROP)" | pattern `<_>class_type:<class_type>\n<_>` | pattern `<_>database_name:<database_name>\n<_>` | pattern `<_>object_name:<object_name>\n<_>` | pattern `<_>schema_name:<schema_name>\n<_>` | pattern `<_>server_instance_name:<server_instance_name>\n<_>` | pattern `<_>server_principal_name:<server_principal_name>\n<_>` | pattern `<_>object_name:<object_name>\n<_>` | pattern `<_>statement:<statement>\nadditional_information<_>` | pattern `<_>succeeded:<succeeded>\n<_>` | label_format class_type=`{{.class_type | trim | replace "DB" "DATABASE" | replace "U" "TABLE" | replace "V" "VIEW" | replace "P" "STORED PROCEDURE"}}`| label_format statement=`{{.statement | replace "\\r\\n" " " | replace "\\r" " " | replace "\\n" " " | replace "\u005c\u005c" "\u005c"}}`

可以透过下列的 T-SQL 指令来查找 action_id 与 class_type 对应的名称

select * from sys.dm_audit_actionsselect * from sys.dm_audit_class_type_map

Table 可视化允许 Pagination 与 Column filter

先点选 labels 栏位看一下，确认想要分析的栏位都有正确解析到。

编辑 Panel 切到 Transform，加入 Extract fields，Source 选择 labels。

如此一来，就可以把解析到的标籤们转换成多个栏位进行使用。

但是这样栏位太多了，可以透过 Transform 加入 Organize fields 隐藏或排序栏位，显示有意义的栏位即可。

画面就乾净许多了，也可以针对栏位进行过滤。

事件纪录 Panel 基本上就搞定了

时间轴 Panel
新增一个 Panel，选择 Time Series 可视化。资料来源选择 Loki，顺便填入 Title 为时间轴。

Graph Style 选择 Bar。Query Legend 填入 {{action_id}}

Label browser 输入 LogQL 语法如下，日誌流选择器带入我们设定的变数。可根据下拉选项的过滤，依照 action_id 的数量进行统计。

sum by (action_id) (    count_over_time({computer=~"$computer", source="MSSQLSERVER", eventID="33205"}    |~ "(?i)$search"     | pattern `<_>action_id:<action_id>\n<_>`     | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`    | action_id =~ "(CREATE|ALTER|DROP)"    [$__interval]))

动作统计 Panel
新增一个 Panel，选择 Pie Chart 可视化。资料来源选择 Loki，顺便填入 Title 为动作统计。

Query Legend 填入 {{action_id}}Value Options Calculation 选择 TotalPie Chart Type 选择 DonutPie Chart Labels 选择 PercentLegend Placement 选择 Right

Label browser 输入 LogQL 语法如下，日誌流选择器带入我们设定的变数。可根据下拉选项的过滤，依照 action_id 的数量进行统计。

sum by (action_id) (    count_over_time({computer=~"$computer", source="MSSQLSERVER", eventID="33205"}    |~ "(?i)$search"     | pattern `<_>action_id:<action_id>\n<_>`     | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`    | action_id =~ "(CREATE|ALTER|DROP)"    [$__interval]))

类别类型统计 Panel
新增一个 Panel，选择 Pie Chart 可视化。资料来源选择 Loki，顺便填入 Title 为类别类型统计。

Query Legend 填入 {{class_type}}其他选项同上

Label browser 输入 LogQL 语法如下，日誌流选择器带入我们设定的变数。可根据下拉选项的过滤，依照 class_type 的数量进行统计。

sum by (class_type) (    count_over_time({computer=~"$computer", source="MSSQLSERVER", eventID="33205"}    |~ "(?i)$search"     | pattern `<_>action_id:<action_id>\n<_>`     | pattern `<_>class_type:<class_type>\n<_>`    | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`    | label_format class_type=`{{ .class_type | trim | replace "DB" "DATABASE" | replace "U" "TABLE" | replace "V" "VIEW" | replace "P" "STORED PROCEDURE"}}`    | action_id =~ "(CREATE|ALTER|DROP)"    [$__interval]))

资料库名称统计 Panel
新增一个 Panel，选择 Pie Chart 可视化。资料来源选择 Loki，顺便填入 Title 为资料库名称统计。

Query Legend 填入 {{database_name}}其他选项同上

Label browser 输入 LogQL 语法如下，日誌流选择器带入我们设定的变数。可根据下拉选项的过滤，依照 database_name 的数量进行统计。

sum by (database_name) (    count_over_time({computer=~"$computer", source="MSSQLSERVER", eventID="33205"}    |~ "(?i)$search"     | pattern `<_>action_id:<action_id>\n<_>`     | pattern `<_>database_name:<database_name>\n<_>`     | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`    | action_id =~ "(CREATE|ALTER|DROP)"    [$__interval]))

物件名称统计 Panel
新增一个 Panel，选择 Pie Chart 可视化。资料来源选择 Loki，顺便填入 Title 为物件名称统计。

Query Legend 填入 {{object_name}}其他选项同上

Label browser 输入 LogQL 语法如下，日誌流选择器带入我们设定的变数。可根据下拉选项的过滤，依照 object_name 的数量进行统计。

sum by (object_name) (    count_over_time({computer=~"$computer", source="MSSQLSERVER", eventID="33205"}    |~ "(?i)$search"     | pattern `<_>action_id:<action_id>\n<_>`     | pattern `<_>object_name:<object_name>\n<_>`     | label_format action_id=`{{.action_id | trim | replace "CR" "CREATE" | replace "AL" "ALTER" | replace "DR" "DROP"}}`    | action_id =~ "(CREATE|ALTER|DROP)"    [$__interval]))

最后完成的仪表板如下，我们也可以使用关键字来进行过滤。

若您是想要分析资料操作语言(DML)，例如 SELECT、INSERT、UPDAT 与 DELETE，只要仿照上面的步骤执行应该不难实作出来的。

例如我们想要稽核资料表执行 SELECT、INSERT、UPDATE 或 DELETE 陈述式时，选择以下稽核动作群组名称。

SCHEMA_OBJECT_ACCESS_GROUP

提供的参考的 LogQL 语法如下

{computer=~"$computer", source="MSSQLSERVER", eventID="33205"}|~ "(?i)$search" | pattern `<_>action_id:<action_id>\n<_>` | label_format action_id=`{{.action_id | trim | replace "SL" "SELECT" | replace "IN" "INSERT" | replace "UP" "UPDATE" | replace "DL" "DELETE"}}`| action_id =~ "(SELECT|INSERT|UPDATE|DELETE)" | pattern `<_>class_type:<class_type>\n<_>` | pattern `<_>database_name:<database_name>\n<_>` | pattern `<_>object_name:<object_name>\n<_>` | pattern `<_>schema_name:<schema_name>\n<_>` | pattern `<_>server_instance_name:<server_instance_name>\n<_>` | pattern `<_>server_principal_name:<server_principal_name>\n<_>` | pattern `<_>object_name:<object_name>\n<_>` | pattern `<_>statement:<statement>\nadditional_information<_>` | pattern `<_>succeeded:<succeeded>\n<_>` | label_format class_type=`{{.class_type | trim | replace "DB" "DATABASE" | replace "U" "TABLE" | replace "V" "VIEW" | replace "P" "STORED PROCEDURE"}}`| label_format statement=`{{.statement | replace "\\r\\n" " " | replace "\\r" " " | replace "\\n" " "}}`

成功登入与失败的稽核纪录当然也没有问题

例如我们想要稽核已成功登入 SQL Server 或者失败的主体，选择以下稽核动作群组名称。

SUCCESSFUL_LOGIN_GROUPFAILED_LOGIN_GROUP

成功登入的 LogQL 语法如下

{computer=~"$computer", source="MSSQLSERVER", eventID="33205"}|~ "(?i)$search" | pattern `<_>action_id:<action_id>\n<_>` | label_format action_id=`{{.action_id | trim | replace "LGIS" "LOGIN SUCCEEDED"}}`| action_id ="LOGIN SUCCEEDED" | pattern `<_>class_type:<class_type>\n<_>` | label_format class_type=`{{.class_type | trim | replace "LX" "LOGIN"}}`| pattern `<_>server_instance_name:<server_instance_name>\n<_>` | pattern `<_>server_principal_name:<server_principal_name>\n<_>` | pattern `<_>network protocol:<network_protocol>\r\n<_>` | label_format network_protocol=`{{.network_protocol | trim}}`| label_format statement =""| pattern `<_>\u003caddress\u003e<ip_address>\u003c/address\u003e<_>` | server_principal_name !=""| network_protocol =~"(TCP/IP|LPC)"| ip_address !="local machine"

登入失败的 LogQL 语法如下

{computer=~"$computer", source="MSSQLSERVER", eventID="33205"}|~ "(?i)$search" | pattern `<_>action_id:<action_id>\n<_>` | label_format action_id=`{{.action_id | trim | replace "LGIF" "LOGIN FAILED"}}`| action_id ="LOGIN FAILED" | pattern `<_>class_type:<class_type>\n<_>` | label_format class_type =`{{.class_type | trim | replace "LX" "LOGIN"}}`| pattern `<_>\nserver_principal_name:<server_principal_name>\n<_>` | pattern `<_>statement:<statement>\n<_>` | pattern `<_>\u003caddress\u003e<ip_address>\u003c/address\u003e<_>`

再透过 Transform 加入 Merge，将这两段的结果整併在同个 Panel 即可。

今天的分享就到这边，希望有帮助到大家。

下一篇再来分享如何使用 Grafana Loki 警报规则并透过 Alertmanager 发送警告

参考文件

https://learn.microsoft.com/en-us/sql/relational-databases/system-dynamic-management-views/sys-dm-audit-actions-transact-sql?view=sql-server-ver16https://learn.microsoft.com/en-us/sql/relational-databases/system-dynamic-management-views/sys-dm-audit-class-type-map-transact-sql?view=sql-server-ver16