1
1)将处理结果写入 ES（先不搞模板）
2
root@Elk03 ~# vim /etc/logstash/conf.d/04-beats-grok-es.conf
3
input {
4
  beats {
5
    port => "6666"
6
  }
7
}
8

9
filter {
10
  # ... grok + date + useragent + geoip 全套 filter ...
11
}
12

13
output {
14
  stdout {
15
     codec => "rubydebug"
16
  }
17

18
  # 我们把数据既输出至屏幕也也输出至ES集群
19
  elasticsearch {
20
      hosts => ["10.0.0.6:9200","10.0.0.7:9200","10.0.0.8:9200"]
21
      index => "kpyun-end-elfk-nginx-%{+YYYY.MM.dd}"
22
  }
23
}
24

25
2)重新发送
26
root@Elk02 ~# rm -rf /var/lib/filebeat/
27
root@Elk02 ~# filebeat -e -c /etc/filebeat/config/07-nginx-to-logstash.yaml
28

29
3)WebUI查看
30
http://10.0.0.6:5601/
31
# 这些索引是按天分布的（脚本具有随机性，是近几天的访问日志）

1
# 删除旧索引 + 清除 Kibana 索引模式
2
`WebUI 操作即可`
3

4
4)删除旧索引 + 重新采集
5
重新采集即可看到地图正常渲染
6
root@Elk03 ~# pkill -f -9 logstash
7
root@Elk03 ~# logstash -rf /etc/logstash/conf.d/04-beats-grok-es.conf
8
root@Elk02 ~# rm -rf /var/lib/filebeat/
9
root@Elk02 ~# filebeat -e -c /etc/filebeat/config/07-nginx-to-logstash.yaml
10

11
5)再次检查 映射（mapping）✅ 已修正
12
root@Elk01 ~# curl -s '10.0.0.6:9200/kpyun-end-elfk-nginx-2026.07.30/_mapping?pretty' | grep -A1 'location\|bytes\|clientip'
13
        "bytes" : {
14
          "type" : "integer",
15
✅️ bytes → integer;
16
        "clientip" : {
17
          "type" : "ip"
18
✅️ clientip → ip;
19
            "location" : {
20
              "type" : "geo_point",
21
✅️ geoip.location → geo_point

1
root@Elk03 ~# cat /etc/logstash/conf.d/01-stdin-to-stdout.conf
2
input {
3
  stdin {}
4
}
5
output {
6
  stdout {
7
    # 指定输出的数据格式，如果不指定，则默认值为: rubydebug
8
  codec => "json"
9
  }
10
}
11
root@Elk03 ~# cp /etc/logstash/conf.d/01-stdin-to-stdout.conf /tmp/instance1.conf
12
root@Elk03 ~# cp /etc/logstash/conf.d/01-stdin-to-stdout.conf /tmp/instance2.conf
13

14
1)启动实例1 — 默认数据目录（/usr/share/logstash/data/）
15
root@Elk03 ~# logstash -f /tmp/instance1.conf
16
# 不指定 --path.data，默认就是 /usr/share/logstash/data/
17

18
2)启动实例2 — 指定独立数据目录
19
root@Elk03 ~# logstash -f /tmp/instance2.conf --path.data /tmp/logstash-02
20
# --path.data 指定独立目录，绕开 .lock 冲突
21

22
3)验证两个实例都在运行
23
root@Elk03 ~# ps -ef | grep [l]ogstash | wc -l
24
2
25
root@Elk03 ~# ps -ef | grep logstash | grep -v grep
26
root  2483  ... Logstash -f /tmp/instance1.conf
27
root  2531  ... Logstash -f /tmp/instance2.conf --path.data /tmp/logstash-02
28
# 两个 logstash 进程，各自独立的数据目录 ✅️

1
1)Logstash 配置 — 3 个 beats input + if/else 输出
2
root@Elk03 ~# vim /etc/logstash/conf.d/05-multiple_input-to-if_es.yaml
3
input {
4
  beats {
5
    port => "6666"
6
    type => "auth"
7
#   这个端口收 auth 日志
8
  }
9

10
  beats {
11
    port => "7777"
12
    type => "syslog"
13
#   这个端口收 syslog
14
  }
15

16
  beats {
17
    port => "8888"
18
    type => "default"
19
#   这个端口收 kern 日志，走 else 兜底
20
  }
21
}
22

