From fb252374f56d2f394a3074eb9089d594dd66f2af Mon Sep 17 00:00:00 2001
From: Sujoy <sujoymajumder474@gmail.com>
Date: Tue, 20 Nov 2018 04:41:30 -0500
Subject: [PATCH 1/2] Added plugings for hadoop log files.

---
 .../mapping/logging_plugins_mapping.yaml      | 273 ++++++++++++++++++
 1 file changed, 273 insertions(+)

diff --git a/config_handler/mapping/logging_plugins_mapping.yaml b/config_handler/mapping/logging_plugins_mapping.yaml
index d2e137c..0dade42 100644
--- a/config_handler/mapping/logging_plugins_mapping.yaml
+++ b/config_handler/mapping/logging_plugins_mapping.yaml
@@ -323,3 +323,276 @@ esalogstore:
     _tag_Name: "${tag_suffix[5].rpartition('_').first}"
   match:
     flush_interval: 30s
+
+# yarn resource manager logging mapping
+yarn-rm:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop-yarn/yarn/yarn-yarn-resourcemanager-manager-0.log'
+    pos_file: '/var/log/td-agent/yarn-rm.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'YarnRmLogs'
+    _documentType: 'YarnRmLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# yarn resource manager audit logging mapping
+yarn-audit:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop-yarn/yarn/rm-audit.log'
+    pos_file: '/var/log/td-agent/rm-audit.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'YarnRmAuditLogs'
+    _documentType: 'YarnRmAuditLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+#Yarn timeline server  logging mapping
+yarn-timeline:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop-yarn/yarn/yarn-yarn-timelineserver-manager-0.log'
+    pos_file: '/var/log/td-agent/yarn-yarn-timelineserver-manager.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'YarmTimelineLogs'
+    _documentType: 'YarmTimelineLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# hdfs audit logging mapping
+hdfs-audit:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop/hdfs/hdfs-audit.log'
+    pos_file: '/var/log/td-agent/hdfs-audit.pos'
+  parse:
+    '@type': multi_format
+    expressions:
+     - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*)\s+(?<message>.*)$/'
+     - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*)\s+(?<message>.*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'HdfsAuditLogs'
+    _documentType: 'HdfsAuditLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# hdfs namenode logging mapping
+hdfs-namenode:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop/hdfs/hadoop-hdfs-namenode-manager-0.log'
+    pos_file: '/var/log/td-agent/hadoop-hdfs-namenode-manager-0.log.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'HdfsNameNodeLogs'
+    _documentType: 'HdfsNameNodeLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+
+# oozie audit logging mapping
+oozie-audit:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie-audit.log'
+    pos_file: '/var/log/td-agent/oozie-audit.log.pos'
+  parse:
+    '@type': regexp
+    expression: '^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*)\s+(?<message>.*)'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieAuditLogs'
+    _documentType: 'OozieAuditLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# oozie Error logging mapping
+oozie-error-logs:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie-error.log'
+    pos_file: '/var/log/td-agent/oozie-error.pos'
+  parse:
+    '@type': regexp
+    expression: '^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*)\s+(?<message>.*)'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieErrorLogs'
+    _documentType: 'OozieErrorLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# oozie logging mapping
+oozie-logs:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie.log'
+    pos_file: '/var/log/td-agent/oozie.pos'
+  parse:
+    '@type': regexp
+    expression: '^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*)\s+(?<message>.*)'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieLogs'
+    _documentType: 'OozieLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+# yarn resource manager logging mapping
+hdfs-journalnode-manager:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop/hdfs/hadoop-hdfs-journalnode-manager-0.log'
+    pos_file: '/var/log/td-agent/hadoop-hdfs-journalnode-manager-0.log'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'hdfsjournalnodemanager'
+    _documentType: 'hdfsjournalnodemanager'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+#hdfs zoopkeeper manager logs
+hdfs-zkfc-manager:
+  source:
+    '@type': tail
+    path: '/var/log/hadoop/hdfs/hadoop-hdfs-zkfc-manager-0.log'
+    pos_file: '/var/log/td-agent/hadoop-hdfs-zkfc-manager-0.log'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'hdfszkfcmanager'
+    _documentType: 'hdfszkfcmanager'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+#Oozie jpa logs
+oozie-jpa:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie-jpa.log'
+    pos_file: '/var/log/td-agent/oozie-jpa.log.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieJpaLogs'
+    _documentType: 'OozieJpaLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+#Oozie instrumentation logs
+oozie-instrumentation:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie-instrumentation.log'
+    pos_file: '/var/log/td-agent/oozie-instrumentation.log.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieInstrumentationLogs'
+    _documentType: 'OozieInstrumentationLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+
+#Oozie ops logs
+oozie-ops:
+  source:
+    '@type': tail
+    path: '/var/log/oozie/oozie-ops.log'
+    pos_file: '/var/log/td-agent/oozie-ops.log.pos'
+  multiline:
+    format_firstline: '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*/'
+    expression:
+      - '/^(?<time>\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\,[^ ]*[ ]*(?<level>[^ ]*) (?<message>[\s\S]*)/'
+  transform:
+    node: '#{Socket.gethostname}'
+    file: '${tag_suffix[1]}'
+    _plugin: 'OozieOpsLogs'
+    _documentType: 'OozieOpsLogs'
+    time: ${require 'time'; time.to_time.to_i}
+    level: ${n_level='warning'; if record['level'].strip().downcase == 'warn' then n_level else record['level'].strip().downcase end}
+    message: ${record['message']}
+  match:
+    flush_interval: 30s
+

From 0b6a06c007bdb500f6ccddae738c2d9dcdfdab49 Mon Sep 17 00:00:00 2001
From: Sujoy <sujoymajumder474@gmail.com>
Date: Tue, 20 Nov 2018 05:38:01 -0500
Subject: [PATCH 2/2] Added services for hadoop logging

---
 service_discovery/discovery.py | 521 ++++++++++++++++++---------------
 1 file changed, 282 insertions(+), 239 deletions(-)

diff --git a/service_discovery/discovery.py b/service_discovery/discovery.py
index a108ad8..7a71220 100644
--- a/service_discovery/discovery.py
+++ b/service_discovery/discovery.py
@@ -1,13 +1,16 @@
+'''
+discover services
+'''
 import os
 import subprocess
+import re
 import psutil
-import socket
-import kafka
 from config_handler import configurator
 from common.util import *
 
 logger = expoter_logging(COLLECTD_MGR)
-service_name = {
+JCMD_PID_DICT = dict()
+SERVICE_NAME = {
     "elasticsearch": "ES",
     "apache": "apache",
     "mysql": "mysql",
@@ -17,11 +20,12 @@
     "tpcc": "tpcc",
     "kafka.Kafka": "kafka",
     "zookeeper": "zookeeper",
-    "hxconnect" : "hxconnect",
+    "hxconnect": "hxconnect",
     "cassandra": "cassandra",
-    "esalogstore": "ESAlogstore"
+    "esalogstore": "ESAlogstore",
+    "knox": "knox",
 }
-services = [
+SERVICES = [
     "elasticsearch",
     "apache",
     "mysql",
@@ -33,12 +37,13 @@
     "zookeeper",
     "hxconnect",
     "cassandra",
-    "esalogstore"
+    "esalogstore",
+    "knox",
 ]
 '''
 Mapping for services and the plugin to be configured for them.
 '''
-service_plugin_mapping = {
+SERVICE_PLUGIN_MAPPING = {
     "elasticsearch": "elasticsearch",
     "apache": "apache",
     "mysql": "mysql",
@@ -48,42 +53,112 @@
     "tpcc": "tpcc",
     "kafka.Kafka": "kafkatopic",
     "zookeeper": "zookeeperjmx",
-    "hxconnect" : "hxconnect",
-    "cassandra" : "cassandra",
+    "hxconnect": "hxconnect",
+    "cassandra": "cassandra",
+    "knox": "oozie"
+}
+
+POLLER_PLUGIN = ["elasticsearch"]
+HADOOP_SERVICE = {
+    "yarn-rm-log": { 
+       "service-name": "org.apache.hadoop.yarn.server.resourcemanager.ResourceManager",
+       "service-list": ["yarn-rm","yarn-audit"]
+     },
+    "yarn-timeline-server": {
+       "service-name": "org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryServer",
+       "service-list": ["yarn-timeline","yarn-audit"]
+     },
+    "hdfs-namenode": {
+       "service-name": "org.apache.hadoop.hdfs.server.namenode.NameNode",
+       "service-list": ["hdfs-namenode","hdfs-audit","hdfs-gc"]
+     },
+    "hdfs-journalnode": {
+       "service-name": "org.apache.hadoop.hdfs.qjournal.server.JournalNode",
+       "service-list": ["hdfs-journalnode","hdfs-audit","hdfs-gc","hdfs-journalnode-manager"]
+     },
+     "oozie-server": {
+       "service-name": "org.apache.catalina.startup.Bootstrap",
+       "service-list": ["oozie-server","oozie-audit", "oozie-error-logs","oozie-logs"],
+       "service-cmd-line": "oozie-server"
+     }
 }
 
-poller_plugin = [
-    "elasticsearch"
-]
 
-def add_pid_usage(pid, service, pid_list):
+def add_pid_usage(pid, pid_list):
     """Add usage stats of each pids"""
-    cmd = "ps auxww | grep %s | grep -v grep" % pid
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
-    (out, err) = p.communicate()
-    for line in out.splitlines():
-        if service in line:
-            pid = {}
-            pid["user"] = line.split()[0]
-            pid["process_id"] = int(line.split()[1])
-            pid["cpuUsage"] = line.split()[2]
-            pid["memUsage"] = line.split()[3]
-            pid["status"] = "running"
-            pid_list.append(pid)
+
+    pid_detail = psutil.Process(pid)
+    if not pid_detail.is_running():
+        return
+    pid_info = {}
+    pid_info["user"] = pid_detail.username()
+    pid_info["process_id"] = pid
+    pid_info["cpuUsage"] = pid_detail.cpu_percent()
+    pid_info["memUsage"] = pid_detail.memory_percent()
+    pid_info["status"] = "running"
+    pid_list.append(pid_info)
 
 
 def check_jmx_enabled(pid):
     """Check if jmx enabled for java process"""
-    ps_cmd = "ps -eaf | grep %s | grep -v grep" % pid
-    p = subprocess.Popen(ps_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
-    (res, err) = p.communicate()
-    if res is not "":
-        res_list = res.split()
-        for element in res_list:
-            if element == "-Dcom.sun.management.jmxremote":
-                return True
+    pid_detail = psutil.Process(pid)
+    if re.search("Dcom.sun.management.jmxremote", str(pid_detail.cmdline())):
+        return True
     return False
+def exec_subprocess(cmd):
+    """ execute subprocess cmd """
+    cmd_output = subprocess.Popen(
+        cmd,
+        shell=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE)
+    res, err = cmd_output.communicate()
+    return res
+
+def parser_jcmd(service,cmd_line=""):
+    """ Parser for jcmd """
+    pid_list = list()
+    if not JCMD_PID_DICT:
+         java_avail = subprocess.check_call(
+             ["java", "-version"],
+             stdout=subprocess.PIPE,
+             stderr=subprocess.PIPE)
+         if java_avail:
+             return JCMD_PID_DICT
+
+         res = exec_subprocess("sudo /usr/lib/java/jdk1.8.0_131/bin/jcmd | awk '{print $1 \" \" $2}'")
+         if not res:
+             return pid_list
+
+         for line in res.splitlines():
+             if not line:
+                continue
+             out_list = line.split()
+             if not cmd_line:
+                JCMD_PID_DICT[out_list[1]] = int(out_list[0])
+                continue
+             pid_detail = psutil.Process(int(out_list[0]))
+             if re.search(cmd_line,str(pid_detail.cmdline())):
+                JCMD_PID_DICT[out_list[1]] = int(out_list[0])
+                print("ItsWorking")
+         print(JCMD_PID_DICT)
+    
+    for service_name, pid in JCMD_PID_DICT.items():
+        if re.search(service,service_name):
+            pid_list.append(pid)
+    print("{} pid list {}".format (service,pid_list))
+    return pid_list
 
+def get_hadoop_running_service_list():
+    '''
+     get hadoop services running in the client machine
+    '''
+    hadoop_running_service_list = list() 
+    for name, service_name in HADOOP_SERVICE.items():
+        if parser_jcmd(service_name['service-name']):
+           hadoop_running_service_list.append(name)
+    return hadoop_running_service_list
+            
 def get_process_id(service):
     '''
     :param service: name of the service
@@ -94,198 +169,183 @@ def get_process_id(service):
     pids = []
 
     if service in ["kafka.Kafka", "zookeeper"]:
-        try:
-                # Common logic for jmx related process
-            processID = []
-            java_avail = subprocess.check_call(["java", "-version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            if not java_avail:
-                jcmd = subprocess.Popen("jcmd | awk '{print $1 \" \" $2}' | grep -w \"%s\"" % service, shell=True,
-                                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-                res, err = jcmd.communicate()
-                if res is not "":
-                    j_pids = res.splitlines()
-                    for j_pid in j_pids:
-                        if j_pid is not "":
-                            pidval = j_pid.split()
-                            if check_jmx_enabled(pidval[0]):
-                                processID.append(pidval[0])
-    
-            for procid in processID:
-                add_pid_usage(procid, service, pids)
-            logger.info("PIDs %s", pids)
-            return pids
-        except:
-            logger.info("PIDs %s", pids)
-            return pids
+        pid_list = parser_jcmd(service)
+
+        for pid in pid_list:
+            if check_jmx_enabled(pid):
+                add_pid_usage(pid, pids)
+
+        logger.info("PIDs %s", pids)
+        return pids
+
+    if service == "apache":
+        service = "httpd"
+        out = exec_subprocess("lsb_release -d")
+        for line in out.splitlines():
+            if "Ubuntu" in line:
+                service = "apache2"
+                break
 
     try:
-        if (service == "apache"):
-            os_cmd = "lsb_release -d"
-            p = subprocess.Popen(os_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
-            (out, err) = p.communicate()
-            for line in out.splitlines():
-                if ("Ubuntu" in line):
-                    service = "apache2"
-                    break
-            # The linux flavour is not Ubuntu could be CentOS or redHat so search for httpd
-            if (service == "apache"):
-                service = "httpd"
-    
-        processID = ""
+        process_id = ""
         for proc in psutil.process_iter(attrs=['pid', 'name', 'username']):
             # Java processes
-            if service in ["elasticsearch", "cassandra"]:
-                if proc.info.get("name") == "java" and proc.info.get("username") == service:
-                    processID = proc.info.get("pid")
+            if service in ["elasticsearch", "cassandra", "knox"]:
+                if proc.info.get("name") == "java" and proc.info.get(
+                        "username") == service:
+                    process_id = proc.info.get("pid")
                     break
             # Postgres process
             elif service in ["postgres"]:
-                if proc.info.get("name") == "postmaster" or proc.info.get("name") == "postgres":
-                    processID = proc.info.get("pid")
+                if proc.info.get("name") == "postmaster" or proc.info.get(
+                        "name") == "postgres":
+                    process_id = proc.info.get("pid")
                     break
             # Non java processes
             elif service in str(proc.info.get("name")):
-                processID = proc.info.get("pid")
+                process_id = proc.info.get("pid")
                 break
-    
-        add_pid_usage(processID, service, pids)
+
+        add_pid_usage(process_id, pids)
         logger.info("PIDs %s", pids)
         return pids
-    except:
+    except BaseException:
         logger.info("PIDs %s", pids)
         return pids
 
 
-def add_status(dict):
+def add_status(proc_dict):
     '''
     Find the status of the PID running, sleeping.
     :param dict: dictionary return by get_process_id
     :return: add status for the PID in th dictionary.
     '''
     # Add state, threads assosciated with the service PID
-    fileobj = open('/proc/%d/status' % (dict["PID"]))
-    if fileobj is None:
-        return
+    fileobj = open('/proc/%d/status' % (proc_dict["PID"]))
+    if not fileobj:
+        return None
     lines = fileobj.readlines()
     for line in lines:
         if line.startswith("State:"):
             state = (line.split())[2]
             state = state.strip("()")
-            dict["state"] = state
+            proc_dict["state"] = state
         elif line.startswith("Threads:"):
             threads = line.split()
             threads = threads[1]
-            dict["threads"] = threads
-    return dict
+            proc_dict["threads"] = threads
+    return proc_dict
 
 
-def add_ports(dict, service):
+def add_ports(service_dict, service):
     '''
     Add listening ports for the PID
     :param dict: dictionary returned by add_status
     :param service: name of the service
     :return: add listening ports for the PID to the dictionary
     '''
-    logger.debug("Add ports %s %s", dict, service)
+    logger.debug("Add ports %s %s", service_dict, service)
     ports = []
-    if(service == "apache"):
-        apache_service = ""
-        os_cmd = "lsb_release -d"
-        p = subprocess.Popen(os_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
-        (out, err) = p.communicate()
+    if service == "apache":
+        cmd = "netstat -anp | grep httpd"
+
+        out = exec_subprocess("lsb_release -d")
         for line in out.splitlines():
-            if("Ubuntu" in line):
-                apache_service = "apache2"
+            if "Ubuntu" in line:
+                cmd = "netstat -anp | grep apache2"
                 break
-        if(apache_service == ""):
-            apache_service = "httpd"
-        cmd = "netstat -anp | grep %s" %(apache_service)
     else:
-        cmd = "netstat -anp | grep %s" %(dict["PID"])
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
-    (out, err) = p.communicate()
+        cmd = "netstat -anp | grep %s" % (service_dict["PID"])
+
+    out = exec_subprocess(cmd)
     for line in out.splitlines():
         line = line.split()
-        if((line[5] == 'LISTEN') and (service == "apache" or str(dict['PID']) in line[6])):
+        if (line[5] == 'LISTEN') and (service == "apache" or str(service_dict['PID']) in line[6]):
             port = (line[3].split(':')[-1])
             if port not in ports:
                 ports.append(port)
-    dict['ports'] = ports
-    return dict
-
-
-def add_logger_config(dict, service):
-    dict["loggerConfig"] = []
-    fluentdPlugins = configurator.get_fluentd_plugins_mapping().keys()
-    for item in fluentdPlugins:
-        if(item.startswith(service.split(".")[0])):
-            logConfig = {}
-            logConfig["name"] = item
-            logConfig["config"] = {}
-            logConfig["config"]["filters"] = {}
-            dict["loggerConfig"].append(logConfig)
-    return dict
-
-
-def add_poller_config(service, dict):
-    dict["pollerConfig"] = {}
-    pollerConfig = {}
-    pollerConfig["config"] = {}
-
-    for key, value in service_plugin_mapping.items():
-        if (key == service):
-            pollerConfig["name"] = value
+    service_dict['ports'] = ports
+    return service_dict
+
+
+def add_logger_config(service_dict, service):
+    '''
+    Add logger config
+    '''
+    service_dict["loggerConfig"] = []
+    fluentd_plugins = configurator.get_fluentd_plugins_mapping().keys()
+    for item in fluentd_plugins:
+        if item.startswith(service.split(".")[0]):
+            log_config = {}
+            log_config["name"] = item
+            log_config["config"] = {}
+            log_config["config"]["filters"] = {}
+            service_dict["loggerConfig"].append(log_config)
+    return service_dict
+
+def add_poller_config(service, service_dict):
+    '''
+    Add poller config
+    '''
+    service_dict["pollerConfig"] = {}
+    poller_config = {}
+    poller_config["config"] = {}
+
+    for key, value in SERVICE_PLUGIN_MAPPING.items():
+        if key == service:
+            poller_config["name"] = value
             break
 
-    config = configurator.get_metrics_plugins_params(pollerConfig["name"])
+    config = configurator.get_metrics_plugins_params(poller_config["name"])
     for item in config["plugins"]:
-        if(item.get("config") and item.get("name") == pollerConfig["name"]):
+        if item.get("config") and item.get("name") == poller_config["name"]:
             for item1 in item["config"]:
-                pollerConfig["config"][item1["fieldName"]] = item1["defaultValue"]
-    dict["pollerConfig"].update(pollerConfig)
+                poller_config["config"][item1["fieldName"]] = item1["defaultValue"]
+    service_dict["pollerConfig"].update(poller_config)
 
-    return dict
+    return service_dict
 
 
-def add_agent_config(service, dict):
+def add_agent_config(service, service_dict):
     '''
     Find the input config for the plugin fieldname:defaultvalue
     :param service: name of the service
     :param dict: poller_dict as the input
     :return:
     '''
-    dict["agentConfig"] = {}
-    agentConfig = {}
-    agentConfig["config"] = {}
-    for key, value in service_plugin_mapping.items():
-        if(key == service):
-            agentConfig["name"] = value
+    service_dict["agentConfig"] = {}
+    agent_config = {}
+    agent_config["config"] = {}
+    for key, value in SERVICE_PLUGIN_MAPPING.items():
+        if key == service:
+            agent_config["name"] = value
             break
-    config = configurator.get_metrics_plugins_params(agentConfig["name"])
+    config = configurator.get_metrics_plugins_params(agent_config["name"])
     for item in config["plugins"]:
-        if(item.get("config") and item.get("name") == agentConfig["name"]):
-            #Config specific to jvm plugin
-            if agentConfig["name"] == "jvm":
-                agentConfig["config"]["process"] = service
+        if item.get("config") and item.get("name") == agent_config["name"]:
+            # Config specific to jvm plugin
+            if agent_config["name"] == "jvm":
+                agent_config["config"]["process"] = service
                 break
-            if agentConfig["name"] == "kafkatopic":
-                agentConfig["config"]["process"] = service
+            if agent_config["name"] == "kafkatopic":
+                agent_config["config"]["process"] = service
                 for parameter in item["config"]:
-                    agentConfig["config"][parameter["fieldName"]] = parameter["defaultValue"]
+                    agent_config["config"][parameter["fieldName"]] = parameter["defaultValue"]
                 break
             for parameter in item["config"]:
-                agentConfig["config"][parameter["fieldName"]] = parameter["defaultValue"]
+                agent_config["config"][parameter["fieldName"]] = parameter["defaultValue"]
 
-    #In apache plugin replace the port default value with the listening ports for apache/httpd,
-    #if there are multiple listening ports for the PID assosciate the first port with the PID
-    if(service == "apache"):
-        if(len(dict["ports"]) != 0):
-            agentConfig["config"]["port"] = dict["ports"][0]
-            if(agentConfig["config"]["port"] == "443"):
-                agentConfig["config"]["secure"] = "true"
+    # In apache plugin replace the port default value with the listening ports for apache/httpd,
+    # if there are multiple listening ports for the PID assosciate the first
+    # port with the PID
+    if service == "apache":
+        if len(service_dict["ports"]) != 0:
+            agent_config["config"]["port"] = service_dict["ports"][0]
+            if agent_config["config"]["port"] == "443":
+                agent_config["config"]["secure"] = "true"
 
-    dict["agentConfig"].update(agentConfig)
-    return dict
+    service_dict["agentConfig"].update(agent_config)
+    return service_dict
 
 
 def discover_services():
@@ -296,91 +356,74 @@ def discover_services():
     '''
     logger.info("Discover service started")
     discovery = {}
-    for service in services:
-        if service == "tpcc":
-            '''
-            If service is tpcc the check for the existance of .tpcc_discovery file
-            '''
-            if os.path.exists("/opt/VDriver/.tpcc_discovery"):
-                port_dict = {}
-                port_dict["loggerConfig"] = []
-                port_dict["agentConfig"] = {}
-                agent_dict = add_agent_config(service, port_dict)
-                final_dict = agent_dict
-                discovery[service_name[service]] = []
-                discovery[service_name[service]].append(final_dict)
-
-	if service == "hxconnect":
-	    '''
-	    If service is hxconnect, check for existance of .hxconnect_discovery file
-	    '''
-	    if os.path.exists("/opt/VDriver/.hxconnect_discovery"):
-                port_dict = {}
-                port_dict["loggerConfig"] = []
-                port_dict["agentConfig"] = {}
-                agent_dict = add_agent_config(service, port_dict)
-                final_dict = agent_dict
-                discovery[service_name[service]] = []
-                discovery[service_name[service]].append(final_dict)
-
-        if service == "esalogstore":
-            '''
-            If service is esalogstore then check for the existance of esa_conf.json file
-            '''
-            if os.path.exists("/opt/esa_conf.json"):
-                port_dict = {}
+    for service in SERVICES:
+        if (service == "tpcc" and os.path.exists("/opt/VDriver/.tpcc_discovery")) or \
+                (service == "hxconnect" and os.path.exists("/opt/VDriver/.hxconnect_discovery")):
+            port_dict = {}
+            port_dict["loggerConfig"] = []
+            port_dict["agentConfig"] = {}
+            final_dict = add_agent_config(service, port_dict)
+            discovery[SERVICE_NAME[service]] = []
+            discovery[SERVICE_NAME[service]].append(final_dict)
+        elif service == "esalogstore" and os.path.exists("/opt/esa_conf.json"):
+            port_dict["loggerConfig"] = []
+            port_dict["agentConfig"] = {}
+            final_dict = add_logger_config(port_dict, service)
+            discovery[SERVICE_NAME[service]] = []
+            discovery[SERVICE_NAME[service]].append(final_dict)
+
+        pid_list = get_process_id(service)
+        if not pid_list:
+            continue
+        discovery[SERVICE_NAME[service]] = []
+        for item in pid_list:
+            service_pid_dict = {}
+            service_pid_dict["PID"] = []
+            service_pid_dict["PID"] = item["process_id"]
+
+            # Add PID, cpuUsage, memUsage, status to service_discovery
+            service_pid_dict["PID"] = item["process_id"]
+            service_pid_dict["user"] = item["user"]
+            service_pid_dict["cpuUsage"] = item["cpuUsage"]
+            service_pid_dict["memUsage"] = item["memUsage"]
+            service_pid_dict["status"] = item["status"]
+
+            # Add state, threads assosciated with the service PID
+            status_dict = add_status(service_pid_dict)
+
+            # Add listening ports assosciated with the service PID
+            port_dict = add_ports(status_dict, service)
+
+            if service in POLLER_PLUGIN:
                 port_dict["loggerConfig"] = []
                 port_dict["agentConfig"] = {}
+                final_dict = add_poller_config(service, port_dict)
+            else:
                 logger_dict = add_logger_config(port_dict, service)
-                final_dict = logger_dict
-                discovery[service_name[service]] = []
-                discovery[service_name[service]].append(final_dict)
-
-        pidList = get_process_id(service)
-        if(len(pidList) != 0):
-            discovery[service_name[service]] = []
-            for item in pidList:
-                service_pid_dict = {}
-                service_pid_dict["PID"] = []
-                service_pid_dict["PID"] = item["process_id"]
-
-                #Add PID, cpuUsage, memUsage, status to service_discovery
-                service_pid_dict["PID"] = item["process_id"]
-                service_pid_dict["user"] = item["user"]
-                service_pid_dict["cpuUsage"] = item["cpuUsage"]
-                service_pid_dict["memUsage"] = item["memUsage"]
-                service_pid_dict["status"] = item["status"]
-
-                #Add state, threads assosciated with the service PID
-                status_dict = add_status(service_pid_dict)
-
-                #Add listening ports assosciated with the service PID
-                port_dict = add_ports(status_dict, service)
-
-                if service not in poller_plugin:
-                    #Add logger config to the service PID
-                    logger_dict = add_logger_config(port_dict, service)
-
-                    #Add poller config to the service dict
-                    logger_dict["pollerConfig"] = {}
-
-                    #Add agent config to the service dict
-                    agent_dict = add_agent_config(service, logger_dict)
-
-                    #final_dict assosciated with the service PID
-                    final_dict = agent_dict
-
-                else:
-                    port_dict["loggerConfig"] = []
-                    port_dict["agentConfig"] = {}
-
-                    # Add poller config to the service dict
-                    poller_dict = add_poller_config(service, port_dict)
-
-                    # final_dict assosciated with the service PID
-                    final_dict = poller_dict
-
-                discovery[service_name[service]].append(final_dict)
+                logger_dict["pollerConfig"] = {}
+                final_dict = add_agent_config(service, logger_dict)
+
+            discovery[SERVICE_NAME[service]].append(final_dict)
+
+    discovery["HadoopLog"] = list()
+    for service_name in get_hadoop_running_service_list():
+        logger.info("Hadoop services are %s" %service_name)
+        print ("Hadoop services are %s" %service_name)
+        print(HADOOP_SERVICE[service_name])
+
+        for service in HADOOP_SERVICE[service_name]['service-list']:
+            logger.info("service detail: {}".format(service))
+            port_dict = dict()
+            port_dict["loggerConfig"] = list()
+            port_dict["agentConfig"] = dict()
+            print(service)
+            final_dict = add_logger_config(port_dict, service)
+            print(final_dict)
+            if not final_dict["loggerConfig"]:
+                continue
+
+            final_dict['pollerConfig'] = dict()
+            discovery["HadoopLog"].append(final_dict)
 
     logger.info("Discovered service %s", discovery)
     return discovery