ambari-common/src/main/python/resource_management/libraries/functions/namenode_ha_utils.py - ambari - Git at Google

 #!/usr/bin/env python

 '''
 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.
 '''
 from resource_management.libraries.functions.is_empty import is_empty
 from resource_management.libraries.functions.format import format
 from resource_management.libraries.functions.jmx import get_value_from_jmx
 from resource_management.core.base import Fail
 from resource_management.core import shell
 from resource_management.core.logger import Logger
 from resource_management.libraries.functions.decorator import retry
 from resource_management.libraries.functions.hdfs_utils import is_https_enabled_in_hdfs


 __all__ = ["get_namenode_states", "get_active_namenode",
            "get_property_for_active_namenode", "get_nameservice"]

 HDFS_NN_STATE_ACTIVE = 'active'
 HDFS_NN_STATE_STANDBY = 'standby'

 NAMENODE_HTTP_NON_HA = 'dfs.namenode.http-address'
 NAMENODE_HTTPS_NON_HA = 'dfs.namenode.https-address'
 DFS_HTTP_POLICY = "dfs.http.policy"

 NAMENODE_HTTP_FRAGMENT = 'dfs.namenode.http-address.{0}.{1}'
 NAMENODE_HTTPS_FRAGMENT = 'dfs.namenode.https-address.{0}.{1}'
 NAMENODE_RPC_FRAGMENT = 'dfs.namenode.rpc-address.{0}.{1}'
 NAMENODE_RPC_NON_HA = 'dfs.namenode.rpc-address'
 JMX_URI_FRAGMENT = "{0}://{1}/jmx?qry=Hadoop:service=NameNode,name=FSNamesystem"
 INADDR_ANY = '0.0.0.0'

 def get_namenode_states(hdfs_site, security_enabled, run_user, times=10, sleep_time=1, backoff_factor=2):
   """
   return format [('nn1', 'hdfs://hostname1:port1'), ('nn2', 'hdfs://hostname2:port2')] , [....], [....]
   """
   @retry(times=times, sleep_time=sleep_time, backoff_factor=backoff_factor, err_class=Fail)
   def doRetries(hdfs_site, security_enabled, run_user):
     doRetries.attempt += 1
     active_namenodes, standby_namenodes, unknown_namenodes = get_namenode_states_noretries(hdfs_site, security_enabled, run_user, doRetries.attempt == times)
     Logger.info(
       "NameNode HA states: active_namenodes = {0}, standby_namenodes = {1}, unknown_namenodes = {2}".format(
         active_namenodes, standby_namenodes, unknown_namenodes))
     if active_namenodes:
       return active_namenodes, standby_namenodes, unknown_namenodes
     elif doRetries.attempt == times:
       Logger.warning("No active NameNode was found after {0} retries. Will return current NameNode HA states".format(times))
       return active_namenodes, standby_namenodes, unknown_namenodes
     raise Fail('No active NameNode was found.')

   doRetries.attempt = 0
   return doRetries(hdfs_site, security_enabled, run_user)

 def get_namenode_states_noretries(hdfs_site, security_enabled, run_user, last_retry=True):
   """
   return format [('nn1', 'hdfs://hostname1:port1'), ('nn2', 'hdfs://hostname2:port2')] , [....], [....]
   """
   active_namenodes = []
   standby_namenodes = []
   unknown_namenodes = []

   name_service = get_nameservice(hdfs_site)
   nn_unique_ids_key = 'dfs.ha.namenodes.' + name_service

   # now we have something like 'nn1,nn2,nn3,nn4'
   # turn it into dfs.namenode.[property].[dfs.nameservices].[nn_unique_id]
   # ie dfs.namenode.http-address.hacluster.nn1
   nn_unique_ids = hdfs_site[nn_unique_ids_key].split(',')
   for nn_unique_id in nn_unique_ids:
     is_https_enabled = is_https_enabled_in_hdfs(hdfs_site['dfs.http.policy'], hdfs_site['dfs.https.enable'])

     rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,nn_unique_id)
     if not is_https_enabled:
       key = NAMENODE_HTTP_FRAGMENT.format(name_service,nn_unique_id)
       protocol = "http"
     else:
       key = NAMENODE_HTTPS_FRAGMENT.format(name_service,nn_unique_id)
       protocol = "https"

     if key in hdfs_site:
       # use str() to ensure that unicode strings do not have the u' in them
       value = str(hdfs_site[key])
       if INADDR_ANY in value and rpc_key in hdfs_site:
         rpc_value = str(hdfs_site[rpc_key])
         if INADDR_ANY not in rpc_value:
           rpc_host = rpc_value.split(":")[0]
           value = value.replace(INADDR_ANY, rpc_host)

       jmx_uri = JMX_URI_FRAGMENT.format(protocol, value)

       state = get_value_from_jmx(jmx_uri, 'tag.HAState', security_enabled, run_user, is_https_enabled, last_retry)
       # If JMX parsing failed
       if not state:
         check_service_cmd = "hdfs haadmin -ns {0} -getServiceState {1}".format(get_nameservice(hdfs_site), nn_unique_id)
         code, out = shell.call(check_service_cmd, logoutput=True, user=run_user)
         if code == 0 and out:
           if HDFS_NN_STATE_STANDBY in out:
             state = HDFS_NN_STATE_STANDBY
           elif HDFS_NN_STATE_ACTIVE in out:
             state = HDFS_NN_STATE_ACTIVE

       if state == HDFS_NN_STATE_ACTIVE:
         active_namenodes.append((nn_unique_id, value))
       elif state == HDFS_NN_STATE_STANDBY:
         standby_namenodes.append((nn_unique_id, value))
       else:
         unknown_namenodes.append((nn_unique_id, value))

   return active_namenodes, standby_namenodes, unknown_namenodes

 def is_ha_enabled(hdfs_site):
   dfs_ha_nameservices = get_nameservice(hdfs_site)

   if not dfs_ha_nameservices or is_empty(dfs_ha_nameservices):
     return False

   dfs_ha_namenode_ids = hdfs_site[format("dfs.ha.namenodes.{dfs_ha_nameservices}")]

   if not is_empty(dfs_ha_namenode_ids):
     dfs_ha_namemodes_ids_list = dfs_ha_namenode_ids.split(",")
     dfs_ha_namenode_ids_array_len = len(dfs_ha_namemodes_ids_list)
     if dfs_ha_namenode_ids_array_len > 1:
       return True

   return False

 def get_active_namenode(hdfs_site, security_enabled, run_user):
   """
   return format is nn_unique_id and it's address ('nn1', 'hdfs://hostname1:port1')
   """
   active_namenodes = get_namenode_states(hdfs_site, security_enabled, run_user)[0]
   if active_namenodes:
     return active_namenodes[0]

   raise Fail('No active NameNode was found.')

 def get_property_for_active_namenode(hdfs_site, property_name, security_enabled, run_user):
   """
   For dfs.namenode.rpc-address:
     - In non-ha mode it will return hdfs_site[dfs.namenode.rpc-address]
     - In ha-mode it will return hdfs_site[dfs.namenode.rpc-address.nnha.nn2], where nnha is the name of HA, and nn2 is id of active NN
   """
   value = None
   rpc_key = None
   if is_ha_enabled(hdfs_site):
     name_service = get_nameservice(hdfs_site)
     active_namenodes = get_namenode_states(hdfs_site, security_enabled, run_user)[0]

     if not len(active_namenodes):
       raise Fail("There is no active namenodes.")

     active_namenode_id = active_namenodes[0][0]
     value = hdfs_site[format("{property_name}.{name_service}.{active_namenode_id}")]
     rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,active_namenode_id)
   else:
     value = hdfs_site[property_name]
     rpc_key = NAMENODE_RPC_NON_HA

   if INADDR_ANY in value and rpc_key in hdfs_site:
     rpc_value = str(hdfs_site[rpc_key])
     if INADDR_ANY not in rpc_value:
       rpc_host = rpc_value.split(":")[0]
       value = value.replace(INADDR_ANY, rpc_host)

   return value

 def get_all_namenode_addresses(hdfs_site):
   """
   - In non-ha mode it will return list of hdfs_site[dfs.namenode.http[s]-address]
   - In ha-mode it will return list of hdfs_site[dfs.namenode.http-address.NS.Uid], where NS is the name of HA, and Uid is id of NameNode
   """
   nn_addresses = []
   http_policy = 'HTTP_ONLY'

   if DFS_HTTP_POLICY in hdfs_site:
     http_policy = hdfs_site[DFS_HTTP_POLICY]

   if is_ha_enabled(hdfs_site):
     name_service = get_nameservice(hdfs_site)
     nn_unique_ids_key = 'dfs.ha.namenodes.' + name_service
     nn_unique_ids = hdfs_site[nn_unique_ids_key].split(',')
     for nn_unique_id in nn_unique_ids:
       rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,nn_unique_id)
       if http_policy == 'HTTPS_ONLY':
         key = NAMENODE_HTTPS_FRAGMENT.format(name_service,nn_unique_id)
       else:
         key = NAMENODE_HTTP_FRAGMENT.format(name_service,nn_unique_id)
       if key in hdfs_site:
         # use str() to ensure that unicode strings do not have the u' in them
         value = str(hdfs_site[key])
         if INADDR_ANY in value and rpc_key in hdfs_site:
           rpc_value = str(hdfs_site[rpc_key])
           if INADDR_ANY not in rpc_value:
             rpc_host = rpc_value.split(":")[0]
             value = value.replace(INADDR_ANY, rpc_host)

         if not value in nn_addresses:
           nn_addresses.append(value)
   else:
     if http_policy == 'HTTPS_ONLY':
       if NAMENODE_HTTPS_NON_HA in hdfs_site:
         nn_addresses.append(hdfs_site[NAMENODE_HTTPS_NON_HA])
     else:
       if NAMENODE_HTTP_NON_HA in hdfs_site:
         nn_addresses.append(hdfs_site[NAMENODE_HTTP_NON_HA])

   return nn_addresses

 def get_nameservice(hdfs_site):
   """
   Multiple nameservices can be configured for example to support seamless distcp
   between two HA clusters. The nameservices are defined as a comma separated
   list in hdfs_site['dfs.nameservices']. The parameter
   hdfs['dfs.internal.nameservices'] was introduced in Hadoop 2.6 to denote the
   nameservice for the current cluster (HDFS-6376).

   This method uses hdfs['dfs.internal.nameservices'] to get the current
   nameservice, if that parameter is not available it tries to splits the value
   in hdfs_site['dfs.nameservices'] returning the first string or what is
   contained in hdfs_site['dfs.namenode.shared.edits.dir'].

   By default hdfs_site['dfs.nameservices'] is returned.
   :param hdfs_site:
   :return: string or empty
   """
   name_service = hdfs_site.get('dfs.internal.nameservices', None)
   if not name_service:
     import re
     name_service = hdfs_site.get('dfs.nameservices', None)
     if name_service:
       for ns in name_service.split(","):
         if 'dfs.namenode.shared.edits.dir' in hdfs_site and re.match(r'.*%s$' % ns, hdfs_site['dfs.namenode.shared.edits.dir']): # better would be core_site['fs.defaultFS'] but it's not available
           return ns
       return name_service.split(",")[0] # default to return the first nameservice
   return name_service
	#!/usr/bin/env python

	'''
	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.
	'''
	from resource_management.libraries.functions.is_empty import is_empty
	from resource_management.libraries.functions.format import format
	from resource_management.libraries.functions.jmx import get_value_from_jmx
	from resource_management.core.base import Fail
	from resource_management.core import shell
	from resource_management.core.logger import Logger
	from resource_management.libraries.functions.decorator import retry
	from resource_management.libraries.functions.hdfs_utils import is_https_enabled_in_hdfs


	__all__ = ["get_namenode_states", "get_active_namenode",
	"get_property_for_active_namenode", "get_nameservice"]

	HDFS_NN_STATE_ACTIVE = 'active'
	HDFS_NN_STATE_STANDBY = 'standby'

	NAMENODE_HTTP_NON_HA = 'dfs.namenode.http-address'
	NAMENODE_HTTPS_NON_HA = 'dfs.namenode.https-address'
	DFS_HTTP_POLICY = "dfs.http.policy"

	NAMENODE_HTTP_FRAGMENT = 'dfs.namenode.http-address.{0}.{1}'
	NAMENODE_HTTPS_FRAGMENT = 'dfs.namenode.https-address.{0}.{1}'
	NAMENODE_RPC_FRAGMENT = 'dfs.namenode.rpc-address.{0}.{1}'
	NAMENODE_RPC_NON_HA = 'dfs.namenode.rpc-address'
	JMX_URI_FRAGMENT = "{0}://{1}/jmx?qry=Hadoop:service=NameNode,name=FSNamesystem"
	INADDR_ANY = '0.0.0.0'

	def get_namenode_states(hdfs_site, security_enabled, run_user, times=10, sleep_time=1, backoff_factor=2):
	"""
	return format [('nn1', 'hdfs://hostname1:port1'), ('nn2', 'hdfs://hostname2:port2')] , [....], [....]
	"""
	@retry(times=times, sleep_time=sleep_time, backoff_factor=backoff_factor, err_class=Fail)
	def doRetries(hdfs_site, security_enabled, run_user):
	doRetries.attempt += 1
	active_namenodes, standby_namenodes, unknown_namenodes = get_namenode_states_noretries(hdfs_site, security_enabled, run_user, doRetries.attempt == times)
	Logger.info(
	"NameNode HA states: active_namenodes = {0}, standby_namenodes = {1}, unknown_namenodes = {2}".format(
	active_namenodes, standby_namenodes, unknown_namenodes))
	if active_namenodes:
	return active_namenodes, standby_namenodes, unknown_namenodes
	elif doRetries.attempt == times:
	Logger.warning("No active NameNode was found after {0} retries. Will return current NameNode HA states".format(times))
	return active_namenodes, standby_namenodes, unknown_namenodes
	raise Fail('No active NameNode was found.')

	doRetries.attempt = 0
	return doRetries(hdfs_site, security_enabled, run_user)

	def get_namenode_states_noretries(hdfs_site, security_enabled, run_user, last_retry=True):
	"""
	return format [('nn1', 'hdfs://hostname1:port1'), ('nn2', 'hdfs://hostname2:port2')] , [....], [....]
	"""
	active_namenodes = []
	standby_namenodes = []
	unknown_namenodes = []

	name_service = get_nameservice(hdfs_site)
	nn_unique_ids_key = 'dfs.ha.namenodes.' + name_service

	# now we have something like 'nn1,nn2,nn3,nn4'
	# turn it into dfs.namenode.[property].[dfs.nameservices].[nn_unique_id]
	# ie dfs.namenode.http-address.hacluster.nn1
	nn_unique_ids = hdfs_site[nn_unique_ids_key].split(',')
	for nn_unique_id in nn_unique_ids:
	is_https_enabled = is_https_enabled_in_hdfs(hdfs_site['dfs.http.policy'], hdfs_site['dfs.https.enable'])

	rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,nn_unique_id)
	if not is_https_enabled:
	key = NAMENODE_HTTP_FRAGMENT.format(name_service,nn_unique_id)
	protocol = "http"
	else:
	key = NAMENODE_HTTPS_FRAGMENT.format(name_service,nn_unique_id)
	protocol = "https"

	if key in hdfs_site:
	# use str() to ensure that unicode strings do not have the u' in them
	value = str(hdfs_site[key])
	if INADDR_ANY in value and rpc_key in hdfs_site:
	rpc_value = str(hdfs_site[rpc_key])
	if INADDR_ANY not in rpc_value:
	rpc_host = rpc_value.split(":")[0]
	value = value.replace(INADDR_ANY, rpc_host)

	jmx_uri = JMX_URI_FRAGMENT.format(protocol, value)

	state = get_value_from_jmx(jmx_uri, 'tag.HAState', security_enabled, run_user, is_https_enabled, last_retry)
	# If JMX parsing failed
	if not state:
	check_service_cmd = "hdfs haadmin -ns {0} -getServiceState {1}".format(get_nameservice(hdfs_site), nn_unique_id)
	code, out = shell.call(check_service_cmd, logoutput=True, user=run_user)
	if code == 0 and out:
	if HDFS_NN_STATE_STANDBY in out:
	state = HDFS_NN_STATE_STANDBY
	elif HDFS_NN_STATE_ACTIVE in out:
	state = HDFS_NN_STATE_ACTIVE

	if state == HDFS_NN_STATE_ACTIVE:
	active_namenodes.append((nn_unique_id, value))
	elif state == HDFS_NN_STATE_STANDBY:
	standby_namenodes.append((nn_unique_id, value))
	else:
	unknown_namenodes.append((nn_unique_id, value))

	return active_namenodes, standby_namenodes, unknown_namenodes

	def is_ha_enabled(hdfs_site):
	dfs_ha_nameservices = get_nameservice(hdfs_site)

	if not dfs_ha_nameservices or is_empty(dfs_ha_nameservices):
	return False

	dfs_ha_namenode_ids = hdfs_site[format("dfs.ha.namenodes.{dfs_ha_nameservices}")]

	if not is_empty(dfs_ha_namenode_ids):
	dfs_ha_namemodes_ids_list = dfs_ha_namenode_ids.split(",")
	dfs_ha_namenode_ids_array_len = len(dfs_ha_namemodes_ids_list)
	if dfs_ha_namenode_ids_array_len > 1:
	return True

	return False

	def get_active_namenode(hdfs_site, security_enabled, run_user):
	"""
	return format is nn_unique_id and it's address ('nn1', 'hdfs://hostname1:port1')
	"""
	active_namenodes = get_namenode_states(hdfs_site, security_enabled, run_user)[0]
	if active_namenodes:
	return active_namenodes[0]

	raise Fail('No active NameNode was found.')

	def get_property_for_active_namenode(hdfs_site, property_name, security_enabled, run_user):
	"""
	For dfs.namenode.rpc-address:
	- In non-ha mode it will return hdfs_site[dfs.namenode.rpc-address]
	- In ha-mode it will return hdfs_site[dfs.namenode.rpc-address.nnha.nn2], where nnha is the name of HA, and nn2 is id of active NN
	"""
	value = None
	rpc_key = None
	if is_ha_enabled(hdfs_site):
	name_service = get_nameservice(hdfs_site)
	active_namenodes = get_namenode_states(hdfs_site, security_enabled, run_user)[0]

	if not len(active_namenodes):
	raise Fail("There is no active namenodes.")

	active_namenode_id = active_namenodes[0][0]
	value = hdfs_site[format("{property_name}.{name_service}.{active_namenode_id}")]
	rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,active_namenode_id)
	else:
	value = hdfs_site[property_name]
	rpc_key = NAMENODE_RPC_NON_HA

	if INADDR_ANY in value and rpc_key in hdfs_site:
	rpc_value = str(hdfs_site[rpc_key])
	if INADDR_ANY not in rpc_value:
	rpc_host = rpc_value.split(":")[0]
	value = value.replace(INADDR_ANY, rpc_host)

	return value

	def get_all_namenode_addresses(hdfs_site):
	"""
	- In non-ha mode it will return list of hdfs_site[dfs.namenode.http[s]-address]
	- In ha-mode it will return list of hdfs_site[dfs.namenode.http-address.NS.Uid], where NS is the name of HA, and Uid is id of NameNode
	"""
	nn_addresses = []
	http_policy = 'HTTP_ONLY'

	if DFS_HTTP_POLICY in hdfs_site:
	http_policy = hdfs_site[DFS_HTTP_POLICY]

	if is_ha_enabled(hdfs_site):
	name_service = get_nameservice(hdfs_site)
	nn_unique_ids_key = 'dfs.ha.namenodes.' + name_service
	nn_unique_ids = hdfs_site[nn_unique_ids_key].split(',')
	for nn_unique_id in nn_unique_ids:
	rpc_key = NAMENODE_RPC_FRAGMENT.format(name_service,nn_unique_id)
	if http_policy == 'HTTPS_ONLY':
	key = NAMENODE_HTTPS_FRAGMENT.format(name_service,nn_unique_id)
	else:
	key = NAMENODE_HTTP_FRAGMENT.format(name_service,nn_unique_id)
	if key in hdfs_site:
	# use str() to ensure that unicode strings do not have the u' in them
	value = str(hdfs_site[key])
	if INADDR_ANY in value and rpc_key in hdfs_site:
	rpc_value = str(hdfs_site[rpc_key])
	if INADDR_ANY not in rpc_value:
	rpc_host = rpc_value.split(":")[0]
	value = value.replace(INADDR_ANY, rpc_host)

	if not value in nn_addresses:
	nn_addresses.append(value)
	else:
	if http_policy == 'HTTPS_ONLY':
	if NAMENODE_HTTPS_NON_HA in hdfs_site:
	nn_addresses.append(hdfs_site[NAMENODE_HTTPS_NON_HA])
	else:
	if NAMENODE_HTTP_NON_HA in hdfs_site:
	nn_addresses.append(hdfs_site[NAMENODE_HTTP_NON_HA])

	return nn_addresses

	def get_nameservice(hdfs_site):
	"""
	Multiple nameservices can be configured for example to support seamless distcp
	between two HA clusters. The nameservices are defined as a comma separated
	list in hdfs_site['dfs.nameservices']. The parameter
	hdfs['dfs.internal.nameservices'] was introduced in Hadoop 2.6 to denote the
	nameservice for the current cluster (HDFS-6376).

	This method uses hdfs['dfs.internal.nameservices'] to get the current
	nameservice, if that parameter is not available it tries to splits the value
	in hdfs_site['dfs.nameservices'] returning the first string or what is
	contained in hdfs_site['dfs.namenode.shared.edits.dir'].

	By default hdfs_site['dfs.nameservices'] is returned.
	:param hdfs_site:
	:return: string or empty
	"""
	name_service = hdfs_site.get('dfs.internal.nameservices', None)
	if not name_service:
	import re
	name_service = hdfs_site.get('dfs.nameservices', None)
	if name_service:
	for ns in name_service.split(","):
	if 'dfs.namenode.shared.edits.dir' in hdfs_site and re.match(r'.*%s$' % ns, hdfs_site['dfs.namenode.shared.edits.dir']): # better would be core_site['fs.defaultFS'] but it's not available
	return ns
	return name_service.split(",")[0] # default to return the first nameservice
	return name_service