/**
 * Copyright 2019 Pinterest, Inc.
 * 
 * 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.pinterest.singer.utils;

import com.pinterest.singer.common.SingerConfigDef;
import com.pinterest.singer.config.ConfigFileServerSet;
import com.pinterest.singer.config.ConfigFileWatcher;
import com.pinterest.singer.config.DirectorySingerConfigurator;
import com.pinterest.singer.environment.EnvironmentProvider;
import com.pinterest.singer.loggingaudit.client.utils.ConfigUtils;
import com.pinterest.singer.loggingaudit.thrift.LoggingAuditStage;
import com.pinterest.singer.loggingaudit.thrift.configuration.LoggingAuditClientConfig;
import com.pinterest.singer.loggingaudit.thrift.configuration.AuditConfig;
import com.pinterest.singer.metrics.StatsPusher;
import com.pinterest.singer.thrift.configuration.AdminConfig;
import com.pinterest.singer.thrift.configuration.MessageTransformerConfig;
import com.pinterest.singer.thrift.configuration.NoOpWriteConfig;
import com.pinterest.singer.thrift.configuration.FileNameMatchMode;
import com.pinterest.singer.thrift.configuration.HeartbeatWriterConfig;
import com.pinterest.singer.thrift.configuration.KafkaProducerConfig;
import com.pinterest.singer.thrift.configuration.KafkaWriterConfig;
import com.pinterest.singer.thrift.configuration.KubeConfig;
import com.pinterest.singer.thrift.configuration.LogMonitorConfig;
import com.pinterest.singer.thrift.configuration.LogStreamProcessorConfig;
import com.pinterest.singer.thrift.configuration.LogStreamReaderConfig;
import com.pinterest.singer.thrift.configuration.LogStreamWriterConfig;
import com.pinterest.singer.thrift.configuration.MemqAuditorConfig;
import com.pinterest.singer.thrift.configuration.MemqWriterConfig;
import com.pinterest.singer.thrift.configuration.PulsarProducerConfig;
import com.pinterest.singer.thrift.configuration.PulsarWriterConfig;
import com.pinterest.singer.thrift.configuration.ReaderType;
import com.pinterest.singer.thrift.configuration.RealpinObjectType;
import com.pinterest.singer.thrift.configuration.RealpinWriterConfig;
import com.pinterest.singer.thrift.configuration.RegexBasedModifierConfig;
import com.pinterest.singer.thrift.configuration.S3WriterConfig;
import com.pinterest.singer.thrift.configuration.SamplingType;
import com.pinterest.singer.thrift.configuration.SingerConfig;
import com.pinterest.singer.thrift.configuration.SingerLogConfig;
import com.pinterest.singer.thrift.configuration.SingerRestartConfig;
import com.pinterest.singer.thrift.configuration.TextLogMessageType;
import com.pinterest.singer.thrift.configuration.TextReaderConfig;
import com.pinterest.singer.thrift.configuration.ThriftReaderConfig;
import com.pinterest.singer.thrift.configuration.TransformType;
import com.pinterest.singer.thrift.configuration.WriterType;

import com.amazonaws.regions.Regions;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.twitter.common.base.MorePreconditions;
import com.twitter.util.Function;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.file.FileSystems;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.PathMatcher;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.Collections;
import java.util.HashMap;
import org.apache.commons.configuration.AbstractConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.configuration.SubsetConfiguration;
import org.apache.commons.configuration.ConfigurationUtils;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.record.CompressionType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.services.s3.model.ObjectCannedACL;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
import java.util.Iterator;
import java.io.StringReader;
import java.util.HashSet;
import java.util.Arrays;
import java.util.stream.Collectors;

/**
 * Utility class for parsing singer related config file. The following is a
 * sample singer configuration file.
 *
 * # /etc/singer/singer.properties # singer.threadPoolSize = 20
 * singer.ostrichPort = 2047
 *
 * # Configuration for LogMonitor. singer.monitor.monitorIntervalInSecs = 10
 *
 * # Watcher interval secs singer.logConfigPollIntervalSecs = 10
 *
 * # stats pusher host ostrichPort singer.statsPusherHostPort = localhost:18126
 *
 * # whether to allow singer to restart itself # the number of errors that
 * singer needs to encounter before restarting # # Singer will restart itself
 * randomly in the time range, in UTC # singer.restart.onFailures=true
 * singer.restart.numberOfFailuresAllowed=100
 *
 * singer.restart.daily=true singer.restart.dailyRestartUtcTimeRangeBegin=02:30
 * singer.restart.dailyRestartUtcTimeRangeEnd=03:30
 *
 * # singer heartbeat configuration
 *
 * singer.heartbeat.intervalInSeconds = 60 singer.heartbeat.writer.writerType =
 * kafka singer.heartbeat.writer.kafka.topic = singer_heartbeat
 * singer.heartbeat.writer.kafka.producerConfig.metadata.broker.serverset =
 * /discovery/datakafka01/prod
 * singer.heartbeat.writer.kafka.producerConfig.acks = 1
 */

