<commit_before>"""
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

"""
import os.path


from resource_management.core import shell
from resource_management.core.source import Template
from resource_management.core.resources.system import File, Execute, Directory
from resource_management.core.resources.service import Service
from resource_management.libraries.functions.format import format
from resource_management.libraries.functions.check_process_status import check_process_status
from resource_management.libraries.resources.execute_hadoop import ExecuteHadoop
from ambari_commons import OSCheck, OSConst
from ambari_commons.os_family_impl import OsFamilyImpl, OsFamilyFuncImpl

if OSCheck.is_windows_family():
  from resource_management.libraries.functions.windows_service_utils import check_windows_service_status

from resource_management.core.shell import as_user
from resource_management.core.exceptions import Fail
from resource_management.core.logger import Logger

from utils import service, safe_zkfc_op
from setup_ranger_hdfs import setup_ranger_hdfs

@OsFamilyFuncImpl(os_family=OsFamilyImpl.DEFAULT)
def namenode(action=None, do_format=True, rolling_restart=False, env=None):
  if action == "configure":
    import params
    #we need this directory to be present before any action(HA manual steps for
    #additional namenode)
    create_name_dirs(params.dfs_name_dir)
  elif action == "start":
    setup_ranger_hdfs(rolling_upgrade = rolling_restart)
    import params
    if do_format:
      format_namenode()
      pass

    File(params.exclude_file_path,
         content=Template("exclude_hosts_list.j2"),
         owner=params.hdfs_user,
         group=params.user_group
    )

    if params.dfs_ha_enabled and \
      params.dfs_ha_namenode_standby is not None and \
      params.hostname == params.dfs_ha_namenode_standby:
        # if the current host is the standby NameNode in an HA deployment
        # run the bootstrap command, to start the NameNode in standby mode
        # this requires that the active NameNode is already up and running,
        # so this execute should be re-tried upon failure, up to a timeout
        success = bootstrap_standby_namenode(params)
        if not success:
          raise Fail("Could not bootstrap standby namenode")

    options = "-rollingUpgrade started" if rolling_restart else ""

    if rolling_restart:
      # Must start Zookeeper Failover Controller if it exists on this host because it could have been killed in order to initiate the failover.
      safe_zkfc_op(action, env)

    service(
      action="start",
      name="namenode",
      user=params.hdfs_user,
      options=options,
      create_pid_dir=True,
      create_log_dir=True
    )

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

    is_namenode_safe_mode_off = format("hadoop dfsadmin -fs {namenode_address} -safemode get | grep 'Safe mode is OFF'")
    is_active_namenode_cmd = as_user(format("hdfs --config {hadoop_conf_dir} haadmin -getServiceState {namenode_id} | grep active"), params.hdfs_user, env={'PATH':params.hadoop_bin_dir})

    # During normal operations, if HA is enabled and it is in standby, then stay in current state, otherwise, leave safemode.
    # During Rolling Upgrade, both namenodes must leave safemode.

    # ___Scenario_________|_Expected safemode state___|_Wait for safemode OFF____|
    # 1 (HA and active)   | ON -> OFF                 | Yes                      |
    # 2 (HA and standby)  | no change (yes during RU) | no check (yes during RU) |
    # 3 (no-HA)           | ON -> OFF                 | Yes                      |
    leave_safe_mode = False
    msg = ""
    if params.dfs_ha_enabled:
      code, out = shell.call(is_active_namenode_cmd, logoutput=True) # If active NN, code will be 0
      if code == 0: # active
        leave_safe_mode = True
        msg = "Must leave safemode since High Availability is enabled and this is the Active NameNode."
      elif rolling_restart:
        leave_safe_mode = True
        msg = "Must leave safemode since High Availability is enabled during a Rolling Upgrade"
    else:
      msg = "Must leave safemode since High Availability is not enabled."
      leave_safe_mode = True

    if not msg:
      msg = "Will remain in the current safemode state."
    Logger.info(msg)

    if leave_safe_mode:
      # First check if Namenode is not in 'safemode OFF' (equivalent to safemode ON), if so, then leave it
      Logger.info("Checking the NameNode safemode status since may need to transition from ON to OFF.")
      code, out = shell.call(is_namenode_safe_mode_off, user=params.hdfs_user)
      if code != 0:
        Logger.info("Will need to leave safemode, state should be OFF.")
        leave_safe_mode_cmd = format("hdfs --config {hadoop_conf_dir} dfsadmin -fs {namenode_address} -safemode leave")
        Execute(leave_safe_mode_cmd,
                tries=10,
                try_sleep=10,
                user=params.hdfs_user,
                path=[params.hadoop_bin_dir],
        )

        Logger.info("Checking if safemode state is now OFF.")
        # Verify if Namenode should be in safemode OFF
        Execute(is_namenode_safe_mode_off,
                tries=40,
                try_sleep=10,
                path=[params.hadoop_bin_dir],
                user=params.hdfs_user
        )

      # Always run on this non-HA, or active NameNode during HA.
      create_hdfs_directories(is_active_namenode_cmd)
  elif action == "stop":
    import params
    service(
      action="stop", name="namenode", 
      user=params.hdfs_user
    )
  elif action == "status":
    import status_params
    check_process_status(status_params.namenode_pid_file)
  elif action == "decommission":
    decommission()

