stop threads (#74)

* Interrupt threads before shutdown * Set logging to debug level on dev.watch() * Use new version of rlo_12 and rlo_13 with differnt way of handling DEW threads to enable graceful shutdown * Removes unnecessary logging messages
teragrep · Jul 27, 2023 · 4c41020 · 4c41020
1 parent 83551b1
commit 4c41020
Show file tree

Hide file tree

Showing 7 changed files with 60 additions and 57 deletions.
diff --git a/example/combined.yaml b/example/combined.yaml
@@ -181,6 +181,8 @@ metadata:
     app: first-pod-appname
     app-stderr: first-pod-appname-stderr
     env: dev
+    hostname: first-pod
+    hostname-stderr: first-pod-stderr
   name: first-pod
   namespace: default
 spec:
@@ -268,7 +270,7 @@ spec:
   restartPolicy: Always
   serviceAccount: kubelogreader
   serviceAccountName: kubelogreader
-  terminationGracePeriodSeconds: 0
+  terminationGracePeriodSeconds: 30
   volumes:
   - configMap:
       name: app-config-8dtf4m92md
@@ -308,6 +310,13 @@ spec:
     - /usr/bin/java
     image: ghcr.io/teragrep/rlp_07/app:latest
     imagePullPolicy: IfNotPresent
+    lifecycle:
+      preStop:
+        exec:
+          command:
+          - sh
+          - -c
+          - echo presleep; sleep 10;
     name: first-pod
     ports:
     - containerPort: 1601
@@ -321,7 +330,7 @@ spec:
   imagePullSecrets:
   - name: ghcr.io
   restartPolicy: Always
-  terminationGracePeriodSeconds: 0
+  terminationGracePeriodSeconds: 30
   volumes:
   - configMap:
       name: receiver-config-cf8g6bh7tg

diff --git a/example/pods/first-pod.json b/example/pods/first-pod.json
@@ -3,6 +3,8 @@
     "name": "first-pod",
     "namespace": "default",
     "labels": {
+      "hostname": "first-pod",
+      "hostname-stderr": "first-pod-stderr",
       "app": "first-pod-appname",
       "app-stderr": "first-pod-appname-stderr",
       "env": "dev"

diff --git a/example/pods/k8s_01.json b/example/pods/k8s_01.json
@@ -15,7 +15,7 @@
     "serviceAccount": "kubelogreader",
     "serviceAccountName": "kubelogreader",
     "dnsPolicy": "ClusterFirst",
-    "terminationGracePeriodSeconds": 0,
+    "terminationGracePeriodSeconds": 30,
     "volumes": [
       {
         "name": "app-config",

diff --git a/example/pods/receiver.json b/example/pods/receiver.json
@@ -12,7 +12,7 @@
   "spec": {
     "hostname": "receiver",
     "dnsPolicy": "ClusterFirst",
-    "terminationGracePeriodSeconds": 0,
+    "terminationGracePeriodSeconds": 30,
     "volumes": [
       {
         "name": "receiver-config",
@@ -39,6 +39,17 @@
             "name": "receiver-config"
           }
         ],
+        "lifecycle": {
+          "preStop": {
+            "exec": {
+              "command": [
+                "sh",
+                "-c",
+                "echo presleep; sleep 10;"
+              ]
+            }
+          }
+        },
         "command": [
           "/usr/bin/java"
         ],

diff --git a/pom.xml b/pom.xml
@@ -45,7 +45,7 @@
     <dependency>
       <groupId>com.teragrep</groupId>
       <artifactId>rlo_13</artifactId>
-      <version>1.2.0</version>
+      <version>1.2.1</version>
     </dependency>
 
     <!-- RELP sending library -->

diff --git a/src/main/java/com/teragrep/k8s_01/KubernetesLogReader.java b/src/main/java/com/teragrep/k8s_01/KubernetesLogReader.java
@@ -41,7 +41,7 @@ public class KubernetesLogReader {
     private static final Logger LOGGER = LoggerFactory.getLogger(KubernetesLogReader.class);
     private static final MetricRegistry metricRegistry = new MetricRegistry();
     static Gson gson = new Gson();
-    public static void main(String[] args) throws IOException {
+    public static void main(String[] args) throws IOException, InterruptedException {
         AppConfig appConfig;
         try {
             try(InputStreamReader isr = new InputStreamReader(Files.newInputStream(Paths.get("etc/config.json")), StandardCharsets.UTF_8)) {
@@ -127,7 +127,7 @@ public static void main(String[] args) throws IOException {
                 Arrays.toString(logfiles)
         );
 
-        List<Thread> threads = new ArrayList<>();
+        List<DirectoryEventWatcher> dews = new ArrayList<>();
         String statesStore = System.getProperty("user.dir") + "/var";
         LOGGER.debug(
                 "Using {} as statestore",
@@ -143,8 +143,20 @@ public static void main(String[] args) throws IOException {
         // Graceful shutdown so Relp sessions are gracefully terminated
         Runtime.getRuntime().addShutdownHook(new Thread(() -> {
             LOGGER.info("Shutting down.");
+            for(DirectoryEventWatcher dew : dews) {
+                LOGGER.debug("Shutting down dew " + dew);
+                try {
+                    dew.stop();
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+            LOGGER.info(
+                    "Disconnecting {} relp threads",
+                    outputThreads
+            );
             for(int i=1; i <= outputThreads; i++) {
-                LOGGER.info(
+                LOGGER.debug(
                         "Disconnecting relp thread #{}/{}",
                         i,
                         outputThreads
@@ -164,52 +176,23 @@ public static void main(String[] args) throws IOException {
 
         // Start a new thread for all logfile watchers
         for (String logfile : logfiles) {
-            Thread thread = new Thread(() -> {
-                LOGGER.debug(
-                        "Starting new DirectoryEventWatcher thread on directory '{}' with pattern '{}'",
-                        appConfig.getKubernetes().getLogdir(),
-                        logfile
-                );
-                try {
-                    DirectoryEventWatcher dew = new DirectoryEventWatcher(
-                            Paths.get(appConfig.getKubernetes().getLogdir()),
-                            false,
-                            Pattern.compile(logfile),
-                            statefulFileReader,
-                            500,
-                            TimeUnit.MILLISECONDS,
-                            appConfig.getKubernetes().getMaxLogReadingThreads()
-                    );
-                    dew.watch();
-                } catch (IOException | InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
-            });
-            thread.setName("DEW-" + threads.size());
-            thread.start();
-            threads.add(thread);
-        }
-
-        // FIXME: Is this necessary
-        for (Thread thread : threads) {
-            if(LOGGER.isTraceEnabled()) {
-                LOGGER.trace(
-                        "Waiting for thread {}#{} to finish",
-                        thread.getName(),
-                        thread.getId()
-                );
-            }
-            try {
-                thread.join();
-            } catch (InterruptedException e) {
-                LOGGER.error(
-                        "Failed to stop thread {}#{}:",
-                        thread.getName(),
-                        thread.getId(),
-                        e
-                );
-                throw new RuntimeException(e);
-            }
+            LOGGER.debug(
+                    "Starting new DirectoryEventWatcher on directory '{}' with pattern '{}'",
+                    appConfig.getKubernetes().getLogdir(),
+                    logfile
+            );
+            DirectoryEventWatcher dew = new DirectoryEventWatcher(
+                    Paths.get(appConfig.getKubernetes().getLogdir()),
+                    false,
+                    Pattern.compile(logfile),
+                    statefulFileReader,
+                    500,
+                    TimeUnit.MILLISECONDS,
+                    appConfig.getKubernetes().getMaxLogReadingThreads()
+            );
+            dew.start();
+            dews.add(dew);
         }
+        Thread.sleep(Long.MAX_VALUE);
     }
 }
diff --git a/src/main/java/com/teragrep/k8s_01/RelpOutput.java b/src/main/java/com/teragrep/k8s_01/RelpOutput.java
@@ -122,13 +122,11 @@ public void disconnect() {
             totalConnections.dec();
             relpConnection.disconnect();
         } catch (IOException | TimeoutException e) {
-            LOGGER.debug(
+            LOGGER.info(
                     "[#{}] Had to teardown connection",
                     getId()
             );
             relpConnection.tearDown();
-            throughputErrors.mark();
-            throw new RuntimeException(e);
         }
     }