def decommission():
  import params

  hdfs_user = params.hdfs_user
  conf_dir = params.hadoop_conf_dir
  user_group = params.user_group
  nn_kinit_cmd = params.nn_kinit_cmd
  
  File(params.exclude_file_path,
       content=Template("exclude_hosts_list.j2"),
       owner=hdfs_user,
       group=user_group
  )
  
  if not params.update_exclude_file_only:
    Execute(nn_kinit_cmd,
            user=hdfs_user
    )

    if params.dfs_ha_enabled:
      # due to a bug in hdfs, refreshNodes will not run on both namenodes so we
      # need to execute each command scoped to a particular namenode
      nn_refresh_cmd = format('dfsadmin -fs hdfs://{namenode_rpc} -refreshNodes')
    else:
      nn_refresh_cmd = format('dfsadmin -fs {namenode_address} -refreshNodes')
    ExecuteHadoop(nn_refresh_cmd,
                  user=hdfs_user,
                  conf_dir=conf_dir,
                  kinit_override=True,
                  bin_dir=params.hadoop_bin_dir)

def setup_java_patch():
  import params

  if params.has_ranger_admin:

    setup_java_patch = format('ambari-python-wrap {kms_home}/db_setup.py -javapatch')

    env_dict = {'RANGER_KMS_HOME':params.kms_home, 'JAVA_HOME': params.java_home}
    if params.db_flavor.lower() == 'sqla':
      env_dict = {'RANGER_KMS_HOME':params.kms_home, 'JAVA_HOME': params.java_home, 'LD_LIBRARY_PATH':params.ld_library_path}

    Execute(setup_java_patch, environment=env_dict, logoutput=True, user=params.kms_user)

    kms_lib_path = format('{kms_home}/ews/webapp/lib/')
    files = os.listdir(kms_lib_path)
    hadoop_jar_files = []

    for x in files:
      if x.startswith('hadoop-common') and x.endswith('.jar'):
        hadoop_jar_files.append(x)

    if len(hadoop_jar_files) != 0:
      for f in hadoop_jar_files:
        Execute((format('{java_home}/bin/jar'),'-uf', format('{kms_home}/ews/webapp/lib/{f}'), format('{kms_home}/ews/webapp/META-INF/services/org.apache.hadoop.crypto.key.KeyProviderFactory')),
          user=params.kms_user)

        File(format('{kms_home}/ews/webapp/lib/{f}'), owner=params.kms_user, group=params.kms_group)

def bootstrap_standby_namenode(params, use_path=False):

  bin_path = os.path.join(params.hadoop_bin_dir, '') if use_path else ""

  try:
    iterations = 50
    bootstrap_cmd = format("{bin_path}hdfs namenode -bootstrapStandby -nonInteractive")
    # Blue print based deployments start both NN in parallel and occasionally
    # the first attempt to bootstrap may fail. Depending on how it fails the
    # second attempt may not succeed (e.g. it may find the folder and decide that
    # bootstrap succeeded). The solution is to call with -force option but only
    # during initial start
    if params.command_phase == "INITIAL_START":
      bootstrap_cmd = format("{bin_path}hdfs namenode -bootstrapStandby -nonInteractive -force")
    Logger.info("Boostrapping standby namenode: %s" % (bootstrap_cmd))
    for i in range(iterations):
      Logger.info('Try %d out of %d' % (i+1, iterations))
      code, out = shell.call(bootstrap_cmd, logoutput=False, user=params.hdfs_user)
      if code == 0:
        Logger.info("Standby namenode bootstrapped successfully")
        return True
      elif code == 5:
        Logger.info("Standby namenode already bootstrapped")
        return True
      else:
        Logger.warning('Bootstrap standby namenode failed with %d error code. Will retry' % (code))
  except Exception as ex:
    Logger.error('Bootstrap standby namenode threw an exception. Reason %s' %(str(ex)))
  return False

    def configure(self, env, upgrade_type=None, config_dir=None):
        from params import params
        env.set_params(params)
        File(format("/etc/default/metron"),
             content=Template("metron.j2")
             )

        File(format("{metron_config_path}/management_ui.yml"),
             mode=0755,
             content=Template("management_ui.yml.j2"),
             owner=params.metron_user,
             group=params.metron_group
             )

        File(format("{metron_management_ui_path}/assets/app-config.json"),
             content=Template("management-ui-app-config.json.j2"),
             owner=params.metron_user,
             group=params.metron_group
             )

        Directory('/var/run/metron',
                  create_parents=False,
                  mode=0755,
                  owner=params.metron_user,
                  group=params.metron_group
                  )

        if params.metron_knox_enabled and not params.metron_ldap_enabled:
            raise Fail("Enabling Metron with Knox requires LDAP authentication.  Please set 'LDAP Enabled' to true in the Metron Security tab.")

