如何将tomcat访问日志发送到kafka

Question

我想将Tomcat访问日志发送到kafka主题。我已经阅读了tomcat日志记录文档，我发现tomcat使用了apache juli。

我想删除默认日志记录并将所有访问日志发送到kafka。

我在server.xml中找到了

 <Valve className="org.apache.catalina.valves.AccessLogValve" directory="logs"
               prefix="localhost_access_log" suffix=".txt"
               pattern="%h %l %u %t &quot;%r&quot; %s %b" />

现在我需要更改此设置但是如何？

Answer 1

您可以浏览tomcat的源代码，您会发现AccessLogValve.java中存在密钥：

对于Tomcat 8：

public void log(CharArrayWriter message) {
    this.rotate();
    if (this.checkExists) {
        synchronized(this) {
            if (this.currentLogFile != null && !this.currentLogFile.exists()) {
                try {
                    this.close(false);
                } catch (Throwable var8) {
                    ExceptionUtils.handleThrowable(var8);
                    log.info(sm.getString("accessLogValve.closeFail"), var8);
                }

                this.dateStamp = this.fileDateFormatter.format(new Date(System.currentTimeMillis()));
                this.open();
            }
        }
    }

    try {
        synchronized(this) {
            if (this.writer != null) {
                message.writeTo(this.writer);
                this.writer.println("");
                if (!this.buffered) {
                    this.writer.flush();
                }
            }
        }
    } catch (IOException var7) {
        log.warn(sm.getString("accessLogValve.writeFail", new Object[]{message.toString()}), var7);
    }

}

您应该记录该日志，然后您将知道如何配置。

然后让我们开始，你应该创建一个类扩展ValveBase实现AccessLog，如：

public class LeKafkaAccesslogValve extends ValveBase implements AccessLog {
    private String topic;
    private String bootstrapServers;

    //  If set to zero then the producer will not wait for any acknowledgment from the server at all.
    private String acks;

    private String producerSize ;

    private String properties;

    private List<Producer<byte[], byte[]>> producerList;
    private AtomicInteger producerIndex = new AtomicInteger(0);
    private int timeoutMillis;
    private boolean enabled = true; 

    private String pattern;
    private AccessLogElement accessLogElement;
    private String localeName;
    private Locale locale = Locale.getDefault();


    @Override
    public void log(Request request, Response response, long l) {
        if (producerList != null && getEnabled() && getState().isAvailable() && null != this.accessLogElement) {
            try {
                getNextProducer().send(new ProducerRecord<byte[], byte[]>(topic, this.accessLogElement.buildLog(request,response,time,this).getBytes(StandardCharsets.UTF_8))).get(timeoutMillis, TimeUnit.MILLISECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error('accesslog in kafka exception', e);
            }
        }
    }

    @Override
    public void setRequestAttributesEnabled(boolean b) {
        //some other code if you would like
    }

    @Override
    public boolean getRequestAttributesEnabled() {
        //some other code if you would like
        return false;
    }

    @Override
    public void invoke(Request request, Response response) throws IOException, ServletException {
        //some other code if you would like
    }
}

然后，您应该将自己的配置添加到server.xml，如：

<Valve className='com.xxx.lekafkavalve.LeKafkaAccesslogValve'         enabled='true'  topic='info' pattern='%{yyyy-MM-dd     HH:mm:ss}t||info||AccessValve||Tomcat||%A||%a||%r||%s||%D' bootstrapServers='kafkaaddress' producerSize='5' properties='acks=0||producer.size=3'/>

Answer 2

好。此外，您可以将日志框架切换到log4j2以获得更高的效率，因此向kafka发送消息不会导致主要速度降低