Merge pull request #473 from jfzunigac/multidirectory_support

Allow multiple directories per log config

Author: jfzunigac

singer/pom.xml

@@ -95,6 +95,11 @@
             <artifactId>commons-text</artifactId>
             <version>1.10.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.12.0</version>
+        </dependency>
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>

singer/src/main/java/com/pinterest/singer/common/LogStream.java

@@ -40,6 +40,7 @@
 import java.io.File;
 import java.io.FileFilter;
 import java.io.IOException;
+import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -94,7 +95,7 @@ public class LogStream {
 
   // The descriptor of the LogStream which consists of SingleLog name and LogStream name.
   // It is globally unique descriptor of the LogStream.
-  private final String logStreamDescriptor;
+  private String logStreamDescriptor;
 
   private Object logFilesInfoLock = new Object();
 
@@ -117,8 +118,25 @@ public class LogStream {
 
   private long lastCompleteCycleTime;
 
+  // the directory that this logStream pertains to
+  private String dir;
+
   private FileNameMatchMode fileNameMatchMode;
 
+  /**
+   * New constructor if there are multiple directories in a config because we need to know the directory through the
+   * full added path to derive the directory, we can no longer use the log config to derive the directory
+   *
+   * Example full added path: "/var/log/singer/singer.log.0"
+   * With the Multi-directory feature, the directory can be "/var/log/*" and the above full added path can be used.
+   * */
+  public LogStream(SingerLog singerLog, Path fullAddedPath) {
+    this(singerLog, fullAddedPath.toFile().getName());
+    this.dir = SingerUtils.extractDirectoryPath(fullAddedPath.toString());
+    this.logStreamDescriptor = String.format("%s:%s:%s", singerLog.getLogName(), fileNamePrefix, this.dir);
+    this.fullPathPrefix = FilenameUtils.concat(dir, fileNamePrefix);
+  }
+
   public LogStream(SingerLog singerLog, String fileNamePrefix) {
     Preconditions.checkArgument(!Strings.isNullOrEmpty(fileNamePrefix));
 
@@ -129,7 +147,8 @@ public LogStream(SingerLog singerLog, String fileNamePrefix) {
     this.logFilePathsIndex = new HashMap<>();
     this.lastStreamModificationTime = -1L;
 
-    String dir = this.singerLog.getSingerLogConfig().getLogDir();
+    this.dir = this.singerLog.getSingerLogConfig().getLogDir();
+    this.logStreamDescriptor = String.format("%s:%s:%s", singerLog.getLogName(), fileNamePrefix, this.dir);
     this.fileNameMatchMode = singerLog.getSingerLogConfig().getFilenameMatchMode();
     this.fileNamePrefix = fileNamePrefix;
     this.fullPathPrefix = FilenameUtils.concat(dir, fileNamePrefix);
@@ -147,7 +166,7 @@ public int size() {
   public void initialize() throws IOException {
     SingerLogConfig singerLogConfig = singerLog.getSingerLogConfig();
     String regexStr = fileNamePrefix;
-    File logDir = new File(singerLogConfig.getLogDir());
+    File logDir = new File(dir);
 
     if (singerLogConfig.getFilenameMatchMode() == FileNameMatchMode.PREFIX) {
       regexStr += ".*";
@@ -283,7 +302,7 @@ public void setLatestProcessedMessageTime(long timestamp) {
   }
 
   public String getLogDir() {
-    return this.singerLog.getSingerLogConfig().getLogDir();
+    return this.dir;
   }
 
   /**

singer/src/main/java/com/pinterest/singer/common/SingerSettings.java

@@ -27,10 +27,12 @@
 import com.pinterest.singer.monitor.LogStreamManager;
 import com.pinterest.singer.thrift.configuration.SingerConfig;
 import com.pinterest.singer.thrift.configuration.SingerLogConfig;
+import com.pinterest.singer.utils.SingerUtils;
 import com.pinterest.singer.writer.KafkaProducerMetricsMonitor;
 import com.twitter.ostrich.stats.Stats;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,6 +43,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.SortedMap;
@@ -92,7 +95,7 @@ public final class SingerSettings {
   public static HeartbeatGenerator heartbeatGenerator;
 
   // initialized here so that unit tests aren't required to call the initialize method below
-  private static SortedMap<String, Collection<SingerLogConfig>> logConfigMap = new TreeMap<>();
+  private static SortedMap<Pair<Integer, String>, Collection<SingerLogConfig>> logConfigMap = new TreeMap<>();
 
   //environment is production by default
   private static Environment environment = new Environment();
@@ -249,17 +252,18 @@ public static Method getLogMonitorStaticInstanceMethod(String monitorClassName)
     return getInstance;
   }
 
-  public static SortedMap<String, Collection<SingerLogConfig>> loadLogConfigMap(SingerConfig config) {
-    SortedMap<String, Collection<SingerLogConfig>> logConfigMap = new TreeMap<>();
-    if(config.getLogConfigs()!=null) {
-        for (SingerLogConfig singerLogConfig : config.getLogConfigs()) {
-            Collection<SingerLogConfig> collection = logConfigMap.get(singerLogConfig.getLogDir());
-            if (collection == null) {
-                collection = new ArrayList<>();
-                logConfigMap.put(singerLogConfig.getLogDir(), collection);
-            }
-            collection.add(singerLogConfig);
-        }
+  public static SortedMap<Pair<Integer, String>, Collection<SingerLogConfig>> loadLogConfigMap(SingerConfig config) {
+    SortedMap<Pair<Integer, String>, Collection<SingerLogConfig>> logConfigMap = new TreeMap<>();
+    List<SingerLogConfig> logConfigs = config.getLogConfigs();
+    if (logConfigs == null) {
+      return logConfigMap;
+    }
+    for (SingerLogConfig singerLogConfig : logConfigs) {
+      List<String> directories = SingerUtils.splitString(singerLogConfig.getLogDir());
+      for (String logDir : directories) {
+        logConfigMap.computeIfAbsent(Pair.of(logDir.split("/").length - 1, logDir),
+            k -> new ArrayList<>()).add(singerLogConfig);
+      }
     }
     return logConfigMap;
   }
@@ -334,7 +338,7 @@ public static void reset() {
     fsMonitorMap.clear();
   }
 
-  public static SortedMap<String, Collection<SingerLogConfig>> getLogConfigMap() {
+  public static SortedMap<Pair<Integer, String>, Collection<SingerLogConfig>> getLogConfigMap() {
     return logConfigMap;
   }