def setup_solr_metrics_support():
    import params

    Directory(
            [
                params.solr_metrics_config_conf_dir,
                params.solr_metrics_config_pid_dir,
                params.solr_metrics_config_log_dir
            ],
            mode=0755,
            cd_access='a',
            owner=params.solr_config_user,
            group=params.solr_config_group,
            create_parents=True
    )

    File(
            format("{solr_metrics_config_conf_dir}/solr.metrics.properties"),
            content=InlineTemplate(params.solr_metrics_properties),
            owner=params.solr_config_user
    )

    File(
            format("{solr_metrics_config_conf_dir}/log4j2.xml"),
            content=Template("log4j2.xml"),
            owner=params.solr_config_user
    )

    if params.security_enabled:
        File(
            format("{solr_metrics_kerberos_jaas_config}"),
            content=Template("solr_metrics_jaas.conf.j2"),
            owner=params.solr_config_user
        )

def remove_solr_ssl_support():
    import params

    if not params.solr_cloud_mode:
        return

    code, output = call(
            format(
                    '{zk_client_prefix} -cmd get {solr_cloud_zk_directory}{clusterprops_json}'
            ),
            env={'JAVA_HOME': params.java64_home},
            timeout=60
    )

    if "NoNodeException" in output:
        return

    Execute(
            format(
                    '{zk_client_prefix} -cmd clear {solr_cloud_zk_directory}{clusterprops_json}'
            ),
            environment={'JAVA_HOME': params.java64_home},
            ignore_failures=True,
            user=params.solr_config_user
    )

