log4j2的GelfAppender - code-craft - SegmentFault 思否

安宇雨 - 随手采集
2019-04-29 18:13:33
随手采集
0000-未整理-等待研究

GelfAppender

 @Override
    public void append(LogEvent event) {
        final Layout<? extends Serializable> layout = getLayout();
        final String formattedMessage;
        if (layout == null) {
            formattedMessage = event.getMessage().getFormattedMessage();
        } else {
            formattedMessage = new String(layout.toByteArray(event), StandardCharsets.UTF_8);
        }

        final GelfMessageBuilder builder = new GelfMessageBuilder(formattedMessage, hostName)
                .timestamp(event.getTimeMillis() / 1000d)
                .level(GelfMessageLevel.fromNumericLevel(Severity.getSeverity(event.getLevel()).getCode()))
                .additionalField("loggerName", event.getLoggerName())
                .additionalField("threadName", event.getThreadName());

        final Marker marker = event.getMarker();
        if (marker != null) {
            builder.additionalField("marker", marker.getName());
        }

        if (includeThreadContext) {
            for (Map.Entry<String, String> entry : event.getContextMap().entrySet()) {
                builder.additionalField(entry.getKey(), entry.getValue());
            }

            // Guard against https://issues.apache.org/jira/browse/LOG4J2-1530
            final ThreadContext.ContextStack contextStack = event.getContextStack();
            if (contextStack != null) {
                final List<String> contextStackItems = contextStack.asList();
                if (contextStackItems != null && !contextStackItems.isEmpty()) {
                    builder.additionalField("contextStack", contextStackItems.toString());
                }
            }
        }

        if (includeSource) {
            final StackTraceElement source = event.getSource();
            if (source != null) {
                builder.additionalField("sourceFileName", source.getFileName());
                builder.additionalField("sourceMethodName", source.getMethodName());
                builder.additionalField("sourceClassName", source.getClassName());
                builder.additionalField("sourceLineNumber", source.getLineNumber());
            }
        }

        @SuppressWarnings("all")
        final Throwable thrown = event.getThrown();
        if (includeStackTrace && thrown != null) {
            String stackTrace;
            if (includeExceptionCause) {
                final StringWriter stringWriter = new StringWriter();
                final PrintWriter printWriter = new PrintWriter(stringWriter);
                thrown.printStackTrace(printWriter);
                stackTrace = stringWriter.toString();
            } else {
                stackTrace = getSimpleStacktraceAsString(thrown);
            }

            builder.additionalField("exceptionClass", thrown.getClass().getCanonicalName());
            builder.additionalField("exceptionMessage", thrown.getMessage());
            builder.additionalField("exceptionStackTrace", stackTrace);

            builder.fullMessage(formattedMessage);
        }

        if (!additionalFields.isEmpty()) {
            builder.additionalFields(additionalFields);
        }

        final GelfMessage gelfMessage = builder.build();
        try {
            final boolean sent = client.trySend(gelfMessage);
            if (!sent) {
                LOG.debug("Couldn't send message: {}", gelfMessage);
            }
        } catch (Exception e) {
            throw new AppenderLoggingException("failed to write log event to GELF server: " + e.getMessage(), e);
        }
    }

gelfclient

基于netty来实现

public class GelfTcpTransport extends AbstractGelfTransport {
    private static final Logger LOG = LoggerFactory.getLogger(GelfTcpTransport.class);

    /**
     * Creates a new TCP GELF transport.
     *
     * @param config the GELF client configuration
     */
    public GelfTcpTransport(GelfConfiguration config) {
        super(config);
    }

    @Override
    protected void createBootstrap(final EventLoopGroup workerGroup) {
        final Bootstrap bootstrap = new Bootstrap();
        final GelfSenderThread senderThread = new GelfSenderThread(queue, config.getMaxInflightSends());

        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout())
                .option(ChannelOption.TCP_NODELAY, config.isTcpNoDelay())
                .option(ChannelOption.SO_KEEPALIVE, config.isTcpKeepAlive())
                .remoteAddress(config.getRemoteAddress())
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        if (config.isTlsEnabled()) {
                            LOG.debug("TLS enabled.");
                            final SslContext sslContext;

                            if (!config.isTlsCertVerificationEnabled()) {
                                // If the cert should not be verified just use an insecure trust manager.
                                LOG.debug("TLS certificate verification disabled!");
                                sslContext = SslContext.newClientContext(InsecureTrustManagerFactory.INSTANCE);
                            } else if (config.getTlsTrustCertChainFile() != null) {
                                // If a cert chain file is set, use it.
                                LOG.debug("TLS certificate chain file: {}", config.getTlsTrustCertChainFile());
                                sslContext = SslContext.newClientContext(config.getTlsTrustCertChainFile());
                            } else {
                                // Otherwise use the JVM default cert chain.
                                sslContext = SslContext.newClientContext();
                            }

                            ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
                        }

                        // The graylog2-server uses '\0'-bytes as delimiter for TCP frames.
                        ch.pipeline().addLast(new GelfTcpFrameDelimiterEncoder());
                        // We cannot use GZIP encoding for TCP because the headers contain '\0'-bytes then.
                        ch.pipeline().addLast(new GelfMessageJsonEncoder());
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
                                // We do not receive data.
                            }

                            @Override
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                senderThread.start(ctx.channel());
                            }

                            @Override
                            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                                LOG.info("Channel disconnected!");
                                senderThread.stop();
                                scheduleReconnect(ctx.channel().eventLoop());
                            }

                            @Override
                            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                                LOG.error("Exception caught", cause);
                            }
                        });
                    }
                });

        if (config.getSendBufferSize() != -1) {
            bootstrap.option(ChannelOption.SO_SNDBUF, config.getSendBufferSize());
        }

        bootstrap.connect().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    LOG.debug("Connected!");
                } else {
                    LOG.error("Connection failed: {}", future.cause().getMessage());
                    scheduleReconnect(future.channel().eventLoop());
                }
            }
        });
    }
}