How to build a custom OTel receiver

[zmoog]

I want to learn how to build a custom OpenTelemetry receiver.

[zmoog]

Why

• collect data from unsupported sources
• learn how receivers work

[zmoog]

Created a GitHub repo using the work at https://zmoog.dev/posts/how-to-build-a-custom-open-telemery-collector/

[zmoog]

https://github.com/zmoog/custom-otel-collector

[zmoog]

The OTel docs contains a detailed guide about how to build a (traces) receiver: https://opentelemetry.io/docs/collector/building/receiver/

Here is my journey.

[zmoog]

Overview

• Set up a Go module for the receiver
• Design the config
• Build the factory
• Implement the receiver
• Add the receiver to the collector
• Running and debugging the receiver

[zmoog]

I want to focus this note to building and running a working no-op receiver component, leaving all data model design to a dedicated note.

[zmoog]

Set up a Go module for the receiver

In this note, I will scaffold a custom receiver for my solar panel inverter build by ZCS.

mkdir -p receiver/zcsreceiver
cd receiver/zcsreceiver 

$ go mod init github.com/zmoog/custom-otel-collector/receiver/zcsreceiver 
go: creating new go.mod: module github.com/zmoog/custom-otel-collector/receiver/zcsreceiver

cd ../..

go work use receiver/zcsreceiver 

[zmoog]

Design the config

Opening the collector's config.yaml file and start writing down the setting your ideal settings is a good starting point:

receivers:
  zcs: # this line represents the ID of your receiver

interval

Since the inverter data lives in the ZCS cloud, this receiver will periodically scrape the metrics from the ZCS cloud API.

At the moment we probably need at least an interval setting.

receivers:
  zcs: # this line represents the ID of your receiver
    interval: 1m

Authentication

The ZCS API requires authentication, so we probably need to also add dedicated options:

receivers:
  zcs: # this line represents the ID of your receiver
    interval: 1m
    auth:
      client_id: ${env:ZCS_CLIENT_ID}
      auth_key: ${env:ZCS_AUTH_KEY}

Filter

Finally, users may have one or more inverters, so we need to include an option to list the inverter serial numbers thay want to collect data from:

receivers:
  zcs: # this line represents the ID of your receiver
    interval: 1m
    auth:
      client_id: ${env:ZCS_CLIENT_ID}
      auth_key: ${env:ZCS_AUTH_KEY}
    thing_ids:
      - ${env:ZCS_THING_KEY}

[zmoog]

Design the config (continued)

We need to create a config.go file to store the Config struct that will hold the receiver config.

touch receiver/zcsreceiver/config.go

The Config struct must have a field for each receiver's settings.

// receiver/zcsreceiver/config.go
package zcsreceiver

// Config represents the receiver config settings within the collector's config.yaml
type Config struct {
	Interval string   `mapstructure:"interval"`
	Auth     Auth     `mapstructure:"auth"`
	ThingIds []string `mapstructure:"thing_ids"`
}

type Auth struct {
	ClientID string `mapstructure:"client_id"`
	AuthKey  string `mapstructure:"auth_key"`
}

We must also validate the config values by implementing the Validate() method according to the optional ConfigValidator interface.

package zcsreceiver

import "fmt"

// Config represents the receiver config settings within the collector's config.yaml
type Config struct {
	Interval string   `mapstructure:"interval"`
	Auth     Auth     `mapstructure:"auth"`
	ThingIds []string `mapstructure:"thing_ids"`
}

// Auth represents the authentication settings for the ZCS receiver
type Auth struct {
	ClientID string `mapstructure:"client_id"`
	AuthKey  string `mapstructure:"auth_key"`
}

// Validate checks if the configuration is valid
func (cfg *Config) Validate() error {
	if cfg.Interval == "" {
		return fmt.Errorf("interval is required")
	}
	if cfg.Auth.ClientID == "" {
		return fmt.Errorf("client_id is required")
	}
	if cfg.Auth.AuthKey == "" {
		return fmt.Errorf("auth_key is required")
	}
	if len(cfg.ThingIds) == 0 {
		return fmt.Errorf("thing_ids is required")
	}

	return nil
}

[zmoog]

Created the PR zmoog/custom-otel-collector#1 for the receiver config.

[zmoog]

Factory

The factory is responsible to set up the receiver, like defining which signals (logs, metrics, and traces) it is going to support.

Add the following code to your factory.go file:

package zcsreceiver

import (
	"go.opentelemetry.io/collector/receiver"
)

// NewFactory creates a factory for zcs receiver.
func NewFactory() receiver.Factory {
	return nil
}

Here's the NewFactory() function explained.

