Added sentinel support (#9)

bfanyuk · web-flow · commit 5df70c48cbc2 · 2020-10-06T15:36:09.000+02:00
#9 Add Redis sentinel support.
diff --git a/README.md b/README.md
@@ -58,6 +58,16 @@ Below you can find a sample `log4j2.xml` snippet employing `RedisAppender`.
 </Configuration>
 ```
 
+If you have Sentinel cluster please configure RedisAppender like this:
+```xml
+<RedisAppender name="REDIS"
+               key="log4j2-messages"
+               sentinelNodes="localhost:63791,localhost:63792"
+               sentinelMaster="mymaster">
+...
+</RedisAppender>
+```   
+
 `RedisAppender` is configured with the following parameters:
 
 | Parameter Name | Type | Default | Description |
@@ -66,6 +76,8 @@ Below you can find a sample `log4j2.xml` snippet employing `RedisAppender`.
 | `key` | String | | Redis queue key |
 | `host` | String | `localhost` | Redis host|
 | `port` | int | 6379 | Redis port |
+| `sentinelNodes` | String | `localhost:63791,localhost:63792` | Redis sentinel nodes as comma-separated list. If specified, `host` and `port` parameters are ignored. |
+| `sentinelMaster` | String | `mymaster` | Redis sentinel master name |
 | `password` | String | `null` | Redis password |
 | `connectionTimeoutSeconds` | int | 2 | initial connection timeout in seconds |
 | `socketTimeoutSeconds` | int | 2 | socket timeout in seconds |
@@ -209,6 +221,7 @@ Contributors
 
 - [t9t](https://github.com/t9t)
 - [Yaroslav Skopets](https://github.com/yskopets)
+- [Boris Faniuk](https://github.com/bfanyuk)
 
 # License
 
diff --git a/appender/src/main/java/com/vlkan/log4j2/redis/appender/RedisAppender.java b/appender/src/main/java/com/vlkan/log4j2/redis/appender/RedisAppender.java
@@ -17,14 +17,20 @@
 import org.apache.logging.log4j.util.Strings;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.JedisSentinelPool;
 import redis.clients.jedis.Protocol;
 import redis.clients.jedis.exceptions.JedisConnectionException;
+import redis.clients.jedis.util.Pool;
 
 import java.io.Serializable;
 import java.nio.charset.Charset;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static com.vlkan.log4j2.redis.appender.Helpers.requireArgument;
 import static java.util.Objects.requireNonNull;
+import static org.apache.logging.log4j.util.Strings.isNotBlank;
 
 @Plugin(name = "RedisAppender",
         category = Core.CATEGORY_NAME,
@@ -56,13 +62,17 @@ public class RedisAppender implements Appender {
 
     private final boolean debugEnabled;
 
+    private final String sentinelNodes;
+
+    private final String sentinelMaster;
+
     private final RedisConnectionPoolConfig poolConfig;
 
     private final DebugLogger logger;
 
     private final RedisThrottler throttler;
 
-    private volatile JedisPool jedisPool;
+    private volatile Pool<Jedis> jedisPool;
 
     private volatile State state;
 
@@ -81,6 +91,8 @@ private RedisAppender(Builder builder) {
         this.socketTimeoutSeconds = builder.socketTimeoutSeconds;
         this.ignoreExceptions = builder.ignoreExceptions;
         this.debugEnabled = builder.debugEnabled;
+        this.sentinelNodes = builder.sentinelNodes;
+        this.sentinelMaster = builder.sentinelMaster;
         this.poolConfig = builder.poolConfig;
         this.logger = new DebugLogger(RedisAppender.class, debugEnabled);
         this.throttler = new RedisThrottler(builder.getThrottlerConfig(), this, ignoreExceptions, debugEnabled);
@@ -201,19 +213,37 @@ private void connect() {
         logger.debug("connecting");
         int connectionTimeoutMillis = 1_000 * connectionTimeoutSeconds;
         int socketTimeoutMillis = 1_000 * socketTimeoutSeconds;
-        jedisPool = new JedisPool(
-                poolConfig.getJedisPoolConfig(),
-                host,
-                port,
-                connectionTimeoutMillis,
-                socketTimeoutMillis,
-                password,
-                0,          // database
-                null,       // clientName
-                false,      // ssl
-                null,       // sslSocketFactory
-                null,       // sslParameters,
-                null);      // hostnameVerifier
+        boolean sentinel = isNotBlank(sentinelNodes);
+        if (sentinel) {
+            Set<String> sentinelNodesAsSet = Stream.of(sentinelNodes.split("\\s*,\\s*"))
+                    .filter(Strings::isNotBlank)
+                    .collect(Collectors.toSet());
+            jedisPool = new JedisSentinelPool(
+                    sentinelMaster,
+                    sentinelNodesAsSet,
+                    poolConfig.getJedisPoolConfig(),
+                    connectionTimeoutMillis,
+                    socketTimeoutMillis,
+                    password,
+                    0,          // database
+                    null        // clientName
+            );
+        } else {
+            jedisPool = new JedisPool(
+                    poolConfig.getJedisPoolConfig(),
+                    host,
+                    port,
+                    connectionTimeoutMillis,
+                    socketTimeoutMillis,
+                    password,
+                    0,          // database
+                    null,       // clientName
+                    false,      // ssl
+                    null,       // sslSocketFactory
+                    null,       // sslParameters,
+                    null       // hostnameVerifier
+            );
+        }
     }
 
     private void disconnect() {
@@ -296,6 +326,12 @@ public static class Builder implements org.apache.logging.log4j.core.util.Builde
         @PluginBuilderAttribute
         private boolean debugEnabled = false;
 
+        @PluginBuilderAttribute
+        private String sentinelNodes;
+
+        @PluginBuilderAttribute
+        private String sentinelMaster;
+
         @PluginElement("RedisConnectionPoolConfig")
         private RedisConnectionPoolConfig poolConfig = RedisConnectionPoolConfig.newBuilder().build();
 
@@ -414,6 +450,24 @@ public Builder setDebugEnabled(boolean debugEnabled) {
             return this;
         }
 
+        public String getSentinelNodes() {
+            return sentinelNodes;
+        }
+
+        public Builder setSentinelNodes(String sentinelNodes) {
+            this.sentinelNodes = sentinelNodes;
+            return this;
+        }
+
+        public String getSentinelMaster() {
+            return sentinelMaster;
+        }
+
+        public Builder setSentinelMaster(String sentinelMaster) {
+            this.sentinelMaster = sentinelMaster;
+            return this;
+        }
+
         public RedisConnectionPoolConfig getPoolConfig() {
             return poolConfig;
         }
@@ -446,6 +500,10 @@ private void check() {
             requireArgument(Strings.isNotBlank(key), "blank key");
             requireArgument(Strings.isNotBlank(host), "blank host");
             requireArgument(port > 0, "expecting: port > 0, found: %d", port);
+            if (sentinelNodes != null) {
+                requireArgument(Strings.isNotBlank(sentinelNodes), "blank sentinel nodes");
+                requireArgument(Strings.isNotBlank(sentinelMaster), "blank sentinel master");
+            }
             requireArgument(connectionTimeoutSeconds > 0, "expecting: connectionTimeoutSeconds > 0, found: %d", connectionTimeoutSeconds);
             requireArgument(socketTimeoutSeconds > 0, "expecting: socketTimeoutSeconds > 0, found: %d", socketTimeoutSeconds);
             requireNonNull(poolConfig, "poolConfig");
diff --git a/appender/src/main/java/com/vlkan/log4j2/redis/appender/RedisThrottler.java b/appender/src/main/java/com/vlkan/log4j2/redis/appender/RedisThrottler.java
@@ -1,6 +1,8 @@
 package com.vlkan.log4j2.redis.appender;
 
 import com.vlkan.log4j2.redis.appender.guava.GuavaRateLimiter;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
 
 import javax.management.InstanceAlreadyExistsException;
 import javax.management.JMX;
@@ -73,9 +75,12 @@ class RedisThrottler implements AutoCloseable {
     private ObjectName createJmxBeanName() {
         String beanName = config.getJmxBeanName();
         if (beanName == null) {
+            LoggerContext loggerContext = appender.getConfig().getLoggerContext();
+            if (loggerContext == null)
+                loggerContext = (LoggerContext) LogManager.getContext(false);
             beanName = String.format(
                     "org.apache.logging.log4j2:type=%s,component=Appenders,name=%s,subtype=RedisThrottler",
-                    appender.getConfig().getLoggerContext().getName(),
+                    loggerContext.getName(),
                     appender.getName());
         }
         try {
diff --git a/appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisAppenderSentinelTest.java b/appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisAppenderSentinelTest.java
@@ -0,0 +1,154 @@
+package com.vlkan.log4j2.redis.appender;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.Appender;
+import org.junit.ClassRule;
+import org.junit.ComparisonFailure;
+import org.junit.Test;
+import org.junit.rules.RuleChain;
+import redis.clients.jedis.Jedis;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RedisAppenderSentinelTest {
+
+    private static final DebugLogger LOGGER = new DebugLogger(RedisAppenderSentinelTest.class, true);
+
+    private static final Random RANDOM = new Random(0);
+
+    private static final int MIN_MESSAGE_COUNT = 1;
+
+    private static final int MAX_MESSAGE_COUNT = 100;
+
+    private static final String REDIS_KEY = "log4j2-messages";
+
+    private static final String REDIS_HOST = "localhost";
+
+    static final String REDIS_PASSWORD = "toosecret";
+
+    static final int REDIS_PORT = 63790;
+
+    static final int SENTINEL_PORT = 63792;
+
+    static final String MASTER_NAME = "mymaster";
+
+    private static final RedisServerResource REDIS_SERVER_RESOURCE = new RedisServerResource(REDIS_PORT, REDIS_PASSWORD);
+
+    private static final RedisSentinelResource REDIS_SENTINEL_RESOURCE = new RedisSentinelResource(SENTINEL_PORT, REDIS_PORT, MASTER_NAME);
+
+    private static final RedisClientResource REDIS_CLIENT_RESOURCE = new RedisClientResource(REDIS_HOST, REDIS_PORT, REDIS_PASSWORD);
+
+    static final URI CONFIG_FILE_URI = createConfigFileUri();
+
+    private static final LoggerContextResource LOGGER_CONTEXT_RESOURCE = new LoggerContextResource(CONFIG_FILE_URI);
+
+    @ClassRule
+    public static final RuleChain RULE_CHAIN = RuleChain
+            .outerRule(REDIS_SERVER_RESOURCE)
+            .around(REDIS_SENTINEL_RESOURCE)
+            .around(REDIS_CLIENT_RESOURCE)
+            .around(LOGGER_CONTEXT_RESOURCE);
+
+    private static class LogMessage {
+
+        private static final Level[] LEVELS = Level.values();
+
+        private static final int MIN_MESSAGE_LENGTH = 10;
+
+        private static final int MAX_MESSAGE_LENGTH = 100;
+
+        private final Level level;
+
+        private final String message;
+
+        private static AtomicInteger COUNTER = new AtomicInteger(0);
+
+        private LogMessage(Level level, String message) {
+            this.level = level;
+            this.message = message;
+        }
+
+        private static LogMessage createRandom() {
+            int levelIndex = RANDOM.nextInt(LEVELS.length);
+            Level level = LEVELS[levelIndex];
+            int messageLength = MIN_MESSAGE_LENGTH + RANDOM.nextInt(MAX_MESSAGE_LENGTH - MIN_MESSAGE_LENGTH);
+            String prefix = String.format("[%d] ", COUNTER.getAndIncrement());
+            StringBuilder messageBuilder = new StringBuilder(prefix);
+            while (messageBuilder.length() < messageLength) {
+                char messageChar = (char) RANDOM.nextInt(Character.MAX_VALUE);
+                if (Character.isLetterOrDigit(messageChar)) {
+                    messageBuilder.append(messageChar);
+                }
+            }
+            String message = messageBuilder.toString();
+            return new LogMessage(level, message);
+        }
+
+        private static LogMessage[] createRandomArray(int count) {
+            LogMessage[] messages = new LogMessage[count];
+            for (int i = 0; i < count; i++) {
+                messages[i] = createRandom();
+            }
+            return messages;
+        }
+
+    }
+
+    private static URI createConfigFileUri() {
+        try {
+            return new URI("classpath:log4j2.RedisAppenderSentinelTest.xml");
+        } catch (URISyntaxException error) {
+            throw new RuntimeException("failed finding Log4j config", error);
+        }
+    }
+
+    @Test
+    public void test_messages_are_enqueued_to_redis() throws InterruptedException {
+
+        LOGGER.debug("creating the logger");
+        Logger logger = LOGGER_CONTEXT_RESOURCE.getLoggerContext().getLogger(RedisAppenderSentinelTest.class.getCanonicalName());
+
+        int expectedMessageCount = MIN_MESSAGE_COUNT + RANDOM.nextInt(MAX_MESSAGE_COUNT - MIN_MESSAGE_COUNT);
+        LOGGER.debug("logging %d messages", expectedMessageCount);
+        LogMessage[] expectedLogMessages = LogMessage.createRandomArray(expectedMessageCount);
+        for (LogMessage expectedLogMessage : expectedLogMessages) {
+            logger.log(expectedLogMessage.level, expectedLogMessage.message);
+        }
+
+        LOGGER.debug("waiting for throttler to kick in");
+        Thread.sleep(1_000);
+
+        LOGGER.debug("checking logged messages");
+        Jedis redisClient = REDIS_CLIENT_RESOURCE.getClient();
+        for (int messageIndex = 0; messageIndex < expectedMessageCount; messageIndex++) {
+            LogMessage expectedLogMessage = expectedLogMessages[messageIndex];
+            String expectedSerializedMessage = String.format("%s %s", expectedLogMessage.level, expectedLogMessage.message);
+            String actualSerializedMessage = redisClient.lpop(REDIS_KEY);
+            try {
+                assertThat(actualSerializedMessage).isEqualTo(expectedSerializedMessage);
+            } catch (ComparisonFailure comparisonFailure) {
+                String message = String.format("comparison failure (messageIndex=%d, messageCount=%d)", messageIndex, expectedMessageCount);
+                throw new RuntimeException(message, comparisonFailure);
+            }
+        }
+
+        Appender appender = LOGGER_CONTEXT_RESOURCE.getLoggerContext().getConfiguration().getAppender("REDIS");
+        assertThat(appender).isInstanceOf(RedisAppender.class);
+        RedisThrottlerJmxBean jmxBean = ((RedisAppender) appender).getJmxBean();
+        assertThat(jmxBean.getTotalEventCount()).isEqualTo(expectedMessageCount);
+        assertThat(jmxBean.getIgnoredEventCount()).isEqualTo(0);
+        assertThat(jmxBean.getEventRateLimitFailureCount()).isEqualTo(0);
+        assertThat(jmxBean.getByteRateLimitFailureCount()).isEqualTo(0);
+        assertThat(jmxBean.getUnavailableBufferSpaceFailureCount()).isEqualTo(0);
+        assertThat(jmxBean.getRedisPushSuccessCount()).isEqualTo(expectedMessageCount);
+        assertThat(jmxBean.getRedisPushFailureCount()).isEqualTo(0);
+
+    }
+
+}
diff --git a/appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisSentinelResource.java b/appender/src/test/java/com/vlkan/log4j2/redis/appender/RedisSentinelResource.java
diff --git a/appender/src/test/resources/log4j2.RedisAppenderSentinelTest.xml b/appender/src/test/resources/log4j2.RedisAppenderSentinelTest.xml
