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

rayman7718 · 2018-06-02T01:05:10Z

This PR shows how the ListGauge can be used to emit exceptions using a DiagnosticsAppender.

DiagnosticsAppender is enabled using a config (diagnostics.appender.enable)
DiagnosticsAppender adds exception-events to a listgauge which is a samza container metric
This ListGauge uses a time-and-count based eviction policy, so that exception-events are not emitted to Kafka(SnapshotReporter) forever.

…pshot reporter. Switched producer-shutdown logic to ensure metric-flush at shutdown-time

…lt policy to retain last N

Added exception metric (as string-guage), to be emitted using the snapshot reporter. Switched producer-shutdown logic to ensure metric-flush at shutdown-time Adding ListGauge and integration with SamzaContainerMetric Typifying ListGauge to ListGauge<T>, adding an eviction policy, default policy to retain last N

…listgauge

…mza container metric) using DiagnosticsAppender

…from ListGauge

… only top-level params

vjagadish1989

It'll be helpful to wrap up the original ListGauge PR first so that this is simpler to compare and review.

vjagadish1989 · 2018-06-04T23:40:38Z

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


+  def startDiagnostics {
+    // TODO: where should this reside, MetricConfig? Log4jSystemConfig? or a new separate config?
+    val DIAGNOSTICS_APPENDER_ENABLE = "diagnostics.appender.enable"


s/diagnostics.appender.enable/samza.diagnostics.enabled

vjagadish1989 · 2018-06-04T23:41:22Z

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

+    return timestamp;
+  }
+
+  public void setTimestamp(long timestamp) {


probably unnecessary to expose setters

Removing ValueInfo class and moving TimestampedValue to samza-api util Adding TimestampedValue.java in core

…listgauge

…istGauge

…ic into listgauge

…et-catch

This PR introduces a ListGauge type, A subsequent PR: #543 shows how this issued in conjunction with a diagnostics appender for error-tracking and dumpting to kafka via SnapshotReporter. Author: Ray Matharu <[email protected]> Reviewers: Jagadish <[email protected]>, Cameron Lee <[email protected]> Closes #541 from rayman7718/listgauge

…ticsappender

…her things

This PR introduces a ListGauge type, A subsequent PR: apache#543 shows how this issued in conjunction with a diagnostics appender for error-tracking and dumpting to kafka via SnapshotReporter. Author: Ray Matharu <[email protected]> Reviewers: Jagadish <[email protected]>, Cameron Lee <[email protected]> Closes apache#541 from rayman7718/listgauge

nickpan47

Fist pass, have a few questions. Will discuss face-to-face tomorrow.

nickpan47 · 2018-06-27T05:06:42Z

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

  }

