Handle Reader thread termination gracefully

build.gradle

@@ -34,7 +34,6 @@ buildscript {
 
     dependencies {
         classpath "org.opensearch.gradle:build-tools:${opensearch_version}"
-        classpath group: 'com.google.guava', name: 'guava', version: '31.1-jre'
     }
 }
 
@@ -186,15 +185,15 @@ jacocoTestCoverageVerification {
                 }
             }
         }
-    } 
+    }
     else {
         violationRules {
             rule {
                 limit {
                     minimum = 0.6
                 }
             }
-        }    
+        }
     }
 }
 
@@ -329,6 +328,7 @@ dependencies {
     }
     def versions = VersionProperties.versions
 
+    def commonslangVersion = "${versions.commonslang}"
     def jacksonVersion = "${versions.jackson}"
     def jacksonDataBindVersion = "${versions.jackson_databind}"
     def nettyVersion = "${versions.netty}"
@@ -350,7 +350,7 @@ dependencies {
     implementation "org.opensearch:performance-analyzer-commons:1.0.0-SNAPSHOT"
     implementation group: 'org.apache.logging.log4j', name: 'log4j-api', version: "${log4jVersion}"
     implementation group: 'org.apache.logging.log4j', name: 'log4j-core', version: "${log4jVersion}"
-    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.9'
+    implementation group: 'org.apache.commons', name: 'commons-lang3', version: "${commonslangVersion}"
     implementation group: 'commons-io', name: 'commons-io', version: '2.7'
     implementation group: 'com.google.errorprone', name: 'error_prone_annotations', version: '2.9.0'
     implementation group: 'com.google.protobuf', name: 'protobuf-java', version: "${protobufVersion}"
@@ -373,16 +373,18 @@ dependencies {
             strictly "2.23.0"
         }
     }
-    testImplementation group: 'org.powermock', name: 'powermock-core', version: '2.0.0'
-    testImplementation group: 'org.powermock', name: 'powermock-api-support', version: '2.0.0'
-    testImplementation group: 'org.powermock', name: 'powermock-module-junit4-common', version: '2.0.0'
-    testImplementation group: 'org.javassist', name: 'javassist', version: '3.24.0-GA'
-    testImplementation group: 'org.powermock', name: 'powermock-reflect', version: '2.0.0'
+    testImplementation group: 'org.powermock', name: 'powermock-core', version: '2.0.2'
+    testImplementation group: 'org.powermock', name: 'powermock-api-support', version: '2.0.2'
+    testImplementation group: 'org.powermock', name: 'powermock-module-junit4-common', version: '2.0.2'
+    testImplementation group: 'org.javassist', name: 'javassist', version: '3.28.0-GA'
+    testImplementation group: 'org.powermock', name: 'powermock-reflect', version: '2.0.2'
     testImplementation group: 'net.bytebuddy', name: 'byte-buddy', version: '1.9.3'
     testImplementation group: 'org.objenesis', name: 'objenesis', version: '3.0.1'
     testImplementation group: 'org.hamcrest', name: 'hamcrest-library', version: '2.1'
     testImplementation group: 'org.hamcrest', name: 'hamcrest', version: '2.1'
     testImplementation group: 'junit', name: 'junit', version: "${junitVersion}"
+    testImplementation group: 'org.xerial', name: 'sqlite-jdbc', version: '3.41.2.2'
+    testImplementation group: 'com.github.stefanbirkner', name: 'system-rules', version: '1.19.0'
 
     // Required for Docker build
     dockerBuild group: 'org.opensearch.plugin', name:'performance-analyzer', version: "${opensearch_build}"

config/supervisord.conf

@@ -7,7 +7,7 @@ chmod=0770
 [supervisord]
 logfile=/usr/share/supervisor/performance_analyzer/supervisord.log ; (main log file;default $CWD/supervisord.log)
 pidfile=/usr/share/supervisor/performance_analyzer/supervisord.pid ; (supervisord pidfile;default supervisord.pid)
