1. 程式人生 > 其它 >logback-spring 整合 ELK、kafka的配置

logback-spring 整合 ELK、kafka的配置

pom.xml

  <dependency>
            <groupId>com.github.danielwegener</groupId>
            <artifactId>logback-kafka-appender</artifactId>
            <version>0.2.0-RC2</version>
        </dependency>

        <dependency>
            <groupId>net.logstash.logback</groupId>
            <artifactId>logstash-logback-encoder</artifactId>
            <version>6.4</version>
        </dependency>

        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.3</version>
        </dependency>

logback-spring.xml的文俊

<?xml version="1.0" encoding="UTF-8"?>
<configuration  scan="true" scanPeriod="10 seconds" debug="false">


    <!--上下文名稱-->
    <contextName>logback</contextName>

    <!--日誌根目錄 -->
    <property name="log.path" value="C:/logs" />
    <springProperty scope="
context" name="servicename" source="spring.application.name" defaultValue="UnknownService"/> <springProperty scope="context" name="env" source="spring.profiles.active" defaultValue="dev"/> <springProperty scope="context" name="bootstrapServers" source="spring.kafka.bootstrap-servers
" defaultValue="localhost:9092"/> <springProperty scope="context" name="serviceport" source="server.port" defaultValue="80"/> <!--獲取伺服器的IP和名稱--> <conversionRule conversionWord="serviceip" converterClass="com.icar.web.makedata.utils.LogIpConfigUtil" /> <!--以上三行需要和yml對應--> <!--輸出日誌到控制檯 --> <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender"> <filter class="ch.qos.logback.classic.filter.ThresholdFilter"> <level>INFO</level> </filter> <encoder> <pattern>%yellow(%date{yyyy-MM-dd HH:mm:ss}) |%highlight(%-5level) |%blue(%thread) |%green(%file:%line) |%magenta(%logger) |%cyan(%msg%n)</pattern> <charset>UTF-8</charset> </encoder> </appender> <!--①.level=INFO的日誌檔案 --> <appender name="INFO_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> <fileNamePattern>${log.path}/info/infolevel_makedata.%d{yyyy-MM-dd}.%i.txt</fileNamePattern> <maxFileSize>100MB</maxFileSize> <maxHistory>15</maxHistory> <totalSizeCap>2GB</totalSizeCap> </rollingPolicy> <!--日誌輸出級別--> <filter class="ch.qos.logback.classic.filter.LevelFilter"> <level>INFO</level> <onMatch>ACCEPT</onMatch> <onMismatch>DENY</onMismatch> </filter> <!--日誌檔案輸出格式--> <encoder> <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %thread %-5level %logger{50} --- %msg%n</pattern> <charset>UTF-8</charset> </encoder> </appender> <!--②.level=WARN的日誌檔案 --> <appender name="WARN_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> <!--基本設定--> <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> <fileNamePattern>${log.path}/warn/warnlevel_makedata.%d{yyyy-MM-dd}.%i.txt</fileNamePattern> <maxFileSize>100MB</maxFileSize> <maxHistory>15</maxHistory> <totalSizeCap>2GB</totalSizeCap> </rollingPolicy> <!--日誌輸出級別--> <filter class="ch.qos.logback.classic.filter.LevelFilter"> <level>WARN</level> <onMatch>ACCEPT</onMatch> <onMismatch>DENY</onMismatch> </filter> <!--日誌檔案輸出格式--> <encoder> <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %thread %-5level %logger{50} - %msg%n</pattern> <charset>UTF-8</charset> </encoder> </appender> <!--③.level=ERROR的日誌檔案 --> <appender name="ERROR_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender"> <!--基本設定--> <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy"> <fileNamePattern>${log.path}/error/errorlever_makedata.%d{yyyy-MM-dd}.%i.txt</fileNamePattern> <maxFileSize>100MB</maxFileSize> <maxHistory>15</maxHistory> <totalSizeCap>2GB</totalSizeCap> </rollingPolicy> <!--日誌輸出級別--> <filter class="ch.qos.logback.classic.filter.LevelFilter"> <level>ERROR</level> <onMatch>ACCEPT</onMatch> <onMismatch>DENY</onMismatch> </filter> <!--日誌檔案輸出格式--> <encoder> <pattern>[%d{yyyy-MM-dd HH:mm:ss.SSS}] %thread %-5level %logger{50} - %msg%n</pattern> <charset>UTF-8</charset> </encoder> </appender> <!-- <appender name="KafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender"> <encoder class="com.github.danielwegener.logback.kafka.encoding.LayoutKafkaMessageEncoder"> <layout class="net.logstash.logback.layout.LogstashLayout" > &lt;!&ndash; 是否包含上下文 &ndash;&gt; <includeContext>true</includeContext> &lt;!&ndash; 是否包含日誌來源 &ndash;&gt; <includeCallerData>true</includeCallerData> &lt;!&ndash; 自定義附加欄位 &ndash;&gt; <customFields>{"system":"test"}</customFields> &lt;!&ndash; 自定義欄位的簡稱 &ndash;&gt; <fieldNames class="net.logstash.logback.fieldnames.ShortenedFieldNames"/> </layout> <charset>UTF-8</charset> </encoder> &lt;!&ndash;kafka topic 需要與配置檔案裡面的topic一致 否則kafka會沉默並鄙視你&ndash;&gt; <topic>applog_dev</topic> <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.HostNameKeyingStrategy" /> <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" /> <producerConfig>bootstrap.servers=124.71.59.186:9092,139.159.249.142:9092,124.71.85.73:9092</producerConfig> &lt;!&ndash; don't wait for a broker to ack the reception of a batch. &ndash;&gt; <producerConfig>acks=0</producerConfig> &lt;!&ndash; wait up to 1000ms and collect log messages before sending them as a batch &ndash;&gt; <producerConfig>linger.ms=1000</producerConfig> &lt;!&ndash; even if the producer buffer runs full, do not block the application but start to drop messages &ndash;&gt; <producerConfig>max.block.ms=0</producerConfig> &lt;!&ndash; define a client-id that you use to identify yourself against the kafka broker &ndash;&gt; <producerConfig>client.id=${HOSTNAME}-${CONTEXT_NAME}-logback-relaxed</producerConfig> this is the fallback appender if kafka is not available. <appender-ref ref="CONSOLE" /> </appender>--> <appender name="KafkaAppender" class="com.github.danielwegener.logback.kafka.KafkaAppender"> <encoder class="net.logstash.logback.encoder.LoggingEventCompositeJsonEncoder"> <providers class="net.logstash.logback.composite.loggingevent.LoggingEventJsonProviders"> <pattern> <!-- <pattern> { "env": "${env}", "servicename":"${servicename}", "type":"${servicename}", "serviceinfo":"%serviceip:${serviceport}", "date":"%d{yyyy-MM-dd HH:mm:ss.SSS}", "level":"%level", "thread": "%thread", "logger": "%logger{36}", "msg":"%msg", "exception":"%exception" } </pattern>--> <pattern> { "env": "${env}", "servicename":"${servicename}", "type":"${servicename}", "serviceinfo":"%serviceip:${serviceport}", "date":"%d{yyyy-MM-dd HH:mm:ss.SSS}", "level":"%level", "thread": "%thread", "msg":"%msg", "exception":"%exception" } </pattern> </pattern> </providers> </encoder> <topic>appdev</topic> <keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy"/> <deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy"/> <producerConfig>acks=0</producerConfig> <producerConfig>linger.ms=1000</producerConfig> <producerConfig>max.block.ms=0</producerConfig> <producerConfig>bootstrap.servers=${bootstrapServers}</producerConfig> </appender> <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender"> <appender-ref ref="KafkaAppender"/> </appender> <root level="INFO"> <appender-ref ref="ASYNC"/> </root> <!-- <root> <appender-ref ref="CONSOLE"/> <appender-ref ref="INFO_FILE"/> <appender-ref ref="WARN_FILE"/> <appender-ref ref="ERROR_FILE"/> </root>--> <!--springProfile表示在dev環境下使用 --> <!-- <springProfile name="dev"> <logger name="com.nick" level="INFO" additivity="true"> <appender-ref ref="KafkaAppender" /> </logger> </springProfile>--> </configuration>
import ch.qos.logback.classic.pattern.ClassicConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;
import lombok.extern.slf4j.Slf4j;

