实时监控：基于流计算 Oceanus（Flink) 实现系统和应用级实时监控

# 监控日志文件配置 
- type: log
  enabled: true
  paths:
    - /tmp/test.log
    #- c:\programdata\elasticsearch\logs\*

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

{
	"@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)
    --'connector.connection-path-prefix' = '/v1'          -- 可选字段, 每次请求时附加的路径前缀                                                        
    '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
}'

pipeline.max-parallelism: 2048
metrics.reporters: promgateway
metrics.reporter.promgateway.host: xx.xx.xx.xx              # Prometheus实例地址 
metrics.reporter.promgateway.port: 9090                     # Prometheus实例端口
metrics.reporter.promgateway.needBasicAuth: true
metrics.reporter.promgateway.password: xxxxxxxxxxx          # Prometheus实例密码
metrics.reporter.promgateway.interval: 10 SECONDS