YARN-8834 Provide Java client for fetching Yarn specific entities from TimelineReader. Contributed by Abhishek Modi

(cherry picked from commit a3edfddcf7822ea13bdf4858672eb82cea5e0b5f)

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;
+}

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;
+  }
+}

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;
+    }
+  }
+}