def check_hbase_acls(params, table, user=None, permissions="READ,WRITE"):
    """
    Validates that HBase table permissions exist for a user. An exception is
    raised if the permissions do not exist.
    :param params:
    :param table: The name of the HBase table.
    :param user: The name of the user.
    :param permissions: The permissions that should exist.
    """
    if user is None:
        user = params.metron_user
    Logger.info("Checking HBase ACLs; table={0}, user={1}, permissions={2}".format(table, user, permissions))

    # if needed kinit as 'hbase'
    if params.security_enabled:
        kinit(params.kinit_path_local,
              params.hbase_keytab_path,
              params.hbase_principal_name,
              execute_user=params.hbase_user)



    template = """echo "user_permission '{0}'" | \
      hbase shell -n | \
      grep " {1} " | \
      grep "actions={2}"
    """
    cmd = template.format(table, user, permissions)
    err_msg = "Missing HBase access; table={0}, user={1}, permissions={2}".format(table, user, permissions)
    execute(cmd, user=params.hbase_user, err_msg=err_msg)

def check_kafka_topics(params, topics):
    """
    Validates that the Kafka topics exist.  An exception is raised if any of the
    topics do not exist.
    :param params:
    :param topics: A list of topic names.
    """

    # if needed kinit as 'metron'
    if params.security_enabled:
        kinit(params.kinit_path_local,
              params.metron_keytab_path,
              params.metron_principal_name,
              execute_user=params.metron_user)

    template = """{0}/kafka-topics.sh \
      --zookeeper {1} \
      --list | \
      awk 'BEGIN {{cnt=0;}} /{2}/ {{cnt++}} END {{if (cnt > 0) {{exit 0}} else {{exit 1}}}}'"""

    for topic in topics:
        Logger.info("Checking existence of Kafka topic '{0}'".format(topic))
        cmd = template.format(params.kafka_bin_dir, params.zookeeper_quorum, topic)
        err_msg = "Missing Kafka topic; topic={0}".format(topic)
        execute(cmd, user=params.kafka_user, err_msg=err_msg)

    def set_hbase_acls(self):
        Logger.info("Setting HBase ACLs")
        if self.__params.security_enabled:
            kinit(self.__params.kinit_path_local,
                  self.__params.hbase_keytab_path,
                  self.__params.hbase_principal_name,
                  execute_user=self.__params.hbase_user)

        cmd = "echo \"grant '{0}', 'RW', '{1}'\" | hbase shell -n"
        add_enrichment_acl_cmd = cmd.format(self.__params.metron_user, self.__params.enrichment_hbase_table)
        Execute(add_enrichment_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,
                try_sleep=5,
                logoutput=False,
                path='/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin',
                user=self.__params.hbase_user
                )

        Logger.info("Done setting HBase ACLs")
        self.set_hbase_acl_configured()