23
filter {
24
  mutate {
25
      split => { "message" => " " }
26
      # 以空格为分隔符切割
27

28
      add_field => {
29
      # 把时间从message拆出来 --> 独立字段date
30
         "date" => "%{[message][0]}"
31
       }
32
  }
33

34
  date {  # 时间格式是 ISO 8601 格式（带时区）
35
    match => [ "date", "ISO8601" ]
36
    target => "@timestamp"
37
 }
38

39
  mutate {  # 删掉被拆过的 `message`
40
      remove_field => [ "message","@version","agent","log","ecs","tags","input" ]
41
  }
42
}
43

44
output {
45
  if [type] == "auth" {
46
      elasticsearch {
47
          hosts => ["10.0.0.6:9200","10.0.0.7:9200","10.0.0.8:9200"]
48
          index => "kpyun-elfk-if-auth-%{+YYYY.MM.dd}"
49
      }  #   条件表达式用 == 做判断
50
  } else if [type] == "syslog" {
51
      elasticsearch {
52
          hosts => ["10.0.0.6:9200","10.0.0.7:9200","10.0.0.8:9200"]
53
          index => "kpyun-elfk-if-syslog-%{+YYYY.MM.dd}"
54
      }
55
  } else {  # 其它都进入这个索引
56
      elasticsearch {
57
          hosts => ["10.0.0.6:9200","10.0.0.7:9200","10.0.0.8:9200"]
58
          index => "kpyun-elfk-if-default-%{+YYYY.MM.dd}"
59
      }
60
   }
61
}

1
2)启动 Logstash
2
root@Elk03 ~# logstash -rf /etc/logstash/conf.d/05-multiple_input-to-if_es.yaml --config.test_and_exit
3
root@Elk03 ~# logstash -rf /etc/logstash/conf.d/05-multiple_input-to-if_es.yaml
4
root@Elk03 ~# ss -lntup | grep -E '6666|7777|8888'
5
LISTEN   *:8888
6
LISTEN   *:6666
7
LISTEN   *:7777
8
# 三个端口全部就绪 ✅️
9

10
3)Filebeat 配置 — 采集系统日志
11
root@Elk01 ~# vim /tmp/66-auth-to-logstash.yaml
12
filebeat.inputs:
13
- type: filestream
14
  id: auth-collector
15
  paths:
16
    - /var/log/auth.log*
17
  prospector.scanner.exclude_files: ['\.gz$']
18
#   排除压缩文件(以gz结尾的文件)（正则）
19

20
output.logstash:
21
  hosts: ["10.0.0.8:6666"]
22
# auth 日志 → 6666（type=auth）
23

24
root@Elk01 ~# vim /tmp/77-syslog-to-logstash.yaml
25
filebeat.inputs:
26
- type: filestream
27
  id: syslog-collector
28
  paths:
29
    - /var/log/syslog*
30
    # 这里没有.log，syslog就是日志文件
31
  prospector.scanner.exclude_files: ['\.gz$']
32

33
output.logstash:
34
  hosts: ["10.0.0.8:7777"]
35
# syslog → 7777（type=syslog）
36

37
root@Elk01 ~# vim /tmp/88-kern-to-logstash.yaml
38
filebeat.inputs:
39
- type: filestream
40
  id: kern-collector
41
  paths:
42
    - /var/log/kern.log*
43
  prospector.scanner.exclude_files: ['\.gz$']
44

45
output.logstash:
46
  hosts: ["10.0.0.8:8888"]
47
# kern → 8888（type=default → 走 else 分支）
48

49
4)启动 Filebeat 多实例 — 每个实例独立 registry
50
# 实例1：auth 日志
51
root@Elk01 ~# rm -rf /var/lib/filebeat/
52
root@Elk01 ~# nohup filebeat -e -c /tmp/66-auth-to-logstash.yaml > /tmp/66.log 2>&1 &
53

54
# 实例2：syslog（独立 --path.data）
55
root@Elk01 ~# rm -rf /tmp/filebeat-syslog
56
root@Elk01 ~# nohup filebeat -e -c /tmp/77-syslog-to-logstash.yaml --path.data /tmp/filebeat-syslog > /tmp/77.log 2>&1 &
57

58
# 实例3：kern（独立 --path.data）
59
root@Elk01 ~# rm -rf /tmp/filebeat-kern
60
root@Elk01 ~# nohup filebeat -e -c /tmp/88-kern-to-logstash.yaml --path.data /tmp/filebeat-kern > /tmp/88.log 2>&1 &
61

62
5)验证三个索引分别收到数据
63
root@Elk01 ~# curl -s '10.0.0.6:9200/_cat/indices/kpyun-elfk-if*?v'
64
health status index                            docs.count
65
green  open   kpyun-elfk-if-auth-2026.07.30         1678
66
green  open   kpyun-elfk-if-syslog-2026.07.30      27800
67
green  open   kpyun-elfk-if-default-2026.07.30     14329
68
# auth→1678 条，syslog→27800 条，kern→14329 条（走 else）✅️
69
'三条数据流各自写入不同的索引，互不干扰'

