Skip to content

YARN-8900. [Router] Federation: routing getContainers REST invocations transparently to multiple RMs #4543

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
goiri merged 8 commits into from
Jul 23, 2022
Merged

YARN-8900. [Router] Federation: routing getContainers REST invocations transparently to multiple RMs #4543

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c1b44fa
YARN-8900. [Router] Federation: routing getContainers REST invocation…
Jul 10, 2022
efc1181
YARN-8890. Fix Error Msg.
Jul 14, 2022
72a1849
YARN-8900. refactor generic methods.
Jul 15, 2022
e0902f5
YARN-8900. Fix CheckStyle.
Jul 16, 2022
210bfd1
YARN-8900. Fix CheckStyle.
Jul 16, 2022
ffd3eb4
YARN-8900. Fix CheckStyle.
Jul 16, 2022
a8cf91d
YARN-8900. Fix CheckStyle.
Jul 17, 2022
8810220
YARN-8900. Fix CheckStyle.
Jul 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions ...-server-common/src/main/java/org/apache/hadoop/yarn/server/webapp/dao/ContainersInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,4 +46,7 @@ public ArrayList<ContainerInfo> getContainers() {
return container;
}

public void addAll(ArrayList<ContainerInfo> containersInfo) {

@goiri goiri Jul 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe just make this Collection<ContainerInfo>

@slfan1989 slfan1989 Jul 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for helping to review the code, I will modify the code.

container.addAll(containersInfo);
}
}
46 changes: 45 additions & 1 deletion .../src/main/java/org/apache/hadoop/yarn/server/router/webapp/FederationInterceptorREST.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1336,7 +1336,51 @@ public AppAttemptInfo getAppAttempt(HttpServletRequest req,
@Override
public ContainersInfo getContainers(HttpServletRequest req,
HttpServletResponse res, String appId, String appAttemptId) {
throw new NotImplementedException("Code is not implemented");
ContainersInfo containersInfo = new ContainersInfo();

Map<SubClusterId, SubClusterInfo> subClustersActive = null;
try {
subClustersActive = federationFacade.getSubClusters(true);
} catch (YarnException e) {
LOG.error("Get All active sub cluster(s) error.", e);
return containersInfo;
}

// Send the requests in parallel
CompletionService<ContainersInfo> compSvc =
new ExecutorCompletionService<>(this.threadpool);

for (final SubClusterInfo info : subClustersActive.values()) {
compSvc.submit(() -> {
DefaultRequestInterceptorREST interceptor =
getOrCreateInterceptorForSubCluster(info.getSubClusterId(), info.getRMWebServiceAddress());
try {
ContainersInfo containers =
interceptor.getContainers(req, res, appId, appAttemptId);
return containers;
} catch (Exception e) {
LOG.error("SubCluster {} failed to return GetContainers.",

@goiri goiri Jul 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One line

@slfan1989 slfan1989 Jul 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will fix it.

info.getSubClusterId());

@goiri goiri Jul 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we print the specific error?

return null;
}
});
}

// Collect all the responses in parallel
for (int i = 0; i < subClustersActive.size(); i++) {
try {
Future<ContainersInfo> future = compSvc.take();
ContainersInfo containersResponse = future.get();

if (containersResponse != null) {
containersInfo.addAll(containersResponse.getContainers());
}
} catch (Throwable e) {
LOG.warn("Failed to get containers report. ", e);

@goiri goiri Jul 14, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Trim the space at the end of the string.

@slfan1989 slfan1989 Jul 14, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK, I will fix it.

}
}

return containersInfo;
}

@Override
Expand Down
48 changes: 48 additions & 0 deletions ...t/java/org/apache/hadoop/yarn/server/router/webapp/MockDefaultRequestInterceptorREST.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,13 +25,20 @@
import java.util.concurrent.atomic.AtomicInteger;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.apache.hadoop.security.authorize.AuthorizationException;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.ContainerState;
import org.apache.hadoop.yarn.api.records.ContainerReport;
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
Expand All @@ -45,6 +52,8 @@
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodesInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ResourceInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ResourceOptionInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainerInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainersInfo;
import org.apache.hadoop.yarn.webapp.NotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -231,4 +240,43 @@ public boolean isRunning() {
public void setRunning(boolean runningMode) {
this.isRunning = runningMode;
}

@Override
public ContainersInfo getContainers(HttpServletRequest req, HttpServletResponse res,
String appId, String appAttemptId) {
if (!isRunning) {
throw new RuntimeException("RM is stopped");
}

// We avoid to check if the Application exists in the system because we need
// to validate that each subCluster returns 1 container.
ContainersInfo containers = new ContainersInfo();

int subClusterId = Integer.valueOf(getSubClusterId().getId());

ContainerId containerId = ContainerId.newContainerId(
ApplicationAttemptId.fromString(appAttemptId), subClusterId);
Resource allocatedResource =
Resource.newInstance(subClusterId, subClusterId);

NodeId assignedNode = NodeId.newInstance("Node", subClusterId);
Priority priority = Priority.newInstance(subClusterId);
long creationTime = subClusterId;
long finishTime = subClusterId;
String diagnosticInfo = "Diagnostic " + subClusterId;
String logUrl = "Log " + subClusterId;
int containerExitStatus = subClusterId;
ContainerState containerState = ContainerState.COMPLETE;
String nodeHttpAddress = "HttpAddress " + subClusterId;

ContainerReport containerReport =
Comment thread
goiri marked this conversation as resolved.
Outdated
ContainerReport.newInstance(containerId, allocatedResource,
assignedNode, priority, creationTime, finishTime, diagnosticInfo,
logUrl, containerExitStatus, containerState, nodeHttpAddress);

ContainerInfo container = new ContainerInfo(containerReport);
containers.add(container);

return containers;
}
}
42 changes: 34 additions & 8 deletions .../test/java/org/apache/hadoop/yarn/server/router/webapp/TestFederationInterceptorREST.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,11 @@

import javax.ws.rs.core.Response;

import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceOption;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.federation.policies.manager.UniformBroadcastPolicyManager;
Expand All @@ -47,6 +49,7 @@
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.NodesInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ResourceInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ResourceOptionInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainersInfo;
import org.apache.hadoop.yarn.util.MonotonicClock;
import org.junit.Assert;
import org.junit.Test;
Expand Down Expand Up @@ -160,7 +163,7 @@ public void testSubmitApplication()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);

ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
Expand All @@ -187,7 +190,7 @@ public void testSubmitApplicationMultipleSubmission()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);
ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
context.setApplicationId(appId.toString());
Expand Down Expand Up @@ -259,7 +262,7 @@ public void testForceKillApplication()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);
ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
context.setApplicationId(appId.toString());
Expand All @@ -286,7 +289,7 @@ public void testForceKillApplicationNotExists()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);
AppState appState = new AppState("KILLED");