func NewFactory(
  cfgType component.Type,
  createDefaultConfig component.CreateDefaultConfigFunc,
  options ...FactoryOption) Factory

• component.Type: the cfgType is the unique identifier of the new component. For example, you will use this identifier in the config.yaml file.
• component.CreateDefaultConfigFunc: a function that is responsible of returning a component.Config instance for the receiver. Usually, the function take the struct from the receiver's config.go plus some additional configuration logic.
• ...FactoryOption: a slice of FactoryOption to set up the receiver, in particular the the signal type the receiver is capable of processing.

component.Type

To keep things simple, to provide the component.Type we can start with a simple var:

var typeStr = component.MustNewType("zcsreceiver")

component.CreateDefaultConfigFunc

As for default settings, you just need to add a function that returns a component.Config holding the default configurations for the zcsreceiver receiver.

func createDefaultConfig() component.Config {
	return &Config{
		Interval: string(defaultInterval),
	}
}

...FactoryOption

A receiver component can process traces, metrics, and logs. The receiver’s factory is responsible for specifying the capabilities that the receiver would provide.

Given that we want to collect metrics from an inverter, we will enable the zcsreceiver to work with metrics only. The OTel receiver package provides the following function to describe the metrics processing capabilities.

func WithMetrics(createMetrics CreateMetricsFunc, sl component.StabilityLevel) FactoryOption

The receiver.WithMetrics() function builds and returns a receiver.FactoryOption and requires the following parameters:

• createMetrics
• sl

The createTracesReceiver function requires:

• ctx context.Context
• params receiver.Settings
• cfg component.Config
• nextConsumer consumer.Metrics

func createMetricsReceiver(ctx context.Context, params receiver.Settings, cfg component.Config, nextConsumer consumer.Metrics) (receiver.Metrics, error) {
	return nil, nil
}

The createMetricsReceiver() will create a receiver for us, but we need to implement the actual receiver in the next section. We'll return to this function a little later.

Wrapping up the factory

We now have all the necessary components to instantiate a receiver factory using the receiver.NewFactory() function.

package zcsreceiver

import (
	"context"
	"time"

	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/receiver"
)

// typeStr is the type string for the zcs receiver.
var typeStr = component.MustNewType("zcsreceiver")

// defaultInterval is the default interval for the zcs receiver.
const defaultInterval = 1 * time.Minute

// createDefaultConfig creates a default config for the zcs receiver.
func createDefaultConfig() component.Config {
	return &Config{
		Interval: string(defaultInterval),
	}
}

// createMetricsReceiver creates a receiver for metrics.
func createMetricsReceiver(
	ctx context.Context,
	params receiver.Settings,
	cfg component.Config,
	nextConsumer consumer.Metrics,
) (receiver.Metrics, error) {
	// TODO: Implement metrics receiver
	return nil, nil
}

// NewFactory creates a factory for zcs receiver.
func NewFactory() receiver.Factory {
	return receiver.NewFactory(
		typeStr,
		createDefaultConfig,
		receiver.WithMetrics(createMetricsReceiver, component.StabilityLevelDevelopment),
	)
}

[zmoog]

Implementing the receiver component

All interfaces receiver.[Traces|Metrics|Logs] just inherit from component.Component to maximize extensibility.

So our receiver only need to implement the following two methods from component.Component:

Start(ctx context.Context, host Host) error
Shutdown(ctx context.Context) error

Both methods actually act as event handlers used by the Collector to communicate with its components as part of their lifecycle.

Start

The collector calls Start() on the receiver to signal it can start processing data.

Start(ctx context.Context, host Host) error

Start() receives two parameters:

• context.Context
• Host

context.Context

About context.Context:

Most of the time, a receiver will be processing a long-running operation, so the recommendation is to ignore this context and actually create a new one from context.Background().

Host

The host is meant to enable the receiver to communicate with the Collector’s host once it’s up and running.

Receivers should store a reference to the Host to interact with the collector.

For example, you can update the receiver status using the go.opentelemetry.io/collector/component/componentstatus package if the receiver hits a fatal failure after starting successfully:

func (fr *functionReceiver) Start(ctx context.Context, host component.Host) error {

        // ....

	go func() {
		defer fr.wg.Done()

		if errHTTP := fr.server.Serve(ln); errHTTP != nil && !errors.Is(errHTTP, http.ErrServerClosed) {
			componentstatus.ReportStatus(host, componentstatus.NewFatalErrorEvent(errHTTP))
		}
	}()

	return nil
}

Shutdown

The collector calls Start() on the receiver to signal it must stop processing data.

Implementation

touch receiver/zcsreceiver/receiver.go

[zmoog]

Updating the Collector’s initialization process