-childlogdir=/usr/share/supervisor/performance_analyzer            ; ('AUTO' child log dir, default $TEMP)
+childlogdir=/usr/share/supervisor/performance_analyzer             ; ('AUTO' child log dir, default $TEMP)
 
 ; the below section must remain in the config file for RPC
 ; (supervisorctl/web interface) to work, additional interfaces may be
@@ -23,10 +23,17 @@ serverurl=/usr/share/supervisord.sock
 ; newlines).  It can also contain wildcards.  The filenames are
 ; interpreted as relative to this file.  Included files *cannot*
 ; include files themselves.
-
 [include]
 files = /etc/supervisor/conf.d/*.conf
 
 [program:performance_analyzer]
 command=/usr/share/opensearch/performance-analyzer-rca/bin/performance-analyzer-agent /usr/share/opensearch
 user=1000
+autostart=true                                                      ; start at supervisord start (default: true)
+autorestart=unexpected                                              ; autorestart if exited after running (def: unexpected)
+exitcodes=1                                                         ; 'expected' exit codes used with autorestart, System.exit(1) called from PerformanceAnalyzerApp.cleanupAndExit() - if Reader Thread crashes.
+
+[eventlistener:stop_supervisord]
+command=bash -c "printf 'READY\n' && while read line; do kill -SIGQUIT $PPID; done < /dev/stdin"
+events=PROCESS_STATE_FATAL
+buffer_size=100

docker/docker-compose.yml

@@ -32,8 +32,6 @@ services:
 
   opensearch2:
     container_name: opensearch2
-    environment:
-      - node.cluster_manager=false
     image: opensearch/pa-rca:3.0
     mem_limit: 4g
     networks:

src/main/java/org/opensearch/performanceanalyzer/LocalhostConnectionUtil.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.performanceanalyzer;
+
+
+import io.netty.handler.codec.http.HttpMethod;
+import java.io.DataOutputStream;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.performanceanalyzer.core.Util;
+
+public class LocalhostConnectionUtil {
+
+    private static final int TIMEOUT_MILLIS = 30000;
+
+    private static final Logger LOG = LogManager.getLogger(LocalhostConnectionUtil.class);
+
+    public static void disablePA() throws InterruptedException {
+        String PA_CONFIG_PATH = Util.PA_BASE_URL + "/cluster/config";
+        String PA_DISABLE_PAYLOAD = "{\"enabled\": false}";
+        int retryCount = 5;
+
+        while (retryCount > 0) {
+            HttpURLConnection connection = null;
+            try {
+                connection = createHTTPConnection(PA_CONFIG_PATH, HttpMethod.POST);
+                DataOutputStream stream = new DataOutputStream(connection.getOutputStream());
+                stream.writeBytes(PA_DISABLE_PAYLOAD);
+                stream.flush();
+                stream.close();
+                LOG.info(
+                        "PA Disable Response: "
+                                + connection.getResponseCode()
+                                + " "
+                                + connection.getResponseMessage());
+                if (connection.getResponseCode() == HttpURLConnection.HTTP_OK) {
+                    return;
+                }
+            } catch (Exception e) {
+                LOG.error("PA Disable Request failed: " + e.getMessage(), e);
+            } finally {
+                if (connection != null) {
+                    connection.disconnect();
+                }
+            }
+            --retryCount;
+            Thread.sleep((int) (60000 * (Math.random() * 2) + 100));
+        }
+        throw new RuntimeException("Failed to disable PA after 5 attempts");
+    }
+
+    private static HttpURLConnection createHTTPConnection(String path, HttpMethod httpMethod) {
+        try {
+            String endPoint = "http://localhost:9200" + path;
+            URL endpointUrl = new URL(endPoint);
+            HttpURLConnection connection = (HttpURLConnection) endpointUrl.openConnection();
+            connection.setRequestMethod(httpMethod.toString());
+            connection.setRequestProperty("Content-Type", "application/json");
+
+            connection.setConnectTimeout(TIMEOUT_MILLIS);
+            connection.setReadTimeout(TIMEOUT_MILLIS);
+            connection.setUseCaches(false);
+            connection.setDoOutput(true);
+            return connection;
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    "Failed to create OpenSearch Connection: " + e.getMessage(), e);
+        }
+    }
+}