导读：当采用 Flink on K8S Application Mode 方案部署作业时 JobMananger、TaskMananger 等都是以容器形式运行。查看容器日志比较麻烦，而且当删除容器时容器内的日志就会消失，因此需要一个日志采集方案将日志采集起来便于排查问题。下文将主要分享笔者的实现思路（组件的安装及部署不在本文讨论之内）。

日志采集方案

在技术方案上笔者选择比较常见的 ELK ，由于日志在容器内，因此需要先将日志挂载到服务器本地，之后通过 Flebeat 采集丢到 Logstash 对日志进行处理并存到 ES 中，最后由 Kibana 进行展示。

容器日志挂载

挂载方式比较简单，通过修改官方 Application cluster resource definitions 提供的 jobmanager-application.yaml 和 taskmanager-job-deployment.yaml 指定日志挂载路径。

volumeMounts:
            - name: volume-log-path
              mountPath: /opt/flink/log

volumes:
        - name: volume-log-path
          hostPath:
            path: /dockerdata/volume/flink/log  # 日志挂载

日志采集及处理要点

笔者在测试时将 JM 和 TM 日志都输出到 /dockerdata/volume/flink/log 目录下，因此该目录下有两种命名类型的文件。

• flink--standalonejob-*-flink-jobmanager-<jobid>-*.log
• flink--taskexecutor-*-flink-taskmanager-<jobid>-*.log

在进行日志采集时有两个需要处理的点：

• 将以日期开头的一段作为一条日志（也就是将非日期开头的日志向前合并为一行）
• 采集日志存入 es 后需支持根据日志的类型 Type = jobmananger/taskmananger 搭配 jobid 对日志进行筛选

为了方便展示，下面具体处理将省去 Fliebeat 环节 ，直接由 Logstash 采集源日志文件。

日志处理

关于日志合并行可通过 Logstash 的 logstash-filter-multiline 插件实现：

• pattern：为正则匹配
• negate： true:表示不匹配正则表达式时，false:匹配正则表达式时
• what：设置未匹配的内容是向前合并还是先后合并,previous向前合并，next向后合并

// 插件安装
./bin/logstash-plugin install logstash-filter-multiline

关于实现 es 筛选条件可在 Logstash 采集日志时根据文件名格式设置不同的 type，对于文件名中的 jobid 字段提取笔者则采用了 ruby 处理，之后在 ES 中就可以对 type 和 jobid 这两列进行条件查询。

logstash flink-log.conf 编写如下：

input {
  file {
    type => "jobmanager"
    path => "/dockerdata/volume/flink/log/flink--standalonejob-*.log" # 匹配 jobMananger 日志
  }

  file {
    type => "taskmanager"
    path => "/dockerdata/volume/flink/log/flink--taskexecutor-*.log" # 匹配 taskMananger 日志
  }
}

filter {
    ruby {
        code => "
            path = event.get('log')['file']['path']
            # 提取文件名中的 jobid
            if (!path.nil?) && (!path.empty?)
                event.set('jobid', path.split('/')[-1].split('-')[-2])
            end
        "
    }

    # 合并日志
    multiline {
        pattern => "^\d{4}-\d{1,2}-\d{1,2}\s\d{1,2}:\d{1,2}:\d{1,2}" # 正则匹配日志
        negate => true
        what => "previous"   # 向前合并
    }
}

output {
    elasticsearch { 
        hosts => ["xxxxxxxxxx:9200"]
        index => "flink-log-test-%{+YYYY.MM.dd}"
    }    
    #stdout { codec => rubydebug }
} 

最后

感谢您的阅读，如果喜欢本文欢迎关注和转发，转载需注明出处，本头条号将持续分享IT技术知识。对于文章内容有其他想法或意见建议等，欢迎提出共同讨论共同进步。

https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
https://www.w3cschool.cn/ruby/
https://blog.csdn.net/dwyane__wade/article/details/80191030

Tags：查看容器日志