接上文https://www.jianshu.com/p/d6f2034ba62b，我们完成了ELK的基本搭建与测试，启动了一个tomcat并采集到了相关日志，本文我们展示其在实际项目中如何更好的使用

1.标准化服务日志

我们的日志是通过log-pilot进行采集，服务的yaml我们是通过模板自动生成的，模板化的好处一方面是统一部署，另一方面是降低了编写与调试yaml的工作量；针对ELK采集部分我们期望所有的服务都能够遵循同样的标准写入日志。
修改YAML模板

#{{name}}为我们的服务名
#deploy
apiVersion: apps/v1
kind: Deployment
 #省略一部分 
    spec:
      #日志部分
      volumes:
      - name: tomcat-log
        emptyDir: {}
      - name: netcore-log
        emptyDir: {}
      - name: java-log
        emptyDir: {}
      containers:
      - name: {{name}}
        image: {{image}}
        imagePullPolicy: Always
        env:
        - name: image
          value: "{{name}}>{{image}}"
        - name: app.id
          value: "{{appId}}"
        - name: TZ
          value: Asia/Shanghai
#日志部分标准配置
        - name: aliyun_logs_mysys-{{name}}-stdout
          value: "stdout"        
        - name: aliyun_logs_mysys-{{name}}-tomcat
          value: "/usr/local/tomcat/logs/*.log"
        - name: aliyun_logs_mysys-{{name}}-netcore
          value: "/app/logs/*.log"
        - name: aliyun_logs_mysys-{{name}}-java
          value: "/logs/*.log"
        - name: aliyun_logs_mysys-{{name}}-stdout_tags
          value: "project={{projName}},app={{name}},lang=all,sourceType=stdout"
        - name: aliyun_logs_mysys-{{name}}-tomcat_tags
          value: "app=project={{projName}},{{name}},lang=java,sourceType=log"
        - name: aliyun_logs_mysys-{{name}}-netcore_tags
          value: "app={{name}},projName={{projName}},lang=net,sourceType=log"
        - name: aliyun_logs_mysys-{{name}}-java_tags
          value: "projName={{projName}},app={{name}},lang=java,sourceType=log"
        volumeMounts:
        - mountPath: "/usr/local/tomcat/logs"
          name: tomcat-log
        - mountPath: "/app/logs"
          name: netcore-log
        - mountPath: "/logs"
          name: java-log

