AMBARI-14140 Test and Adopt FIFO compaction policy for AMS high load tables (dsen)

ambari-metrics/ambari-metrics-timelineservice/conf/unix/ambari-metrics-collector

@@ -32,10 +32,17 @@ HBASE_CONF_DIR=/etc/ams-hbase/conf
 HBASE_CMD=${HBASE_DIR}/bin/hbase
 
 METRIC_TABLES=(METRIC_AGGREGATE_DAILY METRIC_AGGREGATE_HOURLY METRIC_AGGREGATE METRIC_RECORD METRIC_RECORD_DAILY METRIC_RECORD_HOURLY METRIC_RECORD_MINUTE)
+METRIC_FIFO_COMPACTION_TABLES=(METRIC_AGGREGATE METRIC_RECORD METRIC_RECORD_MINUTE)
 METRIC_COLLECTOR=ambari-metrics-collector
 
 AMS_COLLECTOR_LOG_DIR=/var/log/ambari-metrics-collector
 
+AMS_HBASE_NORMALIZER_ENABLED=true
+AMS_HBASE_FIFO_COMPACTION_ENABLED=true
+
+NORMALIZER_ENABLED_STUB_FILE=/var/run/ambari-metrics-collector/normalizer_enabled
+FIFO_ENABLED_STUB_FILE=/var/run/ambari-metrics-collector/fifo_enabled
+
 STOP_TIMEOUT=5
 
 DISTRIBUTED_HBASE=false
@@ -95,6 +102,46 @@ function enable_normalization
   if [ $? -ne 0 ]; then
     echo "WARNING: Failed to enable Ambari Metrics data model normalization."
      >&2 echo "WARNING: Failed to enable Ambari Metrics data model normalization."
+  else
+    touch $NORMALIZER_ENABLED_STUB_FILE
+  fi
+}
+
+function enable_fifo_compaction
+{
+  # Enable FIFO compaction for HIGH load tables
+  command=""
+  for table in "${METRIC_FIFO_COMPACTION_TABLES[@]}"
+  do
+    command="$command \n alter_async '$table', CONFIGURATION => {'hbase.hstore.blockingStoreFiles' => '1000',
+    'hbase.hstore.defaultengine.compactionpolicy.class' =>
+    'org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy'}"
+  done
+  echo -e ${command} | ${HBASE_CMD} --config ${HBASE_CONF_DIR} shell > /dev/null 2>&1
+  if [ $? -ne 0 ]; then
+    echo "WARNING: Failed to enable FIFO compaction policy."
+     >&2 echo "WARNING: Failed to enable FIFO compaction policy."
+  else
+    touch $FIFO_ENABLED_STUB_FILE
+  fi
+}
+
+function disable_fifo_compaction
+{
+  # Disable FIFO compaction for HIGH load tables
+  command=""
+  for table in "${METRIC_FIFO_COMPACTION_TABLES[@]}"
+  do
+    command="$command \n alter_async '$table', CONFIGURATION => {'hbase.hstore.defaultengine.compactionpolicy.class' =>
+    'org.apache.hadoop.hbase.regionserver.compactions.ExploringCompactionPolicy',
+    'hbase.hstore.blockingStoreFiles' => '300'}"
+  done
+  echo -e ${command} | ${HBASE_CMD} --config ${HBASE_CONF_DIR} shell > /dev/null 2>&1
+  if [ $? -ne 0 ]; then
+    echo "WARNING: Failed to disable FIFO compaction policy."
+     >&2 echo "WARNING: Failed to disable FIFO compaction policy."
+  else
+    rm -f $FIFO_ENABLED_STUB_FILE
   fi
 }
 
@@ -238,9 +285,41 @@ function start()
     echo "WARNING: Ambari Metrics data model initialization failed."
      >&2 echo "WARNING: Ambari Metrics data model initialization failed."
   else
-    enable_normalization
-  fi
 
