AMBARI-9158. Add support for AMS operation Modes. (mpapirkovskyy)

ambari-metrics/ambari-metrics-storm-sink/src/main/assemblies/jar-with-common.xml

@@ -28,6 +28,7 @@
       <includes>
         <include>org.apache.ambari:ambari-metrics-common</include>
         <include>org.apache.ambari:ambari-metrics-storm-sink</include>
+        <include>org.codehaus.jackson:jackson-mapper-asl</include> <!--missing in storm classpath-->
       </includes>
     </dependencySet>
   </dependencySets>

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

@@ -35,6 +35,8 @@ AMS_LOG_DIR=/var/log/ambari-metrics-collector
 
 STOP_TIMEOUT=5
 
+DISTRIBUTED_HBASE=false
+
 function hbase_daemon
 {
     local daemon=$1
@@ -156,6 +158,10 @@ while [[ -z "${_ams_configs_done}" ]]; do
         exit 1
       fi
     ;;
+    --distributed)
+      DISTRIBUTED_HBASE=true
+      shift
+    ;;
     *)
       _ams_configs_done=true
     ;;
@@ -186,12 +192,18 @@ case "$1" in
 	start)
 		hadoop_java_setup
 
-		#hbase_daemon "zookeeper" "start"
 
-		hbase_daemon "master" "start"
-		#hbase_daemon "regionserver" "start"
+#     hbase_daemon "zookeeper" "start"
+#		  hbase_daemon "master" "start"
+#		  hbase_daemon "regionserver" "start"
+    if [ !"${DISTRIBUTED_HBASE}" ]; then
+      hbase_daemon "master" "start"
+    else
+      echo "Launching in distributed mode. Assuming Hbase daemons up and running."
+    fi
 
-    sleep 30
+#    sleep 30
+#TODO check issue fixed
 
 		CLASS='org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryServer'
 		# YARN_OPTS="${YARN_OPTS} ${YARN_TIMELINESERVER_OPTS}"
@@ -257,9 +269,9 @@ case "$1" in
       fi
 
       #stop hbase daemons
-      #hbase_daemon "zookeeper" "stop"
-      hbase_daemon "master" "stop"
-      #hbase_daemon "regionserver" "stop"
+      if [ ! "${DISTRIBUTED_HBASE}" ]; then
+        hbase_daemon "master" "stop"
+      fi
 
 
     ;;

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

@@ -21,7 +21,7 @@
 <configuration>
   <property>
     <name>ams_user</name>
-    <value>root</value>
+    <value>ams</value>
     <property-type>USER</property-type>
     <description>AMS User Name.</description>
   </property>

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

@@ -53,7 +53,7 @@
   </property>
   <property>
     <name>hbase_user</name>
-    <value>hbase</value>
+    <value>ams</value>
     <property-type>USER</property-type>
     <description>HBase User Name.</description>
   </property>

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

@@ -28,7 +28,6 @@
       Service Operation modes:
       1) embedded: Metrics stored on local FS, HBase in Standalone mode
       2) distributed: HBase daemons writing to HDFS
-      3) external: External HBase storage backend
     </description>
   </property>
   <property>
@@ -232,4 +231,4 @@
       JDBC resultset prefect size for aggregator queries.
     </description>
   </property>
-</configuration>
+</configuration>

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/ams.py

@@ -180,6 +180,11 @@ def ams(name=None):
               recursive=True
     )
 
