Développer son propre receiver OpenTelemetry : l'exemple d'un receiver NATS

L’OpenTelemetry Collector embarque déjà des dizaines de receivers, mais tôt ou tard vous tomberez sur une source que personne n’a encore couverte : un bus de messages interne, un protocole maison, une file propriétaire. La bonne nouvelle, c’est que le Collector est conçu pour être étendu : écrire son propre receiver est un exercice balisé, presque mécanique une fois qu’on en connaît l’anatomie.

Dans cet article, on construit un receiver complet de A à Z en prenant pour fil rouge un receiver NATS JetStream — il consomme des messages OTLP (traces, métriques, logs) publiés sur NATS et les pousse dans le pipeline du Collector. On détaillera factory.go, config.go, la génération de code avec mdatagen, l’intégration dans une distribution via OCB, le remplissage du YAML, les tests et le workflow CI.

Cet article est la suite logique de OCB : construire son propre Collector sur-mesure. OCB assemble la distribution ; ici, on fabrique le composant qu’on va y brancher.

Anatomie d’un receiver

Un composant du Collector suit toujours la même structure. Pour un receiver, les fichiers clés sont :

Fichier	Rôle
config.go	La structure de configuration et sa validation.
factory.go	La factory : type du composant, config par défaut, constructeurs par signal.
receiver.go	La logique : connexion à la source, désérialisation, push dans le pipeline.
metadata.yaml	Le descripteur du composant (type, stabilité, télémétrie interne).
internal/metadata/	Le code généré par mdatagen à partir de metadata.yaml.
*_test.go	Les tests unitaires et de cycle de vie.

