filebeat采集日志大到ES

filebeat.yml配置

###################### Filebeat Configuration Example #########################
# ============================== Filebeat inputs ===============================

filebeat.inputs:

- type: log
  enabled: true
  paths:
    - /opt/apps/test-*.log
  fields:
     source: test



# ============================== Filebeat modules ==============================

filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: false

  # Period on which files under path should be checked for changes
  #reload.period: 10s

# ======================= Elasticsearch template setting =======================

setup.template.settings:
  index.number_of_shards: 1
  #index.codec: best_compression
  #_source.enabled: false

# ---------------------------- Elasticsearch Output ----------------------------
output.elasticsearch:
  # Array of hosts to connect to.
  hosts: ["127.0.0.1:9200"]
  pipelines: 
# 提取message中的字段的时候用
    - pipeline: "test-pipeline"
  codec.format:
    string: %{[message]}
# 创建到es的索引的名字
  indices:
    - index: "test"
      when.equals:
        fields.source: "test"

# ================================= Processors =================================
processors:
#  - add_host_metadata:
#      when.not.contains.tags: forwarded
#  - add_cloud_metadata: ~
#  - add_docker_metadata: ~
#  - add_kubernetes_metadata: ~
  - drop_fields:
      fields: ["input_type", "log.offset", "host.name", "input.type", "agent.hostname", "agent.type", "ecs.version", "agent.ephemeral_id", "agent.id", "agent.version", "fields.ics", "log.file.path", "log.flags","agent.ephemeral_id","agent.hostname","agent.id","agent.name","agent.type"]
      ignore_missing: false

提取message的字段

新建pipeline.json 文件

语法可参照

{
  "description" : "test-pipeline",
  "processors" : [
    {
      "grok" :{
        "field" : "message",
        "patterns" : ["\|ip=%{DATA:ip}\|cip=%{DATA:cip}\|projectName=%{DATA:projectName}\|traceId=%{DATA:traceId}\|tenantId=%{DATA:tenantId}\|orgId=%{DATA:orgId}\|currentUserId=%{DATA:currentUserId}\|level=%{DATA:level}\|logType=%{DATA:logType}\|bizType=%{DATA:bizType}\|bizTypeDesc=%{DATA:bizTypeDesc}\|retCode=%{DATA:retCode}\|elapsed=%{DATA:elapsed}\|gmtCreate=%{DATA:gmtCreate}\|"]
      }
    }
  ]
}

执行命令

curl -H "Content-Type: application/json" -XPUT http://localhost:9111/_ingest/pipeline/test-pipeline -d@/opt/apps/pipeline.json

test-pipeline与filebeat中的pipeline 要对应