You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2022/09/25 15:01:28 UTC
[james-project] branch master updated (e11041369c -> 583b7d4a58)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
from e11041369c [CONF] Remove vendor URLs from smtpserver.xml default configuration examples (#1204)
new ac4061bd61 JAMES-3817 RabbitMQManagementAPI should parse the queue length property (`messages`) when getting Queue details
new 709609759f JAMES-3817 Implement health check for RabbitMQ dead letter queue of mail queue
new f2659d9d67 JAMES-3817 Implement health check for RabbitMQ dead letter queues of event bus
new 583b7d4a58 JAMES-3817 Integration tests for health check
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../backends/rabbitmq/RabbitMQManagementAPI.java | 7 +
.../james/backends/rabbitmq/RabbitMQTest.java | 20 +++
.../org/apache/james/events/NamingStrategy.java | 2 +
...abbitMQEventBusDeadLetterQueueHealthCheck.java} | 30 ++--
...itMQEventBusDeadLetterQueueHealthCheckTest.java | 155 +++++++++++++++++++++
.../james/modules/event/JMAPEventBusModule.java | 4 +-
.../modules/event/RabbitMQEventBusModule.java | 5 +
.../modules/queue/rabbitmq/RabbitMQModule.java | 5 +-
...itMQWebAdminServerIntegrationImmutableTest.java | 23 +++
...moryWebAdminServerIntegrationImmutableTest.java | 23 +++
...abbitMQMailQueueDeadLetterQueueHealthCheck.java | 33 +++--
...tMQMailQueueDeadLetterQueueHealthCheckTest.java | 133 ++++++++++++++++++
12 files changed, 414 insertions(+), 26 deletions(-)
copy event-bus/{api/src/main/java/org/apache/james/events/EventDeadLettersHealthCheck.java => distributed/src/main/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheck.java} (52%)
create mode 100644 event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheckTest.java
copy event-bus/api/src/main/java/org/apache/james/events/EventDeadLettersHealthCheck.java => server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheck.java (53%)
create mode 100644 server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheckTest.java
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/04: JAMES-3817 Implement health check for RabbitMQ dead letter queues of event bus
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit f2659d9d671bdc49e879b0483e19d4b2078e160b
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Tue Sep 20 17:08:31 2022 +0700
JAMES-3817 Implement health check for RabbitMQ dead letter queues of event bus
Degraded if dead letter queues length are not equal to 0.
---
.../org/apache/james/events/NamingStrategy.java | 2 +
...RabbitMQEventBusDeadLetterQueueHealthCheck.java | 65 +++++++++
...itMQEventBusDeadLetterQueueHealthCheckTest.java | 155 +++++++++++++++++++++
.../james/modules/event/JMAPEventBusModule.java | 4 +-
.../modules/event/RabbitMQEventBusModule.java | 5 +
5 files changed, 229 insertions(+), 2 deletions(-)
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/NamingStrategy.java b/event-bus/distributed/src/main/java/org/apache/james/events/NamingStrategy.java
index 80dc2d2a4f..6e34d15d2d 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/NamingStrategy.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/NamingStrategy.java
@@ -22,6 +22,8 @@ package org.apache.james.events;
import reactor.rabbitmq.QueueSpecification;
public class NamingStrategy {
+ public static final NamingStrategy JMAP_NAMING_STRATEGY = new NamingStrategy("jmapEvent");
+
private final String prefix;
public NamingStrategy(String prefix) {
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheck.java b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheck.java
new file mode 100644
index 0000000000..6eeb2af045
--- /dev/null
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheck.java
@@ -0,0 +1,65 @@
+/****************************************************************
+ * 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.events;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
+import org.apache.james.backends.rabbitmq.RabbitMQManagementAPI;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+
+import reactor.core.publisher.Mono;
+
+public class RabbitMQEventBusDeadLetterQueueHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("RabbitMQEventBusDeadLetterQueueHealthCheck");
+ private static final String DEFAULT_VHOST = "/";
+
+ private final RabbitMQConfiguration configuration;
+ private final NamingStrategy mailboxEventNamingStrategy;
+ private final RabbitMQManagementAPI api;
+
+ @Inject
+ public RabbitMQEventBusDeadLetterQueueHealthCheck(RabbitMQConfiguration configuration, NamingStrategy mailboxEventNamingStrategy) {
+ this.configuration = configuration;
+ this.mailboxEventNamingStrategy = mailboxEventNamingStrategy;
+ this.api = RabbitMQManagementAPI.from(configuration);
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ return Mono.fromCallable(() -> api.queueDetails(configuration.getVhost().orElse(DEFAULT_VHOST), mailboxEventNamingStrategy.deadLetterQueue().getName()).getQueueLength())
+ .mergeWith(Mono.fromCallable(() -> api.queueDetails(configuration.getVhost().orElse(DEFAULT_VHOST), NamingStrategy.JMAP_NAMING_STRATEGY.deadLetterQueue().getName()).getQueueLength()))
+ .reduce(Long::sum)
+ .map(queueSize -> {
+ if (queueSize != 0) {
+ return Result.degraded(COMPONENT_NAME, "RabbitMQ dead letter queue of the event bus contain events. This might indicate transient failure on event processing.");
+ }
+ return Result.healthy(COMPONENT_NAME);
+ })
+ .onErrorResume(e -> Mono.just(Result.unhealthy(COMPONENT_NAME, "Error checking RabbitMQEventBusDeadLetterHealthCheck", e)));
+ }
+}
diff --git a/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheckTest.java b/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheckTest.java
new file mode 100644
index 0000000000..c5f2978fe1
--- /dev/null
+++ b/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusDeadLetterQueueHealthCheckTest.java
@@ -0,0 +1,155 @@
+/****************************************************************
+ * 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.events;
+
+import static com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN;
+import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
+import static org.apache.james.backends.rabbitmq.Constants.DIRECT_EXCHANGE;
+import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
+import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
+import static org.apache.james.backends.rabbitmq.RabbitMQFixture.EXCHANGE_NAME;
+import static org.apache.james.backends.rabbitmq.RabbitMQFixture.awaitAtMostOneMinute;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.james.backends.rabbitmq.DockerRabbitMQ;
+import org.apache.james.backends.rabbitmq.RabbitMQExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.google.common.collect.ImmutableMap;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+class RabbitMQEventBusDeadLetterQueueHealthCheckTest {
+ @RegisterExtension
+ RabbitMQExtension rabbitMQExtension = RabbitMQExtension.singletonRabbitMQ()
+ .isolationPolicy(RabbitMQExtension.IsolationPolicy.STRONG);
+
+ public static final ImmutableMap<String, Object> NO_QUEUE_DECLARE_ARGUMENTS = ImmutableMap.of();
+ public static final NamingStrategy MAILBOX_EVENTS_NAMING_STRATEGY = new NamingStrategy("mailboxEvents");
+ public static final String ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS = "mailboxEventsRoutingKey";
+ public static final String ROUTING_KEY_JMAP_EVENTS_EVENT_BUS = "mailboxEventsRoutingKey";
+
+ private Connection connection;
+ private Channel channel;
+ private RabbitMQEventBusDeadLetterQueueHealthCheck testee;
+
+ @BeforeEach
+ void setup(DockerRabbitMQ rabbitMQ) throws IOException, TimeoutException, URISyntaxException {
+ ConnectionFactory connectionFactory = rabbitMQ.connectionFactory();
+ connectionFactory.setNetworkRecoveryInterval(1000);
+ connection = connectionFactory.newConnection();
+ channel = connection.createChannel();
+ testee = new RabbitMQEventBusDeadLetterQueueHealthCheck(rabbitMQ.getConfiguration(), MAILBOX_EVENTS_NAMING_STRATEGY);
+ }
+
+ @AfterEach
+ void tearDown(DockerRabbitMQ rabbitMQ) throws Exception {
+ closeQuietly(connection, channel);
+ rabbitMQ.reset();
+ }
+
+ @Test
+ void healthCheckShouldReturnUnhealthyWhenRabbitMQIsDown() throws Exception {
+ rabbitMQExtension.getRabbitMQ().stopApp();
+
+ assertThat(testee.check().block().isUnHealthy()).isTrue();
+ }
+
+ @Test
+ void healthCheckShouldReturnHealthyWhenDeadLetterQueuesAreEmpty() throws Exception {
+ createDeadLetterQueue(channel, MAILBOX_EVENTS_NAMING_STRATEGY, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+ createDeadLetterQueue(channel, NamingStrategy.JMAP_NAMING_STRATEGY, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+
+ assertThat(testee.check().block().isHealthy()).isTrue();
+ }
+
+ @Test
+ void healthCheckShouldReturnUnhealthyWhenThereIsNoDeadLetterQueue() {
+ assertThat(testee.check().block().isUnHealthy()).isTrue();
+ }
+
+ @Test
+ void healthCheckShouldReturnDegradedWhenMailboxEventBusDeadLetterQueueIsNotEmpty() throws Exception {
+ createDeadLetterQueue(channel, MAILBOX_EVENTS_NAMING_STRATEGY, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+ createDeadLetterQueue(channel, NamingStrategy.JMAP_NAMING_STRATEGY, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+ publishAMessage(channel, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+
+ awaitAtMostOneMinute.until(() -> testee.check().block().isDegraded());
+ }
+
+ @Test
+ void healthCheckShouldReturnDegradedWhenJmapEventBusDeadLetterQueueIsNotEmpty() throws Exception {
+ createDeadLetterQueue(channel, MAILBOX_EVENTS_NAMING_STRATEGY, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+ createDeadLetterQueue(channel, NamingStrategy.JMAP_NAMING_STRATEGY, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+ publishAMessage(channel, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+
+ awaitAtMostOneMinute.until(() -> testee.check().block().isDegraded());
+ }
+
+ @Test
+ void healthCheckShouldReturnDegradedWhenDeadLetterQueuesAreNotEmpty() throws Exception {
+ createDeadLetterQueue(channel, MAILBOX_EVENTS_NAMING_STRATEGY, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+ createDeadLetterQueue(channel, NamingStrategy.JMAP_NAMING_STRATEGY, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+
+ publishAMessage(channel, ROUTING_KEY_MAILBOX_EVENTS_EVENT_BUS);
+ publishAMessage(channel, ROUTING_KEY_JMAP_EVENTS_EVENT_BUS);
+
+ awaitAtMostOneMinute.until(() -> testee.check().block().isDegraded());
+ }
+
+ private void createDeadLetterQueue(Channel channel, NamingStrategy namingStrategy, String routingKey) throws IOException {
+ channel.exchangeDeclare(EXCHANGE_NAME, DIRECT_EXCHANGE, DURABLE);
+ channel.queueDeclare(namingStrategy.deadLetterQueue().getName(), DURABLE, !EXCLUSIVE, AUTO_DELETE, NO_QUEUE_DECLARE_ARGUMENTS).getQueue();
+ channel.queueBind(namingStrategy.deadLetterQueue().getName(), EXCHANGE_NAME, routingKey);
+ }
+
+ private void publishAMessage(Channel channel, String routingKey) throws IOException {
+ AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder()
+ .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode())
+ .priority(PERSISTENT_TEXT_PLAIN.getPriority())
+ .contentType(PERSISTENT_TEXT_PLAIN.getContentType())
+ .build();
+
+ channel.basicPublish(EXCHANGE_NAME, routingKey, basicProperties, "Hello, world!".getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void closeQuietly(AutoCloseable... closeables) {
+ Arrays.stream(closeables).forEach(this::closeQuietly);
+ }
+
+ private void closeQuietly(AutoCloseable closeable) {
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ //ignore error
+ }
+ }
+}
diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
index dbff6cd7a3..25fa79f7e9 100644
--- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
+++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
@@ -19,6 +19,8 @@
package org.apache.james.modules.event;
+import static org.apache.james.events.NamingStrategy.JMAP_NAMING_STRATEGY;
+
import javax.inject.Named;
import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
@@ -30,7 +32,6 @@ import org.apache.james.events.EventBusId;
import org.apache.james.events.EventBusReconnectionHandler;
import org.apache.james.events.EventDeadLetters;
import org.apache.james.events.KeyReconnectionHandler;
-import org.apache.james.events.NamingStrategy;
import org.apache.james.events.RabbitMQEventBus;
import org.apache.james.events.RetryBackoffConfiguration;
import org.apache.james.events.RoutingKeyConverter;
@@ -52,7 +53,6 @@ import com.google.inject.name.Names;
import reactor.rabbitmq.Sender;
public class JMAPEventBusModule extends AbstractModule {
- public static final NamingStrategy JMAP_NAMING_STRATEGY = new NamingStrategy("jmapEvent");
@Override
protected void configure() {
diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
index 16f3196c88..2eb8c655a0 100644
--- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
+++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
@@ -20,6 +20,7 @@
package org.apache.james.modules.event;
import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.event.json.MailboxEventSerializer;
import org.apache.james.events.EventBus;
import org.apache.james.events.EventBusId;
@@ -28,6 +29,7 @@ import org.apache.james.events.EventSerializer;
import org.apache.james.events.KeyReconnectionHandler;
import org.apache.james.events.NamingStrategy;
import org.apache.james.events.RabbitMQEventBus;
+import org.apache.james.events.RabbitMQEventBusDeadLetterQueueHealthCheck;
import org.apache.james.events.RegistrationKey;
import org.apache.james.events.RetryBackoffConfiguration;
import org.apache.james.mailbox.events.MailboxIdRegistrationKey;
@@ -63,6 +65,9 @@ public class RabbitMQEventBusModule extends AbstractModule {
Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(KeyReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(EventBusReconnectionHandler.class);
+
+ Multibinder.newSetBinder(binder(), HealthCheck.class)
+ .addBinding().to(RabbitMQEventBusDeadLetterQueueHealthCheck.class);
}
@ProvidesIntoSet
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/04: JAMES-3817 Integration tests for health check
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 583b7d4a584afe05f9a2c37ffabe68474e7e62d7
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Wed Sep 21 15:33:25 2022 +0700
JAMES-3817 Integration tests for health check
Make sure dead letter queue healthchecks are binded well
---
...itMQWebAdminServerIntegrationImmutableTest.java | 23 ++++++++++++++++++++++
...moryWebAdminServerIntegrationImmutableTest.java | 23 ++++++++++++++++++++++
2 files changed, 46 insertions(+)
diff --git a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
index 979407793e..8563add768 100644
--- a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
+++ b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
@@ -24,9 +24,12 @@ import static io.restassured.RestAssured.when;
import static io.restassured.RestAssured.with;
import static org.apache.james.JamesServerExtension.Lifecycle.PER_CLASS;
import static org.apache.james.webadmin.Constants.JSON_CONTENT_TYPE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.is;
+import java.util.List;
+
import org.apache.james.CassandraExtension;
import org.apache.james.CassandraRabbitMQJamesConfiguration;
import org.apache.james.CassandraRabbitMQJamesServerMain;
@@ -40,6 +43,7 @@ import org.apache.james.modules.AwsS3BlobStoreExtension;
import org.apache.james.modules.RabbitMQExtension;
import org.apache.james.modules.blobstore.BlobStoreConfiguration;
import org.apache.james.webadmin.integration.WebAdminServerIntegrationImmutableTest;
+import org.apache.james.webadmin.routes.HealthCheckRoutes;
import org.apache.james.webadmin.routes.TasksRoutes;
import org.eclipse.jetty.http.HttpStatus;
import org.junit.jupiter.api.Tag;
@@ -117,4 +121,23 @@ class RabbitMQWebAdminServerIntegrationImmutableTest extends WebAdminServerInteg
.body("additionalInformation.fixedInconsistencies", hasSize(0))
.body("additionalInformation.errors", hasSize(0));
}
+
+ @Test
+ void healthCheckComponentsList() {
+ List<String> listComponentNames =
+ when()
+ .get(HealthCheckRoutes.HEALTHCHECK)
+ .then()
+ .statusCode(HttpStatus.OK_200)
+ .extract()
+ .body()
+ .jsonPath()
+ .getList("checks.componentName", String.class);
+
+ assertThat(listComponentNames).containsOnly("Guice application lifecycle", "EmptyErrorMailRepository",
+ "RabbitMQ backend", "RabbitMQMailQueueDeadLetterQueueHealthCheck",
+ "RabbitMQEventBusDeadLetterQueueHealthCheck", "MailReceptionCheck",
+ "Cassandra backend", "EventDeadLettersHealthCheck", "MessageFastViewProjection",
+ "RabbitMQMailQueue BrowseStart", "OpenSearch Backend");
+ }
}
diff --git a/server/protocols/webadmin-integration-test/memory-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/memory/MemoryWebAdminServerIntegrationImmutableTest.java b/server/protocols/webadmin-integration-test/memory-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/memory/MemoryWebAdminServerIntegrationImmutableTest.java
index 094d4bf961..31046efbd5 100644
--- a/server/protocols/webadmin-integration-test/memory-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/memory/MemoryWebAdminServerIntegrationImmutableTest.java
+++ b/server/protocols/webadmin-integration-test/memory-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/memory/MemoryWebAdminServerIntegrationImmutableTest.java
@@ -19,14 +19,21 @@
package org.apache.james.webadmin.integration.memory;
+import static io.restassured.RestAssured.when;
import static org.apache.james.JamesServerExtension.Lifecycle.PER_CLASS;
import static org.apache.james.data.UsersRepositoryModuleChooser.Implementation.DEFAULT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
import org.apache.james.JamesServerBuilder;
import org.apache.james.JamesServerExtension;
import org.apache.james.MemoryJamesConfiguration;
import org.apache.james.MemoryJamesServerMain;
import org.apache.james.webadmin.integration.WebAdminServerIntegrationImmutableTest;
+import org.apache.james.webadmin.routes.HealthCheckRoutes;
+import org.eclipse.jetty.http.HttpStatus;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
class MemoryWebAdminServerIntegrationImmutableTest extends WebAdminServerIntegrationImmutableTest {
@@ -40,4 +47,20 @@ class MemoryWebAdminServerIntegrationImmutableTest extends WebAdminServerIntegra
.server(MemoryJamesServerMain::createServer)
.lifeCycle(PER_CLASS)
.build();
+
+ @Test
+ void healthCheckComponentsList() {
+ List<String> listComponentNames =
+ when()
+ .get(HealthCheckRoutes.HEALTHCHECK)
+ .then()
+ .statusCode(HttpStatus.OK_200)
+ .extract()
+ .body()
+ .jsonPath()
+ .getList("checks.componentName", String.class);
+
+ assertThat(listComponentNames).containsOnly("Guice application lifecycle", "MailReceptionCheck",
+ "EventDeadLettersHealthCheck", "EmptyErrorMailRepository", "MessageFastViewProjection");
+ }
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/04: JAMES-3817 Implement health check for RabbitMQ dead letter queue of mail queue
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 709609759f56425e5d42840ff11b9f55ca1a40e7
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Tue Sep 20 16:13:41 2022 +0700
JAMES-3817 Implement health check for RabbitMQ dead letter queue of mail queue
Degraded if dead letter queue length is not equal to 0.
---
.../modules/queue/rabbitmq/RabbitMQModule.java | 5 +-
...abbitMQMailQueueDeadLetterQueueHealthCheck.java | 66 ++++++++++
...tMQMailQueueDeadLetterQueueHealthCheckTest.java | 133 +++++++++++++++++++++
3 files changed, 203 insertions(+), 1 deletion(-)
diff --git a/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java b/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
index 712354a21e..7d041c2db8 100644
--- a/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
+++ b/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
@@ -37,6 +37,7 @@ import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.rabbitmq.RabbitMQMailQueue;
+import org.apache.james.queue.rabbitmq.RabbitMQMailQueueDeadLetterQueueHealthCheck;
import org.apache.james.queue.rabbitmq.RabbitMQMailQueueFactory;
import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
import org.apache.james.utils.InitializationOperation;
@@ -65,7 +66,9 @@ public class RabbitMQModule extends AbstractModule {
protected void configure() {
bind(SimpleConnectionPool.class).in(Scopes.SINGLETON);
- Multibinder.newSetBinder(binder(), HealthCheck.class).addBinding().to(RabbitMQHealthCheck.class);
+ Multibinder<HealthCheck> healthCheckMultiBinder = Multibinder.newSetBinder(binder(), HealthCheck.class);
+ healthCheckMultiBinder.addBinding().to(RabbitMQHealthCheck.class);
+ healthCheckMultiBinder.addBinding().to(RabbitMQMailQueueDeadLetterQueueHealthCheck.class);
Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(SpoolerReconnectionHandler.class);
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheck.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheck.java
new file mode 100644
index 0000000000..37d0b3a068
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheck.java
@@ -0,0 +1,66 @@
+/****************************************************************
+ * 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.queue.rabbitmq;
+
+import static org.apache.james.queue.api.MailQueueFactory.SPOOL;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
+import org.apache.james.backends.rabbitmq.RabbitMQManagementAPI;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+public class RabbitMQMailQueueDeadLetterQueueHealthCheck implements HealthCheck {
+ public static final MailQueueName JAMES_MAIL_QUEUE_NAME = MailQueueName.fromString(SPOOL.asString());
+ public static final ComponentName COMPONENT_NAME = new ComponentName("RabbitMQMailQueueDeadLetterQueueHealthCheck");
+ private static final String DEFAULT_VHOST = "/";
+
+ private final RabbitMQConfiguration configuration;
+ private final RabbitMQManagementAPI api;
+
+ @Inject
+ public RabbitMQMailQueueDeadLetterQueueHealthCheck(RabbitMQConfiguration configuration) {
+ this.configuration = configuration;
+ this.api = RabbitMQManagementAPI.from(configuration);
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ return Mono.fromCallable(() -> api.queueDetails(configuration.getVhost().orElse(DEFAULT_VHOST), JAMES_MAIL_QUEUE_NAME.toDeadLetterQueueName()))
+ .map(queueDetails -> {
+ if (queueDetails.getQueueLength() != 0) {
+ return Result.degraded(COMPONENT_NAME, "RabbitMQ dead letter queue of the mail queue contain messages. This might indicate transient failure on mail processing.");
+ }
+ return Result.healthy(COMPONENT_NAME);
+ })
+ .onErrorResume(e -> Mono.just(Result.unhealthy(COMPONENT_NAME, "Error checking RabbitMQMailQueueDeadLetterQueueHealthCheck", e)))
+ .subscribeOn(Schedulers.boundedElastic());
+ }
+}
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheckTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheckTest.java
new file mode 100644
index 0000000000..f32bd40a4e
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheckTest.java
@@ -0,0 +1,133 @@
+/****************************************************************
+ * 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.queue.rabbitmq;
+
+import static com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN;
+import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
+import static org.apache.james.backends.rabbitmq.Constants.DIRECT_EXCHANGE;
+import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
+import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
+import static org.apache.james.backends.rabbitmq.RabbitMQFixture.EXCHANGE_NAME;
+import static org.apache.james.backends.rabbitmq.RabbitMQFixture.ROUTING_KEY;
+import static org.apache.james.backends.rabbitmq.RabbitMQFixture.awaitAtMostOneMinute;
+import static org.apache.james.queue.rabbitmq.RabbitMQMailQueueDeadLetterQueueHealthCheck.JAMES_MAIL_QUEUE_NAME;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.james.backends.rabbitmq.DockerRabbitMQ;
+import org.apache.james.backends.rabbitmq.RabbitMQExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import com.google.common.collect.ImmutableMap;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+class RabbitMQMailQueueDeadLetterQueueHealthCheckTest {
+ @RegisterExtension
+ RabbitMQExtension rabbitMQExtension = RabbitMQExtension.singletonRabbitMQ()
+ .isolationPolicy(RabbitMQExtension.IsolationPolicy.STRONG);
+
+ public static final ImmutableMap<String, Object> NO_QUEUE_DECLARE_ARGUMENTS = ImmutableMap.of();
+
+ private Connection connection;
+ private Channel channel;
+ private RabbitMQMailQueueDeadLetterQueueHealthCheck testee;
+
+ @BeforeEach
+ void setup(DockerRabbitMQ rabbitMQ) throws IOException, TimeoutException, URISyntaxException {
+ ConnectionFactory connectionFactory = rabbitMQ.connectionFactory();
+ connectionFactory.setNetworkRecoveryInterval(1000);
+ connection = connectionFactory.newConnection();
+ channel = connection.createChannel();
+ testee = new RabbitMQMailQueueDeadLetterQueueHealthCheck(rabbitMQ.getConfiguration());
+ }
+
+ @AfterEach
+ void tearDown(DockerRabbitMQ rabbitMQ) throws Exception {
+ closeQuietly(connection, channel);
+ rabbitMQ.reset();
+ }
+
+ @Test
+ void healthCheckShouldReturnUnhealthyWhenRabbitMQIsDown() throws Exception {
+ rabbitMQExtension.getRabbitMQ().stopApp();
+
+ assertThat(testee.check().block().isUnHealthy()).isTrue();
+ }
+
+ @Test
+ void healthCheckShouldReturnHealthyWhenDeadLetterQueueIsEmpty() throws Exception {
+ createDeadLetterQueue(channel);
+
+ assertThat(testee.check().block().isHealthy()).isTrue();
+ }
+
+ @Test
+ void healthCheckShouldReturnDegradedWhenDeadLetterQueueIsNotEmpty() throws Exception {
+ createDeadLetterQueue(channel);
+ publishAMessage(channel);
+ publishAMessage(channel);
+
+ awaitAtMostOneMinute.until(() -> testee.check().block().isDegraded());
+ }
+
+ @Test
+ void healthCheckShouldReturnUnhealthyWhenThereIsNoDeadLetterQueue() {
+ assertThat(testee.check().block().isUnHealthy()).isTrue();
+ }
+
+ private void createDeadLetterQueue(Channel channel) throws IOException {
+ channel.exchangeDeclare(EXCHANGE_NAME, DIRECT_EXCHANGE, DURABLE);
+ channel.queueDeclare(JAMES_MAIL_QUEUE_NAME.toDeadLetterQueueName(), DURABLE, !EXCLUSIVE, AUTO_DELETE, NO_QUEUE_DECLARE_ARGUMENTS).getQueue();
+ channel.queueBind(JAMES_MAIL_QUEUE_NAME.toDeadLetterQueueName(), EXCHANGE_NAME, ROUTING_KEY);
+ }
+
+ private void publishAMessage(Channel channel) throws IOException {
+ AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder()
+ .deliveryMode(PERSISTENT_TEXT_PLAIN.getDeliveryMode())
+ .priority(PERSISTENT_TEXT_PLAIN.getPriority())
+ .contentType(PERSISTENT_TEXT_PLAIN.getContentType())
+ .build();
+
+ channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, basicProperties, "Hello, world!".getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void closeQuietly(AutoCloseable... closeables) {
+ Arrays.stream(closeables).forEach(this::closeQuietly);
+ }
+
+ private void closeQuietly(AutoCloseable closeable) {
+ try {
+ closeable.close();
+ } catch (Exception e) {
+ //ignore error
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/04: JAMES-3817 RabbitMQManagementAPI should parse the queue length property (`messages`) when getting Queue details
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit ac4061bd61856b9c5ef837a85f742991224b7e77
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Tue Sep 20 15:01:19 2022 +0700
JAMES-3817 RabbitMQManagementAPI should parse the queue length property (`messages`) when getting Queue details
---
.../backends/rabbitmq/RabbitMQManagementAPI.java | 7 +++++++
.../apache/james/backends/rabbitmq/RabbitMQTest.java | 20 ++++++++++++++++++++
2 files changed, 27 insertions(+)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
index 59488fbc18..9b7dc78089 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQManagementAPI.java
@@ -125,6 +125,9 @@ public interface RabbitMQManagementAPI {
@JsonProperty("consumer_details")
List<ConsumerDetails> consumerDetails;
+ @JsonProperty("messages")
+ long queueLength;
+
public String getName() {
return name;
}
@@ -152,6 +155,10 @@ public interface RabbitMQManagementAPI {
public List<ConsumerDetails> getConsumerDetails() {
return consumerDetails;
}
+
+ public long getQueueLength() {
+ return queueLength;
+ }
}
class ConsumerDetails {
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQTest.java
index 55d0e372ed..b4c025742c 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQTest.java
@@ -117,6 +117,26 @@ class RabbitMQTest {
awaitAtMostOneMinute.until(() -> messageReceived(channel, queueName));
}
+ @Test
+ void getQueueLengthShouldReturnEmptyWhenEmptyQueue() throws Exception {
+ String queueName = createQueue(channel);
+
+ awaitAtMostOneMinute.until(() -> rabbitMQExtension.managementAPI()
+ .queueDetails("/", queueName)
+ .getQueueLength() == 0);
+ }
+
+ @Test
+ void getQueueLengthShouldReturnExactlyNumberOfMessagesInQueue() throws Exception {
+ String queueName = createQueue(channel);
+ publishAMessage(channel);
+ publishAMessage(channel);
+
+ awaitAtMostOneMinute.until(() -> rabbitMQExtension.managementAPI()
+ .queueDetails("/", queueName)
+ .getQueueLength() == 2);
+ }
+
@Test
void demonstrateDurability(DockerRabbitMQ rabbitMQ) throws Exception {
String queueName = createQueue(channel);
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org