JIRA: SAMZA-1733 Populating ListGauge metric using DiagnosticsAppender for exceptions

samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala

@@ -56,6 +56,7 @@ import org.apache.samza.task._
 import org.apache.samza.util.Util
 import org.apache.samza.util._
 import org.apache.samza.{SamzaContainerStatus, SamzaException}
+import org.apache.samza.diagnostics.DiagnosticsAppender;
 
 import scala.collection.JavaConverters._
 
@@ -757,6 +758,7 @@ class SamzaContainer(
       jmxServer = new JmxServer()
 
       startMetrics
+      startDiagnostics
       startAdmins
       startOffsetManager
       startLocalityManager
@@ -904,6 +906,19 @@ class SamzaContainer(
     })
   }
 
+  def startDiagnostics {
+    // TODO: where should this reside, MetricConfig? Log4jSystemConfig? or a new separate config?
+    val DIAGNOSTICS_APPENDER_ENABLE = "samza.diagnostics.enabled"
+    if (containerContext.config.getBoolean(DIAGNOSTICS_APPENDER_ENABLE, false)) {
+      import org.apache.log4j.Logger
+      val rootLogger = Logger.getRootLogger
+
+      info("Starting Diagnostics Appender.")
+      val diagnosticsAppender = new DiagnosticsAppender(this.metrics)
+      rootLogger.addAppender(diagnosticsAppender)
+    }
+  }
+
   def startOffsetManager {
     info("Registering task instances with offsets.")
 

samza-core/src/main/scala/org/apache/samza/container/SamzaContainerMetrics.scala

@@ -21,6 +21,7 @@ package org.apache.samza.container
 
 import java.util
 
+import org.apache.samza.diagnostics.DiagnosticsExceptionEvent
 import org.apache.samza.metrics.{Gauge, ReadableMetricsRegistry, MetricsRegistryMap, MetricsHelper}
 
 class SamzaContainerMetrics(
@@ -48,7 +49,7 @@ class SamzaContainerMetrics(
 
   val taskStoreRestorationMetrics: util.Map[TaskName, Gauge[Long]] = new util.HashMap[TaskName, Gauge[Long]]()
 
-  val exceptions = newListGauge[String]("exceptions")
+  val exceptions = newListGauge[DiagnosticsExceptionEvent]("exceptions")
 
   def addStoreRestorationGauge(taskName: TaskName, storeName: String) {
     taskStoreRestorationMetrics.put(taskName, newGauge("%s-%s-restore-time" format(taskName.toString, storeName), -1L))

samza-core/src/main/scala/org/apache/samza/diagnostics/DiagnosticsAppender.java

@@ -0,0 +1,81 @@
+/*
+ * 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.samza.diagnostics;
+
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.samza.container.SamzaContainerMetrics;
+import org.apache.samza.metrics.ListGauge;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Provides an in-memory appender that parses LoggingEvents to filter events relevant to diagnostics.
+ * Currently, filters exception related events and update an exception metric ({@link ListGauge}) in
+ * {@link SamzaContainerMetrics}.
+ *
+ * When used inconjunction with {@link org.apache.samza.metrics.reporter.MetricsSnapshotReporter} provides a
+ * stream of diagnostics-related events.
+ */
+public class DiagnosticsAppender extends AppenderSkeleton {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+  private final ListGauge<DiagnosticsExceptionEvent> samzaContainerExceptionMetric;
+
+  public DiagnosticsAppender(SamzaContainerMetrics samzaContainerMetrics) {
+    this.samzaContainerExceptionMetric = samzaContainerMetrics.exceptions();
+  }
+
+  @Override
+  protected void append(LoggingEvent loggingEvent) {
+
+    try {
+
+      // if an event with a non-null throwable is received => exception event
+      if (loggingEvent.getThrowableInformation() != null) {
+        DiagnosticsExceptionEvent diagnosticsExceptionEvent =
+            new DiagnosticsExceptionEvent(loggingEvent.timeStamp, loggingEvent.getThrowableInformation());
+
+        samzaContainerExceptionMetric.add(diagnosticsExceptionEvent);
+        logger.debug("Received DiagnosticsExceptionEvent " + diagnosticsExceptionEvent);
+      } else {
+        logger.debug("Received non-exception event with message " + loggingEvent.getMessage());
+      }
+    } catch (Exception e) {
+      // blanket catch of all exceptions so as to not impact any job
+      logger.error("Exception in logging event parsing", e);
+    }
+  }
+
+  @Override
+  public void close() {
+    // Do nothing.
+  }
+
+  /**
+   * Returns false since this appender requires no layout.
+   * @return false
+   */
+  @Override
+  public boolean requiresLayout() {
+    return false;
+  }
+}

samza-core/src/main/scala/org/apache/samza/diagnostics/DiagnosticsExceptionEvent.java

@@ -0,0 +1,46 @@
+/*
+ * 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.samza.diagnostics;
+
+import org.apache.log4j.spi.ThrowableInformation;
+
+
+/**
+ * This class encapsulates information related to an exception event that is useful for diagnostics.
+ * It used to define container, task, and other metrics as
+ * {@link org.apache.samza.metrics.ListGauge} of type {@link DiagnosticsExceptionEvent}.
+ */
+public class DiagnosticsExceptionEvent {
+
+  private long timestamp; // the timestamp associated with this exception
+  private ThrowableInformation throwableInformation;
+
+  public DiagnosticsExceptionEvent(long timestampMillis, ThrowableInformation throwableInformation) {
+    this.throwableInformation = throwableInformation;
+    this.timestamp = timestampMillis;
+  }
+
+  public long getTimestamp() {
+    return timestamp;
+  }
+
+  public ThrowableInformation getThrowableInformation() {
+    return this.throwableInformation;
+  }
+}