Check for nulls for timestamps and logfiles , validate config and event formats (#61)

* Should fix potential Null pointer dereference when parsing timestamps

* Fixes situations where logfile might be null

* Move logfile null check to more correct location

* Move validation to the config object

* Adds full config validation for configs

* Try/catch Instant.parse instead of nullcheck

* Adds validator for appname and hostname while handling messages

* appname -> appName refactor

* Adds missing exception constructors

* Change all ints to integeres in AppConfig objects, check for null where necessary

Author: StrongestNumber9

example/combined.yaml

@@ -42,6 +42,7 @@ data:
       "kubernetes": {
         "logdir": "/var/log/containers",
         "url": "https://127.0.0.1:8443",
+        "timezone": "Europe/Helsinki",
         "cacheExpireInterval": 300,
         "cacheMaxEntries": 4096,
         "labels": {
@@ -104,7 +105,7 @@ data:
     </Configuration>
 kind: ConfigMap
 metadata:
-  name: app-config-2t5785hm6f
+  name: app-config-58dh7f727h
 ---
 apiVersion: v1
 data:
@@ -271,7 +272,7 @@ spec:
   terminationGracePeriodSeconds: 0
   volumes:
   - configMap:
-      name: app-config-2t5785hm6f
+      name: app-config-58dh7f727h
     name: app-config
   - hostPath:
       path: /var/log/containers

example/config/k8s_01/config.json

@@ -5,6 +5,7 @@
   "kubernetes": {
     "logdir": "/var/log/containers",
     "url": "https://127.0.0.1:8443",
+    "timezone": "Europe/Helsinki",
     "cacheExpireInterval": 300,
     "cacheMaxEntries": 4096,
     "labels": {

src/main/java/com/teragrep/k8s_01/InvalidConfigurationException.java

@@ -0,0 +1,36 @@
+/*
+   Kubernetes log forwarder k8s_01
+   Copyright (C) 2023  Suomen Kanuuna Oy
+
+   Licensed 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 com.teragrep.k8s_01;
+
+public class InvalidConfigurationException extends Exception {
+    public InvalidConfigurationException() {
+        super();
+    }
+
+    public InvalidConfigurationException(String message) {
+        super(message);
+    }
+
+    public InvalidConfigurationException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
+
+    public InvalidConfigurationException(Throwable throwable) {
+        super(throwable);
+    }
+}

src/main/java/com/teragrep/k8s_01/K8SConsumer.java

@@ -36,9 +36,11 @@
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.function.Consumer;
+import java.util.regex.Pattern;
 
 /**
  * Must be thread-safe
@@ -53,6 +55,10 @@ public class K8SConsumer implements Consumer<FileRecord> {
     private static final DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSSxxx");
     private final ZoneId timezoneId;
 
+    // Validators
+    private static final Pattern hostnamePattern = Pattern.compile("^[a-zA-Z0-9.-]+$"); // Not perfect but filters basically all mistakes
+    private static final Pattern appNamePattern = Pattern.compile("^[\\x21-\\x7e]+$"); // DEC 33 - DEC 126 as specified in RFC5424
+
     K8SConsumer(
             AppConfig appConfig,
             KubernetesCachingAPIClient cacheClient,
@@ -131,7 +137,26 @@ public void accept(FileRecord record) {
                     )
                 );
             }
-            Instant instant = Instant.parse(log.getTimestamp());
+            Instant instant;
+            try {
+                instant = Instant.parse(log.getTimestamp());
+            }
+            catch(DateTimeParseException e) {
+                throw new RuntimeException(
+                        String.format(
+                                "[%s] Can't parse timestamp <%s> properly for event from pod <%s/%s> on container <%s> in file %s/%s at offset %s: ",
+                                uuid,
+                                log.getTimestamp(),
+                                namespace,
+                                podname,
+                                containerId,
+                                record.getPath(),
+                                record.getFilename(),
+                                record.getStartOffset()
+                        ),
+                        e
+                );
+            }
             ZonedDateTime zdt = instant.atZone(timezoneId);
             String timestamp = zdt.format(format);
 
@@ -152,34 +177,87 @@ public void accept(FileRecord record) {
             JsonObject dockerMetadata = new JsonObject();
             dockerMetadata.addProperty("container_id", containerId);
 
-            // Handle hostname and appname, use fallback values when labels are empty or if label not found
+            // Handle hostname and appName, use fallback values when labels are empty or if label not found
             String hostname;
-            String appname;
+            String appName;
             if(podMetadataContainer.getLabels() == null) {
                 LOGGER.warn(
-                        "[{}] Can't resolve metadata and/or labels for container <{}>, using fallback values for hostname and appname",
+                        "[{}] Can't resolve metadata and/or labels for container <{}>, using fallback values for hostname and appName",
                         uuid,
                         containerId
                 );
                 hostname = appConfig.getKubernetes().getLabels().getHostname().getFallback();
-                appname = appConfig.getKubernetes().getLabels().getAppname().getFallback();
+                appName = appConfig.getKubernetes().getLabels().getAppName().getFallback();
             }
             else {
                 hostname = podMetadataContainer.getLabels().getOrDefault(
                         appConfig.getKubernetes().getLabels().getHostname().getLabel(log.getStream()),
                         appConfig.getKubernetes().getLabels().getHostname().getFallback()
                 );
-                appname = podMetadataContainer.getLabels().getOrDefault(
-                        appConfig.getKubernetes().getLabels().getAppname().getLabel(log.getStream()),
-                        appConfig.getKubernetes().getLabels().getAppname().getFallback()
+                appName = podMetadataContainer.getLabels().getOrDefault(
+                        appConfig.getKubernetes().getLabels().getAppName().getLabel(log.getStream()),
+                        appConfig.getKubernetes().getLabels().getAppName().getFallback()
+                );
+            }
+            hostname = appConfig.getKubernetes().getLabels().getHostname().getPrefix() + hostname;
+            appName = appConfig.getKubernetes().getLabels().getAppName().getPrefix() + appName;
+
+            if(!hostnamePattern.matcher(hostname).matches()) {
+                throw new RuntimeException(
+                        String.format(
+                                "[%s] Detected hostname <[%s]> from pod <[%s]/[%s]> on container <%s> contains invalid characters, can't continue",
+                                uuid,
+                                hostname,
+                                namespace,
+                                podname,
+                                containerId
+                        )
+                );
+            }
+
+            if(hostname.length() >= 255) {
+                throw new RuntimeException(
+                        String.format(
+                                "[%s] Detected hostname <[%s]...> from pod <[%s]/[%s]> on container <%s> is too long, can't continue",
+                                uuid,
+                                hostname.substring(0,30),
+                                namespace,
+                                podname,
+                                containerId
+                        )
+                );
+            }
+
+            if(!appNamePattern.matcher(appName).matches()) {
+                throw new RuntimeException(
+                        String.format(
+                                "[%s] Detected appName <[%s]> from pod <[%s]/[%s]> on container <%s> contains invalid characters, can't continue",
+                                uuid,
+                                appName,
+                                namespace,
+                                podname,
+                                containerId
+                        )
+                );
+            }
+            if(appName.length() > 48) {
+                throw new RuntimeException(
+                        String.format(
+                                "[%s] Detected appName <[%s]...> from pod <[%s]/[%s]> on container <%s> is too long, can't continue",
+                                uuid,
+                                appName.substring(0,30),
+                                namespace,
+                                podname,
+                                containerId
+                        )
                 );
             }
 
             if(LOGGER.isDebugEnabled()) {
                 LOGGER.debug(
                         "[{}] Resolved message to be {}@{} from {}/{} generated at {}",
                         uuid,
-                        appname,
+                        appName,
                         hostname,
                         namespace,
                         podname,
@@ -205,8 +283,8 @@ public void accept(FileRecord record) {
             SyslogMessage syslog = new SyslogMessage()
                     .withTimestamp(timestamp, true)
                     .withSeverity(Severity.WARNING)
-                    .withHostname(appConfig.getKubernetes().getLabels().getHostname().getPrefix() + hostname)
-                    .withAppName(appConfig.getKubernetes().getLabels().getAppname().getPrefix() + appname)
+                    .withHostname(hostname)
+                    .withAppName(appName)
                     .withFacility(Facility.USER)
                     .withSDElement(SDMetadata)
                     .withMsg(new String(record.getRecord()));

src/main/java/com/teragrep/k8s_01/KubernetesLogReader.java

@@ -42,6 +42,7 @@ public static void main(String[] args) throws IOException {
         AppConfig appConfig;
         try {
             appConfig = gson.fromJson(new FileReader("etc/config.json"), AppConfig.class);
+            appConfig.validate();
         }
         catch (FileNotFoundException e) {
             LOGGER.error(
@@ -57,6 +58,13 @@ public static void main(String[] args) throws IOException {
             );
             return;
         }
+        catch (InvalidConfigurationException e) {
+            LOGGER.error(
+                    "Failed to validate config 'etc/config.json':",
+                    e
+            );
+            return;
+        }
         catch (Exception e) {
             LOGGER.error(
                     "Unknown exception while handling config:",
@@ -108,23 +116,25 @@ public static void main(String[] args) throws IOException {
 
         // consumer supplier, returns always the same instance
         K8SConsumerSupplier consumerSupplier = new K8SConsumerSupplier(appConfig, cacheClient, relpOutputPool);
-
         String[] logfiles = appConfig.getKubernetes().getLogfiles();
         LOGGER.debug(
                 "Monitored logfiles: {}",
                 Arrays.toString(logfiles)
         );
+
         List<Thread> threads = new ArrayList<>();
         String statesStore = System.getProperty("user.dir") + "/var";
         LOGGER.debug(
                 "Using {} as statestore",
                 statesStore
         );
+
         // FIXME: VERIFY: SFR is not in try-with-resources block as it will have weird behaviour with threads.
         StatefulFileReader statefulFileReader = new StatefulFileReader(
             Paths.get(statesStore),
             consumerSupplier
         );
+
         // Start a new thread for all logfile watchers
         for (String logfile : logfiles) {
             Thread thread = new Thread(() -> {

src/main/java/com/teragrep/k8s_01/config/AppConfig.java

@@ -18,22 +18,19 @@
 package com.teragrep.k8s_01.config;
 
 import com.google.gson.Gson;
+import com.teragrep.k8s_01.InvalidConfigurationException;
 
 /* POJO representing the main config.json */
-public class AppConfig {
-    private AppConfigKubernetes kubernetes;
-
+public class AppConfig implements BaseConfig {
+    private AppConfigMetrics metrics;
     public AppConfigMetrics getMetrics() {
         return metrics;
     }
-
-    private AppConfigMetrics metrics;
-    private AppConfigRelp relp;
-
+    private AppConfigKubernetes kubernetes;
     public AppConfigKubernetes getKubernetes() {
         return kubernetes;
     }
-
+    private AppConfigRelp relp;
     public AppConfigRelp getRelp() {
         return relp;
     }
@@ -42,4 +39,22 @@ public AppConfigRelp getRelp() {
     public String toString() {
         return new Gson().toJson(this);
     }
+
+    @Override
+    public void validate() throws InvalidConfigurationException {
+        if(metrics == null) {
+            throw new InvalidConfigurationException("metrics object not found or is null in main config object");
+        }
+        getMetrics().validate();
+
+        if(kubernetes == null) {
+            throw new InvalidConfigurationException("kubernetes object not found or is null in main config object");
+        }
+        getKubernetes().validate();
+
+        if(relp == null) {
+            throw new InvalidConfigurationException("relp object no found or is null in main config object");
+        }
+        getRelp().validate();
+    }
 }