Syslog with Elasticsearch
本指南说明如何使用 Elasticsearch、Kibana 和 Fleet Server 配合 MikroTik RouterOS 设备配置 Syslog 数据收集与分析。内容涵盖先决条件、Elastic 和 RouterOS 组件的设置步骤，包括 agent policy 创建和 Custom UDP logs 集成配置。
将 Syslog 与 Elasticsearch 配合使用
简介
Elasticsearch 是一个流行的 NoSQL 数据库，可用于存储各种数据，包括 Syslog 数据。结合 Kibana，你可以创建一个强大的工具来分析来自 RouterOS 设备的 Syslog 数据。本指南依赖 Elasticsearch 集成，要使其正常工作，你需要一个可用的 Elasticsearch 环境。本指南不会涵盖 Elasticsearch 和 Kibana 的安装说明，但会涵盖设置 Syslog 数据收集与分析的相关步骤。
使用 Elasticsearch 可以进行许多可能的配置，但为了本指南的目的，我们将使用以下原则：
- 一台 RouterOS (10.0.0.1) 设备将 Syslog 数据发送到运行 Custom UDP logs的服务器 (10.0.0.2)。
- 运行 Custom UDP logs 集成的服务器 (10.0.0.2) 摄取 Syslog 数据、处理数据并将其发送到 Fleet Server (10.0.0.3)。
- Fleet Server (10.0.0.3) 将数据存储在 Elasticsearch (10.0.0.4) 中。
- Kibana (10.0.0.5) 从 Elasticsearch (10.0.0.4) 检索数据、分析数据并允许你搜索数据。
信息
本指南不会使用 Logstash 作为分析 Syslog 数据的一部分；它已被 Fleet Server 取代。
信息
可以将 Elasticsearch、Kibana、Fleet Server 和 Custom UDP logs 集成安装在同一台设备上。
先决条件
- 设置 Elasticsearch。
:::info Elasticsearch 在多个平台上得到广泛支持。建议设置一个 Elasticsearch 节点集群。 ::: - 设置 Kibana。
:::info Kibana 可以安装在安装 Elasticsearch 的同一台设备上，但出于性能考虑，也可以将其安装在单独的设备上。虽然不使用 Kibana 也可以分析 Syslog 数据，但需要编写自己的 API 请求，而 Kibana 非常易于使用，并具有广泛的功能。 ::: - 设置 Fleet Server。
:::info 可以在安装 Elasticsearch 和/或 Kibana 的同一台设备上设置 Fleet Server。建议将 Fleet Server 安装在不同的设备上。有关硬件和拓扑要求的建议，请参阅 Elasticsearch 手册。 :::
设置
设置说明分为两部分：Elastic（有关 Elasticsearch、Kibana 和 Fleet Server 的配置）和 RouterOS（与你的 RouterOS 设备相关的配置）。
Elastic
警告
某些步骤可能会随时间变化。请参阅 Elastic 的手册以查找最新的步骤。
-
在主菜单下打开 Fleet 部分。
-
打开 "Agent policies" 部分。
-
按 "Create agent policy" 按钮以创建新的 Agent Policy。
-
为该策略命名，例如 "Syslog policy"，根据需要调整高级设置，然后创建该策略。或者你可以使用下面的 API 请求：
POST kbn:/api/fleet/agent_policies { "name": "Syslog policy", "description": "", "namespace": "default", "monitoring_enabled": [ "logs", "metrics" ], "inactivity_timeout": 1209600, "is_protected": false } -
通过点击名称打开你新创建的策略。
-
按 "Add integration"。
-
搜索 "Custom UDP logs" 并按 "Add Custom UDP logs"。
-
调整配置，并确保：
- 将 "Listen Address" 设置为将要运行 Custom UDP logs 集成的服务器的 IP 地址，在此示例中该地址应为 "10.0.0.2"。
- 将 "Listen Port" 设置为 "5514"。
- 将 "Dataset name" 设置为 "routeros"。
- 将 "Ingest Pipeline" 设置为 "logs-routeros@custom"。
- 将 "Syslog Parsing" 设置为 "Yes"。
-
保存集成。
-
按 "Add Elastic Agent to your host" 按钮。
-
按照说明将 Elastic Agent 添加到你的主机。
:::info 官方 Elastic 手册建议将 Elastic Agent 安装为 Fleet-managed。考虑遵循该建议，因为当 agent 由 Fleet 管理时更容易管理。 ::: -
转到主菜单上的 "Stack Management"，然后打开 "Ingest Pipelines"。
-
通过按 "Create pipeline" 然后 "New pipeline" 来创建新的 Ingest Pipeline。
-
将 "Name" 设置为 "logs-routeros@custom"。
-
按 "Import processors" 并粘贴以下 processors：
{ "processors": [ { "grok": { "field": "message", "patterns": [ "^first L2TP UDP packet received from %{IP:source.ip}$", "^login failure for user %{USERNAME:user.name} from %{IP:source.ip} via %{DATA:service.name}$", "^%{USERNAME:user.name} logged in, %{IP:client.ip} from %{IP:source.ip}$", "^dhcp alert on %{DATA}: discovered unknown dhcp server, mac %{MAC:source.mac}, ip %{IP:source.ip}$", "in:%{DATA} out:%{DATA}, ?(connection-state:%{DATA},|)?(src-mac %{MAC:source.mac},|) proto %{DATA:network.transport} \\(%{DATA}\\), %{IP:source.ip}:?(%{INT:source.port}|)->%{IP:destination.ip}:?(%{INT:destination.port}|), len %{INT:network.bytes}$", "in:%{DATA} out:%{DATA}, ?(connection-state:%{DATA},|)?(src-mac %{MAC:source.mac},|) proto %{DATA:network.transport}, %{IP:source.ip}:?(%{INT:source.port}|)->%{IP:destination.ip}:?(%{INT:destination.port}|), len %{INT:network.bytes}$", "^%{DATA:network.name} (deassigned|assigned) %{IP:client.ip} for %{MAC:client.mac} %{DATA}$", "^%{DATA:user.name} logged out, %{INT:event.duration} %{INT} %{INT} %{INT} %{INT} from %{IP:client.ip}$", "^user %{DATA:user.name} logged out from %{IP:source.ip} via %{DATA:service.name}$", "^user %{DATA:user.name} logged in from %{IP:source.ip} via %{DATA:service.name}$", "^%{DATA:network.name} client %{MAC:client.mac} declines IP address %{IP:client.ip}$", "^%{DATA:network.name} link up \\(speed %{DATA}\\)$", "^%{DATA:network.name} link down$", "^user %{DATA:user.name} authentication failed$", "^%{DATA:network.name} fcs error on link$", "^phase1 negotiation failed due to time up %{IP:source.ip}\\[%{INT:source.port}\\]<=>%{IP:destination.ip}\\[%{INT:destination.port}\\] %{DATA}:%{DATA}$", "^%{DATA:network.name} (learning|forwarding)$", "^user %{DATA:user.name} is already active$", "^%{GREEDYDATA}$" ] } }, { "lowercase": { "field": "network.transport", "ignore_missing": true } }, { "append": { "ignore_failure": true, "field": "event.category", "description": "Enrich logon events", "allow_duplicates": false, "value": [ "authentication" ], "if": "ctx.message =~ /(login failure for user|logged in from|logged in,)/" } }, { "append": { "ignore_failure": true, "field": "event.outcome", "description": "Enrich successful login events", "allow_duplicates": false, "value": [ "success" ], "if": "ctx.message =~ /(logged in from|logged in,)/" } }, { "append": { "ignore_failure": true, "field": "event.outcome", "description": "Enrich failed login events", "allow_duplicates": false, "value": [ "failure" ], "if": "ctx.message =~ /(login failure for user)/" } }, { "append": { "ignore_failure": true, "field": "event.category", "description": "Enrich network events", "allow_duplicates": false, "value": [ "network" ], "if": "ctx.message =~ /( fcs error on link| link down| link up)/" } }, { "append": { "ignore_failure": true, "field": "event.outcome", "description": "Enrich network failures", "allow_duplicates": false, "value": [ "failure" ], "if": "ctx.message =~ /( fcs error on link)/" } }, { "append": { "ignore_failure": true, "field": "event.category", "allow_duplicates": false, "value": [ "session" ], "if": "ctx.message =~ /(logged out)/" } }, { "append": { "ignore_failure": true, "field": "event.category", "allow_duplicates": false, "value": [ "threat" ], "if": "ctx.message =~ /(from address that has not seen before)/" } }, { "append": { "ignore_failure": true, "field": "service.name", "value": [ "l2tp" ], "if": "ctx.message =~ /(^L2TP\\/IPsec VPN)/" } }, { "geoip": { "ignore_failure": true, "ignore_missing": true, "field": "source.ip", "target_field": "source.geo" } }, { "geoip": { "ignore_failure": true, "ignore_missing": true, "field": "destination.ip", "target_field": "destination.geo" } }, { "geoip": { "ignore_failure": true, "ignore_missing": true, "field": "client.ip", "target_field": "client.geo" } } ] } -
按 "Save pipeline"。
-
转到主菜单上的 "Stack Management"，然后选择 "Index Management"，再选择 "Component templates"。
-
通过按 "Create component template" 创建新模板。
-
将 "Name" 设置为 "logs-routeros@custom"。
-
在 "Index settings" 部分下，粘贴以下内容：
{ "index": { "lifecycle": { "name": "logs" }, "default_pipeline": "logs-routeros@custom" } }:::info 将 ILM policy 更改为你需要的策略名称。"logs" 策略是可能被其他日志使用的默认策略。 :::
-
在 "Mappings" 部分下，按 "Load JSON" 并粘贴以下内容：
{ "dynamic_templates": [], "properties": { "service": { "type": "object", "properties": { "name": { "type": "keyword" } } }, "destination": { "type": "object", "properties": { "port": { "type": "long" }, "ip": { "type": "ip" } } }, "host": { "type": "object", "properties": { "ip": { "type": "ip" } } }, "client": { "type": "object", "properties": { "ip": { "type": "ip" }, "mac": { "type": "keyword" } } }, "source": { "type": "object", "properties": { "geo": { "type": "object", "properties": { "continent_name": { "ignore_above": 1024, "type": "keyword" }, "region_iso_code": { "ignore_above": 1024, "type": "keyword" }, "city_name": { "ignore_above": 1024, "type": "keyword" }, "country_iso_code": { "ignore_above": 1024, "type": "keyword" }, "country_name": { "ignore_above": 1024, "type": "keyword" }, "location": { "type": "geo_point" }, "region_name": { "ignore_above": 1024, "type": "keyword" } } }, "as": { "type": "object", "properties": { "number": { "type": "long" }, "organization": { "type": "object", "properties": { "name": { "ignore_above": 1024, "type": "keyword", "fields": { "text": { "type": "match_only_text" } } } } } } }, "address": { "ignore_above": 1024, "type": "keyword" }, "port": { "type": "long" }, "domain": { "ignore_above": 1024, "type": "keyword" }, "ip": { "type": "ip" }, "mac": { "type": "keyword" } } }, "event": { "type": "object", "properties": { "duration": { "type": "long" }, "category": { "type": "keyword" }, "outcome": { "type": "keyword" } } }, "message": { "type": "match_only_text" }, "user": { "type": "object", "properties": { "name": { "type": "keyword" } } }, "network": { "type": "object", "properties": { "bytes": { "type": "long" }, "name": { "type": "keyword" }, "transport": { "type": "keyword" } } }, "tags": { "ignore_above": 1024, "type": "keyword" } } } -
按 "Next"，然后按 "Save component template"。
:::warning 如果存在同名的 component template，则改为编辑现有的模板。 ::: -
转到主菜单上的 "Stack Management"，然后选择 "Index Management"，再选择 "Index templates"。
-
通过按 "Create template" 创建新模板。
-
将 "Name" 设置为 "logs-routeros"。
-
将 "Index patterns" 设置为 "logs-routeros-*"。
-
在 "Component templates" 部分下，将以下模板添加到你的新 Index template：
logs@settings logs-routeros@custom ecs@mappings -
按 "Next"，然后按 "Save template"。
-
确保你已在主机上以及从 RouterOS 设备 (10.0.0.1) 路径中的其他位置打开 5514/UDP 端口。
-
你的 Elastic Agent 现在已准备好接收 Syslog 数据！
RouterOS
-
在你的 RouterOS 设备 (10.0.0.1) 上配置 Logging action 设置：
/system/logging/action set [find where name="remote"] remote-log-format=syslog remote=10.0.0.2 remote-port=5514 syslog-facility=syslog -
添加你希望从 RouterOS 设备 (10.0.0.1) 接收的主题，例如：
/system/logging add action=remote topics=info add action=remote topics=error add action=remote topics=critical add action=remote topics=warning add action=remote topics=bridge,stp -
你现在应该开始看到 Syslog 数据被摄取。
-
继续本指南以开始使用 Kibana。
使用 Kibana
Kibana 允许你搜索已摄取的 Syslog 数据。要查看已摄取的日志，请执行以下操作：
-
从主菜单打开 "Discover"。
-
添加过滤器，并使用以下参数：
Select a field: data_stream.dataset Select operator: IS Select a value: routeros -
为简单起见，我们建议在 Discover 菜单中搜索字段，搜索 "message"，然后将该字段添加到视图中。
-
我们还建议搜索 "log.syslog.hostname" 字段并将其也添加到视图中。
-
考虑保存搜索以便以后更容易访问。
-
完成。
提示
虽然搜索日志可能很有用，但你更可能想要为某些活动创建警报，并创建连接器以将警报发送到电子邮件、webhook、聊天和其他选项。考虑启用 Spike in failed logon events 规则，以对过多的失败登录尝试发出警报。你还可以创建一个阈值规则，并将其设置为在固定数量的失败登录后发出警报。