YARN-9629. Support configurable MIN_LOG_ROLLING_INTERVAL. Contributed by Adam Antal.

Author: szilard-nemeth

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -1411,6 +1411,14 @@ public static boolean isAclEnabled(Configuration conf) {
   public static final long
       DEFAULT_NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS = -1;
 
+  /**
+   * The allowed hard minimum limit for {@link
+   * YarnConfiguration#NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS}.
+   */
+  public static final String MIN_LOG_ROLLING_INTERVAL_SECONDS = NM_PREFIX
+      + "log-aggregation.roll-monitoring-interval-seconds.min";
+  public static final long MIN_LOG_ROLLING_INTERVAL_SECONDS_DEFAULT = 3600;
+
   /**
    * Define how many aggregated log files per application per NM we can have
    * in remote file system.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -3219,14 +3219,25 @@
   <property>
     <description>Defines how often NMs wake up to upload log files.
     The default value is -1. By default, the logs will be uploaded when
-    the application is finished. By setting this configure, logs can be uploaded
-    periodically when the application is running. The minimum rolling-interval-seconds
-    can be set is 3600.
+    the application is finished. By setting this configuration logs can
+    be uploaded periodically while the application is running.
+    The minimum positive accepted value can be configured by the setting
+    "yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds.min".
     </description>
     <name>yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds</name>
     <value>-1</value>
   </property>
 
+  <property>
+    <description>Defines the positive minimum hard limit for
+    "yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds".
+    If this configuration has been set less than its default value (3600)
+    the NodeManager may raise a warning.
+    </description>
+    <name>yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds.min</name>
+    <value>3600</value>
+  </property>
+
   <property>
     <description>Define how many aggregated log files per application per NM
     we can have in remote file system. By default, the total number of

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/LogAggregationService.java

@@ -71,7 +71,6 @@ public class LogAggregationService extends AbstractService implements
 
   private static final Logger LOG =
        LoggerFactory.getLogger(LogAggregationService.class);
-  private static final long MIN_LOG_ROLLING_INTERVAL = 3600;
   // This configuration is for debug and test purpose. By setting
   // this configuration as true. We can break the lower bound of
   // NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS.
@@ -106,40 +105,60 @@ public LogAggregationService(Dispatcher dispatcher, Context context,
     this.invalidTokenApps = ConcurrentHashMap.newKeySet();
   }
 
+  private static long calculateRollingMonitorInterval(Configuration conf) {
+    long interval = conf.getLong(
+        YarnConfiguration.NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS,
+        YarnConfiguration.
+            DEFAULT_NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS);
+
+    if (interval <= 0) {
+      LOG.info("rollingMonitorInterval is set as " + interval
+          + ". The log rolling monitoring interval is disabled. "
+          + "The logs will be aggregated after this application is finished.");
+    } else {
+      boolean logAggregationDebugMode =
+          conf.getBoolean(NM_LOG_AGGREGATION_DEBUG_ENABLED, false);
+      long minRollingMonitorInterval = conf.getLong(
+          YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS,
+          YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS_DEFAULT);
+
+      boolean warnHardMinLimitLowerThanDefault = minRollingMonitorInterval <
+          YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS_DEFAULT &&
+          !logAggregationDebugMode;
+      if (warnHardMinLimitLowerThanDefault) {
+        LOG.warn("{} has been set to {}, which is less than the default "
+            + "minimum value {}. This may impact NodeManager's performance.",
+            YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS,
+            minRollingMonitorInterval,
+            YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS_DEFAULT);
+      }
+      boolean lowerThanHardLimit = interval < minRollingMonitorInterval;
+      if (lowerThanHardLimit) {
+        if (logAggregationDebugMode) {
+          LOG.info("Log aggregation debug mode enabled. " +
+              "Skipped checking minimum limit.");
+        } else {
+          LOG.warn("rollingMonitorInterval should be more than " +
+              "or equal to {} seconds. Using {} seconds instead.",
+              minRollingMonitorInterval, minRollingMonitorInterval);
+          interval = minRollingMonitorInterval;
+        }
+      }
+    }
+    return interval;
+  }
+
   protected void serviceInit(Configuration conf) throws Exception {
     int threadPoolSize = getAggregatorThreadPoolSize(conf);
     this.threadPool = HadoopExecutors.newFixedThreadPool(threadPoolSize,
         new ThreadFactoryBuilder()
             .setNameFormat("LogAggregationService #%d")
             .build());
 
-    rollingMonitorInterval = conf.getLong(
-        YarnConfiguration.NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS,
-        YarnConfiguration.DEFAULT_NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS);
-
-    boolean logAggregationDebugMode =
-        conf.getBoolean(NM_LOG_AGGREGATION_DEBUG_ENABLED, false);
-
-    if (rollingMonitorInterval > 0
-        && rollingMonitorInterval < MIN_LOG_ROLLING_INTERVAL) {
-      if (logAggregationDebugMode) {
-        LOG.info("Log aggregation debug mode enabled. rollingMonitorInterval = "
-            + rollingMonitorInterval);
-      } else {
-        LOG.warn("rollingMonitorInterval should be more than or equal to {} " +
-                "seconds. Using {} seconds instead.",
-                MIN_LOG_ROLLING_INTERVAL, MIN_LOG_ROLLING_INTERVAL);
-        this.rollingMonitorInterval = MIN_LOG_ROLLING_INTERVAL;
-      }
-    } else if (rollingMonitorInterval <= 0) {
-      LOG.info("rollingMonitorInterval is set as " + rollingMonitorInterval
-          + ". The log rolling monitoring interval is disabled. "
-          + "The logs will be aggregated after this application is finished.");
-    } else {
-      LOG.info("rollingMonitorInterval is set as " + rollingMonitorInterval
-          + ". The logs will be aggregated every " + rollingMonitorInterval
-          + " seconds");
-    }
+    rollingMonitorInterval = calculateRollingMonitorInterval(conf);
+    LOG.info("rollingMonitorInterval is set as {}. The logs will be " +
+        "aggregated every {} seconds", rollingMonitorInterval,
+        rollingMonitorInterval);
 
     super.serviceInit(conf);
   }
@@ -413,6 +432,10 @@ public NodeId getNodeId() {
     return this.nodeId;
   }
 
+  @VisibleForTesting
+  public long getRollingMonitorInterval() {
+    return rollingMonitorInterval;
+  }
 
   private int getAggregatorThreadPoolSize(Configuration conf) {
     int threadPoolSize;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/logaggregation/TestLogAggregationService.java

@@ -2641,4 +2641,43 @@ public List<String> getLogFileTypesInLastCycle() {
       return this.logFileTypesInLastCycle;
     }
   }
+
+  @Test
+  public void testRollingMonitorIntervalDefault() {
+    LogAggregationService logAggregationService =
+        new LogAggregationService(dispatcher, this.context, this.delSrvc,
+            super.dirsHandler);
+    logAggregationService.init(this.conf);
+
+    long interval = logAggregationService.getRollingMonitorInterval();
+    assertEquals(-1L, interval);
+  }
+
+  @Test
+  public void testRollingMonitorIntervalGreaterThanSet() {
+    this.conf.set(YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS, "1800");
+    this.conf.set(YarnConfiguration
+        .NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS, "2700");
+    LogAggregationService logAggregationService =
+            new LogAggregationService(dispatcher, this.context, this.delSrvc,
+                    super.dirsHandler);
+    logAggregationService.init(this.conf);
+
+    long interval = logAggregationService.getRollingMonitorInterval();
+    assertEquals(2700L, interval);
+  }
+
+  @Test
+  public void testRollingMonitorIntervalLessThanSet() {
+    this.conf.set(YarnConfiguration.MIN_LOG_ROLLING_INTERVAL_SECONDS, "1800");
+    this.conf.set(YarnConfiguration
+        .NM_LOG_AGGREGATION_ROLL_MONITORING_INTERVAL_SECONDS, "600");
+    LogAggregationService logAggregationService =
+            new LogAggregationService(dispatcher, this.context, this.delSrvc,
+                    super.dirsHandler);
+    logAggregationService.init(this.conf);
+
+    long interval = logAggregationService.getRollingMonitorInterval();
+    assertEquals(1800L, interval);
+  }
 }