Add support for task reports, upload reports to deep storage

api/src/main/java/io/druid/tasklogs/NoopTaskLogs.java

@@ -24,6 +24,7 @@
 import io.druid.java.util.common.logger.Logger;
 
 import java.io.File;
+import java.io.IOException;
 
 public class NoopTaskLogs implements TaskLogs
 {
@@ -41,6 +42,12 @@ public void pushTaskLog(String taskid, File logFile)
     log.info("Not pushing logs for task: %s", taskid);
   }
 
+  @Override
+  public void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+    log.info("Not pushing reports for task: %s", taskid);
+  }
+
   @Override
   public void killAll()
   {

api/src/main/java/io/druid/tasklogs/TaskLogPusher.java

@@ -31,4 +31,8 @@
 public interface TaskLogPusher
 {
   void pushTaskLog(String taskid, File logFile) throws IOException;
+
+  default void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+  }
 }

api/src/main/java/io/druid/tasklogs/TaskLogStreamer.java

@@ -40,4 +40,9 @@ public interface TaskLogStreamer
    * @return input supplier for this log, if available from this provider
    */
   Optional<ByteSource> streamTaskLog(String taskid, long offset) throws IOException;
+
+  default Optional<ByteSource> streamTaskReports(final String taskid) throws IOException
+  {
+    return Optional.absent();
+  }
 }

extensions-contrib/azure-extensions/src/main/java/io/druid/storage/azure/AzureTaskLogs.java

@@ -54,7 +54,19 @@ public void pushTaskLog(final String taskid, final File logFile)
   {
     final String taskKey = getTaskLogKey(taskid);
     log.info("Pushing task log %s to: %s", logFile, taskKey);
+    pushTaskFile(taskid, logFile, taskKey);
+  }
+
+  @Override
+  public void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+    final String taskKey = getTaskReportsKey(taskid);
+    log.info("Pushing task reports %s to: %s", reportFile, taskKey);
+    pushTaskFile(taskid, reportFile, taskKey);
+  }
 
