From e350c58a46cea5d2f306197dd55f913259a8c85f Mon Sep 17 00:00:00 2001 From: Pat Fox Date: Sat, 6 Jan 2018 14:06:39 +0100 Subject: [PATCH] ARTEMIS-1623 ActiveMQServerPlugin impl for logging various broker events Similar concept to the ActiveMQ 5.x loggingBrokerPlugin --- .../impl/LoggingActiveMQServerPlugin.java | 682 ++++++++++++++++++ .../config/impl/ConfigurationImplTest.java | 4 + docs/user-manual/en/broker-plugins.md | 76 ++ .../LoggingActiveMQServerPluginAMQPTest.java | 178 +++++ ...ggingActiveMQServerPluginOpenWireTest.java | 190 +++++ .../LoggingActiveMQServerPluginTest.java | 532 ++++++++++++++ .../plugin/XmlConfigPluginTest.java | 76 +- .../resources/broker-logging-plugin-wrong.xml | 51 ++ .../test/resources/broker-logging-plugin.xml | 56 ++ 9 files changed, 1837 insertions(+), 8 deletions(-) create mode 100644 artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/impl/LoggingActiveMQServerPlugin.java create mode 100644 tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginAMQPTest.java create mode 100644 tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginOpenWireTest.java create mode 100644 tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginTest.java create mode 100644 tests/integration-tests/src/test/resources/broker-logging-plugin-wrong.xml create mode 100644 tests/integration-tests/src/test/resources/broker-logging-plugin.xml diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/impl/LoggingActiveMQServerPlugin.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/impl/LoggingActiveMQServerPlugin.java new file mode 100644 index 0000000000..02070ca9b8 --- /dev/null +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/plugin/impl/LoggingActiveMQServerPlugin.java @@ -0,0 +1,682 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.activemq.artemis.core.server.plugin.impl; + +import java.io.Serializable; +import java.util.Map; + +import org.apache.activemq.artemis.api.core.ActiveMQException; +import org.apache.activemq.artemis.api.core.Message; +import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.api.core.SimpleString; +import org.apache.activemq.artemis.core.config.BridgeConfiguration; +import org.apache.activemq.artemis.core.persistence.OperationContext; +import org.apache.activemq.artemis.core.postoffice.QueueBinding; +import org.apache.activemq.artemis.core.postoffice.RoutingStatus; +import org.apache.activemq.artemis.core.security.SecurityAuth; +import org.apache.activemq.artemis.core.server.MessageReference; +import org.apache.activemq.artemis.core.server.Queue; +import org.apache.activemq.artemis.core.server.QueueConfig; +import org.apache.activemq.artemis.core.server.RoutingContext; +import org.apache.activemq.artemis.core.server.ServerConsumer; +import org.apache.activemq.artemis.core.server.ServerSession; +import org.apache.activemq.artemis.core.server.cluster.Bridge; +import org.apache.activemq.artemis.core.server.impl.AckReason; +import org.apache.activemq.artemis.core.server.plugin.ActiveMQServerPlugin; +import org.apache.activemq.artemis.core.transaction.Transaction; +import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection; +import org.apache.activemq.artemis.spi.core.protocol.SessionCallback; +import org.apache.activemq.artemis.utils.critical.CriticalComponent; +import org.jboss.logging.Logger; + +/** + * plugin to log various events within the broker, configured with the following booleans + * + * LOG_CONNECTION_EVENTS - connections creation/destroy + * LOG_SESSION_EVENTS - sessions creation/close + * LOG_CONSUMER_EVENTS - consumers creation/close + * LOG_DELIVERING_EVENTS - messages delivered to consumer, acked by consumer + * LOG_SENDING_EVENTS - messaged is sent, message is routed + * LOG_INTERNAL_EVENTS - critical failures, bridge deployments, queue creation/destroyed, message expired + */ + +public class LoggingActiveMQServerPlugin implements ActiveMQServerPlugin, Serializable { + + private static final Logger logger = Logger.getLogger(LoggingActiveMQServerPlugin.class); + + private static final long serialVersionUID = 1L; + + public static final String LOG_ALL_EVENTS = "LOG_ALL_EVENTS"; + public static final String LOG_CONNECTION_EVENTS = "LOG_CONNECTION_EVENTS"; + public static final String LOG_SESSION_EVENTS = "LOG_SESSION_EVENTS"; + public static final String LOG_CONSUMER_EVENTS = "LOG_CONSUMER_EVENTS"; + public static final String LOG_DELIVERING_EVENTS = "LOG_DELIVERING_EVENTS"; + public static final String LOG_SENDING_EVENTS = "LOG_SENDING_EVENTS"; + public static final String LOG_INTERNAL_EVENTS = "LOG_INTERNAL_EVENTS"; + + public static final String UNAVAILABLE = "UNAVAILABLE"; + + private boolean logAll = false; + private boolean logConnectionEvents = false; + private boolean logSessionEvents = false; + private boolean logConsumerEvents = false; + private boolean logDeliveringEvents = false; + private boolean logSendingEvents = false; + private boolean logInternalEvents = false; + + public boolean isLogAll() { + return logAll; + } + + public boolean isLogConnectionEvents() { + return logConnectionEvents; + } + + public boolean isLogSessionEvents() { + return logSessionEvents; + } + + public boolean isLogConsumerEvents() { + return logConsumerEvents; + } + + public boolean isLogDeliveringEvents() { + return logDeliveringEvents; + } + + public boolean isLogSendingEvents() { + return logSendingEvents; + } + + public boolean isLogInternalEvents() { + return logInternalEvents; + } + + /** + * used to pass configured properties to Plugin + * + * @param properties + */ + @Override + public void init(Map properties) { + + logAll = Boolean.parseBoolean(properties.getOrDefault(LOG_ALL_EVENTS, "false")); + logConnectionEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_CONNECTION_EVENTS, "false")); + logSessionEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_SESSION_EVENTS, "false")); + logConsumerEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_CONSUMER_EVENTS, "false")); + logDeliveringEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_DELIVERING_EVENTS, "false")); + logSendingEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_SENDING_EVENTS, "false")); + logInternalEvents = Boolean.parseBoolean(properties.getOrDefault(LOG_INTERNAL_EVENTS, "false")); + + if (logger.isDebugEnabled()) { + dumpConfiguration(); + } + + } + + /** + * A connection has been created. + * + * @param connection The newly created connection + * @throws ActiveMQException + */ + @Override + public void afterCreateConnection(RemotingConnection connection) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logConnectionEvents)) { + logger.infof("created connection: %s", connection); + } + } + + /** + * A connection has been destroyed. + * + * @param connection + * @throws ActiveMQException + */ + @Override + public void afterDestroyConnection(RemotingConnection connection) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logConnectionEvents)) { + logger.infof("destroyed connection: %s", connection); + } + } + + /** + * Before a session is created. + * + * @param name + * @param username + * @param minLargeMessageSize + * @param connection + * @param autoCommitSends + * @param autoCommitAcks + * @param preAcknowledge + * @param xa + * @param publicAddress + * @param callback + * @param autoCreateQueues + * @param context + * @param prefixes + * @throws ActiveMQException + */ + @Override + public void beforeCreateSession(String name, + String username, + int minLargeMessageSize, + RemotingConnection connection, + boolean autoCommitSends, + boolean autoCommitAcks, + boolean preAcknowledge, + boolean xa, + String publicAddress, + SessionCallback callback, + boolean autoCreateQueues, + OperationContext context, + Map prefixes) throws ActiveMQException { + + if (logger.isDebugEnabled() && (logAll || logSessionEvents)) { + logger.debugf("beforeCreateSession called with name: %s , username: %s, minLargeMessageSize: %s , connection: %s" + + ", autoCommitSends: %s, autoCommitAcks: %s, preAcknowledge: %s, xa: %s, publicAddress: %s" + + ", , autoCreateQueues: %s, context: %s ", name, username, minLargeMessageSize, connection, + autoCommitSends, autoCommitAcks, preAcknowledge, xa, publicAddress, autoCreateQueues, context); + } + + } + + /** + * After a session has been created. + * + * @param session The newly created session + * @throws ActiveMQException + */ + @Override + public void afterCreateSession(ServerSession session) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logSessionEvents)) { + logger.infof("created session name: %s, session connectionID: %s", + (session == null ? UNAVAILABLE : session.getName()), + (session == null ? UNAVAILABLE : session.getConnectionID())); + } + + } + + /** + * Before a session is closed + * + * @param session + * @param failed + * @throws ActiveMQException + */ + @Override + public void beforeCloseSession(ServerSession session, boolean failed) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logSessionEvents)) { + logger.debugf("beforeCloseSession called with session name : %s, session: %s, failed: %s", + (session == null ? UNAVAILABLE : session.getName()), session, failed); + } + } + + /** + * After a session is closed + * + * @param session + * @param failed + * @throws ActiveMQException + */ + @Override + public void afterCloseSession(ServerSession session, boolean failed) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logSessionEvents)) { + logger.infof("closed session with session name: %s, failed: %s", + (session == null ? UNAVAILABLE : session.getName()), failed); + } + } + + /** + * Before session metadata is added to the session + * + * @param session + * @param key + * @param data + * @throws ActiveMQException + */ + @Override + public void beforeSessionMetadataAdded(ServerSession session, String key, String data) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logSessionEvents)) { + logger.debugf("beforeSessionMetadataAdded called with session name: %s , session: %s, key: %s, data: %s", + (session == null ? UNAVAILABLE : session.getName()), session, key, data); + + } + } + + /** + * After session metadata is added to the session + * + * @param session + * @param key + * @param data + * @throws ActiveMQException + */ + @Override + public void afterSessionMetadataAdded(ServerSession session, String key, String data) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logSessionEvents)) { + logger.debugf("added session metadata for session name : %s, session: %s, key: %s, data: %s", + (session == null ? UNAVAILABLE : session.getName()), session, key, data); + + } else if (logger.isInfoEnabled() && (logAll || logSessionEvents)) { + logger.infof("added session metadata for session name : %s, key: %s, data: %s", + (session == null ? UNAVAILABLE : session.getName()), key, data); + + } + } + + /** + * Before a consumer is created + * + * @param consumerID + * @param queueBinding + * @param filterString + * @param browseOnly + * @param supportLargeMessage + * @throws ActiveMQException + */ + @Override + public void beforeCreateConsumer(long consumerID, + QueueBinding queueBinding, + SimpleString filterString, + boolean browseOnly, + boolean supportLargeMessage) throws ActiveMQException { + + if (logger.isDebugEnabled() && (logAll || logConsumerEvents)) { + logger.debugf("beforeCreateConsumer called with ConsumerID: %s, QueueBinding: %s, filterString: %s," + + " browseOnly: %s, supportLargeMessage: %s", consumerID, queueBinding, filterString, browseOnly, + supportLargeMessage); + } + + } + + /** + * After a consumer has been created + * + * @param consumer the created consumer + * @throws ActiveMQException + */ + @Override + public void afterCreateConsumer(ServerConsumer consumer) throws ActiveMQException { + + if (logger.isInfoEnabled() && (logAll || logConsumerEvents)) { + logger.infof("created consumer with ID: %s, with session name: %s", + (consumer == null ? UNAVAILABLE : consumer.getID()), + (consumer == null ? UNAVAILABLE : consumer.getSessionID())); + } + + } + + /** + * Before a consumer is closed + * + * @param consumer + * @param failed + * @throws ActiveMQException + */ + @Override + public void beforeCloseConsumer(ServerConsumer consumer, boolean failed) throws ActiveMQException { + + if (logger.isDebugEnabled() && (logAll || logConsumerEvents)) { + logger.debugf("beforeCloseConsumer called with consumer: %s, consumer sessionID: %s, failed: %s", + consumer, (consumer == null ? UNAVAILABLE : consumer.getSessionID()), failed); + } + } + + /** + * After a consumer is closed + * + * @param consumer + * @param failed + * @throws ActiveMQException + */ + @Override + public void afterCloseConsumer(ServerConsumer consumer, boolean failed) throws ActiveMQException { + + if (logger.isInfoEnabled() && (logAll || logConsumerEvents)) { + logger.infof("closed consumer ID: %s, with consumer Session: %s, failed: %s", + (consumer == null ? UNAVAILABLE : consumer.getID()), + (consumer == null ? UNAVAILABLE : consumer.getSessionID()), failed); + } + + } + + /** + * Before a queue is created + * + * @param queueConfig + * @throws ActiveMQException + */ + @Override + public void beforeCreateQueue(QueueConfig queueConfig) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logInternalEvents)) { + logger.debugf("beforeCreateQueue called with queueConfig: %s", queueConfig); + } + + } + + /** + * After a queue has been created + * + * @param queue The newly created queue + * @throws ActiveMQException + */ + @Override + public void afterCreateQueue(Queue queue) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logInternalEvents)) { + logger.infof("created queue: %s", queue); + } + } + + /** + * Before a queue is destroyed + * + * @param queueName + * @param session + * @param checkConsumerCount + * @param removeConsumers + * @param autoDeleteAddress + * @throws ActiveMQException + */ + @Override + public void beforeDestroyQueue(SimpleString queueName, + final SecurityAuth session, + boolean checkConsumerCount, + boolean removeConsumers, + boolean autoDeleteAddress) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logInternalEvents)) { + + logger.debugf("beforeDestroyQueue called with queueName: %s, session: %s, checkConsumerCount: %s," + + " removeConsumers: %s, autoDeleteAddress: %s", queueName, session, checkConsumerCount, + removeConsumers, autoDeleteAddress); + } + + } + + /** + * After a queue has been destroyed + * + * @param queue + * @param address + * @param session + * @param checkConsumerCount + * @param removeConsumers + * @param autoDeleteAddress + * @throws ActiveMQException + */ + @Override + public void afterDestroyQueue(Queue queue, + SimpleString address, + final SecurityAuth session, + boolean checkConsumerCount, + boolean removeConsumers, + boolean autoDeleteAddress) throws ActiveMQException { + + if (logger.isInfoEnabled() && (logAll || logInternalEvents)) { + + logger.infof("destroyed queue: %s, with args address: %s, session: %s, checkConsumerCount: %s," + + " removeConsumers: %s, autoDeleteAddress: %s", queue, address, session, checkConsumerCount, + removeConsumers, autoDeleteAddress); + } + + } + + /** + * Before a message is sent + * + * @param session the session that sends the message + * @param tx + * @param message + * @param direct + * @param noAutoCreateQueue + * @throws ActiveMQException + */ + @Override + public void beforeSend(ServerSession session, + Transaction tx, + Message message, + boolean direct, + boolean noAutoCreateQueue) throws ActiveMQException { + + if (logger.isDebugEnabled() && (logAll || logSendingEvents)) { + logger.debugf("beforeSend called with message: %s, tx: %s, session: %s, direct: %s, noAutoCreateQueue: %s", + message, tx, session, direct, noAutoCreateQueue); + } + } + + /** + * After a message is sent + * + * @param session the session that sends the message + * @param tx + * @param message + * @param direct + * @param noAutoCreateQueue + * @param result + * @throws ActiveMQException + */ + @Override + public void afterSend(ServerSession session, + Transaction tx, + Message message, + boolean direct, + boolean noAutoCreateQueue, + RoutingStatus result) throws ActiveMQException { + + if (logger.isDebugEnabled() && (logAll || logSendingEvents)) { + logger.debugf("sent message: %s, session name : %s, session connectionID: %s, with tx: %s, session: %s," + + " direct: %s, noAutoCreateQueue: %s, result: %s", message, + (session == null ? UNAVAILABLE : session.getName()), + (session == null ? UNAVAILABLE : session.getConnectionID()), + tx, session, direct, noAutoCreateQueue, result); + } else if (logger.isInfoEnabled() && (logAll || logSendingEvents)) { + logger.infof("sent message with ID: %s, session name: %s, session connectionID: %s, result: %s", + (message == null ? UNAVAILABLE : message.getMessageID()), + (session == null ? UNAVAILABLE : session.getName()), + (session == null ? UNAVAILABLE : session.getConnectionID()), result); + } + } + + /** + * Before a message is routed + * + * @param message + * @param context + * @param direct + * @param rejectDuplicates + * @throws ActiveMQException + */ + @Override + public void beforeMessageRoute(Message message, + RoutingContext context, + boolean direct, + boolean rejectDuplicates) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logSendingEvents)) { + logger.debugf("beforeMessageRoute called with message: %s, context: %s, direct: %s, rejectDuplicates: %s", + message, context, direct, rejectDuplicates); + } + } + + /** + * After a message is routed + * + * @param message + * @param context + * @param direct + * @param rejectDuplicates + * @param result + * @throws ActiveMQException + */ + @Override + public void afterMessageRoute(Message message, + RoutingContext context, + boolean direct, + boolean rejectDuplicates, + RoutingStatus result) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logSendingEvents)) { + logger.debugf("routed message: %s, with context: %s, direct: %s, rejectDuplicates: %s, result: %s", + message, context, direct, rejectDuplicates, result); + + } else if (logger.isInfoEnabled() && (logAll || logSendingEvents)) { + logger.infof("routed message with ID: %s, result: %s", + (message == null ? UNAVAILABLE : message.getMessageID()), result); + + } + } + + /** + * Before a message is delivered to a client consumer + * + * @param consumer the consumer the message will be delivered to + * @param reference message reference + * @throws ActiveMQException + */ + @Override + public void beforeDeliver(ServerConsumer consumer, MessageReference reference) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logDeliveringEvents)) { + logger.debugf("beforeDeliver called with consumer: %s, reference: %s", consumer, reference); + } + } + + /** + * After a message is delivered to a client consumer + * + * @param consumer the consumer the message was delivered to + * @param reference message reference + * @throws ActiveMQException + */ + @Override + public void afterDeliver(ServerConsumer consumer, MessageReference reference) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logDeliveringEvents)) { + Message message = (reference == null ? null : reference.getMessage()); + + if (consumer != null) { + logger.debugf("delivered message with message ID: %s to consumer on address: %s, queue: %s, consumer sessionID: %s," + + " consumerID: %s, full message: %s, full consumer: %s", (message == null ? UNAVAILABLE : message.getMessageID()), + consumer.getQueueAddress(), consumer.getQueueName(), consumer.getSessionID(), consumer.getID(), reference, consumer); + } else { + logger.debugf("delivered message with message ID: %s, consumer info UNAVAILABLE", (message == null ? UNAVAILABLE : message.getMessageID())); + } + + } else if (logger.isInfoEnabled() && (logAll || logDeliveringEvents)) { + Message message = (reference == null ? null : reference.getMessage()); + + if (consumer != null) { + logger.infof("delivered message with message ID: %s, to consumer on address: %s, queue: %s, consumer sessionID: %s, consumerID: %s", + (message == null ? UNAVAILABLE : message.getMessageID()), consumer.getQueueAddress(), consumer.getQueueName(), + consumer.getSessionID(), consumer.getID()); + } else { + logger.infof("delivered message with message ID: %s, consumer info UNAVAILABLE", (message == null ? UNAVAILABLE : message.getMessageID())); + } + + } + } + + /** + * A message has been expired + * + * @param message The expired message + * @param messageExpiryAddress The message expiry address if exists + * @throws ActiveMQException + */ + @Override + public void messageExpired(MessageReference message, SimpleString messageExpiryAddress) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logInternalEvents)) { + logger.infof("expired message: %s, messageExpiryAddress: %s", message, messageExpiryAddress); + } + } + + /** + * A message has been acknowledged + * + * @param ref The acked message + * @param reason The ack reason + * @throws ActiveMQException + */ + @Override + public void messageAcknowledged(MessageReference ref, AckReason reason) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logDeliveringEvents)) { + logger.debugf("acknowledged message: %s, with ackReason: %s", ref, reason); + } else if (logger.isInfoEnabled() && (logAll || logDeliveringEvents)) { + Message message = (ref == null ? null : ref.getMessage()); + Queue queue = (ref == null ? null : ref.getQueue()); + + logger.infof("acknowledged message ID: %s, with messageRef consumerID: %s, messageRef QueueName: %s, with ackReason: %s", + (message == null ? UNAVAILABLE : message.getMessageID()), (ref == null ? UNAVAILABLE : ref.getConsumerId()), + (queue == null ? UNAVAILABLE : queue.getName()), reason); + } + } + + /** + * Before a bridge is deployed + * + * @param config The bridge configuration + * @throws ActiveMQException + */ + @Override + public void beforeDeployBridge(BridgeConfiguration config) throws ActiveMQException { + if (logger.isDebugEnabled() && (logAll || logInternalEvents)) { + logger.debugf("beforeDeployBridge called with bridgeConfiguration: %s", config); + } + } + + /** + * After a bridge has been deployed + * + * @param bridge The newly deployed bridge + * @throws ActiveMQException + */ + @Override + public void afterDeployBridge(Bridge bridge) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logInternalEvents)) { + logger.infof("deployed bridge: %s", bridge); + } + } + + /** + * A Critical failure has been detected. + * This will be called before the broker is stopped + * + * @param components + * @throws ActiveMQException + */ + @Override + public void criticalFailure(CriticalComponent components) throws ActiveMQException { + if (logger.isInfoEnabled() && (logAll || logInternalEvents)) { + logger.infof("criticalFailure called with criticalComponent: %s", components); + } + } + + /** + * dump the configuration of the logging Plugin + */ + private void dumpConfiguration() { + if (logger.isDebugEnabled()) { + logger.debug("LoggingPlugin logAll=" + logAll); + logger.debug("LoggingPlugin logConnectionEvents=" + logConnectionEvents); + logger.debug("LoggingPlugin logSessionEvents=" + logSessionEvents); + logger.debug("LoggingPlugin logConsumerEvents=" + logConsumerEvents); + logger.debug("LoggingPlugin logSendingEvents=" + logSendingEvents); + logger.debug("LoggingPlugin logDeliveringEvents=" + logDeliveringEvents); + logger.debug("LoggingPlugin logInternalEvents=" + logInternalEvents); + } + + } + +} diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java index 5e9a95a122..847c0bf8fc 100644 --- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java +++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java @@ -27,6 +27,7 @@ import org.apache.activemq.artemis.api.core.SimpleString; import org.apache.activemq.artemis.core.config.Configuration; import org.apache.activemq.artemis.core.config.ha.LiveOnlyPolicyConfiguration; import org.apache.activemq.artemis.core.server.JournalType; +import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin; import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.apache.activemq.artemis.utils.RandomUtil; import org.junit.Assert; @@ -469,6 +470,9 @@ public class ConfigurationImplTest extends ActiveMQTestBase { conf.setClusterPassword(s); Assert.assertEquals(s, conf.getClusterPassword()); + conf.registerBrokerPlugin(new LoggingActiveMQServerPlugin()); + Assert.assertEquals("ensure one plugin registered", 1, conf.getBrokerPlugins().size()); + // This will use serialization to perform a deep copy of the object Configuration conf2 = conf.copy(); diff --git a/docs/user-manual/en/broker-plugins.md b/docs/user-manual/en/broker-plugins.md index c3c1f0e176..95bb6df1f0 100644 --- a/docs/user-manual/en/broker-plugins.md +++ b/docs/user-manual/en/broker-plugins.md @@ -53,4 +53,80 @@ Configuration config = new ConfigurationImpl(); config.registerBrokerPlugin(new UserPlugin()); ``` +## Using the LoggingActiveMQServerPlugin + +The LoggingActiveMQServerPlugin logs specific broker events. + +You can select which events are logged by setting the following configuration properties to `true`. + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
PropertyProperty Description
LOG_CONNECTION_EVENTSLog info when a Connection is created/destroy. Default `false`.
LOG_SESSION_EVENTSLog info when a Session is created/closed. Default `false`.
LOG_CONSUMER_EVENTSLogs info when a Consumer is created/closed. Default `false`.
LOG_DELIVERING_EVENTSLogs info when message is delivered to a consumer and when a message is acknowledged by a consumer. + Default `false`
LOG_SENDING_EVENTSLogs info when a message has been sent to an address and when a message has been routed within the broker. + Default `false`
LOG_INTERNAL_EVENTSLogs info when a queue created/destroyed, when a message is expired, when a bridge is deployed and when a critical + failure occurs. Default `false`
LOG_ALL_EVENTSLogs info for all the above events. Default `false`
+ +By default the LoggingActiveMQServerPlugin wil not log any information. The logging is activated by setting one (or a selection) +of the above configuration properties to `true`. + +To configure the plugin, you can add the following configuration to the broker. In the example below both LOG_DELIVERING_EVENTS +and LOG_SENDING_EVENTS will be logged by the broker. + +```xml + + +... + + + + + + +... + + +``` + +The LoggingActiveMQServerPlugin logs information at Log Level `INFO`. By setting the Logger +"org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin" to `DEBUG` more information for each +event will be logged. diff --git a/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginAMQPTest.java b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginAMQPTest.java new file mode 100644 index 0000000000..aecc81aaab --- /dev/null +++ b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginAMQPTest.java @@ -0,0 +1,178 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.activemq.artemis.tests.extras.byteman; + +import javax.jms.Connection; +import javax.jms.JMSException; + +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin; +import org.apache.qpid.jms.JmsConnectionFactory; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMRules; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; +import org.junit.Test; +import org.junit.runner.RunWith; + +/** + * NOTE: this test should be run at log level INFO + * + * This test checks the LoggingActiveMQServerPlugin is logging expected data with specific property configurations + * when client using AMQP + */ +@RunWith(BMUnitRunner.class) +public class LoggingActiveMQServerPluginAMQPTest extends LoggingActiveMQServerPluginTest { + + /** + * Aim: test all events are logged when plugin configured with + * LOG_ALL_EVENTS + * + * Overridden as behaviour slightly different for AMQP - consumer closed plugin point called twice + * + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test logAll EVENT", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogAll() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_ALL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "log ALL Message_1", 0); + sendAndReceive(true, true, "log ALL Message_2", 0); + + Thread.sleep(500); + + assertEquals("created connections", 2, createdConnectionLogs.size()); + assertEquals("destroyed connections", 2, destroyedConnectionLogs.size()); + assertEquals("created consumer", 2, createdConsumerLogs.size()); + assertEquals("closed consumer", 4, closedConsumerLogs.size()); + assertEquals("delivered message", 2, deliveredLogs.size()); + assertEquals("acked message", 2, ackedLogs.size()); + assertEquals("sending message", 2, sentLogs.size()); + assertEquals("routing message", 2, routedLogs.size()); + assertEquals("queue created", 2, createdQueueLogs.size()); + assertEquals("queue destroyed", 2, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test the consumer create/close events are logged when plugin configured with + * LOG_CONSUMER_EVENTS + * + * Overridden as behaviour slightly different for AMQP - consumer closed plugin point seems to be called twice + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test LOG_CONSUMER_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogConsumerEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_CONSUMER_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "txtMessage", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 1, createdConsumerLogs.size()); + assertEquals("closed consumer", 2, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 0, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test the session create/close events are logged when plugin configured with + * LOG_SESSION_EVENTS + * + * Overriden as addedMetaData does not seem to be invoked for AMQP + * + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test LOG_SESSION_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + + public void testLogSessionEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_SESSION_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(false,false,"test_message",0); + Thread.sleep(500); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 2, createdSessionLogs.size()); + assertEquals("closed sessions", 2, closedSessionLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + @Override + protected Connection createActiveMQConnection() throws JMSException { + JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:61616"); + return factory.createConnection(); + } +} \ No newline at end of file diff --git a/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginOpenWireTest.java b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginOpenWireTest.java new file mode 100644 index 0000000000..9104e8e483 --- /dev/null +++ b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginOpenWireTest.java @@ -0,0 +1,190 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.activemq.artemis.tests.extras.byteman; + +import javax.jms.Connection; +import javax.jms.JMSException; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMRules; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; +import org.junit.Test; +import org.junit.runner.RunWith; + +/** + * + * NOTE: this test should be run at log level INFO + * + * This test checks the LoggingActiveMQServerPlugin is logging expected data with specific property configurations + * when client using OpenWire + * + */ +@RunWith(BMUnitRunner.class) +public class LoggingActiveMQServerPluginOpenWireTest extends LoggingActiveMQServerPluginTest { + + /** + * Aim: test queue creation are logged when plugin configured with + * LOG_INTERNAL_EVENTS + * + * Overiden as Openwire does not seem to destroy the queue. + * + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test queue creation log", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testQueueCreationLog() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_INTERNAL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(false, true, "NO_MESSAGE", 0); + Thread.sleep(500); + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 1, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test the session create/close events are logged when plugin configured with + * LOG_SESSION_EVENTS + * + * Overriden as + * - ceated/closed sessions is invoke 3 times for openwire + * - no meta data added to the session . + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test LOG_SESSION_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + + public void testLogSessionEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_SESSION_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(false,false,"test_message",0); + Thread.sleep(500); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 3, createdSessionLogs.size()); + assertEquals("closed sessions", 3, closedSessionLogs.size()); + assertEquals("added Metadata logs", 0, addedSessionMetaData.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test all events are logged when plugin configured with + * LOG_ALL_EVENTS + * + * Overridden as behaviour slightly different for queue create/destroy with Openwire + * + * @throws Exception + */ + @Override + @Test + @BMRules(rules = {@BMRule(name = "test logAll EVENT", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogAll() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_ALL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "log ALL Message_1", 0); + sendAndReceive(true, true, "log ALL Message_2", 0); + + Thread.sleep(500); + + for (String log : unexpectedLogs) { + System.out.println(" others events logged >>>>" + log); + } + + assertEquals("created connections", 2, createdConnectionLogs.size()); + assertEquals("destroyed connections", 2, destroyedConnectionLogs.size()); + //assertEquals("created sessions", 0, createdSessionLogs.size()); + //assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 2, createdConsumerLogs.size()); + assertEquals("closed consumer", 2, closedConsumerLogs.size()); + assertEquals("delivered message", 2, deliveredLogs.size()); + assertEquals("acked message", 2, ackedLogs.size()); + assertEquals("sending message", 2, sentLogs.size()); + assertEquals("routing message", 2, routedLogs.size()); + assertEquals("queue created", 1, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + @Override + protected Connection createActiveMQConnection() throws JMSException { + ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616?jms.watchTopicAdvisories=false"); + return factory.createConnection(); + } +} \ No newline at end of file diff --git a/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginTest.java b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginTest.java new file mode 100644 index 0000000000..04e0cca2c6 --- /dev/null +++ b/tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/LoggingActiveMQServerPluginTest.java @@ -0,0 +1,532 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + *

+ * http://www.apache.org/licenses/LICENSE-2.0 + *

+ * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.activemq.artemis.tests.extras.byteman; + +import javax.jms.Connection; +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.MessageProducer; +import javax.jms.Queue; +import javax.jms.Session; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import org.apache.activemq.artemis.core.config.Configuration; +import org.apache.activemq.artemis.core.server.ActiveMQServer; +import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin; +import org.apache.activemq.artemis.jms.client.ActiveMQConnection; +import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; +import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMRules; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; +import org.junit.Test; +import org.junit.runner.RunWith; + +/** + * NOTE: this test should be run at log level INFO + * + * This test checks the LoggingActiveMQServerPlugin is logging expected data with specific property configurations + * when using CORE protocol + * + * Byteman is used to intercept the logged message and uses a helper method to put the messages into seperate lists + * based on its content. The test subsequently checks to ensure correct number of logs in each list. + */ +@RunWith(BMUnitRunner.class) +public class LoggingActiveMQServerPluginTest extends ActiveMQTestBase { + + + //buckets for logging + protected static List createdConnectionLogs = new ArrayList<>(); + protected static List destroyedConnectionLogs = new ArrayList<>(); + protected static List createdSessionLogs = new ArrayList<>(); + protected static List closedSessionLogs = new ArrayList<>(); + protected static List createdConsumerLogs = new ArrayList<>(); + protected static List closedConsumerLogs = new ArrayList<>(); + protected static List deliveredLogs = new ArrayList<>(); + protected static List ackedLogs = new ArrayList<>(); + protected static List sentLogs = new ArrayList<>(); + protected static List routedLogs = new ArrayList<>(); + protected static List createdQueueLogs = new ArrayList<>(); + protected static List destroyedQueueLogs = new ArrayList<>(); + protected static List messageExpiredLogs = new ArrayList<>(); + protected static List unexpectedLogs = new ArrayList<>(); + protected static List addedSessionMetaData = new ArrayList<>(); + + /** + * Aim: verify connection creation/destroy is logged when LOG_CONNECTION_EVENTS is set + * + * @throws Exception + */ + + @Test + @BMRules(rules = {@BMRule(name = "test LOG_CONNECTION_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogConnectEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_CONNECTION_EVENTS); + activeMQServer.start(); + + try { + + Connection connection = createActiveMQConnection(); + connection.start(); + Thread.sleep(100); + connection.close(); + + //ensure logs are collected. + Thread.sleep(500); + + assertEquals("created connections", 1, createdConnectionLogs.size()); + assertEquals("destroyed connections", 1, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 0, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test the session create/close events are logged when plugin configured with + * LOG_SESSION_EVENTS + * + * NOTE: Session plugin points seem to be invoked twice - did not expect that. + * + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test LOG_SESSION_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogSessionEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_SESSION_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(false, false, "test_message", 0); + Thread.sleep(500); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 2, createdSessionLogs.size()); + assertEquals("closed sessions", 2, closedSessionLogs.size()); + assertEquals("added Metadata logs", 2, addedSessionMetaData.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test the consumer create/close events are logged when plugin configured with + * LOG_CONSUMER_EVENTS + * @throws Exception + */ + + @Test + @BMRules(rules = {@BMRule(name = "test LOG_CONSUMER_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogConsumerEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_CONSUMER_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "txtMessage", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 1, createdConsumerLogs.size()); + assertEquals("closed consumer", 1, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 0, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test delivering events are logged when plugin configured with + * LOG_DELIVERING_EVENTS + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test LOG_DELIVERING_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogDeliveringEvents() throws Exception { + + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_DELIVERING_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "txtMessage 1", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 1, deliveredLogs.size()); + assertEquals("acked message", 1, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 0, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test sending events are logged when plugin configured with + * LOG_SENDING_EVENTS + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test LOG_SENDING_EVENTS", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogSendingEvents() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_SENDING_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "txtMessage 1", 0); + sendAndReceive(true, true, "txtMessage 2", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 2, sentLogs.size()); + assertEquals("routing message", 2, routedLogs.size()); + assertEquals("queue created", 0, createdQueueLogs.size()); + assertEquals("queue destroyed", 0, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test queue creation/destroy are logged when plugin configured with + * LOG_INTERNAL_EVENTS + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test queue creation log", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testQueueCreationLog() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_INTERNAL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(false, true, "NO_MESSAGE", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 1, createdQueueLogs.size()); + assertEquals("queue destroyed", 1, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test message expiry is logged when plugin configured with + * LOG_INTERNAL_EVENTS + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test queue creation log", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testExpireMessageLog() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_INTERNAL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, false, "message to expiry", 100); + Thread.sleep(1000); + sendAndReceive(false, true, "NO_MESSAGE", 0); + + assertEquals("created connections", 0, createdConnectionLogs.size()); + assertEquals("destroyed connections", 0, destroyedConnectionLogs.size()); + assertEquals("created sessions", 0, createdSessionLogs.size()); + assertEquals("closed sessions", 0, createdSessionLogs.size()); + assertEquals("created consumer", 0, createdConsumerLogs.size()); + assertEquals("closed consumer", 0, closedConsumerLogs.size()); + assertEquals("delivered message", 0, deliveredLogs.size()); + assertEquals("acked message", 0, ackedLogs.size()); + assertEquals("sending message", 0, sentLogs.size()); + assertEquals("routing message", 0, routedLogs.size()); + assertEquals("queue created", 1, createdQueueLogs.size()); + assertEquals("queue destroyed", 1, destroyedQueueLogs.size()); + assertEquals("expired message", 1, messageExpiredLogs.size()); + assertEquals("unexpected logs", 0, unexpectedLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + /** + * Aim: test all events are logged when plugin configured with + * LOG_ALL_EVENTS + * @throws Exception + */ + @Test + @BMRules(rules = {@BMRule(name = "test logAll EVENT", + targetClass = "org.jboss.logging.Logger", + targetMethod = "infof", + targetLocation = "ENTRY", + action = "org.apache.activemq.artemis.tests.extras.byteman.LoggingActiveMQServerPluginTest.infoLog($1, $0)")}) + public void testLogAll() throws Exception { + + //initial plugin + ActiveMQServer activeMQServer = createServerWithLoggingPlugin(LoggingActiveMQServerPlugin.LOG_ALL_EVENTS); + activeMQServer.start(); + + try { + + sendAndReceive(true, true, "log ALL Message_1", 0); + sendAndReceive(true, true, "log ALL Message_2", 0); + + Thread.sleep(500); + + assertEquals("created connections", 2, createdConnectionLogs.size()); + assertEquals("destroyed connections", 2, destroyedConnectionLogs.size()); + assertEquals("created consumer", 2, createdConsumerLogs.size()); + assertEquals("closed consumer", 2, closedConsumerLogs.size()); + assertEquals("delivered message", 2, deliveredLogs.size()); + assertEquals("acked message", 2, ackedLogs.size()); + assertEquals("sending message", 2, sentLogs.size()); + assertEquals("routing message", 2, routedLogs.size()); + assertEquals("queue created", 2, createdQueueLogs.size()); + assertEquals("queue destroyed", 2, destroyedQueueLogs.size()); + assertEquals("expired message", 0, messageExpiredLogs.size()); + + } finally { + activeMQServer.stop(); + //reset the logs lists + clearLogLists(); + } + } + + //collect the log invocation from LoggingActiveMQServerPlugin + public static void infoLog(Object message, org.jboss.logging.Logger logger) { + + //not interested + if (!logger.getName().equals(LoggingActiveMQServerPlugin.class.getName())) { + return; + } + + String stringMessage = (String) message; + + if (stringMessage.startsWith("created connection")) { + createdConnectionLogs.add(stringMessage); + } else if (stringMessage.startsWith("destroyed connection")) { + destroyedConnectionLogs.add(stringMessage); + } else if (stringMessage.startsWith("created session")) { + createdSessionLogs.add(stringMessage); + } else if (stringMessage.startsWith("closed session")) { + closedSessionLogs.add(stringMessage); + } else if (stringMessage.startsWith("added session metadata")) { + addedSessionMetaData.add(stringMessage); + } else if (stringMessage.startsWith("created consumer")) { + createdConsumerLogs.add(stringMessage); + } else if (stringMessage.startsWith("closed consumer")) { + closedConsumerLogs.add(stringMessage); + } else if (stringMessage.startsWith("delivered message")) { + deliveredLogs.add(stringMessage); + } else if (stringMessage.startsWith("acknowledged message")) { + ackedLogs.add(stringMessage); + } else if (stringMessage.startsWith("sent message")) { + sentLogs.add(stringMessage); + } else if (stringMessage.startsWith("routed message")) { + routedLogs.add(stringMessage); + } else if (stringMessage.startsWith("created queue")) { + createdQueueLogs.add(stringMessage); + } else if (stringMessage.startsWith("destroyed queue")) { + destroyedQueueLogs.add(stringMessage); + } else if (stringMessage.startsWith("expired")) { + messageExpiredLogs.add(stringMessage); + } else { + unexpectedLogs.add(stringMessage); + } + + } + + //----- helper methods --- + + protected void clearLogLists() { + createdConnectionLogs.clear(); + destroyedConnectionLogs.clear(); + createdSessionLogs.clear(); + closedSessionLogs.clear(); + createdConsumerLogs.clear(); + closedConsumerLogs.clear(); + deliveredLogs.clear(); + ackedLogs.clear(); + sentLogs.clear(); + routedLogs.clear(); + createdQueueLogs.clear(); + destroyedQueueLogs.clear(); + messageExpiredLogs.clear(); + unexpectedLogs.clear(); + addedSessionMetaData.clear(); + } + + protected ActiveMQServer createServerWithLoggingPlugin(String loggingPluginEventType) throws Exception { + //initial plugin + LoggingActiveMQServerPlugin loggingActiveMQServerPlugin = new LoggingActiveMQServerPlugin(); + Map properties = new HashMap<>(); + properties.put(loggingPluginEventType, "true"); + loggingActiveMQServerPlugin.init(properties); + + //register + Configuration config = createDefaultConfig(true); + config.registerBrokerPlugin(loggingActiveMQServerPlugin); + + return createServer(false, config); + } + + protected void sendAndReceive(boolean send, + boolean receive, + String txtMessage, + long expiry) throws JMSException, InterruptedException { + Connection connection = createActiveMQConnection(); + Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + connection.start(); + Queue queue = session.createQueue("TEST.QUEUE"); + MessageConsumer messageConsumer = null; + + if (receive) { + messageConsumer = session.createConsumer(queue); + Thread.sleep(1000); + } + if (send) { + MessageProducer messageProducer = session.createProducer(queue); + if (expiry > 0) { + messageProducer.setTimeToLive(expiry); + } + messageProducer.send(session.createTextMessage(txtMessage)); + } + if (receive) { + messageConsumer.receive(100); + messageConsumer.close(); + } + + session.close(); + connection.close(); + } + + protected Connection createActiveMQConnection() throws JMSException { + ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616"); + return (ActiveMQConnection) factory.createConnection(); + } +} \ No newline at end of file diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/XmlConfigPluginTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/XmlConfigPluginTest.java index 2dcd9ccfab..25e61a6bda 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/XmlConfigPluginTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/plugin/XmlConfigPluginTest.java @@ -20,6 +20,7 @@ import org.apache.activemq.artemis.core.config.FileDeploymentManager; import org.apache.activemq.artemis.core.config.impl.FileConfiguration; import org.apache.activemq.artemis.core.server.ActiveMQServer; import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl; +import org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin; import org.apache.activemq.artemis.jms.server.config.impl.FileJMSConfiguration; import org.apache.activemq.artemis.tests.util.ActiveMQTestBase; import org.junit.Test; @@ -28,14 +29,7 @@ public class XmlConfigPluginTest extends ActiveMQTestBase { @Test public void testStopStart1() throws Exception { - FileConfiguration fc = new FileConfiguration(); - FileJMSConfiguration fileConfiguration = new FileJMSConfiguration(); - FileDeploymentManager deploymentManager = new FileDeploymentManager("broker-plugins-config.xml"); - deploymentManager.addDeployable(fc); - deploymentManager.addDeployable(fileConfiguration); - deploymentManager.readConfiguration(); - - ActiveMQServer server = addServer(new ActiveMQServerImpl(fc)); + ActiveMQServer server = createServerFromConfig("broker-plugins-config.xml"); try { server.start(); assertEquals(2, server.getBrokerPlugins().size()); @@ -51,4 +45,70 @@ public class XmlConfigPluginTest extends ActiveMQTestBase { } } } + + /** + * Ensure the configuration is bring picked up correctly by LoggingActiveMQServerPlugin + * @throws Exception + */ + @Test + public void testLoggingActiveMQServerPlugin() throws Exception { + ActiveMQServer server = createServerFromConfig("broker-logging-plugin.xml"); + try { + server.start(); + assertEquals("only one plugin should be registered",1, server.getBrokerPlugins().size()); + assertTrue("ensure LoggingActiveMQServerPlugin is registered",server.getBrokerPlugins().get(0) instanceof LoggingActiveMQServerPlugin); + LoggingActiveMQServerPlugin loggingActiveMQServerPlugin = (LoggingActiveMQServerPlugin) server.getBrokerPlugins().get(0); + assertEquals("check logAll", true, loggingActiveMQServerPlugin.isLogAll()); + assertEquals("check logConnectionEvents", true, loggingActiveMQServerPlugin.isLogConnectionEvents()); + assertEquals("check logSessionEvents", true, loggingActiveMQServerPlugin.isLogSessionEvents()); + assertEquals("check logConsumerEvents", true, loggingActiveMQServerPlugin.isLogConsumerEvents()); + assertEquals("check logDeliveringEvents", true, loggingActiveMQServerPlugin.isLogDeliveringEvents()); + assertEquals("check logSendingEvents", true, loggingActiveMQServerPlugin.isLogSendingEvents()); + assertEquals("check logInternalEvents", true, loggingActiveMQServerPlugin.isLogInternalEvents()); + + } finally { + if (server != null) { + server.stop(); + } + } + } + + /** + * ensure the LoggingActiveMQServerPlugin uses default values when configured with incorrect values + * @throws Exception + */ + @Test + public void testLoggingActiveMQServerPluginWrongValue() throws Exception { + ActiveMQServer server = createServerFromConfig("broker-logging-plugin-wrong.xml"); + try { + server.start(); + assertEquals("only one plugin should be registered",1, server.getBrokerPlugins().size()); + assertTrue("ensure LoggingActiveMQServerPlugin is registered",server.getBrokerPlugins().get(0) instanceof LoggingActiveMQServerPlugin); + LoggingActiveMQServerPlugin loggingActiveMQServerPlugin = (LoggingActiveMQServerPlugin) server.getBrokerPlugins().get(0); + assertEquals("check logAll", false, loggingActiveMQServerPlugin.isLogAll()); + assertEquals("check logConnectionEvents", false, loggingActiveMQServerPlugin.isLogConnectionEvents()); + assertEquals("check logSessionEvents", false, loggingActiveMQServerPlugin.isLogSessionEvents()); + assertEquals("check logConsumerEvents", false, loggingActiveMQServerPlugin.isLogConsumerEvents()); + assertEquals("check logDeliveringEvents", false, loggingActiveMQServerPlugin.isLogDeliveringEvents()); + assertEquals("check logSendingEvents", false, loggingActiveMQServerPlugin.isLogSendingEvents()); + assertEquals("check logInternalEvents", false, loggingActiveMQServerPlugin.isLogInternalEvents()); + + } finally { + if (server != null) { + server.stop(); + } + } + } + + private ActiveMQServer createServerFromConfig(String configFileName) throws Exception { + FileConfiguration fc = new FileConfiguration(); + FileJMSConfiguration fileConfiguration = new FileJMSConfiguration(); + FileDeploymentManager deploymentManager = new FileDeploymentManager(configFileName); + deploymentManager.addDeployable(fc); + deploymentManager.addDeployable(fileConfiguration); + deploymentManager.readConfiguration(); + + return addServer(new ActiveMQServerImpl(fc)); + } + } diff --git a/tests/integration-tests/src/test/resources/broker-logging-plugin-wrong.xml b/tests/integration-tests/src/test/resources/broker-logging-plugin-wrong.xml new file mode 100644 index 0000000000..6ae718957e --- /dev/null +++ b/tests/integration-tests/src/test/resources/broker-logging-plugin-wrong.xml @@ -0,0 +1,51 @@ + + + + + + + + + tcp://localhost:61616 + + + ./target/tmp/activemq-unit-test/broker-plugin-test/ + + + tcp://localhost:61616 + + + false + + + +

myAddress
+ + + + + + + + + + + + diff --git a/tests/integration-tests/src/test/resources/broker-logging-plugin.xml b/tests/integration-tests/src/test/resources/broker-logging-plugin.xml new file mode 100644 index 0000000000..445163819f --- /dev/null +++ b/tests/integration-tests/src/test/resources/broker-logging-plugin.xml @@ -0,0 +1,56 @@ + + + + + + + + + tcp://localhost:61616 + + + ./target/tmp/activemq-unit-test/broker-plugin-test/ + + + tcp://localhost:61616 + + + false + + + +
myAddress
+ + + + + + + + + + + + + + + + +