METRON-2050: Automatically populate a list of enrichments from HBase

metron-deployment/packaging/ambari/README.md

@@ -61,7 +61,7 @@ The layout of `/common-services/METRON/CURRENT` is
 * `kerberos.json`
   * Defines the keytabs and other Kerberos configuration to be used when Kerberizing a cluster
 * `metainfo.xml`
-  * Defines the METRON service, along with required packages, services, etc.
+  * Defines the METRON service, along with required packages, services, etc. If you need to have the MPack install a new package (e.g. RPM, DEB), add it here.
 * `service_advisor.py`
   * Handles component layout and validation, along with handling some configurations for other services or that needs configs from other services.
 

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/configuration/metron-enrichment-env.xml

@@ -135,6 +135,30 @@
     <description>The HBase column family which will hold threatintel data in HBase.</description>
     <display-name>HBase Table Column Family</display-name>
   </property>
+  <property>
+    <name>enrichment_list_hbase_provider_impl</name>
+    <value>org.apache.metron.hbase.HTableProvider</value>
+    <description>The HBase table provider implementation to use for obtaining table references.</description>
+    <display-name>Enrichment List HBase Table Provider Implementation</display-name>
+  </property>
+  <property>
+    <name>enrichment_list_hbase_coprocessor_impl</name>
+    <value>org.apache.metron.hbase.coprocessor.EnrichmentCoprocessor</value>
+    <description>The HBase coprocessor implementation to use for managing the enrichment list.</description>
+    <display-name>Enrichment List HBase Coprocessor Implementation</display-name>
+  </property>
+  <property>
+    <name>enrichment_list_hbase_table</name>
+    <value>enrichment_list</value>
+    <description>The HBase table which will hold the list of enrichment types gathered by the enrichment coprocessor.</description>
+    <display-name>Enrichment List HBase Table</display-name>
+  </property>
+  <property>
+    <name>enrichment_list_hbase_cf</name>
+    <value>t</value>
+    <description>The HBase column family which will hold enrichment list data in HBase.</description>
+    <display-name>Enrichment List HBase Column Family</display-name>
+  </property>
 
   <!--
     storm common parameters

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/metainfo.xml

@@ -440,6 +440,9 @@
             <package>
               <name>metron-data-management</name>
             </package>
+            <package>
+              <name>metron-hbase-server</name>
+            </package>
             <package>
               <name>metron-management</name>
             </package>
@@ -497,7 +500,6 @@
             <package>
               <name>metron-performance</name>
             </package>
-
           </packages>
         </osSpecific>
         <osSpecific>

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/enrichment_commands.py

@@ -46,6 +46,7 @@ def __init__(self, params):
         self.__kafka_configured = os.path.isfile(self.__params.enrichment_kafka_configured_flag_file)
         self.__kafka_acl_configured = os.path.isfile(self.__params.enrichment_kafka_acl_configured_flag_file)
         self.__hbase_configured = os.path.isfile(self.__params.enrichment_hbase_configured_flag_file)
+        self.__hbase_coprocessor_configured = os.path.isfile(self.__params.enrichment_hbase_coprocessor_configured_flag_file)
         self.__hbase_acl_configured = os.path.isfile(self.__params.enrichment_hbase_acl_configured_flag_file)
         self.__maxmind_configured = os.path.isfile(self.__params.enrichment_maxmind_configured_flag_file)
 
@@ -64,6 +65,9 @@ def is_kafka_acl_configured(self):
     def is_hbase_configured(self):
         return self.__hbase_configured
 
+    def is_hbase_coprocessor_configured(self):
+        return self.__hbase_coprocessor_configured
+
     def is_hbase_acl_configured(self):
         return self.__hbase_acl_configured
 
@@ -79,6 +83,9 @@ def set_kafka_acl_configured(self):
     def set_hbase_configured(self):
         metron_service.set_configured(self.__params.metron_user, self.__params.enrichment_hbase_configured_flag_file, "Setting HBase configured to True for enrichment")
 
