YARN-7207. Cache the RM proxy server address. (Yufei Gu)
(cherry picked from commit 72d22b753a)
# Conflicts:
# hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContext.java
# hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/RMContextImpl.java
This commit is contained in:
Yufei Gu
parent
c949e8e83a
commit
da0a96968d
|
|
@ -153,4 +153,6 @@ public interface RMContext extends ApplicationMasterServiceContext {
|
|||
String getHAZookeeperConnectionState();
|
||||
|
||||
ResourceManager getResourceManager();
|
||||
|
||||
String getAppProxyUrl(Configuration conf, ApplicationId applicationId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,9 +18,13 @@
|
|||
|
||||
package org.apache.hadoop.yarn.server.resourcemanager;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.apache.hadoop.classification.InterfaceAudience.Private;
|
||||
import org.apache.hadoop.conf.Configuration;
|
||||
import org.apache.hadoop.classification.InterfaceStability.Unstable;
|
||||
|
|
@ -51,7 +55,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRen
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager;
|
||||
import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils;
|
||||
import org.apache.hadoop.yarn.util.Clock;
|
||||
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
||||
|
|
@ -69,6 +75,8 @@ import com.google.common.annotations.VisibleForTesting;
|
|||
*/
|
||||
public class RMContextImpl implements RMContext {
|
||||
|
||||
private static final Log LOG = LogFactory.getLog(RMContextImpl.class);
|
||||
private static final String UNAVAILABLE = "N/A";
|
||||
/**
|
||||
* RM service contexts which runs through out RM life span. These are created
|
||||
* once during start of RM.
|
||||
|
|
@ -81,6 +89,8 @@ public class RMContextImpl implements RMContext {
|
|||
*/
|
||||
private RMActiveServiceContext activeServiceContext;
|
||||
|
||||
private String proxyHostAndPort = null;
|
||||
|
||||
/**
|
||||
* Default constructor. To be used in conjunction with setter methods for
|
||||
* individual fields.
|
||||
|
|
@ -114,7 +124,7 @@ public class RMContextImpl implements RMContext {
|
|||
ConfigurationProvider provider = new LocalConfigurationProvider();
|
||||
setConfigurationProvider(provider);
|
||||
}
|
||||
|
||||
|
||||
@VisibleForTesting
|
||||
// helper constructor for tests
|
||||
public RMContextImpl(Dispatcher rmDispatcher,
|
||||
|
|
@ -511,7 +521,7 @@ public class RMContextImpl implements RMContext {
|
|||
public PlacementManager getQueuePlacementManager() {
|
||||
return this.activeServiceContext.getQueuePlacementManager();
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void setQueuePlacementManager(PlacementManager placementMgr) {
|
||||
this.activeServiceContext.setQueuePlacementManager(placementMgr);
|
||||
|
|
@ -538,5 +548,26 @@ public class RMContextImpl implements RMContext {
|
|||
return this.activeServiceContext.getRMAppLifetimeMonitor();
|
||||
}
|
||||
|
||||
// Note: Read java doc before adding any services over here.
|
||||
String getProxyHostAndPort(Configuration conf) {
|
||||
if (proxyHostAndPort == null) {
|
||||
proxyHostAndPort = WebAppUtils.getProxyHostAndPort(conf);
|
||||
}
|
||||
return proxyHostAndPort;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getAppProxyUrl(Configuration conf, ApplicationId applicationId)
|
||||
{
|
||||
try {
|
||||
final String scheme = WebAppUtils.getHttpSchemePrefix(conf);
|
||||
URI proxyUri = ProxyUriUtils.getUriFromAMUrl(scheme,
|
||||
getProxyHostAndPort(conf));
|
||||
URI result = ProxyUriUtils.getProxyUri(null, proxyUri, applicationId);
|
||||
return result.toASCIIString();
|
||||
} catch(URISyntaxException e) {
|
||||
LOG.warn("Could not generate default proxy tracking URL for " +
|
||||
applicationId);
|
||||
return UNAVAILABLE;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -991,7 +991,7 @@ public class ResourceManager extends CompositeService implements Recoverable {
|
|||
.withCSRFProtection(YarnConfiguration.RM_CSRF_PREFIX)
|
||||
.withXFSProtection(YarnConfiguration.RM_XFS_PREFIX)
|
||||
.at(webAppAddress);
|
||||
String proxyHostAndPort = WebAppUtils.getProxyHostAndPort(conf);
|
||||
String proxyHostAndPort = rmContext.getProxyHostAndPort(conf);
|
||||
if(WebAppUtils.getResolvedRMWebAppURLWithoutScheme(conf).
|
||||
equals(proxyHostAndPort)) {
|
||||
if (HAUtil.isHAEnabled(conf)) {
|
||||
|
|
|
|||
|
|
@ -19,8 +19,6 @@
|
|||
package org.apache.hadoop.yarn.server.resourcemanager.rmapp;
|
||||
|
||||
import java.net.InetAddress;
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
|
|
@ -100,7 +98,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;
|
||||
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppRemovedSchedulerEvent;
|
||||
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
|
||||
import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils;
|
||||
import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
|
||||
import org.apache.hadoop.yarn.state.MultipleArcTransition;
|
||||
import org.apache.hadoop.yarn.state.SingleArcTransition;
|
||||
|
|
@ -110,7 +107,6 @@ import org.apache.hadoop.yarn.util.Clock;
|
|||
import org.apache.hadoop.yarn.util.SystemClock;
|
||||
import org.apache.hadoop.yarn.util.Times;
|
||||
import org.apache.hadoop.yarn.util.resource.Resources;
|
||||
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
||||
|
|
@ -649,7 +645,7 @@ public class RMAppImpl implements RMApp, Recoverable {
|
|||
float progress = 0.0f;
|
||||
org.apache.hadoop.yarn.api.records.Token amrmToken = null;
|
||||
if (allowAccess) {
|
||||
trackingUrl = getDefaultProxyTrackingUrl();
|
||||
trackingUrl = rmContext.getAppProxyUrl(conf, applicationId);
|
||||
if (this.currentAttempt != null) {
|
||||
currentApplicationAttemptId = this.currentAttempt.getAppAttemptId();
|
||||
trackingUrl = this.currentAttempt.getTrackingUrl();
|
||||
|
|
@ -743,20 +739,6 @@ public class RMAppImpl implements RMApp, Recoverable {
|
|||
}
|
||||
}
|
||||
|
||||
private String getDefaultProxyTrackingUrl() {
|
||||
try {
|
||||
final String scheme = WebAppUtils.getHttpSchemePrefix(conf);
|
||||
String proxy = WebAppUtils.getProxyHostAndPort(conf);
|
||||
URI proxyUri = ProxyUriUtils.getUriFromAMUrl(scheme, proxy);
|
||||
URI result = ProxyUriUtils.getProxyUri(null, proxyUri, applicationId);
|
||||
return result.toASCIIString();
|
||||
} catch (URISyntaxException e) {
|
||||
LOG.warn("Could not generate default proxy tracking URL for "
|
||||
+ applicationId);
|
||||
return UNAVAILABLE;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public long getFinishTime() {
|
||||
this.readLock.lock();
|
||||
|
|
|
|||
|
|
@ -21,8 +21,6 @@ package org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt;
|
|||
import static org.apache.hadoop.yarn.util.StringHelper.pjoin;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
|
|
@ -521,7 +519,8 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
|
|||
this.readLock = lock.readLock();
|
||||
this.writeLock = lock.writeLock();
|
||||
|
||||
this.proxiedTrackingUrl = generateProxyUriWithScheme();
|
||||
this.proxiedTrackingUrl = rmContext.getAppProxyUrl(conf,
|
||||
appAttemptId.getApplicationId());
|
||||
this.stateMachine = stateMachineFactory.make(this);
|
||||
|
||||
this.attemptMetrics =
|
||||
|
|
@ -654,24 +653,6 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
|
|||
}
|
||||
}
|
||||
|
||||
private String generateProxyUriWithScheme() {
|
||||
this.readLock.lock();
|
||||
try {
|
||||
final String scheme = WebAppUtils.getHttpSchemePrefix(conf);
|
||||
String proxy = WebAppUtils.getProxyHostAndPort(conf);
|
||||
URI proxyUri = ProxyUriUtils.getUriFromAMUrl(scheme, proxy);
|
||||
URI result = ProxyUriUtils.getProxyUri(null, proxyUri,
|
||||
applicationAttemptId.getApplicationId());
|
||||
return result.toASCIIString();
|
||||
} catch (URISyntaxException e) {
|
||||
LOG.warn("Could not proxify the uri for "
|
||||
+ applicationAttemptId.getApplicationId(), e);
|
||||
return null;
|
||||
} finally {
|
||||
this.readLock.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void setTrackingUrlToRMAppPage(RMAppAttemptState stateToBeStored) {
|
||||
originalTrackingUrl = pjoin(
|
||||
WebAppUtils.getResolvedRMWebAppURLWithScheme(conf),
|
||||
|
|
|
|||
|
|
@ -35,8 +35,6 @@ import static org.mockito.Mockito.times;
|
|||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
import java.net.URI;
|
||||
import java.net.URISyntaxException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
|
|
@ -113,7 +111,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManag
|
|||
import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
|
||||
import org.apache.hadoop.yarn.server.security.MasterKeyData;
|
||||
import org.apache.hadoop.yarn.server.utils.BuilderUtils;
|
||||
import org.apache.hadoop.yarn.server.webproxy.ProxyUriUtils;
|
||||
import org.apache.hadoop.yarn.util.resource.Resources;
|
||||
import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
|
||||
import org.junit.After;
|
||||
|
|
@ -344,19 +341,10 @@ public class TestRMAppAttemptTransitions {
|
|||
((AsyncDispatcher)this.spyRMContext.getDispatcher()).stop();
|
||||
}
|
||||
|
||||
|
||||
private String getProxyUrl(RMAppAttempt appAttempt) {
|
||||
String url = null;
|
||||
final String scheme = WebAppUtils.getHttpSchemePrefix(conf);
|
||||
try {
|
||||
String proxy = WebAppUtils.getProxyHostAndPort(conf);
|
||||
URI proxyUri = ProxyUriUtils.getUriFromAMUrl(scheme, proxy);
|
||||
URI result = ProxyUriUtils.getProxyUri(null, proxyUri, appAttempt
|
||||
.getAppAttemptId().getApplicationId());
|
||||
url = result.toASCIIString();
|
||||
} catch (URISyntaxException ex) {
|
||||
Assert.fail();
|
||||
}
|
||||
String url = rmContext.getAppProxyUrl(conf,
|
||||
appAttempt.getAppAttemptId().getApplicationId());
|
||||
Assert.assertNotEquals("N/A", url);
|
||||
return url;
|
||||
}
|
||||
|
||||
|
|
@ -552,7 +540,7 @@ public class TestRMAppAttemptTransitions {
|
|||
if (unmanagedAM) {
|
||||
verifyUrl(trackingUrl, applicationAttempt.getTrackingUrl());
|
||||
} else {
|
||||
assertEquals(getProxyUrl(applicationAttempt),
|
||||
assertEquals(getProxyUrl(applicationAttempt),
|
||||
applicationAttempt.getTrackingUrl());
|
||||
}
|
||||
// TODO - need to add more checks relevant to this state
|
||||
|
|
|
|||
Loading…
Reference in New Issue