def prepare_rolling_upgrade():
  """
  Perform either an upgrade or a downgrade.

  Rolling Upgrade for HDFS Namenode requires the following.
  0. Namenode must be up
  1. Leave safemode if the safemode status is not OFF
  2. Execute a rolling upgrade "prepare"
  3. Execute a rolling upgrade "query"
  """
  import params

  if not params.upgrade_direction or params.upgrade_direction not in [Direction.UPGRADE, Direction.DOWNGRADE]:
    raise Fail("Could not retrieve upgrade direction: %s" % str(params.upgrade_direction))
  Logger.info(format("Performing a(n) {params.upgrade_direction} of HDFS"))

  if params.security_enabled:
    Execute(format("{params.kinit_path_local} -kt {params.hdfs_user_keytab} {params.hdfs_principal_name}"))


  if params.upgrade_direction == Direction.UPGRADE:
    safemode_transition_successful, original_state = reach_safemode_state(params.hdfs_user, SafeMode.OFF, True)
    if not safemode_transition_successful:
      raise Fail("Could not transition to safemode state %s. Please check logs to make sure namenode is up." % str(SafeMode.OFF))

    prepare = "hdfs dfsadmin -rollingUpgrade prepare"
    query = "hdfs dfsadmin -rollingUpgrade query"
    Execute(prepare,
            user=params.hdfs_user,
            logoutput=True)
    Execute(query,
            user=params.hdfs_user,
            logoutput=True)
  elif params.upgrade_direction == Direction.DOWNGRADE:
    pass

    def load_template(self, env):
        from dashboard.dashboardindex import DashboardIndex

        import params
        env.set_params(params)

        hostname = format("{es_host}")
        port = int(format("{es_port}"))

        Logger.info("Connecting to Elasticsearch on host: %s, port: %s" % (hostname, port))
        di = DashboardIndex(host=hostname, port=port)

        # Loads Kibana Dashboard definition from disk and replaces .kibana on index
        templateFile = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dashboard', 'dashboard.p')
        if not os.path.isfile(templateFile):
            raise IOError(
                errno.ENOENT, os.strerror(errno.ENOENT), templateFile)

        Logger.info("Deleting .kibana index from Elasticsearch")

        di.es.indices.delete(index='.kibana', ignore=[400, 404])

        Logger.info("Loading .kibana index from %s" % templateFile)

        di.put(data=di.load(filespec=templateFile))

    def action_create(self):
        with Environment.get_instance_copy() as env:
            with tempfile.NamedTemporaryFile() as tmpf:
                repo_file_name = format("{repo_file_name}.list", repo_file_name=self.resource.repo_file_name)
                repo_file_path = format("{repo_dir}/{repo_file_name}", repo_dir=self.repo_dir)

                new_content = Template(
                    self.resource.repo_template,
                    package_type=self.package_type,
                    base_url=self.resource.base_url,
                    components=" ".join(self.resource.components),
                ).get_content()
                old_content = ""
                if self.resource.append_to_file and os.path.isfile(repo_file_path):
                    with open(repo_file_path) as repo_file:
                        old_content = repo_file.read() + "\n"

                File(tmpf.name, content=old_content + new_content)

                if not os.path.isfile(repo_file_path) or not filecmp.cmp(tmpf.name, repo_file_path):
                    File(repo_file_path, content=StaticFile(tmpf.name))

                    update_cmd_formatted = [format(x) for x in self.update_cmd]
                    # this is time expensive
                    retcode, out = checked_call(update_cmd_formatted, sudo=True)

                    # add public keys for new repos
                    missing_pkeys = set(re.findall(self.missing_pkey_regex, out))
                    for pkey in missing_pkeys:
                        Execute(
                            format(self.add_pkey_cmd),
                            timeout=15,  # in case we are on the host w/o internet (using localrepo), we should ignore hanging
                            ignore_failures=True,
                        )

    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 post_upgrade_restart(self, env, upgrade_type=None):
    if upgrade_type == "nonrolling":
      return

    Logger.info("Executing Stack Upgrade post-restart")
    import params
    env.set_params(params)
    zk_server_host = random.choice(params.zookeeper_hosts)
    cli_shell = format("{zk_cli_shell} -server {zk_server_host}:{client_port}")
    # Ensure that a quorum is still formed.
    unique = get_unique_id_and_date()
    create_command = format("echo 'create /{unique} mydata' | {cli_shell}")
    list_command = format("echo 'ls /' | {cli_shell}")
    delete_command = format("echo 'delete /{unique} ' | {cli_shell}")

    quorum_err_message = "Failed to establish zookeeper quorum"
    call_and_match_output(create_command, 'Created', quorum_err_message, user=params.zk_user)
    call_and_match_output(list_command, r"\[.*?" + unique + ".*?\]", quorum_err_message, user=params.zk_user)
    shell.call(delete_command, user=params.zk_user)

    if params.client_port:
      check_leader_command = format("echo stat | nc localhost {client_port} | grep Mode")
      code, out = shell.call(check_leader_command, logoutput=False)
      if code == 0 and out:
        Logger.info(out)

def setup_ranger_plugin_keystore(service_name, audit_db_is_enabled, stack_version, credential_file, xa_audit_db_password,
                                ssl_truststore_password, ssl_keystore_password, component_user, component_group, java_home):

  stack_root = Script.get_stack_root()
  service_name = str(service_name).lower()
  cred_lib_path = format('{stack_root}/{stack_version}/ranger-{service_name}-plugin/install/lib/*')
  cred_setup_prefix = (format('{stack_root}/{stack_version}/ranger-{service_name}-plugin/ranger_credential_helper.py'), '-l', cred_lib_path)

  if service_name == 'nifi':
    cred_lib_path = format('{stack_root}/{stack_version}/{service_name}/ext/ranger/install/lib/*')
    cred_setup_prefix = (format('{stack_root}/{stack_version}/{service_name}/ext/ranger/scripts/ranger_credential_helper.py'), '-l', cred_lib_path)

  if audit_db_is_enabled:
    cred_setup = cred_setup_prefix + ('-f', credential_file, '-k', 'auditDBCred', '-v', PasswordString(xa_audit_db_password), '-c', '1')
    Execute(cred_setup, environment={'JAVA_HOME': java_home}, logoutput=True, sudo=True)

  cred_setup = cred_setup_prefix + ('-f', credential_file, '-k', 'sslKeyStore', '-v', PasswordString(ssl_keystore_password), '-c', '1')
  Execute(cred_setup, environment={'JAVA_HOME': java_home}, logoutput=True, sudo=True)

  cred_setup = cred_setup_prefix + ('-f', credential_file, '-k', 'sslTrustStore', '-v', PasswordString(ssl_truststore_password), '-c', '1')
  Execute(cred_setup, environment={'JAVA_HOME': java_home}, logoutput=True, sudo=True)

  File(credential_file,
    owner = component_user,
    group = component_group,
    mode = 0640
  )

