You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ha...@apache.org on 2018/04/20 08:16:08 UTC
[1/2] tez git commit: TEZ-3915. Create protobuf based history event
logger. (Harish Jaiprakash, reviewed by Gunther Hagleitner)
Repository: tez
Updated Branches:
refs/heads/master 871ea80c0 -> 24b872a7f
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestHistoryEventProtoConverter.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestHistoryEventProtoConverter.java b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestHistoryEventProtoConverter.java
new file mode 100644
index 0000000..92d3e30
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestHistoryEventProtoConverter.java
@@ -0,0 +1,716 @@
+/**
+ * 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.tez.dag.history.logging.proto;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+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.NodeId;
+import org.apache.tez.common.ATSConstants;
+import org.apache.tez.common.VersionInfo;
+import org.apache.tez.common.counters.TezCounters;
+import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.OutputDescriptor;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.api.oldrecords.TaskState;
+import org.apache.tez.dag.api.records.DAGProtos.CallerContextProto;
+import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
+import org.apache.tez.dag.app.dag.DAGState;
+import org.apache.tez.dag.app.dag.VertexState;
+import org.apache.tez.dag.app.dag.impl.ServicePluginInfo;
+import org.apache.tez.dag.app.dag.impl.TaskAttemptImpl.DataEventDependencyInfo;
+import org.apache.tez.dag.app.dag.impl.VertexStats;
+import org.apache.tez.dag.app.web.AMWebController;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.AMLaunchedEvent;
+import org.apache.tez.dag.history.events.AMStartedEvent;
+import org.apache.tez.dag.history.events.AppLaunchedEvent;
+import org.apache.tez.dag.history.events.ContainerLaunchedEvent;
+import org.apache.tez.dag.history.events.ContainerStoppedEvent;
+import org.apache.tez.dag.history.events.DAGCommitStartedEvent;
+import org.apache.tez.dag.history.events.DAGFinishedEvent;
+import org.apache.tez.dag.history.events.DAGInitializedEvent;
+import org.apache.tez.dag.history.events.DAGKillRequestEvent;
+import org.apache.tez.dag.history.events.DAGRecoveredEvent;
+import org.apache.tez.dag.history.events.DAGStartedEvent;
+import org.apache.tez.dag.history.events.DAGSubmittedEvent;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.events.TaskAttemptStartedEvent;
+import org.apache.tez.dag.history.events.TaskFinishedEvent;
+import org.apache.tez.dag.history.events.TaskStartedEvent;
+import org.apache.tez.dag.history.events.VertexCommitStartedEvent;
+import org.apache.tez.dag.history.events.VertexConfigurationDoneEvent;
+import org.apache.tez.dag.history.events.VertexFinishedEvent;
+import org.apache.tez.dag.history.events.VertexGroupCommitFinishedEvent;
+import org.apache.tez.dag.history.events.VertexGroupCommitStartedEvent;
+import org.apache.tez.dag.history.events.VertexInitializedEvent;
+import org.apache.tez.dag.history.events.VertexStartedEvent;
+import org.apache.tez.dag.history.logging.EntityTypes;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.HistoryEventProto;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.KVPair;
+import org.apache.tez.dag.records.TaskAttemptTerminationCause;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.apache.tez.runtime.api.TaskFailureType;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+public class TestHistoryEventProtoConverter {
+ private ApplicationAttemptId applicationAttemptId;
+ private ApplicationId applicationId;
+ private String user = "user";
+ private Random random = new Random();
+ private TezDAGID tezDAGID;
+ private TezVertexID tezVertexID;
+ private TezTaskID tezTaskID;
+ private TezTaskAttemptID tezTaskAttemptID;
+ private DAGPlan dagPlan;
+ private ContainerId containerId;
+ private NodeId nodeId;
+ private String containerLogs = "containerLogs";
+ private HistoryEventProtoConverter converter = new HistoryEventProtoConverter();
+
+ @Before
+ public void setup() {
+ applicationId = ApplicationId.newInstance(9999l, 1);
+ applicationAttemptId = ApplicationAttemptId.newInstance(applicationId, 1);
+ tezDAGID = TezDAGID.getInstance(applicationId, random.nextInt());
+ tezVertexID = TezVertexID.getInstance(tezDAGID, random.nextInt());
+ tezTaskID = TezTaskID.getInstance(tezVertexID, random.nextInt());
+ tezTaskAttemptID = TezTaskAttemptID.getInstance(tezTaskID, random.nextInt());
+ CallerContextProto.Builder callerContextProto = CallerContextProto.newBuilder();
+ callerContextProto.setContext("ctxt");
+ callerContextProto.setCallerId("Caller_ID");
+ callerContextProto.setCallerType("Caller_Type");
+ callerContextProto.setBlob("Desc_1");
+ dagPlan = DAGPlan.newBuilder().setName("DAGPlanMock")
+ .setCallerContext(callerContextProto).build();
+ containerId = ContainerId.newContainerId(applicationAttemptId, 111);
+ nodeId = NodeId.newInstance("node", 13435);
+ }
+
+ @Test(timeout = 5000)
+ public void testHandlerExists() {
+ for (HistoryEventType eventType : HistoryEventType.values()) {
+ HistoryEvent event = null;
+ switch (eventType) {
+ case APP_LAUNCHED:
+ event = new AppLaunchedEvent(applicationId, random.nextInt(), random.nextInt(),
+ user, new Configuration(false), null);
+ break;
+ case AM_LAUNCHED:
+ event = new AMLaunchedEvent(applicationAttemptId, random.nextInt(), random.nextInt(),
+ user);
+ break;
+ case AM_STARTED:
+ event = new AMStartedEvent(applicationAttemptId, random.nextInt(), user);
+ break;
+ case DAG_SUBMITTED:
+ event = new DAGSubmittedEvent(tezDAGID, random.nextInt(), dagPlan, applicationAttemptId,
+ null, user, null, containerLogs, null);
+ break;
+ case DAG_INITIALIZED:
+ event = new DAGInitializedEvent(tezDAGID, random.nextInt(), user, dagPlan.getName(), null);
+ break;
+ case DAG_STARTED:
+ event = new DAGStartedEvent(tezDAGID, random.nextInt(), user, dagPlan.getName());
+ break;
+ case DAG_FINISHED:
+ event = new DAGFinishedEvent(tezDAGID, random.nextInt(), random.nextInt(), DAGState.ERROR,
+ null, null, user, dagPlan.getName(), null, applicationAttemptId, dagPlan);
+ break;
+ case VERTEX_INITIALIZED:
+ event = new VertexInitializedEvent(tezVertexID, "v1", random.nextInt(), random.nextInt(),
+ random.nextInt(), "proc", null, null, null);
+ break;
+ case VERTEX_STARTED:
+ event = new VertexStartedEvent(tezVertexID, random.nextInt(), random.nextInt());
+ break;
+ case VERTEX_CONFIGURE_DONE:
+ event = new VertexConfigurationDoneEvent(tezVertexID, 0L, 1, null, null, null, true);
+ break;
+ case VERTEX_FINISHED:
+ event = new VertexFinishedEvent(tezVertexID, "v1", 1, random.nextInt(), random.nextInt(),
+ random.nextInt(), random.nextInt(), random.nextInt(), VertexState.ERROR,
+ null, null, null, null, null);
+ break;
+ case TASK_STARTED:
+ event = new TaskStartedEvent(tezTaskID, "v1", random.nextInt(), random.nextInt());
+ break;
+ case TASK_FINISHED:
+ event = new TaskFinishedEvent(tezTaskID, "v1", random.nextInt(), random.nextInt(),
+ tezTaskAttemptID, TaskState.FAILED, null, null, 0);
+ break;
+ case TASK_ATTEMPT_STARTED:
+ event = new TaskAttemptStartedEvent(tezTaskAttemptID, "v1", random.nextInt(), containerId,
+ nodeId, null, null, "nodeHttpAddress");
+ break;
+ case TASK_ATTEMPT_FINISHED:
+ event = new TaskAttemptFinishedEvent(tezTaskAttemptID, "v1", random.nextInt(),
+ random.nextInt(), TaskAttemptState.FAILED, TaskFailureType.NON_FATAL, TaskAttemptTerminationCause.OUTPUT_LOST,
+ null, null, null, null, 0, null, 0,
+ containerId, nodeId, null, null, "nodeHttpAddress");
+ break;
+ case CONTAINER_LAUNCHED:
+ event = new ContainerLaunchedEvent(containerId, random.nextInt(),
+ applicationAttemptId);
+ break;
+ case CONTAINER_STOPPED:
+ event = new ContainerStoppedEvent(containerId, random.nextInt(), -1, applicationAttemptId);
+ break;
+ case DAG_COMMIT_STARTED:
+ event = new DAGCommitStartedEvent();
+ break;
+ case VERTEX_COMMIT_STARTED:
+ event = new VertexCommitStartedEvent();
+ break;
+ case VERTEX_GROUP_COMMIT_STARTED:
+ event = new VertexGroupCommitStartedEvent();
+ break;
+ case VERTEX_GROUP_COMMIT_FINISHED:
+ event = new VertexGroupCommitFinishedEvent();
+ break;
+ case DAG_RECOVERED:
+ event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID, dagPlan.getName(),
+ user, random.nextLong(), containerLogs);
+ break;
+ case DAG_KILL_REQUEST:
+ event = new DAGKillRequestEvent();
+ break;
+ default:
+ Assert.fail("Unhandled event type " + eventType);
+ }
+ if (event == null || !event.isHistoryEvent()) {
+ continue;
+ }
+ converter.convert(event);
+ }
+ }
+
+ static class MockVersionInfo extends VersionInfo {
+ MockVersionInfo() {
+ super("component", "1.1.0", "rev1", "20120101", "git.apache.org");
+ }
+ }
+
+ private String findEventData(HistoryEventProto proto, String key) {
+ for (KVPair data : proto.getEventDataList()) {
+ if (data.getKey().equals(key)) {
+ return data.getValue();
+ }
+ }
+ return null;
+ }
+
+ private void assertEventData(HistoryEventProto proto, String key, String value) {
+ String evtVal = findEventData(proto, key);
+ if (evtVal == null) {
+ Assert.fail("Cannot find kv pair: " + key);
+ }
+ if (value != null) {
+ Assert.assertEquals(value, evtVal);
+ }
+ }
+
+ private void assertNoEventData(HistoryEventProto proto, String key) {
+ for (KVPair data : proto.getEventDataList()) {
+ if (data.getKey().equals(key)) {
+ Assert.fail("Found find kv pair: " + key);
+ }
+ }
+ }
+
+ private String safeToString(Object obj) {
+ return obj == null ? "" : obj.toString();
+ }
+
+ private void assertCommon(HistoryEventProto proto, HistoryEventType type, long eventTime,
+ EntityTypes entityType, ApplicationAttemptId appAttemptId, String user, int numData) {
+ Assert.assertEquals(type.name(), proto.getEventType());
+ Assert.assertEquals(eventTime, proto.getEventTime());
+ // Assert.assertEquals(safeToString(appId), proto.getAppId());
+ Assert.assertEquals(safeToString(appAttemptId), proto.getAppAttemptId());
+ Assert.assertEquals(safeToString(user), proto.getUser());
+ if (entityType != null) {
+ switch (entityType) { // Intentional fallthrough.
+ case TEZ_TASK_ATTEMPT_ID:
+ Assert.assertEquals(tezTaskAttemptID.toString(), proto.getTaskAttemptId());
+ case TEZ_TASK_ID:
+ Assert.assertEquals(tezTaskID.toString(), proto.getTaskId());
+ case TEZ_VERTEX_ID:
+ Assert.assertEquals(tezVertexID.toString(), proto.getVertexId());
+ case TEZ_DAG_ID:
+ Assert.assertEquals(tezDAGID.toString(), proto.getDagId());
+ case TEZ_APPLICATION:
+ Assert.assertEquals(applicationId.toString(), proto.getAppId());
+ break;
+ default:
+ Assert.fail("Invalid type: " + entityType.name());
+ }
+ }
+ Assert.assertEquals(numData, proto.getEventDataCount());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertAppLaunchedEvent() {
+ long launchTime = random.nextLong();
+ long submitTime = random.nextLong();
+ Configuration conf = new Configuration(false);
+ conf.set("foo", "bar");
+ conf.set("applicationId", "1234");
+
+ MockVersionInfo mockVersionInfo = new MockVersionInfo();
+ AppLaunchedEvent event = new AppLaunchedEvent(applicationId, launchTime, submitTime, user,
+ conf, mockVersionInfo);
+ HistoryEventProto proto = converter.convert(event);
+
+ assertCommon(proto, HistoryEventType.APP_LAUNCHED, launchTime, EntityTypes.TEZ_APPLICATION,
+ null, user, 3);
+ assertEventData(proto, ATSConstants.CONFIG, null);
+ assertEventData(proto, ATSConstants.TEZ_VERSION, null);
+ assertEventData(proto, ATSConstants.DAG_AM_WEB_SERVICE_VERSION, AMWebController.VERSION);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertAMLaunchedEvent() {
+ long launchTime = random.nextLong();
+ long submitTime = random.nextLong();
+ AMLaunchedEvent event = new AMLaunchedEvent(applicationAttemptId, launchTime, submitTime,
+ user);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.AM_LAUNCHED, launchTime, EntityTypes.TEZ_APPLICATION,
+ applicationAttemptId, user, 1);
+ assertEventData(proto, ATSConstants.APP_SUBMIT_TIME, String.valueOf(submitTime));
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertAMStartedEvent() {
+ long startTime = random.nextLong();
+ AMStartedEvent event = new AMStartedEvent(applicationAttemptId, startTime, user);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.AM_STARTED, startTime, EntityTypes.TEZ_APPLICATION,
+ applicationAttemptId, user, 0);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertContainerLaunchedEvent() {
+ long launchTime = random.nextLong();
+ ContainerLaunchedEvent event = new ContainerLaunchedEvent(containerId, launchTime,
+ applicationAttemptId);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.CONTAINER_LAUNCHED, launchTime, EntityTypes.TEZ_APPLICATION,
+ applicationAttemptId, null, 1);
+ assertEventData(proto, ATSConstants.CONTAINER_ID, containerId.toString());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertContainerStoppedEvent() {
+ long stopTime = random.nextLong();
+ int exitStatus = random.nextInt();
+ ContainerStoppedEvent event = new ContainerStoppedEvent(containerId, stopTime, exitStatus,
+ applicationAttemptId);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.CONTAINER_STOPPED, stopTime, EntityTypes.TEZ_APPLICATION,
+ applicationAttemptId, null, 3);
+ assertEventData(proto, ATSConstants.CONTAINER_ID, containerId.toString());
+ assertEventData(proto, ATSConstants.EXIT_STATUS, String.valueOf(exitStatus));
+ assertEventData(proto, ATSConstants.FINISH_TIME, String.valueOf(stopTime));
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGStartedEvent() {
+ long startTime = random.nextLong();
+ String dagName = "testDagName";
+ DAGStartedEvent event = new DAGStartedEvent(tezDAGID, startTime, user, dagName);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_STARTED, startTime, EntityTypes.TEZ_DAG_ID, null,
+ user, 2);
+ assertEventData(proto, ATSConstants.DAG_NAME, dagName);
+ assertEventData(proto, ATSConstants.STATUS, DAGState.RUNNING.name());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGSubmittedEvent() {
+ long submitTime = random.nextLong();
+
+ final String queueName = "TEST_DAG_SUBMITTED";
+ DAGSubmittedEvent event = new DAGSubmittedEvent(tezDAGID, submitTime, dagPlan,
+ applicationAttemptId, null, user, null, containerLogs, queueName);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_SUBMITTED, submitTime, EntityTypes.TEZ_DAG_ID,
+ applicationAttemptId, user, 8);
+
+ assertEventData(proto, ATSConstants.DAG_NAME, dagPlan.getName());
+ assertEventData(proto, ATSConstants.DAG_QUEUE_NAME, event.getQueueName());
+ assertEventData(proto, ATSConstants.DAG_AM_WEB_SERVICE_VERSION, AMWebController.VERSION);
+ assertEventData(proto, ATSConstants.IN_PROGRESS_LOGS_URL + "_"
+ + applicationAttemptId.getAttemptId(), containerLogs);
+ assertEventData(proto, ATSConstants.CALLER_CONTEXT_ID,
+ dagPlan.getCallerContext().getCallerId());
+ assertEventData(proto, ATSConstants.CALLER_CONTEXT_TYPE,
+ dagPlan.getCallerContext().getCallerType());
+ assertEventData(proto, ATSConstants.CALLER_CONTEXT, dagPlan.getCallerContext().getContext());
+ assertEventData(proto, ATSConstants.DAG_PLAN, null);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertTaskAttemptFinishedEvent() {
+ String vertexName = "testVertex";
+ long creationTime = random.nextLong();
+ long startTime = creationTime + 1000;
+ long allocationTime = creationTime + 1001;
+ long finishTime = startTime + 1002;
+ TaskAttemptState state = TaskAttemptState
+ .values()[random.nextInt(TaskAttemptState.values().length)];
+ TaskAttemptTerminationCause error = TaskAttemptTerminationCause
+ .values()[random.nextInt(TaskAttemptTerminationCause.values().length)];
+ String diagnostics = "random diagnostics message";
+ TezCounters counters = new TezCounters();
+ long lastDataEventTime = finishTime - 1;
+ List<DataEventDependencyInfo> events = Lists.newArrayList();
+ events.add(new DataEventDependencyInfo(lastDataEventTime, tezTaskAttemptID));
+ events.add(new DataEventDependencyInfo(lastDataEventTime, tezTaskAttemptID));
+
+ TaskAttemptFinishedEvent event = new TaskAttemptFinishedEvent(tezTaskAttemptID, vertexName,
+ startTime, finishTime, state, TaskFailureType.FATAL, error, diagnostics, counters, events,
+ null, creationTime, tezTaskAttemptID, allocationTime, containerId, nodeId, "inProgressURL",
+ "logsURL", "nodeHttpAddress");
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.TASK_ATTEMPT_FINISHED, finishTime,
+ EntityTypes.TEZ_DAG_ID, null, null, 16);
+
+ assertEventData(proto, ATSConstants.STATUS, state.name());
+ assertEventData(proto, ATSConstants.CREATION_CAUSAL_ATTEMPT, tezTaskAttemptID.toString());
+ assertEventData(proto, ATSConstants.CREATION_TIME, String.valueOf(creationTime));
+ assertEventData(proto, ATSConstants.ALLOCATION_TIME, String.valueOf(allocationTime));
+ assertEventData(proto, ATSConstants.START_TIME, String.valueOf(startTime));
+ assertEventData(proto, ATSConstants.TIME_TAKEN, String.valueOf(finishTime - startTime));
+ assertEventData(proto, ATSConstants.TASK_FAILURE_TYPE, TaskFailureType.FATAL.name());
+ assertEventData(proto, ATSConstants.TASK_ATTEMPT_ERROR_ENUM, error.name());
+ assertEventData(proto, ATSConstants.DIAGNOSTICS, diagnostics);
+ assertEventData(proto, ATSConstants.LAST_DATA_EVENTS, null);
+ assertEventData(proto, ATSConstants.COUNTERS, null);
+ assertEventData(proto, ATSConstants.IN_PROGRESS_LOGS_URL, "inProgressURL");
+ assertEventData(proto, ATSConstants.COMPLETED_LOGS_URL, "logsURL");
+ assertEventData(proto, ATSConstants.NODE_ID, nodeId.toString());
+ assertEventData(proto, ATSConstants.CONTAINER_ID, containerId.toString());
+ assertEventData(proto, ATSConstants.NODE_HTTP_ADDRESS, "nodeHttpAddress");
+
+ TaskAttemptFinishedEvent eventWithNullFailureType =
+ new TaskAttemptFinishedEvent(tezTaskAttemptID, vertexName,
+ startTime, finishTime, state, null, error, diagnostics, counters, events, null,
+ creationTime,
+ tezTaskAttemptID, allocationTime, containerId, nodeId, "inProgressURL", "logsURL",
+ "nodeHttpAddress");
+ proto = converter.convert(eventWithNullFailureType);
+ assertNoEventData(proto, ATSConstants.TASK_FAILURE_TYPE);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGInitializedEvent() {
+ long initTime = random.nextLong();
+
+ Map<String, TezVertexID> nameIdMap = new HashMap<String, TezVertexID>();
+ nameIdMap.put("foo", tezVertexID);
+
+ DAGInitializedEvent event = new DAGInitializedEvent(tezDAGID, initTime, "user", "dagName",
+ nameIdMap);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_INITIALIZED, initTime,
+ EntityTypes.TEZ_DAG_ID, null, user, 2);
+ assertEventData(proto, ATSConstants.DAG_NAME, "dagName");
+ assertEventData(proto, ATSConstants.VERTEX_NAME_ID_MAPPING, null);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGFinishedEvent() {
+ long finishTime = random.nextLong();
+ long startTime = random.nextLong();
+ Map<String, Integer> taskStats = new HashMap<String, Integer>();
+ taskStats.put("FOO", 100);
+ taskStats.put("BAR", 200);
+
+ DAGFinishedEvent event = new DAGFinishedEvent(tezDAGID, startTime, finishTime, DAGState.ERROR,
+ "diagnostics", null, user, dagPlan.getName(), taskStats, applicationAttemptId, dagPlan);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_FINISHED, finishTime,
+ EntityTypes.TEZ_DAG_ID, applicationAttemptId, user, 11);
+
+ assertEventData(proto, ATSConstants.DAG_NAME, dagPlan.getName());
+ assertEventData(proto, ATSConstants.STATUS, DAGState.ERROR.name());
+ assertEventData(proto, ATSConstants.CALLER_CONTEXT_ID,
+ dagPlan.getCallerContext().getCallerId());
+ assertEventData(proto, ATSConstants.CALLER_CONTEXT_TYPE,
+ dagPlan.getCallerContext().getCallerType());
+ assertEventData(proto, ATSConstants.START_TIME, String.valueOf(startTime));
+ assertEventData(proto, ATSConstants.TIME_TAKEN, String.valueOf(finishTime - startTime));
+ assertEventData(proto, ATSConstants.DIAGNOSTICS, "diagnostics");
+ assertEventData(proto, ATSConstants.COMPLETION_APPLICATION_ATTEMPT_ID,
+ applicationAttemptId.toString());
+ assertEventData(proto, "FOO", String.valueOf(100));
+ assertEventData(proto, "BAR", String.valueOf(200));
+ assertEventData(proto, ATSConstants.COUNTERS, null);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertVertexInitializedEvent() {
+ long initRequestedTime = random.nextLong();
+ long initedTime = random.nextLong();
+ int numTasks = random.nextInt();
+ VertexInitializedEvent event = new VertexInitializedEvent(tezVertexID, "v1", initRequestedTime,
+ initedTime, numTasks, "proc", null, null,
+ new ServicePluginInfo().setContainerLauncherName("abc")
+ .setTaskSchedulerName("def").setTaskCommunicatorName("ghi")
+ .setContainerLauncherClassName("abc1")
+ .setTaskSchedulerClassName("def1")
+ .setTaskCommunicatorClassName("ghi1"));
+
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.VERTEX_INITIALIZED, initedTime,
+ EntityTypes.TEZ_VERTEX_ID, null, null, 5);
+
+ assertEventData(proto, ATSConstants.VERTEX_NAME, "v1");
+ assertEventData(proto, ATSConstants.PROCESSOR_CLASS_NAME, "proc");
+ assertEventData(proto, ATSConstants.INIT_REQUESTED_TIME, String.valueOf(initRequestedTime));
+ assertEventData(proto, ATSConstants.NUM_TASKS, String.valueOf(numTasks));
+ assertEventData(proto, ATSConstants.SERVICE_PLUGIN, null);
+
+ /*
+ Assert.assertNotNull(timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN));
+ Assert.assertEquals("abc",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.CONTAINER_LAUNCHER_NAME));
+ Assert.assertEquals("def",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_SCHEDULER_NAME));
+ Assert.assertEquals("ghi",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_COMMUNICATOR_NAME));
+ Assert.assertEquals("abc1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.CONTAINER_LAUNCHER_CLASS_NAME));
+ Assert.assertEquals("def1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_SCHEDULER_CLASS_NAME));
+ Assert.assertEquals("ghi1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_COMMUNICATOR_CLASS_NAME));
+ */
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertVertexStartedEvent() {
+ long startRequestedTime = random.nextLong();
+ long startTime = random.nextLong();
+
+ VertexStartedEvent event = new VertexStartedEvent(tezVertexID, startRequestedTime, startTime);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.VERTEX_STARTED, startTime,
+ EntityTypes.TEZ_VERTEX_ID, null, null, 2);
+ assertEventData(proto, ATSConstants.START_REQUESTED_TIME, String.valueOf(startRequestedTime));
+ assertEventData(proto, ATSConstants.STATUS, VertexState.RUNNING.name());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertVertexFinishedEvent() {
+ String vertexName = "v1";
+ long initRequestedTime = random.nextLong();
+ long initedTime = random.nextLong();
+ long startRequestedTime = random.nextLong();
+ long startTime = random.nextLong();
+ long finishTime = random.nextLong();
+ Map<String, Integer> taskStats = new HashMap<String, Integer>();
+ taskStats.put("FOO", 100);
+ taskStats.put("BAR", 200);
+ VertexStats vertexStats = new VertexStats();
+
+ VertexFinishedEvent event = new VertexFinishedEvent(tezVertexID, vertexName, 1,
+ initRequestedTime, initedTime, startRequestedTime, startTime, finishTime,
+ VertexState.ERROR, "diagnostics", null, vertexStats, taskStats,
+ new ServicePluginInfo().setContainerLauncherName("abc")
+ .setTaskSchedulerName("def").setTaskCommunicatorName("ghi")
+ .setContainerLauncherClassName("abc1")
+ .setTaskSchedulerClassName("def1")
+ .setTaskCommunicatorClassName("ghi1"));
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.VERTEX_FINISHED, finishTime,
+ EntityTypes.TEZ_VERTEX_ID, null, null, 9);
+
+ assertEventData(proto, ATSConstants.VERTEX_NAME, vertexName);
+ assertEventData(proto, ATSConstants.STATUS, VertexState.ERROR.name());
+
+ assertEventData(proto, ATSConstants.TIME_TAKEN, String.valueOf(finishTime - startTime));
+ assertEventData(proto, ATSConstants.DIAGNOSTICS, "diagnostics");
+ assertEventData(proto, ATSConstants.COUNTERS, null);
+ assertEventData(proto, ATSConstants.STATS, null);
+ assertEventData(proto, "FOO", "100");
+ assertEventData(proto, "BAR", "200");
+
+ assertEventData(proto, ATSConstants.SERVICE_PLUGIN, null);
+ /*
+ Assert.assertEquals("abc",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.CONTAINER_LAUNCHER_NAME));
+ Assert.assertEquals("def",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_SCHEDULER_NAME));
+ Assert.assertEquals("ghi",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_COMMUNICATOR_NAME));
+ Assert.assertEquals("abc1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.CONTAINER_LAUNCHER_CLASS_NAME));
+ Assert.assertEquals("def1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_SCHEDULER_CLASS_NAME));
+ Assert.assertEquals("ghi1",
+ ((Map<String, Object>)timelineEntity.getOtherInfo().get(ATSConstants.SERVICE_PLUGIN)).get(
+ ATSConstants.TASK_COMMUNICATOR_CLASS_NAME));
+ */
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertTaskStartedEvent() {
+ long scheduleTime = random.nextLong();
+ long startTime = random.nextLong();
+ TaskStartedEvent event = new TaskStartedEvent(tezTaskID, "v1", scheduleTime, startTime);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.TASK_STARTED, startTime,
+ EntityTypes.TEZ_TASK_ID, null, null, 2);
+
+ assertEventData(proto, ATSConstants.SCHEDULED_TIME, String.valueOf(scheduleTime));
+ assertEventData(proto, ATSConstants.STATUS, TaskState.SCHEDULED.name());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertTaskAttemptStartedEvent() {
+ long startTime = random.nextLong();
+ TaskAttemptStartedEvent event = new TaskAttemptStartedEvent(tezTaskAttemptID, "v1",
+ startTime, containerId, nodeId, "inProgressURL", "logsURL", "nodeHttpAddress");
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.TASK_ATTEMPT_STARTED, startTime,
+ EntityTypes.TEZ_TASK_ATTEMPT_ID, null, null, 6);
+
+ assertEventData(proto, ATSConstants.STATUS, TaskAttemptState.RUNNING.name());
+ assertEventData(proto, ATSConstants.IN_PROGRESS_LOGS_URL, "inProgressURL");
+ assertEventData(proto, ATSConstants.COMPLETED_LOGS_URL, "logsURL");
+ assertEventData(proto, ATSConstants.NODE_ID, nodeId.toString());
+ assertEventData(proto, ATSConstants.CONTAINER_ID, containerId.toString());
+ assertEventData(proto, ATSConstants.NODE_HTTP_ADDRESS, "nodeHttpAddress");
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertTaskFinishedEvent() {
+ String vertexName = "testVertexName";
+ long startTime = random.nextLong();
+ long finishTime = random.nextLong();
+ TaskState state = TaskState.values()[random.nextInt(TaskState.values().length)];
+ String diagnostics = "diagnostics message";
+ TezCounters counters = new TezCounters();
+
+ TaskFinishedEvent event = new TaskFinishedEvent(tezTaskID, vertexName, startTime, finishTime,
+ tezTaskAttemptID, state, diagnostics, counters, 3);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.TASK_FINISHED, finishTime,
+ EntityTypes.TEZ_TASK_ID, null, null, 6);
+
+ assertEventData(proto, ATSConstants.STATUS, state.name());
+ assertEventData(proto, ATSConstants.TIME_TAKEN, String.valueOf(finishTime - startTime));
+ assertEventData(proto, ATSConstants.SUCCESSFUL_ATTEMPT_ID, tezTaskAttemptID.toString());
+ assertEventData(proto, ATSConstants.NUM_FAILED_TASKS_ATTEMPTS, "3");
+ assertEventData(proto, ATSConstants.DIAGNOSTICS, diagnostics);
+ assertEventData(proto, ATSConstants.COUNTERS, null);
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertVertexReconfigreDoneEvent() {
+ TezVertexID vId = tezVertexID;
+ Map<String, EdgeProperty> edgeMgrs =
+ new HashMap<String, EdgeProperty>();
+
+ edgeMgrs.put("a", EdgeProperty.create(EdgeManagerPluginDescriptor.create("a.class")
+ .setHistoryText("text"), DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ OutputDescriptor.create("Out"), InputDescriptor.create("In")));
+ VertexConfigurationDoneEvent event = new VertexConfigurationDoneEvent(vId, 0L, 1, null,
+ edgeMgrs, null, true);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.VERTEX_CONFIGURE_DONE, 0L,
+ EntityTypes.TEZ_VERTEX_ID, null, null, 2);
+ assertEventData(proto, ATSConstants.NUM_TASKS, "1");
+ assertEventData(proto, ATSConstants.UPDATED_EDGE_MANAGERS, null);
+
+ /*
+ Map<String, Object> updatedEdgeMgrs = (Map<String, Object>)
+ evt.getEventInfo().get(ATSConstants.UPDATED_EDGE_MANAGERS);
+ Assert.assertEquals(1, updatedEdgeMgrs.size());
+ Assert.assertTrue(updatedEdgeMgrs.containsKey("a"));
+ Map<String, Object> updatedEdgeMgr = (Map<String, Object>) updatedEdgeMgrs.get("a");
+
+ Assert.assertEquals(DataMovementType.CUSTOM.name(),
+ updatedEdgeMgr.get(DAGUtils.DATA_MOVEMENT_TYPE_KEY));
+ Assert.assertEquals("In", updatedEdgeMgr.get(DAGUtils.EDGE_DESTINATION_CLASS_KEY));
+ Assert.assertEquals("a.class", updatedEdgeMgr.get(DAGUtils.EDGE_MANAGER_CLASS_KEY));
+ */
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGRecoveredEvent() {
+ long recoverTime = random.nextLong();
+ DAGRecoveredEvent event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID,
+ dagPlan.getName(), user, recoverTime, containerLogs);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_RECOVERED, recoverTime,
+ EntityTypes.TEZ_DAG_ID, applicationAttemptId, user, 2);
+ assertEventData(proto, ATSConstants.IN_PROGRESS_LOGS_URL + "_"
+ + applicationAttemptId.getAttemptId(), containerLogs);
+ assertEventData(proto, ATSConstants.DAG_NAME, dagPlan.getName());
+ }
+
+ @Test(timeout = 5000)
+ public void testConvertDAGRecoveredEvent2() {
+ long recoverTime = random.nextLong();
+
+ DAGRecoveredEvent event = new DAGRecoveredEvent(applicationAttemptId, tezDAGID,
+ dagPlan.getName(), user, recoverTime, DAGState.ERROR, "mock reason", containerLogs);
+ HistoryEventProto proto = converter.convert(event);
+ assertCommon(proto, HistoryEventType.DAG_RECOVERED, recoverTime,
+ EntityTypes.TEZ_DAG_ID, applicationAttemptId, user, 4);
+ assertEventData(proto, ATSConstants.DAG_STATE, DAGState.ERROR.name());
+ assertEventData(proto, ATSConstants.RECOVERY_FAILURE_REASON, "mock reason");
+ assertEventData(proto, ATSConstants.IN_PROGRESS_LOGS_URL + "_"
+ + applicationAttemptId.getAttemptId(), containerLogs);
+ assertEventData(proto, ATSConstants.DAG_NAME, dagPlan.getName());
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestProtoHistoryLoggingService.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestProtoHistoryLoggingService.java b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestProtoHistoryLoggingService.java
new file mode 100644
index 0000000..4bd5d4e
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestProtoHistoryLoggingService.java
@@ -0,0 +1,195 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+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.NodeId;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.SystemClock;
+import org.apache.tez.common.VersionInfo;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.records.DAGProtos.DAGPlan;
+import org.apache.tez.dag.app.AppContext;
+import org.apache.tez.dag.app.dag.DAGState;
+import org.apache.tez.dag.history.DAGHistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.AppLaunchedEvent;
+import org.apache.tez.dag.history.events.DAGFinishedEvent;
+import org.apache.tez.dag.history.events.DAGSubmittedEvent;
+import org.apache.tez.dag.history.events.TaskAttemptStartedEvent;
+import org.apache.tez.dag.history.events.TaskStartedEvent;
+import org.apache.tez.dag.history.events.VertexStartedEvent;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.HistoryEventProto;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.ManifestEntryProto;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.apache.tez.hadoop.shim.HadoopShim;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestProtoHistoryLoggingService {
+ private static ApplicationId appId = ApplicationId.newInstance(1000l, 1);
+ private static ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(appId, 1);
+ private static String user = "TEST_USER";
+ private Clock clock;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Test
+ public void testService() throws Exception {
+ ProtoHistoryLoggingService service = createService();
+ service.start();
+ TezDAGID dagId = TezDAGID.getInstance(appId, 0);
+ List<HistoryEventProto> protos = new ArrayList<>();
+ for (DAGHistoryEvent event : makeHistoryEvents(dagId, service)) {
+ protos.add(new HistoryEventProtoConverter().convert(event.getHistoryEvent()));
+ service.handle(event);
+ }
+ service.stop();
+
+ TezProtoLoggers loggers = new TezProtoLoggers();
+ Assert.assertTrue(loggers.setup(service.getConfig(), clock));
+
+ // Verify dag events are logged.
+ DatePartitionedLogger<HistoryEventProto> dagLogger = loggers.getDagEventsLogger();
+ Path dagFilePath = dagLogger.getPathForDate(LocalDate.ofEpochDay(0), dagId.toString());
+ ProtoMessageReader<HistoryEventProto> reader = dagLogger.getReader(dagFilePath);
+ HistoryEventProto evt = reader.readEvent();
+ int ind = 1;
+ while (evt != null) {
+ Assert.assertEquals(protos.get(ind), evt);
+ ind++;
+ evt = reader.readEvent();
+ }
+ reader.close();
+
+ // Verify app events are logged.
+ DatePartitionedLogger<HistoryEventProto> appLogger = loggers.getAppEventsLogger();
+ Path appFilePath = appLogger.getPathForDate(LocalDate.ofEpochDay(0), attemptId.toString());
+ ProtoMessageReader<HistoryEventProto> appReader = appLogger.getReader(appFilePath);
+ long appOffset = appReader.getOffset();
+ Assert.assertEquals(protos.get(0), appReader.readEvent());
+ reader.close();
+
+ // Verify manifest events are logged.
+ DatePartitionedLogger<ManifestEntryProto> manifestLogger = loggers.getManifestEventsLogger();
+ Path manifestFilePath = manifestLogger.getPathForDate(
+ LocalDate.ofEpochDay(0), attemptId.toString());
+ ProtoMessageReader<ManifestEntryProto> reader2 = manifestLogger.getReader(manifestFilePath);
+ ManifestEntryProto manifest = reader2.readEvent();
+ Assert.assertEquals(appId.toString(), manifest.getAppId());
+ Assert.assertEquals(dagId.toString(), manifest.getDagId());
+ Assert.assertEquals(dagFilePath.toString(), manifest.getDagFilePath());
+ Assert.assertEquals(appFilePath.toString(), manifest.getAppFilePath());
+ Assert.assertEquals(appOffset, manifest.getAppLaunchedEventOffset());
+
+ // Verify offsets in manifest logger.
+ reader = dagLogger.getReader(new Path(manifest.getDagFilePath()));
+ reader.setOffset(manifest.getDagSubmittedEventOffset());
+ evt = reader.readEvent();
+ Assert.assertNotNull(evt);
+ Assert.assertEquals(HistoryEventType.DAG_SUBMITTED.name(), evt.getEventType());
+
+ reader.setOffset(manifest.getDagFinishedEventOffset());
+ evt = reader.readEvent();
+ Assert.assertNotNull(evt);
+ Assert.assertEquals(HistoryEventType.DAG_FINISHED.name(), evt.getEventType());
+
+ // Verify manifest file scanner.
+ DagManifesFileScanner scanner = new DagManifesFileScanner(manifestLogger);
+ Assert.assertEquals(manifest, scanner.getNext());
+ Assert.assertNull(scanner.getNext());
+ scanner.close();
+ }
+
+ private List<DAGHistoryEvent> makeHistoryEvents(TezDAGID dagId,
+ ProtoHistoryLoggingService service) {
+ List<DAGHistoryEvent> historyEvents = new ArrayList<>();
+ DAGPlan dagPlan = DAGPlan.newBuilder().setName("DAGPlanMock").build();
+
+ long time = System.currentTimeMillis();
+ Configuration conf = new Configuration(service.getConfig());
+ historyEvents.add(new DAGHistoryEvent(null, new AppLaunchedEvent(appId, time, time, user, conf,
+ new VersionInfo("component", "1.1.0", "rev1", "20120101", "git.apache.org") {})));
+ historyEvents.add(new DAGHistoryEvent(dagId, new DAGSubmittedEvent(dagId, time,
+ DAGPlan.getDefaultInstance(), attemptId, null, user, conf, null, "default")));
+ TezVertexID vertexID = TezVertexID.getInstance(dagId, 1);
+ historyEvents.add(new DAGHistoryEvent(dagId, new VertexStartedEvent(vertexID, time, time)));
+ TezTaskID tezTaskID = TezTaskID.getInstance(vertexID, 1);
+ historyEvents
+ .add(new DAGHistoryEvent(dagId, new TaskStartedEvent(tezTaskID, "test", time, time)));
+ historyEvents.add(new DAGHistoryEvent(dagId,
+ new TaskAttemptStartedEvent(TezTaskAttemptID.getInstance(tezTaskID, 1), "test", time,
+ ContainerId.newContainerId(attemptId, 1), NodeId.newInstance("localhost", 8765), null,
+ null, null)));
+ historyEvents.add(new DAGHistoryEvent(dagId, new DAGFinishedEvent(dagId, time, time,
+ DAGState.ERROR, "diagnostics", null, user, dagPlan.getName(),
+ new HashMap<String, Integer>(), attemptId, dagPlan)));
+ return historyEvents;
+ }
+
+ private static class FixedClock implements Clock {
+ final Clock clock = new SystemClock();
+ final long diff;
+
+ public FixedClock(long startTime) {
+ diff = clock.getTime() - startTime;
+ }
+
+ @Override
+ public long getTime() {
+ return clock.getTime() - diff;
+ }
+ }
+
+ private ProtoHistoryLoggingService createService() throws IOException {
+ ProtoHistoryLoggingService service = new ProtoHistoryLoggingService();
+ clock = new FixedClock(0); // Start time is always first day, easier to write tests.
+ AppContext appContext = mock(AppContext.class);
+ when(appContext.getApplicationID()).thenReturn(appId);
+ when(appContext.getApplicationAttemptId()).thenReturn(attemptId);
+ when(appContext.getUser()).thenReturn(user);
+ when(appContext.getHadoopShim()).thenReturn(new HadoopShim() {});
+ when(appContext.getClock()).thenReturn(clock);
+ service.setAppContext(appContext);
+ Configuration conf = new Configuration(false);
+ String basePath = tempFolder.newFolder().getAbsolutePath();
+ conf.set(TezConfiguration.TEZ_HISTORY_LOGGING_PROTO_BASE_DIR, basePath);
+ service.init(conf);
+ return service;
+ }
+}
[2/2] tez git commit: TEZ-3915. Create protobuf based history event
logger. (Harish Jaiprakash, reviewed by Gunther Hagleitner)
Posted by ha...@apache.org.
TEZ-3915. Create protobuf based history event logger. (Harish Jaiprakash, reviewed by Gunther Hagleitner)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/24b872a7
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/24b872a7
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/24b872a7
Branch: refs/heads/master
Commit: 24b872a7fc7a9bca11cbf4b5da80198386485547
Parents: 871ea80
Author: Harish JP <ha...@gmail.com>
Authored: Fri Apr 20 13:41:37 2018 +0530
Committer: Harish JP <ha...@gmail.com>
Committed: Fri Apr 20 13:41:37 2018 +0530
----------------------------------------------------------------------
pom.xml | 2 +-
.../apache/tez/dag/api/TezConfiguration.java | 21 +
tez-dist/pom.xml | 10 +
tez-plugins/pom.xml | 2 +
.../findbugs-exclude.xml | 50 ++
tez-plugins/tez-protobuf-history-plugin/pom.xml | 92 +++
.../logging/proto/DagManifesFileScanner.java | 150 ++++
.../logging/proto/DatePartitionedLogger.java | 167 +++++
.../proto/HistoryEventProtoConverter.java | 498 +++++++++++++
.../proto/ProtoHistoryLoggingService.java | 211 ++++++
.../logging/proto/ProtoMessageReader.java | 66 ++
.../logging/proto/ProtoMessageWritable.java | 101 +++
.../logging/proto/ProtoMessageWriter.java | 71 ++
.../history/logging/proto/TezProtoLoggers.java | 64 ++
.../src/main/proto/HistoryLogger.proto | 49 ++
.../proto/TestDagManifestFileScanner.java | 118 +++
.../proto/TestHistoryEventProtoConverter.java | 716 +++++++++++++++++++
.../proto/TestProtoHistoryLoggingService.java | 195 +++++
18 files changed, 2582 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 34240cf..16745f0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -37,7 +37,7 @@
<properties>
<maven.test.redirectTestOutputToFile>true</maven.test.redirectTestOutputToFile>
<clover.license>${user.home}/clover.license</clover.license>
- <hadoop.version>2.7.0</hadoop.version>
+ <hadoop.version>2.7.2</hadoop.version>
<jetty.version>9.3.22.v20171030</jetty.version>
<netty.version>3.6.2.Final</netty.version>
<pig.version>0.13.0</pig.version>
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java b/tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
index 6d3050d..243f278 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
@@ -1439,6 +1439,27 @@ public class TezConfiguration extends Configuration {
public static final int TEZ_SIMPLE_HISTORY_LOGGING_MAX_ERRORS_DEFAULT = 10;
/**
+ * String value. The base directory into which history data will be written when proto history
+ * logging service is used for {@link TezConfiguration#TEZ_HISTORY_LOGGING_SERVICE_CLASS}.
+ * If this is not set, then logging is disabled for ProtoHistoryLoggingService.
+ */
+ @ConfigurationScope(Scope.AM)
+ @ConfigurationProperty
+ public static final String TEZ_HISTORY_LOGGING_PROTO_BASE_DIR =
+ TEZ_PREFIX + "history.logging.proto-base-dir";
+
+ /**
+ * Long value. The amount of time in seconds to wait to ensure all events for a day is synced
+ * to disk. This should be maximum time variation b/w machines + maximum time to sync file
+ * content and metadata.
+ */
+ @ConfigurationScope(Scope.AM)
+ @ConfigurationProperty(type="long")
+ public static final String TEZ_HISTORY_LOGGING_PROTO_SYNC_WINDOWN_SECS =
+ TEZ_PREFIX + "history.logging.proto-sync-window-secs";
+ public static final long TEZ_HISTORY_LOGGING_PROTO_SYNC_WINDOWN_SECS_DEFAULT = 60L;
+
+ /**
* Int value. Time, in milliseconds, to wait while flushing YARN ATS data during shutdown.
* Expert level setting.
*/
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-dist/pom.xml
----------------------------------------------------------------------
diff --git a/tez-dist/pom.xml b/tez-dist/pom.xml
index 854a548..9447fe7 100644
--- a/tez-dist/pom.xml
+++ b/tez-dist/pom.xml
@@ -69,6 +69,11 @@
</dependency>
<dependency>
<groupId>org.apache.tez</groupId>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
<artifactId>hadoop-shim-2.7</artifactId>
<version>${project.version}</version>
</dependency>
@@ -97,6 +102,11 @@
</dependency>
<dependency>
<groupId>org.apache.tez</groupId>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
<artifactId>hadoop-shim-2.8</artifactId>
<version>${project.version}</version>
</dependency>
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/pom.xml
----------------------------------------------------------------------
diff --git a/tez-plugins/pom.xml b/tez-plugins/pom.xml
index fb0691a..f90cae7 100644
--- a/tez-plugins/pom.xml
+++ b/tez-plugins/pom.xml
@@ -35,6 +35,7 @@
</property>
</activation>
<modules>
+ <module>tez-protobuf-history-plugin</module>
<module>tez-yarn-timeline-history</module>
<module>tez-yarn-timeline-history-with-acls</module>
<module>tez-history-parser</module>
@@ -47,6 +48,7 @@
<activeByDefault>false</activeByDefault>
</activation>
<modules>
+ <module>tez-protobuf-history-plugin</module>
<module>tez-yarn-timeline-history</module>
<module>tez-yarn-timeline-history-with-acls</module>
<module>tez-yarn-timeline-cache-plugin</module>
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/findbugs-exclude.xml
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/findbugs-exclude.xml b/tez-plugins/tez-protobuf-history-plugin/findbugs-exclude.xml
new file mode 100644
index 0000000..c91265d
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/findbugs-exclude.xml
@@ -0,0 +1,50 @@
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+<FindBugsFilter>
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$KVPair"/>
+ <Field name="unknownFields"/>
+ <Bug pattern="SE_BAD_FIELD"/>
+ </Match>
+
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$.*Proto"/>
+ <Field name="unknownFields"/>
+ <Bug pattern="SE_BAD_FIELD"/>
+ </Match>
+
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$KVPair"/>
+ <Field name="PARSER"/>
+ <Bug pattern="MS_SHOULD_BE_FINAL"/>
+ </Match>
+
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$.*Proto"/>
+ <Field name="PARSER"/>
+ <Bug pattern="MS_SHOULD_BE_FINAL"/>
+ </Match>
+
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$KVPair\$Builder"/>
+ <Method name="maybeForceBuilderInitialization"/>
+ <Bug pattern="UCF_USELESS_CONTROL_FLOW"/>
+ </Match>
+
+ <Match>
+ <Class name="~org\.apache\.tez\.dag\.history\.logging\.proto\.HistoryLoggerProtos\$.*Proto\$Builder"/>
+ <Method name="maybeForceBuilderInitialization"/>
+ <Bug pattern="UCF_USELESS_CONTROL_FLOW"/>
+ </Match>
+</FindBugsFilter>
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/pom.xml
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/pom.xml b/tez-plugins/tez-protobuf-history-plugin/pom.xml
new file mode 100644
index 0000000..880aca9
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/pom.xml
@@ -0,0 +1,92 @@
+<!--
+ Licensed 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. See accompanying LICENSE file.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-plugins</artifactId>
+ <version>0.9.2-SNAPSHOT</version>
+ </parent>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-dag</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-maven-plugins</artifactId>
+ <executions>
+ <execution>
+ <id>compile-protoc</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>protoc</goal>
+ </goals>
+ <configuration>
+ <protocVersion>${protobuf.version}</protocVersion>
+ <protocCommand>${protoc.path}</protocCommand>
+ <imports>
+ <param>${basedir}/src/main/proto</param>
+ </imports>
+ <source>
+ <directory>${basedir}/src/main/proto</directory>
+ <includes>
+ <include>HistoryLogger.proto</include>
+ </includes>
+ </source>
+ <output>${project.build.directory}/generated-sources/java</output>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DagManifesFileScanner.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DagManifesFileScanner.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DagManifesFileScanner.java
new file mode 100644
index 0000000..c8ea02f
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DagManifesFileScanner.java
@@ -0,0 +1,150 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.fs.Path;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.ManifestEntryProto;
+import org.codehaus.jackson.map.ObjectMapper;
+
+/**
+ * Helper class to scan all the dag manifest files to get manifest entries.
+ */
+public class DagManifesFileScanner implements Closeable {
+ private static final int OFFSET_VERSION = 1;
+
+ private final ObjectMapper mapper = new ObjectMapper();
+ private final DatePartitionedLogger<ManifestEntryProto> manifestLogger;
+ private final long syncTime;
+
+ private String scanDir;
+ private Map<String, Long> offsets;
+ private List<Path> newFiles;
+
+ private ProtoMessageReader<ManifestEntryProto> reader;
+
+ public DagManifesFileScanner(DatePartitionedLogger<ManifestEntryProto> manifestLogger) {
+ this.manifestLogger = manifestLogger;
+ this.syncTime = manifestLogger.getConfig().getLong(
+ TezConfiguration.TEZ_HISTORY_LOGGING_PROTO_SYNC_WINDOWN_SECS,
+ TezConfiguration.TEZ_HISTORY_LOGGING_PROTO_SYNC_WINDOWN_SECS_DEFAULT);
+ this.setOffset(LocalDate.ofEpochDay(0));
+ }
+
+ // All public to simplify json conversion.
+ public static class DagManifestOffset {
+ public int version;
+ public String scanDir;
+ public Map<String, Long> offsets;
+ }
+
+ public void setOffset(String offset) {
+ try {
+ DagManifestOffset dagOffset = mapper.readValue(offset, DagManifestOffset.class);
+ if (dagOffset.version != OFFSET_VERSION) {
+ throw new IllegalArgumentException("Version mismatch: " + dagOffset.version);
+ }
+ this.scanDir = dagOffset.scanDir;
+ this.offsets = dagOffset.offsets;
+ this.newFiles = new ArrayList<>();
+ } catch (IOException e) {
+ throw new IllegalArgumentException("Invalid offset", e);
+ }
+ }
+
+ public void setOffset(LocalDate date) {
+ this.scanDir = manifestLogger.getDirForDate(date);
+ this.offsets = new HashMap<>();
+ this.newFiles = new ArrayList<>();
+ }
+
+ public String getOffset() {
+ try {
+ DagManifestOffset offset = new DagManifestOffset();
+ offset.version = OFFSET_VERSION;
+ offset.scanDir = scanDir;
+ offset.offsets = offsets;
+ return mapper.writeValueAsString(offset);
+ } catch (IOException e) {
+ throw new RuntimeException("Unexpected exception while converting to json.", e);
+ }
+ }
+
+ public ManifestEntryProto getNext() throws IOException {
+ while (true) {
+ if (reader != null) {
+ ManifestEntryProto evt = reader.readEvent();
+ if (evt != null) {
+ offsets.put(reader.getFilePath().getName(), reader.getOffset());
+ return evt;
+ } else {
+ IOUtils.closeQuietly(reader);
+ reader = null;
+ }
+ }
+ if (!newFiles.isEmpty()) {
+ this.reader = manifestLogger.getReader(newFiles.remove(0));
+ } else {
+ if (!loadMore()) {
+ return null;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null) {
+ reader.close();
+ reader = null;
+ }
+ }
+
+ private boolean loadMore() throws IOException {
+ newFiles = manifestLogger.scanForChangedFiles(scanDir, offsets);
+ while (newFiles.isEmpty()) {
+ LocalDateTime utcNow = manifestLogger.getNow();
+ if (utcNow.getHour() * 3600 + utcNow.getMinute() * 60 + utcNow.getSecond() < syncTime) {
+ // We are in the delay window for today, do not advance date if we are moving from
+ // yesterday.
+ String yesterDir = manifestLogger.getDirForDate(utcNow.toLocalDate().minusDays(1));
+ if (yesterDir.equals(scanDir)) {
+ return false;
+ }
+ }
+ String nextDir = manifestLogger.getNextDirectory(scanDir);
+ if (nextDir == null) {
+ return false;
+ }
+ scanDir = nextDir;
+ offsets = new HashMap<>();
+ newFiles = manifestLogger.scanForChangedFiles(scanDir, offsets);
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
new file mode 100644
index 0000000..8f89b2e
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
@@ -0,0 +1,167 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.IOException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.yarn.util.Clock;
+
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+/**
+ * Class to create proto reader and writer for a date partitioned directory structure.
+ *
+ * @param <T> The proto message type.
+ */
+public class DatePartitionedLogger<T extends MessageLite> {
+ // Everyone has permission to write, but with sticky set so that delete is restricted.
+ // This is required, since the path is same for all users and everyone writes into it.
+ private static final FsPermission DIR_PERMISSION = FsPermission.createImmutable((short)01777);
+
+ private final Parser<T> parser;
+ private final Path basePath;
+ private final Configuration conf;
+ private final Clock clock;
+ private final FileSystem fileSystem;
+
+ public DatePartitionedLogger(Parser<T> parser, Path baseDir, Configuration conf, Clock clock)
+ throws IOException {
+ this.conf = conf;
+ this.clock = clock;
+ this.parser = parser;
+ this.fileSystem = baseDir.getFileSystem(conf);
+ if (!fileSystem.exists(baseDir)) {
+ fileSystem.mkdirs(baseDir);
+ fileSystem.setPermission(baseDir, DIR_PERMISSION);
+ }
+ this.basePath = fileSystem.resolvePath(baseDir);
+ }
+
+ /**
+ * Creates a writer for the given fileName, with date as today.
+ */
+ public ProtoMessageWriter<T> getWriter(String fileName) throws IOException {
+ Path filePath = getPathForDate(getNow().toLocalDate(), fileName);
+ return new ProtoMessageWriter<>(conf, filePath, parser);
+ }
+
+ /**
+ * Creates a reader for the given filePath, no validation is done.
+ */
+ public ProtoMessageReader<T> getReader(Path filePath) throws IOException {
+ return new ProtoMessageReader<>(conf, filePath, parser);
+ }
+
+ /**
+ * Create a path for the given date and fileName. This can be used to create a reader.
+ */
+ public Path getPathForDate(LocalDate date, String fileName) throws IOException {
+ Path path = new Path(basePath, getDirForDate(date));
+ if (!fileSystem.exists(path)) {
+ fileSystem.mkdirs(path);
+ fileSystem.setPermission(path, DIR_PERMISSION);
+ }
+ return new Path(path, fileName);
+ }
+
+ /**
+ * Extract the date from the directory name, this should be a directory created by this class.
+ */
+ public LocalDate getDateFromDir(String dirName) {
+ if (!dirName.startsWith("date=")) {
+ throw new IllegalArgumentException("Invalid directory: "+ dirName);
+ }
+ return LocalDate.parse(dirName.substring(5), DateTimeFormatter.ISO_LOCAL_DATE);
+ }
+
+ /**
+ * Returns the directory name for a given date.
+ */
+ public String getDirForDate(LocalDate date) {
+ return "date=" + DateTimeFormatter.ISO_LOCAL_DATE.format(date);
+ }
+
+ /**
+ * Find next available directory, after the given directory.
+ */
+ public String getNextDirectory(String currentDir) throws IOException {
+ // Fast check, if the next day directory exists return it.
+ String nextDate = getDirForDate(getDateFromDir(currentDir).plusDays(1));
+ if (fileSystem.exists(new Path(basePath, nextDate))) {
+ return nextDate;
+ }
+ // Have to scan the directory to find min date greater than currentDir.
+ String dirName = null;
+ for (FileStatus status : fileSystem.listStatus(basePath)) {
+ String name = status.getPath().getName();
+ // String comparison is good enough, since its of form date=yyyy-MM-dd
+ if (name.compareTo(currentDir) > 0 && (dirName == null || name.compareTo(dirName) < 0)) {
+ dirName = name;
+ }
+ }
+ return dirName;
+ }
+
+ /**
+ * Returns new or changed files in the given directory. The offsets are used to find
+ * changed files.
+ */
+ public List<Path> scanForChangedFiles(String subDir, Map<String, Long> currentOffsets)
+ throws IOException {
+ Path dirPath = new Path(basePath, subDir);
+ List<Path> newFiles = new ArrayList<>();
+ if (!fileSystem.exists(dirPath)) {
+ return newFiles;
+ }
+ for (FileStatus status : fileSystem.listStatus(dirPath)) {
+ String fileName = status.getPath().getName();
+ Long offset = currentOffsets.get(fileName);
+ // If the offset was never added or offset < fileSize.
+ if (offset == null || offset < status.getLen()) {
+ newFiles.add(new Path(dirPath, fileName));
+ }
+ }
+ return newFiles;
+ }
+
+ /**
+ * Returns the current time, using the underlying clock in UTC time.
+ */
+ public LocalDateTime getNow() {
+ // Use UTC date to ensure reader date is same on all timezones.
+ return LocalDateTime.ofEpochSecond(clock.getTime() / 1000, 0, ZoneOffset.UTC);
+ }
+
+ public Configuration getConfig() {
+ return conf;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/HistoryEventProtoConverter.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/HistoryEventProtoConverter.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/HistoryEventProtoConverter.java
new file mode 100644
index 0000000..44dccb6
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/HistoryEventProtoConverter.java
@@ -0,0 +1,498 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.TreeMap;
+
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.tez.common.ATSConstants;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.TezUncheckedException;
+import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.api.records.DAGProtos.CallerContextProto;
+import org.apache.tez.dag.app.web.AMWebController;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.events.AMLaunchedEvent;
+import org.apache.tez.dag.history.events.AMStartedEvent;
+import org.apache.tez.dag.history.events.AppLaunchedEvent;
+import org.apache.tez.dag.history.events.ContainerLaunchedEvent;
+import org.apache.tez.dag.history.events.ContainerStoppedEvent;
+import org.apache.tez.dag.history.events.DAGFinishedEvent;
+import org.apache.tez.dag.history.events.DAGInitializedEvent;
+import org.apache.tez.dag.history.events.DAGRecoveredEvent;
+import org.apache.tez.dag.history.events.DAGStartedEvent;
+import org.apache.tez.dag.history.events.DAGSubmittedEvent;
+import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
+import org.apache.tez.dag.history.events.TaskAttemptStartedEvent;
+import org.apache.tez.dag.history.events.TaskFinishedEvent;
+import org.apache.tez.dag.history.events.TaskStartedEvent;
+import org.apache.tez.dag.history.events.VertexConfigurationDoneEvent;
+import org.apache.tez.dag.history.events.VertexFinishedEvent;
+import org.apache.tez.dag.history.events.VertexInitializedEvent;
+import org.apache.tez.dag.history.events.VertexStartedEvent;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.HistoryEventProto;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.KVPair;
+import org.apache.tez.dag.history.utils.DAGUtils;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Convert history event into HistoryEventProto message.
+ */
+public class HistoryEventProtoConverter {
+ private static final Logger log =
+ LoggerFactory.getLogger(HistoryEventProtoConverter.class);
+
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ /**
+ * Convert a given history event to HistoryEventProto message.
+ */
+ public HistoryEventProto convert(HistoryEvent historyEvent) {
+ validateEvent(historyEvent);
+ switch (historyEvent.getEventType()) {
+ case APP_LAUNCHED:
+ return convertAppLaunchedEvent((AppLaunchedEvent) historyEvent);
+ case AM_LAUNCHED:
+ return convertAMLaunchedEvent((AMLaunchedEvent) historyEvent);
+ case AM_STARTED:
+ return convertAMStartedEvent((AMStartedEvent) historyEvent);
+ case CONTAINER_LAUNCHED:
+ return convertContainerLaunchedEvent((ContainerLaunchedEvent) historyEvent);
+ case CONTAINER_STOPPED:
+ return convertContainerStoppedEvent((ContainerStoppedEvent) historyEvent);
+ case DAG_SUBMITTED:
+ return convertDAGSubmittedEvent((DAGSubmittedEvent) historyEvent);
+ case DAG_INITIALIZED:
+ return convertDAGInitializedEvent((DAGInitializedEvent) historyEvent);
+ case DAG_STARTED:
+ return convertDAGStartedEvent((DAGStartedEvent) historyEvent);
+ case DAG_FINISHED:
+ return convertDAGFinishedEvent((DAGFinishedEvent) historyEvent);
+ case VERTEX_INITIALIZED:
+ return convertVertexInitializedEvent((VertexInitializedEvent) historyEvent);
+ case VERTEX_STARTED:
+ return convertVertexStartedEvent((VertexStartedEvent) historyEvent);
+ case VERTEX_FINISHED:
+ return convertVertexFinishedEvent((VertexFinishedEvent) historyEvent);
+ case TASK_STARTED:
+ return convertTaskStartedEvent((TaskStartedEvent) historyEvent);
+ case TASK_FINISHED:
+ return convertTaskFinishedEvent((TaskFinishedEvent) historyEvent);
+ case TASK_ATTEMPT_STARTED:
+ return convertTaskAttemptStartedEvent((TaskAttemptStartedEvent) historyEvent);
+ case TASK_ATTEMPT_FINISHED:
+ return convertTaskAttemptFinishedEvent((TaskAttemptFinishedEvent) historyEvent);
+ case VERTEX_CONFIGURE_DONE:
+ return convertVertexReconfigureDoneEvent((VertexConfigurationDoneEvent) historyEvent);
+ case DAG_RECOVERED:
+ return convertDAGRecoveredEvent((DAGRecoveredEvent) historyEvent);
+ case VERTEX_COMMIT_STARTED:
+ case VERTEX_GROUP_COMMIT_STARTED:
+ case VERTEX_GROUP_COMMIT_FINISHED:
+ case DAG_COMMIT_STARTED:
+ case DAG_KILL_REQUEST:
+ throw new UnsupportedOperationException("Invalid Event, does not support history, eventType="
+ + historyEvent.getEventType());
+ // Do not add default, if a new event type is added, we'll get a warning for the
+ // switch.
+ }
+ throw new UnsupportedOperationException(
+ "Unhandled Event, eventType=" + historyEvent.getEventType());
+ }
+
+ private void validateEvent(HistoryEvent event) {
+ if (!event.isHistoryEvent()) {
+ throw new UnsupportedOperationException(
+ "Invalid Event, does not support history" + ", eventType=" + event.getEventType());
+ }
+ }
+
+ private HistoryEventProto.Builder makeBuilderForEvent(HistoryEvent event, long time,
+ TezDAGID dagId, ApplicationId appId, ApplicationAttemptId appAttemptId, TezVertexID vertexId,
+ TezTaskID taskId, TezTaskAttemptID taskAttemptId, String user) {
+ HistoryEventProto.Builder builder = HistoryEventProto.newBuilder();
+ builder.setEventType(event.getEventType().name());
+ builder.setEventTime(time);
+ if (taskAttemptId != null) {
+ builder.setTaskAttemptId(taskAttemptId.toString());
+ taskId = taskAttemptId.getTaskID();
+ }
+ if (taskId != null) {
+ builder.setTaskId(taskId.toString());
+ vertexId = taskId.getVertexID();
+ }
+ if (vertexId != null) {
+ builder.setVertexId(vertexId.toString());
+ dagId = vertexId.getDAGId();
+ }
+ if (dagId != null) {
+ builder.setDagId(dagId.toString());
+ if (appId == null) {
+ appId = dagId.getApplicationId();
+ }
+ }
+ if (appAttemptId != null) {
+ builder.setAppAttemptId(appAttemptId.toString());
+ if (appId == null) {
+ appId = appAttemptId.getApplicationId();
+ }
+ }
+ if (appId != null) {
+ builder.setAppId(appId.toString());
+ }
+ if (user != null) {
+ builder.setUser(user);
+ }
+ return builder;
+ }
+
+ private void addEventData(HistoryEventProto.Builder builder, String key, String value) {
+ if (value == null) {
+ return;
+ }
+ builder.addEventData(KVPair.newBuilder().setKey(key).setValue(value));
+ }
+
+ private void addEventData(HistoryEventProto.Builder builder, String key, Number value) {
+ builder.addEventData(KVPair.newBuilder().setKey(key).setValue(value.toString()));
+ }
+
+ private void addEventData(HistoryEventProto.Builder builder, String key,
+ Map<String, Object> value) {
+ try {
+ builder.addEventData(
+ KVPair.newBuilder().setKey(key).setValue(mapper.writeValueAsString(value)));
+ } catch (IOException e) {
+ log.error("Error converting value for key {} to json: ", key, e);
+ }
+ }
+
+ private HistoryEventProto convertAppLaunchedEvent(AppLaunchedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getLaunchTime(), null,
+ event.getApplicationId(), null, null, null, null, event.getUser());
+ // This is ok as long as we do not modify the underlying map.
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ Map<String, Object> confMap = (Map)DAGUtils.convertConfigurationToATSMap(event.getConf());
+ addEventData(builder, ATSConstants.CONFIG, confMap);
+ if (event.getVersion() != null) {
+ addEventData(builder, ATSConstants.TEZ_VERSION,
+ DAGUtils.convertTezVersionToATSMap(event.getVersion()));
+ }
+ addEventData(builder, ATSConstants.DAG_AM_WEB_SERVICE_VERSION, AMWebController.VERSION);
+ return builder.build();
+ }
+
+ private HistoryEventProto convertAMLaunchedEvent(AMLaunchedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getLaunchTime(), null,
+ null, event.getApplicationAttemptId(), null, null, null, event.getUser());
+ addEventData(builder, ATSConstants.APP_SUBMIT_TIME, event.getAppSubmitTime());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertAMStartedEvent(AMStartedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStartTime(), null,
+ null, event.getApplicationAttemptId(), null, null, null, event.getUser());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertContainerLaunchedEvent(ContainerLaunchedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getLaunchTime(), null,
+ null, event.getApplicationAttemptId(), null, null, null, null);
+ addEventData(builder, ATSConstants.CONTAINER_ID, event.getContainerId().toString());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertContainerStoppedEvent(ContainerStoppedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStoppedTime(), null,
+ null, event.getApplicationAttemptId(), null, null, null, null);
+ addEventData(builder, ATSConstants.CONTAINER_ID, event.getContainerId().toString());
+ addEventData(builder, ATSConstants.EXIT_STATUS, event.getExitStatus());
+ addEventData(builder, ATSConstants.FINISH_TIME, event.getStoppedTime());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertDAGSubmittedEvent(DAGSubmittedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getSubmitTime(),
+ event.getDagID(), null, event.getApplicationAttemptId(), null, null, null,
+ event.getUser());
+ addEventData(builder, ATSConstants.DAG_NAME, event.getDAGName());
+ if (event.getDAGPlan().hasCallerContext() &&
+ event.getDAGPlan().getCallerContext().hasCallerId()) {
+ CallerContextProto callerContext = event.getDagPlan().getCallerContext();
+ addEventData(builder, ATSConstants.CALLER_CONTEXT_ID, callerContext.getCallerId());
+ addEventData(builder, ATSConstants.CALLER_CONTEXT_TYPE, callerContext.getCallerType());
+ addEventData(builder, ATSConstants.CALLER_CONTEXT, callerContext.getContext());
+ }
+ if (event.getQueueName() != null) {
+ addEventData(builder, ATSConstants.DAG_QUEUE_NAME, event.getQueueName());
+ }
+ addEventData(builder, ATSConstants.DAG_AM_WEB_SERVICE_VERSION, AMWebController.VERSION);
+ addEventData(builder, ATSConstants.IN_PROGRESS_LOGS_URL + "_" +
+ event.getApplicationAttemptId().getAttemptId(), event.getContainerLogs());
+ try {
+ addEventData(builder, ATSConstants.DAG_PLAN,
+ DAGUtils.convertDAGPlanToATSMap(event.getDAGPlan()));
+ } catch (IOException e) {
+ throw new TezUncheckedException(e);
+ }
+ return builder.build();
+ }
+
+ private HistoryEventProto convertDAGInitializedEvent(DAGInitializedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getInitTime(),
+ event.getDagID(), null, null, null, null, null, event.getUser());
+ addEventData(builder, ATSConstants.DAG_NAME, event.getDagName());
+
+ if (event.getVertexNameIDMap() != null) {
+ Map<String, Object> nameIdStrMap = new TreeMap<String, Object>();
+ for (Entry<String, TezVertexID> entry : event.getVertexNameIDMap().entrySet()) {
+ nameIdStrMap.put(entry.getKey(), entry.getValue().toString());
+ }
+ addEventData(builder, ATSConstants.VERTEX_NAME_ID_MAPPING, nameIdStrMap);
+ }
+ return builder.build();
+ }
+
+ private HistoryEventProto convertDAGStartedEvent(DAGStartedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStartTime(),
+ event.getDagID(), null, null, null, null, null, event.getUser());
+
+ addEventData(builder, ATSConstants.DAG_NAME, event.getDagName());
+ addEventData(builder, ATSConstants.STATUS, event.getDagState().name());
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertDAGFinishedEvent(DAGFinishedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getFinishTime(),
+ event.getDagID(), null, event.getApplicationAttemptId(), null, null, null,
+ event.getUser());
+ addEventData(builder, ATSConstants.DAG_NAME, event.getDagName());
+ if (event.getDAGPlan().hasCallerContext()) {
+ if (event.getDAGPlan().getCallerContext().hasCallerType()) {
+ addEventData(builder, ATSConstants.CALLER_CONTEXT_TYPE,
+ event.getDAGPlan().getCallerContext().getCallerType());
+ }
+ if (event.getDAGPlan().getCallerContext().hasCallerId()) {
+ addEventData(builder, ATSConstants.CALLER_CONTEXT_ID,
+ event.getDAGPlan().getCallerContext().getCallerId());
+ }
+ }
+ addEventData(builder, ATSConstants.START_TIME, event.getStartTime());
+ addEventData(builder, ATSConstants.TIME_TAKEN, (event.getFinishTime() - event.getStartTime()));
+ addEventData(builder, ATSConstants.STATUS, event.getState().name());
+ addEventData(builder, ATSConstants.DIAGNOSTICS, event.getDiagnostics());
+ addEventData(builder, ATSConstants.COMPLETION_APPLICATION_ATTEMPT_ID,
+ event.getApplicationAttemptId().toString());
+ addEventData(builder, ATSConstants.COUNTERS,
+ DAGUtils.convertCountersToATSMap(event.getTezCounters()));
+ Map<String, Integer> dagTaskStats = event.getDagTaskStats();
+ if (dagTaskStats != null) {
+ for (Entry<String, Integer> entry : dagTaskStats.entrySet()) {
+ addEventData(builder, entry.getKey(), entry.getValue());
+ }
+ }
+ return builder.build();
+ }
+
+ private HistoryEventProto convertTaskAttemptStartedEvent(TaskAttemptStartedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStartTime(),
+ null, null, null, null, null, event.getTaskAttemptID(), null);
+ if (event.getInProgressLogsUrl() != null) {
+ addEventData(builder, ATSConstants.IN_PROGRESS_LOGS_URL, event.getInProgressLogsUrl());
+ }
+ if (event.getCompletedLogsUrl() != null) {
+ addEventData(builder, ATSConstants.COMPLETED_LOGS_URL, event.getCompletedLogsUrl());
+ }
+ addEventData(builder, ATSConstants.NODE_ID, event.getNodeId().toString());
+ addEventData(builder, ATSConstants.NODE_HTTP_ADDRESS, event.getNodeHttpAddress());
+ addEventData(builder, ATSConstants.CONTAINER_ID, event.getContainerId().toString());
+ addEventData(builder, ATSConstants.STATUS, TaskAttemptState.RUNNING.name());
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertTaskAttemptFinishedEvent(TaskAttemptFinishedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getFinishTime(),
+ null, null, null, null, null, event.getTaskAttemptID(), null);
+
+ addEventData(builder, ATSConstants.STATUS, event.getState().name());
+ if (event.getTaskFailureType() != null) {
+ addEventData(builder, ATSConstants.TASK_FAILURE_TYPE, event.getTaskFailureType().name());
+ }
+
+ addEventData(builder, ATSConstants.CREATION_TIME, event.getCreationTime());
+ addEventData(builder, ATSConstants.ALLOCATION_TIME, event.getAllocationTime());
+ addEventData(builder, ATSConstants.START_TIME, event.getStartTime());
+ if (event.getCreationCausalTA() != null) {
+ addEventData(builder, ATSConstants.CREATION_CAUSAL_ATTEMPT,
+ event.getCreationCausalTA().toString());
+ }
+ addEventData(builder, ATSConstants.TIME_TAKEN, (event.getFinishTime() - event.getStartTime()));
+ if (event.getTaskAttemptError() != null) {
+ addEventData(builder, ATSConstants.TASK_ATTEMPT_ERROR_ENUM,
+ event.getTaskAttemptError().name());
+ }
+ addEventData(builder, ATSConstants.DIAGNOSTICS, event.getDiagnostics());
+ addEventData(builder, ATSConstants.COUNTERS,
+ DAGUtils.convertCountersToATSMap(event.getCounters()));
+ if (event.getDataEvents() != null && !event.getDataEvents().isEmpty()) {
+ addEventData(builder, ATSConstants.LAST_DATA_EVENTS,
+ DAGUtils.convertDataEventDependecyInfoToATS(event.getDataEvents()));
+ }
+ if (event.getNodeId() != null) {
+ addEventData(builder, ATSConstants.NODE_ID, event.getNodeId().toString());
+ }
+ if (event.getContainerId() != null) {
+ addEventData(builder, ATSConstants.CONTAINER_ID, event.getContainerId().toString());
+ }
+ if (event.getInProgressLogsUrl() != null) {
+ addEventData(builder, ATSConstants.IN_PROGRESS_LOGS_URL, event.getInProgressLogsUrl());
+ }
+ if (event.getCompletedLogsUrl() != null) {
+ addEventData(builder, ATSConstants.COMPLETED_LOGS_URL, event.getCompletedLogsUrl());
+ }
+ if (event.getNodeHttpAddress() != null) {
+ addEventData(builder, ATSConstants.NODE_HTTP_ADDRESS, event.getNodeHttpAddress());
+ }
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertTaskFinishedEvent(TaskFinishedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getFinishTime(),
+ null, null, null, null, event.getTaskID(), null, null);
+
+ addEventData(builder, ATSConstants.TIME_TAKEN, (event.getFinishTime() - event.getStartTime()));
+ addEventData(builder, ATSConstants.STATUS, event.getState().name());
+ addEventData(builder, ATSConstants.NUM_FAILED_TASKS_ATTEMPTS, event.getNumFailedAttempts());
+ if (event.getSuccessfulAttemptID() != null) {
+ addEventData(builder, ATSConstants.SUCCESSFUL_ATTEMPT_ID,
+ event.getSuccessfulAttemptID().toString());
+ }
+
+ addEventData(builder, ATSConstants.DIAGNOSTICS, event.getDiagnostics());
+ addEventData(builder, ATSConstants.COUNTERS,
+ DAGUtils.convertCountersToATSMap(event.getTezCounters()));
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertTaskStartedEvent(TaskStartedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStartTime(),
+ null, null, null, null, event.getTaskID(), null, null);
+
+ addEventData(builder, ATSConstants.SCHEDULED_TIME, event.getScheduledTime());
+ addEventData(builder, ATSConstants.STATUS, event.getState().name());
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertVertexFinishedEvent(VertexFinishedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getFinishTime(),
+ null, null, null, event.getVertexID(), null, null, null);
+
+ addEventData(builder, ATSConstants.STATUS, event.getState().name());
+ addEventData(builder, ATSConstants.VERTEX_NAME, event.getVertexName());
+ addEventData(builder, ATSConstants.TIME_TAKEN, (event.getFinishTime() - event.getStartTime()));
+ addEventData(builder, ATSConstants.DIAGNOSTICS, event.getDiagnostics());
+ addEventData(builder, ATSConstants.COUNTERS,
+ DAGUtils.convertCountersToATSMap(event.getTezCounters()));
+ addEventData(builder, ATSConstants.STATS,
+ DAGUtils.convertVertexStatsToATSMap(event.getVertexStats()));
+ if (event.getServicePluginInfo() != null) {
+ addEventData(builder, ATSConstants.SERVICE_PLUGIN,
+ DAGUtils.convertServicePluginToATSMap(event.getServicePluginInfo()));
+ }
+
+ final Map<String, Integer> vertexTaskStats = event.getVertexTaskStats();
+ if (vertexTaskStats != null) {
+ for (Entry<String, Integer> entry : vertexTaskStats.entrySet()) {
+ addEventData(builder, entry.getKey(), entry.getValue());
+ }
+ }
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertVertexInitializedEvent(VertexInitializedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getInitedTime(),
+ null, null, null, event.getVertexID(), null, null, null);
+ addEventData(builder, ATSConstants.VERTEX_NAME, event.getVertexName());
+ addEventData(builder, ATSConstants.INIT_REQUESTED_TIME, event.getInitRequestedTime());
+ addEventData(builder, ATSConstants.NUM_TASKS, event.getNumTasks());
+ addEventData(builder, ATSConstants.PROCESSOR_CLASS_NAME, event.getProcessorName());
+ if (event.getServicePluginInfo() != null) {
+ addEventData(builder, ATSConstants.SERVICE_PLUGIN,
+ DAGUtils.convertServicePluginToATSMap(event.getServicePluginInfo()));
+ }
+
+ return builder.build();
+ }
+
+ private HistoryEventProto convertVertexStartedEvent(VertexStartedEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getStartTime(),
+ null, null, null, event.getVertexID(), null, null, null);
+ addEventData(builder, ATSConstants.START_REQUESTED_TIME, event.getStartRequestedTime());
+ addEventData(builder, ATSConstants.STATUS, event.getVertexState().name());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertVertexReconfigureDoneEvent(VertexConfigurationDoneEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getReconfigureDoneTime(),
+ null, null, null, event.getVertexID(), null, null, null);
+ if (event.getSourceEdgeProperties() != null && !event.getSourceEdgeProperties().isEmpty()) {
+ Map<String, Object> updatedEdgeManagers = new HashMap<String, Object>();
+ for (Entry<String, EdgeProperty> entry : event.getSourceEdgeProperties().entrySet()) {
+ updatedEdgeManagers.put(entry.getKey(), DAGUtils.convertEdgeProperty(entry.getValue()));
+ }
+ addEventData(builder, ATSConstants.UPDATED_EDGE_MANAGERS, updatedEdgeManagers);
+ }
+ addEventData(builder, ATSConstants.NUM_TASKS, event.getNumTasks());
+ return builder.build();
+ }
+
+ private HistoryEventProto convertDAGRecoveredEvent(DAGRecoveredEvent event) {
+ HistoryEventProto.Builder builder = makeBuilderForEvent(event, event.getRecoveredTime(),
+ event.getDagID(), null, event.getApplicationAttemptId(), null, null, null,
+ event.getUser());
+ addEventData(builder, ATSConstants.DAG_NAME, event.getDagName());
+ if (event.getRecoveredDagState() != null) {
+ addEventData(builder, ATSConstants.DAG_STATE, event.getRecoveredDagState().name());
+ }
+ if (event.getRecoveryFailureReason() != null) {
+ addEventData(builder, ATSConstants.RECOVERY_FAILURE_REASON,
+ event.getRecoveryFailureReason());
+ }
+ addEventData(builder, ATSConstants.IN_PROGRESS_LOGS_URL + "_" +
+ event.getApplicationAttemptId().getAttemptId(), event.getContainerLogs());
+ return builder.build();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoHistoryLoggingService.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoHistoryLoggingService.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoHistoryLoggingService.java
new file mode 100644
index 0000000..60cbda5
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoHistoryLoggingService.java
@@ -0,0 +1,211 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.IOException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.history.DAGHistoryEvent;
+import org.apache.tez.dag.history.HistoryEvent;
+import org.apache.tez.dag.history.HistoryEventType;
+import org.apache.tez.dag.history.events.DAGFinishedEvent;
+import org.apache.tez.dag.history.logging.HistoryLoggingService;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.HistoryEventProto;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.ManifestEntryProto;
+import org.apache.tez.dag.records.TezDAGID;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Logging service to write history events serialized using protobuf into sequence files.
+ * This can be used as external tables in hive. Or the reader can be used independently to
+ * read the data from these files.
+ */
+public class ProtoHistoryLoggingService extends HistoryLoggingService {
+ private static final Logger LOG = LoggerFactory.getLogger(ProtoHistoryLoggingService.class);
+ private final HistoryEventProtoConverter converter =
+ new HistoryEventProtoConverter();
+ private boolean loggingDisabled = false;
+
+ private final LinkedBlockingQueue<DAGHistoryEvent> eventQueue =
+ new LinkedBlockingQueue<DAGHistoryEvent>(10000);
+ private Thread eventHandlingThread;
+ private final AtomicBoolean stopped = new AtomicBoolean(false);
+
+ private TezProtoLoggers loggers;
+ private ProtoMessageWriter<HistoryEventProto> appEventsWriter;
+ private ProtoMessageWriter<HistoryEventProto> dagEventsWriter;
+ private TezDAGID currentDagId;
+ private long dagSubmittedEventOffset = -1;
+
+ private String appEventsFile;
+ private long appLaunchedEventOffset;
+
+ public ProtoHistoryLoggingService() {
+ super(ProtoHistoryLoggingService.class.getName());
+ }
+
+ @Override
+ protected void serviceInit(Configuration conf) throws Exception {
+ LOG.info("Initing ProtoHistoryLoggingService");
+ setConfig(conf);
+ loggingDisabled = !conf.getBoolean(TezConfiguration.TEZ_AM_HISTORY_LOGGING_ENABLED,
+ TezConfiguration.TEZ_AM_HISTORY_LOGGING_ENABLED_DEFAULT);
+ LOG.info("Inited ProtoHistoryLoggingService");
+ }
+
+ @Override
+ protected void serviceStart() throws Exception {
+ LOG.info("Starting ProtoHistoryLoggingService");
+ if (!loggingDisabled) {
+ loggers = new TezProtoLoggers();
+ if (!loggers.setup(getConfig(), appContext.getClock())) {
+ LOG.warn("Log file location for ProtoHistoryLoggingService not specified, " +
+ "logging disabled");
+ loggingDisabled = true;
+ return;
+ }
+ appEventsWriter = loggers.getAppEventsLogger().getWriter(
+ appContext.getApplicationAttemptId().toString());
+ eventHandlingThread = new Thread(this::loop, "HistoryEventHandlingThread");
+ eventHandlingThread.start();
+ }
+ LOG.info("Started ProtoHistoryLoggingService");
+ }
+
+ @Override
+ protected void serviceStop() throws Exception {
+ LOG.info("Stopping ProtoHistoryLoggingService, eventQueueBacklog=" + eventQueue.size());
+ stopped.set(true);
+ eventHandlingThread.join();
+ IOUtils.closeQuietly(appEventsWriter);
+ IOUtils.closeQuietly(dagEventsWriter);
+ LOG.info("Stopped ProtoHistoryLoggingService");
+ }
+
+ @Override
+ public void handle(DAGHistoryEvent event) {
+ if (loggingDisabled || stopped.get()) {
+ return;
+ }
+ try {
+ eventQueue.add(event);
+ } catch (IllegalStateException e) {
+ LOG.error("Queue capacity filled up, ignoring event: " +
+ event.getHistoryEvent().getEventType());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Queue capacity filled up, ignoring event: {}", event.getHistoryEvent());
+ }
+ }
+ }
+
+ private void loop() {
+ // Keep looping while the service is not stopped.
+ // Drain any left over events after the service has been stopped.
+ while (!stopped.get() || !eventQueue.isEmpty()) {
+ DAGHistoryEvent evt = null;
+ try {
+ evt = eventQueue.poll(100, TimeUnit.MILLISECONDS);
+ if (evt != null) {
+ handleEvent(evt);
+ }
+ } catch (InterruptedException e) {
+ LOG.info("EventQueue poll interrupted, ignoring it.", e);
+ } catch (IOException e) {
+ TezDAGID dagid = evt.getDagID();
+ HistoryEventType type = evt.getHistoryEvent().getEventType();
+ // Retry is hard, because there are several places where this exception can happen
+ // the state will get messed up a lot.
+ LOG.error("Got exception while handling event {} for dag {}.", type, dagid, e);
+ }
+ }
+ }
+
+ private void handleEvent(DAGHistoryEvent event) throws IOException {
+ if (loggingDisabled) {
+ return;
+ }
+ HistoryEvent historyEvent = event.getHistoryEvent();
+ if (event.getDagID() == null) {
+ if (historyEvent.getEventType() == HistoryEventType.APP_LAUNCHED) {
+ appEventsFile = appEventsWriter.getPath().toString();
+ appLaunchedEventOffset = appEventsWriter.getOffset();
+ }
+ appEventsWriter.writeProto(converter.convert(historyEvent));
+ } else {
+ HistoryEventType type = historyEvent.getEventType();
+ TezDAGID dagId = event.getDagID();
+ if (type == HistoryEventType.DAG_FINISHED) {
+ finishCurrentDag((DAGFinishedEvent)historyEvent);
+ } else if (type == HistoryEventType.DAG_SUBMITTED) {
+ finishCurrentDag(null);
+ currentDagId = dagId;
+ dagEventsWriter = loggers.getDagEventsLogger().getWriter(dagId.toString());
+ dagSubmittedEventOffset = dagEventsWriter.getOffset();
+ dagEventsWriter.writeProto(converter.convert(historyEvent));
+ } else if (dagEventsWriter != null) {
+ dagEventsWriter.writeProto(converter.convert(historyEvent));
+ }
+ }
+ }
+
+ private void finishCurrentDag(DAGFinishedEvent event) throws IOException {
+ if (dagEventsWriter == null) {
+ return;
+ }
+ ProtoMessageWriter<ManifestEntryProto> writer = null;
+ try {
+ long finishEventOffset = -1;
+ if (event != null) {
+ finishEventOffset = dagEventsWriter.getOffset();
+ dagEventsWriter.writeProto(converter.convert(event));
+ }
+ // Do not cache this writer, it should be created at the time of writing
+ writer = loggers.getManifestEventsLogger()
+ .getWriter(appContext.getApplicationAttemptId().toString());
+ ManifestEntryProto.Builder entry = ManifestEntryProto.newBuilder()
+ .setDagId(currentDagId.toString())
+ .setAppId(currentDagId.getApplicationId().toString())
+ .setDagSubmittedEventOffset(dagSubmittedEventOffset)
+ .setDagFinishedEventOffset(finishEventOffset)
+ .setDagFilePath(dagEventsWriter.getPath().toString())
+ .setAppFilePath(appEventsFile)
+ .setAppLaunchedEventOffset(appLaunchedEventOffset)
+ .setWriteTime(System.currentTimeMillis());
+ if (event != null) {
+ entry.setDagId(event.getDagID().toString());
+ }
+ writer.writeProto(entry.build());
+ appEventsWriter.hflush();
+ } finally {
+ // On an error, cleanup everything this will ensure, we do not use one dag's writer
+ // into another dag.
+ IOUtils.closeQuietly(dagEventsWriter);
+ IOUtils.closeQuietly(writer);
+ dagEventsWriter = null;
+ currentDagId = null;
+ dagSubmittedEventOffset = -1;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
new file mode 100644
index 0000000..e5f5e6b
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
@@ -0,0 +1,66 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+public class ProtoMessageReader<T extends MessageLite> implements Closeable {
+ private final Path filePath;
+ private final SequenceFile.Reader reader;
+ private final ProtoMessageWritable<T> writable;
+
+ ProtoMessageReader(Configuration conf, Path filePath, Parser<T> parser) throws IOException {
+ this.filePath = filePath;
+ this.reader = new SequenceFile.Reader(conf, SequenceFile.Reader.file(filePath));
+ this.writable = new ProtoMessageWritable<>(parser);
+ }
+
+ public Path getFilePath() {
+ return filePath;
+ }
+
+ public void setOffset(long offset) throws IOException {
+ reader.seek(offset);
+ }
+
+ public long getOffset() throws IOException {
+ return reader.getPosition();
+ }
+
+ public T readEvent() throws IOException {
+ if (!reader.next(NullWritable.get(), writable)) {
+ return null;
+ }
+ return writable.getMessage();
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
new file mode 100644
index 0000000..34e4701
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
@@ -0,0 +1,101 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import org.apache.hadoop.io.Writable;
+
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.CodedOutputStream;
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+public class ProtoMessageWritable<T extends MessageLite> implements Writable {
+ private T message;
+ private final Parser<T> parser;
+ private DataOutputStream dos;
+ private CodedOutputStream cos;
+ private DataInputStream din;
+ private CodedInputStream cin;
+
+ ProtoMessageWritable(Parser<T> parser) {
+ this.parser = parser;
+ }
+
+ public T getMessage() {
+ return message;
+ }
+
+ public void setMessage(T message) {
+ this.message = message;
+ }
+
+ private static class DataOutputStream extends OutputStream {
+ DataOutput out;
+ @Override
+ public void write(int b) throws IOException {
+ out.write(b);
+ }
+
+ @Override
+ public void write(byte b[], int off, int len) throws IOException {
+ out.write(b, off, len);
+ }
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ if (dos == null) {
+ dos = new DataOutputStream();
+ cos = CodedOutputStream.newInstance(dos);
+ }
+ dos.out = out;
+ cos.writeMessageNoTag(message);
+ cos.flush();
+ }
+
+ private static class DataInputStream extends InputStream {
+ DataInput in;
+ @Override
+ public int read() throws IOException {
+ try {
+ return in.readUnsignedByte();
+ } catch (EOFException e) {
+ return -1;
+ }
+ }
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ if (din == null) {
+ din = new DataInputStream();
+ cin = CodedInputStream.newInstance(din);
+ cin.setSizeLimit(Integer.MAX_VALUE);
+ }
+ din.in = in;
+ message = cin.readMessage(parser, null);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
new file mode 100644
index 0000000..ca9ba61
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
@@ -0,0 +1,71 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
+
+import com.google.protobuf.MessageLite;
+import com.google.protobuf.Parser;
+
+public class ProtoMessageWriter<T extends MessageLite> implements Closeable {
+ private final Path filePath;
+ private final SequenceFile.Writer writer;
+ private final ProtoMessageWritable<T> writable;
+
+ ProtoMessageWriter(Configuration conf, Path filePath, Parser<T> parser) throws IOException {
+ this.filePath = filePath;
+ this.writer = SequenceFile.createWriter(
+ conf,
+ SequenceFile.Writer.file(filePath),
+ SequenceFile.Writer.keyClass(NullWritable.class),
+ SequenceFile.Writer.valueClass(ProtoMessageWritable.class),
+ SequenceFile.Writer.appendIfExists(true),
+ SequenceFile.Writer.compression(CompressionType.RECORD));
+ this.writable = new ProtoMessageWritable<>(parser);
+ }
+
+ public Path getPath() {
+ return filePath;
+ }
+
+ public long getOffset() throws IOException {
+ return writer.getLength();
+ }
+
+ public void writeProto(T message) throws IOException {
+ writable.setMessage(message);
+ writer.append(NullWritable.get(), writable);
+ }
+
+ public void hflush() throws IOException {
+ writer.hflush();
+ }
+
+ @Override
+ public void close() throws IOException {
+ writer.close();
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/TezProtoLoggers.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/TezProtoLoggers.java b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/TezProtoLoggers.java
new file mode 100644
index 0000000..44390fc
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/java/org/apache/tez/dag/history/logging/proto/TezProtoLoggers.java
@@ -0,0 +1,64 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.HistoryEventProto;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.ManifestEntryProto;
+
+/**
+ * Helper class to create the logger for tez, we would use this to read the events outside
+ * tez and hence isolating the configuration and the paths in this.
+ */
+public class TezProtoLoggers {
+ private DatePartitionedLogger<HistoryEventProto> appEventsLogger;
+ private DatePartitionedLogger<HistoryEventProto> dagEventsLogger;
+ private DatePartitionedLogger<ManifestEntryProto> manifestEventsLogger;
+
+ public boolean setup(Configuration conf, Clock clock) throws IOException {
+ String logDir = conf.get(TezConfiguration.TEZ_HISTORY_LOGGING_PROTO_BASE_DIR);
+ if (logDir == null) {
+ return false;
+ }
+ appEventsLogger = new DatePartitionedLogger<>(HistoryEventProto.PARSER,
+ new Path(logDir, "app_data"), conf, clock);
+ dagEventsLogger = new DatePartitionedLogger<>(HistoryEventProto.PARSER,
+ new Path(logDir, "dag_data"), conf, clock);
+ manifestEventsLogger = new DatePartitionedLogger<>(ManifestEntryProto.PARSER,
+ new Path(logDir, "dag_meta"), conf, clock);
+ return true;
+ }
+
+ public DatePartitionedLogger<HistoryEventProto> getAppEventsLogger() {
+ return appEventsLogger;
+ }
+
+ public DatePartitionedLogger<HistoryEventProto> getDagEventsLogger() {
+ return dagEventsLogger;
+ }
+
+ public DatePartitionedLogger<ManifestEntryProto> getManifestEventsLogger() {
+ return manifestEventsLogger;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/main/proto/HistoryLogger.proto
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/main/proto/HistoryLogger.proto b/tez-plugins/tez-protobuf-history-plugin/src/main/proto/HistoryLogger.proto
new file mode 100644
index 0000000..a5bbe34
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/main/proto/HistoryLogger.proto
@@ -0,0 +1,49 @@
+/**
+ * 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.
+ */
+
+option java_package = "org.apache.tez.dag.history.logging.proto";
+option java_outer_classname = "HistoryLoggerProtos";
+
+message KVPair {
+ optional string key = 1;
+ optional string value = 2;
+}
+
+message HistoryEventProto {
+ optional string event_type = 1;
+ optional int64 event_time = 2;
+ optional string user = 3;
+ optional string app_id = 4;
+ optional string app_attempt_id = 5;
+ optional string dag_id = 6;
+ optional string vertex_id = 7;
+ optional string task_id = 8;
+ optional string task_attempt_id = 9;
+ repeated KVPair event_data = 10;
+}
+
+message ManifestEntryProto {
+ optional string dag_id = 1;
+ optional string app_id = 2;
+ optional int64 dag_submitted_event_offset = 3;
+ optional int64 dag_finished_event_offset = 4;
+ optional string dag_file_path = 5;
+ optional int64 writeTime = 6;
+ optional string app_file_path = 7;
+ optional int64 app_launched_event_offset = 8;
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/24b872a7/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestDagManifestFileScanner.java
----------------------------------------------------------------------
diff --git a/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestDagManifestFileScanner.java b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestDagManifestFileScanner.java
new file mode 100644
index 0000000..fcaa315
--- /dev/null
+++ b/tez-plugins/tez-protobuf-history-plugin/src/test/java/org/apache/tez/dag/history/logging/proto/TestDagManifestFileScanner.java
@@ -0,0 +1,118 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.tez.dag.history.logging.proto;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.history.logging.proto.HistoryLoggerProtos.ManifestEntryProto;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+public class TestDagManifestFileScanner {
+ private MockClock clock;
+ private DatePartitionedLogger<ManifestEntryProto> manifestLogger;
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Before
+ public void setupTest() throws Exception {
+ String basePath = tempFolder.newFolder().getAbsolutePath();
+ clock = new MockClock();
+ Configuration conf = new Configuration(false);
+ conf.set(TezConfiguration.TEZ_HISTORY_LOGGING_PROTO_BASE_DIR, basePath);
+ TezProtoLoggers loggers = new TezProtoLoggers();
+ loggers.setup(conf, clock);
+ manifestLogger = loggers.getManifestEventsLogger();
+ }
+
+ @Test
+ public void testNormal() throws Exception {
+ clock.setTime(0); // 0th day.
+ createManifestEvents(0, 8);
+ clock.setTime((24 * 60 * 60 + 1) * 1000); // 1 day 1 sec.
+ createManifestEvents(24 * 3600, 5);
+ DagManifesFileScanner scanner = new DagManifesFileScanner(manifestLogger);
+ int count = 0;
+ while (scanner.getNext() != null) {
+ ++count;
+ }
+ Assert.assertEquals(8, count);
+
+ // Save offset for later use.
+ String offset = scanner.getOffset();
+
+ // Move time outside the window, no changes and it will give more events.
+ clock.setTime((24 * 60 * 60 + 61) * 1000); // 1 day 61 sec.
+ count = 0;
+ while (scanner.getNext() != null) {
+ ++count;
+ }
+ Assert.assertEquals(5, count);
+
+ // Reset the offset
+ scanner.setOffset(offset);
+ count = 0;
+ while (scanner.getNext() != null) {
+ ++count;
+ }
+ Assert.assertEquals(5, count);
+
+ scanner.close();
+
+ // Not able to test append since the LocalFileSystem does not implement append.
+ }
+
+ private void createManifestEvents(long time, int numEvents) throws IOException {
+ for (int i = 0; i < numEvents; ++i) {
+ ApplicationId appId = ApplicationId.newInstance(1000l, i);
+ ManifestEntryProto proto = ManifestEntryProto.newBuilder()
+ .setAppId(appId.toString())
+ .setDagFilePath("dummy_dag_path_" + i)
+ .setDagSubmittedEventOffset(0)
+ .setDagFinishedEventOffset(1)
+ .setAppFilePath("dummp_app_path_" + i)
+ .setAppLaunchedEventOffset(2)
+ .setWriteTime(clock.getTime())
+ .build();
+ ProtoMessageWriter<ManifestEntryProto> writer = manifestLogger.getWriter(appId.toString());
+ writer.writeProto(proto);
+ writer.close();
+ }
+ }
+
+ private static class MockClock implements Clock {
+ private long time = 0;
+
+ void setTime(long time) {
+ this.time = time;
+ }
+
+ @Override
+ public long getTime() {
+ return time;
+ }
+ }
+}