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 2023/08/18 08:39:16 UTC
[james-project] branch master updated: JAMES-3693 RedisHealthCheck (#1671)
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
The following commit(s) were added to refs/heads/master by this push:
new 38ed64d6fc JAMES-3693 RedisHealthCheck (#1671)
38ed64d6fc is described below
commit 38ed64d6fc68b4e076d421c459b9217c28001d71
Author: Trần Hồng Quân <55...@users.noreply.github.com>
AuthorDate: Fri Aug 18 15:39:09 2023 +0700
JAMES-3693 RedisHealthCheck (#1671)
---
.../rate/limiter/redis/RedisHealthCheck.scala | 64 +++++++++++++
.../rate/limiter/redis/RedisRateLimiter.scala | 13 ++-
.../org/apache/james/rate/limiter/DockerRedis.java | 15 +++
.../james/rate/limiter/RedisHealthCheckTest.scala | 102 +++++++++++++++++++++
4 files changed, 192 insertions(+), 2 deletions(-)
diff --git a/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisHealthCheck.scala b/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisHealthCheck.scala
new file mode 100644
index 0000000000..93690aa579
--- /dev/null
+++ b/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisHealthCheck.scala
@@ -0,0 +1,64 @@
+/****************************************************************
+ * 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.rate.limiter.redis
+
+import java.time.Duration
+
+import io.lettuce.core.api.StatefulConnection
+import io.lettuce.core.cluster.RedisClusterClient
+import io.lettuce.core.codec.StringCodec
+import io.lettuce.core.{RedisClient, RedisURI}
+import javax.inject.Inject
+import org.apache.james.core.healthcheck.{ComponentName, HealthCheck, Result}
+import org.reactivestreams.Publisher
+import reactor.core.scala.publisher.SMono
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.jdk.CollectionConverters._
+import scala.jdk.FutureConverters._
+
+class RedisHealthCheck @Inject()(redisConfiguration: RedisRateLimiterConfiguration) extends HealthCheck {
+ private val redisComponent: ComponentName = new ComponentName("Redis")
+ private val healthcheckTimeout = Duration.ofSeconds(3)
+
+ override def componentName(): ComponentName = redisComponent
+
+ override def check(): Publisher[Result] =
+ connectRedis()
+ .`then`(SMono.just(Result.healthy(redisComponent)))
+ .onErrorResume(_ => SMono.just(Result.degraded(redisComponent, "Can not connect to Redis.")))
+
+ private def connectRedis(): SMono[StatefulConnection[String, String]] =
+ if (redisConfiguration.isCluster) {
+ val redisUris = redisConfiguration.redisURI.value.asJava
+ redisUris.forEach(redisUri => redisUri.setTimeout(healthcheckTimeout))
+ val redisClusterClient = RedisClusterClient.create(redisUris)
+
+ SMono.fromFuture(redisClusterClient.connectAsync(StringCodec.UTF8).asScala)
+ .doOnTerminate(() => redisClusterClient.shutdownAsync())
+ } else {
+ val redisUri: RedisURI = redisConfiguration.redisURI.value.last
+ redisUri.setTimeout(healthcheckTimeout)
+ val redisClient = RedisClient.create(redisUri)
+
+ SMono.fromFuture(redisClient.connectAsync(StringCodec.UTF8, redisUri).asScala)
+ .doOnTerminate(() => redisClient.shutdownAsync())
+ }
+}
diff --git a/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisRateLimiter.scala b/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisRateLimiter.scala
index 26a0d7b8ed..f04b86f0ce 100644
--- a/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisRateLimiter.scala
+++ b/server/mailet/rate-limiter-redis/src/main/scala/org/apache/james/rate/limiter/redis/RedisRateLimiter.scala
@@ -21,13 +21,15 @@ package org.apache.james.rate.limiter.redis
import java.time.Duration
-import com.google.inject.{AbstractModule, Provides}
+import com.google.inject.multibindings.Multibinder
+import com.google.inject.{AbstractModule, Provides, Scopes}
import es.moki.ratelimitj.core.limiter.request.{AbstractRequestRateLimiterFactory, ReactiveRequestRateLimiter, RequestLimitRule}
import es.moki.ratelimitj.redis.request.{RedisClusterRateLimiterFactory, RedisSlidingWindowRequestRateLimiter, RedisRateLimiterFactory => RedisSingleInstanceRateLimitjFactory}
import io.lettuce.core.RedisClient
import io.lettuce.core.cluster.RedisClusterClient
import io.lettuce.core.resource.ClientResources
import javax.inject.Inject
+import org.apache.james.core.healthcheck.HealthCheck
import org.apache.james.rate.limiter.api.Increment.Increment
import org.apache.james.rate.limiter.api.{AcceptableRate, RateExceeded, RateLimiter, RateLimiterFactory, RateLimitingKey, RateLimitingResult, Rule, Rules}
import org.apache.james.util.concurrent.NamedThreadFactory
@@ -38,10 +40,17 @@ import reactor.core.scala.publisher.SMono
import scala.jdk.CollectionConverters._
class RedisRateLimiterModule() extends AbstractModule {
- override def configure(): Unit =
+ override def configure(): Unit = {
bind(classOf[RateLimiterFactory])
.to(classOf[RedisRateLimiterFactory])
+ bind(classOf[RedisRateLimiterFactory]).in(Scopes.SINGLETON)
+
+ Multibinder.newSetBinder(binder(), classOf[HealthCheck])
+ .addBinding()
+ .to(classOf[RedisHealthCheck])
+ }
+
@Provides
def provideConfig(propertiesProvider: PropertiesProvider): RedisRateLimiterConfiguration =
RedisRateLimiterConfiguration.from(propertiesProvider.getConfiguration("redis"))
diff --git a/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/DockerRedis.java b/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/DockerRedis.java
index 1dc0ceb932..139a4cf0fb 100644
--- a/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/DockerRedis.java
+++ b/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/DockerRedis.java
@@ -77,6 +77,21 @@ public class DockerRedis {
container.stop();
}
+ public void pause() {
+ container.getDockerClient().pauseContainerCmd(container.getContainerId()).exec();
+ }
+
+ public void unPause() {
+ container.getDockerClient().unpauseContainerCmd(container.getContainerId()).exec();
+ }
+
+ public boolean isPaused() {
+ return container.getDockerClient().inspectContainerCmd(container.getContainerId())
+ .exec()
+ .getState()
+ .getPaused();
+ }
+
public RedisCommands<String, String> createClient() {
return RedisClient.create(redisURI().toString())
.connect().sync();
diff --git a/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/RedisHealthCheckTest.scala b/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/RedisHealthCheckTest.scala
new file mode 100644
index 0000000000..aa950d8747
--- /dev/null
+++ b/server/mailet/rate-limiter-redis/src/test/java/org/apache/james/rate/limiter/RedisHealthCheckTest.scala
@@ -0,0 +1,102 @@
+/****************************************************************
+ * 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.rate.limiter
+
+import org.apache.james.core.healthcheck.Result
+import org.apache.james.rate.limiter.redis.{RedisHealthCheck, RedisRateLimiterConfiguration}
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.SoftAssertions
+import org.junit.jupiter.api.extension.ExtendWith
+import org.junit.jupiter.api.{AfterEach, BeforeEach, Test}
+import reactor.core.scala.publisher.SMono
+
+@ExtendWith(Array(classOf[RedisExtension]))
+class RedisHealthCheckTest {
+
+ var redisHealthCheck: RedisHealthCheck = _
+
+ @BeforeEach
+ def setup(redis: DockerRedis): Unit = {
+ val redisConfiguration: RedisRateLimiterConfiguration = RedisRateLimiterConfiguration.from(redis.redisURI().toString, isCluster = false)
+
+ redisHealthCheck = new RedisHealthCheck(redisConfiguration)
+ }
+
+ @AfterEach
+ def cleaner(redis: DockerRedis): Unit =
+ if (redis.isPaused) {
+ redis.unPause()
+ }
+
+ @Test
+ def checkShouldReturnHealthyWhenRedisIsRunning(): Unit = {
+ val result: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ assertThat(result.isHealthy).isTrue
+ }
+
+ @Test
+ def multipleCheckInShortPeriodShouldReturnHealthyWhenRedisIsRunning(): Unit = {
+ val check1: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+ val check2: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+ val check3: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ SoftAssertions.assertSoftly(softly => {
+ softly.assertThat(check1.isHealthy).isTrue
+ softly.assertThat(check2.isHealthy).isTrue
+ softly.assertThat(check3.isHealthy).isTrue
+ })
+ }
+
+ @Test
+ def checkShouldReturnDegradedWhenRedisIsDown(redis: DockerRedis): Unit = {
+ redis.pause()
+ val result: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ assertThat(result.isDegraded).isTrue
+ }
+
+ @Test
+ def checkShouldReturnHealthyWhenRedisIsRecovered(redis: DockerRedis): Unit = {
+ redis.pause()
+ redis.unPause()
+ val result: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ assertThat(result.isHealthy).isTrue
+ }
+
+ @Test
+ def multipleCheckInShortPeriodShouldReturnMixedResultWhenRedisIsUnstable(redis: DockerRedis): Unit = {
+ val check1: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ redis.pause()
+ val check2: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ redis.unPause()
+ val check3: Result = SMono.fromPublisher(redisHealthCheck.check()).block()
+
+ SoftAssertions.assertSoftly(softly => {
+ softly.assertThat(check1.isHealthy).isTrue
+ softly.assertThat(check2.isDegraded).isTrue
+ softly.assertThat(check3.isHealthy).isTrue
+ })
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org