def setup_solr_cloud():
    import params

    code, output = call(
            format(
                    '{zk_client_prefix} -cmd get {solr_cloud_zk_directory}{clusterstate_json}'
            ),
            env={'JAVA_HOME': params.java64_home},
            timeout=60
    )

    if not ("NoNodeException" in output):
        Logger.info(
                format(
                        "ZK node {solr_cloud_zk_directory}{clusterstate_json} already exists, skipping ..."
                )
        )
        return

    Execute(
            format(
                    '{zk_client_prefix} -cmd makepath {solr_cloud_zk_directory}'
            ),
            environment={'JAVA_HOME': params.java64_home},
            ignore_failures=True,
            user=params.solr_config_user
    )

def kill_zkfc(zkfc_user):
  """
  There are two potential methods for failing over the namenode, especially during a Rolling Upgrade.
  Option 1. Kill zkfc on primary namenode provided that the secondary is up and has zkfc running on it.
  Option 2. Silent failover (not supported as of HDP 2.2.0.0)
  :param zkfc_user: User that started the ZKFC process.
  :return: Return True if ZKFC was killed, otherwise, false.
  """
  import params
  if params.dfs_ha_enabled:
    zkfc_pid_file = get_service_pid_file("zkfc", zkfc_user)
    if zkfc_pid_file:
      check_process = as_user(format("ls {zkfc_pid_file} > /dev/null 2>&1 && ps -p `cat {zkfc_pid_file}` > /dev/null 2>&1"), user=zkfc_user)
      code, out = shell.call(check_process)
      if code == 0:
        Logger.debug("ZKFC is running and will be killed.")
        kill_command = format("kill -15 `cat {zkfc_pid_file}`")
        Execute(kill_command,
             user=zkfc_user
        )
        File(zkfc_pid_file,
             action = "delete",
        )
        return True
  return False

    def service_check(self, env):
        import params
        env.set_params(params)

        if not os.path.isfile(params.solr_config_pid_file):
            Logger.error(format("PID file {solr_config_pid_file} does not exist"))
            exit(1)

        if not params.solr_collection_sample_create:
            Logger.info("Create sample collection unchecked, skipping ...")
            return

        if exists_collection(params.solr_collection_name):
            Logger.warning(format("Collection {solr_collection_name} already exists, skipping ..."))
            return

        if not params.solr_cloud_mode:
            Execute(
                    format(
                            '{solr_config_bin_dir}/solr create_core -c {solr_collection_name}' +
                            ' -d {solr_collection_config_dir} -p {solr_config_port} >> {solr_config_service_log_file} 2>&1'
                    ),
                    environment={'JAVA_HOME': params.java64_home},
                    user=params.solr_config_user
            )
        else:
            Execute(format(
                    '{solr_config_bin_dir}/solr create_collection -c {solr_collection_name}' +
                    ' -d {solr_collection_config_dir} -p {solr_config_port}' +
                    ' -s {solr_collection_shards} -rf {solr_collection_replicas}' +
                    ' >> {solr_config_service_log_file} 2>&1'),
                    environment={'JAVA_HOME': params.java64_home},
                    user=params.solr_config_user
            )

def is_active_namenode(hdfs_binary):
  """
  Checks if current NameNode is active. Waits up to 30 seconds. If other NameNode is active returns False.
  :return: True if current NameNode is active, False otherwise
  """
  import params

  if params.dfs_ha_enabled:
    is_active_this_namenode_cmd = as_user(format("{hdfs_binary} --config {hadoop_conf_dir} haadmin -getServiceState {namenode_id} | grep active"), params.hdfs_user, env={'PATH':params.hadoop_bin_dir})
    is_active_other_namenode_cmd = as_user(format("{hdfs_binary} --config {hadoop_conf_dir} haadmin -getServiceState {other_namenode_id} | grep active"), params.hdfs_user, env={'PATH':params.hadoop_bin_dir})

    for i in range(0, 5):
      code, out = shell.call(is_active_this_namenode_cmd) # If active NN, code will be 0
      if code == 0: # active
        return True

      code, out = shell.call(is_active_other_namenode_cmd) # If other NN is active, code will be 0
      if code == 0: # other NN is active
        return False

      if i < 4: # Do not sleep after last iteration
        time.sleep(6)

    Logger.info("Active NameNode is not found.")
    return False

  else:
    return True