+    #
+    # if hbase.normalizer.enabled = true
+    #   if stub file DOES NOT exist
+    #     enable_normalizer + create stub file
+    # else
+    #   if stub file exists
+    #     delete Stub file.
+    if [[ "${AMS_HBASE_NORMALIZER_ENABLED}" == "true" || "${AMS_HBASE_NORMALIZER_ENABLED}" == "True" ]]
+    then
+      if [ ! -f "$NORMALIZER_ENABLED_STUB_FILE" ]
+      then
+        enable_normalization
+      fi
+    else
+      if [ -f "$NORMALIZER_ENABLED_STUB_FILE" ]
+       then
+          rm -f $NORMALIZER_ENABLED_STUB_FILE
+      fi
+    fi
+
+    #Similarly for HBase FIFO Compaction policy
+    if [[ "${AMS_HBASE_FIFO_COMPACTION_ENABLED}" == "true" || "${AMS_HBASE_FIFO_COMPACTION_ENABLED}" == "True" ]]
+    then
+      if [ ! -f "$FIFO_ENABLED_STUB_FILE" ]
+      then
+        enable_fifo_compaction
+      fi
+    else
+      if [ -f "$FIFO_ENABLED_STUB_FILE" ]
+       then
+        disable_fifo_compaction
+      fi
+    fi
+  fi
   }
 
 function stop()
@@ -309,6 +388,8 @@ fi
 # set pid dir path
 if [[ -n "${AMS_COLLECTOR_PID_DIR}" ]]; then
   PIDFILE=${AMS_COLLECTOR_PID_DIR}/ambari-metrics-collector.pid
+  NORMALIZER_ENABLED_STUB_FILE=${AMS_COLLECTOR_PID_DIR}/normalizer_enabled
+  FIFO_ENABLED_STUB_FILE=${AMS_COLLECTOR_PID_DIR}/fifo_enabled
 fi
 
 if [[ -n "${AMS_HBASE_PID_DIR}" ]]; then

ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog213.java

@@ -94,6 +94,9 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
   private static final String AMS_HBASE_SITE = "ams-hbase-site";
   private static final String AMS_HBASE_SITE_ZK_TIMEOUT_PROPERTY =
     "zookeeper.session.timeout.localHBaseCluster";
+  private static final String AMS_HBASE_SITE_NORMALIZER_ENABLED_PROPERTY = "hbase.normalizer.enabled";
+  private static final String AMS_HBASE_SITE_NORMALIZER_PERIOD_PROPERTY = "hbase.normalizer.period";
+  private static final String AMS_HBASE_SITE_NORMALIZER_CLASS_PROPERTY = "hbase.master.normalizer.class";
   private static final String HBASE_ENV_CONFIG = "hbase-env";
   private static final String FLUME_ENV_CONFIG = "flume-env";
   private static final String HIVE_SITE_CONFIG = "hive-site";
@@ -790,7 +793,7 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
       long clusterID = cluster.getClusterId();
 
       final AlertDefinitionEntity journalNodeProcessAlertDefinitionEntity = alertDefinitionDAO.findByName(
-          clusterID, "journalnode_process");
+        clusterID, "journalnode_process");
       final AlertDefinitionEntity hostDiskUsageAlertDefinitionEntity = alertDefinitionDAO.findByName(
           clusterID, "ambari_agent_disk_usage");
 
