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.flume.appender; 018 019import java.io.Serializable; 020import java.util.Locale; 021 022import org.apache.logging.log4j.core.Filter; 023import org.apache.logging.log4j.core.Layout; 024import org.apache.logging.log4j.core.LogEvent; 025import org.apache.logging.log4j.core.appender.AbstractAppender; 026import org.apache.logging.log4j.core.config.Property; 027import org.apache.logging.log4j.core.config.plugins.Plugin; 028import org.apache.logging.log4j.core.config.plugins.PluginAliases; 029import org.apache.logging.log4j.core.config.plugins.PluginAttribute; 030import org.apache.logging.log4j.core.config.plugins.PluginElement; 031import org.apache.logging.log4j.core.config.plugins.PluginFactory; 032import org.apache.logging.log4j.core.layout.Rfc5424Layout; 033import org.apache.logging.log4j.core.net.Facility; 034import org.apache.logging.log4j.core.util.Booleans; 035import org.apache.logging.log4j.core.util.Integers; 036 037/** 038 * An Appender that uses the Avro protocol to route events to Flume. 039 */ 040@Plugin(name = "Flume", category = "Core", elementType = "appender", printObject = true) 041public final class FlumeAppender extends AbstractAppender implements FlumeEventFactory { 042 043 private static final String[] EXCLUDED_PACKAGES = {"org.apache.flume", "org.apache.avro"}; 044 private static final int DEFAULT_MAX_DELAY = 60000; 045 046 private static final int DEFAULT_LOCK_TIMEOUT_RETRY_COUNT = 5; 047 048 private final AbstractFlumeManager manager; 049 050 private final String mdcIncludes; 051 private final String mdcExcludes; 052 private final String mdcRequired; 053 054 private final String eventPrefix; 055 056 private final String mdcPrefix; 057 058 private final boolean compressBody; 059 060 private final FlumeEventFactory factory; 061 062 /** 063 * Which Manager will be used by the appender instance. 064 */ 065 private enum ManagerType { 066 AVRO, EMBEDDED, PERSISTENT; 067 068 public static ManagerType getType(final String type) { 069 return valueOf(type.toUpperCase(Locale.US)); 070 } 071 } 072 073 private FlumeAppender(final String name, final Filter filter, final Layout<? extends Serializable> layout, 074 final boolean ignoreExceptions, final String includes, final String excludes, 075 final String required, final String mdcPrefix, final String eventPrefix, 076 final boolean compress, final FlumeEventFactory factory, final AbstractFlumeManager manager) { 077 super(name, filter, layout, ignoreExceptions); 078 this.manager = manager; 079 this.mdcIncludes = includes; 080 this.mdcExcludes = excludes; 081 this.mdcRequired = required; 082 this.eventPrefix = eventPrefix; 083 this.mdcPrefix = mdcPrefix; 084 this.compressBody = compress; 085 this.factory = factory == null ? this : factory; 086 } 087 088 /** 089 * Publish the event. 090 * @param event The LogEvent. 091 */ 092 @Override 093 public void append(final LogEvent event) { 094 final String name = event.getLoggerName(); 095 if (name != null) { 096 for (final String pkg : EXCLUDED_PACKAGES) { 097 if (name.startsWith(pkg)) { 098 return; 099 } 100 } 101 } 102 final FlumeEvent flumeEvent = factory.createEvent(event, mdcIncludes, mdcExcludes, mdcRequired, mdcPrefix, 103 eventPrefix, compressBody); 104 flumeEvent.setBody(getLayout().toByteArray(flumeEvent)); 105 manager.send(flumeEvent); 106 } 107 108 @Override 109 public void stop() { 110 super.stop(); 111 manager.release(); 112 } 113 114 /** 115 * Create a Flume event. 116 * @param event The Log4j LogEvent. 117 * @param includes comma separated list of mdc elements to include. 118 * @param excludes comma separated list of mdc elements to exclude. 119 * @param required comma separated list of mdc elements that must be present with a value. 120 * @param mdcPrefix The prefix to add to MDC key names. 121 * @param eventPrefix The prefix to add to event fields. 122 * @param compress If true the body will be compressed. 123 * @return A Flume Event. 124 */ 125 @Override 126 public FlumeEvent createEvent(final LogEvent event, final String includes, final String excludes, 127 final String required, final String mdcPrefix, final String eventPrefix, 128 final boolean compress) { 129 return new FlumeEvent(event, mdcIncludes, mdcExcludes, mdcRequired, mdcPrefix, 130 eventPrefix, compressBody); 131 } 132 133 /** 134 * Create a Flume Avro Appender. 135 * @param agents An array of Agents. 136 * @param properties Properties to pass to the embedded agent. 137 * @param embedded true if the embedded agent manager should be used. otherwise the Avro manager will be used. 138 * <b>Note: </b><i>The embedded attribute is deprecated in favor of specifying the type attribute.</i> 139 * @param type Avro (default), Embedded, or Persistent. 140 * @param dataDir The directory where the Flume FileChannel should write its data. 141 * @param connectionTimeoutMillis The amount of time in milliseconds to wait before a connection times out. Minimum is 142 * 1000. 143 * @param requestTimeoutMillis The amount of time in milliseconds to wait before a request times out. Minimum is 1000. 144 * @param agentRetries The number of times to retry an agent before failing to the next agent. 145 * @param maxDelayMillis The maximum number of milliseconds to wait for a complete batch. 146 * @param name The name of the Appender. 147 * @param ignore If {@code "true"} (default) exceptions encountered when appending events are logged; otherwise 148 * they are propagated to the caller. 149 * @param excludes A comma separated list of MDC elements to exclude. 150 * @param includes A comma separated list of MDC elements to include. 151 * @param required A comma separated list of MDC elements that are required. 152 * @param mdcPrefix The prefix to add to MDC key names. 153 * @param eventPrefix The prefix to add to event key names. 154 * @param compressBody If true the event body will be compressed. 155 * @param batchSize Number of events to include in a batch. Defaults to 1. 156 * @param lockTimeoutRetries Times to retry a lock timeout when writing to Berkeley DB. 157 * @param factory The factory to use to create Flume events. 158 * @param layout The layout to format the event. 159 * @param filter A Filter to filter events. 160 * 161 * @return A Flume Avro Appender. 162 */ 163 @PluginFactory 164 public static FlumeAppender createAppender(@PluginElement("Agents") Agent[] agents, 165 @PluginElement("Properties") final Property[] properties, 166 @PluginAttribute("embedded") final String embedded, 167 @PluginAttribute("type") final String type, 168 @PluginAttribute("dataDir") final String dataDir, 169 @PluginAliases("connectTimeout") 170 @PluginAttribute("connectTimeoutMillis") final String connectionTimeoutMillis, 171 @PluginAliases("requestTimeout") 172 @PluginAttribute("requestTimeoutMillis") final String requestTimeoutMillis, 173 @PluginAttribute("agentRetries") final String agentRetries, 174 @PluginAliases("maxDelay") // deprecated 175 @PluginAttribute("maxDelayMillis") final String maxDelayMillis, 176 @PluginAttribute("name") final String name, 177 @PluginAttribute("ignoreExceptions") final String ignore, 178 @PluginAttribute("mdcExcludes") final String excludes, 179 @PluginAttribute("mdcIncludes") final String includes, 180 @PluginAttribute("mdcRequired") final String required, 181 @PluginAttribute("mdcPrefix") final String mdcPrefix, 182 @PluginAttribute("eventPrefix") final String eventPrefix, 183 @PluginAttribute("compress") final String compressBody, 184 @PluginAttribute("batchSize") final String batchSize, 185 @PluginAttribute("lockTimeoutRetries") final String lockTimeoutRetries, 186 @PluginElement("FlumeEventFactory") final FlumeEventFactory factory, 187 @PluginElement("Layout") Layout<? extends Serializable> layout, 188 @PluginElement("Filter") final Filter filter) { 189 190 final boolean embed = embedded != null ? Boolean.parseBoolean(embedded) : 191 (agents == null || agents.length == 0) && properties != null && properties.length > 0; 192 final boolean ignoreExceptions = Booleans.parseBoolean(ignore, true); 193 final boolean compress = Booleans.parseBoolean(compressBody, true); 194 ManagerType managerType; 195 if (type != null) { 196 if (embed && embedded != null) { 197 try { 198 managerType = ManagerType.getType(type); 199 LOGGER.warn("Embedded and type attributes are mutually exclusive. Using type " + type); 200 } catch (final Exception ex) { 201 LOGGER.warn("Embedded and type attributes are mutually exclusive and type " + type + 202 " is invalid."); 203 managerType = ManagerType.EMBEDDED; 204 } 205 } else { 206 try { 207 managerType = ManagerType.getType(type); 208 } catch (final Exception ex) { 209 LOGGER.warn("Type " + type + " is invalid."); 210 managerType = ManagerType.EMBEDDED; 211 } 212 } 213 } else if (embed) { 214 managerType = ManagerType.EMBEDDED; 215 } else { 216 managerType = ManagerType.AVRO; 217 } 218 219 final int batchCount = Integers.parseInt(batchSize, 1); 220 final int connectTimeoutMillis = Integers.parseInt(connectionTimeoutMillis, 0); 221 final int reqTimeoutMillis = Integers.parseInt(requestTimeoutMillis, 0); 222 final int retries = Integers.parseInt(agentRetries, 0); 223 final int lockTimeoutRetryCount = Integers.parseInt(lockTimeoutRetries, DEFAULT_LOCK_TIMEOUT_RETRY_COUNT); 224 final int delayMillis = Integers.parseInt(maxDelayMillis, DEFAULT_MAX_DELAY); 225 226 if (layout == null) { 227 final int enterpriseNumber = Rfc5424Layout.DEFAULT_ENTERPRISE_NUMBER; 228 layout = Rfc5424Layout.createLayout(Facility.LOCAL0, null, enterpriseNumber, true, Rfc5424Layout.DEFAULT_MDCID, 229 mdcPrefix, eventPrefix, false, null, null, null, excludes, includes, required, null, false, null, 230 null); 231 } 232 233 if (name == null) { 234 LOGGER.error("No name provided for Appender"); 235 return null; 236 } 237 238 AbstractFlumeManager manager; 239 240 switch (managerType) { 241 case EMBEDDED: 242 manager = FlumeEmbeddedManager.getManager(name, agents, properties, batchCount, dataDir); 243 break; 244 case AVRO: 245 if (agents == null || agents.length == 0) { 246 LOGGER.debug("No agents provided, using defaults"); 247 agents = new Agent[] {Agent.createAgent(null, null)}; 248 } 249 manager = FlumeAvroManager.getManager(name, agents, batchCount, delayMillis, retries, connectTimeoutMillis, reqTimeoutMillis); 250 break; 251 case PERSISTENT: 252 if (agents == null || agents.length == 0) { 253 LOGGER.debug("No agents provided, using defaults"); 254 agents = new Agent[] {Agent.createAgent(null, null)}; 255 } 256 manager = FlumePersistentManager.getManager(name, agents, properties, batchCount, retries, 257 connectTimeoutMillis, reqTimeoutMillis, delayMillis, lockTimeoutRetryCount, dataDir); 258 break; 259 default: 260 LOGGER.debug("No manager type specified. Defaulting to AVRO"); 261 if (agents == null || agents.length == 0) { 262 LOGGER.debug("No agents provided, using defaults"); 263 agents = new Agent[] {Agent.createAgent(null, null)}; 264 } 265 manager = FlumeAvroManager.getManager(name, agents, batchCount, delayMillis, retries, connectTimeoutMillis, reqTimeoutMillis); 266 } 267 268 if (manager == null) { 269 return null; 270 } 271 272 return new FlumeAppender(name, filter, layout, ignoreExceptions, includes, 273 excludes, required, mdcPrefix, eventPrefix, compress, factory, manager); 274 } 275}