Skip to content

YARN-7614. [RESERVATION] Support ListReservation APIs in Federation Router. #4843

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 17 commits into from
Sep 12, 2022
Merged

YARN-7614. [RESERVATION] Support ListReservation APIs in Federation Router. #4843

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
918c782
YARN-7614. [RESERVATION] Support ListReservation APIs in Federation R…
Sep 1, 2022
e8cb452
YARN-7614. Fix CheckStyle.
Sep 1, 2022
36cb3a8
YARN-7614. Fix CheckStyle.
Sep 1, 2022
902cdca
YARN-7614. Fix CheckStyle.
Sep 1, 2022
1088900
Merge branch 'apache:trunk' into YARN-7614
slfan1989 Sep 2, 2022
633b91f
YARN-7614. Fix CheckStyle.
Sep 3, 2022
b58dfd0
YARN-7614. Fix CheckStyle.
Sep 3, 2022
9cfa000
YARN-7614. Fix CheckStyle.
Sep 4, 2022
89291c2
YARN-7614. Add ListReservation Metrics.
Sep 5, 2022
1084bf2
YARN-7614. Fix CheckStyle.
Sep 5, 2022
46623a3
YARN-7614. Fix Comments.
Sep 5, 2022
4aa8042
Merge branch 'apache:trunk' into YARN-7614
slfan1989 Sep 7, 2022
332d193
YARN-7614. Fix CodeStyle.
Sep 8, 2022
04533a1
Merge remote-tracking branch 'origin/YARN-7614' into YARN-7614
Sep 8, 2022
42925e9
Merge branch 'apache:trunk' into YARN-7614
slfan1989 Sep 11, 2022
9cdbf88
YARN-7614. Fix CheckStyle.
Sep 11, 2022
397c4be
Merge branch 'apache:trunk' into YARN-7614
slfan1989 Sep 12, 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
7 changes: 7 additions & 0 deletions ...r-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRM.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

import static org.apache.hadoop.yarn.server.resourcemanager.MockNM.createMockNodeStatus;

import org.apache.hadoop.classification.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.security.token.Token;
Expand Down Expand Up @@ -64,6 +65,7 @@
import org.apache.hadoop.yarn.server.resourcemanager.recovery.MemoryRMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystem;
import org.apache.hadoop.yarn.server.resourcemanager.resource.TestResourceProfiles;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
Expand Down Expand Up @@ -1035,4 +1037,9 @@ protected void serviceInit(Configuration conf) throws Exception {
public RMStateStore getRMStateStore() {
return getRMContext().getStateStore();
}

@VisibleForTesting
public ReservationSystem getReservationSystem(){
return this.reservationSystem;
}
}
49 changes: 48 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 @@ -53,6 +53,7 @@
import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.NodeLabel;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
Expand Down Expand Up @@ -1483,7 +1484,27 @@ public Response deleteReservation(ReservationDeleteRequestInfo resContext,
public Response listReservation(String queue, String reservationId,
long startTime, long endTime, boolean includeResourceAllocations,
HttpServletRequest hsr) throws Exception {
throw new NotImplementedException("Code is not implemented");

if (queue == null || queue.isEmpty()) {
throw new IllegalArgumentException("Parameter error, the queue is empty or null.");
}

if (reservationId == null || reservationId.isEmpty()) {
throw new IllegalArgumentException("Parameter error, the reservationId is empty or null.");
}

try {
SubClusterInfo subClusterInfo = getHomeSubClusterInfoByReservationId(reservationId);
DefaultRequestInterceptorREST interceptor = getOrCreateInterceptorForSubCluster(
subClusterInfo.getSubClusterId(), subClusterInfo.getRMWebServiceAddress());
HttpServletRequest hsrCopy = clone(hsr);
return interceptor.listReservation(queue, reservationId, startTime, endTime,
includeResourceAllocations, hsrCopy);
} catch (YarnException e) {
RouterServerUtil.logAndThrowRunTimeException("listReservation Failed.", e);
}

return null;
}

@Override
Expand Down Expand Up @@ -1808,6 +1829,32 @@ private SubClusterInfo getHomeSubClusterInfoByAppId(String appId)
throw new YarnException("Unable to get subCluster by applicationId = " + appId);
}

/**
* get the HomeSubCluster according to ReservationId.
*
* @param resId reservationId
* @return HomeSubCluster
* @throws YarnException on failure
*/
private SubClusterInfo getHomeSubClusterInfoByReservationId(String resId)
throws YarnException {
SubClusterInfo subClusterInfo = null;

@goiri goiri Sep 2, 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.

Declare it where we use it or even just return.

@slfan1989 slfan1989 Sep 3, 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.

try {
ReservationId reservationId = ReservationId.parseReservationId(resId);
SubClusterId subClusterId = federationFacade.getReservationHomeSubCluster(reservationId);
if (subClusterId == null) {
RouterServerUtil.logAndThrowException(null,
"Can't get HomeSubCluster by reservationId %s", resId);
}
subClusterInfo = federationFacade.getSubCluster(subClusterId);
return subClusterInfo;
} catch (YarnException | IOException e) {
RouterServerUtil.logAndThrowException(e,
"Get HomeSubClusterInfo by reservationId %s failed.", resId);
}
throw new YarnException("Unable to get subCluster by reservationId = " + resId);
}

