filebeat采集udp/tcp数据_filebeat采集udp端口

背景：在 win10下docker操作filebeat做日志采集 基础上进行

一、配置文件

1. D:\usr\local\etc\filebeat目录下docker-compose-filebeat.yml文件，新增暴露端口

version: '3'
services:
  filebeat:
    image: elastic/filebeat:7.6.2
    container_name: filebeat
    volumes:
        - /d/usr/local/logs/filebeat:/usr/share/filebeat/logs
        - /d/usr/local/etc/filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml
        - /d/usr/local/var/filebeat/data:/usr/share/filebeat/data
    ports:
      - 8080:8080/udp
      - 9000:9000

2. D:\usr\local\etc\filebeat目录下filebeat.yml文件，新增UDP、TCP数据源

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/share/filebeat/logs/*.log
  # 自定义属性
  fields:
    type: log
    document_type: spring
- type: tcp
  enabled: true
  max_message_size: 10MiB
  host: "0.0.0.0:9000"
  fields:
    type: tcp
- type: udp
  enabled: true
  max_message_size: 10KiB
  host: "0.0.0.0:8080"
  fields:
    type: udp
output.logstash:
    hosts: ["192.168.1.110:5044"]

3. D:\usr\local\etc\logstash\pipeline目录下新增UDP、TCP输出源

input { 
    stdin { } 
    beats {
        port => 5044
        type => "filebeat"
    }
    # tcp {
        # mode => "server"
        # host => "0.0.0.0"
        # # 从5044端口取日志
        # port => 5044
        # # 需要安装logstash-codec-json_lines插件
        # codec => json_lines
        # type => "tcp"
    # }
    file {
        # 容器中日志所在目录的文件
        path => ["/usr/share/logstash/logs/*.log"]
        #codec => "json"
        sincedb_path => "NUL"
        type => "spring"
        start_position => "beginning"
    }
}
 
output {
    if [type] == "spring" {
        elasticsearch {
            hosts => ["es:9200"]
            index => "spring-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "tcp" {
        elasticsearch { 
            hosts => ["es:9200"] 
            index => "logstash-%{+YYYY.MM.dd}"  
        }
    }
    if [fields][type] == "log" {
        elasticsearch { 
            hosts => ["es:9200"] 
            manage_template => false
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-log-%{+YYYY.MM.dd}"
        }
    }
    if [fields][type] == "tcp" {
        elasticsearch { 
            hosts => ["es:9200"] 
            manage_template => false
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-tcp-%{+YYYY.MM.dd}"
        }
    }
    if [fields][type] == "udp" {
        elasticsearch { 
            hosts => ["es:9200"] 
            manage_template => false
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-udp-%{+YYYY.MM.dd}"
        }
    }
    if [fields][document_type] == "spring" {
        elasticsearch { 
            hosts => ["es:9200"] 
            manage_template => false
            index => "%{[@metadata][beat]}-%{[@metadata][version]}-spring-%{+YYYY.MM.dd}"
        }
    }
	stdout { codec => rubydebug }
}

二、重启容器

1. 重启logstash

docker restart logstash

 

2. 重启filebeat

docker restart filebeat

三、发送数据

1. UDP数据发送脚本send_udp.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import socket  # 网络通信 TCP，UDP
 
# SOCK_DGRAM表示UDP，SOCKET_STREAM表示TCP
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 
while True:
    msg = input('>>: ').strip()
    if not msg:
        continue
 
    client.sendto(msg.encode('utf-8'), ('192.168.1.110', 8080))
 
    if msg == 'close':
        break
client.close()

2. TCP数据发送脚本send_tcp.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-
import socket  # 网络通信 TCP，UDP
 
# SOCK_DGRAM表示UDP，SOCKET_STREAM表示TCP
host = ('127.0.0.1', 9000)
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(host)
 
while True:
    msg = input('>>: ').strip()
    if not msg:
        continue
 
    client.sendto(msg.encode('utf-8'), host)
 
    if msg == 'close':
        break
client.close()

3. 执行send_udp.py脚本

4. 执行send_tcp.py脚本

四、查看结果

1. 接收udp数据结果

2. 接收tcp数据结果

从以上结果可以看出配置成功