def setup_hdp_install_directory():
  # This is a name of marker file.
  SELECT_ALL_PERFORMED_MARKER = "/var/lib/ambari-agent/data/hdp-select-set-all.performed"
  import params
  if params.hdp_stack_version != "" and compare_versions(params.hdp_stack_version, '2.2') >= 0:
    Execute(as_sudo(['touch', SELECT_ALL_PERFORMED_MARKER]) + ' ; ' +
                   format('{sudo} /usr/bin/hdp-select set all `ambari-python-wrap /usr/bin/hdp-select versions | grep ^{stack_version_unformatted} | tail -1`'),
            only_if=format('ls -d /usr/hdp/{stack_version_unformatted}*'),   # If any HDP version is installed
            not_if=format("test -f {SELECT_ALL_PERFORMED_MARKER}")           # Do that only once (otherwise we break rolling upgrade logic)
    )

def setup_ranger_plugin_jar_symblink(hdp_version, service_name, component_list):

  jar_files = os.listdir(format('/usr/hdp/{hdp_version}/ranger-{service_name}-plugin/lib'))

  for jar_file in jar_files:
    for component in component_list:
      Execute(('ln','-sf',format('/usr/hdp/{hdp_version}/ranger-{service_name}-plugin/lib/{jar_file}'),format('/usr/hdp/current/{component}/lib/{jar_file}')),
      not_if=format('ls /usr/hdp/current/{component}/lib/{jar_file}'),
      only_if=format('ls /usr/hdp/{hdp_version}/ranger-{service_name}-plugin/lib/{jar_file}'),
      sudo=True)

def setup_ranger_plugin(component_select_name, service_name,
                        downloaded_custom_connector, driver_curl_source,
                        driver_curl_target, java_home,
                        repo_name, plugin_repo_dict, 
                        ranger_env_properties, plugin_properties,
                        policy_user, policymgr_mgr_url,
                        plugin_enabled, component_user, component_group, api_version=None, skip_if_rangeradmin_down = True, **kwargs):
  File(downloaded_custom_connector,
      content = DownloadSource(driver_curl_source),
      mode = 0644
  )

  Execute(('cp', '--remove-destination', downloaded_custom_connector, driver_curl_target),
          path=["/bin", "/usr/bin/"],
          sudo=True
  )

  File(driver_curl_target, mode=0644)

  hdp_version = get_hdp_version(component_select_name)
  file_path = format('/usr/hdp/{hdp_version}/ranger-{service_name}-plugin/install.properties')
  
  if not os.path.isfile(file_path):
    raise Fail(format('Ranger {service_name} plugin install.properties file does not exist at {file_path}'))
  
  ModifyPropertiesFile(file_path,
    properties = plugin_properties
  )

  custom_plugin_properties = dict()
  custom_plugin_properties['CUSTOM_USER'] = component_user
  custom_plugin_properties['CUSTOM_GROUP'] = component_group
  ModifyPropertiesFile(file_path,properties = custom_plugin_properties)

  if plugin_enabled:
    cmd = (format('enable-{service_name}-plugin.sh'),)
    if api_version == 'v2' and api_version is not None:
      ranger_adm_obj = RangeradminV2(url=policymgr_mgr_url, skip_if_rangeradmin_down = skip_if_rangeradmin_down)
    else:
      ranger_adm_obj = Rangeradmin(url=policymgr_mgr_url, skip_if_rangeradmin_down = skip_if_rangeradmin_down)

    ranger_adm_obj.create_ranger_repository(service_name, repo_name, plugin_repo_dict,
                                            ranger_env_properties['ranger_admin_username'], ranger_env_properties['ranger_admin_password'], 
                                            ranger_env_properties['admin_username'], ranger_env_properties['admin_password'], 
                                            policy_user)
  else:
    cmd = (format('disable-{service_name}-plugin.sh'),)
    
  cmd_env = {'JAVA_HOME': java_home, 'PWD': format('/usr/hdp/{hdp_version}/ranger-{service_name}-plugin'), 'PATH': format('/usr/hdp/{hdp_version}/ranger-{service_name}-plugin')}
  
  Execute(cmd, 
        environment=cmd_env, 
        logoutput=True,
        sudo=True,
  )

def get_service_pid_file(name, user):
  """
  Get the pid file path that was used to start the service by the user.
  :param name: Service name
  :param user: User that started the service.
  :return: PID file path
  """
  import params
  pid_dir = format("{hadoop_pid_dir_prefix}/{user}")
  pid_file = format("{pid_dir}/hadoop-{user}-{name}.pid")
  return pid_file