MAILBOX-376 Metrics for RabbitMQ event-bus listener execution time
Project: http://git-wip-us.apache.org/repos/asf/james-project/repo Commit: http://git-wip-us.apache.org/repos/asf/james-project/commit/21603694 Tree: http://git-wip-us.apache.org/repos/asf/james-project/tree/21603694 Diff: http://git-wip-us.apache.org/repos/asf/james-project/diff/21603694 Branch: refs/heads/master Commit: 21603694c754c93347bc0c29162dca73d9ee2b31 Parents: 97b33f8 Author: Benoit Tellier <[email protected]> Authored: Wed Jan 23 15:10:45 2019 +0700 Committer: Benoit Tellier <[email protected]> Committed: Wed Jan 23 17:40:01 2019 +0700 ---------------------------------------------------------------------- .../apache/james/mailbox/events/EventBus.java | 6 +++ .../events/delivery/InVmEventDelivery.java | 5 +- mailbox/event/event-rabbitmq/pom.xml | 4 ++ .../james/mailbox/events/EventDispatcher.java | 17 +++---- .../james/mailbox/events/GroupRegistration.java | 18 +++---- .../events/GroupRegistrationHandler.java | 7 ++- .../mailbox/events/KeyRegistrationHandler.java | 17 +++---- .../mailbox/events/MailboxListenerExecutor.java | 52 ++++++++++++++++++++ .../james/mailbox/events/RabbitMQEventBus.java | 11 +++-- .../mailbox/events/RabbitMQEventBusTest.java | 3 +- 10 files changed, 102 insertions(+), 38 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java ---------------------------------------------------------------------- diff --git a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java index 2563fa6..2333736 100644 --- a/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java +++ b/mailbox/api/src/main/java/org/apache/james/mailbox/events/EventBus.java @@ -39,6 +39,12 @@ public interface EventBus { String REGISTRATION_KEY = "registrationKey"; } + interface Metrics { + static String timerName(MailboxListener mailboxListener) { + return "mailbox-listener-" + mailboxListener.getClass().getSimpleName(); + } + } + Registration register(MailboxListener listener, RegistrationKey key); Registration register(MailboxListener listener, Group group) throws GroupAlreadyRegistered; http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-memory/src/main/java/org/apache/james/mailbox/events/delivery/InVmEventDelivery.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-memory/src/main/java/org/apache/james/mailbox/events/delivery/InVmEventDelivery.java b/mailbox/event/event-memory/src/main/java/org/apache/james/mailbox/events/delivery/InVmEventDelivery.java index 90ae81f..f0b1b1e 100644 --- a/mailbox/event/event-memory/src/main/java/org/apache/james/mailbox/events/delivery/InVmEventDelivery.java +++ b/mailbox/event/event-memory/src/main/java/org/apache/james/mailbox/events/delivery/InVmEventDelivery.java @@ -19,6 +19,8 @@ package org.apache.james.mailbox.events.delivery; +import static org.apache.james.mailbox.events.EventBus.Metrics.timerName; + import java.io.Closeable; import javax.inject.Inject; @@ -81,7 +83,7 @@ public class InVmEventDelivery implements EventDelivery { } private void doDeliverToListener(MailboxListener mailboxListener, Event event) { - TimeMetric timer = metricFactory.timer("mailbox-listener-" + mailboxListener.getClass().getSimpleName()); + TimeMetric timer = metricFactory.timer(timerName(mailboxListener)); try (Closeable mdc = MDCBuilder.create() .addContext(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) .addContext(EventBus.StructuredLoggingFields.EVENT_CLASS, event.getClass()) @@ -96,6 +98,7 @@ public class InVmEventDelivery implements EventDelivery { timer.stopAndPublish(); } } + private StructuredLogger structuredLogger(Event event, MailboxListener mailboxListener) { return MDCStructuredLogger.forLogger(LOGGER) .addField(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/pom.xml ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/pom.xml b/mailbox/event/event-rabbitmq/pom.xml index 605fe6d..157be73 100644 --- a/mailbox/event/event-rabbitmq/pom.xml +++ b/mailbox/event/event-rabbitmq/pom.xml @@ -67,6 +67,10 @@ <scope>test</scope> </dependency> <dependency> + <groupId>${project.groupId}</groupId> + <artifactId>metrics-api</artifactId> + </dependency> + <dependency> <groupId>com.jayway.awaitility</groupId> <artifactId>awaitility</artifactId> <scope>test</scope> http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java index 1331e29..628bab3 100644 --- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/EventDispatcher.java @@ -24,7 +24,6 @@ import static org.apache.james.backend.rabbitmq.Constants.DURABLE; import static org.apache.james.mailbox.events.RabbitMQEventBus.EVENT_BUS_ID; import static org.apache.james.mailbox.events.RabbitMQEventBus.MAILBOX_EVENT_EXCHANGE_NAME; -import java.io.Closeable; import java.nio.charset.StandardCharsets; import java.util.Set; @@ -57,14 +56,16 @@ class EventDispatcher { private final Sender sender; private final MailboxListenerRegistry mailboxListenerRegistry; private final AMQP.BasicProperties basicProperties; + private final MailboxListenerExecutor mailboxListenerExecutor; - EventDispatcher(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, MailboxListenerRegistry mailboxListenerRegistry) { + EventDispatcher(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, MailboxListenerRegistry mailboxListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) { this.eventSerializer = eventSerializer; this.sender = sender; this.mailboxListenerRegistry = mailboxListenerRegistry; this.basicProperties = new AMQP.BasicProperties.Builder() .headers(ImmutableMap.of(EVENT_BUS_ID, eventBusId.asString())) .build(); + this.mailboxListenerExecutor = mailboxListenerExecutor; } void start() { @@ -98,14 +99,10 @@ class EventDispatcher { } private void executeListener(Event event, MailboxListener mailboxListener, RegistrationKey registrationKey) throws Exception { - try (Closeable mdc = MDCBuilder.create() - .addContext(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) - .addContext(EventBus.StructuredLoggingFields.EVENT_CLASS, event.getClass()) - .addContext(EventBus.StructuredLoggingFields.USER, event.getUser()) - .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, registrationKey) - .build()) { - mailboxListener.event(event); - } + mailboxListenerExecutor.execute(mailboxListener, + MDCBuilder.create() + .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, registrationKey), + event); } private StructuredLogger structuredLogger(Event event, Set<RegistrationKey> keys) { http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java index d975ef7..ee3013d 100644 --- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistration.java @@ -27,7 +27,6 @@ import static org.apache.james.backend.rabbitmq.Constants.NO_ARGUMENTS; import static org.apache.james.mailbox.events.RabbitMQEventBus.MAILBOX_EVENT; import static org.apache.james.mailbox.events.RabbitMQEventBus.MAILBOX_EVENT_EXCHANGE_NAME; -import java.io.Closeable; import java.nio.charset.StandardCharsets; import java.util.Objects; import java.util.Optional; @@ -86,17 +85,19 @@ class GroupRegistration implements Registration { private final GroupConsumerRetry retryHandler; private final WaitDelayGenerator delayGenerator; private final Group group; + private final MailboxListenerExecutor mailboxListenerExecutor; private Optional<Disposable> receiverSubscriber; GroupRegistration(Mono<Connection> connectionSupplier, Sender sender, EventSerializer eventSerializer, MailboxListener mailboxListener, Group group, RetryBackoffConfiguration retryBackoff, EventDeadLetters eventDeadLetters, - Runnable unregisterGroup) { + Runnable unregisterGroup, MailboxListenerExecutor mailboxListenerExecutor) { this.eventSerializer = eventSerializer; this.mailboxListener = mailboxListener; this.queueName = WorkQueueName.of(group); this.sender = sender; this.receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connectionSupplier)); + this.mailboxListenerExecutor = mailboxListenerExecutor; this.receiverSubscriber = Optional.empty(); this.unregisterGroup = unregisterGroup; this.retryHandler = new GroupConsumerRetry(sender, group, retryBackoff, eventDeadLetters); @@ -149,14 +150,11 @@ class GroupRegistration implements Registration { } private void runListener(Event event) throws Exception { - try (Closeable mdc = MDCBuilder.create() - .addContext(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) - .addContext(EventBus.StructuredLoggingFields.EVENT_CLASS, event.getClass()) - .addContext(EventBus.StructuredLoggingFields.USER, event.getUser()) - .addContext(EventBus.StructuredLoggingFields.GROUP, group) - .build()) { - mailboxListener.event(event); - } + mailboxListenerExecutor.execute( + mailboxListener, + MDCBuilder.create() + .addContext(EventBus.StructuredLoggingFields.GROUP, group), + event); } private int getRetryCount(AcknowledgableDelivery acknowledgableDelivery) { http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistrationHandler.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistrationHandler.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistrationHandler.java index 9e3e062..1950aab 100644 --- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistrationHandler.java +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/GroupRegistrationHandler.java @@ -37,15 +37,17 @@ class GroupRegistrationHandler { private final Mono<Connection> connectionMono; private final RetryBackoffConfiguration retryBackoff; private final EventDeadLetters eventDeadLetters; + private final MailboxListenerExecutor mailboxListenerExecutor; GroupRegistrationHandler(EventSerializer eventSerializer, Sender sender, Mono<Connection> connectionMono, RetryBackoffConfiguration retryBackoff, - EventDeadLetters eventDeadLetters) { + EventDeadLetters eventDeadLetters, MailboxListenerExecutor mailboxListenerExecutor) { this.eventSerializer = eventSerializer; this.sender = sender; this.connectionMono = connectionMono; this.retryBackoff = retryBackoff; this.eventDeadLetters = eventDeadLetters; + this.mailboxListenerExecutor = mailboxListenerExecutor; this.groupRegistrations = new ConcurrentHashMap<>(); } @@ -73,6 +75,7 @@ class GroupRegistrationHandler { group, retryBackoff, eventDeadLetters, - () -> groupRegistrations.remove(group)); + () -> groupRegistrations.remove(group), + mailboxListenerExecutor); } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java index b12e5b4..6e84254 100644 --- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/KeyRegistrationHandler.java @@ -25,7 +25,6 @@ import static org.apache.james.backend.rabbitmq.Constants.EXCLUSIVE; import static org.apache.james.backend.rabbitmq.Constants.NO_ARGUMENTS; import static org.apache.james.mailbox.events.RabbitMQEventBus.EVENT_BUS_ID; -import java.io.Closeable; import java.nio.charset.StandardCharsets; import java.util.Optional; @@ -63,15 +62,17 @@ class KeyRegistrationHandler { private final Receiver receiver; private final RegistrationQueueName registrationQueue; private final RegistrationBinder registrationBinder; + private final MailboxListenerExecutor mailboxListenerExecutor; private Optional<Disposable> receiverSubscriber; - KeyRegistrationHandler(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, Mono<Connection> connectionMono, RoutingKeyConverter routingKeyConverter, MailboxListenerRegistry mailboxListenerRegistry) { + KeyRegistrationHandler(EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, Mono<Connection> connectionMono, RoutingKeyConverter routingKeyConverter, MailboxListenerRegistry mailboxListenerRegistry, MailboxListenerExecutor mailboxListenerExecutor) { this.eventBusId = eventBusId; this.eventSerializer = eventSerializer; this.sender = sender; this.routingKeyConverter = routingKeyConverter; this.mailboxListenerRegistry = mailboxListenerRegistry; this.receiver = RabbitFlux.createReceiver(new ReceiverOptions().connectionMono(connectionMono)); + this.mailboxListenerExecutor = mailboxListenerExecutor; this.registrationQueue = new RegistrationQueueName(); this.registrationBinder = new RegistrationBinder(sender, registrationQueue); } @@ -133,14 +134,10 @@ class KeyRegistrationHandler { } private void executeListener(MailboxListener listener, Event event, RegistrationKey key) throws Exception { - try (Closeable mdc = MDCBuilder.create() - .addContext(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) - .addContext(EventBus.StructuredLoggingFields.EVENT_CLASS, event.getClass()) - .addContext(EventBus.StructuredLoggingFields.USER, event.getUser()) - .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, key) - .build()) { - listener.event(event); - } + mailboxListenerExecutor.execute(listener, + MDCBuilder.create() + .addContext(EventBus.StructuredLoggingFields.REGISTRATION_KEY, key), + event); } private boolean isLocalSynchronousListeners(EventBusId eventBusId, MailboxListener listener) { http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerExecutor.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerExecutor.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerExecutor.java new file mode 100644 index 0000000..cea4d9e --- /dev/null +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/MailboxListenerExecutor.java @@ -0,0 +1,52 @@ +/**************************************************************** + * 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.james.mailbox.events; + +import static org.apache.james.mailbox.events.EventBus.Metrics.timerName; + +import java.io.Closeable; + +import org.apache.james.mailbox.Event; +import org.apache.james.mailbox.MailboxListener; +import org.apache.james.metrics.api.MetricFactory; +import org.apache.james.metrics.api.TimeMetric; +import org.apache.james.util.MDCBuilder; + +public class MailboxListenerExecutor { + private final MetricFactory metricFactory; + + public MailboxListenerExecutor(MetricFactory metricFactory) { + this.metricFactory = metricFactory; + } + + void execute(MailboxListener listener, MDCBuilder mdcBuilder, Event event) throws Exception { + TimeMetric timer = metricFactory.timer(timerName(listener)); + try (Closeable mdc = mdcBuilder + .addContext(EventBus.StructuredLoggingFields.EVENT_ID, event.getEventId()) + .addContext(EventBus.StructuredLoggingFields.EVENT_CLASS, event.getClass()) + .addContext(EventBus.StructuredLoggingFields.USER, event.getUser()) + .addContext(EventBus.StructuredLoggingFields.LISTENER_CLASS, listener.getClass()) + .build()) { + listener.event(event); + } finally { + timer.stopAndPublish(); + } + } +} http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java index 122ba04..8fba81d 100644 --- a/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java +++ b/mailbox/event/event-rabbitmq/src/main/java/org/apache/james/mailbox/events/RabbitMQEventBus.java @@ -29,6 +29,7 @@ import org.apache.james.backend.rabbitmq.RabbitMQConnectionFactory; import org.apache.james.event.json.EventSerializer; import org.apache.james.mailbox.Event; import org.apache.james.mailbox.MailboxListener; +import org.apache.james.metrics.api.MetricFactory; import com.rabbitmq.client.Connection; @@ -49,6 +50,7 @@ public class RabbitMQEventBus implements EventBus { private final RetryBackoffConfiguration retryBackoff; private final EventBusId eventBusId; private final EventDeadLetters eventDeadLetters; + private final MailboxListenerExecutor mailboxListenerExecutor; private MailboxListenerRegistry mailboxListenerRegistry; private GroupRegistrationHandler groupRegistrationHandler; @@ -60,7 +62,8 @@ public class RabbitMQEventBus implements EventBus { RabbitMQEventBus(RabbitMQConnectionFactory rabbitMQConnectionFactory, EventSerializer eventSerializer, RetryBackoffConfiguration retryBackoff, RoutingKeyConverter routingKeyConverter, - EventDeadLetters eventDeadLetters) { + EventDeadLetters eventDeadLetters, MetricFactory metricFactory) { + this.mailboxListenerExecutor = new MailboxListenerExecutor(metricFactory); this.eventBusId = EventBusId.random(); this.connectionMono = Mono.fromSupplier(rabbitMQConnectionFactory::create).cache(); this.eventSerializer = eventSerializer; @@ -74,9 +77,9 @@ public class RabbitMQEventBus implements EventBus { if (!isRunning.get()) { sender = RabbitFlux.createSender(new SenderOptions().connectionMono(connectionMono)); mailboxListenerRegistry = new MailboxListenerRegistry(); - keyRegistrationHandler = new KeyRegistrationHandler(eventBusId, eventSerializer, sender, connectionMono, routingKeyConverter, mailboxListenerRegistry); - groupRegistrationHandler = new GroupRegistrationHandler(eventSerializer, sender, connectionMono, retryBackoff, eventDeadLetters); - eventDispatcher = new EventDispatcher(eventBusId, eventSerializer, sender, mailboxListenerRegistry); + keyRegistrationHandler = new KeyRegistrationHandler(eventBusId, eventSerializer, sender, connectionMono, routingKeyConverter, mailboxListenerRegistry, mailboxListenerExecutor); + groupRegistrationHandler = new GroupRegistrationHandler(eventSerializer, sender, connectionMono, retryBackoff, eventDeadLetters, mailboxListenerExecutor); + eventDispatcher = new EventDispatcher(eventBusId, eventSerializer, sender, mailboxListenerRegistry, mailboxListenerExecutor); eventDispatcher.start(); keyRegistrationHandler.start(); http://git-wip-us.apache.org/repos/asf/james-project/blob/21603694/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java ---------------------------------------------------------------------- diff --git a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java index efe5296..69204ad 100644 --- a/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java +++ b/mailbox/event/event-rabbitmq/src/test/java/org/apache/james/mailbox/events/RabbitMQEventBusTest.java @@ -56,6 +56,7 @@ import org.apache.james.mailbox.Event; import org.apache.james.mailbox.MailboxListener; import org.apache.james.mailbox.model.TestId; import org.apache.james.mailbox.model.TestMessageId; +import org.apache.james.metrics.api.NoopMetricFactory; import org.apache.james.util.concurrency.ConcurrentTestRunner; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -126,7 +127,7 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract, } private RabbitMQEventBus newEventBus() { - return new RabbitMQEventBus(connectionFactory, eventSerializer, RetryBackoffConfiguration.DEFAULT, routingKeyConverter, memoryEventDeadLetters); + return new RabbitMQEventBus(connectionFactory, eventSerializer, RetryBackoffConfiguration.DEFAULT, routingKeyConverter, memoryEventDeadLetters, new NoopMetricFactory()); } @Override --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