public class LogConfigUtils {

  private static final int MAX_SEED_BROKERS = 5;
  public static final String DEFAULT_SHADOW_SERVERSET = "default";
  private static final Logger LOG = LoggerFactory.getLogger(LogConfigUtils.class);
  public static String DEFAULT_SERVERSET_DIR = "/var/serverset";
  private static final String DEFAULT_ACKS = "1";
  private static final String ACKS_ALL = "all";
  private static final long MaximumProcessingTimeSliceInMilliseconds = 864000000L;
  private static final ConcurrentMap<String, Set<String>> KAFKA_SERVER_SETS = Maps
      .newConcurrentMap();
  private static final Set<String> MEMQ_SERVER_SETS = ConcurrentHashMap.newKeySet();
  public static final Properties SHADOW_SERVERSET_MAPPING = new Properties();
  // Minimum value that can be set for S3WriterConfig.minUploadTimeInSeconds
  private static final int MIN_UPLOAD_TIME_IN_SECONDS = 30;
  // Minimum value that can be set for S3WriterConfig.maxFileSizeInMB
  private static final int MIN_MAX_FILE_SIZE_IN_MB = 5;
  // Minimum value that can be set for S3WriterConfig.maxRetries
  private static final int MIN_MAX_RETRIES = 5;
  public static boolean SHADOW_MODE_ENABLED;
  // this is used to check if the wildcard characters are present in dir config provided by user
  public static final Pattern WILDCARD_SUPPORTED_CHARS = Pattern.compile("[*?\\]\\[{}]");

  private LogConfigUtils() {
  }

