Added logic to fail back based on a pid file.

Author: viragtripathi

pom.xml

@@ -63,6 +63,11 @@
             <artifactId>commons-pool2</artifactId>
             <version>2.11.1</version>
         </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.11.0</version>
+        </dependency>
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-simple</artifactId>

src/main/java/com/redis/benchmark/utils/FileEventListener.java

@@ -0,0 +1,51 @@
+package com.redis.benchmark.utils;
+
+import java.io.File;
+import org.apache.commons.io.monitor.FileAlterationListener;
+import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
+import org.apache.commons.io.monitor.FileAlterationMonitor;
+import org.apache.commons.io.monitor.FileAlterationObserver;
+
+public final class FileEventListener {
+
+    public static final FileEventListener FILE_EVENT_LISTENER = new FileEventListener();
+
+    public FileEventListener() {
+    }
+
+    /*
+    Apache commons monitoring uses a polling mechanism with a configurable polling interval.
+    In every poll, it calls listFiles() method of File class and compares with the listFiles()
+    output of the previous iteration to identify file creation, modification and deletion.
+    The algorithm is robust enough. Works even on network drives.
+     */
+    public void start(String dir, int pollInterval) throws Exception {
+        FileAlterationObserver observer = new FileAlterationObserver(dir);
+        FileAlterationMonitor monitor = new FileAlterationMonitor(pollInterval);
+        FileAlterationListener listener = new FileAlterationListenerAdaptor() {
+            @Override
+            public void onFileCreate(File file) {
+                System.out.println("\nDetected file create event. File: " + file);
+                int failBackIndex = Integer.parseInt(file.getName().substring(0, 1));
+                JedisConnectionManagement.firstActiveIndex = failBackIndex;
+                System.out.println("User have requested a fail-back event. Setting the active multi cluster index to " + failBackIndex);
+                JedisConnectionManagement.provider.setActiveMultiClusterIndex(JedisConnectionManagement.firstActiveIndex);
+            }
+
+            @Override
+            public void onFileDelete(File file) {
+                System.out.println("\nDetected file delete event. File: " + file);
+            }
+
+            @Override
+            public void onFileChange(File file) {
+                System.out.println("\nDetected file change event. File: " + file);
+            }
+        };
+        observer.addListener(listener);
+        monitor.addObserver(observer);
+        monitor.start();
+        System.out.println("Starting File Listener Service on " + observer.getDirectory());
+    }
+
+}

src/main/java/com/redis/benchmark/utils/JedisConnectionManagement.java

@@ -8,18 +8,26 @@
 import redis.clients.jedis.commands.JedisCommands;
 import redis.clients.jedis.MultiClusterJedisClientConfig;
 import redis.clients.jedis.providers.MultiClusterPooledConnectionProvider;
+
+import java.io.File;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Set;
 
 public final class JedisConnectionManagement {
     private static final JedisConnectionManagement connectionManagement = new JedisConnectionManagement();
     private static Boolean connectionCreated = false;
     private UnifiedJedis unifiedJedis;
+    public static MultiClusterPooledConnectionProvider provider;
+    public static int firstActiveIndex;
+    public static String pidPath;
+    public static String pidFile;
+
 
     private JedisConnectionManagement() {
     }
 
     private void createJedisConnection() {
-        MultiClusterPooledConnectionProvider provider;
         MultiClusterJedisClientConfig.Builder multiClusterJedisClientConfig;
         Set<HostAndPort> hostAndPorts = BenchmarkConfiguration.get().getRedisHostAndPorts();
         int index = 0;
@@ -52,6 +60,13 @@ private void createJedisConnection() {
                 provider = new MultiClusterPooledConnectionProvider(multiClusterJedisClientConfig.build());
 
                 connectionManagement.unifiedJedis = new UnifiedJedis(provider);
+
+                firstActiveIndex = Integer.parseInt(provider.getClusterCircuitBreaker().getName().split(":")[1]);
+                pidPath = System.getProperty("user.dir") + File.separator + "jedisPid" + File.separator;
+                pidFile = pidPath + firstActiveIndex + ".pid";
+                PidFile.create(Path.of(pidFile), true, firstActiveIndex);
+
+                FileEventListener.FILE_EVENT_LISTENER.start(pidPath, 1000);
             }
         } catch (Exception e) {
             System.err.println("------------------- Failed UnifiedJedis " + e.getMessage());
@@ -63,6 +78,9 @@ public static JedisCommands getCommands() {
             connectionManagement.createJedisConnection();
             connectionCreated = Boolean.TRUE;
         }
+        if (!Files.exists(Path.of(pidFile)) && !Files.isRegularFile(Path.of(pidFile))) {
+            provider.setActiveMultiClusterIndex(firstActiveIndex);
+        }
         return connectionManagement.unifiedJedis;
     }
 }