+    Directory(format("{ams_monitor_dir}/psutil/build"),
+              owner=params.ams_user,
+              group=params.user_group,
+              recursive=True)
+
     TemplateConfig(
       format("{ams_monitor_conf_dir}/metric_monitor.ini"),
       owner=params.ams_user,

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/ams_service.py

@@ -21,6 +21,7 @@ limitations under the License.
 from resource_management import *
 from ambari_commons import OSConst
 from ambari_commons.os_family_impl import OsFamilyFuncImpl, OsFamilyImpl
+from hbase_service import hbase_service
 
 @OsFamilyFuncImpl(os_family=OSConst.WINSRV_FAMILY)
 def ams_service(name, action):
@@ -39,6 +40,12 @@ def ams_service(name, action):
     pid_file = format("{ams_collector_pid_dir}/ambari-metrics-collector.pid")
     no_op_test = format("ls {pid_file} >/dev/null 2>&1 && ps `cat {pid_file}` >/dev/null 2>&1")
 
+    if params.is_hbase_distributed:
+      hbase_service('zookeeper', action=action)
+      hbase_service('master', action=action)
+      hbase_service('regionserver', action=action)
+      cmd = format("{cmd} --distributed")
+
     if action == 'start':
       daemon_cmd = format("{cmd} start")
       Execute(daemon_cmd,

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/hbase.py

@@ -158,6 +158,29 @@ def hbase(name=None # 'master' or 'regionserver' or 'client'
   if params.security_enabled:
     hbase_TemplateConfig( format("hbase_{name}_jaas.conf"), user=params.hbase_user)
   
+  if name in ["master","regionserver"]:
+
+    if params.is_hbase_distributed:
+
+      params.HdfsDirectory(params.hbase_root_dir,
+                           action="create_delayed",
+                           owner=params.hbase_user,
+                           mode=0775
+      )
+      params.HdfsDirectory(None, action="create")
+
+    else:
+
+      local_root_dir = params.hbase_root_dir
+      #cut protocol name
+      if local_root_dir.startswith("file://"):
+        local_root_dir = local_root_dir[7:]
+        #otherwise assume dir name is provided as is
+
+      Directory(local_root_dir,
+                owner = params.hbase_user,
+                recursive = True)
+
   if name != "client":
     Directory( params.hbase_pid_dir,
       owner = params.hbase_user,
@@ -169,14 +192,14 @@ def hbase(name=None # 'master' or 'regionserver' or 'client'
       recursive = True
     )
 
-  if (params.hbase_log4j_props != None):
+  if params.hbase_log4j_props is not None:
     File(format("{params.hbase_conf_dir}/log4j.properties"),
          mode=0644,
          group=params.user_group,
          owner=params.hbase_user,
          content=params.hbase_log4j_props
     )
-  elif (os.path.exists(format("{params.hbase_conf_dir}/log4j.properties"))):
+  elif os.path.exists(format("{params.hbase_conf_dir}/log4j.properties")):
     File(format("{params.hbase_conf_dir}/log4j.properties"),
       mode=0644,
       group=params.user_group,

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/hbase_service.py

@@ -28,7 +28,7 @@ def hbase_service(
   
     role = name
     cmd = format("{daemon_script} --config {hbase_conf_dir}")
-    pid_file = format("{pid_dir}/hbase-{hbase_user}-{role}.pid")
+    pid_file = format("{hbase_pid_dir}/hbase-{hbase_user}-{role}.pid")
     no_op_test = format("ls {pid_file} >/dev/null 2>&1 && ps `cat {pid_file}` >/dev/null 2>&1")
     
     if action == 'start':

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/metric_collector.py

@@ -49,9 +49,7 @@ class AmsCollector(Script):
         import params
         env.set_params(params)
 
-        ams_service( 'collector',
-                       action = 'stop'
-        )
+        ams_service( 'collector', action = 'stop')
 
     def status(self, env):
         import status_params

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

@@ -61,8 +61,9 @@ hbase_included_hosts = config['commandParams']['included_hosts']
 
 hbase_user = status_params.hbase_user
 smokeuser = config['configurations']['cluster-env']['smokeuser']
-hbase_hdfs_root_dir = config['configurations']['ams-hbase-site']['hbase.rootdir']
-is_hbase_distributed = hbase_hdfs_root_dir.startswith('hdfs://')
+hbase_root_dir = config['configurations']['ams-hbase-site']['hbase.rootdir']
+
+is_hbase_distributed = hbase_root_dir.startswith('hdfs://')
 
 # security is disabled for embedded mode, when HBase is backed by file
 security_enabled = False if not is_hbase_distributed else config['configurations']['cluster-env']['security_enabled'] 
@@ -140,18 +141,19 @@ hdfs_user_keytab = config['configurations']['hadoop-env']['hdfs_user_keytab']
 hdfs_user = config['configurations']['hadoop-env']['hdfs_user']
 hdfs_principal_name = config['configurations']['hadoop-env']['hdfs_principal_name']
 kinit_path_local = functions.get_kinit_path(["/usr/bin", "/usr/kerberos/bin", "/usr/sbin"])
+
 import functools
-#create partial functions with common arguments for every HdfsDirectory call
-#to create hdfs directory we need to call params.HdfsDirectory in code
-# HdfsDirectory = functools.partial(
-#   HdfsDirectory,
-#   conf_dir=hadoop_conf_dir,
-#   hdfs_user=hdfs_user,
-#   security_enabled = security_enabled,
-#   keytab = hdfs_user_keytab,
-#   kinit_path_local = kinit_path_local,
-#   bin_dir = hadoop_bin_dir
-# )
+# create partial functions with common arguments for every HdfsDirectory call
+# to create hdfs directory we need to call params.HdfsDirectory in code
+HdfsDirectory = functools.partial(
+  HdfsDirectory,
+  conf_dir=hadoop_conf_dir,
+  hdfs_user=hdfs_user,
+  security_enabled = security_enabled,
+  keytab = hdfs_user_keytab,
+  kinit_path_local = kinit_path_local,
+  bin_dir = hadoop_bin_dir
+)
 
 
 

ambari-server/src/main/resources/common-services/AMS/0.1.0/package/scripts/params_linux.py

@@ -34,20 +34,12 @@ if rpm_version is not None:
   #RPM versioning support
   rpm_version = default("/configurations/hadoop-env/rpm_version", None)
 
-if rpm_version is not None:
-  hadoop_native_lib = format("/usr/hdp/current/hadoop-client/lib/native/")
-  hadoop_bin_dir = format("/usr/hdp/current/hadoop/bin")
-  daemon_script = format('/usr/hdp/current/hbase/bin/hbase-daemon.sh')
-  region_mover = format('/usr/hdp/current/hbase/bin/region_mover.rb')
-  region_drainer = format('/usr/hdp/current/hbase/bin/draining_servers.rb')
-  hbase_cmd = format('/usr/hdp/current/hbase/bin/hbase')
-else:
-  hadoop_native_lib = format("/usr/lib/hadoop/lib/native")
-  hadoop_bin_dir = "/usr/bin"
-  daemon_script = "/usr/lib/hbase/bin/hbase-daemon.sh"
-  region_mover = "/usr/lib/hbase/bin/region_mover.rb"
-  region_drainer = "/usr/lib/hbase/bin/draining_servers.rb"
-  hbase_cmd = "/usr/lib/hbase/bin/hbase"
+hadoop_native_lib = format("/usr/lib/hadoop/lib/native")
+hadoop_bin_dir = "/usr/bin"
+daemon_script = "/usr/lib/ams-hbase/bin/hbase-daemon.sh"
+region_mover = "/usr/lib/ams-hbase/bin/region_mover.rb"
+region_drainer = "/usr/lib/ams-hbase/bin/draining_servers.rb"
+hbase_cmd = "/usr/lib/ams-hbase/bin/hbase"
 
 hadoop_conf_dir = "/etc/hadoop/conf"
 hbase_conf_dir = "/etc/ams-hbase/conf"

ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py

@@ -254,8 +254,10 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
     if not propertyName in properties:
       return self.getErrorItem("Value should be set")
     dir = properties[propertyName]
-    if not dir.startswith("hdfs://"):
-      dir = re.sub("^file://", "", dir, count=1)
+    if dir.startswith("hdfs://"):
+      return None #TODO following code fails for hdfs://, is this valid check for hdfs?
+
+    dir = re.sub("^file://", "", dir, count=1)
     mountPoints = {}
     for mountPoint in hostInfo["disk_info"]:
       mountPoints[mountPoint["mountpoint"]] = to_number(mountPoint["available"])

ambari-server/src/main/resources/stacks/HDP/2.2/role_command_order.json

@@ -4,6 +4,7 @@
   "general_deps" : {
     "_comment" : "dependencies for all cases",
     "FALCON_SERVER-START": ["NAMENODE-START", "DATANODE-START", "OOZIE_SERVER-START"],
+    "METRIC_COLLECTOR-START": ["NAMENODE-START", "DATANODE-START"],
     "WEBHCAT_SERVICE_CHECK-SERVICE_CHECK": ["WEBHCAT_SERVER-START"],
     "FLUME_SERVICE_CHECK-SERVICE_CHECK": ["FLUME_HANDLER-START"],
     "FALCON_SERVICE_CHECK-SERVICE_CHECK": ["FALCON_SERVER-START"],

ambari-server/src/main/resources/stacks/HDP/2.2/services/stack_advisor.py

@@ -101,7 +101,8 @@ class HDP22StackAdvisor(HDP21StackAdvisor):
                "hadoop-env": self.validateHDFSConfigurationsEnv},
       "MAPREDUCE2": {"mapred-site": self.validateMapReduce2Configurations},
       "AMS": {"ams-hbase-site": self.validateAmsHbaseSiteConfigurations,
-              "ams-hbase-env": self.validateAmsHbaseEnvConfigurations},
+              "ams-hbase-env": self.validateAmsHbaseEnvConfigurations,
+              "ams-site": self.validateAmsSiteConfigurations},
       "TEZ": {"tez-site": self.validateTezConfigurations}
     }
     parentValidators.update(childValidators)
@@ -114,9 +115,23 @@ class HDP22StackAdvisor(HDP21StackAdvisor):
                         {"config-name": 'tez.runtime.unordered.output.buffer.size-mb', "item": self.validatorLessThenDefaultValue(properties, recommendedDefaults, 'tez.runtime.unordered.output.buffer.size-mb')},]
     return self.toConfigurationValidationProblems(validationItems, "tez-site")
 
+  def validateAmsSiteConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):
+    validationItems = []
+
+    op_mode = properties.get("timeline.metrics.service.operation.mode")
+    correct_op_mode_item = None
+    if op_mode not in ("embedded", "distributed"):
+      correct_op_mode_item = self.getErrorItem("Correct value should be set.")
+      pass
+
+    validationItems.extend([{"config-name":'timeline.metrics.service.operation.mode', "item": correct_op_mode_item }])
+    return self.toConfigurationValidationProblems(validationItems, "ams-site")
+
   def validateAmsHbaseSiteConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):
 
     amsCollectorHosts = self.getComponentHostNames(services, "AMS", "METRIC_COLLECTOR")
+    ams_site = getSiteProperties(configurations, "ams-site")
+
     recommendedDiskSpace = 10485760
     # TODO validate configuration for multiple AMS collectors
     if len(amsCollectorHosts) > 1:
@@ -138,6 +153,21 @@ class HDP22StackAdvisor(HDP21StackAdvisor):
           validationItems.extend([ {"config-name": 'hbase.rootdir', "item": self.validatorEnoughDiskSpace(properties, 'hbase.rootdir', host["Hosts"], recommendedDiskSpace)}])
           break
 
+    rootdir_item = None
+    op_mode = ams_site.get("timeline.metrics.service.operation.mode")
+    hbase_rootdir = properties.get("hbase.rootdir")
+    if op_mode == "distributed" and not hbase_rootdir.startswith("hdfs://"):
+      rootdir_item = self.getWarnItem("In distributed mode hbase.rootdir should point to HDFS. Collector will operate in embedded mode otherwise.")
+      pass
+
+    distributed_item = None
+    distributed = properties.get("hbase.cluster.distributed")
+    if hbase_rootdir.startswith("hdfs://") and not distributed.lower() == "true":
+      distributed_item = self.getErrorItem("Distributed property should be set to true if hbase.rootdir points to HDFS.")
+
+    validationItems.extend([{"config-name":'hbase.rootdir', "item": rootdir_item },
+                            {"config-name":'hbase.cluster.distributed', "item": distributed_item }])
+
     return self.toConfigurationValidationProblems(validationItems, "ams-hbase-site")
 
   def validateAmsHbaseEnvConfigurations(self, properties, recommendedDefaults, configurations, services, hosts):