+    def set_hbase_coprocessor_configured(self):
+        metron_service.set_configured(self.__params.metron_user, self.__params.enrichment_hbase_coprocessor_configured_flag_file, "Setting HBase coprocessor configured to True for enrichment")
+
     def set_hbase_acl_configured(self):
         metron_service.set_configured(self.__params.metron_user, self.__params.enrichment_hbase_acl_configured_flag_file, "Setting HBase ACL configured to True for enrichment")
 
@@ -206,14 +213,64 @@ def is_topology_active(self, env):
     def create_hbase_tables(self):
         Logger.info("Creating HBase Tables")
         metron_service.create_hbase_table(self.__params,
-                                        self.__params.enrichment_hbase_table,
-                                        self.__params.enrichment_hbase_cf)
+                                          self.__params.enrichment_hbase_table,
+                                          self.__params.enrichment_hbase_cf)
+        metron_service.create_hbase_table(self.__params,
+                                        self.__params.enrichment_list_hbase_table,
+                                        self.__params.enrichment_list_hbase_cf)
         metron_service.create_hbase_table(self.__params,
                                         self.__params.threatintel_hbase_table,
                                         self.__params.threatintel_hbase_cf)
         Logger.info("Done creating HBase Tables")
         self.set_hbase_configured()
 
+    def load_enrichment_coprocessor(self):
+        Logger.info("Creating HDFS location for enrichment coprocessor and loading from local disk")
+
+        self.__params.HdfsResource(self.__params.hbase_coprocessor_hdfs_dir,
+                                   type="directory",
+                                   action="create_on_execute",
+                                   owner=self.__params.metron_user,
+                                   group=self.__params.metron_group,
+                                   mode=0755,
+                                   source=self.__params.hbase_coprocessor_local_dir,
+                                   recursive_chown = True)
+
+        Logger.info("Loading HBase coprocessor for enrichments")
+        Logger.info("See https://hbase.apache.org/1.1/book.html#load_coprocessor_in_shell for more detail")
+
+        Logger.info("HBase coprocessor setup - first disabling the enrichments HBase table.")
+        command_template = "echo \"disable '{0}'\" | hbase shell -n"
+        command = command_template.format(self.__params.enrichment_hbase_table)
+        Logger.info("Executing command " + command)
+        Execute(command, user=self.__params.metron_user, tries=1, logoutput=True)
+
+        Logger.info("HBase coprocessor setup - altering table and adding coprocessor.")
+        command_template = "{0}/bin/load_enrichment_coprocessor.sh {1} {2} {3} {4} {5}"
+        command = command_template.format(self.__params.metron_home,
+                                          self.__params.enrichment_hbase_table,
+                                          self.__params.hdfs_url,
+                                          self.__params.hbase_coprocessor_hdfs_dir,
+                                          self.__params.enrichment_list_hbase_coprocessor_impl,
+                                          self.__params.zookeeper_quorum)
+        Logger.info("Executing command " + command)
+        Execute(command, user=self.__params.metron_user, tries=1, logoutput=True)
+
+        Logger.info("HBase coprocessor setup - re-enabling enrichments table.")
+        command_template = "echo \"enable'{0}'\" | hbase shell -n"
+        command = command_template.format(self.__params.enrichment_hbase_table)
+        Logger.info("Executing command " + command)
+        Execute(command, user=self.__params.metron_user, tries=1, logoutput=True)
+
+        Logger.info("HBase coprocessor setup - verifying coprocessor was loaded. The coprocessor should be listed in the TABLE_ATTRIBUTES.")
+        command_template = "echo \"describe '{0}'\" | hbase shell -n"
+        command = command_template.format(self.__params.enrichment_hbase_table)
+        Logger.info("Executing command " + command)
+        Execute(command, user=self.__params.metron_user, tries=1, logoutput=True)
+
+        Logger.info("Done loading HBase coprocessor for enrichments")
+        self.set_hbase_coprocessor_configured()
+
     def set_hbase_acls(self):
         Logger.info("Setting HBase ACLs")
         if self.__params.security_enabled:
