WAF防火墙数据接入腾讯云ES最佳实践（上）

rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

[logstash-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install logstash

input {
  beats {
    port => 5044
  }
}

input {
  syslog {
    port => 5144
    syslog_field => "syslog"
  }
}

input {
  file {
    path => "/var/log/*/*.log"
  }
}

input {
  elasticsearch {
    hosts => "localhost"
    query => '{ "query": {"match": { "statuscode": 200 } }, "sort": ["_doc" ] }'
  }
}

input {
  kafka {
    bootstrap_servers =>"xxxxx:9992 "
    topics => ["test"]
    codec => "json"
    security_protocol =>"SASL_PLAINTEXT"
    sasl_mechanism => "PLAIN"
    jaas_path =>"/etc/logstash/kafka-client-jaas.conf"
    consumer_threads => 2
    group_id => "logstash-bigdata"
    auto_offset_reset => "latest"
    max_poll_records => "500"
    max_poll_interval_ms =>"30000"
    session_timeout_ms => "30000"
    request_timeout_ms => "60000"
    auto_commit_interval_ms =>"5000"
    check_crcs => "false"
    heartbeat_interval_ms =>"9000"
    partition_assignment_strategy =>"org.apache.kafka.clients.consumer.RoundRobinAssignor"
    }
}

input {
  jdbc {
    jdbc_driver_library =>"mysql-connector-java-5.1.36-bin.jar"
    jdbc_driver_class =>"com.mysql.jdbc.Driver"
    jdbc_connection_string =>"jdbc:mysql://localhost:3306/mydb"
    jdbc_user => "mysql"
    parameters => {"favorite_artist" => "Beethoven" }
    schedule => "* * * * *"
    statement => "SELECT * from songswhere artist = :favorite_artist"
  }
}

input {
  s3 {
    "access_key_id" => "xxx"
    "secret_access_key" => "xxxxx"
    "endpoint" => "https://cos.ap-guangzhou.myqcloud.com"
    "bucket" => "my-bucket"
    "region" => "ap-guangzhou"
  }
}

output {
if [log_type] == "tomcat" and([department] == "zxbi" or [department] == "jfbi") {
  elasticsearch {
    hosts =>["https://rjjd-node01:9200"]
    index =>"jfbi.prd-%{+YYYY.MM}"
    user => "beatuser"
    password =>"xxxxx"
    http_compression => "true"
    sniffing => "false"
    ssl => true
    ssl_certificate_verification =>"true"
    cacert => "/etc/ssl/xxx_ca.crt"
    id => "zxbi"
  }
 }
}

output {
file {
  path => “/tmp/stdout.log”
  codec => line { 
      format => "customformat: %{message}"
    }
  }
}

output {
  stdout { codec => json }
}

filter {
  date {
      match => ["[creatime]",                     # 时间字段
          "yyyy-MM-dd HH:mm:ss",                  #  
          "yyyy-MM-dd HH:mm:ss Z",                # 
          "MMM dd, yyyy HH:mm:ss aa",             # Oct 16, 2020 11:59:53 PM
          "yyyy-MM-dd HH:mm:ss.SSS",      
          "ISO8601",                              # 2018-06-17T03:44:01.103Z ( Z 后面可以有 "08:00" 也可以没 )
          "UNIX",                                 # UNIX时间戳格式记录的是从 1970 年起始至今的总秒数
          "UNIX_MS"                               # 从 1970 年起始至今的总毫秒数
      ]
      target => "@timestamp"                      # 目标字段
      timezone => "Asia/Shanghai"                 # 时区设置
  }
}

filter {
  if [loglevel] == "debug" {
    drop { }
  }
}

filter {
  geoip {
    database =>"/usr/local/GeoLite2-City/GeoLite2-City.mmdb"
    fields => ["ip","longitute","city_name"]
    source => "ip"
    target => "geoip"
  }
}

filter {
  json {
    source => "message"
  }
}

filter {
  kv {
    default_keys => [“host”,”ip”]
    exclude_keys => [“times”]
    field_split => “&?”
  }
}

filter {
  ruby {
    code => "event.set('eventHappenedAt',event.get('createdDateTime').to_i*1000)"
  }
  ruby {
    id => "135453"
    path => "/usr/local/rubyCode/rubyLogParse.rb"
  }
}

   def filter(event)
      #取出日志信息
      logdata=event.get('message')
      if (logdata.include?'reqMessage' )
        #解析日志级别和请求时间,并保存到event对象中
        logleve=logdata[(logdata.index('[')+1)...logdata.index(']')]
        event.set('logleve',logleve)
        event.set('requestTime',logdata[0,logdata.index(' - [')])
        #如果是rest接口json日志
        if(logdata.include?'restJson')
          parseJsonLog event,logdata
        else
          parseXmlLog event,logdata
        end
      end
    return [event]
   #解析json格式日志
 def parseJsonLog(event,logdata)
    #获取日志内容
    loginfo=logdata[(logdata.index('] ')+2)...(logdata.rindex('}')+1)]
    #转成json对象
    logInfoJson=JSON.parse loginfo
    #获取流水号
    event.set('transactionID',logInfoJson['transactionID'])
    event.set('channel',logInfoJson['reqMessage']['channel'])
    #获取返回报文
    event.set('respMessage',logInfoJson['respMessage'].to_json)
    #获取请求报文
    event.set('reqMessage',logInfoJson['reqMessage'].to_json)
​
    .....................省略部分代码
​
    #设置标示位
    event.set('tags','log_success')
  end

filter {
  mutate {
    split => ["hostname","."]
    add_field => {"shortHostname" => "%{hostname[0]}" }
  }
  mutate {
    rename =>["shortHostname", "hostname" ]
  }
}