src/main/java/com/redis/benchmark/utils/PidFile.java

@@ -0,0 +1,100 @@
+package com.redis.benchmark.utils;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.management.ManagementFactory;
+import java.nio.charset.Charset;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+/**
+ * Process ID file abstraction that writes the current pid into a file and optionally
+ * removes it on system exit.
+ */
+public final class PidFile {
+
+    private final long pid;
+    private final Path path;
+    private final boolean deleteOnExit;
+
+    private PidFile(Path path, boolean deleteOnExit, long pid) {
+        this.path = path;
+        this.deleteOnExit = deleteOnExit;
+        this.pid = pid;
+    }
+
+    /**
+     * Creates a new PidFile and writes the current process ID into the provided path
+     *
+     * @param path         the path to the pid file. The file is newly created or truncated if it already exists
+     * @param deleteOnExit if <code>true</code> the pid file is deleted with the best effort on system exit
+     * @throws IOException if an IOException occurs
+     */
+    public static PidFile create(Path path, boolean deleteOnExit) throws IOException {
+        return create(path, deleteOnExit, ManagementFactory.getRuntimeMXBean().getPid());
+    }
+
+    static PidFile create(Path path, boolean deleteOnExit, long pid) throws IOException {
+        Path parent = path.getParent();
+        if (parent != null) {
+            if (Files.exists(parent) && !Files.isDirectory(parent)) {
+                throw new IllegalArgumentException(parent + " exists but is not a directory");
+            }
+            if (!Files.exists(parent)) {
+                // only do this if it doesn't exist we get a better exception further down
+                // if there are security issues etc. this also doesn't work if the parent exists
+                // and is a soft-link like on many linux systems /var/run can be a link and that should
+                // not prevent us from writing the PID
+                Files.createDirectories(parent);
+            }
+        }
+        if (Files.exists(path) && !Files.isRegularFile(path))
+            throw new IllegalArgumentException(path + " exists but is not a regular file");
+
+        try (OutputStream stream = Files.newOutputStream(path, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)) {
+            stream.write(Long.toString(pid).getBytes(Charset.defaultCharset()));
+        }
+
+        if (deleteOnExit)
+            addShutdownHook(path);
+
+        return new PidFile(path, deleteOnExit, pid);
+    }
+
+
+    /**
+     * Returns the current process id
+     */
+    public long getPid() {
+        return pid;
+    }
+
+    /**
+     * Returns the process id file path
+     */
+    public Path getPath() {
+        return path;
+    }
+
+    /**
+     * Returns <code>true</code> if the process id file is deleted on system exit. Otherwise <code>false</code>.
+     */
+    public boolean isDeleteOnExit() {
+        return deleteOnExit;
+    }
+
+    private static void addShutdownHook(final Path path) {
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            try {
+                Files.deleteIfExists(path);
+            } catch (IOException e) {
+                try {
+                    throw new Exception("Failed to delete pid file " + path, e);
+                } catch (Exception ex) {
+                    throw new RuntimeException(ex);
+                }
+            }
+        }));
+    }
+}