+  def startDiagnostics {
+    // TODO: where should this reside, MetricConfig? Log4jSystemConfig? or a new separate config?


Suggestion: is this a per job configuration? If yes, it would probably be JobConfig and should be job.diagnostics.enabled.

nickpan47 · 2018-06-27T08:25:05Z

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

+ */
+public class DiagnosticsAppender extends AppenderSkeleton {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());


We have been using the code convention like: logger = LoggerFactory.getLog(DiagnosticsAppender.class.getName()).

nickpan47 · 2018-06-27T08:31:33Z

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

+      // if an event with a non-null throwable is received => exception event
+      if (loggingEvent.getThrowableInformation() != null) {
+        DiagnosticsExceptionEvent diagnosticsExceptionEvent =
+            new DiagnosticsExceptionEvent(loggingEvent.timeStamp, loggingEvent.getThrowableInformation());


So, how do we get the context info regarding to the exception? Through the stack trace? How do we add the logic component context like sub-query, task instance, container, and thread context?

nickpan47 · 2018-06-27T08:46:53Z

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

+public class DiagnosticsAppender extends AppenderSkeleton {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+  private final ListGauge<DiagnosticsExceptionEvent> samzaContainerExceptionMetric;


Question: how is this metric serialized and sent to diagnostic topic? If yes, shouldn't DiagnosticsExceptionEvent be serializable?

Existing StreamAppender implementation uses LoggingEventJsonSerde to serialize the loggingEvent. Would be better to use the same if it satisfies all the requirement here(or add fields ContainerContext and operatorContext if we need it).

shanthoosh · 2018-06-28T19:04:44Z

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

+ */
+package org.apache.samza.diagnostics;
+
+import org.apache.log4j.spi.ThrowableInformation;


Would be better not to depend on log4j specific objects in your model. If we choose to update to log4j2 in samza or switch to some other logging library(logback2), then this has to be changed(along with Appender implementation).

shanthoosh · 2018-06-28T19:11:41Z

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

+    // 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


nit: Better to move the imports to the top.

shanthoosh · 2018-06-28T19:19:11Z

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

+ */
+public class DiagnosticsAppender extends AppenderSkeleton {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());


It's a good practice to make the logger variable static and share it across all instances of the class.

shanthoosh · 2018-06-28T19:22:08Z

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

+ * When used inconjunction with {@link org.apache.samza.metrics.reporter.MetricsSnapshotReporter} provides a
+ * stream of diagnostics-related events.
+ */
+public class DiagnosticsAppender extends AppenderSkeleton {


Just curious. Do we need to have separate implementation for samza jobs using log4j2?

the same implementation can be adopted, as of log4j2.4.

shanthoosh · 2018-06-28T19:24:27Z

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

+public class DiagnosticsAppender extends AppenderSkeleton {
+
+  private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+  private final ListGauge<DiagnosticsExceptionEvent> samzaContainerExceptionMetric;


Existing StreamAppender implementation uses LoggingEventJsonSerde to serialize the loggingEvent. Would be better to use the same if it satisfies all the requirement here(or add fields ContainerContext and operatorContext if we need it).

shanthoosh · 2018-06-29T01:15:06Z

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

+
+      info("Starting Diagnostics Appender.")
+      val diagnosticsAppender = new DiagnosticsAppender(this.metrics)
+      rootLogger.addAppender(diagnosticsAppender)


In standalone, there're plans to run more than one StreamProcessor in a jvm(which translates to more than one SamzaContainer thread in jvm).
So it will be better to add diagnosticsAppender to rootLogger if it is not already added.

…aking exception event generic.

rayman7718 · 2018-07-10T00:33:14Z

Made the following changes:

Depend directly on using java objects rather than log4j objects in appender (done)
Check if logger already attached, only than attach appender to root logger (done)
JobConfig for samza.diagnostics.enabled (done)
SnapshotSerdeV2 to serialize Throwables with type and other relevant info for error tracking (done), with unit test.
Obtain/store an exception's context in appender

Container-level context is already present in MetricsHeaders.
Adding TaskContext to SamzaException didnt quite work because TaskContext doesn't apply to many components (like KafkaSysCons).
Instead, we use the MDC map associated with the exception. Context can then be added by the components to MDC map, e.g., topicName, espressoTableName, etc. LOg4j's LoggingEvent takes care of multi-threading and async-appender.
In cases, such as KafkaSysConsumer the poll is for a set of SSPs and the exception message (with the erroring topic name) is populated by KafkaConsumer (which Samza has no control over).

…iagnostics, with unittest

nickpan47

@rayman7718 thanks for the update! I have a few comments regarding the usage of log4j in Samza code base. Please check. Thanks!

nickpan47 · 2018-07-23T17:36:22Z

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

 import org.apache.samza.util._
 import org.apache.samza.{SamzaContainerStatus, SamzaException}
+import org.apache.samza.diagnostics.DiagnosticsAppender;
+import org.apache.log4j.Logger


We are not using log4j directly in samza-core. We always use slf4j libraries in all our source code, except for the samza-log4j module.

nickpan47 · 2018-07-23T17:43:41Z

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

+    if (containerContext.config.getDiagnosticsEnabled) {
+      val rootLogger = Logger.getRootLogger
+
+      if (rootLogger.getAppender(classOf[DiagnosticsAppender].getName) == null) {


It seems that in log4j2, the API to get appender dynamically also changed: https://logging.apache.org/log4j/2.x/log4j-core/apidocs/org/apache/logging/log4j/core/config/Configuration.html#getAppender(java.lang.String). Please check w/ @PawasChhokra to see what's the best way to make sure the code here is log4j version agnostic.

nickpan47 · 2018-07-23T17:44:19Z

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

+    if (containerContext.config.getDiagnosticsEnabled) {
+      val rootLogger = Logger.getRootLogger
+
+      if (rootLogger.getAppender(classOf[DiagnosticsAppender].getName) == null) {


Shouldn't this condition be (rootLogger.getAppender(classOf[DiagnosticsAppender].getName) != null)?

nickpan47 · 2018-07-23T17:47:13Z

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

 import java.util.Collections
 import java.util.HashMap
 import java.util.Map
+


nit: whitespace change should be avoided.

nickpan47 · 2018-07-23T17:50:32Z

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


  def getAsMap(): Map[String, Map[String, Object]] = Collections.unmodifiableMap(immutableMetrics)
+
+  def this() {


I did a code search and couldn't find any reference to this default constructor. What's the reason to add it?

nickpan47 · 2018-07-23T17:51:06Z

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


 package org.apache.samza.metrics.reporter

+import java.util


This seems to be redundant, if we remove the prefix to HashMap in line 59.

…n to instantiate

rayman7718 · 2018-07-24T17:15:53Z

Addressed the log4j dependency issue by using a config for the Appender class, this removes the use of log4j Logger from SamzaContainer
For log4j2, verified with Pawas that the getappender and addAppender apis are consistent.
Addressed the other Metrics comments.

nickpan47

@rayman7718 thanks a lot for the updates. Still have a few minor comments, but looks good overall.

nickpan47 · 2018-07-25T17:43:28Z

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

+
+  def getDiagnosticsEnabled = { getBoolean(JobConfig.JOB_DIAGNOSTICS_ENABLED, false) }
+
+  def getDiagnosticsAppenderClass = { getOption(JobConfig.DIAGNOSTICS_APPENDER_CLASS) }


I think that it would make sense to provide a default class name here as well, for log4j 1.2.x (since in open source, we still depend on log4j 1.2.x).

nickpan47 · 2018-07-25T17:49:35Z

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

+  def startDiagnostics {
+    if (containerContext.config.getDiagnosticsEnabled) {
+      info("Starting diagnostics.")
+      val diagnosticsAppender = Class.forName(containerContext.config.getDiagnosticsAppenderClass.get).


It would be nice to capture the class loading exceptions and log a line here, then throw them as ConfigException(msg, e).

nickpan47 · 2018-07-25T17:53:39Z

samza-log4j/src/main/java/org/apache/samza/logging/log4j/SimpleDiagnosticsAppender.java

+    this.setName(SimpleDiagnosticsAppender.class.getName());
+
+    // ensure appender is attached only once per JVM (regardless of #containers)
+    if (org.apache.log4j.Logger.getRootLogger().getAppender(SimpleDiagnosticsAppender.class.getName()) == null) {


Just realize: is this code section safe for multi-threaded environment? Ideally, we should make it threadsafe. If not, we better make 100% sure that the loading of this appender only happens in one single thread and make the usage documentation here very clear.

…nder instantiation, adding default value in jobConfig

rayman7718 · 2018-07-26T01:00:05Z

Added simple synchronization to make appender attachment thread safe (from constructor).
Added default value for appender class path in job config.
Adding error logging in samza-container for appender instantiation.

nickpan47

Thanks for the fixes! Feel free to check after fix the minor comments.

nickpan47 · 2018-07-26T21:41:45Z

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

+  def getDiagnosticsEnabled = { getBoolean(JobConfig.JOB_DIAGNOSTICS_ENABLED, false) }
+
+  def getDiagnosticsAppenderClass = {
+    getOrDefault(JobConfig.DIAGNOSTICS_APPENDER_CLASS, "org.apache.samza.logging.log4j.SimpleDiagnosticsAppender")


nit: define a static constant for the default value in the class.

nickpan47 · 2018-07-26T21:43:00Z

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

+      catch {
+        case e@(_: ClassNotFoundException | _: InstantiationException | _: InvocationTargetException) => {
+          error("Failed to instantiate diagnostic appender", e)
+          throw new ConfigException("Failed to instantiate diagnostic appender class specified in config", e)


nit: add the configured DiagnosticsAppenderClass name in the log and exception would make it easier for debugging.

rmatharu added 7 commits May 31, 2018 15:44

Added exception metric (as string-guage), to be emitted using the sna…

5ae1df9

…pshot reporter. Switched producer-shutdown logic to ensure metric-flush at shutdown-time

Adding ListGauge and integration with SamzaContainerMetric

5a3ef7b

Typifying ListGauge to ListGauge<T>, adding an eviction policy, defau…

75ef5d2

…lt policy to retain last N

Merge branch 'listgauge' of https://github.com/rayman7718/samza into …

d2922c5

…listgauge

Adding DiagnosticsAppender, Populating ListGauge exception metric (sa…

6602fa0

…mza container metric) using DiagnosticsAppender

Adding DiagnosticsExceptionEventEvictionPolicy to evict stale events …

054d507

…from ListGauge

rayman7718 mentioned this pull request Jun 2, 2018

SAMZA-1733 New metric type Listgauge #541

Closed

rmatharu added 4 commits June 4, 2018 13:03

Removing remove() and policy-configurability from ListGauge, exposing…

1f08eae

… only top-level params

Merge branch 'listgauge' into diagnosticsappender

b598b92

Renaming valueList to elements, and getValue to getValues

6f19ba0

Merge branch 'listgauge' into diagnosticsappender

c22e252

vjagadish1989 reviewed Jun 4, 2018

View reviewed changes

rmatharu added 17 commits June 4, 2018 17:28

Removing the interface for ListGaugeEvictionPolicy

4f08cd1

Merge branch 'listgauge' into diagnosticsappender

9f98c57

Removing ValueInfo class and moving TimestampedValue to samza-api util

906afeb

Removing ValueInfo class and moving TimestampedValue to samza-api util

bdc85d4

Removing ValueInfo class and moving TimestampedValue to samza-api util

5c1ed0a

Removing ValueInfo class and moving TimestampedValue to samza-api util Adding TimestampedValue.java in core

Merge branch 'listgauge' of https://github.com/rayman7718/samza into …

c1884e4

…listgauge

Merge branch 'listgauge' into diagnosticsappender

b0695b4

Renaming diagnostics.appender.enable to samza.diagnostics.enabled

508b6b3

Removing periodic pruning from Listgauge, pruning on add and get in L…

c7821ba

…istGauge

Updating testListGauge for new listgauge

9ba8e5b

Merge branch 'listgauge' into diagnosticsappender

64b1cb8

Complete removing evictionpolicy separation, merging all eviction log…

fd77c53

…ic into listgauge

Reverting superflous comment changes, indentation changes

cc4d881

Merge branch 'listgauge' into diagnosticsappender

172ab20

Minor changes

863d850

Merge branch 'listgauge' into diagnosticsappender

cac4ecb

Adding throwable cause messages and class names to exception event

925473c

Fixing minor bug around diagnostic-event-initialization, adding blank…

2ad0264

…et-catch

rmatharu added 3 commits June 21, 2018 14:40

Merge branch 'master' of https://github.com/apache/samza into diagnos…

60f281c

…ticsappender

Removing thread name

cd58669

Adding throwableInformation to DiagnosticsExceptionEvent, removing ot…

8a66d16

…her things

nickpan47 reviewed Jun 27, 2018

View reviewed changes

shanthoosh reviewed Jun 28, 2018

View reviewed changes

shanthoosh reviewed Jun 29, 2018

View reviewed changes

rmatharu added 2 commits June 29, 2018 16:36

Moving diagnostics.enabled to jobconfig, adding appender-add check, m…

48cbf3e

…aking exception event generic.

Adding MDCMap to DiagnosticExceptionEvent for storing context

346affd

rmatharu added 4 commits July 11, 2018 15:19

Adding default constructor to enable serializing and deserializing

7f7885b

Adding SnapshotSerdeV2 to enable serializing of ExceptionEvents for d…

92a29b4

…iagnostics, with unittest

Serializing exceptions as throwables, storing classtype separately.

e9f4004

Minor changes to ensure build

28c4fec

nickpan47 requested changes Jul 23, 2018

View reviewed changes

rmatharu added 3 commits July 23, 2018 16:03

Fixing dependencies in Metrics.scala

edbd200

Moving addAppender to DiagnosticsAppender, using config and reflectio…

f8ace28

…n to instantiate

efactoring useless code out

a4d11b0

nickpan47 requested changes Jul 25, 2018

View reviewed changes

Making appender-attachment thread safe, adding error logging for appe…

6e274e7

…nder instantiation, adding default value in jobConfig

nickpan47 approved these changes Jul 26, 2018

View reviewed changes

Adding constant to store class, adding class name to log message

7c727f7

rayman7718 changed the title ~~Populating ListGauge metric using DiagnosticsAppender for exceptions~~ JIRA: SAMZA-1733 Populating ListGauge metric using DiagnosticsAppender for exceptions Jul 26, 2018

asfgit closed this in e0ff4c5 Jul 27, 2018


		def getAsMap(): Map[String, Map[String, Object]] = Collections.unmodifiableMap(immutableMetrics)

		def this() {


		def getDiagnosticsEnabled = { getBoolean(JobConfig.JOB_DIAGNOSTICS_ENABLED, false) }

		def getDiagnosticsAppenderClass = { getOption(JobConfig.DIAGNOSTICS_APPENDER_CLASS) }

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

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

Uh oh!

Conversation

rayman7718 commented Jun 2, 2018

Uh oh!

vjagadish1989 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayman7718 commented Jul 10, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayman7718 commented Jul 24, 2018

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rayman7718 commented Jul 26, 2018

Uh oh!

nickpan47 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

rayman7718 commented Jul 10, 2018 •

edited

Loading