@@ -1028,8 +1031,8 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
 
           Config amsEnv = cluster.getDesiredConfigByType(AMS_ENV);
           if (amsHbaseEnv != null) {
-            Map<String, String> amsHbaseEnvProperties = amsEnv.getProperties();
-            String content = amsHbaseEnvProperties.get("content");
+            Map<String, String> amsEnvProperties = amsEnv.getProperties();
+            String content = amsEnvProperties.get("content");
             Map<String, String> newProperties = new HashMap<>();
             newProperties.put("content", updateAmsEnvContent(content));
             updateConfigurationPropertiesForCluster(cluster, AMS_ENV, newProperties, true, true);
@@ -1037,10 +1040,15 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
 
           Config amsSite = cluster.getDesiredConfigByType(AMS_SITE);
           if (amsSite != null) {
+            Map<String, String> currentAmsSiteProperties = amsSite.getProperties();
             Map<String, String> newProperties = new HashMap<>();
 
             //Changed AMS result set limit from 5760 to 15840.
-            newProperties.put("timeline.metrics.service.default.result.limit", String.valueOf(15840));
+            if(currentAmsSiteProperties.containsKey("timeline.metrics.service.default.result.limit") &&
+              currentAmsSiteProperties.get("timeline.metrics.service.default.result.limit").equals(String.valueOf(5760))) {
+              LOG.info("Updating timeline.metrics.service.default.result.limit to 15840");
+              newProperties.put("timeline.metrics.service.default.result.limit", String.valueOf(15840));
+            }
 
             //Interval
             newProperties.put("timeline.metrics.cluster.aggregator.second.interval", String.valueOf(120));
@@ -1057,19 +1065,40 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
             //disabled
             newProperties.put("timeline.metrics.cluster.aggregator.second.disabled", String.valueOf(false));
 
+            //Add compaction policy property
+            newProperties.put("hbase.fifo.compaction.policy.enabled", String.valueOf(true));
+
             updateConfigurationPropertiesForCluster(cluster, AMS_SITE, newProperties, true, true);
           }
 
           Config amsHbaseSite = cluster.getDesiredConfigByType(AMS_HBASE_SITE);
           if (amsHbaseSite != null) {
             Map<String, String> amsHbaseSiteProperties = amsHbaseSite.getProperties();
+            Map<String, String> newProperties = new HashMap<>();
+
             String zkTimeout = amsHbaseSiteProperties.get(AMS_HBASE_SITE_ZK_TIMEOUT_PROPERTY);
             // if old default, set new default
             if ("20000".equals(zkTimeout)) {
-              Map<String, String> newProperties = new HashMap<>();
               newProperties.put(AMS_HBASE_SITE_ZK_TIMEOUT_PROPERTY, "120000");
-              updateConfigurationPropertiesForCluster(cluster, AMS_HBASE_SITE, newProperties, true, true);
             }
+
+            //Adding hbase.normalizer.period to upgrade
+            if(!amsHbaseSiteProperties.containsKey(AMS_HBASE_SITE_NORMALIZER_ENABLED_PROPERTY)) {
+              LOG.info("Enabling " + AMS_HBASE_SITE_NORMALIZER_ENABLED_PROPERTY);
+              newProperties.put(AMS_HBASE_SITE_NORMALIZER_ENABLED_PROPERTY, String.valueOf(true));
+            }
+
+            if(!amsHbaseSiteProperties.containsKey(AMS_HBASE_SITE_NORMALIZER_PERIOD_PROPERTY)) {
+              LOG.info("Updating " + AMS_HBASE_SITE_NORMALIZER_PERIOD_PROPERTY);
+              newProperties.put(AMS_HBASE_SITE_NORMALIZER_PERIOD_PROPERTY, String.valueOf(600000));
+            }
+
+            if(!amsHbaseSiteProperties.containsKey(AMS_HBASE_SITE_NORMALIZER_CLASS_PROPERTY)) {
+              LOG.info("Updating " + AMS_HBASE_SITE_NORMALIZER_CLASS_PROPERTY);
+              newProperties.put(AMS_HBASE_SITE_NORMALIZER_CLASS_PROPERTY,
+                "org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer");
+            }
+            updateConfigurationPropertiesForCluster(cluster, AMS_HBASE_SITE, newProperties, true, true);
           }
         }
       }
@@ -1103,6 +1132,19 @@ public class UpgradeCatalog213 extends AbstractUpgradeCatalog {
         "-Xloggc:{{ams_collector_log_dir}}/collector-gc.log-`date +'%Y%m%d%H%M'`\"\n" +
         "export AMS_COLLECTOR_OPTS=\"$AMS_COLLECTOR_OPTS $AMS_COLLECTOR_GC_OPTS\"\n";
     }
+
+    if (!content.contains("HBASE_NORMALIZATION_ENABLED")) {
+      content += "\n" +
+        "# HBase compaction policy enabled\n" +
+        "export HBASE_NORMALIZATION_ENABLED={{ams_hbase_normalizer_enabled}}\n";
+    }
+
+    if (!content.contains("HBASE_FIFO_COMPACTION_POLICY_ENABLED")) {
+      content += "\n" +
+        "# HBase compaction policy enabled\n" +
+        "export HBASE_FIFO_COMPACTION_POLICY_ENABLED={{ams_hbase_fifo_compaction_policy_enabled}}\n";
+    }
+
     return content;
   }
 

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-env.xml

@@ -100,6 +100,12 @@ export AMS_HBASE_PID_DIR={{hbase_pid_dir}}
 # AMS Collector heapsize
 export AMS_COLLECTOR_HEAPSIZE={{metrics_collector_heapsize}}
 
+# HBase compaction policy enabled
+export AMS_HBASE_NORMALIZER_ENABLED={{ams_hbase_normalizer_enabled}}
+
+# HBase compaction policy enabled
+export AMS_HBASE_FIFO_COMPACTION_ENABLED={{ams_hbase_fifo_compaction_enabled}}
+
 # AMS Collector options
 export AMS_COLLECTOR_OPTS="-Djava.library.path=/usr/lib/ams-hbase/lib/hadoop-native"
 {% if security_enabled %}

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-site.xml

