注意：与 Logstash 的集成需要 Manticore Buddy。如果集成不工作，请确保 Buddy 已安装。

Logstash 是一种日志管理工具，它从各种来源收集数据，实时转换数据，并发送到您想要的目标。它通常用作 Elasticsearch（一个开源分析和搜索引擎）的数据管道。

现在，Manticore 支持使用 Logstash 作为处理管道。这允许收集和转换后的数据像发送给 Elasticsearch 一样发送给 Manticore。目前，支持的版本是 7.6 到 9.2。

让我们来看看一个用于索引 dpkg.log 的简单 Logstash 配置文件示例，dpkg.log 是 Debian 包管理器的标准日志文件。该日志本身结构简单，如下所示：

2023-05-31 10:42:55 status triggers-awaited ca-certificates-java:all 20190405ubuntu1.1
2023-05-31 10:42:55 trigproc libc-bin:amd64 2.31-0ubuntu9.9 <none>
2023-05-31 10:42:55 status half-configured libc-bin:amd64 2.31-0ubuntu9.9
2023-05-31 10:42:55 status installed libc-bin:amd64 2.31-0ubuntu9.9
2023-05-31 10:42:55 trigproc systemd:amd64 245.4-4ubuntu3.21 <none>

下面是一个 Logstash 配置示例：