  public static SingerConfig parseDirBasedSingerConfigHeader(String singerConfigFile) throws ConfigurationException {
    List<String> errors = new ArrayList<>();
    PropertiesConfiguration configHeader = new PropertiesConfiguration(singerConfigFile);
    AbstractConfiguration singerConfiguration = new SubsetConfiguration(configHeader,
        SingerConfigDef.SINGER_CONFIGURATION_PREFIX);

    SingerConfig result;
    try {
      result = parseCommonSingerConfigHeader(singerConfiguration);
    } catch (NoSuchElementException x) {
      result = new SingerConfig();
      errors
          .add(SingerConfigDef.MONITOR_INTERVAL_IN_SECS + " is required for Singer Configuration");
    }
    if (singerConfiguration.containsKey("logConfigPollIntervalSecs")) {
      result.setLogConfigPollIntervalSecs(singerConfiguration.getInt("logConfigPollIntervalSecs"));
    } else {
      errors.add("singer.setLogConfigPollIntervalSecs " + "is required for Singer Configuration");
    }

    try {
      result.singerRestartConfig = parseSingerRestartConfig(configHeader);
    } catch (ConfigurationException x) {
      errors.add(x.getMessage());
    }
    try {
      // if running in Kubernetes mode then set the pod directory to an empty string
      // if this line is removed and the configuration has a kubernetes log directory;
      // Singer will point to the wrong log path; with the kubernetes log path
      // pre-pended to the path
      if (result.isKubernetesEnabled()) {
        result.setKubeConfig(parseKubeConfig(configHeader));
      }
    } catch (ConfigurationException e) {
      errors.add(e.getMessage());
    }

    try {
      if(result.isAdminEnabled()) {
        result.setAdminConfig(parseAdminConfig(new SubsetConfiguration(configHeader, SingerConfigDef.SINGER_ADMIN_CONFIG_PREFIX)));
      }
    } catch (ConfigurationException e) {
      errors.add(e.getMessage());
    }
    // check if heartbeat is enabled, don't auto enable if configuration is set
    // by default heartbeat is enabled in thrift so it will need to be explicitly
    // disabled
    if (result.isHeartbeatEnabled()) {
      try {
        AbstractConfiguration heartbeatConfig = new SubsetConfiguration(configHeader,
            "singer.heartbeat.");
        int intervalInSeconds = heartbeatConfig.getInt("intervalInSeconds");
        result.setHeartbeatIntervalInSeconds(intervalInSeconds);

        AbstractConfiguration heartbeatWriterConfig = new SubsetConfiguration(heartbeatConfig,
            "writer.");
        String writerType = heartbeatWriterConfig.getString("writerType");
        WriterType type = WriterType.valueOf(writerType.toUpperCase());
        HeartbeatWriterConfig writerConfig = new HeartbeatWriterConfig(type);
        switch (type) {
        case KAFKA08:
        case KAFKA:
          try {
            KafkaWriterConfig kafkaWriterConf = parseKafkaWriterConfig(
                new SubsetConfiguration(heartbeatWriterConfig, writerType + "."));
            writerConfig.setKafkaWriterConfig(kafkaWriterConf);
          } catch (ConfigurationException x) {
            errors.add(x.getMessage());
          }
          break;
        default:
          errors.add("Unsupported heartbeat writer type.");
        }
        result.setHeartbeatWriterConfig(writerConfig);
        result.setHeartbeatEnabled(true);
      } catch (java.util.NoSuchElementException e) {
        LOG.error("singer.heartbeat is not configured correctly.", e);
        result.setHeartbeatEnabled(false);
      }
    }

    if (result.isLoggingAuditEnabled()){
      AbstractConfiguration loggingAuditConf = new SubsetConfiguration(configHeader, "singer.loggingAudit.");
      try{
        LoggingAuditClientConfig loggingAuditClientConfig = ConfigUtils.parseLoggingAuditClientConfig(loggingAuditConf, "");
        if (loggingAuditClientConfig.getStage() != LoggingAuditStage.SINGER){
          LOG.warn("LoggingAudit stage should be SINGER.");
          loggingAuditClientConfig.setStage(LoggingAuditStage.SINGER);
        }
        result.setLoggingAuditClientConfig(loggingAuditClientConfig);
      } catch (ConfigurationException e){
        LOG.error("loggingAudit is not configured correctly.", e);
        result.setLoggingAuditEnabled(false);
      }
    }

    if (errors.size() > 0) {
      String errorMessage = Joiner.on('\n').join(errors);
      throw new ConfigurationException(errorMessage);
    }
    return result;
  }