@@ -232,6 +289,15 @@ def set_hbase_acls(self):
                 user=self.__params.hbase_user
                 )
 
+        add_enrichment_list_acl_cmd = cmd.format(self.__params.metron_user, self.__params.enrichment_list_hbase_table)
+        Execute(add_enrichment_list_acl_cmd,
+                tries=3,
+                try_sleep=5,
+                logoutput=False,
+                path='/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin',
+                user=self.__params.hbase_user
+                )
+
         add_threatintel_acl_cmd = cmd.format(self.__params.metron_user, self.__params.threatintel_hbase_table)
         Execute(add_threatintel_acl_cmd,
                 tries=3,
@@ -260,31 +326,42 @@ def service_check(self, env):
 
         Logger.info("Checking HBase for Enrichment")
         metron_service.check_hbase_table(
-          self.__params,
-          self.__params.enrichment_hbase_table)
+            self.__params,
+            self.__params.enrichment_hbase_table)
         metron_service.check_hbase_column_family(
-          self.__params,
-          self.__params.enrichment_hbase_table,
-          self.__params.enrichment_hbase_cf)
+            self.__params,
+            self.__params.enrichment_hbase_table,
+            self.__params.enrichment_hbase_cf)
+
+        Logger.info("Checking HBase for Enrichment List")
+        metron_service.check_hbase_table(
+            self.__params,
+            self.__params.enrichment_list_hbase_table)
+        metron_service.check_hbase_column_family(
+            self.__params,
+            self.__params.enrichment_list_hbase_table,
+            self.__params.enrichment_list_hbase_cf)
 
         Logger.info("Checking HBase for Threat Intel")
         metron_service.check_hbase_table(
-          self.__params,
-          self.__params.threatintel_hbase_table)
+            self.__params,
+            self.__params.threatintel_hbase_table)
         metron_service.check_hbase_column_family(
-          self.__params,
-          self.__params.threatintel_hbase_table,
-          self.__params.threatintel_hbase_cf)
+            self.__params,
+            self.__params.threatintel_hbase_table,
+            self.__params.threatintel_hbase_cf)
 
         if self.__params.security_enabled:
-
-          Logger.info('Checking Kafka ACLs for Enrichment')
-          metron_service.check_kafka_acls(self.__params, self.__get_topics())
-          metron_service.check_kafka_acl_groups(self.__params, self.__get_kafka_acl_groups())
-
-          Logger.info("Checking HBase ACLs for Enrichment")
-          metron_service.check_hbase_acls(self.__params, self.__params.enrichment_hbase_table)
-          metron_service.check_hbase_acls(self.__params, self.__params.threatintel_hbase_table)
+            Logger.info('Checking Kafka ACLs for Enrichment')
+            metron_service.check_kafka_acls(self.__params, self.__get_topics())
+            metron_service.check_kafka_acl_groups(self.__params, self.__get_kafka_acl_groups())
+
+            Logger.info("Checking HBase ACLs for Enrichment")
+            metron_service.check_hbase_acls(self.__params, self.__params.enrichment_hbase_table)
+            Logger.info("Checking HBase ACLs for Enrichment List")
+            metron_service.check_hbase_acls(self.__params, self.__params.enrichment_list_hbase_table)
+            Logger.info("Checking HBase ACLs for Threat Intel")
+            metron_service.check_hbase_acls(self.__params, self.__params.threatintel_hbase_table)
 
         Logger.info("Checking for Enrichment topology")
         if not self.is_topology_active(env):

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/enrichment_master.py

@@ -49,8 +49,8 @@ def configure(self, env, upgrade_type=None, config_dir=None):
             group=params.metron_group)
 
         if not metron_service.is_zk_configured(params):
-          metron_service.init_zk_config(params)
-          metron_service.set_zk_configured(params)
+            metron_service.init_zk_config(params)
+            metron_service.set_zk_configured(params)
         metron_service.refresh_configs(params)
 
         Logger.info("Calling security setup")