def get_running_topologies(params):
  Logger.info('Getting Running Storm Topologies from Storm REST Server')
  Logger.info('Security enabled? ' + str(params.security_enabled))

  # Want to sudo to the metron user and kinit as them so we aren't polluting root with Metron's Kerberos tickets.
  # This is becuase we need to run a command with a return as the metron user. Sigh
  negotiate = '--negotiate -u : ' if params.security_enabled else ''
  cmd = ambari_format(
    'curl --max-time 3 ' + negotiate + '{storm_rest_addr}/api/v1/topology/summary')

  if params.security_enabled:
    kinit(params.kinit_path_local,
          params.metron_keytab_path,
          params.metron_principal_name,
          execute_user=params.metron_user)

  Logger.info('Running cmd: ' + cmd)
  return_code, stdout, stderr = get_user_call_output(cmd,
                                                     user=params.metron_user,
                                                     is_checked_call=False)

  if (return_code != 0):
    return {}

  try:
    stormjson = json.loads(stdout)
  except ValueError, e:
    Logger.info('Stdout: ' + str(stdout))
    Logger.info('Stderr: ' + str(stderr))
    Logger.exception(str(e))
    return {}

    def start_parser_topologies(self, env):
        Logger.info("Starting Metron parser topologies: {0}".format(self.__get_aggr_parsers(self.__params)))
        start_cmd_template = """{0}/bin/start_parser_topology.sh \
                                    -k {1} \
                                    -z {2} \
                                    -s {3} \
                                    -ksp {4}"""
        if self.__params.security_enabled:
            # Append the extra configs needed for secured cluster.
            start_cmd_template = start_cmd_template + ' -e ~' + self.__params.metron_user + '/.storm/storm.config'
            metron_security.kinit(self.__params.kinit_path_local,
                                  self.__params.metron_keytab_path,
                                  self.__params.metron_principal_name,
                                  execute_user=self.__params.metron_user)

        stopped_parsers = set(self.__get_aggr_parsers(self.__params)) - self.get_running_topology_names(env)
        Logger.info('Parsers that need started: ' + str(stopped_parsers))

        for parser in stopped_parsers:
            Logger.info('Starting ' + parser)
            start_cmd = start_cmd_template.format(self.__params.metron_home,
                                                  self.__params.kafka_brokers,
                                                  self.__params.zookeeper_quorum,
                                                  parser,
                                                  self.__params.kafka_security_protocol)
            Execute(start_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)

        Logger.info('Finished starting parser topologies')

    def start_rest_application(self):
        """
        Start the REST application
        """
        Logger.info('Starting REST application')

        if self.__params.security_enabled:
            kinit(self.__params.kinit_path_local,
            self.__params.metron_keytab_path,
            self.__params.metron_principal_name,
            execute_user=self.__params.metron_user)

        # Get the PID associated with the service
        pid_file = format("{metron_rest_pid_dir}/{metron_rest_pid}")
        pid = get_user_call_output.get_user_call_output(format("cat {pid_file}"), user=self.__params.metron_user, is_checked_call=False)[1]
        process_id_exists_command = format("ls {pid_file} >/dev/null 2>&1 && ps -p {pid} >/dev/null 2>&1")

        # Set the password with env variable instead of param to avoid it showing in ps
        cmd = format((
          "export METRON_JDBC_PASSWORD={metron_jdbc_password!p};"
          "export JAVA_HOME={java_home};"
          "export METRON_REST_CLASSPATH={metron_rest_classpath};"
          "export METRON_INDEX_CP={metron_indexing_classpath};"
          "export METRON_LOG_DIR={metron_log_dir};"
          "export METRON_PID_FILE={pid_file};"
          "{metron_home}/bin/metron-rest.sh;"
          "unset METRON_JDBC_PASSWORD;"
        ))

        Execute(cmd,
                user = self.__params.metron_user,
                logoutput=True,
                not_if = process_id_exists_command,
                timeout=60)
        Logger.info('Done starting REST application')

    def solr_schema_install(self, env):
        from params import params
        env.set_params(params)
        Logger.info("Installing Solr schemas")
        if self.__params.security_enabled:
            metron_security.kinit(self.__params.kinit_path_local,
                                  self.__params.solr_keytab_path,
                                  self.__params.solr_principal_name,
                                  self.__params.solr_user)

        try:
            commands = IndexingCommands(params)
            for collection_name in commands.get_solr_schemas():
                # install the schema
                cmd = format((
                "export ZOOKEEPER={solr_zookeeper_url};"
                "export SECURITY_ENABLED={security_enabled};"
                ))
                cmd += "{0}/bin/create_collection.sh {1};"

                Execute(
                cmd.format(params.metron_home, collection_name),
                user=self.__params.solr_user)
            return True

        except Exception as e:
            msg = "WARNING: Solr schemas could not be installed.  " \
                  "Is Solr running?  Will reattempt install on next start.  error={0}"
            Logger.warning(msg.format(e))
            return False

    def status(self, env):
        from params import status_params
        env.set_params(status_params)
        commands = EnrichmentCommands(status_params)

        if status_params.security_enabled:
            metron_security.kinit(status_params.kinit_path_local,
                                  status_params.metron_keytab_path,
                                  status_params.metron_principal_name,
                                  execute_user=status_params.metron_user)

        if not commands.is_topology_active(env):
            raise ComponentIsNotRunning()

    def stop(self, env, upgrade_type=None):
        from params import params

        env.set_params(params)
        commands = EnrichmentCommands(params)

        if params.security_enabled:
            metron_security.kinit(params.kinit_path_local,
                                  params.metron_keytab_path,
                                  params.metron_principal_name,
                                  execute_user=params.metron_user)

        commands.stop_enrichment_topology(env)

    def start_pcap_topology(self, env):
        Logger.info('Starting Metron PCAP topology')
        start_cmd_template = """{0}/bin/start_pcap_topology.sh"""
        if not self.is_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            start_cmd = start_cmd_template.format(self.__params.metron_home)
            Execute(start_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)
        else :
            Logger.info('PCAP topology already started')

        Logger.info('Finished starting pcap topologies')

    def stop_pcap_topology(self, env):
        Logger.info('Stopping Metron PCAP topology')
        if self.is_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            stop_cmd = 'storm kill ' + self.__pcap_topology
            Execute(stop_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)

        else :
            Logger.info('PCAP topology already stopped')

        Logger.info('Finished starting PCAP topologies')

    def stop_random_access_indexing_topology(self, env):
        Logger.info('Stopping ' + self.__random_access_indexing_topology)

        if self.is_random_access_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            stop_cmd = 'storm kill ' + self.__random_access_indexing_topology
            Execute(stop_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)

        else:
            Logger.info("Random Access Indexing topology already stopped")

        Logger.info('Done stopping random access indexing topologies')

    def stop_parser_topologies(self, env):
        Logger.info('Stopping parsers')

        running_parsers = set(self.get_parser_aggr_topology_names(self.__params)) & self.get_running_topology_names(env)
        Logger.info('Parsers that need stopped: ' + str(running_parsers))

        for parser in running_parsers:
            Logger.info('Stopping ' + parser)
            stop_cmd = 'storm kill ' + parser
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            Execute(stop_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)
        Logger.info('Done stopping parser topologies')

    def stop_rest_application(self):
        """
        Stop the REST application
        """
        Logger.info('Stopping REST application')

        # Get the pid associated with the service
        pid_file = format("{metron_rest_pid_dir}/{metron_rest_pid}")
        pid = get_user_call_output.get_user_call_output(format("cat {pid_file}"), user=self.__params.metron_user, is_checked_call=False)[1]
        process_id_exists_command = format("ls {pid_file} >/dev/null 2>&1 && ps -p {pid} >/dev/null 2>&1")

        if self.__params.security_enabled:
            kinit(self.__params.kinit_path_local,
            self.__params.metron_keytab_path,
            self.__params.metron_principal_name,
            execute_user=self.__params.metron_user)

        # Politely kill
        kill_cmd = ('kill', format("{pid}"))
        Execute(kill_cmd,
                sudo=True,
                not_if = format("! ({process_id_exists_command})")
                )

        # Violently kill
        hard_kill_cmd = ('kill', '-9', format("{pid}"))
        wait_time = 5
        Execute(hard_kill_cmd,
                not_if = format("! ({process_id_exists_command}) || ( sleep {wait_time} && ! ({process_id_exists_command}) )"),
                sudo=True,
                ignore_failures = True
                )

        try:
            # check if stopped the process, else fail the task
            Execute(format("! ({process_id_exists_command})"),
                tries=20,
                try_sleep=3,
                  )
        except:
            show_logs(self.__params.metron_log_dir, self.__params.metron_user)
            raise

        File(pid_file, action = "delete")
        Logger.info('Done stopping REST application')

    def start_batch_indexing_topology(self, env):
        Logger.info('Starting ' + self.__batch_indexing_topology)

        if not self.is_batch_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)

            start_cmd_template = """{0}/bin/start_hdfs_topology.sh"""
            start_cmd = start_cmd_template.format(self.__params.metron_home)
            Execute(start_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)

        else:
            Logger.info('Batch Indexing topology already running')

        Logger.info('Finished starting batch indexing topology')

    def start(self, env, upgrade_type=None):
        from params import params
        env.set_params(params)
        self.configure(env)
        commands = ProfilerCommands(params)
        if params.security_enabled:
            metron_security.kinit(params.kinit_path_local,
                                  params.metron_keytab_path,
                                  params.metron_principal_name,
                                  execute_user=params.metron_user)

        if params.security_enabled and not commands.is_hbase_acl_configured():
            commands.set_hbase_acls()
        if params.security_enabled and not commands.is_acl_configured():
            commands.init_kafka_acls()
            commands.set_acl_configured()

        commands.start_profiler_topology(env)

    def create_hbase_tables(self):
        Logger.info("Creating HBase Tables for indexing")
        if self.__params.security_enabled:
            metron_security.kinit(self.__params.kinit_path_local,
                  self.__params.hbase_keytab_path,
                  self.__params.hbase_principal_name,
                  execute_user=self.__params.hbase_user)
        cmd = "echo \"create '{0}','{1}'\" | hbase shell -n"
        add_update_cmd = cmd.format(self.__params.update_hbase_table, self.__params.update_hbase_cf)
        Execute(add_update_cmd,
                tries=3,
                try_sleep=5,
                logoutput=False,
                path='/usr/sbin:/sbin:/usr/local/bin:/bin:/usr/bin',
                user=self.__params.hbase_user
                )

        Logger.info("Done creating HBase Tables for indexing")
        self.set_hbase_configured()

    def start_random_access_indexing_topology(self, env):
        Logger.info('Starting ' + self.__random_access_indexing_topology)

        if not self.is_random_access_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            start_cmd_template = """{0}/bin/start_elasticsearch_topology.sh"""
            if self.__params.ra_indexing_writer == 'Solr':
                start_cmd_template = """{0}/bin/start_solr_topology.sh"""
            start_cmd = start_cmd_template.format(self.__params.metron_home)
            Execute(start_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)

        else:
            Logger.info('Random Access Indexing topology already running')

        Logger.info('Finished starting random access indexing topology')

    def start_profiler_topology(self, env):
        Logger.info('Starting ' + self.__profiler_topology)

        if not self.is_topology_active(env):
            if self.__params.security_enabled:
                metron_security.kinit(self.__params.kinit_path_local,
                                      self.__params.metron_keytab_path,
                                      self.__params.metron_principal_name,
                                      execute_user=self.__params.metron_user)
            start_cmd_template = """{0}/bin/start_profiler_topology.sh \
                                    -s {1} \
                                    -z {2}"""
            start_cmd = start_cmd_template.format(self.__params.metron_home,
                                                  self.__profiler_topology,
                                                  self.__params.zookeeper_quorum)
            Execute(start_cmd, user=self.__params.metron_user, tries=3, try_sleep=5, logoutput=True)
        else:
            Logger.info('Profiler topology already running')

        Logger.info('Finished starting profiler topology')

def check_hbase_table(params, table):
    """
    Validates that an HBase table exists.  An exception is raised if the table
    does not exist.
    :param params:
    :param table: The name of the HBase table.
    """
    Logger.info("Checking HBase table '{0}'".format(table))

    # if needed kinit as 'hbase'
    if params.security_enabled:
        kinit(params.kinit_path_local,
              params.hbase_keytab_path,
              params.hbase_principal_name,
              execute_user=params.hbase_user)

    template = "echo \"exists '{0}'\" | hbase shell -n | grep 'Table {1} does exist'"
    cmd = template.format(table, table)
    err_msg = "Missing HBase table; table={0}".format(table)
    execute(cmd, user=params.hbase_user, err_msg=err_msg)