feat: custom message based throughput measurements

.gitignore

@@ -25,4 +25,6 @@ release_deps
 # OpAmp Files
 collector*.yaml
 manager*.yaml
-logging*.yaml
+logging*.yaml
+config.yaml
+storage

.goreleaser.yml

@@ -14,6 +14,8 @@ builds:
     env:
       - CGO_ENABLED=0
     mod_timestamp: "{{ .CommitTimestamp }}"
+    tags:
+      - bindplane
     goos:
       - windows
       - linux

Makefile

@@ -32,7 +32,7 @@ VERSION ?= $(if $(CURRENT_TAG),$(CURRENT_TAG),$(PREVIOUS_TAG))
 # Builds just the agent for current GOOS/GOARCH pair
 .PHONY: agent
 agent:
-	go build -ldflags "-s -w -X github.com/observiq/bindplane-agent/internal/version.version=$(VERSION)" -o $(OUTDIR)/collector_$(GOOS)_$(GOARCH)$(EXT) ./cmd/collector
+	go build -ldflags "-s -w -X github.com/observiq/bindplane-agent/internal/version.version=$(VERSION)" -tags bindplane -o $(OUTDIR)/collector_$(GOOS)_$(GOARCH)$(EXT) ./cmd/collector
 
 # Builds just the updater for current GOOS/GOARCH pair
 .PHONY: updater

collector/collector.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/observiq/bindplane-agent/factories"
+	"github.com/observiq/bindplane-agent/internal/measurements"
 	"go.opentelemetry.io/collector/otelcol"
 	"go.uber.org/zap"
 )
@@ -199,6 +200,9 @@ func (c *collector) Stop(ctx context.Context) {
 	close(shutdownCompleteChan)
 
 	c.svc = nil
+
+	// After shutting down, we reset the throughputs measurements registry so it's fresh for the next collector startup.
+	measurements.BindplaneAgentThroughputMeasurementsRegistry.Reset()
 }
 
 // Restart will restart the collector. It will also reset the status channel.

extension/bindplaneextension/config.go

@@ -14,7 +14,32 @@
 
 package bindplaneextension
 
+import (
+	"errors"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+)
+
 // Config is the configuration for the bindplane extension
 type Config struct {
+	// Labels in "k1=v1,k2=v2" format
 	Labels string `mapstructure:"labels"`
+	// Component ID of the opamp extension. If not specified, then
+	// this extension will not generate any custom messages for throughput metrics.
+	OpAMP component.ID `mapstructure:"opamp"`
+	// MeasurementsInterval is the interval on which to report measurements.
+	// Measurements reporting is disabled if this duration is 0.
+	MeasurementsInterval time.Duration `mapstructure:"measurements_interval"`
+	// ExtraMeasurementsAttributes are a map of key-value pairs to add to all reported measurements.
+	ExtraMeasurementsAttributes map[string]string `mapstructure:"extra_measurements_attributes,omitempty"`
+}
+
+// Validate returns an error if the config is invalid
+func (c Config) Validate() error {
+	if c.MeasurementsInterval < 0 {
+		return errors.New("measurements interval must be postitive or 0")
+	}
+
+	return nil
 }

extension/bindplaneextension/extension.go

@@ -16,16 +16,151 @@ package bindplaneextension
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
 
+	"github.com/golang/snappy"
+	"github.com/observiq/bindplane-agent/internal/measurements"
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages"
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.uber.org/zap"
 )
 