@@ -74,6 +74,8 @@ def start(self, env, upgrade_type=None):
             commands.init_kafka_acls()
         if not commands.is_hbase_configured():
             commands.create_hbase_tables()
+        if not commands.is_hbase_coprocessor_configured():
+            commands.load_enrichment_coprocessor()
         if params.security_enabled and not commands.is_hbase_acl_configured():
             commands.set_hbase_acls()
         if not commands.is_maxmind_configured():

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/metron_service.py

@@ -115,6 +115,21 @@ def build_global_config_patch(params, patch_file):
         "path": "/parser.error.topic",
         "value": "{{parser_error_topic}}"
     },
+    {
+        "op": "add",
+        "path": "/enrichment.list.hbase.provider.impl",
+        "value": "{{enrichment_list_hbase_provider_impl}}"
+    },
+    {
+        "op": "add",
+        "path": "/enrichment.list.hbase.table",
+        "value": "{{enrichment_list_hbase_table}}"
+    },
+    {
+        "op": "add",
+        "path": "/enrichment.list.hbase.cf",
+        "value": "{{enrichment_list_hbase_cf}}"
+    },
     {
         "op": "add",
         "path": "/update.hbase.table",
@@ -384,7 +399,6 @@ def create_hbase_table(params, table, cf):
             user=params.hbase_user
             )
 