@OsFamilyFuncImpl(os_family=OSConst.WINSRV_FAMILY)
def namenode(action=None, do_format=True, rolling_restart=False, env=None):
  if action == "configure":
    pass
  elif action == "start":
    import params
    #TODO: Replace with format_namenode()
    namenode_format_marker = os.path.join(params.hadoop_conf_dir,"NN_FORMATTED")
    if not os.path.exists(namenode_format_marker):<commit_msg>AMBARI-11704. Mahout service check fails due to bad folder permission for /user/ambari-qa (afernandez via smohanty)<commit_after>"""
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

"""
import os.path


from resource_management.core import shell
from resource_management.core.source import Template
from resource_management.core.resources.system import File, Execute, Directory
from resource_management.core.resources.service import Service
from resource_management.libraries.functions.format import format
from resource_management.libraries.functions.check_process_status import check_process_status
from resource_management.libraries.resources.execute_hadoop import ExecuteHadoop
from ambari_commons import OSCheck, OSConst
from ambari_commons.os_family_impl import OsFamilyImpl, OsFamilyFuncImpl

if OSCheck.is_windows_family():
  from resource_management.libraries.functions.windows_service_utils import check_windows_service_status

from resource_management.core.shell import as_user
from resource_management.core.exceptions import Fail
from resource_management.core.logger import Logger

from utils import service, safe_zkfc_op
from setup_ranger_hdfs import setup_ranger_hdfs