flowchart TD
    M[metadata.yaml] -->|mdatagen| G[internal/metadata/*.go]
    CFG[config.go] --> F[factory.go]
    F --> R[receiver.go]
    G --> R
    F -->|NewFactory| OCB[Distribution OCB]

Le Collector ne connaît votre composant qu’à travers une interface : il appelle votre NewFactory(), lui demande une config par défaut, puis lui demande de créer un receiver pour chaque signal. Tout le reste est votre implémentation.

config.go : décrire et valider la configuration

La configuration est une simple structure Go annotée de tags mapstructure — c’est ce qui permet au Collector de mapper le YAML sur vos champs. Voici une version condensée pour le receiver NATS :

package natsreceiver

import (
	"errors"
	"fmt"
	"time"

	"go.opentelemetry.io/collector/config/configopaque"
	"go.opentelemetry.io/collector/config/configtls"
)

// Config définit la configuration du receiver NATS JetStream.
type Config struct {
	URL            string                 `mapstructure:"url"`
	ConnectTimeout time.Duration          `mapstructure:"connect_timeout"`
	MaxReconnects  int                    `mapstructure:"max_reconnects"`
	ReconnectWait  time.Duration          `mapstructure:"reconnect_wait"`
	TLS            configtls.ClientConfig `mapstructure:"tls"`
	Auth           NATSAuth               `mapstructure:"auth"`
	SubjectPrefix  string                 `mapstructure:"subject_prefix"`
	Encoding       string                 `mapstructure:"encoding"`
	DurableName    string                 `mapstructure:"durable_name"`
}

// NATSAuth porte la configuration d'authentification (une seule méthode à la fois).
type NATSAuth struct {
	NKeyFile        string              `mapstructure:"nkey_file"`
	CredentialsFile string              `mapstructure:"credentials_file"`
	User            string              `mapstructure:"user"`
	Password        configopaque.String `mapstructure:"password"`
	Token           configopaque.String `mapstructure:"token"`
}

Trois bonnes pratiques sautent aux yeux :

• Réutilisez les types du Collector. configtls.ClientConfig vous offre toute la configuration TLS standard (cert_file, ca_file, insecure…) gratuitement, et de façon cohérente avec les autres composants.
• Protégez les secrets avec configopaque.String. Un champ de ce type est automatiquement masqué ([REDACTED]) dans les logs et les dumps de config : un mot de passe ou un token ne fuitera pas par accident.
• Validez tôt. Implémentez l’interface Validate() error : le Collector l’appelle au démarrage, avant même d’instancier le composant. Une config invalide fait échouer le boot avec un message clair plutôt que de planter en production.

// Validate vérifie que la configuration est cohérente.
func (cfg *Config) Validate() error {
	if cfg.URL == "" {
		return errors.New("url must not be empty")
	}
	if cfg.Encoding != "otlp_proto" && cfg.Encoding != "otlp_json" {
		return fmt.Errorf("unsupported encoding %q, supported values: otlp_proto, otlp_json", cfg.Encoding)
	}
	if cfg.ConnectTimeout <= 0 {
		return errors.New("connect_timeout must be positive")
	}
	if cfg.DurableName == "" {
		return errors.New("durable_name must not be empty")
	}
	// Une seule méthode d'auth à la fois (credentials_file, nkey_file, user/password, token).
	if err := cfg.Auth.validate(); err != nil {
		return fmt.Errorf("auth: %w", err)
	}
	return nil
}

factory.go : le point d’entrée du composant

La factory est ce que le Collector instancie. Elle déclare le type du composant (le mot-clé YAML), la config par défaut, et un constructeur par signal supporté (traces, métriques, logs).

package natsreceiver

import (
	"context"
	"time"

	"github.com/votre-org/natsreceiver/internal/metadata"
	"go.opentelemetry.io/collector/component"
	"go.opentelemetry.io/collector/consumer"
	"go.opentelemetry.io/collector/receiver"
)

// NewFactory crée la factory du receiver NATS JetStream.
func NewFactory() receiver.Factory {
	return receiver.NewFactory(
		metadata.Type, // = component.MustNewType("nats")
		createDefaultConfig,
		receiver.WithTraces(createTracesReceiver, metadata.TracesStability),
		receiver.WithMetrics(createMetricsReceiver, metadata.MetricsStability),
		receiver.WithLogs(createLogsReceiver, metadata.LogsStability),
	)
}

func createDefaultConfig() component.Config {
	return &Config{
		URL:            "nats://localhost:4222",
		ConnectTimeout: 5 * time.Second,
		MaxReconnects:  -1,
		ReconnectWait:  2 * time.Second,
		SubjectPrefix:  "otel",
		Encoding:       "otlp_proto",
		DurableName:    "otel-collector",
	}
}

func createTracesReceiver(
	_ context.Context,
	set receiver.Settings,
	cfg component.Config,
	nextConsumer consumer.Traces,
) (receiver.Traces, error) {
	rcv, err := getOrCreateReceiver(cfg.(*Config), set)
	if err != nil {
		return nil, err
	}
	rcv.tracesConsumer = nextConsumer
	return rcv, nil
}

Quelques points importants :

• metadata.Type et metadata.*Stability proviennent du code généré par mdatagen (voir plus bas). Le type "nats" est exactement le mot-clé que l’utilisateur écrira dans receivers: nats:.
• createDefaultConfig doit renvoyer une config fonctionnelle par défaut. L’utilisateur ne redéfinit que ce qu’il veut changer ; tout le reste hérite de ces valeurs.
• Le nextConsumer est le maillon suivant du pipeline. Votre receiver reçoit la donnée brute, la transforme en pdata, puis appelle nextConsumer.ConsumeTraces(ctx, td).

Astuce : un receiver partagé entre plusieurs signaux

Le Collector appelle createTracesReceiver, createMetricsReceiver et createLogsReceiver séparément, mais leur passe le même pointeur de *Config quand ils sont déclarés sous le même bloc. On peut s’en servir pour ne créer qu’une seule connexion partagée entre les trois signaux :

var (
	receiversMu sync.Mutex
	receivers   = map[*Config]*natsReceiver{}
)

// getOrCreateReceiver renvoie un receiver partagé par identité de pointeur de config.
func getOrCreateReceiver(cfg *Config, set receiver.Settings) (*natsReceiver, error) {
	receiversMu.Lock()
	defer receiversMu.Unlock()
	if rcv, ok := receivers[cfg]; ok {
		return rcv, nil
	}
	rcv, err := newNATSReceiver(cfg, set)
	if err != nil {
		return nil, err
	}
	receivers[cfg] = rcv
	return rcv, nil
}

Côté receiver.go, on protège alors le démarrage par un sync.Once pour que la connexion ne s’ouvre qu’une fois, même si les trois signaux pointent vers le même receiver.

metadata.yaml + mdatagen : générer le code répétitif

mdatagen est l’outil officiel qui génère, à partir d’un descripteur YAML, tout le code « plomberie » d’un composant : le type, les niveaux de stabilité, et surtout les compteurs de télémétrie interne (combien de messages reçus, combien d’échecs…). On décrit, il génère.

# metadata.yaml
type: nats

status:
  class: receiver
  stability:
    alpha: [metrics, logs, traces]

telemetry:
  metrics:
    nats_receiver_received_count:
      enabled: true
      description: Number of messages successfully received from NATS JetStream.
      unit: "{message}"
      sum:
        value_type: int
        monotonic: true
      attributes: [signal_type]
    nats_receiver_failed_count:
      enabled: true
      description: Number of messages that failed to be processed.
      unit: "{message}"
      sum:
        value_type: int
        monotonic: true
      attributes: [signal_type]

attributes:
  signal_type:
    description: The type of OpenTelemetry signal (traces, metrics, or logs).
    type: string
    enum: [traces, metrics, logs]

On l’installe puis on génère le code via une directive go:generate :

# Installer mdatagen
go install go.opentelemetry.io/collector/cmd/mdatagen@latest

// Dans un fichier du package (par ex. doc.go)
//go:generate mdatagen metadata.yaml

# Générer (ou régénérer) le code
go generate ./...

mdatagen produit alors le dossier internal/metadata/ avec, entre autres, un TelemetryBuilder prêt à l’emploi :

// Code generated by mdatagen. DO NOT EDIT.
type TelemetryBuilder struct {
	NatsReceiverReceivedCount metric.Int64Counter
	NatsReceiverFailedCount   metric.Int64Counter
	// ...
}

func NewTelemetryBuilder(settings component.TelemetrySettings, ...) (*TelemetryBuilder, error) {
	// instruments créés à partir de metadata.yaml :
	//   otelcol_nats_receiver_received_count
	//   otelcol_nats_receiver_failed_count
}

Dans receiver.go, on construit ce builder une fois et on incrémente les compteurs au fil de l’eau :

tb, err := metadata.NewTelemetryBuilder(set.TelemetrySettings)
// ...
tb.NatsReceiverReceivedCount.Add(ctx, 1, attrSignalTraces)

Avantage : votre receiver expose immédiatement sa propre télémétrie, au même format et avec le même préfixe otelcol_ que tous les composants natifs. C’est exactement ce qu’on veut pour observer le Collector lui-même. Bonus : mdatagen génère aussi des tests de cycle de vie (generated_component_test.go, generated_package_test.go) qui valident que votre factory respecte le contrat du Collector.

Installer le receiver dans une distribution (OCB)

Votre receiver est un module Go publiable comme n’importe quel autre. Pour l’embarquer dans un Collector, on l’ajoute au manifeste OCB sous receivers: :

# builder-config.yaml
receivers:
  - gomod: github.com/votre-org/natsreceiver v0.1.0

En développement, avant même de publier une version, on peut pointer vers le code local grâce à la section replaces du manifeste :

replaces:
  - github.com/votre-org/natsreceiver => ../natsreceiver

Puis on (re)construit la distribution :

./ocb --config builder-config.yaml

OCB génère le code du Collector, ajoute votre module au go.mod, télécharge les dépendances et compile un binaire qui contient votre receiver. (Le détail du manifeste et de l’installation d’OCB est couvert dans l’article précédent.)

Remplir le YAML : configurer le receiver

Une fois le composant dans la distribution, l’utilisateur final n’a plus qu’à le déclarer. Le type (nats) et les clés correspondent exactement aux tags mapstructure de config.go. Exemple minimal — tout sur les valeurs par défaut :

receivers:
  nats:
    url: "nats://localhost:4222"

exporters:
  debug:
    verbosity: detailed

service:
  pipelines:
    traces:
      receivers: [nats]
      exporters: [debug]
    metrics:
      receivers: [nats]
      exporters: [debug]
    logs:
      receivers: [nats]
      exporters: [debug]

Exemple complet — TLS, authentification et réglages avancés :

receivers:
  nats:
    url: "nats://nats.example.com:4222"
    subject_prefix: "telemetry"
    encoding: "otlp_json"
    durable_name: "my-collector"
    connect_timeout: 10s
    max_reconnects: 5
    reconnect_wait: 3s
    tls:
      cert_file: /etc/tls/client.crt
      key_file: /etc/tls/client.key
      ca_file: /etc/tls/ca.crt
    auth:
      credentials_file: /etc/nats/user.creds

Écrire les tests

Un composant du Collector se teste à trois niveaux. La règle d’or : plus c’est rapide et déterministe, plus on en écrit.

1. La validation de config (rapide, exhaustive)

On couvre chaque règle de Validate(). C’est le filet de sécurité le plus rentable :

func newValidConfig() *Config {
	return &Config{
		URL: "nats://localhost:4222", SubjectPrefix: "otel",
		Encoding: "otlp_proto", DurableName: "otel-collector",
		ConnectTimeout: 5 * time.Second, ReconnectWait: 2 * time.Second,
	}
}

func TestConfigValidate_EncodingUnsupported(t *testing.T) {
	cfg := newValidConfig()
	cfg.Encoding = "avro"
	err := cfg.Validate()
	require.Error(t, err)
	assert.Contains(t, err.Error(), "unsupported encoding")
}

func TestConfigValidate_AuthConflict(t *testing.T) {
	cfg := newValidConfig()
	cfg.Auth.CredentialsFile = "/path/creds"
	cfg.Auth.NKeyFile = "/path/nkey"
	err := cfg.Validate()
	require.Error(t, err)
	assert.Contains(t, err.Error(), "only one authentication method")
}

2. Le chargement depuis un YAML réel (confmaptest)

On vérifie que le YAML se mappe bien sur la structure, en chargeant un fichier de testdata/ :

# testdata/config.yaml
nats:
  url: "nats://nats.example.com:4222"
  subject_prefix: "telemetry"
  encoding: "otlp_json"
  auth:
    token: "secret-token"

func TestLoadConfig(t *testing.T) {
	cm, err := confmaptest.LoadConf("testdata/config.yaml")
	require.NoError(t, err)

	sub, err := cm.Sub(component.NewIDWithName(componentType, "").String())
	require.NoError(t, err)

	cfg := createDefaultConfig().(*Config)
	require.NoError(t, sub.Unmarshal(cfg))

	assert.Equal(t, "telemetry", cfg.SubjectPrefix)
	assert.Equal(t, configopaque.String("secret-token"), cfg.Auth.Token)
	require.NoError(t, cfg.Validate())
}

3. Le cycle de vie de bout en bout

Pour la factory, on s’assure que chaque signal crée bien un receiver, puis on fait un test d’intégration complet : démarrer un broker NATS embarqué, publier un message, démarrer le receiver et vérifier qu’il atterrit dans un sink de test.

func startTestServer(t *testing.T) *server.Server {
	opts := natsservertest.DefaultTestOptions
	opts.Port = -1            // port aléatoire
	opts.JetStream = true
	opts.StoreDir = t.TempDir()
	ns := natsservertest.RunServer(&opts)
	t.Cleanup(ns.Shutdown)
	return ns
}

func TestFactoryLifecycle_Traces(t *testing.T) {
	ns := startTestServer(t)
	// ... publier un message OTLP sur "otel.traces.>" ...

	factory := NewFactory()
	cfg := factory.CreateDefaultConfig().(*Config)
	cfg.URL = ns.ClientURL()

	set := receivertest.NewNopSettings(componentType)
	sink := &consumertest.TracesSink{}
	rcv, err := factory.CreateTraces(context.Background(), set, cfg, sink)
	require.NoError(t, err)

	require.NoError(t, rcv.Start(context.Background(), componenttest.NewNopHost()))
	t.Cleanup(func() { _ = rcv.Shutdown(context.Background()) })

	require.Eventually(t, func() bool {
		return len(sink.AllTraces()) > 0
	}, 5*time.Second, 50*time.Millisecond)
}

Deux réflexes utiles : lancer go test -race pour traquer les accès concurrents (un receiver est intrinsèquement concurrent), et ajouter go.uber.org/goleak pour détecter les goroutines qui ne se ferment pas au Shutdown().

Configurer le workflow

Garde-fous locaux avec lefthook

Avant même la CI, lefthook fait tourner lint, tests et go mod tidy à chaque commit — les erreurs sont attrapées sur le poste du développeur :

# lefthook.yml
pre-commit:
  parallel: true
  commands:
    lint:
      glob: "*.go"
      run: golangci-lint run --fix --timeout=5m
      stage_fixed: true
    test:
      glob: "*.go"
      run: go test -race -short ./...
    tidy:
      glob: "*.go"
      run: go mod tidy && git diff --exit-code go.mod go.sum

pre-push:
  parallel: true
  commands:
    test-full:
      run: go test -race ./...
    build:
      run: go build ./...

Intégration continue avec GitHub Actions

Un workflow autonome suffit : lint, tests sous le détecteur de data races, build, et — réflexe propre aux composants générés — vérifier que le code mdatagen est à jour (régénérer et exiger un diff vide) :

# .github/workflows/ci.yaml
name: CI
on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  build-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-go@v5
        with:
          go-version: '1.24'
          cache: true

      - name: Lint
        uses: golangci/golangci-lint-action@v6

      - name: Vérifier le code généré (mdatagen)
        run: |
          go install go.opentelemetry.io/collector/cmd/mdatagen@latest
          go generate ./...
          git diff --exit-code

      - name: Tests (race + couverture)
        run: go test -race -covermode=atomic -coverprofile=coverage.out ./...

      - name: Build
        run: go build ./...

Quand un composant grandit en plusieurs receivers/exporters partageant la même mécanique, on factorise ces étapes dans un workflow réutilisable (workflow_call) appelé par chaque dépôt — exactement le pattern décrit dans l’article sur OCB. Le dépôt du composant ne garde alors qu’un appelant de quelques lignes.

Conclusion

Écrire un receiver OpenTelemetry, c’est assembler quatre briques bien délimitées : une config validée, une factory qui expose le composant, une logique qui parle à votre source, et un metadata.yaml que mdatagen transforme en télémétrie interne. Le reste — intégration via OCB, YAML, tests, CI — suit des patterns réutilisables d’un composant à l’autre.

Une fois ce squelette en place, vous pouvez instrumenter n’importe quelle source et l’amener dans votre pipeline d’observabilité, sans dépendre du catalogue officiel ni d’un vendor.

Si vous voulez développer un composant sur-mesure du Collector ou industrialiser votre plateforme d’observabilité, parlons-en.