Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

YARN-7974. Allow updating application tracking url after registration. Contributed by Jonathan Hung

This commit is contained in:
Jonathan Hung 2018-07-30 17:41:01 -07:00
parent ee53602a81
commit 3e06a5dcea
13 changed files with 298 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/AllocateRequest.java
View File

@ -73,7 +73,21 @@ public abstract class AllocateRequest {
.releaseList(containersToBeReleased)
.resourceBlacklistRequest(resourceBlacklistRequest).build();
}
@Public
@Unstable
public static AllocateRequest newInstance(int responseID, float appProgress,
List<ResourceRequest> resourceAsk,
List<ContainerId> containersToBeReleased,
ResourceBlacklistRequest resourceBlacklistRequest,
String trackingUrl) {
return AllocateRequest.newBuilder().responseId(responseID)
.progress(appProgress).askList(resourceAsk)
.releaseList(containersToBeReleased)
.resourceBlacklistRequest(resourceBlacklistRequest)
.trackingUrl(trackingUrl).build();
}
@Public
@Unstable
public static AllocateRequest newInstance(int responseID, float appProgress,
@ -240,6 +254,22 @@ public abstract class AllocateRequest {
List<SchedulingRequest> schedulingRequests) {
}
/**
* Get the tracking url update for this heartbeat.
* @return tracking url to update this application with
*/
@Public
@Unstable
public abstract String getTrackingUrl();
/**
* Set the new tracking url for this application.
* @param trackingUrl the new tracking url
*/
@Public
@Unstable
public abstract void setTrackingUrl(String trackingUrl);
@Public
@Unstable
public static AllocateRequestBuilder newBuilder() {
@ -355,6 +385,19 @@ public abstract class AllocateRequest {
return this;
}
/**
* Set the <code>trackingUrl</code> of the request.
* @see AllocateRequest#setTrackingUrl(String)
* @param trackingUrl new tracking url
* @return {@link AllocateRequestBuilder}
*/
@Public
@Unstable
public AllocateRequestBuilder trackingUrl(String trackingUrl) {
allocateRequest.setTrackingUrl(trackingUrl);
return this;
}
/**
* Return generated {@link AllocateRequest} object.
* @return {@link AllocateRequest}
@ -365,4 +408,4 @@ public abstract class AllocateRequest {
return allocateRequest;
}
}
}
}

1
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
View File

@ -92,6 +92,7 @@ message AllocateRequestProto {
optional float progress = 5;
repeated UpdateContainerRequestProto update_requests = 7;
repeated SchedulingRequestProto scheduling_requests = 10;
optional string tracking_url = 11;
}
message NMTokenProto {

11
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/AMRMClient.java
View File

@ -804,6 +804,17 @@ public abstract class AMRMClient<T extends AMRMClient.ContainerRequest> extends
return this.timelineV2Client;
}
/**
* Update application's tracking url on next heartbeat.
*
* @param trackingUrl new tracking url for this application
*/
@Public
@InterfaceStability.Unstable
public void updateTrackingUrl(String trackingUrl) {
// Unimplemented.
}
/**
* Wait for <code>check</code> to return true for each 1000 ms.
* See also {@link #waitFor(java.util.function.Supplier, int)}

11
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/AMRMClientAsync.java
View File

@ -412,6 +412,17 @@ extends AbstractService {
public abstract void updateBlacklist(List<String> blacklistAdditions,
List<String> blacklistRemovals);
/**
* Update application's tracking url on next heartbeat.
*
* @param trackingUrl new tracking url for this application
*/
@Public
@Unstable
public void updateTrackingUrl(String trackingUrl) {
// Unimplemented.
}
/**
* Wait for <code>check</code> to return true for each 1000 ms.
* See also {@link #waitFor(java.util.function.Supplier, int)}

5
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/async/impl/AMRMClientAsyncImpl.java
View File

@ -286,6 +286,11 @@ extends AMRMClientAsync<T> {
List<String> blacklistRemovals) {
client.updateBlacklist(blacklistAdditions, blacklistRemovals);
}
@Override
public void updateTrackingUrl(String trackingUrl) {
client.updateTrackingUrl(trackingUrl);
}
private class HeartbeatThread extends Thread {
public HeartbeatThread() {

11
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/main/java/org/apache/hadoop/yarn/client/api/impl/AMRMClientImpl.java
View File

@ -99,6 +99,7 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
protected String appHostName;
protected int appHostPort;
protected String appTrackingUrl;
protected String newTrackingUrl;
protected ApplicationMasterProtocol rmClient;
protected Resource clusterAvailableResources;
@ -308,6 +309,11 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
.releaseList(releaseList).updateRequests(updateList)
.schedulingRequests(schedulingRequestList).build();
if (this.newTrackingUrl != null) {
allocateRequest.setTrackingUrl(this.newTrackingUrl);
this.appTrackingUrl = this.newTrackingUrl;
this.newTrackingUrl = null;
}
// clear blacklistAdditions and blacklistRemovals before
// unsynchronized part
blacklistAdditions.clear();
@ -1008,6 +1014,11 @@ public class AMRMClientImpl<T extends ContainerRequest> extends AMRMClient<T> {
}
}
@Override
public synchronized void updateTrackingUrl(String trackingUrl) {
this.newTrackingUrl = trackingUrl;
}
private void updateAMRMToken(Token token) throws IOException {
org.apache.hadoop.security.token.Token<AMRMTokenIdentifier> amrmToken =
new org.apache.hadoop.security.token.Token<AMRMTokenIdentifier>(token

77
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-client/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestAMRMClient.java
View File

@ -20,10 +20,12 @@ package org.apache.hadoop.yarn.client.api.impl;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
@ -79,6 +81,7 @@ import org.junit.Assume;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.eclipse.jetty.util.log.Log;
@ -1994,4 +1997,78 @@ public class TestAMRMClient extends BaseAMRMClientTest{
}
}
}
@Test(timeout = 60000)
public void testNoUpdateTrackingUrl() {
try {
AMRMClientImpl<ContainerRequest> amClient = null;
amClient = new AMRMClientImpl<>();
amClient.init(conf);
amClient.start();
amClient.registerApplicationMaster("Host", 10000, "");
assertEquals("", amClient.appTrackingUrl);
ApplicationMasterProtocol mockRM = mock(ApplicationMasterProtocol.class);
AllocateResponse mockResponse = mock(AllocateResponse.class);
when(mockRM.allocate(any(AllocateRequest.class)))
.thenReturn(mockResponse);
ApplicationMasterProtocol realRM = amClient.rmClient;
amClient.rmClient = mockRM;
// Do allocate without updated tracking url
amClient.allocate(0.1f);
ArgumentCaptor<AllocateRequest> argument =
ArgumentCaptor.forClass(AllocateRequest.class);
verify(mockRM).allocate(argument.capture());
assertNull(argument.getValue().getTrackingUrl());
amClient.rmClient = realRM;
amClient
.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, null,
null);
} catch (IOException | YarnException e) {
throw new AssertionError(
"testNoUpdateTrackingUrl unexpectedly threw exception: " + e);
}
}
@Test(timeout = 60000)
public void testUpdateTrackingUrl() {
try {
AMRMClientImpl<ContainerRequest> amClient = null;
amClient = new AMRMClientImpl<>();
amClient.init(conf);
amClient.start();
amClient.registerApplicationMaster("Host", 10000, "");
String trackingUrl = "hadoop.apache.org";
assertEquals("", amClient.appTrackingUrl);
ApplicationMasterProtocol mockRM = mock(ApplicationMasterProtocol.class);
AllocateResponse mockResponse = mock(AllocateResponse.class);
when(mockRM.allocate(any(AllocateRequest.class)))
.thenReturn(mockResponse);
ApplicationMasterProtocol realRM = amClient.rmClient;
amClient.rmClient = mockRM;
// Do allocate with updated tracking url
amClient.updateTrackingUrl(trackingUrl);
assertEquals(trackingUrl, amClient.newTrackingUrl);
assertEquals("", amClient.appTrackingUrl);
amClient.allocate(0.1f);
assertNull(amClient.newTrackingUrl);
assertEquals(trackingUrl, amClient.appTrackingUrl);
ArgumentCaptor<AllocateRequest> argument
= ArgumentCaptor.forClass(AllocateRequest.class);
verify(mockRM).allocate(argument.capture());
assertEquals(trackingUrl, argument.getValue().getTrackingUrl());
amClient.rmClient = realRM;
amClient
.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, null,
null);
} catch (IOException | YarnException e) {
throw new AssertionError(
"testUpdateTrackingUrl unexpectedly threw exception: " + e);
}
}
}

27
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateRequestPBImpl.java
View File

@ -58,6 +58,7 @@ public class AllocateRequestPBImpl extends AllocateRequest {
private List<UpdateContainerRequest> updateRequests = null;
private List<SchedulingRequest> schedulingRequests = null;
private ResourceBlacklistRequest blacklistRequest = null;
private String trackingUrl = null;
public AllocateRequestPBImpl() {
builder = AllocateRequestProto.newBuilder();
@ -111,6 +112,9 @@ public class AllocateRequestPBImpl extends AllocateRequest {
if (this.blacklistRequest != null) {
builder.setBlacklistRequest(convertToProtoFormat(this.blacklistRequest));
}
if (this.trackingUrl != null) {
builder.setTrackingUrl(this.trackingUrl);
}
}
private void mergeLocalToProto() {
@ -398,7 +402,28 @@ public class AllocateRequestPBImpl extends AllocateRequest {
this.release.add(convertFromProtoFormat(c));
}
}
@Override
public String getTrackingUrl() {
AllocateRequestProtoOrBuilder p = viaProto ? proto : builder;
if (this.trackingUrl != null) {
return this.trackingUrl;
}
if (p.hasTrackingUrl()) {
this.trackingUrl = p.getTrackingUrl();
}
return this.trackingUrl;
}
@Override
public void setTrackingUrl(String trackingUrl) {
maybeInitBuilder();
if (trackingUrl == null) {
builder.clearTrackingUrl();
}
this.trackingUrl = trackingUrl;
}
private void addReleasesToProto() {
maybeInitBuilder();
builder.clearRelease();

2
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/DefaultAMSProcessor.java
View File

@ -401,7 +401,7 @@ final class DefaultAMSProcessor implements ApplicationMasterServiceProcessor {
// Send the status update to the appAttempt.
getRmContext().getDispatcher().getEventHandler().handle(
new RMAppAttemptStatusupdateEvent(appAttemptId, request
.getProgress()));
.getProgress(), request.getTrackingUrl()));
}
@Override

20
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptImpl.java
View File

@ -1823,6 +1823,26 @@ public class RMAppAttemptImpl implements RMAppAttempt, Recoverable {
// Update progress
appAttempt.progress = statusUpdateEvent.getProgress();
// Update tracking url if changed and save it to state store
String newTrackingUrl = statusUpdateEvent.getTrackingUrl();
if (newTrackingUrl != null &&
!newTrackingUrl.equals(appAttempt.originalTrackingUrl)) {
appAttempt.originalTrackingUrl = newTrackingUrl;
ApplicationAttemptStateData attemptState = ApplicationAttemptStateData
.newInstance(appAttempt.applicationAttemptId,
appAttempt.getMasterContainer(),
appAttempt.rmContext.getStateStore()
.getCredentialsFromAppAttempt(appAttempt),
appAttempt.startTime, appAttempt.recoveredFinalState,
newTrackingUrl, appAttempt.getDiagnostics(), null,
ContainerExitStatus.INVALID, appAttempt.getFinishTime(),
appAttempt.attemptMetrics.getAggregateAppResourceUsage()
.getResourceUsageSecondsMap(),
appAttempt.attemptMetrics.getPreemptedResourceSecondsMap());
appAttempt.rmContext.getStateStore()
.updateApplicationAttemptState(attemptState);
}
// Ping to AMLivelinessMonitor
appAttempt.rmContext.getAMLivelinessMonitor().receivedPing(
statusUpdateEvent.getApplicationAttemptId());

11
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/event/RMAppAttemptStatusupdateEvent.java
View File

@ -25,15 +25,26 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptE
public class RMAppAttemptStatusupdateEvent extends RMAppAttemptEvent {
private final float progress;
private final String trackingUrl;
public RMAppAttemptStatusupdateEvent(ApplicationAttemptId appAttemptId,
float progress) {
this(appAttemptId, progress, null);
}
public RMAppAttemptStatusupdateEvent(ApplicationAttemptId appAttemptId,
float progress, String trackingUrl) {
super(appAttemptId, RMAppAttemptEventType.STATUS_UPDATE);
this.progress = progress;
this.trackingUrl = trackingUrl;
}
public float getProgress() {
return this.progress;
}
public String getTrackingUrl() {
return this.trackingUrl;
}
}

34
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestApplicationMasterService.java
View File

@ -956,4 +956,38 @@ public class TestApplicationMasterService {
fail("Cannot find RMContainer");
}
}
@Test(timeout = 300000)
public void testUpdateTrackingUrl() throws Exception {
conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class,
ResourceScheduler.class);
MockRM rm = new MockRM(conf);
rm.start();
// Register node1
MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB);
RMApp app1 = rm.submitApp(2048);
nm1.nodeHeartbeat(true);
RMAppAttempt attempt1 = app1.getCurrentAppAttempt();
MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId());
am1.registerAppAttempt();
Assert.assertEquals("N/A", rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getOriginalTrackingUrl());
AllocateRequestPBImpl allocateRequest = new AllocateRequestPBImpl();
String newTrackingUrl = "hadoop.apache.org";
allocateRequest.setTrackingUrl(newTrackingUrl);
am1.allocate(allocateRequest);
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getOriginalTrackingUrl());
// Send it again
am1.allocate(allocateRequest);
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getOriginalTrackingUrl());
rm.stop();
}
}

45
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMRestart.java
View File

@ -2698,6 +2698,51 @@ public class TestRMRestart extends ParameterizedSchedulerTestBase {
rm2.stop();
}
@Test(timeout = 20000)
public void testRMRestartAfterUpdateTrackingUrl() throws Exception {
MockRM rm = new MockRM(conf);
rm.start();
MemoryRMStateStore memStore = (MemoryRMStateStore) rm.getRMStateStore();
// Register node1
MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * 1024);
RMApp app1 = rm.submitApp(2048);
nm1.nodeHeartbeat(true);
RMAppAttempt attempt1 = app1.getCurrentAppAttempt();
MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId());
am1.registerAppAttempt();
AllocateRequestPBImpl allocateRequest = new AllocateRequestPBImpl();
String newTrackingUrl = "hadoop.apache.org";
allocateRequest.setTrackingUrl(newTrackingUrl);
am1.allocate(allocateRequest);
// Check in-memory and stored tracking url
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getOriginalTrackingUrl());
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getCurrentAppAttempt()
.getOriginalTrackingUrl());
Assert.assertEquals(newTrackingUrl, memStore.getState()
.getApplicationState().get(app1.getApplicationId())
.getAttempt(attempt1.getAppAttemptId()).getFinalTrackingUrl());
// Start new RM, should recover updated tracking url
MockRM rm2 = new MockRM(conf, memStore);
rm2.start();
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getOriginalTrackingUrl());
Assert.assertEquals(newTrackingUrl, rm.getRMContext().getRMApps().get(
app1.getApplicationId()).getCurrentAppAttempt()
.getOriginalTrackingUrl());
rm.stop();
rm2.stop();
}
private Credentials getCreds() throws IOException {
Credentials ts = new Credentials();
DataOutputBuffer dob = new DataOutputBuffer();