@OsFamilyFuncImpl(os_family=OsFamilyImpl.DEFAULT)
def namenode(action=None, do_format=True, rolling_restart=False, env=None):
  if action == "configure":
    import params
    #we need this directory to be present before any action(HA manual steps for
    #additional namenode)
    create_name_dirs(params.dfs_name_dir)
  elif action == "start":
    setup_ranger_hdfs(rolling_upgrade = rolling_restart)
    import params
    if do_format:
      format_namenode()
      pass

    File(params.exclude_file_path,
         content=Template("exclude_hosts_list.j2"),
         owner=params.hdfs_user,
         group=params.user_group
    )

    if params.dfs_ha_enabled and \
      params.dfs_ha_namenode_standby is not None and \
      params.hostname == params.dfs_ha_namenode_standby:
        # if the current host is the standby NameNode in an HA deployment
        # run the bootstrap command, to start the NameNode in standby mode
        # this requires that the active NameNode is already up and running,
        # so this execute should be re-tried upon failure, up to a timeout
        success = bootstrap_standby_namenode(params)
        if not success:
          raise Fail("Could not bootstrap standby namenode")

    options = "-rollingUpgrade started" if rolling_restart else ""

    if rolling_restart:
      # Must start Zookeeper Failover Controller if it exists on this host because it could have been killed in order to initiate the failover.
      safe_zkfc_op(action, env)

    service(
      action="start",
      name="namenode",
      user=params.hdfs_user,
      options=options,
      create_pid_dir=True,
      create_log_dir=True
    )

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

    is_namenode_safe_mode_off = format("hadoop dfsadmin -fs {namenode_address} -safemode get | grep 'Safe mode is OFF'")
    is_active_namenode_cmd = as_user(format("hdfs --config {hadoop_conf_dir} haadmin -getServiceState {namenode_id} | grep active"), params.hdfs_user, env={'PATH':params.hadoop_bin_dir})

    # During normal operations, if HA is enabled and it is in standby, then stay in current state, otherwise, leave safemode.
    # During Rolling Upgrade, both namenodes must leave safemode.

    # ___Scenario_________|_Expected safemode state___|_Wait for safemode OFF____|
    # 1 (HA and active)   | ON -> OFF                 | Yes                      |
    # 2 (HA and standby)  | no change (yes during RU) | no check (yes during RU) |
    # 3 (no-HA)           | ON -> OFF                 | Yes                      |
    leave_safe_mode = False
    msg = ""
    if params.dfs_ha_enabled:
      code, out = shell.call(is_active_namenode_cmd, logoutput=True) # If active NN, code will be 0
      if code == 0: # active
        leave_safe_mode = True
        msg = "Must leave safemode since High Availability is enabled and this is the Active NameNode."
      elif rolling_restart:
        leave_safe_mode = True
        msg = "Must leave safemode since High Availability is enabled during a Rolling Upgrade"
    else:
      msg = "Must leave safemode since High Availability is not enabled."
      leave_safe_mode = True

    if not msg:
      msg = "Will remain in the current safemode state."
    Logger.info(msg)

    if leave_safe_mode:
      # First check if Namenode is not in 'safemode OFF' (equivalent to safemode ON), if so, then leave it
      Logger.info("Checking the NameNode safemode status since may need to transition from ON to OFF.")
      code, out = shell.call(is_namenode_safe_mode_off, user=params.hdfs_user)
      if code != 0:
        Logger.info("Will need to leave safemode, state should be OFF.")
        leave_safe_mode_cmd = format("hdfs --config {hadoop_conf_dir} dfsadmin -fs {namenode_address} -safemode leave")
        Execute(leave_safe_mode_cmd,
                tries=10,
                try_sleep=10,
                user=params.hdfs_user,
                path=[params.hadoop_bin_dir],
        )

        Logger.info("Checking if safemode state is now OFF.")
        # Verify if Namenode should be in safemode OFF
        Execute(is_namenode_safe_mode_off,
                tries=40,
                try_sleep=10,
                path=[params.hadoop_bin_dir],
                user=params.hdfs_user
        )
        pass
      pass

    # Always run this on non-HA, or active NameNode during HA.
    create_hdfs_directories(is_active_namenode_cmd)

  elif action == "stop":
    import params
    service(
      action="stop", name="namenode", 
      user=params.hdfs_user
    )
  elif action == "status":
    import status_params
    check_process_status(status_params.namenode_pid_file)
  elif action == "decommission":
    decommission()

@OsFamilyFuncImpl(os_family=OSConst.WINSRV_FAMILY)
def namenode(action=None, do_format=True, rolling_restart=False, env=None):
  if action == "configure":
    pass
  elif action == "start":
    import params
    #TODO: Replace with format_namenode()
    namenode_format_marker = os.path.join(params.hadoop_conf_dir,"NN_FORMATTED")
    if not os.path.exists(namenode_format_marker):
{'fraction_non_alphanumeric': 0.0694098383996583, 'fraction_numerical': 0.003061151847369545, 'mean_word_length': 3.355968992248062, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2898249', 'n_tokens_mistral': 4160, 'n_tokens_neox': 4038, 'n_words': 1293}