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.ByteArrayOutputStream; 020import java.io.IOException; 021import java.util.ArrayList; 022import java.util.HashMap; 023import java.util.List; 024import java.util.Map; 025import java.util.zip.GZIPOutputStream; 026 027import org.apache.flume.event.SimpleEvent; 028import org.apache.logging.log4j.Level; 029import org.apache.logging.log4j.LoggingException; 030import org.apache.logging.log4j.Marker; 031import org.apache.logging.log4j.ThreadContext; 032import org.apache.logging.log4j.core.LogEvent; 033import org.apache.logging.log4j.core.impl.Log4jLogEvent; 034import org.apache.logging.log4j.core.impl.ThrowableProxy; 035import org.apache.logging.log4j.core.util.Patterns; 036import org.apache.logging.log4j.core.util.UuidUtil; 037import org.apache.logging.log4j.message.MapMessage; 038import org.apache.logging.log4j.message.Message; 039import org.apache.logging.log4j.message.StructuredDataId; 040import org.apache.logging.log4j.message.StructuredDataMessage; 041import org.apache.logging.log4j.util.ReadOnlyStringMap; 042import org.apache.logging.log4j.util.Strings; 043 044/** 045 * Class that is both a Flume and Log4j Event. 046 */ 047public class FlumeEvent extends SimpleEvent implements LogEvent { 048 049 static final String GUID = "guId"; 050 /** 051 * Generated serial version ID. 052 */ 053 private static final long serialVersionUID = -8988674608627854140L; 054 055 private static final String DEFAULT_MDC_PREFIX = Strings.EMPTY; 056 057 private static final String DEFAULT_EVENT_PREFIX = Strings.EMPTY; 058 059 private static final String EVENT_TYPE = "eventType"; 060 061 private static final String EVENT_ID = "eventId"; 062 063 private static final String TIMESTAMP = "timeStamp"; 064 065 private final LogEvent event; 066 067 private final Map<String, String> contextMap = new HashMap<>(); 068 069 private final boolean compress; 070 071 /** 072 * Construct the FlumeEvent. 073 * @param event The Log4j LogEvent. 074 * @param includes A comma separated list of MDC elements to include. 075 * @param excludes A comma separated list of MDC elements to exclude. 076 * @param required A comma separated list of MDC elements that are required to be defined. 077 * @param mdcPrefix The value to prefix to MDC keys. 078 * @param eventPrefix The value to prefix to event keys. 079 * @param compress If true the event body should be compressed. 080 */ 081 public FlumeEvent(final LogEvent event, final String includes, final String excludes, final String required, 082 String mdcPrefix, String eventPrefix, final boolean compress) { 083 this.event = event; 084 this.compress = compress; 085 final Map<String, String> headers = getHeaders(); 086 headers.put(TIMESTAMP, Long.toString(event.getTimeMillis())); 087 if (mdcPrefix == null) { 088 mdcPrefix = DEFAULT_MDC_PREFIX; 089 } 090 if (eventPrefix == null) { 091 eventPrefix = DEFAULT_EVENT_PREFIX; 092 } 093 final Map<String, String> mdc = event.getContextData().toMap(); 094 if (includes != null) { 095 final String[] array = includes.split(Patterns.COMMA_SEPARATOR); 096 if (array.length > 0) { 097 for (String str : array) { 098 str = str.trim(); 099 if (mdc.containsKey(str)) { 100 contextMap.put(str, mdc.get(str)); 101 } 102 } 103 } 104 } else if (excludes != null) { 105 final String[] array = excludes.split(Patterns.COMMA_SEPARATOR); 106 if (array.length > 0) { 107 final List<String> list = new ArrayList<>(array.length); 108 for (final String value : array) { 109 list.add(value.trim()); 110 } 111 for (final Map.Entry<String, String> entry : mdc.entrySet()) { 112 if (!list.contains(entry.getKey())) { 113 contextMap.put(entry.getKey(), entry.getValue()); 114 } 115 } 116 } 117 } else { 118 contextMap.putAll(mdc); 119 } 120 121 if (required != null) { 122 final String[] array = required.split(Patterns.COMMA_SEPARATOR); 123 if (array.length > 0) { 124 for (String str : array) { 125 str = str.trim(); 126 if (!mdc.containsKey(str)) { 127 throw new LoggingException("Required key " + str + " is missing from the MDC"); 128 } 129 } 130 } 131 } 132 final String guid = UuidUtil.getTimeBasedUuid().toString(); 133 final Message message = event.getMessage(); 134 if (message instanceof MapMessage) { 135 // Add the guid to the Map so that it can be included in the Layout. 136 @SuppressWarnings("unchecked") 137 final 138 MapMessage<?, String> stringMapMessage = (MapMessage<?, String>) message; 139 stringMapMessage.put(GUID, guid); 140 if (message instanceof StructuredDataMessage) { 141 addStructuredData(eventPrefix, headers, (StructuredDataMessage) message); 142 } 143 addMapData(eventPrefix, headers, stringMapMessage); 144 } else { 145 headers.put(GUID, guid); 146 } 147 148 addContextData(mdcPrefix, headers, contextMap); 149 } 150 151 protected void addStructuredData(final String prefix, final Map<String, String> fields, 152 final StructuredDataMessage msg) { 153 fields.put(prefix + EVENT_TYPE, msg.getType()); 154 final StructuredDataId id = msg.getId(); 155 fields.put(prefix + EVENT_ID, id.getName()); 156 } 157 158 protected void addMapData(final String prefix, final Map<String, String> fields, final MapMessage<?, String> msg) { 159 final Map<String, String> data = msg.getData(); 160 for (final Map.Entry<String, String> entry : data.entrySet()) { 161 fields.put(prefix + entry.getKey(), entry.getValue()); 162 } 163 } 164 165 protected void addContextData(final String prefix, final Map<String, String> fields, 166 final Map<String, String> context) { 167 final Map<String, String> map = new HashMap<>(); 168 for (final Map.Entry<String, String> entry : context.entrySet()) { 169 if (entry.getKey() != null && entry.getValue() != null) { 170 fields.put(prefix + entry.getKey(), entry.getValue()); 171 map.put(prefix + entry.getKey(), entry.getValue()); 172 } 173 } 174 context.clear(); 175 context.putAll(map); 176 } 177 178 @Override 179 public LogEvent toImmutable() { 180 return Log4jLogEvent.createMemento(this); 181 } 182 183 /** 184 * Set the body in the event. 185 * @param body The body to add to the event. 186 */ 187 @Override 188 public void setBody(final byte[] body) { 189 if (body == null || body.length == 0) { 190 super.setBody(new byte[0]); 191 return; 192 } 193 if (compress) { 194 final ByteArrayOutputStream baos = new ByteArrayOutputStream(); 195 try (GZIPOutputStream os = new GZIPOutputStream(baos)) { 196 os.write(body); 197 } catch (final IOException ioe) { 198 throw new LoggingException("Unable to compress message", ioe); 199 } 200 super.setBody(baos.toByteArray()); 201 } else { 202 super.setBody(body); 203 } 204 } 205 206 /** 207 * Get the Frequently Qualified Class Name. 208 * @return the FQCN String. 209 */ 210 @Override 211 public String getLoggerFqcn() { 212 return event.getLoggerFqcn(); 213 } 214 215 /** 216 * Returns the logging Level. 217 * @return the Level. 218 */ 219 @Override 220 public Level getLevel() { 221 return event.getLevel(); 222 } 223 224 /** 225 * Returns the logger name. 226 * @return the logger name. 227 */ 228 @Override 229 public String getLoggerName() { 230 return event.getLoggerName(); 231 } 232 233 /** 234 * Returns the StackTraceElement for the caller of the logging API. 235 * @return the StackTraceElement of the caller. 236 */ 237 @Override 238 public StackTraceElement getSource() { 239 return event.getSource(); 240 } 241 242 /** 243 * Returns the Message. 244 * @return the Message. 245 */ 246 @Override 247 public Message getMessage() { 248 return event.getMessage(); 249 } 250 251 /** 252 * Returns the Marker. 253 * @return the Marker. 254 */ 255 @Override 256 public Marker getMarker() { 257 return event.getMarker(); 258 } 259 260 /** 261 * Returns the ID of the Thread. 262 * @return the ID of the Thread. 263 */ 264 @Override 265 public long getThreadId() { 266 return event.getThreadId(); 267 } 268 269 /** 270 * Returns the priority of the Thread. 271 * @return the priority of the Thread. 272 */ 273 @Override 274 public int getThreadPriority() { 275 return event.getThreadPriority(); 276 } 277 278 /** 279 * Returns the name of the Thread. 280 * @return the name of the Thread. 281 */ 282 @Override 283 public String getThreadName() { 284 return event.getThreadName(); 285 } 286 287 /** 288 * Returns the event timestamp. 289 * @return the event timestamp. 290 */ 291 @Override 292 public long getTimeMillis() { 293 return event.getTimeMillis(); 294 } 295 296 /** 297 * Returns the value of the running Java Virtual Machine's high-resolution time source when this event was created, 298 * or a dummy value if it is known that this value will not be used downstream. 299 * @return the event nanosecond timestamp. 300 */ 301 @Override 302 public long getNanoTime() { 303 return event.getNanoTime(); 304 } 305 306 /** 307 * Returns the Throwable associated with the event, if any. 308 * @return the Throwable. 309 */ 310 @Override 311 public Throwable getThrown() { 312 return event.getThrown(); 313 } 314 315 /** 316 * Returns the Throwable associated with the event, if any. 317 * @return the Throwable. 318 */ 319 @Override 320 public ThrowableProxy getThrownProxy() { 321 return event.getThrownProxy(); 322 } 323 324 /** 325 * Returns a copy of the context Map. 326 * @return a copy of the context Map. 327 */ 328 @Override 329 public Map<String, String> getContextMap() { 330 return contextMap; 331 } 332 333 /** 334 * Returns the context data of the {@code LogEvent} that this {@code FlumeEvent} was constructed with. 335 * @return the context data of the {@code LogEvent} that this {@code FlumeEvent} was constructed with. 336 */ 337 @Override 338 public ReadOnlyStringMap getContextData() { 339 return event.getContextData(); 340 } 341 342 /** 343 * Returns a copy of the context stack. 344 * @return a copy of the context stack. 345 */ 346 @Override 347 public ThreadContext.ContextStack getContextStack() { 348 return event.getContextStack(); 349 } 350 351 @Override 352 public boolean isIncludeLocation() { 353 return event.isIncludeLocation(); 354 } 355 356 @Override 357 public void setIncludeLocation(final boolean includeLocation) { 358 event.setIncludeLocation(includeLocation); 359 } 360 361 @Override 362 public boolean isEndOfBatch() { 363 return event.isEndOfBatch(); 364 } 365 366 @Override 367 public void setEndOfBatch(final boolean endOfBatch) { 368 event.setEndOfBatch(endOfBatch); 369 } 370 371}