-type bindplaneExtension struct{}
+type bindplaneExtension struct {
+	logger                  *zap.Logger
+	cfg                     *Config
+	ctmr                    *measurements.ResettableThroughputMeasurementsRegistry
+	customCapabilityHandler opampcustommessages.CustomCapabilityHandler
+	doneChan                chan struct{}
+	wg                      *sync.WaitGroup
+}
+
+func newBindplaneExtension(logger *zap.Logger, cfg *Config) *bindplaneExtension {
+	return &bindplaneExtension{
+		logger:   logger,
+		cfg:      cfg,
+		ctmr:     measurements.NewResettableThroughputMeasurementsRegistry(false),
+		doneChan: make(chan struct{}),
+		wg:       &sync.WaitGroup{},
+	}
+}
+
+func (b *bindplaneExtension) Start(_ context.Context, host component.Host) error {
+	var emptyComponentID component.ID
+
+	// Set up measurements if enabled
+	if b.cfg.OpAMP != emptyComponentID && b.cfg.MeasurementsInterval > 0 {
+		err := b.setupCustomCapabilities(host)
+		if err != nil {
+			return fmt.Errorf("setup capability handler: %w", err)
+		}
+
+		b.wg.Add(1)
+		go b.reportMetricsLoop()
+	}
 
-func (bindplaneExtension) Start(_ context.Context, _ component.Host) error {
 	return nil
 }
 
-func (bindplaneExtension) Shutdown(_ context.Context) error {
+func (b *bindplaneExtension) RegisterThroughputMeasurements(processorID string, measurements *measurements.ThroughputMeasurements) {
+	b.ctmr.RegisterThroughputMeasurements(processorID, measurements)
+}
+
+func (b *bindplaneExtension) setupCustomCapabilities(host component.Host) error {
+	ext, ok := host.GetExtensions()[b.cfg.OpAMP]
+	if !ok {
+		return fmt.Errorf("opamp extension %q does not exist", b.cfg.OpAMP)
+	}
+
+	registry, ok := ext.(opampcustommessages.CustomCapabilityRegistry)
+	if !ok {
+		return fmt.Errorf("extension %q is not an custom message registry", b.cfg.OpAMP)
+	}
+
+	var err error
+	b.customCapabilityHandler, err = registry.Register(measurements.ReportMeasurementsV1Capability)
+	if err != nil {
+		return fmt.Errorf("register custom capability: %w", err)
+	}
+
+	return nil
+}
+
+func (b *bindplaneExtension) Dependencies() []component.ID {
+	var emptyComponentID component.ID
+	if b.cfg.OpAMP == emptyComponentID {
+		return nil
+	}
+
+	return []component.ID{b.cfg.OpAMP}
+}
+
+func (b *bindplaneExtension) reportMetricsLoop() {
+	defer b.wg.Done()
+
+	t := time.NewTicker(b.cfg.MeasurementsInterval)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-t.C:
+			err := b.reportMetrics()
+			if err != nil {
+				b.logger.Error("Failed to report throughput metrics.", zap.Error(err))
+			}
+		case <-b.doneChan:
+			return
+		}
+	}
+}
+
+func (b *bindplaneExtension) reportMetrics() error {
+	m := b.ctmr.OTLPMeasurements(b.cfg.ExtraMeasurementsAttributes)
+
+	// Send metrics as snappy-encoded otlp proto
+	marshaller := pmetric.ProtoMarshaler{}
+	marshalled, err := marshaller.MarshalMetrics(m)
+	if err != nil {
+		return fmt.Errorf("marshal metrics: %w", err)
+	}
+
+	encoded := snappy.Encode(nil, marshalled)
+	for {
+		sendingChannel, err := b.customCapabilityHandler.SendMessage(measurements.ReportMeasurementsType, encoded)
+		switch {
+		case err == nil:
+			return nil
+		case errors.Is(err, types.ErrCustomMessagePending):
+			<-sendingChannel
+			continue
+		default:
+			return fmt.Errorf("send custom message: %w", err)
+		}
+	}
+}
+
+func (b *bindplaneExtension) Shutdown(ctx context.Context) error {
+	close(b.doneChan)
+
+	waitgroupDone := make(chan struct{})
+	go func() {
+		defer close(waitgroupDone)
+		b.wg.Wait()
+	}()
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-waitgroupDone: // OK
+	}
+
+	if b.customCapabilityHandler != nil {
+		b.customCapabilityHandler.Unregister()
+	}
+
 	return nil
 }

extension/bindplaneextension/factory.go

@@ -36,6 +36,8 @@ func defaultConfig() component.Config {
 	return &Config{}
 }
 
