Repository: hadoop Updated Branches: refs/heads/trunk fb18cc5ea -> a3edfddcf
YARN-8834 Provide Java client for fetching Yarn specific entities from TimelineReader. Contributed by Abhishek Modi Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a3edfddc Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a3edfddc Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a3edfddc Branch: refs/heads/trunk Commit: a3edfddcf7822ea13bdf4858672eb82cea5e0b5f Parents: fb18cc5 Author: Vrushali C <vrush...@apache.org> Authored: Thu Oct 11 20:21:00 2018 -0700 Committer: Vrushali C <vrush...@apache.org> Committed: Thu Oct 11 20:21:00 2018 -0700 ---------------------------------------------------------------------- .../yarn/client/api/TimelineReaderClient.java | 120 ++++++++++ .../api/impl/TimelineReaderClientImpl.java | 239 +++++++++++++++++++ .../api/impl/TestTimelineReaderClientImpl.java | 157 ++++++++++++ 3 files changed, 516 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/a3edfddc/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineReaderClient.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineReaderClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineReaderClient.java new file mode 100644 index 0000000..f73c2d3 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/TimelineReaderClient.java @@ -0,0 +1,120 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.client.api; + +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.service.CompositeService; +import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ContainerId; +import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity; +import org.apache.hadoop.yarn.client.api.impl.TimelineReaderClientImpl; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * A client library that can be used to get Timeline Entities associated with + * application, application attempt or containers. This client library needs to + * be used along with time line v.2 server version. + */ +@InterfaceAudience.Public +@InterfaceStability.Evolving +public abstract class TimelineReaderClient extends CompositeService { + + /** + * Create a new instance of Timeline Reader Client. + */ + @InterfaceAudience.Public + public static TimelineReaderClient createTimelineReaderClient() { + return new TimelineReaderClientImpl(); + } + + @InterfaceAudience.Private + public TimelineReaderClient(String name) { + super(name); + } + + /** + * Gets application entity. + * @param appId application id + * @param fields Fields to be fetched. Defaults to INFO. + * @param filters Filters to be applied while fetching entities. + * @return entity of the application + * @throws IOException + */ + public abstract TimelineEntity getApplicationEntity( + ApplicationId appId, String fields, Map<String, String> filters) + throws IOException; + + /** + * Gets application attempt entity. + * @param appAttemptId application attempt id + * @param fields Fields to be fetched. Defaults to INFO. + * @param filters Filters to be applied while fetching entities. + * @return entity associated with application attempt + * @throws IOException + */ + public abstract TimelineEntity getApplicationAttemptEntity( + ApplicationAttemptId appAttemptId, String fields, + Map<String, String> filters) throws IOException; + + /** + * Gets application attempt entities. + * @param appId application id + * @param fields Fields to be fetched. Defaults to INFO. + * @param filters Filters to be applied while fetching entities. + * @param limit Number of entities to return. + * @param fromId Retrieve next set of generic ids from given fromId + * @return list of application attempt entities + * @throws IOException + */ + public abstract List<TimelineEntity> getApplicationAttemptEntities( + ApplicationId appId, String fields, Map<String, String> filters, + long limit, String fromId) throws IOException; + + /** + * Gets Timeline entity for the container. + * @param containerId container id + * @param fields Fields to be fetched. Defaults to INFO. + * @param filters Filters to be applied while fetching entities. + * @return timeline entity for container + * @throws IOException + */ + public abstract TimelineEntity getContainerEntity( + ContainerId containerId, String fields, Map<String, String> filters) + throws IOException; + + /** + * Gets container entities for an application. + * @param appId application id + * @param fields Fields to be fetched. Defaults to INFO. + * @param filters Filters to be applied while fetching entities. + * @param limit Number of entities to return. + * @param fromId Retrieve next set of generic ids from given fromId + * @return list of entities + * @throws IOException + */ + public abstract List<TimelineEntity> getContainerEntities( + ApplicationId appId, String fields, + Map<String, String> filters, + long limit, String fromId) throws IOException; +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/a3edfddc/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineReaderClientImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineReaderClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineReaderClientImpl.java new file mode 100644 index 0000000..b138a54 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/client/api/impl/TimelineReaderClientImpl.java @@ -0,0 +1,239 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.yarn.client.api.impl; + +import com.google.common.annotations.VisibleForTesting; +import com.sun.jersey.core.util.MultivaluedMapImpl; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.classification.InterfaceStability; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticatedURL; +import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ContainerId; +import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity; +import org.apache.hadoop.yarn.client.api.TimelineReaderClient; +import com.sun.jersey.api.client.ClientResponse; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.apache.hadoop.yarn.webapp.util.WebAppUtils; + +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.MultivaluedMap; +import java.io.IOException; +import java.net.URI; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntityType.YARN_APPLICATION_ATTEMPT; +import static org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntityType.YARN_CONTAINER; +import static org.apache.hadoop.yarn.util.StringHelper.PATH_JOINER; + +/** + * Implementation of TimelineReaderClient interface. + */ +@InterfaceAudience.Public +@InterfaceStability.Evolving +public class TimelineReaderClientImpl extends TimelineReaderClient { + private static final Log LOG = + LogFactory.getLog(TimelineReaderClientImpl.class); + + private static final String RESOURCE_URI_STR_V2 = "/ws/v2/timeline/"; + + private TimelineConnector connector; + private URI baseUri; + private String clusterId; + + public TimelineReaderClientImpl() { + super(TimelineReaderClientImpl.class.getName()); + } + + @Override + protected void serviceInit(Configuration conf) throws Exception { + if (!YarnConfiguration.timelineServiceV2Enabled(conf)) { + throw new IOException("Timeline V2 client is not properly configured. " + + "Either timeline service is not enabled or version is not set to" + + " 2"); + } + UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); + UserGroupInformation realUgi = ugi.getRealUser(); + String doAsUser; + UserGroupInformation authUgi; + if (realUgi != null) { + authUgi = realUgi; + doAsUser = ugi.getShortUserName(); + } else { + authUgi = ugi; + doAsUser = null; + } + DelegationTokenAuthenticatedURL.Token token = + new DelegationTokenAuthenticatedURL.Token(); + connector = new TimelineConnector(false, authUgi, doAsUser, token); + addIfService(connector); + String timelineReaderWebAppAddress = + WebAppUtils.getTimelineReaderWebAppURLWithoutScheme(conf); + baseUri = TimelineConnector.constructResURI( + conf, timelineReaderWebAppAddress, RESOURCE_URI_STR_V2); + clusterId = conf.get(YarnConfiguration.RM_CLUSTER_ID, + YarnConfiguration.DEFAULT_RM_CLUSTER_ID); + super.serviceInit(conf); + } + + @Override + public TimelineEntity getApplicationEntity(ApplicationId appId, String fields, + Map<String, String> filters) + throws IOException { + String path = PATH_JOINER.join("clusters", clusterId, "apps", appId); + + if (fields == null || fields.isEmpty()) { + fields = "INFO"; + } + MultivaluedMap<String, String> params = new MultivaluedMapImpl(); + params.add("fields", fields); + mergeFilters(params, filters); + + ClientResponse response = doGetUri(baseUri, path, params); + TimelineEntity entity = response.getEntity(TimelineEntity.class); + return entity; + } + + @Override + public TimelineEntity getApplicationAttemptEntity( + ApplicationAttemptId appAttemptId, + String fields, Map<String, String> filters) throws IOException { + ApplicationId appId = appAttemptId.getApplicationId(); + String path = PATH_JOINER.join("clusters", clusterId, "apps", + appId, "entities", YARN_APPLICATION_ATTEMPT, appAttemptId); + + if (fields == null || fields.isEmpty()) { + fields = "INFO"; + } + MultivaluedMap<String, String> params = new MultivaluedMapImpl(); + params.add("fields", fields); + mergeFilters(params, filters); + + ClientResponse response = doGetUri(baseUri, path, params); + TimelineEntity entity = response.getEntity(TimelineEntity.class); + return entity; + } + + @Override + public List<TimelineEntity> getApplicationAttemptEntities( + ApplicationId appId, String fields, Map<String, String> filters, + long limit, String fromId) throws IOException { + String path = PATH_JOINER.join("clusters", clusterId, "apps", + appId, "entities", YARN_APPLICATION_ATTEMPT); + + if (fields == null || fields.isEmpty()) { + fields = "INFO"; + } + MultivaluedMap<String, String> params = new MultivaluedMapImpl(); + params.add("fields", fields); + if (limit > 0) { + params.add("limit", Long.toString(limit)); + } + if (fromId != null && !fromId.isEmpty()) { + params.add("fromid", fromId); + } + mergeFilters(params, filters); + + ClientResponse response = doGetUri(baseUri, path, params); + TimelineEntity[] entities = response.getEntity(TimelineEntity[].class); + return Arrays.asList(entities); + } + + @Override + public TimelineEntity getContainerEntity(ContainerId containerId, + String fields, Map<String, String> filters) throws IOException { + ApplicationId appId = containerId.getApplicationAttemptId(). + getApplicationId(); + String path = PATH_JOINER.join("clusters", clusterId, "apps", + appId, "entities", YARN_CONTAINER, containerId); + + if (fields == null || fields.isEmpty()) { + fields = "INFO"; + } + MultivaluedMap<String, String> params = new MultivaluedMapImpl(); + params.add("fields", fields); + mergeFilters(params, filters); + + ClientResponse response = doGetUri(baseUri, path, params); + TimelineEntity entity = response.getEntity(TimelineEntity.class); + return entity; + } + + @Override + public List<TimelineEntity> getContainerEntities( + ApplicationId appId, String fields, + Map<String, String> filters, + long limit, String fromId) throws IOException { + String path = PATH_JOINER.join("clusters", clusterId, "apps", + appId, "entities", YARN_CONTAINER); + + if (fields == null || fields.isEmpty()) { + fields = "INFO"; + } + MultivaluedMap<String, String> params = new MultivaluedMapImpl(); + params.add("fields", fields); + if (limit > 0) { + params.add("limit", Long.toString(limit)); + } + if (fromId != null && !fromId.isEmpty()) { + params.add("fromid", fromId); + } + mergeFilters(params, filters); + + ClientResponse response = doGetUri(baseUri, path, params); + TimelineEntity[] entity = response.getEntity(TimelineEntity[].class); + return Arrays.asList(entity); + } + + private void mergeFilters(MultivaluedMap<String, String> defaults, + Map<String, String> filters) { + if (filters != null && !filters.isEmpty()) { + for (Map.Entry<String, String> entry : filters.entrySet()) { + if (!defaults.containsKey(entry.getKey())) { + defaults.add(entry.getKey(), filters.get(entry.getValue())); + } + } + } + } + + @VisibleForTesting + protected ClientResponse doGetUri(URI base, String path, + MultivaluedMap<String, String> params) throws IOException { + ClientResponse resp = connector.getClient().resource(base).path(path) + .queryParams(params).accept(MediaType.APPLICATION_JSON) + .get(ClientResponse.class); + if (resp == null || + resp.getStatusInfo().getStatusCode() != ClientResponse.Status.OK + .getStatusCode()) { + String msg = + "Response from the timeline reader server is " + + ((resp == null) ? "null" : "not successful," + + " HTTP error code: " + resp.getStatus() + + ", Server response:\n" + resp.getEntity(String.class)); + LOG.error(msg); + throw new IOException(msg); + } + return resp; + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/a3edfddc/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineReaderClientImpl.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineReaderClientImpl.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineReaderClientImpl.java new file mode 100644 index 0000000..f668472 --- /dev/null +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/test/java/org/apache/hadoop/yarn/client/api/impl/TestTimelineReaderClientImpl.java @@ -0,0 +1,157 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.yarn.client.api.impl; + +import static org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntityType.YARN_APPLICATION_ATTEMPT; +import static org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntityType.YARN_CONTAINER; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.sun.jersey.api.client.ClientResponse; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; +import org.apache.hadoop.yarn.api.records.ApplicationId; +import org.apache.hadoop.yarn.api.records.ContainerId; +import org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity; +import org.apache.hadoop.yarn.client.api.TimelineReaderClient; +import org.apache.hadoop.yarn.conf.YarnConfiguration; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import javax.ws.rs.core.MultivaluedMap; +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.List; + +/** + * Test class for Timeline Reader Client. + */ +public class TestTimelineReaderClientImpl { + + private TimelineReaderClient client; + + @Before + public void setup() { + client = new MockTimelineReaderClient(); + Configuration conf = new YarnConfiguration(); + conf.setBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED, true); + conf.setFloat(YarnConfiguration.TIMELINE_SERVICE_VERSION, 2.0f); + client.init(conf); + client.start(); + } + + @Test + public void testGetApplication() throws Exception { + ApplicationId applicationId = + ApplicationId.fromString("application_1234_0001"); + TimelineEntity entity = client.getApplicationEntity(applicationId, + null, null); + Assert.assertEquals("mockApp1", entity.getId()); + } + + @Test + public void getApplicationAttemptEntity() throws Exception { + ApplicationAttemptId attemptId = + ApplicationAttemptId.fromString("appattempt_1234_0001_000001"); + TimelineEntity entity = client.getApplicationAttemptEntity(attemptId, + null, null); + Assert.assertEquals("mockAppAttempt1", entity.getId()); + } + + @Test + public void getApplicationAttemptEntities() throws Exception { + ApplicationId applicationId = + ApplicationId.fromString("application_1234_0001"); + List<TimelineEntity> entities = + client.getApplicationAttemptEntities(applicationId, null, + null, 0, null); + Assert.assertEquals(2, entities.size()); + Assert.assertEquals("mockAppAttempt2", entities.get(1).getId()); + } + + @Test + public void testGetContainer() throws Exception { + ContainerId containerId = + ContainerId.fromString("container_1234_0001_01_000001"); + TimelineEntity entity = client.getContainerEntity(containerId, + null, null); + Assert.assertEquals("mockContainer1", entity.getId()); + } + + @Test + public void testGetContainers() throws Exception { + ApplicationId appId = + ApplicationId.fromString("application_1234_0001"); + List<TimelineEntity> entities = client.getContainerEntities(appId, + null, null, 0, null); + Assert.assertEquals(2, entities.size()); + Assert.assertEquals("mockContainer2", entities.get(1).getId()); + } + + @After + public void tearDown() { + if (client != null) { + client.stop(); + } + } + + private static TimelineEntity createTimelineEntity(String id) { + TimelineEntity entity = new TimelineEntity(); + entity.setId(id); + return entity; + } + + private static TimelineEntity[] createTimelineEntities(String... ids) { + List<TimelineEntity> entities = new ArrayList<>(); + for (String id : ids) { + TimelineEntity entity = new TimelineEntity(); + entity.setId(id); + entities.add(entity); + } + return entities.toArray(new TimelineEntity[entities.size()]); + } + + private class MockTimelineReaderClient extends TimelineReaderClientImpl { + @Override + protected ClientResponse doGetUri(URI base, String path, + MultivaluedMap<String, String> params) throws IOException { + ClientResponse mockClientResponse = mock(ClientResponse.class); + if (path.contains(YARN_CONTAINER.toString())) { + when(mockClientResponse.getEntity(TimelineEntity.class)).thenReturn( + createTimelineEntity("mockContainer1")); + when(mockClientResponse.getEntity(TimelineEntity[].class)).thenReturn( + createTimelineEntities("mockContainer1", "mockContainer2")); + } else if (path.contains(YARN_APPLICATION_ATTEMPT.toString())) { + when(mockClientResponse.getEntity(TimelineEntity.class)).thenReturn( + createTimelineEntity("mockAppAttempt1")); + when(mockClientResponse.getEntity(TimelineEntity[].class)).thenReturn( + createTimelineEntities("mockAppAttempt1", "mockAppAttempt2")); + } else { + when(mockClientResponse.getEntity(TimelineEntity.class)).thenReturn( + createTimelineEntity("mockApp1")); + when(mockClientResponse.getEntity(TimelineEntity[].class)).thenReturn( + createTimelineEntities("mockApp1", "mockApp2")); + } + return mockClientResponse; + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: common-commits-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-commits-h...@hadoop.apache.org
