Monitoring

Monitoring Samza Applications

This section provides details on monitoring of Samza jobs, not to be confused with Samza Monitors (components of the Samza-REST service that provide cluster-wide monitoring capabilities).

Like any other production software, it is critical to monitor the health of our Samza jobs. Samza relies on metrics for monitoring and includes an extensible metrics library. While a few standard metrics are provided out-of-the-box, it is easy to define metrics specific to your application.

• A. Metrics Reporters
  • A.1 Reporting Metrics to JMX (JMX Reporter)
    • Enabling the JMX Reporter
    • Using the JMX Reporter
  • A.2 Reporting Metrics to Kafka (MetricsSnapshot Reporter)
    • Enabling the MetricsSnapshot Reporter
  • A.3 Creating a Custom MetricsReporter
• B. Metric Types in Samza
• C. Adding User-Defined Metrics
  • Low Level Task API
  • High Level Streams API
• D. Key Internal Samza Metrics
  • D.1 Vital Metrics
  • D.2 Store Metrics
  • D.3 Operator Metrics
• E. Metrics Reference Sheet

A. Metrics Reporters

Samza’s metrics library encapsulates the metrics collection and sampling logic. Metrics Reporters in Samza are responsible for emitting metrics to external services which may archive, process, visualize the metrics’ values, or trigger alerts based on them.

Samza includes default implementations for two such Metrics Reporters:

1. a) A JMXReporter (detailed below) which allows using standard JMX clients for probing containers to retrieve metrics encoded as JMX MBeans. Visualization tools such as Grafana could also be used to visualize this JMX data.

2. b) A MetricsSnapshot reporter (detailed below) which allows periodically publishing all metrics to Kafka. A downstream Samza job could then consume and publish these metrics to other metrics management systems such as Prometheus and Graphite.

Note that Samza allows multiple Metrics Reporters to be used simultaneously.

A.1 Reporting Metrics to JMX (JMX Reporter)

This reporter encodes all its internal and user-defined metrics as JMX MBeans and hosts a JMX MBean server. Standard JMX clients (such as JConsole, VisualVM) can thus be used to probe Samza’s containers and YARN-ApplicationMaster to retrieve these metrics’ values. JMX also provides additional profiling capabilities (e.g., for CPU and memory utilization), which are also enabled by this reporter.

Enabling the JMX Reporter

JMXReporter can be enabled by adding the following configuration.

#Define a Samza metrics reporter called "jxm", which publishes to JMX
metrics.reporter.jmx.class=org.apache.samza.metrics.reporter.JmxReporterFactory

# Use the jmx reporter (if using multiple reporters, separate them with commas)
metrics.reporters=jmx

Using the JMX Reporter

To connect to the JMX MBean server, first obtain the JMX Server URL and port, published in the container logs:

2018-08-14 11:30:49.888 [main] JmxServer [INFO] Started JmxServer registry port=54661 server port=54662 url=service:jmx:rmi://localhost:54662/jndi/rmi://localhost:54661/jmxrmi

If using the JConsole JMX client, launch it with the service URL as:

jconsole service:jmx:rmi://localhost:54662/jndi/rmi://localhost:54661/jmxrmi

If using the VisualVM JMX client, run:

jvisualvm

After VisualVM starts, click the “Add JMX Connection” button and paste in your JMX server URL (obtained from the logs). Install the VisualVM-MBeans plugin (Tools->Plugin) to view the metrics MBeans.

A.2 Reporting Metrics to Kafka (MetricsSnapshot Reporter)

This reporter publishes metrics to Kafka.

Enabling the MetricsSnapshot Reporter

To enable this reporter, simply append the following to your job’s configuration.

#Define a metrics reporter called "snapshot"
metrics.reporters=snapshot
metrics.reporter.snapshot.class=org.apache.samza.metrics.reporter.MetricsSnapshotReporterFactory

Specify the kafka topic to which the reporter should publish to

metrics.reporter.snapshot.stream=kafka.metrics

Specify the serializer to be used for the metrics data

serializers.registry.metrics.class=org.apache.samza.serializers.MetricsSnapshotSerdeFactory
systems.kafka.streams.metrics.samza.msg.serde=metrics

With this configuration, all containers (including the YARN-ApplicationMaster) will publish their JSON-encoded metrics to a Kafka topic called “metrics” every 60 seconds. The following is an example of such a metrics message:

{
  "header": {
    "container-name": "samza-container-0",

    "exec-env-container-id": "YARN-generated containerID",
    "host": "samza-grid-1234.example.com",
    "job-id": "1",
    "job-name": "my-samza-job",
    "reset-time": 1401729000347,
    "samza-version": "0.0.1",
    "source": "TaskName-Partition1",
    "time": 1401729420566,
    "version": "0.0.1"
  },
  "metrics": {
    "org.apache.samza.container.TaskInstanceMetrics": {
      "commit-calls": 1,
      "window-calls": 0,
      "process-calls": 14,

      "messages-actually-processed": 14,
      "send-calls": 0,

      "flush-calls": 1,
      "pending-messages": 0,
      "messages-in-flight": 0,
      "async-callback-complete-calls": 14,
        "wikipedia-#en.wikipedia-0-offset": 8979,
    }
  }
}