1
1)生成证书文件
2
root@Elk01 ~# /usr/share/elasticsearch/bin/elasticsearch-certutil cert \
3
    -out /etc/elasticsearch/elastic-certificates.p12 \
4
    -pass "" --days 36500
5
# -pass "" : 空密码（生产环境建议设密码）
6
# --days 36500 : 100 年有效期
7
........
8
Certificates written to /etc/elasticsearch/elastic-certificates.p12
9
........
10
root@Elk01 ~# ls -l /etc/elasticsearch/elastic-certificates.p12
11
-rw------- 1 root elasticsearch 3596 Jul 30 10:05 elastic-certificates.p12
12
# 3600 字节的 PKCS#12 证书文件
13
'权限600'  # 属主root，属组elasticsearch
14
root@Elk01 ~# egrep -i 'group|user' /lib/systemd/system/elasticsearch.service
15
User=elasticsearch
16
Group=elasticsearch
17
# ES集群运行用户是elasticsearch
18
root@Elk01 ~# chmod 640 /etc/elasticsearch/elastic-certificates.p12
19
-rw-r----- 1 root elasticsearch 3596 Aug  1 09:31 elastic-certificates.p12
20

21
2)拷贝证书到所有节点
22
root@Elk01 ~# scp -p /etc/elasticsearch/elastic-certificates.p12 10.0.0.7:/etc/elasticsearch/
23
-p：    #（小写的p）保留源文件的权限
24
root@Elk01 ~# scp -p /etc/elasticsearch/elastic-certificates.p12 10.0.0.8:/etc/elasticsearch/
25
# 三台机器的证书文件必须一致
26
root@Elk02 ~# ls -l /etc/elasticsearch/elastic-certificates.p12
27
-rw-r----- 1 root elasticsearch 3596 Aug  1 09:31 elastic-certificates.p12
28
# 权限、属主、属组正确
29

30
3)修改 ES 配置文件 — 三台都要加
31
root@Elk01 ~# vim /etc/elasticsearch/elasticsearch.yml
32
# 在文件末尾追加：
33
xpack.security.enabled: true
34
xpack.security.transport.ssl.enabled: true
35
# 启用安全认证，认证的方式是证书认证
36
xpack.security.transport.ssl.verification_mode: certificate
37
xpack.security.transport.ssl.keystore.path: elastic-certificates.p12
38
xpack.security.transport.ssl.truststore.path: elastic-certificates.p12
39
# 同步到其他节点
40
root@Elk01 ~# scp /etc/elasticsearch/elasticsearch.yml 10.0.0.7:/etc/elasticsearch/
41
root@Elk01 ~# scp /etc/elasticsearch/elasticsearch.yml 10.0.0.8:/etc/elasticsearch/
42

43
4)三台全部重启 ES
44
root@Elk01 ~# systemctl restart elasticsearch.service
45
root@Elk02 ~# systemctl restart elasticsearch.service
46
root@Elk03 ~# systemctl restart elasticsearch.service
47

48
5)验证 — 没有认证直接被拒
49
root@Elk01 ~# curl 10.0.0.6:9200/_cat/nodes?v
50
{"error":{"type":"security_exception","reason":"missing authentication credentials ..."},"status":401}
51
# 401 Unauthorized ✅️ 认证已生效
52

53
6)生成随机密码
54
root@Elk01 ~# /usr/share/elasticsearch/bin/elasticsearch-setup-passwords auto
55
Please confirm that you would like to continue [y/N] y
56

57
Changed password for user apm_system
58
PASSWORD apm_system = lmpdGQWoBWR5B7PjGpVi
59

60
Changed password for user kibana_system
61
PASSWORD kibana_system = iYLp0QpWVR0IUnpidbi6
62

63
Changed password for user kibana
64
PASSWORD kibana = iYLp0QpWVR0IUnpidbi6
65

66
Changed password for user logstash_system
67
PASSWORD logstash_system = aotLNka89X00g2zdutTU
68

69
Changed password for user beats_system
70
PASSWORD beats_system = YErtC18PxGbBSb9QioRh
71

72
Changed password for user remote_monitoring_user
73
PASSWORD remote_monitoring_user = qI2N1Q77Y3dDtDfQtknC
74

75
Changed password for user elastic
76
PASSWORD elastic = XYVMlZSaxUnA17uYP3Ti
77
# 把这些密码记下来！后面 Kibana / Logstash / Filebeat 都要用