+  private void pushTaskFile(final String taskId, final File logFile, String taskKey)
+  {
     try {
       AzureUtils.retryAzureOperation(
           () -> {
@@ -71,9 +83,19 @@ public void pushTaskLog(final String taskid, final File logFile)
 
   @Override
   public Optional<ByteSource> streamTaskLog(final String taskid, final long offset) throws IOException
+  {
+    return streamTaskFile(taskid, offset, getTaskLogKey(taskid));
+  }
+
+  @Override
+  public Optional<ByteSource> streamTaskReports(String taskid) throws IOException
+  {
+    return streamTaskFile(taskid, 0, getTaskReportsKey(taskid));
+  }
+
+  private Optional<ByteSource> streamTaskFile(final String taskid, final long offset, String taskKey) throws IOException
   {
     final String container = config.getContainer();
-    final String taskKey = getTaskLogKey(taskid);
 
     try {
       if (!azureStorage.getBlobExists(container, taskKey)) {
@@ -116,12 +138,16 @@ public InputStream openStream() throws IOException
     }
   }
 
-
   private String getTaskLogKey(String taskid)
   {
     return StringUtils.format("%s/%s/log", config.getPrefix(), taskid);
   }
 
+  private String getTaskReportsKey(String taskid)
+  {
+    return StringUtils.format("%s/%s/report.json", config.getPrefix(), taskid);
+  }
+
   @Override
   public void killAll()
   {

extensions-contrib/google-extensions/src/main/java/io/druid/storage/google/GoogleTaskLogs.java

@@ -51,7 +51,19 @@ public void pushTaskLog(final String taskid, final File logFile) throws IOExcept
   {
     final String taskKey = getTaskLogKey(taskid);
     LOG.info("Pushing task log %s to: %s", logFile, taskKey);
+    pushTaskFile(taskid, logFile, taskKey);
+  }
+
+  @Override
+  public void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+    final String taskKey = getTaskReportKey(taskid);
+    LOG.info("Pushing task reports %s to: %s", reportFile, taskKey);
+    pushTaskFile(taskid, reportFile, taskKey);
+  }
 
+  private void pushTaskFile(final String taskid, final File logFile, final String taskKey) throws IOException
+  {
     FileInputStream fileSteam = new FileInputStream(logFile);
 
     InputStreamContent mediaContent = new InputStreamContent("text/plain", fileSteam);
@@ -64,7 +76,18 @@ public void pushTaskLog(final String taskid, final File logFile) throws IOExcept
   public Optional<ByteSource> streamTaskLog(final String taskid, final long offset) throws IOException
   {
     final String taskKey = getTaskLogKey(taskid);
+    return streamTaskFile(taskid, offset, taskKey);
+  }
 
+  @Override
+  public Optional<ByteSource> streamTaskReports(String taskid) throws IOException
+  {
+    final String taskKey = getTaskReportKey(taskid);
+    return streamTaskFile(taskid, 0, taskKey);
+  }
+
+  private Optional<ByteSource> streamTaskFile(final String taskid, final long offset, String taskKey) throws IOException
+  {
     try {
       if (!storage.exists(config.getBucket(), taskKey)) {
         return Optional.absent();
@@ -111,6 +134,11 @@ private String getTaskLogKey(String taskid)
     return config.getPrefix() + "/" + taskid.replaceAll(":", "_");
   }
 
+  private String getTaskReportKey(String taskid)
+  {
+    return config.getPrefix() + "/" + taskid.replaceAll(":", "_") + ".report.json";
+  }
+
   @Override
   public void killAll()
   {

extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/tasklog/HdfsTaskLogs.java

@@ -61,21 +61,46 @@ public void pushTaskLog(String taskId, File logFile) throws IOException
   {
     final Path path = getTaskLogFileFromId(taskId);
     log.info("Writing task log to: %s", path);
+    pushTaskFile(path, logFile);
+    log.info("Wrote task log to: %s", path);
+  }
+
+  @Override
+  public void pushTaskReports(String taskId, File reportFile) throws IOException
+  {
+    final Path path = getTaskReportsFileFromId(taskId);
+    log.info("Writing task reports to: %s", path);
+    pushTaskFile(path, reportFile);
+    log.info("Wrote task reports to: %s", path);
+  }
+
+  private void pushTaskFile(Path path, File logFile) throws IOException
+  {
     final FileSystem fs = path.getFileSystem(hadoopConfig);
     try (
         final InputStream in = new FileInputStream(logFile);
         final OutputStream out = fs.create(path, true)
     ) {
       ByteStreams.copy(in, out);
     }
-
-    log.info("Wrote task log to: %s", path);
   }
 
   @Override
   public Optional<ByteSource> streamTaskLog(final String taskId, final long offset) throws IOException
   {
     final Path path = getTaskLogFileFromId(taskId);
+    return streamTaskFile(path, offset);
+  }
+
+  @Override
+  public Optional<ByteSource> streamTaskReports(String taskId) throws IOException
+  {
+    final Path path = getTaskReportsFileFromId(taskId);
+    return streamTaskFile(path, 0);
+  }
+
+  private Optional<ByteSource> streamTaskFile(final Path path, final long offset) throws IOException
+  {
     final FileSystem fs = path.getFileSystem(hadoopConfig);
     if (fs.exists(path)) {
       return Optional.<ByteSource>of(
@@ -113,6 +138,15 @@ private Path getTaskLogFileFromId(String taskId)
     return new Path(mergePaths(config.getDirectory(), taskId.replaceAll(":", "_")));
   }
 
+  /**
+   * Due to https://issues.apache.org/jira/browse/HDFS-13 ":" are not allowed in
+   * path names. So we format paths differently for HDFS.
+   */
+  private Path getTaskReportsFileFromId(String taskId)
+  {
+    return new Path(mergePaths(config.getDirectory(), taskId.replaceAll(":", "_") + ".reports.json"));
+  }
+
   // some hadoop version Path.mergePaths does not exist
   private static String mergePaths(String path1, String path2)
   {

extensions-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java

@@ -904,6 +904,7 @@ public void onFailure(Throwable t)
       toolbox.getDataSegmentServerAnnouncer().unannounce();
     }
 
+    toolbox.getTaskReportFileWriter().write(null);
     return success();
   }
 
@@ -1272,6 +1273,7 @@ public String apply(DataSegment input)
       toolbox.getDataSegmentServerAnnouncer().unannounce();
     }
 
+    toolbox.getTaskReportFileWriter().write(null);
     return success();
   }
 

extensions-core/kafka-indexing-service/src/test/java/io/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -57,6 +57,7 @@
 import io.druid.indexing.common.actions.TaskActionToolbox;
 import io.druid.indexing.common.config.TaskConfig;
 import io.druid.indexing.common.config.TaskStorageConfig;
+import io.druid.indexing.common.task.NoopTestTaskFileWriter;
 import io.druid.indexing.common.task.Task;
 import io.druid.indexing.kafka.supervisor.KafkaSupervisor;
 import io.druid.indexing.kafka.test.TestBroker;
@@ -2032,7 +2033,8 @@ public List<StorageLocationConfig> getLocations()
         EasyMock.createNiceMock(DruidNodeAnnouncer.class),
         EasyMock.createNiceMock(DruidNode.class),
         new LookupNodeService("tier"),
-        new DataNodeService("tier", 1, ServerType.INDEXER_EXECUTOR, 0)
+        new DataNodeService("tier", 1, ServerType.INDEXER_EXECUTOR, 0),
+        new NoopTestTaskFileWriter()
     );
   }
 

extensions-core/s3-extensions/src/main/java/io/druid/storage/s3/S3TaskLogs.java

@@ -57,8 +57,19 @@ public S3TaskLogs(S3TaskLogsConfig config, AmazonS3 service)
   @Override
   public Optional<ByteSource> streamTaskLog(final String taskid, final long offset) throws IOException
   {
-    final String taskKey = getTaskLogKey(taskid);
+    final String taskKey = getTaskLogKey(taskid, "log");
+    return streamTaskFile(offset, taskKey);
+  }
+
+  @Override
+  public Optional<ByteSource> streamTaskReports(String taskid) throws IOException
+  {
+    final String taskKey = getTaskLogKey(taskid, "report.json");
+    return streamTaskFile(0, taskKey);
+  }
 
+  private Optional<ByteSource> streamTaskFile(final long offset, String taskKey) throws IOException
+  {
     try {
       final ObjectMetadata objectMetadata = service.getObjectMetadata(config.getS3Bucket(), taskKey);
 
@@ -107,9 +118,21 @@ public InputStream openStream() throws IOException
   @Override
   public void pushTaskLog(final String taskid, final File logFile) throws IOException
   {
-    final String taskKey = getTaskLogKey(taskid);
+    final String taskKey = getTaskLogKey(taskid, "log");
     log.info("Pushing task log %s to: %s", logFile, taskKey);
+    pushTaskFile(logFile, taskKey);
+  }
+
+  @Override
+  public void pushTaskReports(String taskid, File reportFile) throws IOException
+  {
+    final String taskKey = getTaskLogKey(taskid, "report.json");
+    log.info("Pushing task reports %s to: %s", reportFile, taskKey);
+    pushTaskFile(reportFile, taskKey);
+  }
 
+  private void pushTaskFile(final File logFile, String taskKey) throws IOException
+  {
     try {
       S3Utils.retryS3Operation(
           () -> {
@@ -124,9 +147,9 @@ public void pushTaskLog(final String taskid, final File logFile) throws IOExcept
     }
   }
 
-  private String getTaskLogKey(String taskid)
+  private String getTaskLogKey(String taskid, String filename)
   {
-    return StringUtils.format("%s/%s/log", config.getS3Prefix(), taskid);
+    return StringUtils.format("%s/%s/%s", config.getS3Prefix(), taskid, filename);
   }
 
   @Override

indexing-service/src/main/java/io/druid/indexing/common/TaskReport.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.indexing.common;
+
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.google.common.collect.Maps;
+
+import java.util.Map;
+
+/**
+ * TaskReport objects contain additional information about an indexing task, such as row statistics, errors, and
+ * published segments. They are kept in deep storage along with task logs.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
+@JsonSubTypes(value = {
+})
+public interface TaskReport
+{
+  String getTaskId();
+
+  String getReportKey();
+
+  /**
+   * @return A JSON-serializable Object that contains a TaskReport's information
+   */
+  Object getPayload();
+
+  static Map<String, TaskReport> buildTaskReports(TaskReport... taskReports)
+  {
+    Map<String, TaskReport> taskReportMap = Maps.newHashMap();
+    for (TaskReport taskReport : taskReports) {
+      taskReportMap.put(taskReport.getReportKey(), taskReport);
+    }
+    return taskReportMap;
+  }
+}