-
 def check_hbase_table(params, table):
     """
     Validates that an HBase table exists.  An exception is raised if the table

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/params/params_linux.py

@@ -43,11 +43,14 @@
 
 hostname = config['hostname']
 metron_home = status_params.metron_home
+metron_apps_hdfs_dir = config['configurations']['metron-env']['metron_apps_hdfs_dir']
 
 parsers = status_params.parsers
 parser_error_topic = config['configurations']['metron-parsers-env']['parser_error_topic']
-geoip_hdfs_dir = "/apps/metron/geo/default/"
-asn_hdfs_dir = "/apps/metron/asn/default/"
+geoip_hdfs_dir = metron_apps_hdfs_dir + "/geo/default/"
+asn_hdfs_dir = metron_apps_hdfs_dir + "/asn/default/"
+hbase_coprocessor_local_dir = format("{metron_home}/coprocessor")
+hbase_coprocessor_hdfs_dir = metron_apps_hdfs_dir + "/coprocessor"
 metron_user = status_params.metron_user
 metron_group = config['configurations']['metron-env']['metron_group']
 metron_log_dir = config['configurations']['metron-env']['metron_log_dir']
@@ -89,6 +92,7 @@
 enrichment_kafka_configured_flag_file = status_params.enrichment_kafka_configured_flag_file
 enrichment_kafka_acl_configured_flag_file = status_params.enrichment_kafka_acl_configured_flag_file
 enrichment_hbase_configured_flag_file = status_params.enrichment_hbase_configured_flag_file
+enrichment_hbase_coprocessor_configured_flag_file = status_params.enrichment_hbase_coprocessor_configured_flag_file
 enrichment_hbase_acl_configured_flag_file = status_params.enrichment_hbase_acl_configured_flag_file
 enrichment_maxmind_configured_flag_file = status_params.enrichment_maxmind_configured_flag_file
 indexing_configured_flag_file = status_params.indexing_configured_flag_file
@@ -146,6 +150,9 @@
 solr_principal_name = config['configurations']['solr-config-env']['solr_principal_name']
 solr_keytab_path = config['configurations']['solr-config-env']['solr_keytab_path']
 
+# HDFS
+hdfs_url = status_params.hdfs_url
+
 # Storm
 storm_rest_addr = status_params.storm_rest_addr
 
@@ -164,8 +171,6 @@
     kafka_brokers = (':' + kafka_broker_port + ',').join(config['clusterHostInfo']['kafka_broker_hosts'])
     kafka_brokers += ':' + kafka_broker_port
 
-metron_apps_hdfs_dir = config['configurations']['metron-env']['metron_apps_hdfs_dir']
-
 # the double "format" is not an error - we are pulling in a jinja-templated param. This is a bit of a hack, but works
 # well enough until we find a better way via Ambari
 metron_temp_grok_path = format(format(config['configurations']['metron-rest-env']['metron_temp_grok_path']))
@@ -208,6 +213,11 @@
 enrichment_hbase_provider_impl = 'org.apache.metron.hbase.HTableProvider'
 enrichment_hbase_table = status_params.enrichment_hbase_table
 enrichment_hbase_cf = status_params.enrichment_hbase_cf
+# coprocessor config for enrichment list
+enrichment_list_hbase_provider_impl = status_params.enrichment_list_hbase_provider_impl
+enrichment_list_hbase_coprocessor_impl = status_params.enrichment_list_hbase_coprocessor_impl
+enrichment_list_hbase_table = status_params.enrichment_list_hbase_table
+enrichment_list_hbase_cf = status_params.enrichment_list_hbase_cf
 update_hbase_table = status_params.update_hbase_table
 update_hbase_cf = status_params.update_hbase_cf
 
@@ -475,4 +485,4 @@
 knox_hosts = default("/clusterHostInfo/knox_gateway_hosts", [])
 knox_host = ''
 if not len(knox_hosts) == 0:
-    knox_host = knox_hosts[0]
+    knox_host = knox_hosts[0]

metron-deployment/packaging/ambari/metron-mpack/src/main/resources/common-services/METRON/CURRENT/package/scripts/params/status_params.py

@@ -44,11 +44,16 @@
 enrichment_kafka_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_kafka_configured'
 enrichment_kafka_acl_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_kafka_acl_configured'
 enrichment_hbase_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_hbase_configured'
+enrichment_hbase_coprocessor_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_hbase_coprocessor_configured'
 enrichment_hbase_acl_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_hbase_acl_configured'
 enrichment_maxmind_configured_flag_file = metron_zookeeper_config_path + '/../metron_enrichment_maxmind_configured'
 
 enrichment_hbase_table = config['configurations']['metron-enrichment-env']['enrichment_hbase_table']
 enrichment_hbase_cf = config['configurations']['metron-enrichment-env']['enrichment_hbase_cf']
+enrichment_list_hbase_provider_impl = config['configurations']['metron-enrichment-env']['enrichment_list_hbase_provider_impl']
+enrichment_list_hbase_coprocessor_impl = config['configurations']['metron-enrichment-env']['enrichment_list_hbase_coprocessor_impl']
+enrichment_list_hbase_table = config['configurations']['metron-enrichment-env']['enrichment_list_hbase_table']
+enrichment_list_hbase_cf = config['configurations']['metron-enrichment-env']['enrichment_list_hbase_cf']
 threatintel_hbase_table = config['configurations']['metron-enrichment-env']['threatintel_hbase_table']
 threatintel_hbase_cf = config['configurations']['metron-enrichment-env']['threatintel_hbase_cf']
 update_hbase_table = config['configurations']['metron-indexing-env']['update_hbase_table']
@@ -99,6 +104,12 @@
 metron_management_ui_host = default("/clusterHostInfo/metron_management_ui_hosts", [hostname])[0]
 metron_management_ui_port = config['configurations']['metron-management-ui-env']['metron_management_ui_port']
 
+# HDFS
+# This should always grab the full namenode url, e.g. "hdfs://mynamenodehost:8020"
+# In the case of namenode HA, it will refer to the nameservice ID
+# https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HDFSHighAvailabilityWithNFS.html
+hdfs_url = config["configurations"]["core-site"]["fs.defaultFS"]
+
 # Storm
 storm_rest_addr = config['configurations']['metron-env']['storm_rest_addr']
 
@@ -130,4 +141,4 @@
 metron_user_hdfs_dir_configured_flag_file = metron_zookeeper_config_path + '/../metron_user_hdfs_dir_configured'
 
 # Knox
-metron_knox_installed_flag_file = metron_zookeeper_config_path + '/../metron_knox_installed'
+metron_knox_installed_flag_file = metron_zookeeper_config_path + '/../metron_knox_installed'