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

samza-core/src/main/scala/org/apache/samza/config/JobConfig.scala

@@ -88,6 +88,9 @@ object JobConfig {
   // across application restarts
   val JOB_LOGGED_STORE_BASE_DIR = "job.logged.store.base.dir"
 
+  // Enables diagnostic appender for logging exception events
+  val JOB_DIAGNOSTICS_ENABLED = "job.diagnostics.enabled"
+
   implicit def Config2Job(config: Config) = new JobConfig(config)
 
   /**
@@ -186,4 +189,6 @@ class JobConfig(config: Config) extends ScalaMapConfig(config) with Logging {
   def getNonLoggedStorePath = getOption(JobConfig.JOB_NON_LOGGED_STORE_BASE_DIR)
 
   def getLoggedStorePath = getOption(JobConfig.JOB_LOGGED_STORE_BASE_DIR)
+
+  def getDiagnosticsEnabled = { getBoolean(JobConfig.JOB_DIAGNOSTICS_ENABLED, false) }
 }

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

@@ -56,6 +56,8 @@ 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 org.apache.log4j.Logger
 
 import scala.collection.JavaConverters._
 
@@ -757,6 +759,7 @@ class SamzaContainer(
       jmxServer = new JmxServer()
 
       startMetrics
+      startDiagnostics
       startAdmins
       startOffsetManager
       startLocalityManager
@@ -904,6 +907,17 @@ class SamzaContainer(
     })
   }
 
+  def startDiagnostics {
+    if (containerContext.config.getDiagnosticsEnabled) {
+      val rootLogger = Logger.getRootLogger
+
+      if (rootLogger.getAppender(classOf[DiagnosticsAppender].getName) == null) {
+        info("Starting diagnostics appender.")
+        rootLogger.addAppender(new DiagnosticsAppender(this.metrics))
+      }
+    }
+  }
+
   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,82 @@
+/*
+ * 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 static final Logger LOG = LoggerFactory.getLogger(DiagnosticsAppender.class);
+  private final ListGauge<DiagnosticsExceptionEvent> samzaContainerExceptionMetric;
+
+  public DiagnosticsAppender(SamzaContainerMetrics samzaContainerMetrics) {
+    this.samzaContainerExceptionMetric = samzaContainerMetrics.exceptions();
+    this.setName(DiagnosticsAppender.class.getName());
+  }
+
+  @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().getThrowable(),
+                loggingEvent.getProperties());
+
+        samzaContainerExceptionMetric.add(diagnosticsExceptionEvent);
+        LOG.debug("Received DiagnosticsExceptionEvent " + diagnosticsExceptionEvent);
+      } else {
+        LOG.debug("Received non-exception event with message " + loggingEvent.getMessage());
+      }
+    } catch (Exception e) {
+      // blanket catch of all exceptions so as to not impact any job
+      LOG.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,86 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+
+/**
+ * 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 Class exceptionType; // store the exception type separately
+  private Throwable throwable;
+  private Map mdcMap;
+  // the MDC map associated with this exception, used to store/obtain any context associated with the throwable
+
+  public DiagnosticsExceptionEvent() {
+  }
+
+  public DiagnosticsExceptionEvent(long timestampMillis, Throwable throwable, Map mdcMap) {
+    this.throwable = throwable;
+    this.exceptionType = throwable.getClass();
+    this.timestamp = timestampMillis;
+    this.mdcMap = new HashMap(mdcMap);
+  }
+
+  public long getTimestamp() {
+    return timestamp;
+  }
+
+  public Throwable getThrowable() {
+    return this.throwable;
+  }
+
+  public Class getExceptionType() {
+    return this.exceptionType;
+  }
+
+  public Map getMdcMap() {
+    return mdcMap;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    DiagnosticsExceptionEvent that = (DiagnosticsExceptionEvent) o;
+
+    // Throwable provides no equals impl, so we assume message & stacktrace equality suffices
+    return timestamp == that.timestamp && this.exceptionType.equals(that.exceptionType) && mdcMap.equals(that.mdcMap)
+        && this.throwable.getMessage().equals(that.throwable.getMessage()) && Arrays.equals(
+        this.throwable.getStackTrace(), that.throwable.getStackTrace());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(timestamp, exceptionType, throwable, mdcMap);
+  }
+}

samza-core/src/main/scala/org/apache/samza/metrics/reporter/Metrics.scala

@@ -19,9 +19,11 @@
 
 package org.apache.samza.metrics.reporter
 
+import java.util
 import java.util.Collections
 import java.util.HashMap
 import java.util.Map
+
 import scala.collection.JavaConverters._
 
 object Metrics {
@@ -52,4 +54,8 @@ class Metrics(metrics: Map[String, Map[String, Object]]) {
   def get(group: String) = immutableMetrics.get(group)
 
   def getAsMap(): Map[String, Map[String, Object]] = Collections.unmodifiableMap(immutableMetrics)
+
+  def this() {
+    this(new util.HashMap[String, Map[String, Object]]())
+  }
 }

samza-core/src/main/scala/org/apache/samza/serializers/MetricsSnapshotSerdeV2.java

@@ -0,0 +1,75 @@
+/*
+ * 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.serializers;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.samza.metrics.reporter.MetricsSnapshot;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class MetricsSnapshotSerdeV2 implements Serde<MetricsSnapshot> {
+
+  private static final Logger LOG = LoggerFactory.getLogger(MetricsSnapshotSerdeV2.class);
+  private final ObjectMapper objectMapper;
+
+  public MetricsSnapshotSerdeV2() {
+    objectMapper = new ObjectMapper();
+    objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.OBJECT_AND_NON_CONCRETE);
+  }
+
+  @Override
+  public MetricsSnapshot fromBytes(byte[] bytes) {
+    try {
+      return MetricsSnapshot.fromMap(
+          objectMapper.readValue(bytes, new HashMap<String, Map<String, Object>>().getClass()));
+    } catch (IOException e) {
+      LOG.info("Exception while deserializing", e);
+    }
+    return null;
+  }
+
+  @Override
+  public byte[] toBytes(MetricsSnapshot metricsSnapshot) {
+    try {
+      return objectMapper.writeValueAsString(convertMap(metricsSnapshot.getAsMap())).getBytes("UTF-8");
+    } catch (IOException e) {
+      LOG.info("Exception while serializing", e);
+    }
+    return null;
+  }
+
+  /** Metrics returns an UnmodifiableMap.
+   * Unmodifiable maps should not be serialized with type, because UnmodifiableMap cannot be deserialized.
+   * So we convert to HashMap.
+   */
+  private HashMap convertMap(Map<String, Object> map) {
+    HashMap retVal = new HashMap(map);
+    for (Map.Entry<String, Object> entry : map.entrySet()) {
+      if (entry.getValue() instanceof Map) {
+        retVal.put(entry.getKey(), convertMap((Map<String, Object>) entry.getValue()));
+      }
+    }
+    return retVal;
+  }
+}

samza-core/src/main/scala/org/apache/samza/serializers/MetricsSnapshotSerdeV2Factory.java

@@ -0,0 +1,31 @@
+/*
+ * 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.serializers;
+
+import org.apache.samza.config.Config;
+import org.apache.samza.metrics.reporter.MetricsSnapshot;
+
+
+public class MetricsSnapshotSerdeV2Factory implements SerdeFactory<MetricsSnapshot> {
+  @Override
+  public Serde<MetricsSnapshot> getSerde(String name, Config config) {
+    return new MetricsSnapshotSerdeV2();
+  }
+}