View Javadoc
1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one or more
3    * contributor license agreements. See the NOTICE file distributed with
4    * this work for additional information regarding copyright ownership.
5    * The ASF licenses this file to You under the Apache license, Version 2.0
6    * (the "License"); you may not use this file except in compliance with
7    * the License. You may obtain a copy of the License at
8    *
9    *      http://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS,
13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14   * See the license for the specific language governing permissions and
15   * limitations under the license.
16   */
17  package org.apache.logging.log4j.core.appender;
18  
19  import java.io.Serializable;
20  import java.util.ArrayList;
21  import java.util.List;
22  import java.util.Map;
23  import java.util.concurrent.ArrayBlockingQueue;
24  import java.util.concurrent.BlockingQueue;
25  import java.util.concurrent.atomic.AtomicLong;
26  
27  import org.apache.logging.log4j.core.Appender;
28  import org.apache.logging.log4j.core.Filter;
29  import org.apache.logging.log4j.core.LogEvent;
30  import org.apache.logging.log4j.core.async.RingBufferLogEvent;
31  import org.apache.logging.log4j.core.config.AppenderControl;
32  import org.apache.logging.log4j.core.config.AppenderRef;
33  import org.apache.logging.log4j.core.config.Configuration;
34  import org.apache.logging.log4j.core.config.ConfigurationException;
35  import org.apache.logging.log4j.core.config.plugins.Plugin;
36  import org.apache.logging.log4j.core.config.plugins.PluginAliases;
37  import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
38  import org.apache.logging.log4j.core.config.plugins.PluginConfiguration;
39  import org.apache.logging.log4j.core.config.plugins.PluginElement;
40  import org.apache.logging.log4j.core.config.plugins.PluginFactory;
41  import org.apache.logging.log4j.core.impl.Log4jLogEvent;
42  
43  /**
44   * Appends to one or more Appenders asynchronously.  You can configure an
45   * AsyncAppender with one or more Appenders and an Appender to append to if the
46   * queue is full. The AsyncAppender does not allow a filter to be specified on
47   * the Appender references.
48   */
49  @Plugin(name = "Async", category = "Core", elementType = "appender", printObject = true)
50  public final class AsyncAppender extends AbstractAppender {
51  
52      private static final long serialVersionUID = 1L;
53      private static final int DEFAULT_QUEUE_SIZE = 128;
54      private static final String SHUTDOWN = "Shutdown";
55  
56      private final BlockingQueue<Serializable> queue;
57      private final int queueSize;
58      private final boolean blocking;
59      private final Configuration config;
60      private final AppenderRef[] appenderRefs;
61      private final String errorRef;
62      private final boolean includeLocation;
63      private AppenderControl errorAppender;
64      private AsyncThread thread;
65      private static final AtomicLong threadSequence = new AtomicLong(1);
66      private static ThreadLocal<Boolean> isAppenderThread = new ThreadLocal<>();
67  
68  
69      private AsyncAppender(final String name, final Filter filter, final AppenderRef[] appenderRefs,
70                             final String errorRef, final int queueSize, final boolean blocking,
71                             final boolean ignoreExceptions, final Configuration config,
72                             final boolean includeLocation) {
73          super(name, filter, null, ignoreExceptions);
74          this.queue = new ArrayBlockingQueue<>(queueSize);
75          this.queueSize = queueSize;
76          this.blocking = blocking;
77          this.config = config;
78          this.appenderRefs = appenderRefs;
79          this.errorRef = errorRef;
80          this.includeLocation = includeLocation;
81      }
82  
83      @Override
84      public void start() {
85          final Map<String, Appender> map = config.getAppenders();
86          final List<AppenderControl> appenders = new ArrayList<>();
87          for (final AppenderRef appenderRef : appenderRefs) {
88              final Appender appender = map.get(appenderRef.getRef());
89              if (appender != null) {
90                  appenders.add(new AppenderControl(appender, appenderRef.getLevel(), appenderRef.getFilter()));
91              } else {
92                  LOGGER.error("No appender named {} was configured", appenderRef);
93              }
94          }
95          if (errorRef != null) {
96              final Appender appender = map.get(errorRef);
97              if (appender != null) {
98                  errorAppender = new AppenderControl(appender, null, null);
99              } 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();
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
131      *        The LogEvent.
132      */
133     @Override
134     public void append(LogEvent logEvent) {
135         if (!isStarted()) {
136             throw new IllegalStateException("AsyncAppender " + getName() + " is not active");
137         }
138         if (!(logEvent instanceof Log4jLogEvent)) {
139             if (!(logEvent instanceof RingBufferLogEvent)) {
140                 return; // only know how to Serialize Log4jLogEvents and RingBufferLogEvents
141             }
142             logEvent = ((RingBufferLogEvent) logEvent).createMemento();
143         }
144         logEvent.getMessage().getFormattedMessage(); // LOG4J2-763: ask message to freeze parameters
145         final Log4jLogEvent coreEvent = (Log4jLogEvent) logEvent;
146         boolean appendSuccessful = false;
147         if (blocking) {
148             if (isAppenderThread.get() == Boolean.TRUE && queue.remainingCapacity() == 0) {
149                 // LOG4J2-485: avoid deadlock that would result from trying
150                 // to add to a full queue from appender thread
151                 coreEvent.setEndOfBatch(false); // queue is definitely not empty!
152                 appendSuccessful = thread.callAppenders(coreEvent);
153             } else {
154                 final Serializable serialized = Log4jLogEvent.serialize(coreEvent, includeLocation);
155                 try {
156                     // wait for free slots in the queue
157                     queue.put(serialized);
158                     appendSuccessful = true;
159                 } catch (final InterruptedException e) {
160                     // LOG4J2-1049: Some applications use Thread.interrupt() to send
161                     // messages between application threads. This does not necessarily
162                     // mean that the queue is full. To prevent dropping a log message,
163                     // quickly try to offer the event to the queue again.
164                     // (Yes, this means there is a possibility the same event is logged twice.)
165                     //
166                     // Finally, catching the InterruptedException means the
167                     // interrupted flag has been cleared on the current thread.
168                     // This may interfere with the application's expectation of
169                     // being interrupted, so when we are done, we set the interrupted
170                     // flag again.
171                     appendSuccessful = queue.offer(serialized);
172                     if (!appendSuccessful) {
173                         LOGGER.warn("Interrupted while waiting for a free slot in the AsyncAppender LogEvent-queue {}",
174                         getName());
175                     }
176                     // set the interrupted flag again.
177                     Thread.currentThread().interrupt();
178                 }
179             }
180         } else {
181             appendSuccessful = queue.offer(Log4jLogEvent.serialize(coreEvent, includeLocation));
182             if (!appendSuccessful) {
183                 error("Appender " + getName() + " is unable to write primary appenders. queue is full");
184             }
185         }
186         if (!appendSuccessful && errorAppender != null) {
187             errorAppender.callAppender(coreEvent);
188         }
189     }
190 
191     /**
192      * Create an AsyncAppender.
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 size The size of the event queue. The default is 128.
197      * @param name The name of the Appender.
198      * @param includeLocation whether to include location information. The default is false.
199      * @param filter The Filter or null.
200      * @param config The Configuration.
201      * @param ignoreExceptions If {@code "true"} (default) exceptions encountered when appending events are logged;
202      *                         otherwise they are propagated to the caller.
203      * @return The AsyncAppender.
204      */
205     @PluginFactory
206     public static AsyncAppender createAppender(@PluginElement("AppenderRef") final AppenderRef[] appenderRefs,
207             @PluginAttribute("errorRef") @PluginAliases("error-ref") final String errorRef,
208             @PluginAttribute(value = "blocking", defaultBoolean = true) final boolean blocking,
209             @PluginAttribute(value = "bufferSize", defaultInt = DEFAULT_QUEUE_SIZE) final int size,
210             @PluginAttribute("name") final String name,
211             @PluginAttribute(value = "includeLocation", defaultBoolean = false) final boolean includeLocation,
212             @PluginElement("Filter") final Filter filter,
213             @PluginConfiguration final Configuration config,
214             @PluginAttribute(value = "ignoreExceptions", defaultBoolean = true) final boolean ignoreExceptions) {
215         if (name == null) {
216             LOGGER.error("No name provided for AsyncAppender");
217             return null;
218         }
219         if (appenderRefs == null) {
220             LOGGER.error("No appender references provided to AsyncAppender {}", name);
221         }
222 
223         return new AsyncAppender(name, filter, appenderRefs, errorRef,
224                 size, blocking, ignoreExceptions, config, includeLocation);
225     }
226 
227     /**
228      * Thread that calls the Appenders.
229      */
230     private class AsyncThread extends Thread {
231 
232         private volatile boolean shutdown = false;
233         private final List<AppenderControl> appenders;
234         private final BlockingQueue<Serializable> queue;
235 
236         public AsyncThread(final List<AppenderControl> appenders, final BlockingQueue<Serializable> queue) {
237             this.appenders = appenders;
238             this.queue = queue;
239             setDaemon(true);
240             setName("AsyncAppenderThread" + threadSequence.getAndIncrement());
241         }
242 
243         @Override
244         public void run() {
245             isAppenderThread.set(Boolean.TRUE); // LOG4J2-485
246             while (!shutdown) {
247                 Serializable s;
248                 try {
249                     s = queue.take();
250                     if (s != null && s instanceof String && SHUTDOWN.equals(s.toString())) {
251                         shutdown = true;
252                         continue;
253                     }
254                 } catch (final InterruptedException ex) {
255                     break; // LOG4J2-830
256                 }
257                 final Log4jLogEvent event = Log4jLogEvent.deserialize(s);
258                 event.setEndOfBatch(queue.isEmpty());
259                 final boolean success = callAppenders(event);
260                 if (!success && errorAppender != null) {
261                     try {
262                         errorAppender.callAppender(event);
263                     } catch (final Exception ex) {
264                         // Silently accept the error.
265                     }
266                 }
267             }
268             // Process any remaining items in the queue.
269             LOGGER.trace("AsyncAppender.AsyncThread shutting down. Processing remaining {} queue events.",
270                     queue.size());
271             int count= 0;
272             int ignored = 0;
273             while (!queue.isEmpty()) {
274                 try {
275                     final Serializable s = queue.take();
276                     if (Log4jLogEvent.canDeserialize(s)) {
277                         final Log4jLogEvent event = Log4jLogEvent.deserialize(s);
278                         event.setEndOfBatch(queue.isEmpty());
279                         callAppenders(event);
280                         count++;
281                     } else {
282                         ignored++;
283                         LOGGER.trace("Ignoring event of class {}", s.getClass().getName());
284                     }
285                 } catch (final InterruptedException ex) {
286                     // May have been interrupted to shut down.
287                     // Here we ignore interrupts and try to process all remaining events.
288                 }
289             }
290             LOGGER.trace("AsyncAppender.AsyncThread stopped. Queue has {} events remaining. " +
291             		"Processed {} and ignored {} events since shutdown started.",
292             		queue.size(), count, ignored);
293         }
294 
295         /**
296          * Calls {@link AppenderControl#callAppender(LogEvent) callAppender} on
297          * all registered {@code AppenderControl} objects, and returns {@code true}
298          * if at least one appender call was successful, {@code false} otherwise.
299          * Any 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
327      * as an array of Strings.
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
340      * every log event to determine the class and method where the logging call
341      * was made.
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,
350      * or {@code false} if events are dropped when the queue is full.
351      * @return whether this AsyncAppender will block or drop events when the queue is full.
352      */
353     public boolean isBlocking() {
354         return blocking;
355     }
356 
357     /**
358      * Returns the name of the appender that any errors are logged to or {@code null}.
359      * @return the name of the appender that any errors are logged to or {@code null}
360      */
361     public String getErrorRef() {
362         return errorRef;
363     }
364 
365     public int getQueueCapacity() {
366         return queueSize;
367     }
368 
369     public int getQueueRemainingCapacity() {
370         return queue.remainingCapacity();
371     }
372 }