@@ -143,7 +143,7 @@
   </property>
   <property>
     <name>hbase.normalizer.period</name>
-    <value>3600000</value>
+    <value>600000</value>
     <description>Period in ms at which the region normalizer runs in the Master.</description>
   </property>
   <property>

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml

@@ -519,5 +519,12 @@
       Default resolution is 30 seconds.
     </description>
   </property>
+  <property>
+    <name>timeline.metrics.hbase.fifo.compaction.enabled</name>
+    <value>true</value>
+    <description>
+      Enable Compaction policy for lower for Precision and Minute aggregate tables.
+    </description>
+  </property>
 
 </configuration>

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/params.py

@@ -74,6 +74,9 @@ ams_monitor_script = "/usr/sbin/ambari-metrics-monitor"
 
 ams_hbase_home_dir = "/usr/lib/ams-hbase/"
 
+ams_hbase_normalizer_enabled = default("/configurations/ams-hbase-site/hbase.normalizer.enabled", None)
+ams_hbase_fifo_compaction_enabled = default("/configurations/ams-site/timeline.metrics.hbase.fifo.compaction.enabled", None)
+
 #hadoop params
 
 hbase_excluded_hosts = config['commandParams']['excluded_hosts']

ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog213Test.java

@@ -647,6 +647,7 @@ public class UpgradeCatalog213Test {
         put("timeline.metrics.cluster.aggregator.minute.ttl", String.valueOf(7776000));
         put("timeline.metrics.cluster.aggregator.second.checkpointCutOffMultiplier", String.valueOf(2));
         put("timeline.metrics.cluster.aggregator.second.disabled", String.valueOf(false));
+        put("hbase.fifo.compaction.policy.enabled", String.valueOf(true));
       }
     };
     EasyMockSupport easyMockSupport = new EasyMockSupport();
@@ -659,7 +660,7 @@ public class UpgradeCatalog213Test {
       put("normal", cluster);
     }}).once();
     expect(cluster.getDesiredConfigByType("ams-site")).andReturn(mockAmsSite).atLeastOnce();
-    expect(mockAmsSite.getProperties()).andReturn(oldPropertiesAmsSite).times(1);
+    expect(mockAmsSite.getProperties()).andReturn(oldPropertiesAmsSite).times(2);
 
     Injector injector = easyMockSupport.createNiceMock(Injector.class);
     expect(injector.getInstance(Gson.class)).andReturn(null).anyTimes();
@@ -704,6 +705,10 @@ public class UpgradeCatalog213Test {
     Map<String, String> newPropertiesAmsSite = new HashMap<String, String>() {
       {
         put("zookeeper.session.timeout.localHBaseCluster", String.valueOf(120000));
+        put("hbase.normalizer.enabled", String.valueOf(true));
+        put("hbase.normalizer.period", String.valueOf(600000));
+        put("hbase.master.normalizer.class", "org.apache.hadoop.hbase.master.normalizer.SimpleRegionNormalizer");
+
       }
     };
     EasyMockSupport easyMockSupport = new EasyMockSupport();
@@ -823,7 +828,14 @@ public class UpgradeCatalog213Test {
       "-XX:+UseCMSInitiatingOccupancyOnly -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCDateStamps " +
       "-XX:+UseGCLogFileRotation -XX:GCLogFileSize=10M " +
       "-Xloggc:{{ams_collector_log_dir}}/collector-gc.log-`date +'%Y%m%d%H%M'`\"\n" +
-      "export AMS_COLLECTOR_OPTS=\"$AMS_COLLECTOR_OPTS $AMS_COLLECTOR_GC_OPTS\"\n";
+      "export AMS_COLLECTOR_OPTS=\"$AMS_COLLECTOR_OPTS $AMS_COLLECTOR_GC_OPTS\"\n"+
+      "\n" +
+      "# HBase compaction policy enabled\n" +
+      "export HBASE_NORMALIZATION_ENABLED={{ams_hbase_normalizer_enabled}}\n" +
+      "\n" +
+      "# HBase compaction policy enabled\n" +
+      "export HBASE_FIFO_COMPACTION_POLICY_ENABLED={{ams_hbase_fifo_compaction_policy_enabled}}\n";
+
     String result = (String) updateAmsEnvContent.invoke(upgradeCatalog213, oldContent);
     Assert.assertEquals(expectedContent, result);
   }