@VisibleForTesting
public LRUCacheHashMap<RouterAppInfoCacheKey, AppsInfo> getAppInfosCaches() {
return appInfosCaches;
Expand Down
94 changes: 94 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 @@ -40,7 +40,9 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.security.authorize.AuthorizationException;
import org.apache.hadoop.util.Sets;
import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
Expand All @@ -57,10 +59,18 @@
import org.apache.hadoop.yarn.api.records.YarnApplicationAttemptState;
import org.apache.hadoop.yarn.api.records.ApplicationTimeoutType;
import org.apache.hadoop.yarn.api.records.ApplicationTimeout;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationSubmissionRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationListRequest;
import org.apache.hadoop.yarn.api.protocolrecords.ReservationListResponse;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.server.federation.store.records.SubClusterId;
import org.apache.hadoop.yarn.server.resourcemanager.ClientRMService;
import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystem;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystemTestUtil;
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
Expand All @@ -70,6 +80,8 @@
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityState;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityLevel;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
Expand Down Expand Up @@ -97,6 +109,7 @@
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.StatisticsItemInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ApplicationStatisticsInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppActivitiesInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationListInfo;
import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
import org.apache.hadoop.yarn.server.webapp.dao.AppAttemptInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainerInfo;
Expand Down Expand Up @@ -126,12 +139,30 @@ public class MockDefaultRequestInterceptorREST
private Map<ApplicationId, ApplicationReport> applicationMap = new HashMap<>();
public static final String APP_STATE_RUNNING = "RUNNING";

private static final String QUEUE_DEFAULT = "default";
private static final String QUEUE_DEFAULT_FULL = CapacitySchedulerConfiguration.ROOT +
CapacitySchedulerConfiguration.DOT + QUEUE_DEFAULT;
private static final String QUEUE_DEDICATED = "dedicated";
public static final String QUEUE_DEDICATED_FULL = CapacitySchedulerConfiguration.ROOT +
CapacitySchedulerConfiguration.DOT + QUEUE_DEDICATED;
private MockRM mockRM;

private void validateRunning() throws ConnectException {
if (!isRunning) {
throw new ConnectException("RM is stopped");
}
}

public MockDefaultRequestInterceptorREST(){
super();
try {
mockRM = setupResourceManager();
} catch (Exception ex) {
mockRM = null;
LOG.error("mockRM init failed", ex);
}
}

@Override
public Response createNewApplication(HttpServletRequest hsr)
throws AuthorizationException, IOException, InterruptedException {
Expand Down Expand Up @@ -788,4 +819,67 @@ public AppActivitiesInfo getAppActivities(

return appActivitiesInfo;
}

@Override
public Response listReservation(String queue, String reservationId, long startTime, long endTime,
boolean includeResourceAllocations, HttpServletRequest hsr) throws Exception {

if (!isRunning) {
throw new RuntimeException("RM is stopped");
}

if (!StringUtils.equals(queue, QUEUE_DEDICATED_FULL)) {
throw new RuntimeException("The specified queue: " + queue +
" is not managed by reservation system." +
" Please try again with a valid reservable queue.");
}

ReservationId reservationID = ReservationId.parseReservationId(reservationId);
ReservationSystem reservationSystem = mockRM.getReservationSystem();
reservationSystem.synchronizePlan(QUEUE_DEDICATED_FULL, true);

// Generate reserved resources
ClientRMService clientService = mockRM.getClientRMService();
long arrival = Time.now();
long duration = 60000;

@goiri goiri Sep 2, 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.

We probably want to have reasons for these constants even though is just a mock.

long deadline = (long) (arrival + 1.05 * duration);
ReservationSubmissionRequest submissionRequest =
ReservationSystemTestUtil.createSimpleReservationRequest(reservationID, 4,
arrival, deadline, duration);

@goiri goiri Sep 2, 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.

The indentation is not correct.

@slfan1989 slfan1989 Sep 3, 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.

clientService.submitReservation(submissionRequest);

// listReservations
ReservationListRequest request = ReservationListRequest.newInstance(
queue, reservationID.toString(), startTime, endTime, includeResourceAllocations);
ReservationListResponse resRespInfo = clientService.listReservations(request);
ReservationListInfo resResponse =
new ReservationListInfo(resRespInfo, includeResourceAllocations);
return Response.status(Status.OK).entity(resResponse).build();
}

private MockRM setupResourceManager() throws Exception {
CapacitySchedulerConfiguration conf = new CapacitySchedulerConfiguration();

// Define default queue
conf.setCapacity(QUEUE_DEFAULT_FULL, 20);
// Define dedicated queues
conf.setQueues(CapacitySchedulerConfiguration.ROOT,
new String[] {QUEUE_DEFAULT, QUEUE_DEDICATED});
conf.setCapacity(QUEUE_DEDICATED_FULL, 80);
conf.setReservable(QUEUE_DEDICATED_FULL, true);

conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class);
conf.setBoolean(YarnConfiguration.RM_RESERVATION_SYSTEM_ENABLE, true);
MockRM rm = new MockRM(conf);
rm.start();
rm.registerNode("127.0.0.1:5678", 100*1024, 100);
return rm;
}

@Override
public void shutdown() {
if (mockRM != null) {
mockRM.stop();
}
}
}
70 changes: 70 additions & 0 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 @@ -28,9 +28,11 @@
import java.util.Collections;