  private static KubeConfig parseKubeConfig(PropertiesConfiguration configHeader) throws ConfigurationException {
    KubeConfig config = new KubeConfig();
    AbstractConfiguration subsetConfig = new SubsetConfiguration(configHeader,
        SingerConfigDef.SINGER_KUBE_CONFIG_PREFIX);
    if (subsetConfig.containsKey(SingerConfigDef.KUBE_POLL_FREQUENCY_SECONDS)) {
      config.setPollFrequencyInSeconds(
          subsetConfig.getInt(SingerConfigDef.KUBE_POLL_FREQUENCY_SECONDS));
    }

    if (subsetConfig.containsKey(SingerConfigDef.POD_METADATA_FIELDS)) {
      config.setPodMetadataFields(subsetConfig.getList(SingerConfigDef.POD_METADATA_FIELDS).stream()
          .map(Object::toString)
          .collect(Collectors.toList()));
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_POD_LOG_DIR)) {
      String logDirectory = subsetConfig.getString(SingerConfigDef.KUBE_POD_LOG_DIR);
      // normalize path before setting the property
      logDirectory = new File(logDirectory).toPath().normalize().toString() + "/";
      if (!logDirectory.isEmpty() && !logDirectory.endsWith("/")) {
        logDirectory += "/";
      }
      config.setPodLogDirectory(logDirectory);
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_DEFAULT_DELETION_TIMEOUT)) {
      config.setDefaultDeletionTimeoutInSeconds(
          subsetConfig.getInt(SingerConfigDef.KUBE_DEFAULT_DELETION_TIMEOUT));
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_KUBELET_PORT)) {
      config.setKubeletPort(subsetConfig.getString(SingerConfigDef.KUBE_KUBELET_PORT));
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_USE_SECURE_CONNECTION)
        && subsetConfig.getBoolean(SingerConfigDef.KUBE_USE_SECURE_CONNECTION)) {
      config.setUseSecureConnection(
          subsetConfig.getBoolean(SingerConfigDef.KUBE_USE_SECURE_CONNECTION));
      if (!subsetConfig.containsKey(SingerConfigDef.KUBE_SERVICE_ACCOUNT_TOKEN_PATH) ||
          !subsetConfig.containsKey(SingerConfigDef.KUBE_SERVICE_ACCOUNT_CA_CERT_PATH)) {
        throw new ConfigurationException("When using secure connection, "
            + SingerConfigDef.KUBE_SERVICE_ACCOUNT_TOKEN_PATH + " must be set.");
      }
      // Validate the service account token and CA cert paths
      String tokenPath = subsetConfig.getString(SingerConfigDef.KUBE_SERVICE_ACCOUNT_TOKEN_PATH);
      File tokenFile = new File(tokenPath);
      if (!tokenFile.exists() || !tokenFile.isFile()) {
        throw new ConfigurationException(
            "Service account token file does not exist: " + tokenPath);
      }
      config.setServiceAccountTokenPath(tokenPath);
      String caCertPath = subsetConfig.getString(
          SingerConfigDef.KUBE_SERVICE_ACCOUNT_CA_CERT_PATH);
      File caCertFile = new File(caCertPath);
      if (!caCertFile.exists() || !caCertFile.isFile()) {
        throw new ConfigurationException(
            "Service account CA cert file does not exist: " + caCertPath);
      }
      config.setServiceAccountCaCertPath(caCertPath);
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_SERVICE_ACCOUNT_TOKEN_PATH)) {
      String tokenPath = subsetConfig.getString(SingerConfigDef.KUBE_SERVICE_ACCOUNT_TOKEN_PATH);
      File tokenFile = new File(tokenPath);
      if (!tokenFile.exists() || !tokenFile.isFile()) {
        throw new ConfigurationException("Service account token file does not exist: " + tokenPath);
      }
      config.setServiceAccountTokenPath(tokenPath);
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_SERVICE_ACCOUNT_CA_CERT_PATH)) {
      String caCertPath = subsetConfig.getString(SingerConfigDef.KUBE_SERVICE_ACCOUNT_CA_CERT_PATH);
      File caCertFile = new File(caCertPath);
      if (!caCertFile.exists() || !caCertFile.isFile()) {
        throw new ConfigurationException("Service account CA cert file does not exist: " + caCertPath);
      }
      config.setServiceAccountCaCertPath(caCertPath);
    }

    if (subsetConfig.containsKey(SingerConfigDef.KUBE_ENABLE_POD_LOG_DIRECTORY_CLEANUP)) {
        config.setEnablePodLogDirectoryCleanup(
            subsetConfig.getBoolean(SingerConfigDef.KUBE_ENABLE_POD_LOG_DIRECTORY_CLEANUP));
    }
    return config;
  }

  private static AdminConfig parseAdminConfig(SubsetConfiguration configuration) throws ConfigurationException {
    AdminConfig adminConfig = new AdminConfig();
    if (configuration.containsKey(SingerConfigDef.ADMIN_SOCKET_FILE)) {
      adminConfig.setSocketFile(configuration.getString(SingerConfigDef.ADMIN_SOCKET_FILE));
    }

    if (configuration.containsKey(SingerConfigDef.ADMIN_ALLOWED_UIDS)) {
      List<Long> allowedUids = Arrays.stream(configuration.getString(SingerConfigDef.ADMIN_ALLOWED_UIDS).split(",")).map(Long::parseLong).collect(Collectors.toList());
      adminConfig.setAllowedUids(allowedUids);
    }

    if (configuration.containsKey(SingerConfigDef.ADMIN_DEFAULT_DELETION_TIMEOUT)) {
      adminConfig.setDefaultDeletionTimeoutInSeconds(configuration.getInt(SingerConfigDef.ADMIN_DEFAULT_DELETION_TIMEOUT));
    }

    if (configuration.containsKey(SingerConfigDef.ADMIN_DELETION_CHECK_INTERVAL)) {
      adminConfig.setDeletionCheckIntervalInSeconds(configuration.getInt(SingerConfigDef.ADMIN_DELETION_CHECK_INTERVAL));
    }
    return adminConfig;
  }

  public static MemqWriterConfig parseMemqWriterConfig(SubsetConfiguration configuration) throws ConfigurationException {
    configuration.setThrowExceptionOnMissing(false);
    MemqWriterConfig config = new MemqWriterConfig();
    config.setAuditingEnabled(false);
    if (!configuration.containsKey("cluster")) {
      throw new ConfigurationException("Missing cluster name");
    }

    if (!configuration.containsKey("environment")) {
      throw new ConfigurationException("Missing environment name");
    }

    String clustername = configuration.getString("cluster");
    config.setCluster(clustername);
    String environment = configuration.getString("environment");
    if (!Arrays.asList("dev", "test", "prod").contains(environment)) {
      throw new ConfigurationException("Invalid environment name");
    }
    // e.g. discovery.memq.dev.prototype.prod_rich_data
    String serverSetFilePath = DEFAULT_SERVERSET_DIR + "/discovery.memq." + environment + "." + clustername
        + ".prod_rich_data";
    config.setServerset(serverSetFilePath);
    
    File serversetFile = new File(serverSetFilePath);
    if (!serversetFile.exists()) {
      throw new ConfigurationException("Serverset doesn't exist:" + serverSetFilePath);
    }
    
    if (!configuration.containsKey(SingerConfigDef.TOPIC)) {
      throw new ConfigurationException("Missing topic name");
    }
    config.setTopic(configuration.getString(SingerConfigDef.TOPIC));
    
    try {
      if (!MEMQ_SERVER_SETS.contains(serverSetFilePath)) {
        final byte[] serversetBytes = Files.readAllBytes(serversetFile.toPath());
        ConfigFileWatcher watcher = ConfigFileWatcher.defaultInstance();
        watcher.addWatch(serverSetFilePath, new Function<byte[], Void>() {

          @Override
          public Void apply(byte[] data) {
            if (!Arrays.equals(serversetBytes, data)) {
              LOG.warn("Serverset:" + serverSetFilePath + " updated, Singer will restart");
              System.exit(0);
            }
            return null;
          }
        });
        MEMQ_SERVER_SETS.add(serverSetFilePath);
      }
    } catch (IOException e) {
      throw new ConfigurationException("Failed to load serverset");
    }
    
    config.setMaxInFlightRequests(configuration.getInt("maxInFlightRequests", 1));
    if (configuration.containsKey("maxPayLoadBytes")) {
      config.setMaxPayLoadBytes(configuration.getInt("maxPayLoadBytes"));
    }
    if (configuration.containsKey("compression")) {
      config.setCompression(configuration.getString("compression").toUpperCase());
    }
    if (configuration.containsKey("disableAcks")) {
      config.setDisableAcks(configuration.getBoolean("disableAcks"));
    }
    if (configuration.containsKey("ackCheckPollInterval")) {
      config.setAckCheckPollInterval(configuration.getInt("ackCheckPollInterval"));
    }
    if (configuration.containsKey("clientType")) {
      config.setClientType(configuration.getString("clientType").toUpperCase());
    }
    if (configuration.containsKey("auditor.enabled")) {
      parseMemqAuditorConfigs(config, new SubsetConfiguration(configuration, "auditor"));
    }
    if (configuration.containsKey("maxInFlightRequestsMemoryBytes")) {
      config.setMaxInFlightRequestsMemoryBytes(configuration.getInt("maxInFlightRequestsMemoryBytes"));
    }
    if (configuration.containsKey("maxBlockMs")) {
      config.setMaxBlockMs(configuration.getInt("maxBlockMs"));
    }
    if (configuration.containsKey("numWriteEndpoints")) {
      config.setNumWriteEndpoints(configuration.getInt("numWriteEndpoints"));
    }
    return config;
  }
  
  public static void parseMemqAuditorConfigs(MemqWriterConfig config,
                                         AbstractConfiguration configuration) throws ConfigurationException {
    if (configuration == null || configuration.isEmpty()) {
      LOG.warn("Memq auditor configuration is empty, auditing will not be enabled for:"
          + config.getTopic());
      return;
    }
    if (!configuration.getBoolean(".enabled")) {
      return;
    }
    MemqAuditorConfig auditorConfig = new MemqAuditorConfig();
    auditorConfig.setAuditorClass("com.pinterest.memq.client.commons.audit.KafkaBackedAuditor");
    if (!configuration.containsKey(".topic")) {
      throw new ConfigurationException("Missing auditor topic");
    }
    auditorConfig.setTopic(configuration.getString(".topic"));
    if (!configuration.containsKey(".serverset")) {
      throw new ConfigurationException("Missing auditor serverset");
    }
    auditorConfig.setServerset(configuration.getString(".serverset"));
    config.setAuditorConfig(auditorConfig);
  }

  /**
   * Singer can restart itself if # of failures exceeds threshold, and in daily
   * cadence. The following is singer restart related configuration:
   *
   * singer.restart.onFailures=true singer.restart.numberOfFailuresAllowed=100
   * singer.restart.daily=true singer.restart.dailyRestartUtcTimeRangeBegin=02:30
   * singer.restart.dailyRestartUtcTimeRangeEnd=03:30
   */
  private static SingerRestartConfig parseSingerRestartConfig(PropertiesConfiguration configHeader) throws ConfigurationException {
    SingerRestartConfig restartConfig = new SingerRestartConfig();
    AbstractConfiguration subsetConfig = new SubsetConfiguration(configHeader,
        SingerConfigDef.SINGER_RESTART_PREFIX);

    if (subsetConfig.containsKey(SingerConfigDef.ON_FAILURES)) {
      restartConfig.restartOnFailures = subsetConfig.getBoolean(SingerConfigDef.ON_FAILURES);
    }
    if (subsetConfig.containsKey(SingerConfigDef.NUMBER_OF_FAILURES_ALLOWED)) {
      restartConfig.numOfFailuesAllowed = subsetConfig
          .getInt(SingerConfigDef.NUMBER_OF_FAILURES_ALLOWED);
    }
    if (subsetConfig.containsKey(SingerConfigDef.DAILY_RESTART_FLAG)) {
      restartConfig.restartDaily = subsetConfig.getBoolean(SingerConfigDef.DAILY_RESTART_FLAG);
    }

    if (restartConfig.restartDaily) {
      if (!subsetConfig.containsKey(SingerConfigDef.DAILY_RESTART_TIME_BEGIN)
          || !subsetConfig.containsKey(SingerConfigDef.DAILY_RESTART_TIME_END)) {
        throw new ConfigurationException("Daily restart time range is not set correctly");
      }

      restartConfig.dailyRestartUtcTimeRangeBegin = subsetConfig
          .getString(SingerConfigDef.DAILY_RESTART_TIME_BEGIN);
      restartConfig.dailyRestartUtcTimeRangeEnd = subsetConfig
          .getString(SingerConfigDef.DAILY_RESTART_TIME_END);
      Date startTime = SingerUtils.convertToDate(restartConfig.dailyRestartUtcTimeRangeBegin);
      Date endTime = SingerUtils.convertToDate(restartConfig.dailyRestartUtcTimeRangeEnd);
      if (endTime.compareTo(startTime) <= 0) {
        throw new ConfigurationException("Daily restart end time is not later than start time");
      }
    }
    return restartConfig;
  }

  /**
   * Finds full-path directories matching the given list of paths that may include wildcards
   *
   * @param pathsWithWildcards List of directory paths that may include wildcard characters  (e.g. /mnt/thrift_logger/*)
   * @return A set of fully resolved directory paths matching any wildcard expressions.
   *      If a path does not contain wildcards, it will be included in the result directly.
   * @throws IllegalArgumentException If any path is not a valid directory path.
   *
   * For example, given the directory structure:
   *     /var/logs/app1/
   *     /var/logs/app2/
   *     /mnt/logs/tmp1/
   *
   * Calling findDirectories(Arrays.asList("/var/logs/*", /mnt/thrift_logs/*")) will return:
   * Set { "/var/logs/app1/", "/var/logs/app2/", "/mnt/logs/tmp1/" }
   */
  public static Set<String> findDirectories(List<String> pathsWithWildcards) {
    Set<String> result = new HashSet<>();
    for (String pathAndWildcard : pathsWithWildcards) {
      if (!WILDCARD_SUPPORTED_CHARS.matcher(pathAndWildcard).find()) {
        result.add(pathAndWildcard);
      } else {
        // Compute base path for wildcard directory matching
        String [] dirParts = pathAndWildcard.split("/");
        StringBuilder basePath = new StringBuilder();
        for (String part : dirParts) {
          if (WILDCARD_SUPPORTED_CHARS.matcher(part).find()) {
            break;
          }
          basePath.append(part).append("/");
        }
        result.addAll(wildcardDirectoryMatcher(pathAndWildcard, Paths.get(basePath.toString())));
      }
    }
    return result;
  }

  /**
   * Finds directories matching the given wildcard pattern within the specified base path.
   * Uses Files.walkFileTree to traverse the directory tree and match against the wildcard pattern.
   *
   * @param pathAndWildcard The path with wildcard to match against.
   * @param basePath The base directory to start searching from.
   * @return A set of matching directory paths.
   */
  public static Set<String> wildcardDirectoryMatcher(String pathAndWildcard, Path basePath) {
    Set<String> matchingDirectories = new HashSet<>();
    PathMatcher matcher = FileSystems.getDefault().getPathMatcher("glob:" + pathAndWildcard);
    if (basePath == null || !Files.exists(basePath)) {
      return matchingDirectories;
    }

    try {
      Files.walkFileTree(basePath, new SimpleFileVisitor<Path>() {
        @Override
        public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) {
          File directory = dir.toFile();
          if (directory.isDirectory() && matcher.matches(dir)) {
            matchingDirectories.add(directory.getPath());
          }
          return FileVisitResult.CONTINUE;
        }
      });
    } catch (IOException e) {
      LOG.error("Error while discovering directories in file tree for " + pathAndWildcard, e);
    }
    LOG.info("Wildcard directory matching found " + matchingDirectories.size() + " directories for "
        + pathAndWildcard + ": " + matchingDirectories);
    return matchingDirectories;
  }

  public static SingerConfig parseFileBasedSingerConfig(String singerConfigFile) throws ConfigurationException {
    AbstractConfiguration singerConfiguration = new SubsetConfiguration(
        new PropertiesConfiguration(singerConfigFile), SingerConfigDef.SINGER_CONFIGURATION_PREFIX);
    singerConfiguration.setThrowExceptionOnMissing(true);
    SingerConfig result = parseCommonSingerConfigHeader(singerConfiguration);
    Set<String> logs = Sets.newHashSet(singerConfiguration.getStringArray("logs"));
    for (String log : logs) {
      AbstractConfiguration logConfiguration = new SubsetConfiguration(singerConfiguration,
          log + ".");
      result.addToLogConfigs(LogConfigUtils.parseLogConfig(log, logConfiguration));
    }
    return result;
  }

  public static LogMonitorConfig parseLogMonitorConfig(AbstractConfiguration monitorConfiguration) {
    int monitorIntervalInSecs = monitorConfiguration.getInt("monitorIntervalInSecs");
    return new LogMonitorConfig(monitorIntervalInSecs);
  }

  public static SingerLogConfig parseLogConfigFromFile(File logConfigFile) throws ConfigurationException {
    AbstractConfiguration logConfig = new PropertiesConfiguration(logConfigFile);
    return parseLogConfig(parseLogNamespace(logConfigFile), logConfig);
  }

  public static SingerLogConfig parseLogConfigFromFileWithOverrides(File logConfigFile, List<Consumer<AbstractConfiguration>> overrides) throws ConfigurationException {
    AbstractConfiguration logConfig = new PropertiesConfiguration(logConfigFile);
    for (Consumer<AbstractConfiguration> consumer : overrides) {
      consumer.accept(logConfig);
    }
    return parseLogConfig(parseLogNamespace(logConfigFile), logConfig);
  }

  public static SingerLogConfig[] parseLogStreamConfigFromFile(String logConfigFile) throws ConfigurationException {
    PropertiesConfiguration logConfig = new PropertiesConfiguration();
    logConfig.setDelimiterParsingDisabled(true);
    logConfig.load(new StringReader(logConfigFile));
    return parseLogStreamConfig(logConfig);
  }

  public static String parseLogNamespace(File logConfigFile) {
    return FilenameUtils.getBaseName(logConfigFile.getAbsolutePath());
  }

  public static SingerLogConfig parseLogConfig(String logName,
                                               AbstractConfiguration logConfiguration) throws ConfigurationException {
    logConfiguration.setThrowExceptionOnMissing(true);
    
    long ts = System.currentTimeMillis();

    String local_dir = null;
    if (logConfiguration.containsKey("logDir")) {
      // Retrieve the list of strings associated with the key
      List<Object> list = logConfiguration.getList("logDir");

      // Optionally, convert the List<Object> to List<String> (Apache Commons Configuration
      // can store any type of object, so casting may be necessary)
      List<String> stringList = list.stream()
          .map(Object::toString)
          .collect(Collectors.toList());

      for (int i = 0; i < stringList.size(); i++) {
        String dir = stringList.get(i);
        if (dir.endsWith("/")) {
          stringList.set(i, dir.substring(0, dir.length() - 1));
        }
      }

      LOG.info("Found directories for logDir: {}", stringList);
      List<String> sortedPaths = new ArrayList<>(stringList);
      Collections.sort(sortedPaths); // Sort the list
      local_dir = String.join(",", sortedPaths);
    } else if (logConfiguration.containsKey("local_dir")) {
      local_dir = logConfiguration.getString("local_dir");
    } else {
      throw new ConfigurationException("missing logDir/local_dir");
    }
    if (local_dir != null && local_dir.endsWith("/")) {
      local_dir = local_dir.substring(0, local_dir.length() - 1);
    }

    String logfile_regex;
    if (logConfiguration.containsKey("logStreamRegex")) {
      logfile_regex = logConfiguration.getString("logStreamRegex");
    } else if (logConfiguration.containsKey("logfile_regex")) {
      logfile_regex = logConfiguration.getString("logfile_regex");
    } else {
      throw new ConfigurationException("missing logStreamRegex/logfile_regex");
    }

    try {
      Pattern.compile(logfile_regex);
    } catch (PatternSyntaxException e) {
      throw new ConfigurationException("Invalid logStreamRegex: " + logfile_regex);
    }

    LogStreamProcessorConfig processorConfig = parseLogStreamProcessorConfig(
        new SubsetConfiguration(logConfiguration, "processor."));
    LogStreamReaderConfig readerConfig = parseLogStreamReaderConfig(
        new SubsetConfiguration(logConfiguration, "reader."));
    LogStreamWriterConfig writerConfig = parseLogStreamWriterConfig(
        new SubsetConfiguration(logConfiguration, "writer."));

    // ThriftReader and S3Writer configs not allowed.
    if (readerConfig.isSetThriftReaderConfig() && writerConfig.isSetS3WriterConfig()) {
      throw new ConfigurationException("ThriftReader and S3Writer cannot be used together");
    }

    // initialize the optional fields