如上面的yaml文件描述：
1.我们采集了stdout标准控制台输出日志，输出索引名：mysys-{{name}}-stdout
2.我们采集了/usr/local/tomcat/logs/.log，输出索引名：mysys-{{name}}-tomcat
3.我们采集了/app/logs/.log，输出索引名为：mysys-{{name}}-netcore
4.我们采集了/logs/*.log，输出索引名为：mysys-{{name}}-java
5.我们对不同语言环境日志添加了标签，这些标签会自动生成ES字段，用于kibana筛选或logstash表达式处理
projName：代表一个项目或应用，下游可以从项目维度使用索引过滤
app:为服务名，下游可以从服务维度使用索引过滤
lang:代表开发语言，下游可以从语言维度过滤，通常是因为不同语言输出
sourceType: 来源控制台还是日志文件
日志的格式不同，需要特殊处理时使用这个判断
添加mysys前缀的目的是为了kibana显示时，我们可以匹配所有的日志。
模板为了兼容.net与java不同的输出路径，我们将不同语言的日志输出目录均进行采集

如果平台团队地位够高，更建议服务开发团队进行一些约定：
1.强制约定大家写日志的路径相关，比如/logs目录
2.强制统一日志输出格式
这对于未来日志采集与统一处理会有很大的意义

优化建议
如果我们能够约定开发人员不同日志类型文件名格式不同，我们可以更方便的将日志采集到不同的索引中，而不需要写logstash的判断。
比如文件名约定log-ERROR-YYYY-MM-dd.log log-INFO-YYYY-MM-dd.log，我们yaml就可以分类型采集到不同的索引，本文我们为了演练logstash，因而索引生成希望后移到logstash中处理，故不采用文件名区分模式。

  - name: aliyun_logs_mysys-ERROR-{{name}}-java
    value: "/logs/*ERROR-.log"
  - name: aliyun_logs_mysys-INFO-{{name}}-java
    value: "/logs/*INFO-.log"

2.日志输出格式统一

因为我们平台需要支持多种语言，但不同语言日志组件不同，且输出格式不统一，我们建议对不同环境日志输出格式统一，否则logstash处理起来会比较复杂：
.net我们使用了NLog组件，输出配置如下：

  <targets>
    <default-wrapper xsi:type="BufferingWrapper" bufferSize="100" FlushTimeout="5000"/>
    <!--write logs to file-->
    <target xsi:type="File" name="infoFile" fileName="logs/info-${shortdate}.log"
                 layout="[${longdate}]|${logger}|${uppercase:${level}}|${message} ${exception}" />
    <target xsi:type="File" name="errorFile" fileName="logs/error-${shortdate}.log"
                 layout="[${longdate}]|${logger}|${uppercase:${level}}|${message} ${exception}" />
    <target xsi:type="File" name="traceFile" fileName="logs/trace-${shortdate}.log"
                    layout="[${longdate}]|${logger}|${uppercase:${level}}|${message} ${exception}" />
    <target xsi:type="Null" name="blackhole" />
  </targets>

java我们使用了logback，摘了info段的配置

 <appender name="INFO_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 正在记录的日志文件的路径及文件名 -->
        <file>${LOG_PATH}/info.log</file>
        <!--日志文件输出格式-->
        <encoder>
            <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] |%thread|%level{5}| %logger{50} - %msg%n</pattern>
            <charset>UTF-8</charset> <!-- 此处设置字符集 -->
        </encoder>
        <!-- 日志记录器的滚动策略，按日期，按大小记录 -->
        <rollingPolicy
                class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy"> <!-- 归档的日志文件的路径，例如今天是2018-10-26日志，当前写的日志文件路径为file节点指定，可以将此文件与file指定文件路径设置为不同路径，从而将当前日志文件或归档日志文件置不同的目录。 而2018-10-26的日志文件在由fileNamePattern指定。%d{yyyy-MM-dd}指定日期格式，%i指定索引 -->
            <fileNamePattern>${LOG_PATH}/info-%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <!-- 除按日志记录之外，还配置了日志文件不能超过50M，若超过50M，日志文件会以索引0开始， 命名日志文件，例如log-error-2018-10-26.0.log -->
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>20MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy> 
            <!--日志文件保留天数-->
            <maxHistory>30</maxHistory>
        </rollingPolicy>
        <!-- 此日志文件只记录info级别的 -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>info</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>

然后我们看到输出日志格式实现了统一

***********.Net日志**************
[2020-04-05 12:59:04.7338]|netCore07.LogHelper|INFO|请求Healthy:04/05/2020 12:59:04 
[2020-04-05 12:59:16.3861]|netCore07.LogHelper|INFO|请求DefaultIndex:04/05/2020 12:59:16 
[2020-04-05 12:59:17.5992]|netCore07.LogHelper|INFO|请求DefaultIndex:04/05/2020 12:59:17 
[2020-04-05 12:59:18.1920]|netCore07.LogHelper|INFO|请求DefaultIndex:04/05/2020 12:59:18 
[2020-04-05 12:59:21.6663]|netCore07.LogHelper|INFO|请求Healthy:04/05/2020 12:59:21 
***********Java日志**************
[2020-04-05 13:16:58.351] |http-nio-8020-exec-1|INFO| o.a.c.core.ContainerBase.[Tomcat].[localhost].[/] - Initializing Spring DispatcherServlet 'dispatcherServlet'
[2020-04-05 13:16:58.351] |http-nio-8020-exec-1|INFO| org.springframework.web.servlet.DispatcherServlet - Initializing Servlet 'dispatcherServlet'
[2020-04-05 13:16:58.375] |http-nio-8020-exec-1|INFO| org.springframework.web.servlet.DispatcherServlet - Completed initialization in 22 ms
[2020-04-05 13:16:58.434] |http-nio-8020-exec-1|INFO| com.xxx.controller.StudentController - File-springboot-say:xx

3.Logstash表达式

logstash处理过程为：input（输入数据接收）--->filter（数据加工过滤）---->output（数据接收与存储）
logstash前期测试我们只使用了input和output，没有对数据加工，因而在kibana中看到的日志效果如下图，所有日志信息均显示在message字段中，比较杂乱：

没有使用filter过滤效果

由于我们采集时没有按不同的日志文件名分别采集到不同索引，即无法识别ERROR/INFO/DEBUG等日志类型，因此我们第一个目标是拆分出日志类型字段给ES。
我们需要将数据从message串中提取并抽取成结构化的字段，这个过程处理是在filter环节进行的，而filter最常用的过滤器为grok。
grok使用的是正则过滤语法，且其内置了一系列常用的匹配表达式，可以直接使用。

关于grok语法调试器：grok调试起来相当麻烦，官网调试器国内不好用（http://grokdebug.herokuapp.com/），另外kibana的开发者工具中也有grok调试器，测试时也有问题，估计连接的也是官方后台。
国内可以使用这个地址http://grok.51vagaa.com/或http://47.112.11.147:9999/
或者您也可以自行安装一个调试器（如果能有一个客户端软件就完美了）。

在线调试grok
如上图所示，我们通过如下表达式对日志进行匹配验证，并成功提取出了logLevel字段
.{0,3}%{TIMESTAMP_ISO8601:logTime}.{0,10}\|.*\|%{LOGLEVEL:logLevel}\|%{GREEDYDATA:message}
关于grok内置表达式可以参考https://www.cnblogs.com/dyh004/p/9700110.html详细了解；

我们编写logstash脚本如下：

***default.conf***
input {
   beats {
       port => 5044
        codec => plain{
           charset=>"UTF-8"
       } 
  }
}
filter {
  grok {   
     match => {"message" => ".{0,3}%{TIMESTAMP_ISO8601:logTime}.{0,10}\|.*\|%{LOGLEVEL:logLevel}\|%{GREEDYDATA:message}"}
     #用上面提取的message覆盖原message字段
     overwrite => ["message"]
 }
}
output {
   elasticsearch {
      action => "index"
      hosts  => ["192.168.0.230:9200"]
      index  => "%{index}-%{+YYYY.MM.dd}"   #使用log-pilot给的索引名+年月日
      codec => plain {
            format => "%{message}"
            charset => "UTF-8"
        }
      user => "elastic"
      password => "123456"
    } 
}

修改后重启logstash容器，再次请求服务URL产生一些日志，我们到kibana中查看，发现相关字段已经可以提取出来了

使用filter将字段抽取后效果
多项目日志管理
如前面我们提到，一个项目或应用会有多个服务构成，我们通过log-pilot的tag为服务加上了projName和appName，在logstash的output中我们就可以直接用这些属性构建索引名，例如：
output {
index => "%{projName}-%{appName}-%{logLevel}-%{+YYYY.MM.dd}" #使用log-pilot给的索引名+年月日
}
这个模式的优点在于查看维度的灵活性：
1.如果想看整体项目的日志，可以配置索引模式projName查看
2.如果想看某一个服务的日志，可以配置索引模式projName-appName查看
3.如果想查看某一类型的日志，如错误日志，可以配置索引模式projName--ERROR查看

4.摘录:logstash工作原理

以下内容摘自[https://blog.csdn.net/herojuice/article/details/85113376]
Inputs 数据输入端：
1、file：从文件中读取
2、syslog：监听在514端口的系统日志信息，并解析成RFC3164格式
3、redis：从redis-server list中获取
4、beat：接收来自Filebeat的事件
Filter 数据中转层，主要进行格式处理，数据类型转换、数据过滤、字段添加，修改等，常用的过滤器如下：
1、grok：通过正则解析和结构化任何文件。Grok目前是logstash最好的方式对非结构化日志数据解析成结构化和可查询化。logstash内置了120个匹配模式，满足大部分需求
2、mutate：在事件字段执行一般的转换。可以重命名、删除、替换和修改事件字段
3、drop：完全丢弃事件，如debug事件
4、clone：复制事件，可能添加或者删除字段
5、geoip：添加有关IP地址地理位置信息
Output 是Logstash工作的最后一个阶段，负责将数据输出到指定位置，兼容大多数应用，常用的有：
1、elasticsearch：发送事件数据到ElasticSearch，便于查询、分析、绘图
2、file：将事件数据写入到磁盘文件上
3、mongodb：将事件数据发送至高性能NoSQL mongodb，便于永久存储、查询、分析、大数据分片
4、redis：将数据发送至redis-server
5、statsd：发送事件数据到statsd
6、graphite：发送事件数据到graphite