Springboot 整合ELK+kafka采集日志

package com.elk.log;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.Charset;

public class KafkaOutputStream extends OutputStream {


    Producer logProducer;
    String topic;

    public KafkaOutputStream(Producer producer, String topic) {
        this.logProducer = producer;
        this.topic = topic;
    }


    @Override
    public void write(int b) throws IOException {
        this.logProducer.send(new ProducerRecord<>(this.topic, b));
    }

    @Override
    public void write(byte[] b) throws IOException {
        this.logProducer.send(new ProducerRecord<String, String>(this.topic, new String(b, Charset.defaultCharset())));
    }

    @Override
    public void flush() throws IOException {
        this.logProducer.flush();
    }
}

package com.elk.log;

import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.Layout;
import ch.qos.logback.core.OutputStreamAppender;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.util.StringUtils;
import java.io.OutputStream;
import java.util.Properties;

public class KafkaAppender<E> extends OutputStreamAppender<E> {
   private Producer logProducer;
   private  String bootstrapServers;
   private 	Layout<E> layout;
   private  String topic;

    public void setLayout(Layout<E> layout) {
        this.layout = layout;
    }

    public void setBootstrapServers(String bootstrapServers) {
        this.bootstrapServers = bootstrapServers;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    @Override
    protected void append(E event) {
        if (event instanceof ILoggingEvent) {
            String msg = layout.doLayout(event);
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, 0,((ILoggingEvent) event).getLevel().toString(), msg);
            logProducer.send(producerRecord);
        }
    }

    @Override
    public void start() {

        if (StringUtils.isEmpty(topic)) {
            topic = "Kafka-app-log";
        }
        if (StringUtils.isEmpty(bootstrapServers)) {
            bootstrapServers = "localhost:9092";
        }
        logProducer = createProducer();

        OutputStream targetStream = new KafkaOutputStream(logProducer, topic);
        super.setOutputStream(targetStream);
        super.start();
    }

    @Override
    public void stop() {
        super.stop();
        if (logProducer != null) {
            logProducer.close();
        }
    }

    //创建生产者
    private Producer createProducer() {
        synchronized (this) {
            if (logProducer != null) {
                return logProducer;
            }

            Properties props = new Properties();
            props.put("bootstrap.servers", bootstrapServers);
            //判断是否成功，我们指定了“all”将会阻塞消息 0.关闭 1.主broker确认 -1（all）.所在节点都确认
            props.put("acks", "0");
            //失败重试次数
            props.put("retries", 0);
            //延迟100ms，100ms内数据会缓存进行发送
            props.put("linger.ms", 100);
            //超时关闭连接
			//props.put("connections.max.idle.ms", 10000);
            props.put("batch.size", 16384);
            props.put("buffer.memory", 33554432);
            //该属性对性能影响非常大，如果吞吐量不够，消息生产过快，超过本地buffer.memory时，将阻塞1000毫秒，等待有空闲容量再继续
            props.put("max.block.ms",1000);
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            return new KafkaProducer<String, String>(props);
        }
    }
}

package com.elk.log;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/test")
public class TestController {
    @GetMapping("/testLog")
    public String testLog() {
        log.warn("gotest");
        return "ok";
    }

    @GetMapping("/testLog1")
    public Integer testLog1() {
        int i = 1/0;
        return i;
    }
}

public class LogMDCInterceptor implements HandlerInterceptor {
    private static final long serialVersionUID = 1L;
  	public static final String TRACE_ID = "TRACE_ID";

    @Override
    public boolean preHandle(HttpServletRequest request,HttpServletResponse response,
                             Object handler) throws Exception {
      // 如果有上层调用就用上层的ID，微服务feign调用场景，同一个请求链路用同一个traceId
      
      String traceId = request.getHeader(TRACE_ID);
      if(traceId == null){
        traceId = UUID.randomUUID().toString().replace("-","");
      }
      // MDC本质是一个threadLocal线程变量
      MDC.put(TRACE_ID, traceId);
      // 把traceId放入response中，前端也可查看
      response.setHeader(TRACE_ID,traceId);
     	return true;
    }
    
    @Override
    public void afterCompletion(HttpServletRequest request,HttpServletResponse response,
                                Object handler, Exception ex) throws Exception {
      	 // MDC本质是一个threadLocal线程变量,使用后需要清理，避免线程复用出现脏数据（request请求数据是放在tomcat 线程池中 ）
        MDC.clear();
    }
}

@Configuration
public class AutoConfiguration implements WebMvcConfigurer{
    @Override
    public void addInterceptors(InterceptorRegistry registry) {
        registry.addInterceptor(new LogMDCInterceptor()).addPathPatterns("/**");
    }
}

@Component
public class FeignRequestExtend implements RequestInterceptor {
  @Override
  public void apply(RequestTemplate requestTemplate){
    String traceId = MDC.get(LogMDCInterceptor.TRACE_ID);
    requestTemplate.header(LogMDCInterceptor.TRACE_ID,traceId);
  }
}

input {
    kafka {
        bootstrap_servers => "10.10.47.6:9092,10.10.47.9:9092"
        topics => ["cloud_logs"]   # appender中配置的topic
        consumer_threads => 5
        decorate_events => true
        codec => json {charset => "UTF-8"}
        auto_offset_reset => "latest"
        client_id => "kafka_log_collect"
    }
}

output {
    elasticsearch {
        hosts => "10.10.47.10:9200"
        #旧版本的logstash和新版的格式不一样，新版要增加[kafka]
        index => "%{[@metadata][topic]}-%{+YYYY-MM-dd}"
        index => "%{[@metadata][kafka][topic]}-%{+YYYY-MM-dd}"
    }
    #stdout {codec=>rubydebug}
}

{
  "appName": "demo",
  "method": "restartedMain",
  "level": "INFO",
  "className": "org.springframework.cloud.context.scope.GenericScope",
  "dateTime": "2020-06-09 17:36:07",
  "line": "295",
  "message": "BeanFactory id\u003d507e95a4-a9a1-3a1c-80dd-fcb5d8753cf2",
  "traceId": "5413d7d744ea497e84e2bca954587cd5"
}

curl -XPUT "http://192.168.0.111:9200/_template/cloud_logs" -H 'Content-Type: application/json' -d'{  "index_patterns": [    "cloud_logs*"  ],  "mappings": {    "properties": {        "line": {          "type": "long"        },        "@timestamp": {          "type": "date"        },        "dateTime": {          "type": "text"        },        "appName": {          "type": "text"        },        "message": {          "type": "text",          "analyzer": "ik_max_word",          "search_analyzer": "ik_max_word"        },        "className": {          "type": "text"        },        "level": {          "type": "text"        },        "method": {          "type": "text"        },        "traceId": {          "type": "text"        },        "timestamp": {           "type": "long"        }      }  },  "aliases": {}}'