Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1014: Add addMdcAsHeaders into appenders

Browse Source 
Fixes https://github.com/spring-projects/spring-amqp/issues/1014

**Cherry-pick to 2.1.x, 2.0.x & 1.7.x**

GH-1014 minor changes

GH-1014 renamed property to addMdcAsHeaders

GH-1014 added addMdcAsHeaders into documentation

GH-1014 added addMdcAsHeaders into logback appender. added integration test

GH-1014 updated documentation

GH-1014 minor fix

GH-1014 updated documentation

GH-1014 minor fix

GH-1014 removed this prefix

* Made addMdcAsHeaders true by default

* Polishing

# Conflicts:
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderIntegrationTests.java

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
#	src/reference/asciidoc/logging.adoc
#	src/reference/asciidoc/whats-new.adoc

# Conflicts:
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
#	spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j2/AmqpAppenderTests.java
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderIntegrationTests.java
#	spring-rabbit/src/test/resources/log4j2-amqp-appender.xml
#	spring-rabbit/src/test/resources/logback-test.xml
This commit is contained in:
Eugene Gusev
2019-06-07 11:14:17 +03:00
committed by Artem Bilan
parent 3fa628e4a6
commit 34e614a30a
6 changed files with 149 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java
View File

@@ -75,6 +75,7 @@ import com.rabbitmq.client.ConnectionFactory;
* @author Stephen Oakey
* @author Artem Bilan
* @author Nicolas Ristock
* @author Eugene Gusev
*
* @since 1.6
*/
@@ -156,7 +157,8 @@ public class AmqpAppender extends AbstractAppender {
@PluginAttribute("contentEncoding") String contentEncoding,
@PluginAttribute("clientConnectionProperties") String clientConnectionProperties,
@PluginAttribute("async") boolean async,
@PluginAttribute("charset") String charset) {
@PluginAttribute("charset") String charset,
@PluginAttribute(value = "addMdcAsHeaders", defaultBoolean = true) boolean addMdcAsHeaders) {
if (name == null) {
LOGGER.error("No name for AmqpAppender");
}
@@ -187,6 +189,7 @@ public class AmqpAppender extends AbstractAppender {
manager.clientConnectionProperties = clientConnectionProperties;
manager.charset = charset;
manager.async = async;
manager.addMdcAsHeaders = addMdcAsHeaders;
AmqpAppender appender = new AmqpAppender(name, filter, theLayout, ignoreExceptions, manager);
if (manager.activateOptions()) {
appender.startSenders();
@@ -235,7 +238,7 @@ public class AmqpAppender extends AbstractAppender {
return message;
}
private void sendEvent(final Event event, Map<?, ?> properties) {
protected void sendEvent(Event event, Map<?, ?> properties) {
LogEvent logEvent = event.getEvent();
String name = logEvent.getLoggerName();
Level level = logEvent.getLevel();
@@ -264,8 +267,10 @@ public class AmqpAppender extends AbstractAppender {
amqpProps.setTimestamp(tstamp.getTime());
// Copy properties in from MDC
for (Entry<?, ?> entry : properties.entrySet()) {
amqpProps.setHeader(entry.getKey().toString(), entry.getValue());
if (this.manager.addMdcAsHeaders) {
for (Entry<?, ?> entry : properties.entrySet()) {
amqpProps.setHeader(entry.getKey().toString(), entry.getValue());
}
}
if (logEvent.getSource() != null) {
amqpProps.setHeader(
@@ -275,6 +280,10 @@ public class AmqpAppender extends AbstractAppender {
logEvent.getSource().getLineNumber()));
}
doSend(event, logEvent, amqpProps);
}
protected void doSend(final Event event, LogEvent logEvent, MessageProperties amqpProps) {
StringBuilder msgBody;
String routingKey;
@@ -491,6 +500,11 @@ public class AmqpAppender extends AbstractAppender {
*/
private String charset = Charset.defaultCharset().name();
/**
* Whether or not add MDC properties into message headers. true by default for backward compatibility
*/
private boolean addMdcAsHeaders = true;
private boolean durable = true;
private MessageDeliveryMode deliveryMode = MessageDeliveryMode.PERSISTENT;

26
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java
View File

@@ -77,6 +77,7 @@ import ch.qos.logback.core.Layout;
* @author Artem Bilan
* @author Gary Russell
* @author Nicolas Ristock
* @author Eugene Gusev
*
* @since 1.4
*/
@@ -222,6 +223,11 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
*/
private String charset;
/**
* Whether or not add MDC properties into message headers. true by default for backward compatibility
*/
private boolean addMdcAsHeaders = true;
private boolean durable = true;
private MessageDeliveryMode deliveryMode = MessageDeliveryMode.PERSISTENT;
@@ -359,6 +365,14 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
this.maxSenderRetries = maxSenderRetries;
}
public boolean isAddMdcAsHeaders() {
return this.addMdcAsHeaders;
}
public void setAddMdcAsHeaders(boolean addMdcAsHeaders) {
this.addMdcAsHeaders = addMdcAsHeaders;
}
public boolean isDurable() {
return this.durable;
}
@@ -579,10 +593,12 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
amqpProps.setTimestamp(tstamp.getTime());
// Copy properties in from MDC
Map<String, String> props = event.getProperties();
Set<Entry<String, String>> entrySet = props.entrySet();
for (Entry<String, String> entry : entrySet) {
amqpProps.setHeader(entry.getKey(), entry.getValue());
if (AmqpAppender.this.addMdcAsHeaders) {
Map<String, String> props = event.getProperties();
Set<Entry<String, String>> entrySet = props.entrySet();
for (Entry<String, String> entry : entrySet) {
amqpProps.setHeader(entry.getKey(), entry.getValue());
}
}
String[] location = AmqpAppender.this.locationLayout.doLayout(logEvent).split("\\|");
if (!"?".equals(location[0])) {
@@ -629,6 +645,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
if (retries < AmqpAppender.this.maxSenderRetries) {
// Schedule a retry based on the number of times I've tried to re-send this
AmqpAppender.this.retryTimer.schedule(new TimerTask() {
@Override
public void run() {
AmqpAppender.this.events.add(event);
@@ -646,6 +663,7 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
Thread.currentThread().interrupt();
}
}
}
/**

10
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderConfiguration.java
View File

@@ -25,6 +25,7 @@ import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.context.annotation.Bean;
@@ -33,6 +34,7 @@ import org.springframework.context.annotation.Scope;
/**
* @author Jon Brisbin
* @author Artem Bilan
*/
@Configuration
public class AmqpAppenderConfiguration {
@@ -104,4 +106,12 @@ public class AmqpAppenderConfiguration {
public TestListener testListener(int count) {
return new TestListener(count);
}
@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
rabbitTemplate.setReceiveTimeout(10_000);
return rabbitTemplate;
}
}

47
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderIntegrationTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.amqp.rabbit.logback;
import static org.hamcrest.Matchers.hasEntry;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.startsWith;
@@ -37,6 +38,9 @@ import org.slf4j.MDC;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.log4j.AmqpAppenderConfiguration;
@@ -52,6 +56,7 @@ import ch.qos.logback.classic.Logger;
/**
* @author Artem Bilan
* @author Nicolas Ristock
* @author Eugene Gusev
*
* @since 1.4
*/
@@ -69,15 +74,23 @@ public class AmqpAppenderIntegrationTests {
@Autowired
private ApplicationContext applicationContext;
@Autowired
private RabbitTemplate template;
@Autowired
private Queue testQueue;
private SimpleMessageListenerContainer listenerContainer;
@Before
public void setUp() throws Exception {
listenerContainer = applicationContext.getBean(SimpleMessageListenerContainer.class);
public void setUp() {
this.listenerContainer = this.applicationContext.getBean(SimpleMessageListenerContainer.class);
MDC.clear();
}
@After
public void tearDown() {
MDC.clear();
listenerContainer.shutdown();
}
@@ -120,7 +133,9 @@ public class AmqpAppenderIntegrationTests {
assertNotNull(location);
assertThat(location, instanceOf(String.class));
assertThat((String) location,
startsWith("org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests.testAppenderWithProps()"));
startsWith("org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests" +
".testAppenderWithProps" +
"()"));
Object threadName = messageProperties.getHeaders().get("thread");
assertNotNull(threadName);
assertThat(threadName, instanceOf(String.class));
@@ -144,6 +159,32 @@ public class AmqpAppenderIntegrationTests {
assertEquals(0xbf, body[body.length - 3 - lineSeparatorExtraBytes] & 0xff);
}
@Test
public void testAddMdcAsHeaders() {
this.applicationContext.getBean(SingleConnectionFactory.class).createConnection().close();
Logger logWithMdc = (Logger) LoggerFactory.getLogger("withMdc");
Logger logWithoutMdc = (Logger) LoggerFactory.getLogger("withoutMdc");
MDC.put("mdc1", "test1");
MDC.put("mdc2", "test2");
logWithMdc.info("test message with MDC in headers");
Message received1 = this.template.receive(this.testQueue.getName());
assertNotNull(received1);
assertEquals("test message with MDC in headers", new String(received1.getBody()));
assertThat(received1.getMessageProperties().getHeaders(), hasEntry("mdc1", "test1"));
assertThat(received1.getMessageProperties().getHeaders(), hasEntry("mdc2", "test2"));
logWithoutMdc.info("test message without MDC in headers");
Message received2 = this.template.receive(this.testQueue.getName());
assertNotNull(received2);
assertEquals("test message without MDC in headers", new String(received2.getBody()));
assertThat(received1.getMessageProperties().getHeaders(), hasEntry("mdc1", "test1"));
assertThat(received1.getMessageProperties().getHeaders(), hasEntry("mdc2", "test2"));
}
public static class EnhancedAppender extends AmqpAppender {
private String foo;

42
spring-rabbit/src/test/resources/logback-test.xml
View File

@@ -25,10 +25,52 @@
<foo>bar</foo>
</appender>
<appender name="AMQPWithMdc" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
<layout>
<pattern>%m</pattern>
</layout>
<addresses>localhost:5672</addresses>
<abbreviation>36</abbreviation>
<includeCallerData>true</includeCallerData>
<applicationId>AmqpAppenderTest</applicationId>
<routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
<generateId>true</generateId>
<charset>UTF-8</charset>
<durable>false</durable>
<deliveryMode>NON_PERSISTENT</deliveryMode>
<declareExchange>false</declareExchange>
<addMdcAsHeaders>true</addMdcAsHeaders>
</appender>
<appender name="AMQPWithoutMdc" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
<layout>
<pattern>%m</pattern>
</layout>
<addresses>localhost:5672</addresses>
<abbreviation>36</abbreviation>
<includeCallerData>true</includeCallerData>
<applicationId>AmqpAppenderTest</applicationId>
<routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
<generateId>true</generateId>
<charset>UTF-8</charset>
<durable>false</durable>
<deliveryMode>NON_PERSISTENT</deliveryMode>
<declareExchange>false</declareExchange>
<addMdcAsHeaders>false</addMdcAsHeaders>
</appender>
<logger name="org.springframework.amqp.rabbit.logback" level="DEBUG" additivity="false">
<appender-ref ref="AMQP"/>
</logger>
<logger name="withMdc" level="DEBUG" additivity="false">
<appender-ref ref="AMQPWithMdc"/>
</logger>
<logger name="withoutMdc" level="DEBUG" additivity="false">
<appender-ref ref="AMQPWithoutMdc"/>
</logger>
<root level="INFO">
<appender-ref ref="CONSOLE"/>
</root>

16
src/reference/asciidoc/logging.adoc
View File

@@ -7,8 +7,7 @@ The framework provides logging appenders for several popular logging subsystems:
- logback (since Spring AMQP _version 1.4_)
- log4j2 (since Spring AMQP _version 1.6_)
The appenders are configured using the normal mechanisms for the logging subsystem, available properties are specified
in the following sections.
The appenders are configured by using the normal mechanisms for the logging subsystem, available properties are specified in the following sections.
==== Common properties
@@ -109,6 +108,15 @@ If the charset is unsupported on the current platform, we fall back to using the
| null
| A comma-delimited list of `key:value` pairs for custom client properties to the RabbitMQ connection.
| addMdcAsHeaders
| true
| MDC properties were always added into RabbitMQ message headers until this property was introduced.
It can lead to issues for big MDC as while RabbitMQ has limited buffer size for all headers and this buffer is pretty small.
This property was introduced to avoid issues in cases of big MDC.
By default this value set to `true` for backward compatibility.
The `false` turns off serialization MDC into headers.
Please note, the `JsonLayout` adds MDC into the message by default.
|===
==== Log4j Appender
@@ -144,7 +152,8 @@ NOTE: This appender is deprecated and will be removed in _version 2.0_.
applicationId="myAppId" routingKeyPattern="%X{applicationId}.%c.%p"
contentType="text/plain" contentEncoding="UTF-8" generateId="true" deliveryMode="NON_PERSISTENT"
charset="UTF-8"
senderPoolSize="3" maxSenderRetries="5">
senderPoolSize="3" maxSenderRetries="5"
addMdcAsHeaders="false">
</RabbitMQ>
</Appenders>
----
@@ -179,6 +188,7 @@ If async publishing is used with the `ReusableLogEventFactory`, events will have
<durable>false</durable>
<deliveryMode>NON_PERSISTENT</deliveryMode>
<declareExchange>true</declareExchange>
<addMdcAsHeaders>false</addMdcAsHeaders>
</appender>
----