input {
  file {
    path => ["/var/log/dpkg.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    mode => "read"
    exit_after_read => "true"
   file_completed_action => "log"
   file_completed_log_path => "/dev/null"
  }
}
output {
  elasticsearch {
   index => " dpkg_log"
   hosts => ["http://localhost:9308"]
   ilm_enabled => false
   manage_template => false
  }
}

请注意，在继续之前，需要强调一个关键注意事项：Manticore 不支持 Elasticsearch 的日志模板管理和索引生命周期管理功能。由于这些功能在 Logstash 中默认启用，必须在配置中显式禁用它们。此外，output 配置段中的 hosts 选项必须对应 Manticore 的 HTTP 监听端口（默认是 localhost:9308）。

配置根据您使用的 Logstash 版本而有所不同。

对于 Logstash 7.17，基本配置相当简单，不需要额外的 ILM 设置：

input {
  file {
    path => ["/var/log/dpkg.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    mode => "read"
    exit_after_read => "true"
    file_completed_action => "log"
    file_completed_log_path => "/dev/null"
  }
}
output {
  elasticsearch {
    index => "dpkg_log"
    hosts => ["http://localhost:9308"]
  }
}

运行命令：

logstash -f logstash.conf

从 8.0 版本开始，ILM（索引生命周期管理）和模板管理默认启用，必须显式禁用以兼容 Manticore：

input {
  file {
    path => ["/var/log/dpkg.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    mode => "read"
    exit_after_read => "true"
    file_completed_action => "log"
    file_completed_log_path => "/dev/null"
  }
}
output {
  elasticsearch {
    index => "dpkg_log"
    hosts => ["http://localhost:9308"]
    ilm_enabled => false
    manage_template => false
  }
}

对于 9.0 和 9.1 版本，Logstash 需要以超级用户身份运行。在启动前设置环境变量：

export ALLOW_SUPERUSER=1
logstash -f logstash.conf

从 9.2 版本开始，推荐通过配置文件来配置超级用户设置，而不是使用环境变量，提供了更持久和便于管理的方案。

配置文件（例如 logstash.conf）：

input {
  file {
    path => ["/var/log/dpkg.log"]
    start_position => "beginning"
    sincedb_path => "/dev/null"
    mode => "read"
    exit_after_read => "true"
    file_completed_action => "log"
    file_completed_log_path => "/dev/null"
  }
}
output {
  elasticsearch {
    index => "dpkg_log"
    hosts => ["http://localhost:9308"]
    ilm_enabled => false
    manage_template => false
  }
}

创建 /etc/logstash/logstash.yml：

allow_superuser: true

运行命令：

logstash --path.settings=/etc/logstash -f logstash.conf

按照上述方式调整配置后，您可以运行 Logstash，dpkg 日志中的数据将传递给 Manticore 并正确索引。

下面是创建的表的结果模式以及插入文档的示例：

mysql> DESCRIBE dpkg_log;
+------------------+--------+---------------------+
| Field            | Type   | Properties          |
+------------------+--------+---------------------+
| id               | bigint |                     |
| message          | text   | indexed stored      |
| @version         | text   | indexed stored      |
| @timestamp       | text   | indexed stored      |
| path             | text   | indexed stored      |
| host             | text   | indexed stored      |
+------------------+--------+---------------------+

mysql> SELECT * FROM dpkg_log LIMIT 1\G
*************************** 1. row ***************************
id: 7280000849080746110
host: logstash-db848f65f-lnlf9
message: 2023-04-12 02:03:21 status unpacked libc-bin:amd64 2.31-0ubuntu9
path: /var/log/dpkg.log
@timestamp: 2023-06-16T09:23:57.405Z
@version: 1

注意：与 Filebeat 的集成需要 Manticore Buddy。如果无法正常工作，请确保已安装 Buddy。

Filebeat 是一个轻量级的记录数据转发和集中工具。安装作为代理后，它会监控您指定的日志文件或位置，收集日志事件，并将其转发以供索引，通常是发送到 Elasticsearch 或 Logstash。

现在，Manticore 也支持将 Filebeat 用作处理管道。这允许将收集和转换后的数据像发送到 Elasticsearch 一样发送到 Manticore。目前支持的版本为 7.17-9.2。

配置根据您使用的 Filebeat 版本而异。

重要：Filebeat 版本 7.17.0、8.0.0 和 8.1.0 在使用 glibc 2.35+（用于 Ubuntu 22.04 及更高版本的发行版）时存在已知问题。这些版本可能会因“Fatal glibc error: rseq registration failed”而崩溃。为解决此问题，请添加如下所示的 seccomp 配置。

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/dpkg.log
  close_eof: true
  scan_frequency: 1s
output.elasticsearch:
  hosts: ["http://localhost:9308"]
  index: "dpkg_log"
  compression_level: 0
  allow_older_versions: true  # Required for 8.1
# Fix for glibc 2.35+ compatibility (Ubuntu 22.04+)
seccomp:
  default_action: allow
  syscalls:
    - action: allow
      names:
        - rseq
setup.ilm.enabled: false
setup.template.enabled: false
setup.template.name: "dpkg_log"
setup.template.pattern: "dpkg_log"

参考：Issue #30576，PR #30620

对于 8.1 到 8.10 版本，需要添加 allow_older_versions 选项：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/dpkg.log
  close_eof: true
  scan_frequency: 1s
output.elasticsearch:
  hosts: ["http://localhost:9308"]
  index: "dpkg_log"
  compression_level: 0
  allow_older_versions: true
setup.ilm.enabled: false
setup.template.enabled: false
setup.template.name: "dpkg_log"
setup.template.pattern: "dpkg_log"

从 8.11 版本起，默认启用输出压缩，因此您必须显式设置 compression_level: 0 以保证与 Manticore 的兼容：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/dpkg.log
  close_eof: true
  scan_frequency: 1s
output.elasticsearch:
  hosts: ["http://localhost:9308"]
  index: "dpkg_log"
  compression_level: 0
  allow_older_versions: true
setup.ilm.enabled: false
setup.template.enabled: false
setup.template.name: "dpkg_log"
setup.template.pattern: "dpkg_log"

Filebeat 9.0 引入了重大架构更改，替换了 log 输入类型为 filestream。从 9.0 版本开始，默认的文件识别方法也更改为指纹识别，要求文件至少为 1024 字节（参见 issue #44780）。为了使 Manticore 兼容任意大小的文件，您必须禁用指纹识别。

以下是 Filebeat 9.0 及之后所有版本所需的配置：

filebeat.inputs:
- type: filestream
  id: dpkg-log-input
  enabled: true
  paths:
    - /var/log/dpkg.log
  prospector.scanner.check_interval: 1s
  prospector.scanner.fingerprint.enabled: false
output.elasticsearch:
  hosts: ["http://localhost:9308"]
  index: "dpkg_log"
  compression_level: 0
  allow_older_versions: true
setup.ilm.enabled: false
setup.template.enabled: false
setup.template.name: "dpkg_log"
setup.template.pattern: "dpkg_log"

针对 Filebeat 9.0+ 的重要说明：

• type: filestream 输入替代了旧的 type: log
• prospector.scanner.fingerprint.enabled: false 设置是必须的，用以禁用基于指纹的文件识别，确保对小于 1024 字节的文件可靠处理
• filestream 输入要求有 id 字段，且必须唯一

一旦使用此配置启动 Filebeat，日志数据将被发送到 Manticore 并正确建立索引。以下是 Manticore 创建的表的结果模式及插入文档示例：

mysql> DESCRIBE dpkg_log;
+------------------+--------+--------------------+
| Field            | Type   | Properties         |
+------------------+--------+--------------------+
| id               | bigint |                    |
| @timestamp       | text   | indexed stored     |
| message          | text   | indexed stored     |
| log              | json   |                    |
| input            | json   |                    |
| ecs              | json   |                    |
| host             | json   |                    |
| agent            | json   |                    |
+------------------+--------+--------------------+

mysql> SELECT * FROM dpkg_log LIMIT 1\G
*************************** 1. row ***************************
id: 7280000849080753116
@timestamp: 2023-06-16T09:27:38.792Z
message: 2023-04-12 02:06:08 status half-installed libhogweed5:amd64 3.5.1+really3.5.1-2
input: {"type":"filestream"}
ecs: {"version":"1.6.0"}
host: {"name":"logstash-db848f65f-lnlf9"}
agent: {"ephemeral_id":"587c2ebc-e7e2-4e27-b772-19c611115996","id":"2e3d985b-3610-4b8b-aa3b-2e45804edd2c","name":"logstash-db848f65f-lnlf9","type":"filebeat","version":"7.10.0","hostname":"logstash-db848f65f-lnlf9"}
log: {"offset":80,"file":{"path":"/var/log/dpkg.log"}}

Kibana 是一个可视化界面，允许您探索、可视化并为日志数据创建仪表盘。将 Kibana 与 Manticore Search 集成，可以使 Kibana 可视化加载速度比 Elasticsearch 快最多 3 倍，如此 演示 所示。此集成使用户能够通过交互式仪表盘、自定义可视化和实时搜索功能无缝分析数据。它还通过支持 Logstash 和 Filebeat 等工具简化了处理多样化数据源的过程，实现了流畅的数据摄取，是日志分析工作流的理想选择。

1. 下载 Kibana：确保下载与 Manticore 兼容的 Kibana 版本。目前，推荐并测试的版本是 7.6.0。其他 7.x 版本可能可用，但可能会引入问题。不支持 8.x 版本。
2. 验证 Manticore：确保您的 Manticore 实例正在运行且其 HTTP API 可访问（默认地址：http://localhost:9308）。

1. 打开 Kibana 配置文件 (kibana.yml)。
2. 设置您的 Manticore 实例的 URL：

   elasticsearch.hosts: ["http://localhost:9308"]

3. 启动 Kibana 并在浏览器中打开 http://localhost:5601。如有必要，将 localhost 替换为您的服务器 IP 或主机名。

注意：Manticore 在与 Kibana 配合使用时不需要设置认证。同时请注意，Manticore 必须以实时模式运行，才能与 Kibana 集成。

    listen = 127.0.0.1:9308:http
    pid_file = /var/run/manticore/searchd.pid
    data_dir = /var/lib/manticore
 }

## Supported Features
### Discover
- Use the **Discover** tab in Kibana to search and filter your data interactively.
- Refine your searches using the query bar with simple queries in the %Kibana query language [https://www.elastic.co/guide/en/kibana/current/kuery-query.html] %.

### Visualizations
- Navigate to **Visualizations** to create custom visualizations:
  - Create a table pattern (it’s called an 'index pattern' in Kibana) if one doesn’t already exist to define your data source.
  - Choose a visualization type (e.g., bar chart, line chart, or pie chart).
  - Configure your visualization, execute it, and explore your data.
  - Save your visualizations for future use.

### Dashboards
- Access **Dashboards** to create or view interactive dashboards:
  - Add visualizations, filters, or controls for a personalized experience.
  - Interact with your data directly from the dashboard.
  - Save dashboards for future use.

### Management
- Go to **Management > Kibana** to customize settings like default time zones and visualization preferences.

## Limitations
- Currently, Kibana version 7.6.0 is tested and recommended. Other 7.x versions may work but could cause issues. Versions 8.x are not supported.
- The following Elasticsearch-specific field types are not supported:
  - %Spatial data types [https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#spatial_datatypes] %
  - %Structured data types [https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#structured-data-types] %
  - %Document ranking types [https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#document-ranking-types] %
  - %Text search types [https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#text-search-types] % (except for plain 'text')
  - %Relational data types [https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-types.html#object-types] %
- Metric aggregation functions are limited to %those supported by Manticore [../Searching/Grouping.md#Aggregation-functions] %.
- The following Kibana tools are not supported:
  - %Canvas [https://www.elastic.co/guide/en/kibana/7.6/canvas.html] % – A visualization and presentation tool for combining data with colors and images.
  - %Elastic Maps [https://www.elastic.co/guide/en/kibana/7.6/maps.html] % – A tool for analyzing geographical data.
  - %Metrics [https://www.elastic.co/guide/en/kibana/7.6/xpack-infra.html] % – An app for monitoring infrastructure metrics.
  - %Logs [https://www.elastic.co/guide/en/kibana/7.6/xpack-logs.html] % – A console-like display for exploring logs from common services.
  - Monitoring:
    - %Uptime [https://www.elastic.co/guide/en/kibana/7.6/xpack-uptime.html] % – Monitors the status of network endpoints via HTTP/S, TCP, and ICMP.
    - %APM (Application Performance Monitoring) [https://www.elastic.co/guide/en/kibana/7.6/xpack-apm.html] % – Collects in-depth performance metrics from applications.
    - %SIEM (Security Information and Event Management) [https://www.elastic.co/guide/en/kibana/7.6/xpack-siem.html] % – An interactive workspace for security teams to triage events and conduct initial investigations.
    - %ILM (Index lifecycle management) [https://www.elastic.co/guide/en/elasticsearch/reference/7.6/index-lifecycle-management.html] % - Automatically manage indices according to performance, resiliency, and retention requirements.
    - %Stack Monitoring [https://www.elastic.co/guide/en/kibana/7.6/xpack-monitoring.html] % – Provides visualizations of monitoring data across the Elastic Stack.
  - %Elasticsearch Management [https://www.elastic.co/guide/en/kibana/7.6/management.html] % – A UI for managing Elastic Stack objects, including ILM (Index Lifecycle Management), etc.

## Data Ingestion and Exploration
Integrate Manticore with tools like %Logstash [../Integration/Logstash.md] %, %Filebeat [../Integration/Filebeat.md] %, %Fluentbit [https://manticoresearch.com/blog/integration-of-manticore-with-fluentbit/] %, or %Vector.dev [https://manticoresearch.com/blog/integration-of-manticore-with-vectordev/] % to ingest data from sources like web logs. Once the data is loaded into Manticore, you can explore and visualize it in Kibana.

注意：此功能需要 Manticore Buddy。如果无法使用，请确保已安装 Buddy。

Manticore 支持通过 Kafka 源和物化视图与 Apache Kafka 实时数据摄取集成，实现实时数据索引和搜索。目前，apache/kafka 版本 3.7.0-4.1.0 已测试并支持。

开始之前，您需要：

1. 定义源： 指定 Manticore Search 将读取消息的 Kafka 主题。此设置包括代理的主机、端口和主题名称等详细信息。
2. 设置目标表： 选择一个 Manticore 实时表来存储传入的 Kafka 数据。
3. 创建物化视图： 设置物化视图（mv）以处理从 Kafka 到 Manticore Search 目标表的数据转换和映射。在这里，您将定义字段映射、数据转换以及传入数据流的任何过滤器或条件。

CREATE SOURCE <source name> [(column type, ...)] [source_options]

allowed_field_name 'original JSON key name with special symbols' type

price_field '$price' float    -- maps JSON key '$price' to field 'price_field'
field_123 '123field' text     -- maps JSON key '123field' to field 'field_123'

CREATE SOURCE kafka
(id bigint, term text, abbrev '$abbrev' text, GlossDef json)
type='kafka'
broker_list='kafka:9092'
topic_list='my-data'
consumer_group='manticore'
num_consumers='2'
batch=50

Query OK, 2 rows affected (0.02 sec)

CREATE TABLE destination_kafka
(id bigint, name text, short_name text, received_at text, size multi);

Query OK, 0 rows affected (0.02 sec)

CREATE MATERIALIZED VIEW <materialized view name>
TO <destination table name> AS
SELECT [column|function [as <new name>], ...] FROM <source name>

CREATE MATERIALIZED VIEW view_table
TO destination_kafka AS
SELECT id, term as name, abbrev as short_name,
       UTC_TIMESTAMP() as received_at, GlossDef.size as size FROM kafka

Query OK, 2 rows affected (0.02 sec)

数据以批次方式从 Kafka 传输到 Manticore Search，每次运行后批次被清空。对于跨批次的计算（如 AVG），请谨慎使用，因为批次处理可能导致结果不符合预期。

以下是基于上述示例的映射表：

SHOW SOURCES

+-------+
| name  |
+-------+
| kafka |
+-------+

SHOW SOURCE kafka;

+--------+-------------------------------------------------------------------+
| Source | Create Table                                                      |
+--------+-------------------------------------------------------------------+
| kafka  | CREATE SOURCE kafka                                               |
|        | (id bigint, term text, abbrev '$abbrev' text, GlossDef json)      |
|        | type='kafka'                                                      |
|        | broker_list='kafka:9092'                                          |
|        | topic_list='my-data'                                              |
|        | consumer_group='manticore'                                        |
|        | num_consumers='2'                                                 |
|        | batch=50                                                          |
+--------+-------------------------------------------------------------------+

SHOW MVS

+------------+
| name       |
+------------+
| view_table |
+------------+

SHOW MV view_table

+------------+--------------------------------------------------------------------------------------------------------+-----------+
| View       | Create Table                                                                                           | suspended |
+------------+--------------------------------------------------------------------------------------------------------+-----------+
| view_table | CREATE MATERIALIZED VIEW view_table TO destination_kafka AS                                            | 0         |
|            | SELECT id, term as name, abbrev as short_name, UTC_TIMESTAMP() as received_at, GlossDef.size as size   |           |
|            | FROM kafka                                                                                             |           |
+------------+--------------------------------------------------------------------------------------------------------+-----------+

ALTER MATERIALIZED VIEW view_table suspended=1

Query OK (0.02 sec)

您还可以为每个 Kafka 主题指定 partition_list。 这种方法的主要优点之一是能够通过 Kafka 实现表的 sharding（分片）。 为此，您应为每个分片创建一条独立的链：source → materialized view → destination table：

源：

CREATE SOURCE kafka_p1 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='0' batch=50;
CREATE SOURCE kafka_p2 (id bigint, term text)
  type='kafka' broker_list='kafka:9092' topic_list='my-data'
  consumer_group='manticore' num_consumers='1' partition_list='1' batch=50;

目标表：

CREATE TABLE destination_shard_1 (id bigint, name text);
CREATE TABLE destination_shard_2 (id bigint, name text);

物化视图：

CREATE MATERIALIZED VIEW mv_1 TO destination_shard_1 AS SELECT id, term AS name FROM kafka_p1;
CREATE MATERIALIZED VIEW mv_2 TO destination_shard_2 AS SELECT id, term AS name FROM kafka_p2;

• 在此设置中，重新平衡必须手动管理。
• Kafka 默认不使用轮询（round-robin）策略分发消息。
• 若要在发送数据时实现类似轮询的分发，请确保您的 Kafka 生产者配置了：
  • parse.key=true
  • key.separator={your_delimiter}

否则，Kafka 会根据其内部规则分发消息，可能导致分区不均。

Kafka 在每个批次后或处理超时后提交偏移量。如果物化视图查询过程中进程意外停止，可能会出现重复条目。为避免此情况，请在您的模式中包含 id 字段，使 Manticore Search 能防止表中出现重复。

• 工作线程初始化： 配置源和物化视图后，Manticore Search 会设置专用工作线程处理来自 Kafka 的数据摄取。
• 消息映射： 根据源配置的模式映射消息，将其转换为结构化格式。
• 批处理： 将消息分组为批次以提高处理效率。批次大小可根据性能和延迟需求调整。
• 缓冲： 映射后的数据批次存储在缓冲表中，以便高效批量操作。
• 物化视图处理： 对缓冲表中的数据应用视图逻辑，执行任何转换或过滤。
• 数据传输： 处理后的数据随后传输到目标实时表。
• 清理： 每个批次处理后清空缓冲表，确保为下一批数据做好准备。