Response response =
Expand Down Expand Up @@ -317,7 +320,7 @@ public void testForceKillApplicationWrongFormat()
public void testForceKillApplicationEmptyRequest()
throws YarnException, IOException, InterruptedException {
ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);

ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
Expand All @@ -341,7 +344,7 @@ public void testGetApplicationReport()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);
ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
context.setApplicationId(appId.toString());
Expand Down Expand Up @@ -478,7 +481,7 @@ public void testGetApplicationState()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);
ApplicationSubmissionContextInfo context =
new ApplicationSubmissionContextInfo();
context.setApplicationId(appId.toString());
Expand All @@ -505,7 +508,7 @@ public void testGetApplicationStateNotExists()
throws YarnException, IOException, InterruptedException {

ApplicationId appId =
ApplicationId.newInstance(System.currentTimeMillis(), 1);
ApplicationId.newInstance(Time.now(), 1);

AppState response = interceptor.getAppState(null, appId.toString());

Expand Down Expand Up @@ -560,4 +563,27 @@ SubClusterState.SC_RUNNING, new MonotonicClock().getTime(),
SubClusterRegisterRequest.newInstance(subClusterInfo));
}

@Test
public void testGetContainers()
throws YarnException, IOException, InterruptedException {

ApplicationId appId = ApplicationId.newInstance(Time.now(), 1);
ApplicationSubmissionContextInfo context =
Comment thread
goiri marked this conversation as resolved.
new ApplicationSubmissionContextInfo();
context.setApplicationId(appId.toString());

// Submit the application we want the report later
Response response = interceptor.submitApplication(context, null);

Assert.assertNotNull(response);
Assert.assertNotNull(stateStoreUtil.queryApplicationHomeSC(appId));

ApplicationAttemptId appAttempt =

@goiri goiri Jul 15, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One line

ApplicationAttemptId.newInstance(appId, 1);

ContainersInfo responseGet = interceptor.getContainers(null, null,

@goiri goiri Jul 15, 2022

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move all the args to the second line:

ContainersInfo responseGet = interceptor.getContainers(
    null, null, appId.toString(), appAttempt.toString());

@slfan1989 slfan1989 Jul 16, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will fix it.

appId.toString(), appAttempt.toString());

Assert.assertEquals(4, responseGet.getContainers().size());
}
}