-func createBindPlaneExtension(_ context.Context, _ extension.Settings, _ component.Config) (extension.Extension, error) {
-	return bindplaneExtension{}, nil
+func createBindPlaneExtension(_ context.Context, cs extension.Settings, cfg component.Config) (extension.Extension, error) {
+	oCfg := cfg.(*Config)
+
+	return newBindplaneExtension(cs.Logger, oCfg), nil
 }

extension/bindplaneextension/go.mod

@@ -3,11 +3,17 @@ module github.com/observiq/bindplane-agent/extension/bindplaneextension
 go 1.21.9
 
 require (
+	github.com/golang/snappy v0.0.4
+	github.com/observiq/bindplane-agent/internal/measurements v1.59.0
+	github.com/open-telemetry/opamp-go v0.15.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages v0.107.0
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/collector/component v0.107.0
 	go.opentelemetry.io/collector/confmap v0.107.0
 	go.opentelemetry.io/collector/extension v0.107.0
+	go.opentelemetry.io/collector/pdata v1.13.0
 	go.uber.org/goleak v1.3.0
+	go.uber.org/zap v1.27.0
 )
 
 require (
@@ -20,11 +26,14 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/go-version v1.7.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
 	github.com/knadh/koanf/v2 v2.1.1 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.19.1 // indirect
@@ -34,15 +43,13 @@ require (
 	go.opentelemetry.io/collector/config/configtelemetry v0.107.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.13.0 // indirect
 	go.opentelemetry.io/collector/internal/globalgates v0.107.0 // indirect
-	go.opentelemetry.io/collector/pdata v1.13.0 // indirect
 	go.opentelemetry.io/otel v1.28.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.50.0 // indirect
 	go.opentelemetry.io/otel/metric v1.28.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.28.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.28.0 // indirect
 	go.opentelemetry.io/otel/trace v1.28.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/net v0.26.0 // indirect
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
@@ -51,3 +58,5 @@ require (
 	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+replace github.com/observiq/bindplane-agent/internal/measurements => ../../internal/measurements

extension/bindplaneextension/go.sum

@@ -2,6 +2,7 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
@@ -13,12 +14,17 @@ github.com/go-viper/mapstructure/v2 v2.0.0 h1:dhn8MZ1gZ0mzeodTG3jt5Vj/o87xZKuNAp
 github.com/go-viper/mapstructure/v2 v2.0.0/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
+github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hashicorp/go-version v1.7.0 h1:5tqGy27NaOTB8yJKUZELlFAS/LTKJkrmONwQKeRZfjY=
 github.com/hashicorp/go-version v1.7.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/knadh/koanf/maps v0.1.1 h1:G5TjmUh2D7G2YWf5SQQqSiHRJEjaicvU0KpypqB3NIs=
@@ -35,8 +41,23 @@ github.com/mitchellh/copystructure v1.2.0 h1:vpKXTN4ewci03Vljg/q9QvCGUDttBOGBIa1
 github.com/mitchellh/copystructure v1.2.0/go.mod h1:qLl+cE2AmVv+CoeAwDPye/v+N2HKCj9FbZEVFJRxO9s=
 github.com/mitchellh/reflectwalk v1.0.2 h1:G2LzWKi524PWgd3mLHV8Y5k7s6XUvT0Gef6zxSIeXaQ=
 github.com/mitchellh/reflectwalk v1.0.2/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/open-telemetry/opamp-go v0.15.0 h1:X2TWhEsGQ8GP7Uos3Ic9v/1aFUqoECZXKS7xAF5HqsA=
+github.com/open-telemetry/opamp-go v0.15.0/go.mod h1:QyPeN56JXlcZt5yG5RMdZ50Ju+zMFs1Ihy/hwHyF8Oo=
+github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages v0.107.0 h1:ijv/+GY9xqCXmy6PdyBBomw0OWVZ7Cq4TQvVh6np74Y=
+github.com/open-telemetry/opentelemetry-collector-contrib/extension/opampcustommessages v0.107.0/go.mod h1:1qSfNuJtzPlGadFK6cSZWtKmLNBhrQ5CgpOtLyn5baE=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.107.0 h1:Zo2UJk4AOxU5M6q3LNiFgTZfQicY9zz/BDqJWsmWzY4=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.107.0/go.mod h1:qj9lEtkVjQUzZ7FdJTeDqqTUq9xVU9kE4F8zZnHFB9M=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.107.0 h1:g1pkpFfe+dnhpfvo+f9yFIkbvTdiOvNmFOUFNzVAgvk=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.107.0/go.mod h1:oG/PliNiIOUHVARyDrFdvxFvG8DUPEjMGlmxjEqeoKM=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.107.0 h1:zTeRh4V3rMlXgNvfbDBnET6nvhOeZpYIbKTjVbSl9Ws=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.107.0/go.mod h1:/RtBag3LuHIkqN4bo8Erd3jCzA3gea70l9WyJ9TncXM=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v1.19.1 h1:wZWJDwK+NameRJuPGDhlnFgx8e8HN3XHQeLaYJFJBOE=
@@ -49,6 +70,8 @@ github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0leargg
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
 github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=