Fluentd日志处理-tail拉取

Posted 2021-01-19

利用tail内置插件来获取日志

tail插件相当于tail -f，它会不断的获取更新的日志，

<source>
    @type     tail
    path      /log-dir/*-app.log
    pos_file  /log-dir/app.log.pos
    tagidaas
    refresh_interval 10s
    read_from_head true
    path_key path
    <parse>
            @type json      #把日志格式直接解析为json格式
    </parse>
</source>
<source>
  @type     tail
  path      /log-dir/*-debug.log
  pos_file  /log-dir/debug.log.pos
  tagdebug
  multiline_flush_interval 2s
  read_from_head true
  path_key path
    <parse>
            @type   multiline               #multiline 相当于logstash的multiline 
            format_firstline /^(?<level>(INFO|WARN|ERROR)+)/
            format1 /(?<level>[a-zA-Z]+)s*[(?<date>[0-9/-: ,]+)] (?<logger>[a-zA-Z.]+):(?<message>[dDs]+)/
    </parse>
</source>
<source>
    @type     tail
    path      /log-dir/*-requests.log
    pos_file  /log-dir/request.log.pos
    tagrequest
    refresh_interval 10s
    read_from_head true
    path_key path
    <parse>
        @type regexp
        expression /(?<message>.*)/
    </parse>
</source>

第一个filter是为日志添加字段，tag和宿主机的名字，这个可能需要调docker，直接取只会取到docker的ID

<filter *>
    @type record_transformer             
    <record>
        tag ${tag}
        hostname "#{Socket.gethostname}"
    </record>
</filter>
    <filter request>
    @type    grep                         #排除掉一些不需要的日志
    <exclude>
        key message
        pattern /.*healthcheck.*|.*prometheusMetrics.*|.*(v1+/)+(configurations)+(/+versions).*/
    </exclude>
</filter>
<filter request>
    @type parser
    key_name message
    reserve_data yes
    <parse>   
        @type regexp
        expression  /(?<ip>[^|]+)|(?<date>[^|]+)|(?<statusCode>[^|]+)|(?<contentLength>[^|]+)|(?<reqURI>[^|]+)|(?<referer>[^|]+)|(?<userAgent>[^|]+)|(?<reqId>[^|]+)|(?<internalIp>[^|]+)|(?<reqHost>[^|]+)|(?<reqOrigin>[^|]+)|(?<reqTime>[^|]+) |.*|(?<requestMethod>[w]+)/
    </parse>
</filter>
<match idaas>
    @type rewrite_tag_filter        #重写tag，匹配的重写tag为app.token,不匹配的重写标app.idaas
    <rule>
        key     thread_name
        pattern /token/
        tag     app.token
    </rule>
    <rule>
         key     thread_name
         pattern /token/
         tag     app.idaas
         invert  true
    </rule>
</match>

上面已经把idaas进行分流处理，这里我们把app.token进行一次过滤，然后和app.idaas一起输入到ES中

<filter app.token>
    @type parser
    key_name thread_name
    reserve_data yes
    <parse>
        @type regexp
        expression /(?<thread_name>[A-Za-z0-9.-_=/? ]+.)/
    </parse>
</filter>
<match request>
    @type elasticsearch
    host elasticsearchlog-lb.elasticsearch-log
    index_name    s3-fluentd-request
    type_name     s3-fluentd-request
    flush_interval 2s
    include_timestamp true
    ssl_verify    false
</match>
<match debug>
    @type elasticsearch
    host elasticsearchlog-lb.elasticsearch-log
    index_name    s3-fluentd-debug
    type_name     s3-fluentd-debug
    flush_interval 2s
    include_timestamp true
    ssl_verify    false
</match>
<match app.*>
    @type elasticsearch
    host elasticsearchlog-lb.elasticsearch-log
    index_name    s3-fluentd-idaas
    type_name     s3-fluentd-idaas
    flush_interval 2s
    include_timestamp true
    ssl_verify    false
</match>