import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;

import org.apache.hadoop.util.Time;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceOption;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
Expand All @@ -48,6 +50,8 @@
import org.apache.hadoop.yarn.server.federation.store.records.GetApplicationHomeSubClusterRequest;
import org.apache.hadoop.yarn.server.federation.store.records.GetApplicationHomeSubClusterResponse;
import org.apache.hadoop.yarn.server.federation.store.records.ApplicationHomeSubCluster;
import org.apache.hadoop.yarn.server.federation.store.records.ReservationHomeSubCluster;
import org.apache.hadoop.yarn.server.federation.store.records.AddReservationHomeSubClusterRequest;
import org.apache.hadoop.yarn.server.federation.utils.FederationStateStoreFacade;
import org.apache.hadoop.yarn.server.federation.utils.FederationStateStoreTestUtil;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppInfo;
Expand All @@ -71,10 +75,15 @@
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.StatisticsItemInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppPriority;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppQueue;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationListInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.NodeIDsInfo;
import org.apache.hadoop.yarn.server.router.webapp.cache.RouterAppInfoCacheKey;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ApplicationStatisticsInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppActivitiesInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationDefinitionInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationRequestsInfo;
import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.ReservationRequestInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainerInfo;
import org.apache.hadoop.yarn.server.webapp.dao.ContainersInfo;
import org.apache.hadoop.yarn.util.LRUCacheHashMap;
Expand Down Expand Up @@ -1056,4 +1065,65 @@ public void testGetAppActivities() throws IOException, InterruptedException {
Assert.assertEquals(appId.toString(), appActivitiesInfo.getApplicationId());
Assert.assertEquals(10, appActivitiesInfo.getAllocations().size());
}

@Test
public void testListReservation() throws Exception {

// Add ReservationId In stateStore
ReservationId reservationId = ReservationId.newInstance(Time.now(), 1);
SubClusterId homeSubClusterId = subClusters.get(0);
ReservationHomeSubCluster reservationHomeSubCluster =
ReservationHomeSubCluster.newInstance(reservationId, homeSubClusterId);
AddReservationHomeSubClusterRequest request =
AddReservationHomeSubClusterRequest.newInstance(reservationHomeSubCluster);
stateStore.addReservationHomeSubCluster(request);

// Call the listReservation method
String applyReservationId = reservationId.toString();
Response listReservationResponse =
interceptor.listReservation(MockDefaultRequestInterceptorREST.QUEUE_DEDICATED_FULL,
applyReservationId, -1, -1, false, null);

@goiri goiri Sep 7, 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.

Indentation

Assert.assertNotNull(listReservationResponse);
Assert.assertNotNull(listReservationResponse.getStatus());
Status status = Status.fromStatusCode(listReservationResponse.getStatus());
Assert.assertEquals(Status.OK, status);

Object entity = listReservationResponse.getEntity();
Assert.assertNotNull(entity);
Assert.assertNotNull(entity instanceof ReservationListInfo);

ReservationListInfo listInfo = (ReservationListInfo) entity;
Assert.assertNotNull(listInfo);

List<ReservationInfo> reservationInfoList = listInfo.getReservations();
Assert.assertNotNull(reservationInfoList);
Assert.assertEquals(1, reservationInfoList.size());

ReservationInfo reservationInfo = reservationInfoList.get(0);
Assert.assertNotNull(reservationInfo);
Assert.assertEquals(applyReservationId, reservationInfo.getReservationId());

ReservationDefinitionInfo definitionInfo = reservationInfo.getReservationDefinition();
Assert.assertNotNull(definitionInfo);

ReservationRequestsInfo reservationRequestsInfo = definitionInfo.getReservationRequests();
Assert.assertNotNull(reservationRequestsInfo);

ArrayList<ReservationRequestInfo> reservationRequestInfoList =
reservationRequestsInfo.getReservationRequest();
Assert.assertNotNull(reservationRequestInfoList);
Assert.assertEquals(1, reservationRequestInfoList.size());

ReservationRequestInfo reservationRequestInfo = reservationRequestInfoList.get(0);
Assert.assertNotNull(reservationRequestInfo);
Assert.assertEquals(4, reservationRequestInfo.getNumContainers());

ResourceInfo resourceInfo = reservationRequestInfo.getCapability();
Assert.assertNotNull(resourceInfo);

int vCore = resourceInfo.getvCores();
long memory = resourceInfo.getMemorySize();
Assert.assertEquals(1, vCore);
Assert.assertEquals(1024, memory);
}
}