Each message contains a header which includes information about the job, time, and container from which the metrics were obtained. The remainder of the message contains the metric values, grouped by their types, such as TaskInstanceMetrics, SamzaContainerMetrics, KeyValueStoreMetrics, JVMMetrics, etc. Detailed descriptions of the various metric categories and metrics are available here.

It is possible to configure the MetricsSnapshot reporter to use a different serializer using this configuration

serializers.registry.metrics.class=<classpath-to-my-custom-serializer-factory>

To configure the reporter to publish with a different frequency (default 60 seconds), add the following to your job’s configuration

metrics.reporter.snapshot.interval=<publish frequency in seconds>

Similarly, to limit the set of metrics emitted you can use the regex based blacklist supported by this reporter. For example, to limit it to publishing only SamzaContainerMetrics use:

metrics.reporter.snapshot.blacklist=^(?!.\*?(?:SamzaContainerMetrics)).\*$

A.3 Creating a Custom MetricsReporter

Creating a custom MetricsReporter entails implementing the MetricsReporter interface. The lifecycle of Metrics Reporters is managed by Samza and is aligned with the Samza container lifecycle. Metrics Reporters can poll metric values and can receive callbacks when new metrics are added at runtime, e.g., user-defined metrics. Metrics Reporters are responsible for maintaining executor pools, IO connections, and any in-memory state that they require in order to export metrics to the desired external system, and managing the lifecycles of such components.

After implementation, a custom reporter can be enabled by appending the following to the Samza job’s configuration:

#Define a metrics reporter with a desired name
metrics.reporter.<my-custom-reporter-name>.class=<classpath-of-my-custom-reporter-factory>


#Enable its use for metrics reporting
metrics.reporters=<my-custom-reporter-name>

B. Metric Types in Samza

Metrics in Samza are divided into three types – Gauges, Counters, and Timers.

Gauges are useful when measuring the magnitude of a certain system property, e.g., the current queue length, or a buffer size.

Counters are useful in measuring metrics that are cumulative values, e.g., the number of messages processed since container startup. Certain counters are also useful when visualized with their rate-of-change, e.g., the rate of message processing.

Timers are useful for storing and reporting a sliding-window of timing values. Samza also supports a ListGauge type metric, which can be used to store and report a list of any primitive-type such as strings.

C. Adding User-Defined Metrics

To add a new metric, you can simply use the MetricsRegistry in the provided TaskContext of the init() method to register new metrics. The code snippets below show examples of registering and updating a user-defined Counter metric. Timers and gauges can similarly be used from within your task class.

Low Level Task API

Simply have your task implement the InitableTask interface and access the MetricsRegistry from the TaskContext.

public class MyJavaStreamTask implements StreamTask, InitableTask {

  private Counter messageCount;
  public void init(Config config, TaskContext context) {
    this.messageCount = context.getMetricsRegistry().newCounter(getClass().getName(), "message-count");

  }

  public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
    messageCount.inc();
  }

}

High Level Streams API

In the High Level Streams API, you can define a ContextManager and access the MetricsRegistry from the TaskContext, using which you can add and update your metrics.

public class MyJavaStreamApp implements StreamApplication {

  private Counter messageCount = null;

  @Override
  public void init(StreamGraph graph, Config config) {
    graph.withContextManager(new DemoContextManager());
    MessageStream<IndexedRecord> viewEvent = ...;
    viewEvent
        .map(this::countMessage)
        ...;
  }

  public final class DemoContextManager implements ContextManager {

  @Override
  public void init(Config config, TaskContext context) {
      messageCount = context.getMetricsRegistry().
      newCounter(getClass().getName(), "message-count");
  }

  private IndexedRecord countMessage(IndexedRecord value) {
    messageCount.inc();
    return value;
  }

  @Override
  public void close() { }

  }

D. Key Internal Samza Metrics

Samza’s internal metrics allow for detailed monitoring of a Samza job and all its components. Detailed descriptions of all internal metrics are listed in a reference sheet here. However, a small subset of internal metrics facilitates easy high-level monitoring of a job.

These key metrics can be grouped into three categories: Vital metrics, Storemetrics, and Operator metrics. We explain each of these categories in detail below.

D.1. Vital Metrics

These metrics indicate the vital signs of a Samza job’s health. Note that these metrics are categorized into different groups based on the Samza component they are emitted by, (e.g. SamzaContainerMetrics, TaskInstanceMetrics, ApplicationMaster metrics, etc).

D.2. Store Metrics

Stateful Samza jobs typically use RocksDB backed KV stores for storing state. Therefore, timing metrics associated with KV stores can be useful for monitoring input lag. These are some key metrics for KV stores. The metrics reference sheet here details all metrics for KV stores.

D.3. Operator Metrics

If your Samza job uses Samza’s Fluent API or Samza-SQL, Samza creates a DAG (directed acyclic graph) of operators to form the required data processing pipeline. In such cases, operator metrics allow fine-grained monitoring of such operators. Key operator metrics are listed below, while a detailed list is present in the metrics reference sheet.

E. Metrics Reference Sheet

Suffixes “-ms” and “-ns” to metric names indicated milliseconds and nanoseconds respectively. All “average time” metrics are calculated over a sliding time window of 300 seconds.

All <system>, <stream>, <partition>, <store-name>, <topic>, are populated with the corresponding actual values at runtime.