windows下部署安装 ELK，nginx，tomcat日志分析

1.安装包

如上就是elk- windows下部署需要用到的安装包（ps:注意版本需要对应，另外es7版本对应是 jdk8，若更高版本请自行查询版本对应号）。

下载地址：

Past Releases of Elastic Stack Software | Elastic

此地址可下载elk的全部安装包（不包含分词器），请按照对应系统选择对应版本。

2.安装es

ES 在windows下，压缩包直接解压，进入bin目录启动 elasticsearch.bat 即可。

关于分词器的添加，解压分词器，重命名为 ik，将这个ik文件夹放入 elasticsearch-7.9.3\plugins 目录下即可

如上，显示分词器添加成功

启动成功后，访问：

3.安装filebeat - 读取日志

压缩包解压即可，修改 filebeat.yml 文件，具体修改如下（如下只是作为测试，此配置为主要配置的参数，并非全部，请勿全部复制，具体根据自己的需求自行配置）

============================== Filebeat inputs ===============================
# 单个日志配置 采集nginx
# filebeat.inputs:
# - type: log
#   enabled: true
#   paths:
#     # - /var/log/*.log
#     # - c:\programdata\elasticsearch\logs\*
#     - D:\tools\jiankong_tools\jiankong-nginx-1.26.2\logs\access.log

# 示例
# - type: log
#   enabled: true
#   paths:
#     - C:\path\to\tomcat\logs\*.log  # 替换为实际的Tomcat日志路径
#   fields:
#     type: tomcat
#     service: tomcat

# 定义全局字段，所有输入都将包含这些字段
fields:
  environment: "production" # 如果适用，定义环境（如生产、测试）

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - D:/nginx-1.26.2/logs/access.log*
    - D:/nginx-1.26.2/logs/error.log*
  tags: ["nginx"] # 标签，用于Logstash中的条件判断
  fields:
    service: nginx # 指定服务名，用于后续过滤

# 如果你选择在Filebeat中解析，可以添加如下processors
  processors:
    - dissect:
        # tokenizer: '%{ip} - %{user} [%{timestamp}] "%{method} %{uri} HTTP/%{http_version}" %{status} %{size} "%{referrer}" "%{user_agent}" rt=%{response_time}'
        tokenizer: 'rt=%{response_time}'
        field: 'message'
        target_prefix: 'nginx' 
  # multiline.pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2}' # 匹配日期格式的行作为新消息的开始
  # multiline.negate: true # 反转模式匹配，即非日期开头的行被视为同一事件的一部分
  # multiline.match: after # 在找到匹配行后，将随后的行附加到该事件中

- type: log
  enabled: true
  paths:
    - D:/apache-tomcat-9.0.83/xoalogs/*.log # Tomcat自定义日志路径
    - D:/apache-tomcat-9.0.83/logs/*.log # Tomcat标准日志路径
  tags: ["tomcat"] # 标签，用于Logstash中的条件判断
  fields:
    service: tomcat # 指定服务名，用于后续过滤
  multiline.pattern: '^\s*(\d{4}|\d{2})-(\d{2}|[a-zA-Z]{3})-(\d{2}|\d{4})' # 匹配日期格式的行作为新消息的开始
  multiline.negate: true # 反转模式匹配，即非日期开头的行被视为同一事件的一部分
  multiline.match: after # 在找到匹配行后，将随后的行附加到该事件中
  
  
  # ---------------------------- Elasticsearch Output ----------------------------
# 此为默认配置，注释此配置 ，添加下面的 写入logstash 端口为5044
# output.elasticsearch:
#   # Array of hosts to connect to.
#   hosts: ["localhost:9200"]

  # Protocol - either `http` (default) or `https`.
  #protocol: "https"

  # Authentication credentials - either API key or username/password.
  #api_key: "id:api_key"
  #username: "elastic"
  #password: "changeme"

# ------------------------------ Logstash Output -------------------------------
output.logstash:
  # The Logstash hosts
  hosts: ["localhost:5044"]

启动filebeat：

进入解压的目录执行注意配置指向路径改为自己的：

.\filebeat.exe -e -c D:\xxx\filebeat-7.9.3-windows-x86_64\filebeat.yml

4.安装logstash - 接收来自filebeat的日志

logstash-7.9.3\config 目录下已经有一个 logstash-sample.conf 示例了，也可自行添加一个 conf文件，我添加了一个 log.conf文件配置如下：

input {
  beats {
    port => 5044
  }
}

# 单独写入一个 es的索引
# output {
#   elasticsearch {
#     hosts => ["http://localhost:9200"]
#     index => "test123"
#     #user => "elastic"
#     #password => "changeme"
#   }
# }

filter {
  if "nginx" in [tags] {
      grok {
        match => { "message" => '%{IPORHOST:clientip} - %{DATA:user} $%{HTTPDATE:timestamp}$ "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response:int} %{NUMBER:bytes:int} "%{DATA:referrer}" "%{DATA:agent}" rt=%{NUMBER:response_time:float}' }
      }
      date {
        match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
        target => "@timestamp"
      }
    }

  if "tomcat" in [tags] {
    grok {
      match => { "message" => "%{TIMESTAMP_ISO8601:log_timestamp} $%{DATA:thread}$ %{LOGLEVEL:level} %{JAVACLASS:class} - %{GREEDYDATA:message}" } # 解析Tomcat日志，包括线程、日志级别、类名和消息
    }
    date {
      match => [ "log_timestamp", "yyyy-MM-dd HH:mm:ss,SSS", "yyyy-MM-dd HH:mm:ss" ] # 支持两种时间格式：带或不带毫秒
      target => "@timestamp"
    }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"] # Elasticsearch主机地址
    index => "%{[fields][service]}" # 索引命名模式
  }
  stdout { codec => rubydebug } # 输出到控制台，仅用于调试目的，上线前应移除
}