[
https://issues.apache.org/jira/browse/YARN-11391?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17646460#comment-17646460
]
ASF GitHub Bot commented on YARN-11391:
---------------------------------------
lnbest0707 commented on code in PR #5213:
URL: https://github.com/apache/hadoop/pull/5213#discussion_r1046726153
##########
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/HAUtil.java:
##########
@@ -262,6 +301,179 @@ public static String getRMHAId(Configuration conf) {
return currentRMId;
}
+ /**
+ * This function resolves all RMIds with their address. For multi-A DNS
records,
+ * it will resolve all of them, and generate a new Id for each of them.
+ *
+ * @param conf Configuration
+ * @return Map key as RMId, value as its address
+ */
+ public static Map<String, InetSocketAddress> getResolvedRMIdPairs(
+ Configuration conf) {
+ boolean resolveNeeded = conf.getBoolean(
+ YarnConfiguration.RESOLVE_RM_ADDRESS_NEEDED_KEY,
+ YarnConfiguration.RESOLVE_RM_ADDRESS_NEEDED_DEFAULT);
+ boolean requireFQDN = conf.getBoolean(
+ YarnConfiguration.RESOLVE_RM_ADDRESS_TO_FQDN,
+ YarnConfiguration.RESOLVE_RM_ADDRESS_TO_FQDN_DEFAULT);
+ // In case client using DIFFERENT addresses for each service address
+ // need to categorize them first
+ Map<List<String>, List<String>> addressesConfigKeysMap = new HashMap<>();
+ Collection<String> rmIds = getOriginalRMHAIds(conf);
+ for (String configKey : RM_ADDRESS_CONFIG_KEYS) {
+ List<String> addresses = new ArrayList<>();
+ for (String rmId : rmIds) {
+ String keyToRead = addSuffix(configKey, rmId);
+ InetSocketAddress address = getInetSocketAddressFromString(
+ conf.get(keyToRead));
+ if (address != null) {
+ addresses.add(address.getHostName());
+ }
+ }
+ Collections.sort(addresses);
+ List<String> configKeysOfTheseAddresses =
addressesConfigKeysMap.get(addresses);
+ if (configKeysOfTheseAddresses == null) {
+ configKeysOfTheseAddresses = new ArrayList<>();
+ addressesConfigKeysMap.put(addresses, configKeysOfTheseAddresses);
+ }
+ configKeysOfTheseAddresses.add(configKey);
+ }
+ // We need to resolve and override by group (categorized by their input
host)
+ // But since the function is called from "getRMHAId",
+ // this function would only return value which is corresponded to
YarnConfiguration.RM_ADDRESS
+ Map<String, InetSocketAddress> ret = null;
+ for (List<String> configKeys : addressesConfigKeysMap.values()) {
+ Map<String, InetSocketAddress> res = getResolvedIdPairs(conf,
resolveNeeded, requireFQDN, getOriginalRMHAIds(conf),
+ configKeys.get(0), YarnConfiguration.RM_HA_IDS, configKeys);
+ if (configKeys.contains(YarnConfiguration.RM_ADDRESS)) {
+ ret = res;
+ }
+ }
+ return ret;
+ }
+
+ private static Map<String, InetSocketAddress> getResolvedIdPairs(
+ Configuration conf, boolean resolveNeeded, boolean requireFQDN,
Collection<String> ids,
+ String configKey, String configKeyToReplace, List<String>
listOfConfigKeysToReplace) {
+ Map<String, InetSocketAddress> idAddressPairs = new HashMap<>();
+ Map<String, String> generatedIdToOriginalId = new HashMap<>();
+ for (String id : ids) {
+ String key = addSuffix(configKey, id);
+ String addr = conf.get(key); // string with port
+ InetSocketAddress address = getInetSocketAddressFromString(addr);
+ if (address == null) {
+ continue;
+ }
+ if (resolveNeeded) {
+ if (dnr == null) {
+ setDnrByConfiguration(conf);
+ }
+ // If the address needs to be resolved, get all of the IP addresses
+ // from this address and pass them into the map
+ LOG.info("Multi-A domain name " + addr +
+ " will be resolved by " + dnr.getClass().getName());
+ int port = address.getPort();
+ String[] resolvedHostNames;
+ try {
+ resolvedHostNames = dnr.getAllResolvedHostnameByDomainName(
+ address.getHostName(), requireFQDN);
+ } catch (UnknownHostException e) {
+ LOG.warn("Exception in resolving socket address "
+ + address.getHostName(), e);
+ continue;
+ }
+ LOG.info("Resolved addresses for " + addr +
+ " is " + Arrays.toString(resolvedHostNames));
+ if (resolvedHostNames == null || resolvedHostNames.length < 1) {
+ LOG.warn("Cannot resolve from address " + address.getHostName());
+ } else {
+ // If multiple address resolved, corresponding id needs to be created
+ for (int i = 0; i < resolvedHostNames.length; i++) {
+ String generatedRMId = id + "_resolved_" + (i + 1);
+ idAddressPairs.put(generatedRMId,
+ new InetSocketAddress(resolvedHostNames[i], port));
+ generatedIdToOriginalId.put(generatedRMId, id);
+ }
+ }
+ overrideIdsInConfiguration(
+ idAddressPairs, generatedIdToOriginalId, configKeyToReplace,
+ listOfConfigKeysToReplace, conf);
+ } else {
+ idAddressPairs.put(id, address);
+ }
+ }
+ return idAddressPairs;
+ }
+
+ /**
+ * This function override all RMIds and their addresses by the input Map.
+ *
+ * @param idAddressPairs key as Id, value as its address
+ * @param generatedIdToOriginalId key as generated rmId from multi-A,
+ * value as its original input Id
+ * @param configKeyToReplace key to replace
+ * @param listOfConfigKeysToReplace list of keys to replace/add
+ * @param conf Configuration
+ */
+ synchronized static void overrideIdsInConfiguration(
+ Map<String, InetSocketAddress> idAddressPairs,
+ Map<String, String> generatedIdToOriginalId,
+ String configKeyToReplace, List<String> listOfConfigKeysToReplace,
+ Configuration conf) {
+ Collection<String> currentIds = getRMHAIds(conf);
+ Set<String> resolvedIds = new HashSet<>(idAddressPairs.keySet());
+ // override rm-ids
+ if (originalRMIDs == null) {
+ originalRMIDs = currentIds;
+ }
+ // if it is already resolved, we need to form a superset
+ resolvedIds.addAll((currentIds));
+ resolvedIds.removeAll(generatedIdToOriginalId.values());
+ conf.setStrings(configKeyToReplace,
+ resolvedIds.toArray(new String[0]));
+ // override/add address configuration entries for each rm-id
+ for (Map.Entry<String, InetSocketAddress> entry :
idAddressPairs.entrySet()) {
+ String rmId = entry.getKey();
+ String addr = entry.getValue().getHostName();
+ String originalRMId = generatedIdToOriginalId.get(rmId);
+ if (originalRMId != null) {
+ // for each required configKeys, get its port and then set it back
+ for (String configKey : listOfConfigKeysToReplace) {
+ String keyToRead = addSuffix(configKey, originalRMId);
+ InetSocketAddress originalAddress = getInetSocketAddressFromString(
+ conf.get(keyToRead));
+ if (originalAddress == null) {
+ LOG.warn("Missing configuration for key " + keyToRead);
+ continue;
+ }
+ int port = originalAddress.getPort();
+ String keyToWrite = addSuffix(configKey, rmId);
+ conf.setSocketAddr(keyToWrite, new InetSocketAddress(addr, port));
+ }
+ }
+ }
+ }
+
+ /**
+ * Helper function to create InetsocketAddress from string address.
+ *
+ * @param addr string format of address accepted by NetUtils.createSocketAddr
+ * @return InetSocketAddress of input, would return null upon any kinds of
invalid inout
+ */
+ public static InetSocketAddress getInetSocketAddressFromString(String addr) {
+ if (addr == null) {
+ return null;
+ }
+ InetSocketAddress address;
+ try {
+ address = NetUtils.createSocketAddr(addr);
+ } catch (Exception e) {
+ LOG.warn("Exception in creating socket address " + addr, e);
Review Comment:
Forgot to pick the patch for new pr. Thanks for the comments, revised.
> Support using DNS to resolve yarn services to IP addresses
> ----------------------------------------------------------
>
> Key: YARN-11391
> URL: https://issues.apache.org/jira/browse/YARN-11391
> Project: Hadoop YARN
> Issue Type: New Feature
> Reporter: Shuyi Zhang
> Priority: Major
> Labels: pull-request-available
>
> We have been using A-record (hostname) for discovering YARN control
> planes(RM, JHS) from both the server and client sides. The components
> discover each other using hostnames (DNS A-records) which are preconfigured
> in YARN configurations (mapred-site.xml and yarn-site.xml). However, the
> hard-coded address mapping is prohibitive to perform appropriate failure
> recovery and node replacement. Any host replacement, adding, or removing
> requires client integration, which is a quite painful process.
> It is essential to find a solution where the discovery configuration of
> different components in the cluster stays intact and is able to resolve with
> the latest hostname/records.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
