diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml index 80baf8d1d2d4..82a3b96607b0 100644 --- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/pom.xml @@ -82,6 +82,11 @@ nifi-record-path 2.0.0-SNAPSHOT + + org.apache.nifi + nifi-ssl-context-service-api + provided + org.apache.nifi @@ -96,15 +101,15 @@ test - com.github.kstyrc - embedded-redis - 0.6 + org.testcontainers + testcontainers + ${testcontainers.version} test - org.apache.nifi - nifi-ssl-context-service-api - compile + org.testcontainers + junit-jupiter + test diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java index af9b61e7a067..30e6f7b00985 100644 --- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java @@ -35,6 +35,7 @@ import org.springframework.data.redis.connection.RedisStandaloneConfiguration; import org.springframework.data.redis.connection.jedis.JedisClientConfiguration; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.lang.Nullable; import redis.clients.jedis.JedisPoolConfig; import javax.net.ssl.SSLContext; @@ -134,7 +135,16 @@ public class RedisUtils { public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() .name("Password") .displayName("Password") - .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.") + .description("The password used to authenticate to the Redis server. See the 'requirepass' property in redis.conf.") + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .sensitive(true) + .build(); + + public static final PropertyDescriptor SENTINEL_PASSWORD = new PropertyDescriptor.Builder() + .name("Sentinel Password") + .displayName("Sentinel Password") + .description("The password used to authenticate to the Redis Sentinel server. See the 'requirepass' and 'sentinel sentinel-pass' properties in sentinel.conf.") .addValidator(StandardValidators.NON_BLANK_VALIDATOR) .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) .sensitive(true) @@ -275,6 +285,7 @@ public class RedisUtils { props.add(RedisUtils.CLUSTER_MAX_REDIRECTS); props.add(RedisUtils.SENTINEL_MASTER); props.add(RedisUtils.PASSWORD); + props.add(RedisUtils.SENTINEL_PASSWORD); props.add(RedisUtils.SSL_CONTEXT_SERVICE); props.add(RedisUtils.POOL_MAX_TOTAL); props.add(RedisUtils.POOL_MAX_IDLE); @@ -297,6 +308,7 @@ public static JedisConnectionFactory createConnectionFactory(final PropertyConte final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue(); final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger(); final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue(); + final String sentinelPassword = context.getProperty(RedisUtils.SENTINEL_PASSWORD).evaluateAttributeExpressions().getValue(); final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); final JedisPoolConfig poolConfig = createJedisPoolConfig(context); @@ -323,7 +335,7 @@ public static JedisConnectionFactory createConnectionFactory(final PropertyConte final String host = hostAndPortSplit[0].trim(); final Integer port = Integer.parseInt(hostAndPortSplit[1].trim()); final RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(host, port); - enrichRedisConfiguration(redisStandaloneConfiguration, dbIndex, password); + enrichRedisConfiguration(redisStandaloneConfiguration, dbIndex, password, sentinelPassword); connectionFactory = new JedisConnectionFactory(redisStandaloneConfiguration, jedisClientConfiguration); @@ -331,7 +343,7 @@ public static JedisConnectionFactory createConnectionFactory(final PropertyConte final String[] sentinels = connectionString.split("[,]"); final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue(); final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(getTrimmedValues(sentinels))); - enrichRedisConfiguration(sentinelConfiguration, dbIndex, password); + enrichRedisConfiguration(sentinelConfiguration, dbIndex, password, sentinelPassword); logger.info("Connecting to Redis in sentinel mode..."); logger.info("Redis master = " + sentinelMaster); @@ -347,7 +359,7 @@ public static JedisConnectionFactory createConnectionFactory(final PropertyConte final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger(); final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(getTrimmedValues(clusterNodes)); - enrichRedisConfiguration(clusterConfiguration, dbIndex, password); + enrichRedisConfiguration(clusterConfiguration, dbIndex, password, sentinelPassword); clusterConfiguration.setMaxRedirects(maxRedirects); logger.info("Connecting to Redis in clustered mode..."); @@ -373,13 +385,17 @@ private static List getTrimmedValues(final String[] values) { private static void enrichRedisConfiguration(final RedisConfiguration redisConfiguration, final Integer dbIndex, - final String password) { + @Nullable final String password, + @Nullable final String sentinelPassword) { if (redisConfiguration instanceof RedisConfiguration.WithDatabaseIndex) { ((RedisConfiguration.WithDatabaseIndex) redisConfiguration).setDatabase(dbIndex); } if (redisConfiguration instanceof RedisConfiguration.WithPassword) { ((RedisConfiguration.WithPassword) redisConfiguration).setPassword(RedisPassword.of(password)); } + if (redisConfiguration instanceof RedisConfiguration.SentinelConfiguration) { + ((RedisConfiguration.SentinelConfiguration) redisConfiguration).setSentinelPassword(RedisPassword.of(sentinelPassword)); + } } private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) { diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java index aec018954ef0..f14187f6edba 100644 --- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java @@ -30,14 +30,19 @@ import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.redis.RedisConnectionPool; +import org.apache.nifi.redis.testcontainers.RedisContainer; +import org.apache.nifi.redis.testcontainers.RedisReplicaContainer; +import org.apache.nifi.redis.testcontainers.RedisSentinelContainer; import org.apache.nifi.redis.util.RedisUtils; import org.apache.nifi.reporting.InitializationException; import org.apache.nifi.util.TestRunner; import org.apache.nifi.util.TestRunners; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import redis.embedded.RedisServer; +import org.junit.jupiter.api.io.TempDir; +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -46,6 +51,8 @@ import java.net.StandardSocketOptions; import java.nio.channels.SocketChannel; import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.HashMap; @@ -53,8 +60,10 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Consumer; import static org.apache.nifi.redis.util.RedisUtils.REDIS_CONNECTION_POOL; +import static org.apache.nifi.redis.util.RedisUtils.REDIS_MODE_SENTINEL; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -64,22 +73,156 @@ */ public class ITRedisDistributedMapCacheClientService { - private TestRedisProcessor proc; - private TestRunner testRunner; - private RedisServer redisServer; + public static final String CONTAINER_IMAGE_TAG = "redis:7.0.12-alpine"; + + private static final String masterName = "redisLeader"; + @TempDir + + private Path testDirectory; + + private final TestRedisProcessor proc = new TestRedisProcessor(); + private final TestRunner testRunner = TestRunners.newTestRunner(proc); + + private final List redisContainers = new ArrayList<>(); private RedisConnectionPoolService redisConnectionPool; - private RedisDistributedMapCacheClientService redisMapCacheClientService; - private int redisPort; - @BeforeEach - public void setup() throws IOException { - this.redisPort = getAvailablePort(); + @Test + public void testStandaloneRedis() throws InitializationException, IOException { + int redisPort = setupStandaloneRedis(null).port; + setUpRedisConnectionPool(portsToConnectionString(redisPort), pool -> { + // uncomment this to test using a different database index than the default 0 + // testRunner.setProperty(pool, RedisUtils.DATABASE, "1"); + }); + setupRedisMapCacheClientService(); + + executeProcessor(); + } + + @Test + public void testStandaloneRedisWithAuthentication() throws InitializationException, IOException { + final String redisPassword = "foobared"; + final int redisPort = setupStandaloneRedis(redisPassword).port; + setUpRedisConnectionPool(portsToConnectionString(redisPort), pool -> { + testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD, redisPassword); + }); + setupRedisMapCacheClientService(); + + executeProcessor(); + } + + @Test + public void testSentinelRedis() throws InitializationException, IOException { + RedisContainer redisMasterContainer = setupStandaloneRedis(null); + String masterHost = "127.0.0.1"; + int masterPort = redisMasterContainer.port; + setUpRedisReplica(masterHost, masterPort, null); + setUpRedisReplica(masterHost, masterPort, null); + + int sentinelAPort = setUpSentinel(masterHost, masterPort, null, 2, null).port; + int sentinelBPort = setUpSentinel(masterHost, masterPort, null, 2, null).port; + int sentinelCPort = setUpSentinel(masterHost, masterPort, null, 2, null).port; + + setUpRedisConnectionPool(portsToConnectionString(sentinelAPort, sentinelBPort, sentinelCPort), pool -> { + testRunner.setProperty(redisConnectionPool, RedisUtils.REDIS_MODE, REDIS_MODE_SENTINEL); + testRunner.setProperty(redisConnectionPool, RedisUtils.SENTINEL_MASTER, masterName); + }); + + setupRedisMapCacheClientService(); + + executeProcessor(); + } - this.redisServer = new RedisServer(redisPort); - redisServer.start(); + @Test + public void testSentinelRedisWithAuthentication() throws InitializationException, IOException { + String redisPassword = "t0p_53cr35"; + String sentinelPassword = "otherPassword"; + + RedisContainer redisMasterContainer = setupStandaloneRedis(redisPassword); + String masterHost = "127.0.0.1"; + int masterPort = redisMasterContainer.port; + setUpRedisReplica(masterHost, masterPort, redisPassword); + setUpRedisReplica(masterHost, masterPort, redisPassword); + + int sentinelAPort = setUpSentinel(masterHost, masterPort, redisPassword, 2, sentinelPassword).port; + int sentinelBPort = setUpSentinel(masterHost, masterPort, redisPassword, 2, sentinelPassword).port; + int sentinelCPort = setUpSentinel(masterHost, masterPort, redisPassword, 2, sentinelPassword).port; + + setUpRedisConnectionPool(portsToConnectionString(sentinelAPort, sentinelBPort, sentinelCPort), pool -> { + testRunner.setProperty(redisConnectionPool, RedisUtils.REDIS_MODE, REDIS_MODE_SENTINEL); + testRunner.setProperty(redisConnectionPool, RedisUtils.SENTINEL_MASTER, masterName); + + testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD, redisPassword); + testRunner.setProperty(redisConnectionPool, RedisUtils.SENTINEL_PASSWORD, sentinelPassword); + }); + setupRedisMapCacheClientService(); + + executeProcessor(); + } + + @AfterEach + public void teardown() { + if (redisConnectionPool != null) { + redisConnectionPool.onDisabled(); + } - proc = new TestRedisProcessor(); - testRunner = TestRunners.newTestRunner(proc); + redisContainers.forEach(RedisContainer::stop); + } + + private RedisContainer setupStandaloneRedis(final @Nullable String redisPassword) throws IOException { + int redisPort = getAvailablePort(); + + RedisContainer redisContainer = new RedisContainer(CONTAINER_IMAGE_TAG); + redisContainer.mountConfigurationFrom(testDirectory); + redisContainer.setPort(redisPort); + redisContainer.addPortBinding(redisPort, redisPort); + redisContainer.setPassword(redisPassword); + + redisContainers.add(redisContainer); + redisContainer.start(); + + return redisContainer; + } + + private RedisReplicaContainer setUpRedisReplica(final @NonNull String masterHost, + final int masterPort, + final @Nullable String redisPassword) throws IOException { + int replicaPort = getAvailablePort(); + + RedisReplicaContainer redisReplicaContainer = new RedisReplicaContainer(CONTAINER_IMAGE_TAG); + redisReplicaContainer.mountConfigurationFrom(testDirectory); + redisReplicaContainer.setPort(replicaPort); + redisReplicaContainer.addPortBinding(replicaPort, replicaPort); + redisReplicaContainer.setReplicaOf(masterHost, masterPort); + redisReplicaContainer.setPassword(redisPassword); + + redisContainers.add(redisReplicaContainer); + redisReplicaContainer.start(); + + return redisReplicaContainer; + } + + private RedisSentinelContainer setUpSentinel(final @NonNull String masterHost, + final int masterPort, + final @Nullable String redisPassword, + final int quorumSize, + final @Nullable String sentinelPassword) throws IOException { + int sentinelPort = getAvailablePort(); + + RedisSentinelContainer redisSentinelContainer = new RedisSentinelContainer(CONTAINER_IMAGE_TAG); + redisSentinelContainer.mountConfigurationFrom(testDirectory); + redisSentinelContainer.setPort(sentinelPort); + redisSentinelContainer.addPortBinding(sentinelPort, sentinelPort); + redisSentinelContainer.setMasterHost(masterHost); + redisSentinelContainer.setMasterPort(masterPort); + redisSentinelContainer.setMasterName(masterName); + redisSentinelContainer.setQuorumSize(quorumSize); + redisSentinelContainer.setPassword(redisPassword); + redisSentinelContainer.setSentinelPassword(sentinelPassword); + + redisContainers.add(redisSentinelContainer); + redisSentinelContainer.start(); + + return redisSentinelContainer; } private int getAvailablePort() throws IOException { @@ -90,41 +233,37 @@ private int getAvailablePort() throws IOException { } } - @AfterEach - public void teardown() throws IOException { - if (redisServer != null) { - redisServer.stop(); - } - } - - @Test - public void testStandaloneRedis() throws InitializationException { - try { - // create, configure, and enable the RedisConnectionPool service - redisConnectionPool = new RedisConnectionPoolService(); - testRunner.addControllerService("redis-connection-pool", redisConnectionPool); - testRunner.setProperty(redisConnectionPool, RedisUtils.CONNECTION_STRING, "localhost:" + redisPort); + private String portsToConnectionString(int... ports) { + StringBuilder stringBuilder = new StringBuilder(); - // uncomment this to test using a different database index than the default 0 - //testRunner.setProperty(redisConnectionPool, RedisUtils.DATABASE, "1"); + for (int i = 0; i < ports.length; i++) { + if (i > 0) { + stringBuilder.append(','); + } + stringBuilder.append("localhost"); + stringBuilder.append(':'); + stringBuilder.append(ports[i]); + } - // uncomment this to test using a password to authenticate to redis - //testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD, "foobared"); + return stringBuilder.toString(); + } - testRunner.enableControllerService(redisConnectionPool); + public void setUpRedisConnectionPool(String connectionString, Consumer initializer) throws InitializationException { + // create, configure, and enable the RedisConnectionPool service + redisConnectionPool = new RedisConnectionPoolService(); + testRunner.addControllerService("redis-connection-pool", redisConnectionPool); + testRunner.setProperty(redisConnectionPool, RedisUtils.CONNECTION_STRING, connectionString); - setupRedisMapCacheClientService(); - executeProcessor(); - } finally { - if (redisConnectionPool != null) { - redisConnectionPool.onDisabled(); - } + if (initializer != null) { + initializer.accept(redisConnectionPool); } + + testRunner.enableControllerService(redisConnectionPool); } private void setupRedisMapCacheClientService() throws InitializationException { // create, configure, and enable the RedisDistributedMapCacheClient service - redisMapCacheClientService = new RedisDistributedMapCacheClientService(); + RedisDistributedMapCacheClientService redisMapCacheClientService = new RedisDistributedMapCacheClientService(); testRunner.addControllerService("redis-map-cache-client", redisMapCacheClientService); testRunner.setProperty(redisMapCacheClientService, REDIS_CONNECTION_POOL, "redis-connection-pool"); testRunner.enableControllerService(redisMapCacheClientService); @@ -218,12 +357,12 @@ public void onTrigger(final ProcessContext context, final ProcessSession session assertEquals(value, cacheClient.get(keyThatDoesntExist, stringSerializer, stringDeserializer)); // verify atomic fetch returns the correct entry - final AtomicCacheEntry entry = cacheClient.fetch(key, stringSerializer, stringDeserializer); + final AtomicCacheEntry entry = cacheClient.fetch(key, stringSerializer, stringDeserializer); assertEquals(key, entry.getKey()); assertEquals(value, entry.getValue()); assertTrue(Arrays.equals(value.getBytes(StandardCharsets.UTF_8), entry.getRevision().orElse(null))); - final AtomicCacheEntry notLatestEntry = new AtomicCacheEntry<>(entry.getKey(), entry.getValue(), "not previous".getBytes(StandardCharsets.UTF_8)); + final AtomicCacheEntry notLatestEntry = new AtomicCacheEntry<>(entry.getKey(), entry.getValue(), "not previous".getBytes(StandardCharsets.UTF_8)); // verify atomic replace does not replace when previous value is not equal assertFalse(cacheClient.replace(notLatestEntry, stringSerializer, stringSerializer)); @@ -237,12 +376,12 @@ public void onTrigger(final ProcessContext context, final ProcessSession session // verify atomic replace does replace no value previous existed final String replaceKeyDoesntExist = key + "_REPLACE_DOES_NOT_EXIST"; - final AtomicCacheEntry entryDoesNotExist = new AtomicCacheEntry<>(replaceKeyDoesntExist, replacementValue, null); + final AtomicCacheEntry entryDoesNotExist = new AtomicCacheEntry<>(replaceKeyDoesntExist, replacementValue, null); assertTrue(cacheClient.replace(entryDoesNotExist, stringSerializer, stringSerializer)); assertEquals(replacementValue, cacheClient.get(replaceKeyDoesntExist, stringSerializer, stringDeserializer)); final int numToDelete = 2000; - for (int i=0; i < numToDelete; i++) { + for (int i = 0; i < numToDelete; i++) { cacheClient.put(key + "-" + i, value, stringSerializer, stringSerializer); } @@ -294,3 +433,4 @@ public String deserialize(byte[] input) throws DeserializationException, IOExcep } } } + diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java index 599e5ec6ab70..5315439f1fd5 100644 --- a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java @@ -23,18 +23,18 @@ import org.apache.nifi.components.state.StateProvider; import org.apache.nifi.components.state.StateProviderInitializationContext; import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.redis.service.ITRedisDistributedMapCacheClientService; +import org.apache.nifi.redis.testcontainers.RedisContainer; import org.apache.nifi.redis.util.RedisUtils; import org.apache.nifi.util.MockComponentLog; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import redis.embedded.RedisServer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; import javax.net.ssl.SSLContext; import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.StandardSocketOptions; -import java.nio.channels.SocketChannel; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -51,22 +51,21 @@ * NOTE: These test cases should be kept in-sync with AbstractTestStateProvider which is in the framework * and couldn't be extended here. */ +@Testcontainers public class ITRedisStateProvider { protected final String componentId = "111111111-1111-1111-1111-111111111111"; - private RedisServer redisServer; + @Container + public RedisContainer redisContainer = new RedisContainer(ITRedisDistributedMapCacheClientService.CONTAINER_IMAGE_TAG) + .withExposedPorts(6379); + private RedisStateProvider provider; @BeforeEach - public void setup() throws Exception { - final int redisPort = getAvailablePort(); - - this.redisServer = new RedisServer(redisPort); - redisServer.start(); - + public void setup() { final Map properties = new HashMap<>(); - properties.put(RedisUtils.CONNECTION_STRING, "localhost:" + redisPort); + properties.put(RedisUtils.CONNECTION_STRING, redisContainer.getHost() + ":" + redisContainer.getFirstMappedPort()); this.provider = createProvider(properties); } @@ -80,10 +79,6 @@ public void teardown() { provider.disable(); provider.shutdown(); } - - if (redisServer != null) { - redisServer.stop(); - } } public StateProvider getProvider() { @@ -173,7 +168,7 @@ public void testToMap() throws IOException { assertEquals(1, map.size()); assertEquals("value", map.get(key)); - provider.setState(Collections. emptyMap(), componentId); + provider.setState(Collections.emptyMap(), componentId); final StateMap stateMap = provider.getState(componentId); map = stateMap.toMap(); @@ -284,8 +279,8 @@ public Map getProperties() { } @Override - public Map getAllProperties() { - final Map propValueMap = new LinkedHashMap<>(); + public Map getAllProperties() { + final Map propValueMap = new LinkedHashMap<>(); for (final Map.Entry entry : properties.entrySet()) { propValueMap.put(entry.getKey().getName(), entry.getValue()); } @@ -321,13 +316,4 @@ private RedisStateProvider createProvider(final Map provider.enable(); return provider; } - - private int getAvailablePort() throws IOException { - try (SocketChannel socket = SocketChannel.open()) { - socket.setOption(StandardSocketOptions.SO_REUSEADDR, true); - socket.bind(new InetSocketAddress("localhost", 0)); - return socket.socket().getLocalPort(); - } - } - } diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java new file mode 100644 index 000000000000..24f5b68a98ce --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisContainer.java @@ -0,0 +1,113 @@ +/* + * 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.nifi.redis.testcontainers; + +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.testcontainers.containers.BindMode; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.utility.DockerImageName; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; + +public class RedisContainer extends GenericContainer { + + public static final int REDIS_PORT = 6379; + + public RedisContainer(@NonNull DockerImageName dockerImageName) { + super(dockerImageName); + } + + public RedisContainer(@NonNull String fullImageName) { + this(DockerImageName.parse(fullImageName)); + } + + public int port = REDIS_PORT; + + @Nullable + protected String password = null; + @Nullable + protected Path configurationMountDirectory = null; + + protected final List configurationOptions = new ArrayList<>(); + + public void setPassword(final @Nullable String password) { + this.password = password; + } + + public void setPort(final int port) { + this.port = port; + } + + public void mountConfigurationFrom(final Path mountDirectory) { + this.configurationMountDirectory = mountDirectory; + } + + public void addConfigurationOption(final String configurationOption) { + this.configurationOptions.add(configurationOption); + } + + protected void adjustConfiguration() { + addConfigurationOption("port " + port); + + if (password != null) { + addConfigurationOption("requirepass " + password); + } + } + + /** + * Sets up a static binding between a port on the host and one in the container. + * In order for auto-discovery mechanisms of Redis to work, 1-to-1 mapped ports are useful. + */ + public void addPortBinding(int hostPort, int containerPort) { + addFixedExposedPort(hostPort, containerPort); + } + + @Override + protected void configure() { + adjustConfiguration(); + + Path configurationFilePath = writeConfigurationFile().toAbsolutePath(); + String hostPath = configurationFilePath.toString(); + String containerPath = "/usr/local/etc/redis/redis.conf"; + addFileSystemBind(hostPath, containerPath, BindMode.READ_WRITE); + + setCommand(containerPath); + } + + protected Path writeConfigurationFile() { + try { + Path mountDirectory = this.configurationMountDirectory; + if (mountDirectory == null) { + mountDirectory = Files.createTempDirectory("redis-container-configuration"); + } + + Path configFile = mountDirectory.resolve("redis-" + UUID.randomUUID() + ".conf"); + Files.write(configFile, configurationOptions, StandardCharsets.UTF_8); + + return configFile; + } catch (IOException ioException) { + throw new IllegalStateException("Cannot start container because configuration could not be written", ioException); + } + } +} diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java new file mode 100644 index 000000000000..815fe33f8528 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisReplicaContainer.java @@ -0,0 +1,60 @@ +/* + * 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.nifi.redis.testcontainers; + +import org.springframework.lang.NonNull; +import org.testcontainers.utility.DockerImageName; + +public class RedisReplicaContainer extends RedisContainer { + + public RedisReplicaContainer(@NonNull DockerImageName dockerImageName) { + super(dockerImageName); + } + + public RedisReplicaContainer(@NonNull String fullImageName) { + this(DockerImageName.parse(fullImageName)); + } + + @NonNull + protected String masterHost = "localhost"; + protected int masterPort = REDIS_PORT; + + public void setMasterHost(@NonNull String masterHost) { + this.masterHost = masterHost; + } + + public void setMasterPort(int masterPort) { + this.masterPort = masterPort; + } + + public void setReplicaOf(@NonNull String masterHost, int masterPort) { + setMasterHost(masterHost); + setMasterPort(masterPort); + } + + @Override + protected void adjustConfiguration() { + addConfigurationOption("port " + port); + + if (password != null) { + addConfigurationOption("requirepass " + password); + addConfigurationOption("masterauth " + password); + } + + addConfigurationOption("replicaof " + masterHost + " " + masterPort); + } +} diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java new file mode 100644 index 000000000000..06b802dc3ba3 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/testcontainers/RedisSentinelContainer.java @@ -0,0 +1,113 @@ +/* + * 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.nifi.redis.testcontainers; + +import org.springframework.lang.NonNull; +import org.springframework.lang.Nullable; +import org.testcontainers.utility.DockerImageName; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public class RedisSentinelContainer extends RedisContainer { + + public static final int REDIS_SENTINEL_PORT = 26379; + + public RedisSentinelContainer(final @NonNull DockerImageName dockerImageName) { + super(dockerImageName); + + setPort(REDIS_SENTINEL_PORT); + } + + public RedisSentinelContainer(final @NonNull String fullImageName) { + this(DockerImageName.parse(fullImageName)); + } + + @NonNull + protected String masterHost = "localhost"; + protected int masterPort = REDIS_PORT; + @NonNull + protected String masterName = "mymaster"; + @Nullable + protected String sentinelPassword = null; + private long downAfterMilliseconds = 60000L; + private long failoverTimeout = 180000L; + private int parallelSyncs = 1; + private int quorumSize = 1; + + public void setMasterHost(final @NonNull String masterHost) { + this.masterHost = masterHost; + } + + public void setMasterPort(final int masterPort) { + this.masterPort = masterPort; + } + + public void setMasterName(final @NonNull String masterName) { + this.masterName = masterName; + } + + public void setSentinelPassword(final @Nullable String sentinelPassword) { + this.sentinelPassword = sentinelPassword; + } + + public void setQuorumSize(final int quorumSize) { + this.quorumSize = quorumSize; + } + + public void setDownAfterMilliseconds(final long downAfterMilliseconds) { + this.downAfterMilliseconds = downAfterMilliseconds; + } + + public void setFailoverTimeout(final long failoverTimeout) { + this.failoverTimeout = failoverTimeout; + } + + public void setParallelSyncs(final int parallelSyncs) { + this.parallelSyncs = parallelSyncs; + } + + + @Override + protected void adjustConfiguration() { + addConfigurationOption("port " + port); + + addConfigurationOption(String.format("sentinel monitor %s %s %d %d", masterName, masterHost, masterPort, quorumSize)); + addConfigurationOption(String.format("sentinel down-after-milliseconds %s %d", masterName, downAfterMilliseconds)); + addConfigurationOption(String.format("sentinel failover-timeout %s %d", masterName, failoverTimeout)); + addConfigurationOption(String.format("sentinel parallel-syncs %s %d", masterName, parallelSyncs)); + + if (password != null) { + addConfigurationOption("sentinel auth-pass " + masterName + " " + password); + } + if (sentinelPassword != null) { + addConfigurationOption("requirepass " + sentinelPassword); + addConfigurationOption("sentinel sentinel-pass " + sentinelPassword); + } + } + + @Override + protected void configure() { + super.configure(); + + + List commandParts = new ArrayList<>(Arrays.asList(getCommandParts())); + commandParts.add("--sentinel"); + setCommandParts(commandParts.toArray(new String[0])); + } +}