import java.io.Console;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.UnknownHostException;
import java.util.Enumeration;

/**
 * 獲取伺服器IP以及伺服器名稱
 */
//@Slf4j
public class LogIpConfigUtil extends ClassicConverter {

    public  static  String serviceIp;

    static {
        try {
           /* Enumeration<NetworkInterface> allNetInterfaces = NetworkInterface.getNetworkInterfaces();
            InetAddress ip = null;
            while (allNetInterfaces.hasMoreElements()) {
                NetworkInterface netInterface = (NetworkInterface) allNetInterfaces.nextElement();
                if (netInterface.isLoopback() || netInterface.isVirtual() || !netInterface.isUp()) {
                    continue;
                } else {
                    Enumeration<InetAddress> addresses = netInterface.getInetAddresses();
                    while (addresses.hasMoreElements()) {
                        ip = addresses.nextElement();
                        if (ip != null && ip instanceof Inet4Address) {
                            return ip.getHostAddress();
                        }
                    }
                }
            }*/

            InetAddress addr = InetAddress.getLocalHost();
            serviceIp =addr.getHostName()+"/" +addr.getHostAddress();

        } catch (Exception e) {
            //log.error("IP地址獲取失敗" + e.toString());
        }
    }

    @Override
    public String convert(ILoggingEvent iLoggingEvent) {
        return  serviceIp;
    }
}

logstash-es.conf

input {
  kafka {
    group_id => "test-consumer-group"
    topics => ["appdev"]   
    bootstrap_servers => "localhost:9092"
    codec => "json"
  }
}
filter {
}
output {
  stdout {  codec => rubydebug }
  if [type] == "xxxx" {
    elasticsearch {
      hosts => [ "localhost:9200" ]
      index => "xxx"
    }
  }
}