001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache license, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the license for the specific language governing permissions and
015 * limitations under the license.
016 */
017package org.apache.logging.log4j.core.appender;
018
019import java.io.Serializable;
020import java.util.ArrayList;
021import java.util.List;
022import java.util.Map;
023import java.util.concurrent.ArrayBlockingQueue;
024import java.util.concurrent.BlockingQueue;
025import java.util.concurrent.atomic.AtomicLong;
026
027import org.apache.logging.log4j.core.Appender;
028import org.apache.logging.log4j.core.Filter;
029import org.apache.logging.log4j.core.LogEvent;
030import org.apache.logging.log4j.core.async.RingBufferLogEvent;
031import org.apache.logging.log4j.core.config.AppenderControl;
032import org.apache.logging.log4j.core.config.AppenderRef;
033import org.apache.logging.log4j.core.config.Configuration;
034import org.apache.logging.log4j.core.config.ConfigurationException;
035import org.apache.logging.log4j.core.config.plugins.Plugin;
036import org.apache.logging.log4j.core.config.plugins.PluginAliases;
037import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
038import org.apache.logging.log4j.core.config.plugins.PluginConfiguration;
039import org.apache.logging.log4j.core.config.plugins.PluginElement;
040import org.apache.logging.log4j.core.config.plugins.PluginFactory;
041import org.apache.logging.log4j.core.impl.Log4jLogEvent;
042
043/**
044 * Appends to one or more Appenders asynchronously. You can configure an AsyncAppender with one or more Appenders and an
045 * Appender to append to if the queue is full. The AsyncAppender does not allow a filter to be specified on the Appender
046 * references.
047 */
048@Plugin(name = "Async", category = "Core", elementType = "appender", printObject = true)
049public final class AsyncAppender extends AbstractAppender {
050
051    private static final long serialVersionUID = 1L;
052    private static final int DEFAULT_QUEUE_SIZE = 128;
053    private static final String SHUTDOWN = "Shutdown";
054
055    private static final AtomicLong THREAD_SEQUENCE = new AtomicLong(1);
056    private static ThreadLocal<Boolean> isAppenderThread = new ThreadLocal<>();
057
058    private final BlockingQueue<Serializable> queue;
059    private final int queueSize;
060    private final boolean blocking;
061    private final long shutdownTimeout;
062    private final Configuration config;
063    private final AppenderRef[] appenderRefs;
064    private final String errorRef;
065    private final boolean includeLocation;
066    private AppenderControl errorAppender;
067    private AsyncThread thread;
068
069    private AsyncAppender(final String name, final Filter filter, final AppenderRef[] appenderRefs,
070            final String errorRef, final int queueSize, final boolean blocking, final boolean ignoreExceptions,
071            final long shutdownTimeout, final Configuration config, final boolean includeLocation) {
072        super(name, filter, null, ignoreExceptions);
073        this.queue = new ArrayBlockingQueue<>(queueSize);
074        this.queueSize = queueSize;
075        this.blocking = blocking;
076        this.shutdownTimeout = shutdownTimeout;
077        this.config = config;
078        this.appenderRefs = appenderRefs;
079        this.errorRef = errorRef;
080        this.includeLocation = includeLocation;
081    }
082
083    @Override
084    public void start() {
085        final Map<String, Appender> map = config.getAppenders();
086        final List<AppenderControl> appenders = new ArrayList<>();
087        for (final AppenderRef appenderRef : appenderRefs) {
088            final Appender appender = map.get(appenderRef.getRef());
089            if (appender != null) {
090                appenders.add(new AppenderControl(appender, appenderRef.getLevel(), appenderRef.getFilter()));
091            } else {
092                LOGGER.error("No appender named {} was configured", appenderRef);
093            }
094        }
095        if (errorRef != null) {
096            final Appender appender = map.get(errorRef);
097            if (appender != null) {
098                errorAppender = new AppenderControl(appender, null, null);
099            } else {
100                LOGGER.error("Unable to set up error Appender. No appender named {} was configured", errorRef);
101            }
102        }
103        if (appenders.size() > 0) {
104            thread = new AsyncThread(appenders, queue);
105            thread.setName("AsyncAppender-" + getName());
106        } else if (errorRef == null) {
107            throw new ConfigurationException("No appenders are available for AsyncAppender " + getName());
108        }
109
110        thread.start();
111        super.start();
112    }
113
114    @Override
115    public void stop() {
116        super.stop();
117        LOGGER.trace("AsyncAppender stopping. Queue still has {} events.", queue.size());
118        thread.shutdown();
119        try {
120            thread.join(shutdownTimeout);
121        } catch (final InterruptedException ex) {
122            LOGGER.warn("Interrupted while stopping AsyncAppender {}", getName());
123        }
124        LOGGER.trace("AsyncAppender stopped. Queue has {} events.", queue.size());
125    }
126
127    /**
128     * Actual writing occurs here.
129     * 
130     * @param logEvent The LogEvent.
131     */
132    @Override
133    public void append(LogEvent logEvent) {
134        if (!isStarted()) {
135            throw new IllegalStateException("AsyncAppender " + getName() + " is not active");
136        }
137        if (!(logEvent instanceof Log4jLogEvent)) {
138            if (!(logEvent instanceof RingBufferLogEvent)) {
139                return; // only know how to Serialize Log4jLogEvents and RingBufferLogEvents
140            }
141            logEvent = ((RingBufferLogEvent) logEvent).createMemento();
142        }
143        logEvent.getMessage().getFormattedMessage(); // LOG4J2-763: ask message to freeze parameters
144        final Log4jLogEvent coreEvent = (Log4jLogEvent) logEvent;
145        boolean appendSuccessful = false;
146        if (blocking) {
147            if (isAppenderThread.get() == Boolean.TRUE && queue.remainingCapacity() == 0) {
148                // LOG4J2-485: avoid deadlock that would result from trying
149                // to add to a full queue from appender thread
150                coreEvent.setEndOfBatch(false); // queue is definitely not empty!
151                appendSuccessful = thread.callAppenders(coreEvent);
152            } else {
153                final Serializable serialized = Log4jLogEvent.serialize(coreEvent, includeLocation);
154                try {
155                    // wait for free slots in the queue
156                    queue.put(serialized);
157                    appendSuccessful = true;
158                } catch (final InterruptedException e) {
159                    // LOG4J2-1049: Some applications use Thread.interrupt() to send
160                    // messages between application threads. This does not necessarily
161                    // mean that the queue is full. To prevent dropping a log message,
162                    // quickly try to offer the event to the queue again.
163                    // (Yes, this means there is a possibility the same event is logged twice.)
164                    //
165                    // Finally, catching the InterruptedException means the
166                    // interrupted flag has been cleared on the current thread.
167                    // This may interfere with the application's expectation of
168                    // being interrupted, so when we are done, we set the interrupted
169                    // flag again.
170                    appendSuccessful = queue.offer(serialized);
171                    if (!appendSuccessful) {
172                        LOGGER.warn("Interrupted while waiting for a free slot in the AsyncAppender LogEvent-queue {}",
173                                getName());
174                    }
175                    // set the interrupted flag again.
176                    Thread.currentThread().interrupt();
177                }
178            }
179        } else {
180            appendSuccessful = queue.offer(Log4jLogEvent.serialize(coreEvent, includeLocation));
181            if (!appendSuccessful) {
182                error("Appender " + getName() + " is unable to write primary appenders. queue is full");
183            }
184        }
185        if (!appendSuccessful && errorAppender != null) {
186            errorAppender.callAppender(coreEvent);
187        }
188    }
189
190    /**
191     * Create an AsyncAppender.
192     * 
193     * @param appenderRefs The Appenders to reference.
194     * @param errorRef An optional Appender to write to if the queue is full or other errors occur.
195     * @param blocking True if the Appender should wait when the queue is full. The default is true.
196     * @param shutdownTimeout How many milliseconds the Appender should wait to flush outstanding log events
197     *                        in the queue on shutdown. The default is zero which means to wait forever.
198     * @param size The size of the event queue. The default is 128.
199     * @param name The name of the Appender.
200     * @param includeLocation whether to include location information. The default is false.
201     * @param filter The Filter or null.
202     * @param config The Configuration.
203     * @param ignoreExceptions If {@code "true"} (default) exceptions encountered when appending events are logged;
204     *            otherwise they are propagated to the caller.
205     * @return The AsyncAppender.
206     */
207    @PluginFactory
208    public static AsyncAppender createAppender(@PluginElement("AppenderRef") final AppenderRef[] appenderRefs,
209            @PluginAttribute("errorRef") @PluginAliases("error-ref") final String errorRef,
210            @PluginAttribute(value = "blocking", defaultBoolean = true) final boolean blocking,
211            @PluginAttribute(value = "shutdownTimeout", defaultLong = 0L) final long shutdownTimeout,
212            @PluginAttribute(value = "bufferSize", defaultInt = DEFAULT_QUEUE_SIZE) final int size,
213            @PluginAttribute("name") final String name,
214            @PluginAttribute(value = "includeLocation", defaultBoolean = false) final boolean includeLocation,
215            @PluginElement("Filter") final Filter filter, @PluginConfiguration final Configuration config,
216            @PluginAttribute(value = "ignoreExceptions", defaultBoolean = true) final boolean ignoreExceptions) {
217        if (name == null) {
218            LOGGER.error("No name provided for AsyncAppender");
219            return null;
220        }
221        if (appenderRefs == null) {
222            LOGGER.error("No appender references provided to AsyncAppender {}", name);
223        }
224
225        return new AsyncAppender(name, filter, appenderRefs, errorRef, size, blocking, ignoreExceptions,
226                shutdownTimeout, config, includeLocation);
227    }
228
229    /**
230     * Thread that calls the Appenders.
231     */
232    private class AsyncThread extends Thread {
233
234        private volatile boolean shutdown = false;
235        private final List<AppenderControl> appenders;
236        private final BlockingQueue<Serializable> queue;
237
238        public AsyncThread(final List<AppenderControl> appenders, final BlockingQueue<Serializable> queue) {
239            this.appenders = appenders;
240            this.queue = queue;
241            setDaemon(true);
242            setName("AsyncAppenderThread" + THREAD_SEQUENCE.getAndIncrement());
243        }
244
245        @Override
246        public void run() {
247            isAppenderThread.set(Boolean.TRUE); // LOG4J2-485
248            while (!shutdown) {
249                Serializable s;
250                try {
251                    s = queue.take();
252                    if (s != null && s instanceof String && SHUTDOWN.equals(s.toString())) {
253                        shutdown = true;
254                        continue;
255                    }
256                } catch (final InterruptedException ex) {
257                    break; // LOG4J2-830
258                }
259                final Log4jLogEvent event = Log4jLogEvent.deserialize(s);
260                event.setEndOfBatch(queue.isEmpty());
261                final boolean success = callAppenders(event);
262                if (!success && errorAppender != null) {
263                    try {
264                        errorAppender.callAppender(event);
265                    } catch (final Exception ex) {
266                        // Silently accept the error.
267                    }
268                }
269            }
270            // Process any remaining items in the queue.
271            LOGGER.trace("AsyncAppender.AsyncThread shutting down. Processing remaining {} queue events.",
272                    queue.size());
273            int count = 0;
274            int ignored = 0;
275            while (!queue.isEmpty()) {
276                try {
277                    final Serializable s = queue.take();
278                    if (Log4jLogEvent.canDeserialize(s)) {
279                        final Log4jLogEvent event = Log4jLogEvent.deserialize(s);
280                        event.setEndOfBatch(queue.isEmpty());
281                        callAppenders(event);
282                        count++;
283                    } else {
284                        ignored++;
285                        LOGGER.trace("Ignoring event of class {}", s.getClass().getName());
286                    }
287                } catch (final InterruptedException ex) {
288                    // May have been interrupted to shut down.
289                    // Here we ignore interrupts and try to process all remaining events.
290                }
291            }
292            LOGGER.trace("AsyncAppender.AsyncThread stopped. Queue has {} events remaining. "
293                    + "Processed {} and ignored {} events since shutdown started.", queue.size(), count, ignored);
294        }
295
296        /**
297         * Calls {@link AppenderControl#callAppender(LogEvent) callAppender} on all registered {@code AppenderControl}
298         * objects, and returns {@code true} if at least one appender call was successful, {@code false} otherwise. Any
299         * exceptions are silently ignored.
300         *
301         * @param event the event to forward to the registered appenders
302         * @return {@code true} if at least one appender call succeeded, {@code false} otherwise
303         */
304        boolean callAppenders(final Log4jLogEvent event) {
305            boolean success = false;
306            for (final AppenderControl control : appenders) {
307                try {
308                    control.callAppender(event);
309                    success = true;
310                } catch (final Exception ex) {
311                    // If no appender is successful the error appender will get it.
312                }
313            }
314            return success;
315        }
316
317        public void shutdown() {
318            shutdown = true;
319            if (queue.isEmpty()) {
320                queue.offer(SHUTDOWN);
321            }
322        }
323    }
324
325    /**
326     * Returns the names of the appenders that this asyncAppender delegates to as an array of Strings.
327     * 
328     * @return the names of the sink appenders
329     */
330    public String[] getAppenderRefStrings() {
331        final String[] result = new String[appenderRefs.length];
332        for (int i = 0; i < result.length; i++) {
333            result[i] = appenderRefs[i].getRef();
334        }
335        return result;
336    }
337
338    /**
339     * Returns {@code true} if this AsyncAppender will take a snapshot of the stack with every log event to determine
340     * the class and method where the logging call was made.
341     * 
342     * @return {@code true} if location is included with every event, {@code false} otherwise
343     */
344    public boolean isIncludeLocation() {
345        return includeLocation;
346    }
347
348    /**
349     * Returns {@code true} if this AsyncAppender will block when the queue is full, or {@code false} if events are
350     * dropped when the queue is full.
351     * 
352     * @return whether this AsyncAppender will block or drop events when the queue is full.
353     */
354    public boolean isBlocking() {
355        return blocking;
356    }
357
358    /**
359     * Returns the name of the appender that any errors are logged to or {@code null}.
360     * 
361     * @return the name of the appender that any errors are logged to or {@code null}
362     */
363    public String getErrorRef() {
364        return errorRef;
365    }
366
367    public int getQueueCapacity() {
368        return queueSize;
369    }
370
371    public int getQueueRemainingCapacity() {
372        return queue.remainingCapacity();
373    }
374}