Metrics

Overview

Conduit comes with a number of already defined metrics. The metrics available are exposed through an HTTP API and ready to be scraped by Prometheus. It's also possible to easily define new metrics with existing types, or just create a completely new metric type.

Accessing metrics

Metrics are exposed at /metrics. For example, if you're running Conduit locally, you can get metrics if you run curl localhost:8080/metrics.

Available metrics

• Conduit metrics: We currently have a number of high level pipeline, processor and connector metrics, all of which are defined in measure.go. Those are:

  Pipeline name	Type	Description
  conduit_pipelines	Gauge	Number of pipelines by status.
  conduit_connectors	Gauge	Number of connectors by type (source, destination).
  conduit_processors	Gauge	Number of processors by name and type.
  conduit_connector_bytes	Histogram	Number of bytes* a connector processed by pipeline name, plugin and type (source, destination).
  conduit_dlq_bytes	Histogram	Number of bytes* a DLQ connector processed per pipeline and plugin.
  conduit_pipeline_execution_duration_seconds	Histogram	Amount of time records spent in a pipeline.
  conduit_connector_execution_duration_seconds	Histogram	Amount of time spent reading or writing records per pipeline, plugin and connector type (source, destination).
  conduit_processor_execution_duration_seconds	Histogram	Amount of time spent on processing records per pipeline and processor.
  conduit_dlq_execution_duration_seconds	Histogram	Amount of time spent writing records to DLQ connector per pipeline and plugin.
  conduit_inspector_sessions	Gauge	Number of inspector sessions by ID of pipeline component (connector or processor)

  *We calculate bytes based on the JSON representation of the record payload and key.

• Go runtime metrics: The default metrics exposed by Prometheus' official Go package, client_golang.

• gRPC metrics: The gRPC instrumentation package we use is promgrpc. The metrics exposed are listed here.

• HTTP API metrics: We use promhttp, Prometheus' official package for instrumentation of HTTP servers.

Adding new metrics

Currently, we have a number of metric types already defined in metrics.go. Those are: counter, gauge, timer and histogram and their "labeled" versions too. A labeled metric is one where labels must be set before usage. In many cases, the already present metric types should be sufficient.

Adding a new metric of an existing type is simple. Let's say we want to count number of messages processed, per pipeline. To do so we will define a labeled counter and increase the counter in source nodes, each time a message is read.

Create a new labeled counter

To do so, add the following code to measure.go.

PipelineMsgMetrics = metrics.NewLabeledCounter(
	"conduit_pipeline_msg_counter",
	"Number of messages per pipeline.",
	[]string{"pipeline_name"},
)

The labeled counter created here:

• has the name conduit_pipeline_msg_counter,
• has the description Number of messages per pipeline.,
• accepts a pipeline_name label.

Instantiate a counter with a label

Think of the labeled counter as of a factory for counters. It lets us create counters where the label it defines is set to a specific value (a pipeline name in our case).

In other words, for each pipeline we will have a separate counter (for which the pipeline_name label is set to the pipeline name). To do so, when building a source node in lifecycle.go, we can add the following:

sourceNode := stream.SourceNode{
	// initialize other fields
	PipelineMsgMetrics: measure.PipelineMsgMetrics.WithValues(pl.Config.Name),
}

Increment the counter

When a message is successfully read in a source node, we can increment the counter:

r, err := n.Source.Read(ctx)
if err == nil {
	n.PipelineMsgMetrics.Inc()
}

Check the metrics

Assuming you have a pipeline running locally, you can execute curl -Ss localhost:8080/metrics | grep conduit_pipeline_msg_counter to check your newly created metrics. You will see something along the lines of:

# HELP conduit_pipeline_msg_counter Number of messages per pipeline.
# TYPE conduit_pipeline_msg_counter counter
conduit_pipeline_msg_counter{pipeline_name="my-new-pipeline"} 84