基于流计算 Oceanus 和 Elasticsearch Service 实现实时监控系统

# 监控日志文件配置
- type: log
  enabled: true
  paths:
    - /tmp/test.log

# 监控数据输出项配置
output.kafka:
  version: 2.0.0                           # kafka版本号
  hosts: ["xx.xx.xx.xx:xxxx"]              # 请填写实际的IP地址+端口
  topic: 'topic-app-info'                  # 请填写实际的topic

# filebeat 启动
./filebeat -e -c filebeat.yml

# 监控系统信息写入 test.log 文件
top -d 10 >>/tmp/test.log

{
	"@timestamp": "2021-08-30T10:22:52.888Z",
	"@metadata": {
		"beat": "filebeat",
		"type": "_doc",
		"version": "7.14.0"
	},
	"input": {
		"type": "log"
	},
	"host": {
		"ip": ["xx.xx.xx.xx", "xx::xx:xx:xx:xx"],
		"mac": ["xx:xx:xx:xx:xx:xx"],
		"hostname": "xx.xx.xx.xx",
		"architecture": "x86_64",
		"os": {
			"type": "linux",
			"platform": "centos",
			"version": "7(Core)",
			"family": "redhat",
			"name": "CentOSLinux",
			"kernel": "3.10.0-1062.9.1.el7.x86_64",
			"codename": "Core"
		},
		"id": "0ea734564f9a4e2881b866b82d679dfc",
		"name": "xx.xx.xx.xx",
		"containerized": false
	},
	"agent": {
		"name": "xx.xx.xx.xx",
		"type": "filebeat",
		"version": "7.14.0",
		"hostname": "xx.xx.xx.xx",
		"ephemeral_id": "6c0922a6-17af-4474-9e88-1fc3b1c3b1a9",
		"id": "6b23463c-0654-4f8b-83a9-84ec75721311"
	},
	"ecs": {
		"version": "1.10.0"
	},
	"log": {
		"offset": 2449931,
		"file": {
			"path": "/tmp/test.log"
		}
	},
	"message": "(B16root0-20000S0.00.00:00.00kworker/1:0H(B"
}

 CREATE TABLE DataInput (
     `@timestamp` VARCHAR,
     `host`       ROW<id VARCHAR,ip ARRAY<VARCHAR>>,
     `log`        ROW<`offset` INTEGER,file ROW<path VARCHAR>>,
     `message`    VARCHAR
 ) WITH (
     'connector' = 'kafka',         -- 可选 'kafka','kafka-0.11'. 注意选择对应的内置  Connector
     'topic' = 'topic-app-info',    -- 替换为您要消费的 Topic
     'scan.startup.mode' = 'earliest-offset',            -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets 的任何一种
     'properties.bootstrap.servers' = '10.0.0.29:9092',  -- 替换为您的 Kafka 连接地址
     'properties.group.id' = 'oceanus_group2',           -- 必选参数, 一定要指定 Group ID
     -- 定义数据格式 (JSON 格式)
     'format' = 'json',
     'json.ignore-parse-errors' = 'true',     -- 忽略 JSON 结构解析异常
     'json.fail-on-missing-field' = 'false'   -- 如果设置为 true, 则遇到缺失字段会报错 设置为 false 则缺失字段设置为 null
 );

CREATE TABLE es_output (
    `id`         VARCHAR,
    `ip`         ARRAY<VARCHAR>,
    `path`       VARCHAR,
    `num`        INTEGER,
    `message`    VARCHAR,
    `createTime` VARCHAR
) WITH (
    'connector.type' = 'elasticsearch',             -- 输出到 Elasticsearch
    'connector.version' = '6',                      -- 指定 Elasticsearch 的版本, 例如 '6', '7'. 
    'connector.hosts' = 'http://10.0.0.175:9200',   -- Elasticsearch 的连接地址
    'connector.index' = 'oceanus_test2',            -- Elasticsearch 的 Index 名
    'connector.document-type' = '_doc',             -- Elasticsearch 的 Document 类型
    'connector.username' = 'elastic',  
    'connector.password' = 'yourpassword', 
    'update-mode' = 'upsert',                       -- 可选无主键的 'append' 模式，或有主键的 'upsert' 模式     
    'connector.key-delimiter' = '$',                -- 可选参数, 复合主键的连接字符 (默认是 _ 符号, 例如 key1_key2_key3)
    'connector.key-null-literal' = 'n/a',           -- 主键为 null 时的替代字符串，默认是 'null'
    'connector.failure-handler' = 'retry-rejected', -- 可选的错误处理。可选择 'fail' （抛出异常）、'ignore'（忽略任何错误）、'retry-rejected'（重试）

    'connector.flush-on-checkpoint' = 'true',       -- 可选参数, 快照时不允许批量写入（flush）, 默认为 true
    'connector.bulk-flush.max-actions' = '42',      -- 可选参数, 每批次最多的条数
    'connector.bulk-flush.max-size' = '42 mb',      -- 可选参数, 每批次的累计最大大小 (只支持 mb)
    'connector.bulk-flush.interval' = '60000',      -- 可选参数, 批量写入的间隔 (ms)
    'connector.connection-max-retry-timeout' = '1000',    -- 每次请求的最大超时时间 (ms)                                                  
    'format.type' = 'json'                          -- 输出数据格式, 目前只支持 'json'
);

INSERT INTO es_output
SELECT 
  host.id       AS `id`,
  host.ip       AS `ip`,
  log.file.path AS `path`,
  log.`offset`  AS `num`,
  message,
  `@timestamp`  AS `createTime`
from DataInput;

# 查询索引  username:password 请替换为实际账号密码
curl -XGET -u username:password http://xx.xx.xx.xx:xxxx/oceanus_test2/_search -H 'Content-Type: application/json' -d'
{
    "query": { "match_all": {}},
    "size":  10
}
'