Replayer logging changes

TrafficCapture/trafficReplayer/src/main/java/org/opensearch/migrations/replay/TrafficReplayer.java

@@ -1,5 +1,7 @@
 package org.opensearch.migrations.replay;
 
+import java.io.File;
+import java.io.IOException;
 import java.lang.ref.WeakReference;
 import java.net.URI;
 import java.time.Clock;
@@ -36,6 +38,10 @@
 import io.netty.util.concurrent.DefaultThreadFactory;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.ConfigurationFactory;
+import org.apache.logging.log4j.core.config.ConfigurationSource;
 import org.slf4j.LoggerFactory;
 import org.slf4j.event.Level;
 import software.amazon.awssdk.arns.Arn;
@@ -45,6 +51,7 @@
 @Slf4j
 public class TrafficReplayer {
     private static final String ALL_ACTIVE_CONTEXTS_MONITOR_LOGGER = "AllActiveWorkMonitor";
+    public static final String LOG4J2_PROPERTIES_FILE = "LOG4J2_PROPERTIES_FILE";
 
     public static final String SIGV_4_AUTH_HEADER_SERVICE_REGION_ARG = "--sigv4-auth-header-service-region";
     public static final String AUTH_HEADER_VALUE_ARG = "--auth-header-value";
@@ -91,6 +98,21 @@ public static boolean validateRequiredKafkaParams(String brokers, String topic,
         return true;
     }
 
+    public static void reloadConfiguration(String configFile) throws IOException {
+        ConfigurationSource source;
+        var file = configFile == null || configFile.isEmpty() ? null : new File(configFile);
+        if (file != null && file.exists()) {
+            source = new ConfigurationSource(file.toURI().toURL().openStream(), file.toURI().toURL());
+            LoggerContext context = (LoggerContext) LogManager.getContext(false);
+            var config = ConfigurationFactory.getInstance().getConfiguration(context, source);
+            context.start(config);
+            log.atInfo().setMessage("Successfully reloaded Log4j2 configuration from: {}")
+                .addArgument(configFile).log();
+        } else {
+            log.atInfo().setMessage("Using the default Log4j2 configuration").log();
+        }
+    }
+
     public static class Parameters {
         @Parameter(
             required = true,
@@ -333,6 +355,8 @@ private static Parameters parseArgs(String[] args) {
 
     public static void main(String[] args) throws Exception {
         System.err.println("Got args: " + String.join("; ", args));
+        reloadConfiguration(System.getenv(LOG4J2_PROPERTIES_FILE));
+
         final var workerId = ProcessHelpers.getNodeInstanceName();
         log.info("Starting Traffic Replayer with id=" + workerId);
 

TrafficCapture/trafficReplayer/src/main/resources/log4j2.properties

@@ -2,6 +2,7 @@ status = warn
 
 property.logsDir = ${env:SHARED_LOGS_DIR_PATH:-./logs}
 property.tuplesFileNamePrefix = ${logsDir}/${hostName}/tuples/tuples
+property.tempDir = ./logs/temp
 
 appenders = console, ReplayerLogFile, OUTPUT_TUPLES, TRANSACTION_SUMMARIES, TRANSACTION_SUMMARIES_LOGFILE, ALL_ACTIVE_WORK_MONITOR_LOGFILE
 
@@ -14,8 +15,9 @@ appender.console.direct = true
 
 appender.ReplayerLogFile.type = RollingRandomAccessFile
 appender.ReplayerLogFile.name = ReplayerLogFile
-appender.ReplayerLogFile.fileName = logs/replayer.log
-appender.ReplayerLogFile.filePattern = logs/%d{yyyy-MM}{UTC}/replayer-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz
+appender.ReplayerLogFile.fileName = ${tempDir}/replayer.log
+appender.ReplayerLogFile.filePattern = ${logsDir}/${hostName}/replayer-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz
+appender.ReplayerLogFile.tempCompressedFilePattern = ${tempDir}/replayer-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz.tmp
 appender.ReplayerLogFile.layout.type = PatternLayout
 appender.ReplayerLogFile.layout.pattern = [%-5level] %d{DEFAULT_MICROS}{UTC} [%t] %c{1} - %msg%equals{ ctx=%mdc}{ ctx={}}{}%n
 appender.ReplayerLogFile.policies.type = Policies
@@ -24,18 +26,21 @@ appender.ReplayerLogFile.policies.time.interval = 60
 appender.ReplayerLogFile.policies.time.modulate = true
 appender.ReplayerLogFile.strategy.type = DefaultRolloverStrategy
 appender.ReplayerLogFile.strategy.max = 288
+appender.ReplayerLogFile.strategy.compressionLevel = 9
 appender.ReplayerLogFile.immediateFlush = false
 
 appender.OUTPUT_TUPLES.type = RollingRandomAccessFile
 appender.OUTPUT_TUPLES.name = OUTPUT_TUPLES
-appender.OUTPUT_TUPLES.fileName = ${tuplesFileNamePrefix}.log
-appender.OUTPUT_TUPLES.filePattern = ${tuplesFileNamePrefix}-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz
+appender.OUTPUT_TUPLES.fileName = ${tempDir}/tuples.log
+appender.OUTPUT_TUPLES.filePattern = ${logsDir}/${hostName}/tuples/tuples-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz
+appender.OUTPUT_TUPLES.tempCompressedFilePattern = ${tempDir}/tuples-%d{yyyy-MM-dd-HH-mm}{UTC}-%i.log.gz.tmp
 appender.OUTPUT_TUPLES.layout.type = PatternLayout
 appender.OUTPUT_TUPLES.layout.pattern = %m%n
 appender.OUTPUT_TUPLES.policies.type = Policies
 appender.OUTPUT_TUPLES.policies.size.type = SizeBasedTriggeringPolicy
 appender.OUTPUT_TUPLES.policies.size.size = 10 MB
 appender.OUTPUT_TUPLES.strategy.type = DefaultRolloverStrategy
+appender.OUTPUT_TUPLES.strategy.compressionLevel = 9
 appender.OUTPUT_TUPLES.immediateFlush = false
 
 appender.TRANSACTION_SUMMARIES.type = Console

deployment/k8s/charts/components/replayer/templates/deployment.yaml

@@ -23,7 +23,7 @@
         - name: wait-for-kafka
           image: bitnami/kubectl:latest  # or any image with curl/kubectl
           command: [ 'sh', '-c',
                      'until kubectl wait --for=condition=Ready kafka/captured-traffic -n {{.Release.Namespace }} --timeout=10s; do echo waiting for kafka cluster is ready; sleep 1; done' ]
       containers:
         - name: replayer
           image: migrations/traffic_replayer:latest
@@ -47,6 +47,9 @@
             - name: shared-logs
               mountPath: /shared-logs-output
             {{- end }}
+          env:
+          - name: LOG4J2_PROPERTIES_FILE
+            value: "/shared-logs-output/replayer_log4j.properties"
       volumes:
         - name: {{ $envMountName }}
           emptyDir: {}