Repository: honeycombio/refinery
Pull request #1528 (2025-04-15 22:26:42)

feat: add experimental support for opamp
## Which problem is this PR solving?

- This adds OpAMP configuration support to refinery.

## Short description of the changes

- add configuration options for opamp
- update documentat
- update tests




Patch:
diff --git a/agent/agent.go b/agent/agent.go
new file mode 100644
index 0000000..3e1c5f0
--- /dev/null
+++ b/agent/agent.go
@@ -0,0 +1,437 @@
+package agent
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"net/http"
+	"os"
+	"runtime"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/internal/health"
+	"github.com/honeycombio/refinery/metrics"
+	"github.com/jonboulle/clockwork"
+	"github.com/open-telemetry/opamp-go/client"
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
+)
+
+const (
+	serviceName                  = "refinery"
+	sendAgentTelemetryCapability = "io.honeycomb.capabilities.sendAgentTelemetry"
+	defaultHealthCheckInterval   = 15 * time.Second
+	defaultReportUsageInterval   = 15 * time.Second
+)
+
+type Agent struct {
+	clock                    clockwork.Clock
+	agentType                string
+	agentVersion             string
+	instanceId               uuid.UUID
+	hostname                 string
+	effectiveConfig          config.Config
+	agentDescription         *protobufs.AgentDescription
+	opampClient              client.OpAMPClient
+	remoteConfigStatus       *protobufs.RemoteConfigStatus
+	lastRemoteConfigReceived *protobufs.AgentRemoteConfig
+
+	//	opampClientCert     *tls.Certificate
+	//	caCertPath          string
+	//	certRequested       bool
+	//	clientPrivateKeyPEM []byte
+	lastHealth *protobufs.ComponentHealth
+
+	logger       Logger
+	ctx          context.Context
+	cancel       context.CancelFunc
+	metrics      metrics.Metrics
+	usageTracker *usageTracker
+	health       health.Reporter
+
+	healthCheckInterval time.Duration
+	reportUsageInterval time.Duration
+}
+
+func NewAgent(refineryLogger Logger, agentVersion string, currentConfig config.Config, metrics metrics.Metrics, health health.Reporter) *Agent {
+	ctx, cancel := context.WithCancel(context.Background())
+	agent := &Agent{
+		ctx:                 ctx,
+		cancel:              cancel,
+		clock:               clockwork.NewRealClock(),
+		logger:              refineryLogger,
+		agentType:           serviceName,
+		agentVersion:        agentVersion,
+		effectiveConfig:     currentConfig,
+		metrics:             metrics,
+		health:              health,
+		usageTracker:        newUsageTracker(),
+		healthCheckInterval: defaultHealthCheckInterval,
+		reportUsageInterval: defaultReportUsageInterval,
+	}
+	agent.createAgentIdentity()
+	agent.logger.Debugf(context.Background(), "starting opamp client, id=%v", agent.instanceId)
+	if err := agent.connect(); err != nil {
+		agent.logger.Errorf(context.Background(), "Failed to connect to OpAMP Server: %v", err)
+		return nil
+	}
+	return agent
+}
+
+func (agent *Agent) createAgentIdentity() {
+	uid, err := uuid.NewV7()
+	if err != nil {
+		panic(err)
+	}
+	agent.instanceId = uid
+	hostname, _ := os.Hostname()
+	agent.hostname = hostname
+	agent.agentDescription = &protobufs.AgentDescription{
+		IdentifyingAttributes: []*protobufs.KeyValue{
+			{
+				Key: "service.name",
+				Value: &protobufs.AnyValue{
+					Value: &protobufs.AnyValue_StringValue{StringValue: agent.agentType},
+				},
+			},
+			{
+				Key: "service.version",
+				Value: &protobufs.AnyValue{
+					Value: &protobufs.AnyValue_StringValue{StringValue: agent.agentVersion},
+				},
+			},
+		},
+		NonIdentifyingAttributes: []*protobufs.KeyValue{
+			{
+				Key: "os.type",
+				Value: &protobufs.AnyValue{
+					Value: &protobufs.AnyValue_StringValue{StringValue: runtime.GOOS},
+				},
+			},
+			{
+				Key: "host.name",
+				Value: &protobufs.AnyValue{
+					Value: &protobufs.AnyValue_StringValue{StringValue: agent.hostname},
+				},
+			},
+		},
+	}
+}
+
+func (agent *Agent) connect() error {
+	agent.opampClient = client.NewWebSocket(&agent.logger)
+
+	settings := types.StartSettings{
+		Header:         http.Header{"agent": []string{"refinery"}},
+		OpAMPServerURL: agent.effectiveConfig.GetOpAMPConfig().Endpoint,
+		InstanceUid:    types.InstanceUid(agent.instanceId),
+		Callbacks: types.Callbacks{
+			OnConnect: func(ctx context.Context) {
+				agent.logger.Debugf(ctx, "connected to OpAMP server")
+			},
+			OnConnectFailed: func(ctx context.Context, err error) {
+				agent.logger.Errorf(ctx, "Failed to connect to server: %v", err)
+			},
+			OnError: func(ctx context.Context, err *protobufs.ServerErrorResponse) {
+				agent.logger.Errorf(ctx, "Received error from server: %v", err)
+			},
+			// TODO: when will this get called??
+			SaveRemoteConfigStatus: func(ctx context.Context, status *protobufs.RemoteConfigStatus) {
+				agent.logger.Debugf(ctx, "got remote config status: %v", status)
+				agent.remoteConfigStatus = status
+			},
+			GetEffectiveConfig: func(ctx context.Context) (*protobufs.EffectiveConfig, error) {
+				return agent.composeEffectiveConfig(), nil
+			},
+			OnMessage:                 agent.onMessage,
+			OnOpampConnectionSettings: agent.onOpampConnectionSettings,
+		},
+		RemoteConfigStatus: agent.remoteConfigStatus,
+		Capabilities: protobufs.AgentCapabilities_AgentCapabilities_AcceptsRemoteConfig |
+			protobufs.AgentCapabilities_AgentCapabilities_ReportsRemoteConfig |
+			protobufs.AgentCapabilities_AgentCapabilities_ReportsEffectiveConfig |
+			protobufs.AgentCapabilities_AgentCapabilities_ReportsOwnMetrics |
+			protobufs.AgentCapabilities_AgentCapabilities_AcceptsOpAMPConnectionSettings |
+			protobufs.AgentCapabilities_AgentCapabilities_ReportsHealth,
+	}
+
+	err := agent.opampClient.SetAgentDescription(agent.agentDescription)
+	if err != nil {
+		return err
+	}
+	err = agent.opampClient.SetHealth(healthMessage(false))
+	if err != nil {
+		return err
+	}
+
+	agent.opampClient.SetCustomCapabilities(&protobufs.CustomCapabilities{
+		Capabilities: []string{sendAgentTelemetryCapability},
+	})
+
+	agent.logger.Debugf(context.Background(), "starting opamp client")
+
+	err = agent.opampClient.Start(context.Background(), settings)
+	if err != nil {
+		return err
+	}
+	agent.logger.Debugf(context.Background(), "started opamp client")
+
+	go agent.healthCheck()
+	go agent.reportUsagePeriodically()
+	return nil
+}
+
+func (agent *Agent) Stop(ctx context.Context) {
+	agent.logger.Debugf(ctx, "disconnecting from OpAMP server")
+	err := agent.opampClient.SetHealth(
+		&protobufs.ComponentHealth{
+			Healthy: false, LastError: "Refinery is shutdown",
+		},
+	)
+	if err != nil {
+		agent.logger.Errorf(ctx, "Could not report health to OpAMP server: %v", err)
+	}
+	err = agent.opampClient.Stop(ctx)
+	if err != nil {
+		agent.logger.Errorf(ctx, "Failed to stop OpAMP client: %v", err)
+	}
+
+	agent.cancel()
+}
+
+func (agent *Agent) healthCheck() {
+	timer := agent.clock.NewTicker(agent.healthCheckInterval)
+	for {
+		select {
+		case <-agent.ctx.Done():
+		case <-timer.Chan():
+			report := agent.calculateHealth()
+			if report != nil {
+				agent.lastHealth = report
+				if err := agent.opampClient.SetHealth(report); err != nil {
+					agent.logger.Errorf(context.Background(), "Could not report health to OpAMP server: %v", err)
+				}
+			}
+
+			traceUsage, ok := agent.metrics.Get("bytes_received_traces")
+			if !ok {
+				agent.logger.Errorf(context.Background(), "unexpected missing trace usage metric")
+			}
+			logUsage, ok := agent.metrics.Get("bytes_received_logs")
+			if !ok {
+				agent.logger.Errorf(context.Background(), "unexpected missing log usage metric")
+			}
+
+			agent.usageTracker.Add(signal_traces, traceUsage)
+			agent.usageTracker.Add(signal_logs, logUsage)
+		}
+	}
+}
+
+func (agent *Agent) reportUsagePeriodically() {
+	timer := agent.clock.NewTicker(agent.reportUsageInterval)
+	defer timer.Stop()
+
+	for {
+		select {
+		case <-agent.ctx.Done():
+			// TODO: drain the existing reports
+			return
+		case <-timer.Chan():
+			if err := agent.sendUsageReport(); err != nil {
+				if errors.Is(err, errNoData) {
+					agent.logger.Debugf(context.Background(), "No data to report")
+					continue
+				}
+				agent.logger.Errorf(context.Background(), "Could not send usage report: %v", err)
+			}
+
+		}
+	}
+}
+
+func (agent *Agent) sendUsageReport() error {
+	usageReport, err := agent.usageTracker.NewReport(agent.agentType, agent.agentVersion, agent.hostname, agent.clock.Now())
+	if err != nil {
+		return err
+	}
+
+	isSent, err := agent.opampClient.SendCustomMessage(&protobufs.CustomMessage{
+		Capability: sendAgentTelemetryCapability,
+		Data:       usageReport,
+	})
+
+	if err != nil {
+		if errors.Is(err, types.ErrCustomMessagePending) {
+			agent.logger.Debugf(context.Background(), "Usage report is pending")
+			select {
+			case <-agent.ctx.Done():
+				// TODO: we probably need to drain the existing reports
+				return agent.ctx.Err()
+			case <-isSent:
+				// Retry sending the message once
+				isSent, err = agent.opampClient.SendCustomMessage(&protobufs.CustomMessage{
+					Capability: sendAgentTelemetryCapability,
+					Data:       usageReport,
+				})
+				if err != nil {
+					return err
+				}
+			}
+		} else {
+			return err
+		}
+	}
+
+	select {
+	case <-agent.ctx.Done():
+		return agent.ctx.Err()
+	case <-isSent:
+		agent.usageTracker.completeSend()
+		return nil
+	}
+}
+
+func (agent *Agent) calculateHealth() *protobufs.ComponentHealth {
+	lastHealth := agent.lastHealth
+	report := healthMessage(agent.health.IsAlive())
+	if report.GetHealthy() {
+		report.Healthy = agent.health.IsReady()
+	}
+
+	// report health only if it has changed
+	if lastHealth == nil || lastHealth.GetHealthy() != report.GetHealthy() {
+		return report
+	}
+
+	return nil
+}
+
+func (agent *Agent) composeEffectiveConfig() *protobufs.EffectiveConfig {
+	configYAML, err := config.SerializeToYAML(agent.effectiveConfig)
+	if err != nil {
+		agent.logger.Errorf(context.Background(), "Failed to marshal effective config: %v", err)
+		return nil
+	}
+	return &protobufs.EffectiveConfig{
+		ConfigMap: &protobufs.AgentConfigMap{
+			ConfigMap: map[string]*protobufs.AgentConfigFile{
+				"": {Body: []byte(configYAML)},
+			},
+		},
+	}
+}
+
+func (agent *Agent) reportConfigStatus(status protobufs.RemoteConfigStatuses, errorMessage string) {
+	err := agent.opampClient.SetAgentDescription(agent.agentDescription)
+	if err != nil {
+		agent.logger.Errorf(context.Background(), "Could not report OpAMP remote config status: %s", err)
+	}
+	remoteConfigstatus := &protobufs.RemoteConfigStatus{
+		LastRemoteConfigHash: agent.lastRemoteConfigReceived.GetConfigHash(),
+		Status:               status,
+		ErrorMessage:         errorMessage,
+	}
+	err = agent.opampClient.SetRemoteConfigStatus(remoteConfigstatus)
+	if err != nil {
+		agent.logger.Errorf(context.Background(), "Could not report OpAMP remote config status: %s", err)
+		return
+	}
+	agent.remoteConfigStatus = remoteConfigstatus
+}
+
+func (agent *Agent) onMessage(ctx context.Context, msg *types.MessageData) {
+	if msg.OwnMetricsConnSettings != nil {
+		agent.logger.Debugf(ctx, "got own metrics connection settings")
+	}
+	if msg.AgentIdentification != nil {
+		agent.logger.Debugf(ctx, "got agent identification")
+		uid, err := uuid.FromBytes(msg.AgentIdentification.NewInstanceUid)
+		if err != nil {
+			agent.logger.Errorf(ctx, "Failed to parse new instance uid: %v", err)
+			return
+		}
+		agent.updateAgentIdentity(ctx, uid)
+	}
+
+	agent.updateRemoteConfig(ctx, msg)
+
+}
+
+func (agent *Agent) onOpampConnectionSettings(ctx context.Context, settings *protobufs.OpAMPConnectionSettings) error {
+	agent.logger.Debugf(ctx, "got connection settings")
+	return nil
+}
+
+func (agent *Agent) updateRemoteConfig(ctx context.Context, msg *types.MessageData) {
+	if msg.RemoteConfig == nil {
+		agent.logger.Debugf(context.Background(), "updateRemoteConfig: no remote config in message")
+		return
+	}
+
+	if msg.RemoteConfig.GetConfig().GetConfigMap() != nil {
+		// deserialize the config and call ReloadConfig
+		agent.logger.Debugf(ctx, "onMessage got remote config: %v", msg)
+
+		confMap := msg.RemoteConfig.GetConfig().GetConfigMap()
+
+		if !agent.isConfigChanged(msg.RemoteConfig.GetConfigHash()) {
+			agent.logger.Debugf(ctx, "onMessage remote config is the same as the last one, skipping")
+			return
+		}
+
+		var opts []config.ReloadedConfigDataOption
+		if c, ok := confMap["refinery_rules"]; ok {
+			opts = append(opts, config.WithRulesData(config.NewConfigData(c.GetBody(), config.FormatYAML, "opamp://rules")))
+		}
+		if c, ok := confMap["refinery_config"]; ok {
+			opts = append(opts, config.WithConfigData(config.NewConfigData(c.GetBody(), config.FormatYAML, "opamp://config")))
+		}
+
+		agent.lastRemoteConfigReceived = msg.RemoteConfig
+		agent.logger.Debugf(ctx, "onMessage config opts: %v", opts)
+		if len(opts) > 0 {
+			agent.reportConfigStatus(protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLYING, "")
+			err := agent.effectiveConfig.Reload(opts...)
+			if err != nil {
+				agent.logger.Errorf(ctx, "Failed to reload config: %v", err)
+				agent.reportConfigStatus(protobufs.RemoteConfigStatuses_RemoteConfigStatuses_FAILED, err.Error())
+			} else {
+				agent.logger.Logger.Info().Logf("Successfully reloaded config")
+				agent.reportConfigStatus(protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED, "")
+			}
+		}
+	}
+}
+
+func (agent *Agent) isConfigChanged(newConfigHash []byte) bool {
+	if agent.lastRemoteConfigReceived == nil {
+		return true
+	}
+	if !bytes.Equal(agent.remoteConfigStatus.GetLastRemoteConfigHash(), newConfigHash) {
+		return true
+	}
+	if agent.remoteConfigStatus.GetStatus() == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED {
+		return false
+	}
+
+	return true
+
+}
+
+func (agent *Agent) updateAgentIdentity(ctx context.Context, instanceId uuid.UUID) {
+	agent.logger.Debugf(ctx, "Agent identify is being changed from id=%v to id=%v",
+		agent.instanceId,
+		instanceId)
+	agent.instanceId = instanceId
+
+	// TODO: update metrics setting when identity changes
+}
+
+func healthMessage(healthy bool) *protobufs.ComponentHealth {
+	return &protobufs.ComponentHealth{
+		Healthy: healthy,
+	}
+}
diff --git a/agent/logger.go b/agent/logger.go
new file mode 100644
index 0000000..6e2a736
--- /dev/null
+++ b/agent/logger.go
@@ -0,0 +1,22 @@
+package agent
+
+import (
+	"context"
+
+	"github.com/honeycombio/refinery/logger"
+	"github.com/open-telemetry/opamp-go/client/types"
+)
+
+var _ types.Logger = &Logger{}
+
+type Logger struct {
+	Logger logger.Logger
+}
+
+func (l *Logger) Debugf(_ context.Context, format string, v ...interface{}) {
+	l.Logger.Debug().Logf(format, v...)
+}
+
+func (l *Logger) Errorf(_ context.Context, format string, v ...interface{}) {
+	l.Logger.Error().Logf(format, v...)
+}
diff --git a/agent/otlp_metrics.go b/agent/otlp_metrics.go
new file mode 100644
index 0000000..7e9e814
--- /dev/null
+++ b/agent/otlp_metrics.go
@@ -0,0 +1,55 @@
+package agent
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+type otlpMetrics struct {
+	metrics pmetric.Metrics
+	ms      pmetric.Sum
+}
+
+func newOTLPMetrics(serviceName, version, hostname string) *otlpMetrics {
+	metrics := pmetric.NewMetrics()
+	rm := metrics.ResourceMetrics().AppendEmpty()
+	resourceAttrs := rm.Resource().Attributes()
+	resourceAttrs.PutStr("service.name", serviceName)
+	resourceAttrs.PutStr("service.version", version)
+	resourceAttrs.PutStr("host.name", hostname)
+	sm := rm.ScopeMetrics().AppendEmpty()
+	ms := sm.Metrics().AppendEmpty()
+	ms.SetName("bytes_received")
+	sum := ms.SetEmptySum()
+	sum.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+	return &otlpMetrics{
+		metrics: metrics,
+		ms:      sum,
+	}
+}
+
+func (om *otlpMetrics) addOTLPSum(timestamp time.Time, value float64, signal usageSignal) error {
+	intVal, err := convertFloat64ToInt64(value)
+	if err != nil {
+		return err
+	}
+	d := om.ms.DataPoints().AppendEmpty()
+	d.SetTimestamp(pcommon.NewTimestampFromTime(timestamp))
+	d.SetIntValue(intVal)
+	d.Attributes().PutStr("signal", string(signal))
+	return nil
+}
+
+func convertFloat64ToInt64(value float64) (int64, error) {
+	if value > math.MaxInt64 {
+		return 0, fmt.Errorf("value %f is too large to convert to int64", value)
+	}
+	if value < 0 {
+		return 0, fmt.Errorf("invalid negative value %f", value)
+	}
+	return int64(value), nil
+}
diff --git a/agent/usage_report.go b/agent/usage_report.go
new file mode 100644
index 0000000..0d947a0
--- /dev/null
+++ b/agent/usage_report.go
@@ -0,0 +1,94 @@
+package agent
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+var (
+	errNoData     = errors.New("no data to report")
+	jsonMarshaler = &pmetric.JSONMarshaler{}
+)
+
+// usageTracker is a store for usage data. It keeps track of the last usage data and the delta between each update.
+type usageTracker struct {
+	// lastUsageData is a map of the last cumulative usage data for each signal.
+	lastUsageData map[usageSignal]float64
+
+	mut               sync.Mutex
+	lastDataPoints    map[usageSignal]float64
+	currentDataPoints map[usageSignal]float64
+}
+
+func newUsageTracker() *usageTracker {
+	return &usageTracker{
+		lastUsageData:     make(map[usageSignal]float64),
+		currentDataPoints: make(map[usageSignal]float64),
+		lastDataPoints:    make(map[usageSignal]float64),
+	}
+}
+
+// Add records the current cumulative usage and calculates the delta between the last update.
+func (ur *usageTracker) Add(signal usageSignal, data float64) {
+	ur.mut.Lock()
+	defer ur.mut.Unlock()
+
+	if data == 0 {
+		return
+	}
+
+	deltaTraceUsage := data - ur.lastUsageData[signal]
+	ur.currentDataPoints[signal] += deltaTraceUsage
+	ur.lastUsageData[signal] = data
+}
+
+// NewReport creates a new usage report with the current delta usage data.
+func (ur *usageTracker) NewReport(serviceName, version, hostname string, now time.Time) ([]byte, error) {
+	ur.mut.Lock()
+	defer ur.mut.Unlock()
+
+	if len(ur.currentDataPoints) == 0 && len(ur.lastDataPoints) == 0 {
+		return nil, errNoData
+	}
+
+	otlpMetrics := newOTLPMetrics(serviceName, version, hostname)
+	for signal, usage := range ur.currentDataPoints {
+		err := otlpMetrics.addOTLPSum(now, usage, signal)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	for signal, usage := range ur.lastDataPoints {
+		err := otlpMetrics.addOTLPSum(now, usage, signal)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	data, err := jsonMarshaler.MarshalMetrics(otlpMetrics.metrics)
+	if err != nil {
+		return nil, err
+	}
+	// clear the current data points and keep the last data points until we know the report was sent
+	ur.lastDataPoints = ur.currentDataPoints
+	ur.currentDataPoints = make(map[usageSignal]float64)
+	return data, nil
+}
+
+// completeSend clears the last data points after the report is sent.
+func (ur *usageTracker) completeSend() {
+	ur.mut.Lock()
+	defer ur.mut.Unlock()
+	ur.lastDataPoints = make(map[usageSignal]float64)
+}
+
+type usageSignal string
+
+var (
+	signal_traces usageSignal = "traces"
+	signal_logs   usageSignal = "logs"
+)
diff --git a/app/app.go b/app/app.go
index 8469454..486fc07 100644
--- a/app/app.go
+++ b/app/app.go
@@ -1,6 +1,9 @@
 package app
 
 import (
+	"context"
+
+	"github.com/honeycombio/refinery/agent"
 	"github.com/honeycombio/refinery/collect"
 	"github.com/honeycombio/refinery/config"
 	"github.com/honeycombio/refinery/logger"
@@ -19,6 +22,10 @@ type App struct {
 	// Version is the build ID for Refinery so that the running process may answer
 	// requests for the version
 	Version string
+
+	// opampAgent communicates with the opamp server configured if OpAMP
+	// support is enabled via configuration.
+	opampAgent *agent.Agent
 }
 
 // Start on the App object should block until the proxy is shutting down. After
@@ -58,11 +65,19 @@ func (a *App) Start() error {
 	a.IncomingRouter.LnS("incoming")
 	a.PeerRouter.LnS("peer")
 
+	// only enable the opamp agent if it's configured
+	if a.Config.GetOpAMPConfig().Enabled {
+		a.opampAgent = agent.NewAgent(agent.Logger{Logger: a.Logger}, a.Version, a.Config, a.Metrics, a.IncomingRouter.Health)
+	}
+
 	return nil
 }
 
 func (a *App) Stop() error {
 	a.Logger.Debug().Logf("Shutting down App...")
+	if a.Config.GetOpAMPConfig().Enabled {
+		a.opampAgent.Stop(context.Background())
+	}
 	return nil
 }
 
diff --git a/cmd/refinery/main.go b/cmd/refinery/main.go
index e30c119..2f63dd3 100644
--- a/cmd/refinery/main.go
+++ b/cmd/refinery/main.go
@@ -99,11 +99,7 @@ func main() {
 		Version: version,
 	}
 
-	c, err := config.NewConfig(opts, func(err error) {
-		if a.Logger != nil {
-			a.Logger.Error().WithField("error", err).Logf("error loading config")
-		}
-	})
+	c, err := config.NewConfig(opts)
 	if err != nil {
 		fmt.Printf("%+v\n", err)
 		os.Exit(1)
diff --git a/config.md b/config.md
index f215fef..8341ba8 100644
--- a/config.md
+++ b/config.md
@@ -27,6 +27,7 @@ The remainder of this document describes the sections within the file and the fi
 ## Table of Contents
 - [General Configuration](#general-configuration)
 - [Network Configuration](#network-configuration)
+- [OpAMP Configuration](#opamp-configuration)
 - [Access Key Configuration](#access-key-configuration)
 - [Refinery Telemetry](#refinery-telemetry)
 - [Traces](#traces)
@@ -150,6 +151,48 @@ This setting is the destination to which Refinery sends all events that it decid
 - Environment variable: `REFINERY_HONEYCOMB_API`
 - Command line switch: `--honeycomb-api`
 
+## OpAMP Configuration
+
+`OpAMP` contains OpAMP configuration options.
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+### `Enabled`
+
+Enabled controls whether to enable OpAMP support.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Not eligible for live reload.
+- Type: `bool`
+
+### `RecordUsage`
+
+RecordUsage controls whether to record usage metrics.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Eligible for live reload.
+- Type: `bool`
+- Default: `true`
+
+### `Endpoint`
+
+Endpoint is the URL of the OpAMP server for this client.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Not eligible for live reload.
+- Type: `string`
+- Default: `wss://127.0.0.1:4320/v1/opamp`
+
 ## Access Key Configuration
 
 `AccessKeys` contains access keys -- API keys that the proxy will treat specially, and other flags that control how the proxy handles API keys.
diff --git a/config/cmdenv.go b/config/cmdenv.go
index 6ed264e..8ecb93f 100644
--- a/config/cmdenv.go
+++ b/config/cmdenv.go
@@ -40,6 +40,7 @@ type CmdEnv struct {
 	HoneycombAPIKey       string     `long:"honeycomb-api-key" env:"REFINERY_HONEYCOMB_API_KEY" description:"Honeycomb API key (for logger and metrics). Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
 	HoneycombLoggerAPIKey string     `long:"logger-api-key" env:"REFINERY_HONEYCOMB_LOGGER_API_KEY" description:"Honeycomb logger API key. Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
 	LegacyMetricsAPIKey   string     `long:"legacy-metrics-api-key" env:"REFINERY_HONEYCOMB_METRICS_API_KEY" description:"API key for legacy Honeycomb metrics. Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
+	OpAMPEndpoint         string     `long:"opamp-server-url" env:"REFINERY_OPAMP_ENDPOINT" description:"URL of the OpAMP server to use for remote management."`
 	OTelMetricsAPIKey     string     `long:"otel-metrics-api-key" env:"REFINERY_OTEL_METRICS_API_KEY" description:"API key for OTel metrics if being sent to Honeycomb. Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
 	OTelTracesAPIKey      string     `long:"otel-traces-api-key" env:"REFINERY_OTEL_TRACES_API_KEY" description:"API key for OTel traces if being sent to Honeycomb. Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
 	QueryAuthToken        string     `long:"query-auth-token" env:"REFINERY_QUERY_AUTH_TOKEN" description:"Token for debug/management queries. Setting this value via a flag may expose credentials - it is recommended to use the env var or a configuration file."`
diff --git a/config/config.go b/config/config.go
index 1fa5998..2099a20 100644
--- a/config/config.go
+++ b/config/config.go
@@ -24,7 +24,7 @@ type Config interface {
 
 	// Reload forces the config to attempt to reload its values. If the config
 	// checksum has changed, the reload callbacks will be called.
-	Reload()
+	Reload(opts ...ReloadedConfigDataOption) error
 
 	// GetHashes returns the current config and rule hashes
 	GetHashes() (cfg string, rules string)
@@ -156,6 +156,8 @@ type Config interface {
 	GetTraceIdFieldNames() []string
 
 	GetParentIdFieldNames() []string
+
+	GetOpAMPConfig() OpAMPConfig
 }
 
 type ConfigReloadCallback func(configHash, ruleCfgHash string)
@@ -166,3 +168,24 @@ type ConfigMetadata struct {
 	Hash     string `json:"hash"`
 	LoadedAt string `json:"loaded_at"`
 }
+
+// ReloadedConfigData holds the new config data that will be applied to
+// the Config instance through `Reload` method.
+type ReloadedConfigData struct {
+	configs []configData
+	rules   []configData
+}
+
+// ReloadedConfigDataOption is a function that allows setting the new config data
+type ReloadedConfigDataOption func(*ReloadedConfigData)
+
+func WithConfigData(in configData) ReloadedConfigDataOption {
+	return func(c *ReloadedConfigData) {
+		c.configs = append(c.configs, in)
+	}
+}
+func WithRulesData(in configData) ReloadedConfigDataOption {
+	return func(c *ReloadedConfigData) {
+		c.rules = append(c.rules, in)
+	}
+}
diff --git a/config/configLoadHelpers.go b/config/configLoadHelpers.go
index 6150360..e5566b8 100644
--- a/config/configLoadHelpers.go
+++ b/config/configLoadHelpers.go
@@ -134,21 +134,29 @@ type configData struct {
 	location string
 }
 
+func NewConfigData(data []byte, format Format, location string) configData {
+	return configData{
+		data:     data,
+		format:   format,
+		location: location,
+	}
+}
+
 // getConfigDataForLocations returns a slice of configData grabbed from each location.
 func getConfigDataForLocations(locations []string) ([]configData, error) {
-	results := make([]configData, len(locations))
-	for i, location := range locations {
+	results := make([]configData, 0, len(locations))
+	for _, location := range locations {
 		// trim leading and trailing whitespace just in case
 		location := strings.TrimSpace(location)
 		data, format, err := getBytesFor(location)
 		if err != nil {
 			return nil, err
 		}
-		results[i] = configData{
+		results = append(results, configData{
 			data:     data,
 			format:   format,
 			location: location,
-		}
+		})
 	}
 	return results, nil
 }
diff --git a/config/config_serializer.go b/config/config_serializer.go
new file mode 100644
index 0000000..ca4b05d
--- /dev/null
+++ b/config/config_serializer.go
@@ -0,0 +1,103 @@
+package config
+
+import (
+	"fmt"
+
+	"gopkg.in/yaml.v3"
+)
+
+// SerializeToYAML serializes the Config to a YAML string
+func SerializeToYAML(cfg Config) (string, error) {
+	// Create a configContents struct and populate it from the Config interface
+	contents := populateConfigContents(cfg)
+
+	yamlBytes, err := yaml.Marshal(contents)
+	if err != nil {
+		return "", fmt.Errorf("error serializing config to YAML: %w", err)
+	}
+	return string(yamlBytes), nil
+}
+
+// populateConfigContents creates a configContents struct from a Config interface
+func populateConfigContents(cfg Config) configContents {
+	opAMPConfig := cfg.GetOpAMPConfig()
+
+	return configContents{
+		General: cfg.GetGeneralConfig(),
+		Network: NetworkConfig{
+			ListenAddr:      cfg.GetListenAddr(),
+			PeerListenAddr:  cfg.GetPeerListenAddr(),
+			HoneycombAPI:    cfg.GetHoneycombAPI(),
+			HTTPIdleTimeout: Duration(cfg.GetHTTPIdleTimeout()),
+		},
+		OpAMP: OpAMPConfig{
+			Endpoint:    opAMPConfig.Endpoint,
+			Enabled:     opAMPConfig.Enabled,
+			RecordUsage: getDefaultTrueValue(cfg.GetOpAMPConfig().RecordUsage.Get()),
+		},
+		AccessKeys: cfg.GetAccessKeyConfig(),
+		Telemetry:  getRefineryTelemetryConfig(cfg),
+		Traces:     cfg.GetTracesConfig(),
+		Debugging: DebuggingConfig{
+			DebugServiceAddr:      cfg.GetDebugServiceAddr(),
+			QueryAuthToken:        cfg.GetQueryAuthToken(),
+			AdditionalErrorFields: cfg.GetAdditionalErrorFields(),
+			DryRun:                cfg.GetIsDryRun(),
+		},
+		Logger: LoggerConfig{
+			Type:  cfg.GetLoggerType(),
+			Level: cfg.GetLoggerLevel(),
+		},
+		HoneycombLogger:     cfg.GetHoneycombLoggerConfig(),
+		StdoutLogger:        cfg.GetStdoutLoggerConfig(),
+		PrometheusMetrics:   cfg.GetPrometheusMetricsConfig(),
+		LegacyMetrics:       cfg.GetLegacyMetricsConfig(),
+		OTelMetrics:         cfg.GetOTelMetricsConfig(),
+		OTelTracing:         cfg.GetOTelTracingConfig(),
+		PeerManagement:      getPeerManagementConfig(cfg),
+		RedisPeerManagement: cfg.GetRedisPeerManagement(),
+		Collection:          cfg.GetCollectionConfig(),
+		BufferSizes: BufferSizeConfig{
+			UpstreamBufferSize: cfg.GetUpstreamBufferSize(),
+			PeerBufferSize:     cfg.GetPeerBufferSize(),
+		},
+		Specialized: SpecializedConfig{
+			EnvironmentCacheTTL:       Duration(cfg.GetEnvironmentCacheTTL()),
+			CompressPeerCommunication: getDefaultTrueValue(cfg.GetCompressPeerCommunication()),
+			AdditionalAttributes:      cfg.GetAdditionalAttributes(),
+		},
+		IDFieldNames: IDFieldsConfig{
+			TraceNames:  cfg.GetTraceIdFieldNames(),
+			ParentNames: cfg.GetParentIdFieldNames(),
+		},
+		GRPCServerParameters: cfg.GetGRPCConfig(),
+		SampleCache:          cfg.GetSampleCacheConfig(),
+		StressRelief:         cfg.GetStressReliefConfig(),
+	}
+}
+
+// Helper function to get RefineryTelemetryConfig
+func getRefineryTelemetryConfig(cfg Config) RefineryTelemetryConfig {
+	return RefineryTelemetryConfig{
+		AddRuleReasonToTrace:   cfg.GetAddRuleReasonToTrace(),
+		AddSpanCountToRoot:     getDefaultTrueValue(cfg.GetAddSpanCountToRoot()),
+		AddCountsToRoot:        cfg.GetAddCountsToRoot(),
+		AddHostMetadataToTrace: getDefaultTrueValue(cfg.GetAddHostMetadataToTrace()),
+	}
+}
+
+// Helper function to get PeerManagementConfig
+func getPeerManagementConfig(cfg Config) PeerManagementConfig {
+	return PeerManagementConfig{
+		Type:                    cfg.GetPeerManagementType(),
+		Identifier:              cfg.GetRedisIdentifier(),
+		IdentifierInterfaceName: cfg.GetIdentifierInterfaceName(),
+		UseIPV6Identifier:       cfg.GetUseIPV6Identifier(),
+		Peers:                   cfg.GetPeers(),
+	}
+}
+
+func getDefaultTrueValue(value bool) *DefaultTrue {
+	dt := DefaultTrue(value)
+	return &dt
+}
diff --git a/config/file_config.go b/config/file_config.go
index c82634f..37cd28a 100644
--- a/config/file_config.go
+++ b/config/file_config.go
@@ -31,15 +31,14 @@ func (d *Duration) UnmarshalText(text []byte) error {
 }
 
 type fileConfig struct {
-	mainConfig    *configContents
-	mainHash      string
-	rulesConfig   *V2SamplerConfig
-	rulesHash     string
-	opts          *CmdEnv
-	callbacks     []ConfigReloadCallback
-	errorCallback func(error)
-	mux           sync.RWMutex
-	lastLoadTime  time.Time
+	mainConfig   *configContents
+	mainHash     string
+	rulesConfig  *V2SamplerConfig
+	rulesHash    string
+	opts         *CmdEnv
+	callbacks    []ConfigReloadCallback
+	mux          sync.RWMutex
+	lastLoadTime time.Time
 }
 
 // ensure that fileConfig implements Config
@@ -48,6 +47,7 @@ var _ Config = (*fileConfig)(nil)
 type configContents struct {
 	General              GeneralConfig             `yaml:"General"`
 	Network              NetworkConfig             `yaml:"Network"`
+	OpAMP                OpAMPConfig               `yaml:"OpAMP"`
 	AccessKeys           AccessKeyConfig           `yaml:"AccessKeys"`
 	Telemetry            RefineryTelemetryConfig   `yaml:"RefineryTelemetry"`
 	Traces               TracesConfig              `yaml:"Traces"`
@@ -77,6 +77,12 @@ type GeneralConfig struct {
 	ConfigReloadInterval Duration `yaml:"ConfigReloadInterval" default:"15s"`
 }
 
+type OpAMPConfig struct {
+	Endpoint    string       `yaml:"Endpoint" cmdenv:"OpAMPEndpoint" default:"wss://127.0.0.1:4320/v1/opamp"`
+	Enabled     bool         `yaml:"Enabled" default:"false"`
+	RecordUsage *DefaultTrue `yaml:"RecordUsage" default:"true"`
+}
+
 type NetworkConfig struct {
 	ListenAddr      string   `yaml:"ListenAddr" default:"0.0.0.0:8080" cmdenv:"HTTPListenAddr"`
 	PeerListenAddr  string   `yaml:"PeerListenAddr" default:"0.0.0.0:8081" cmdenv:"PeerListenAddr"`
@@ -447,24 +453,27 @@ func (e *FileConfigError) Error() string {
 	return msg.String()
 }
 
-// newFileConfig does the work of creating and loading the start of a config object
-// from the given arguments.
-// It's used by both the main init as well as the reload code.
-// In order to do proper validation, we actually process the data twice -- once as
-// a map, and once as the actual config object.
-func newFileConfig(opts *CmdEnv) (*fileConfig, error) {
-	configData, err := getConfigDataForLocations(opts.ConfigLocations)
+func newConfigAndRules(opts *CmdEnv) ([]configData, []configData, error) {
+	cData, err := getConfigDataForLocations(opts.ConfigLocations)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	rulesData, err := getConfigDataForLocations(opts.RulesLocations)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
+	return cData, rulesData, err
+}
 
+// newFileConfig does the work of creating and loading the start of a config object
+// from the given arguments.
+// It's used by both the main init as well as the reload code.
+// In order to do proper validation, we actually process the data twice -- once as
+// a map, and once as the actual config object.
+func newFileConfig(opts *CmdEnv, cData, rulesData []configData) (*fileConfig, error) {
 	// If we're not validating, skip this part
 	if !opts.NoValidate {
-		cfgFails, err := validateConfigs(configData, opts)
+		cfgFails, err := validateConfigs(cData, opts)
 		if err != nil {
 			return nil, err
 		}
@@ -486,7 +495,7 @@ func newFileConfig(opts *CmdEnv) (*fileConfig, error) {
 
 	// Now load the files
 	mainconf := &configContents{}
-	mainhash, err := applyConfigInto(mainconf, configData, opts)
+	mainhash, err := applyConfigInto(mainconf, cData, opts)
 	if err != nil {
 		return nil, err
 	}
@@ -524,8 +533,13 @@ func writeYAMLToFile(data any, filename string) error {
 // nil, it uses the command line arguments.
 // It also dumps the config and rules to the given files, if specified, which
 // will cause the program to exit.
-func NewConfig(opts *CmdEnv, errorCallback func(error)) (Config, error) {
-	cfg, err := newFileConfig(opts)
+func NewConfig(opts *CmdEnv) (Config, error) {
+	cData, rData, err := newConfigAndRules(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg, err := newFileConfig(opts, cData, rData)
 	// only exit if we have no config at all; if it fails validation, we'll
 	// do the rest and return it anyway
 	if err != nil && cfg == nil {
@@ -549,24 +563,36 @@ func NewConfig(opts *CmdEnv, errorCallback func(error)) (Config, error) {
 	}
 
 	cfg.callbacks = make([]ConfigReloadCallback, 0)
-	cfg.errorCallback = errorCallback
 
 	return cfg, err
 }
 
 // Reload attempts to reload the configuration; if it has changed, it stores the
 // new data and calls the reload callbacks.
-func (f *fileConfig) Reload() {
+func (f *fileConfig) Reload(opts ...ReloadedConfigDataOption) error {
+	cData, rData, err := newConfigAndRules(f.opts)
+	if err != nil {
+		return err
+	}
+
+	newData := &ReloadedConfigData{
+		configs: cData,
+		rules:   rData,
+	}
+
+	for _, opt := range opts {
+		opt(newData)
+	}
+
 	// reread the configs
-	cfg, err := newFileConfig(f.opts)
+	cfg, err := newFileConfig(f.opts, newData.configs, newData.rules)
 	if err != nil {
-		f.errorCallback(err)
-		return
+		return err
 	}
 
 	// if nothing's changed, we're fine
 	if f.mainHash == cfg.mainHash && f.rulesHash == cfg.rulesHash {
-		return
+		return nil
 	}
 
 	// otherwise, update our state and call the callbacks
@@ -580,6 +606,7 @@ func (f *fileConfig) Reload() {
 	for _, cb := range f.callbacks {
 		cb(cfg.mainHash, cfg.rulesHash)
 	}
+	return nil
 }
 
 // GetHashes returns the current hash values for the main and rules configs.
@@ -925,6 +952,17 @@ func (f *fileConfig) GetEnvironmentCacheTTL() time.Duration {
 	return time.Duration(f.mainConfig.Specialized.EnvironmentCacheTTL)
 }
 
+func (f *fileConfig) GetOpAMPConfig() OpAMPConfig {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return OpAMPConfig{
+		Enabled:     f.mainConfig.OpAMP.Enabled,
+		Endpoint:    f.mainConfig.OpAMP.Endpoint,
+		RecordUsage: f.mainConfig.OpAMP.RecordUsage,
+	}
+}
+
 func (f *fileConfig) GetOTelTracingConfig() OTelTracingConfig {
 	f.mux.RLock()
 	defer f.mux.RUnlock()
diff --git a/config/metadata/configMeta.yaml b/config/metadata/configMeta.yaml
index 0453b36..aea7663 100644
--- a/config/metadata/configMeta.yaml
+++ b/config/metadata/configMeta.yaml
@@ -142,6 +142,43 @@ groups:
         description: >
           This setting is the destination to which Refinery sends all events
           that it decides to keep.
+  - name: OpAMP
+    title: "OpAMP Configuration"
+    description: > 
+      contains OpAMP configuration options. OpAMP support is an experimental
+      feature, subject to breaking changes at any time. Not intended or
+      supported for customer use. **FOR INTERNAL USE ONLY**
+    fields:
+      - name: Enabled
+        type: bool
+        valuetype: nondefault
+        default: false
+        reload: false
+        firstversion: v2.9.4
+        summary: controls whether to enable OpAMP support.
+        description: >
+          OpAMP support is an experimental feature, subject to breaking changes at
+          any time. Not intended or supported for customer use. **FOR INTERNAL USE ONLY**
+      - name: RecordUsage
+        type: defaulttrue
+        valuetype: nondefault
+        default: true
+        reload: true
+        firstversion: v2.9.4
+        summary: controls whether to record usage metrics.
+        description: >
+          OpAMP support is an experimental feature, subject to breaking changes at
+          any time. Not intended or supported for customer use. **FOR INTERNAL USE ONLY**
+      - name: Endpoint
+        type: string
+        valuetype: assigndefault
+        default: "wss://127.0.0.1:4320/v1/opamp"
+        reload: false
+        firstversion: v2.9.4
+        summary: is the URL of the OpAMP server for this client.
+        description: >
+          OpAMP support is an experimental feature, subject to breaking changes at
+          any time. Not intended or supported for customer use. **FOR INTERNAL USE ONLY**
 
   - name: AccessKeys
     title: "Access Key Configuration"
@@ -746,6 +783,7 @@ groups:
         description: >
           Between 1 and 60 seconds is typical.
 
+
   - name: OTelMetrics
     title: "OpenTelemetry Metrics"
     description: >
diff --git a/config/mock.go b/config/mock.go
index 34497a3..08d84bd 100644
--- a/config/mock.go
+++ b/config/mock.go
@@ -33,6 +33,7 @@ type MockConfig struct {
 	GetGeneralConfigVal              GeneralConfig
 	GetLegacyMetricsConfigVal        LegacyMetricsConfig
 	GetPrometheusMetricsConfigVal    PrometheusMetricsConfig
+	GetOpAmpConfigVal                OpAMPConfig
 	GetOTelMetricsConfigVal          OTelMetricsConfig
 	GetOTelTracingConfigVal          OTelTracingConfig
 	GetUpstreamBufferSizeVal         int
@@ -69,13 +70,15 @@ type MockConfig struct {
 // assert that MockConfig implements Config
 var _ Config = (*MockConfig)(nil)
 
-func (m *MockConfig) Reload() {
+func (m *MockConfig) Reload(opts ...ReloadedConfigDataOption) error {
 	m.Mux.RLock()
 	defer m.Mux.RUnlock()
 
 	for _, callback := range m.Callbacks {
 		callback("", "")
 	}
+
+	return nil
 }
 
 func (m *MockConfig) RegisterReloadCallback(callback ConfigReloadCallback) {
@@ -447,3 +450,10 @@ func (f *MockConfig) GetAdditionalAttributes() map[string]string {
 
 	return f.AdditionalAttributes
 }
+
+func (f *MockConfig) GetOpAMPConfig() OpAMPConfig {
+	f.Mux.RLock()
+	defer f.Mux.RUnlock()
+
+	return f.GetOpAmpConfigVal
+}
diff --git a/config_complete.yaml b/config_complete.yaml
index ef921d9..091cdb9 100644
--- a/config_complete.yaml
+++ b/config_complete.yaml
@@ -128,6 +128,44 @@ Network:
     ## Eligible for live reload.
     # HoneycombAPI: "https://api.honeycomb.io"
 
+#########################
+## OpAMP Configuration ##
+#########################
+OpAMP:
+  ## OpAMP contains OpAMP configuration options. OpAMP support is an
+  ## experimental feature, subject to breaking changes at any time. Not
+  ## intended or supported for customer use. **FOR INTERNAL USE ONLY**
+  ##
+  ####
+    ## Enabled controls whether to enable OpAMP support.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## Not eligible for live reload.
+    # Enabled: false
+
+    ## RecordUsage controls whether to record usage metrics.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## default: true
+    ## Eligible for live reload.
+    # RecordUsage: true
+
+    ## Endpoint is the URL of the OpAMP server for this client.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## default: wss://127.0.0.1:4320/v1/opamp
+    ## Not eligible for live reload.
+    Endpoint: wss://127.0.0.1:4320/v1/opamp
+
 ##############################
 ## Access Key Configuration ##
 ##############################
diff --git a/go.mod b/go.mod
index a0efb0c..89303e6 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,7 @@
 module github.com/honeycombio/refinery
 
 go 1.24.0
+
 require (
 	github.com/agnivade/levenshtein v1.2.1
 	github.com/creasty/defaults v1.8.0
@@ -19,6 +20,8 @@ require (
 	github.com/jonboulle/clockwork v0.5.0
 	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.18.0
+	github.com/open-telemetry/opamp-go v0.19.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.123.0
 	github.com/panmari/cuckoofilter v1.0.6
 	github.com/pelletier/go-toml/v2 v2.2.3
 	github.com/pkg/errors v0.9.1
@@ -51,14 +54,18 @@ require (
 )
 
 require (
+	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/go-licenses/v2 v2.0.0-alpha.1 // indirect
 	github.com/google/licenseclassifier/v2 v2.0.0 // indirect
+	github.com/gorilla/websocket v1.5.3 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.123.0 // indirect
 	github.com/otiai10/copy v1.10.0 // indirect
 	github.com/sergi/go-diff v1.2.0 // indirect
 	github.com/spf13/cobra v1.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	k8s.io/klog/v2 v2.90.1 // indirect
@@ -91,6 +98,7 @@ require (
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.29.0
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/mod v0.24.0 // indirect
 	golang.org/x/net v0.37.0 // indirect
diff --git a/go.sum b/go.sum
index c3a7051..0891d78 100644
--- a/go.sum
+++ b/go.sum
@@ -63,6 +63,8 @@ github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY=
 github.com/go-logr/logr v1.4.2/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
 github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
+github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
@@ -107,6 +109,8 @@ 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/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
 github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grafana/pyroscope-go/godeltaprof v0.1.8 h1:iwOtYXeeVSAeYefJNaxDytgjKtUuKQbJqgAIjlnicKg=
 github.com/grafana/pyroscope-go/godeltaprof v0.1.8/go.mod h1:2+l7K7twW49Ct4wFluZD3tZ6e0SjanjcUUBPVD/UuGU=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.1 h1:e9Rjr40Z98/clHv5Yg79Is0NtosR5LXRvdr7o/6NwbA=
@@ -129,6 +133,8 @@ github.com/jonboulle/clockwork v0.5.0 h1:Hyh9A8u51kptdkR+cqRpT1EebBwTn1oK9YfGYbd
 github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7XN3SzBPjZF60=
 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/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
 github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -147,6 +153,14 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
 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.19.0 h1:8LvQKDwqi+BU3Yy159SU31e2XB0vgnk+PN45pnKilPs=
+github.com/open-telemetry/opamp-go v0.19.0/go.mod h1:9/1G6T5dnJz4cJtoYSr6AX18kHdOxnxxETJPZSHyEUg=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.123.0 h1:bt/LdA/x3Nem2rm7xqw1S2aMwpx76+y0M4ITBxgMlvs=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.123.0/go.mod h1:EAzpBsYIXoXQbyVTkvfOksaiq+jctnoxnPevB+/xsqw=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.123.0 h1:zISmQ6FC0ZySjbNA48KvHLrHWYaSv7vXBNqAxdqA5cE=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest v0.123.0/go.mod h1:yv3tSTQdVXHvNT+OZ95Drnwb5ntWdlhTthKAi/sB5QY=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.123.0 h1:dFNC9MEqpyVfnfu29Yxw69+Na1OkEXg/NaW9mo/fbP4=
+github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.123.0/go.mod h1:hXC2pbybf1KcGnXB5RvHhR++/QMgS1HP0iFf/DSXBFg=
 github.com/otiai10/copy v1.10.0 h1:znyI7l134wNg/wDktoVQPxPkgvhDfGCYUasey+h0rDQ=
 github.com/otiai10/copy v1.10.0/go.mod h1:rSaLseMUsZFFbsFGc7wCJnnkTAvdc5L6VWxPE4308Ww=
 github.com/otiai10/mint v1.5.1 h1:XaPLeE+9vGbuyEHem1JNk3bYc7KKqyI/na0/mLd/Kks=
@@ -193,6 +207,8 @@ github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
@@ -211,10 +227,14 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
 github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
+github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=
 go.opencensus.io v0.24.0/go.mod h1:vNK8G9p7aAivkbmorf4v+7Hgx+Zs0yY+0fOtgBfjQKo=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/collector/pdata v1.29.0 h1:ZXSZ2fROdAEbv4JKKiCspBpjIjYZ5XaNt71LNH4RpQw=
+go.opentelemetry.io/collector/pdata v1.29.0/go.mod h1:9kb3zMtLFXBPA6WGWkBHbkFwlwwYL/OHk1m0ASWZpeY=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0 h1:x7wzEgXfnzJcHDwStJT+mxOz4etr2EcexjqhBvmoakw=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.60.0/go.mod h1:rg+RlpR5dKwaS95IyyZqj5Wd4E13lk/msnTS0Xl9lJM=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU=
@@ -242,6 +262,7 @@ go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20231127185646-65229373498e h1:Gvh4YaCaXNs6dKTlfgismwWZKyjVZXwOPfIyUaqU3No=
@@ -249,6 +270,8 @@ golang.org/x/exp v0.0.0-20231127185646-65229373498e/go.mod h1:iRJReGqOEeBhDZGkGb
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU=
 golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -256,6 +279,9 @@ golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.37.0 h1:1zLorHbz+LYj7MQlSf1+2tPIIgibq2eL5xkrGk6f+2c=
 golang.org/x/net v0.37.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
@@ -263,6 +289,8 @@ golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAG
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
 golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -281,9 +309,15 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
+golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.31.0 h1:0EedkvKDbh+qistFTd0Bcwe/YLh4vHwWEkiI0toFIBU=
 golang.org/x/tools v0.31.0/go.mod h1:naFTU+Cev749tSJRXJlna0T3WxKvb1kWEx15xA4SdmQ=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
diff --git a/internal/configwatcher/watcher.go b/internal/configwatcher/watcher.go
index aa6f5df..670edf0 100644
--- a/internal/configwatcher/watcher.go
+++ b/internal/configwatcher/watcher.go
@@ -9,6 +9,7 @@ import (
 	"github.com/facebookgo/startstop"
 	"github.com/honeycombio/refinery/config"
 	"github.com/honeycombio/refinery/internal/otelutil"
+	"github.com/honeycombio/refinery/logger"
 	"github.com/honeycombio/refinery/pubsub"
 	"github.com/jonboulle/clockwork"
 	"go.opentelemetry.io/otel/trace"
@@ -24,6 +25,7 @@ const ConfigPubsubTopic = "cfg_update"
 // It avoids sending duplicate messages by comparing the hash of the configs.
 type ConfigWatcher struct {
 	Config  config.Config   `inject:""`
+	Logger  logger.Logger   `inject:""`
 	PubSub  pubsub.PubSub   `inject:""`
 	Tracer  trace.Tracer    `inject:"tracer"`
 	Clock   clockwork.Clock `inject:""`
@@ -78,7 +80,10 @@ func (cw *ConfigWatcher) SubscriptionListener(ctx context.Context, msg string) {
 	cw.mut.Unlock()
 	// maybe reload the config (it will only reload if the hashes are different,
 	// and if they were, it will call the ReloadCallback)
-	cw.Config.Reload()
+	err = cw.Config.Reload()
+	if err != nil {
+		cw.Logger.Error().Logf("error reloading config: %s", err)
+	}
 }
 
 // Monitor periodically wakes up and tells the config to reload itself.
@@ -94,7 +99,9 @@ func (cw *ConfigWatcher) monitor() {
 		case <-cw.done:
 			return
 		case <-ticker.C:
-			cw.Config.Reload()
+			if err := cw.Config.Reload(); err != nil {
+				cw.Logger.Error().Logf("error reloading config: %s", err)
+			}
 		}
 	}
 }
@@ -103,6 +110,9 @@ func (cw *ConfigWatcher) Start() error {
 	if cw.Tracer == nil {
 		cw.Tracer = noop.NewTracerProvider().Tracer("test")
 	}
+	if cw.Config.GetOpAMPConfig().Enabled {
+		return nil
+	}
 	if cw.Config.GetGeneralConfig().ConfigReloadInterval != 0 {
 		go cw.monitor()
 	}
@@ -112,7 +122,11 @@ func (cw *ConfigWatcher) Start() error {
 }
 
 func (cw *ConfigWatcher) Stop() error {
-	close(cw.done)
-	cw.subscr.Close()
+	if cw.done != nil {
+		close(cw.done)
+	}
+	if cw.subscr != nil {
+		cw.subscr.Close()
+	}
 	return nil
 }
diff --git a/internal/health/mock.go b/internal/health/mock.go
new file mode 100644
index 0000000..8c3490f
--- /dev/null
+++ b/internal/health/mock.go
@@ -0,0 +1,33 @@
+package health
+
+import "sync"
+
+type MockHealthReporter struct {
+	isAlive bool
+	isReady bool
+	mutex   sync.Mutex
+}
+
+func (m *MockHealthReporter) SetAlive(isAlive bool) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.isAlive = isAlive
+}
+
+func (m *MockHealthReporter) IsAlive() bool {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	return m.isAlive
+}
+
+func (m *MockHealthReporter) SetReady(isReady bool) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.isReady = isReady
+}
+
+func (m *MockHealthReporter) IsReady() bool {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	return m.isReady
+}
diff --git a/metrics.md b/metrics.md
index 13a367e..0ca617f 100644
--- a/metrics.md
+++ b/metrics.md
@@ -10,6 +10,8 @@ This table includes metrics with fully defined names.
 
 | Name | Type | Unit | Description |
 |------|------|------|-------------|
+| is_ready | Gauge | Dimensionless | Whether the system is ready to receive traffic |
+| is_alive | Gauge | Dimensionless | Whether the system is alive and reporting in |
 | collect_cache_entries | Histogram | Dimensionless | The number of traces currently stored in the cache |
 | cuckoo_current_capacity | Gauge | Dimensionless | current capacity of the cuckoo filter |
 | cuckoo_future_load_factor | Gauge | Percent | the fraction of slots occupied in the future cuckoo filter |
@@ -18,8 +20,6 @@ This table includes metrics with fully defined names.
 | cuckoo_addqueue_locktime_uS | Histogram | Microseconds | the time spent holding the add queue lock |
 | cache_recent_dropped_traces | Gauge | Dimensionless | the current size of the most recent dropped trace cache |
 | collect_sent_reasons_cache_entries | Histogram | Dimensionless | Number of entries in the sent reasons cache |
-| is_ready | Gauge | Dimensionless | Whether the system is ready to receive traffic |
-| is_alive | Gauge | Dimensionless | Whether the system is alive and reporting in |
 | redis_pubsub_published | Counter | Dimensionless | Number of messages published to Redis PubSub |
 | redis_pubsub_received | Counter | Dimensionless | Number of messages received from Redis PubSub |
 | local_pubsub_published | Counter | Dimensionless | The total number of messages sent via the local pubsub implementation |
@@ -104,6 +104,8 @@ Metrics in this table don't contain their expected prefixes. This is because the
 | _router_peer | Counter | Dimensionless | the number of spans proxied to a peer |
 | _router_batch | Counter | Dimensionless | the number of batches of events received |
 | _router_otlp | Counter | Dimensionless | the number of batches of otlp requests received |
+| bytes_received_traces | Counter | Bytes | the number of bytes received in trace events |
+| bytes_received_logs | Counter | Bytes | the number of bytes received in log events |
 | queue_length | Gauge | Dimensionless | number of events waiting to be sent to destination |
 | queue_overflow | Counter | Dimensionless | number of events dropped due to queue overflow |
 | send_errors | Counter | Dimensionless | number of errors encountered while sending events to destination |
diff --git a/metrics/multi_metrics.go b/metrics/multi_metrics.go
index 3f94c46..87cef92 100644
--- a/metrics/multi_metrics.go
+++ b/metrics/multi_metrics.go
@@ -12,9 +12,9 @@ var _ Metrics = (*MultiMetrics)(nil)
 // underlying metrics provider (StoreMetrics). It can be configured to send
 // metrics to multiple providers at once.
 //
-// It also stores the values saved with Store(), gauges, and updown counters,
-// which can then be retrieved with Get(). This is for use with StressRelief. It
-// does not track histograms or counters, which are reset after each scrape.
+// It also stores the values saved with Store(), counters, gauges, and updown counters,
+// which can then be retrieved with Get(). This is for use with StressRelief and OpAMP agent. It
+// does not track histograms, which are reset after each scrape.
 type MultiMetrics struct {
 	Config        config.Config `inject:""`
 	LegacyMetrics Metrics       `inject:"legacyMetrics"`
@@ -76,6 +76,9 @@ func (m *MultiMetrics) Increment(name string) { // for counters
 	for _, ch := range m.children {
 		ch.Increment(name)
 	}
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.values[name]++
 }
 
 func (m *MultiMetrics) Gauge(name string, val interface{}) { // for gauges
@@ -91,6 +94,9 @@ func (m *MultiMetrics) Count(name string, n interface{}) { // for counters
 	for _, ch := range m.children {
 		ch.Count(name, n)
 	}
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.values[name] += ConvertNumeric(n)
 }
 
 func (m *MultiMetrics) Histogram(name string, obs interface{}) { // for histogram
diff --git a/refinery_config.md b/refinery_config.md
index bd17940..1b69547 100644
--- a/refinery_config.md
+++ b/refinery_config.md
@@ -126,6 +126,48 @@ This setting is the destination to which Refinery sends all events that it decid
 - Environment variable: `REFINERY_HONEYCOMB_API`
 - Command line switch: `--honeycomb-api`
 
+## OpAMP Configuration
+
+`OpAMP` contains OpAMP configuration options.
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+### `Enabled`
+
+`Enabled` controls whether to enable OpAMP support.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Not eligible for live reload.
+- Type: `bool`
+
+### `RecordUsage`
+
+`RecordUsage` controls whether to record usage metrics.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Eligible for live reload.
+- Type: `defaulttrue`
+- Default: `true`
+
+### `Endpoint`
+
+`Endpoint` is the URL of the OpAMP server for this client.
+
+OpAMP support is an experimental feature, subject to breaking changes at any time.
+Not intended or supported for customer use.
+**FOR INTERNAL USE ONLY**
+
+- Not eligible for live reload.
+- Type: `string`
+- Default: `wss://127.0.0.1:4320/v1/opamp`
+
 ## Access Key Configuration
 
 `AccessKeys` contains access keys -- API keys that the proxy will treat specially, and other flags that control how the proxy handles API keys.
diff --git a/route/route.go b/route/route.go
index 0ca1895..1e40403 100644
--- a/route/route.go
+++ b/route/route.go
@@ -135,6 +135,8 @@ var routerMetrics = []metrics.Metadata{
 	{Name: "_router_peer", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of spans proxied to a peer"},
 	{Name: "_router_batch", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of batches of events received"},
 	{Name: "_router_otlp", Type: metrics.Counter, Unit: metrics.Dimensionless, Description: "the number of batches of otlp requests received"},
+	{Name: "bytes_received_traces", Type: metrics.Counter, Unit: metrics.Bytes, Description: "the number of bytes received in trace events"},
+	{Name: "bytes_received_logs", Type: metrics.Counter, Unit: metrics.Bytes, Description: "the number of bytes received in log events"},
 }
 
 // LnS spins up the Listen and Serve portion of the router. A router is
@@ -162,7 +164,9 @@ func (r *Router) LnS(incomingOrPeer string) {
 	}
 
 	for _, metric := range routerMetrics {
-		metric.Name = r.incomingOrPeer + metric.Name
+		if strings.HasPrefix(metric.Name, "_") {
+			metric.Name = r.incomingOrPeer + metric.Name
+		}
 		r.Metrics.Register(metric)
 	}
 
@@ -607,6 +611,15 @@ func (r *Router) processEvent(ev *types.Event, reqID interface{}) error {
 		IsRoot:  isRootSpan(ev, r.Config),
 	}
 
+	// only record bytes received for incoming traffic when opamp is enabled and record usage is set to true
+	if r.incomingOrPeer == "incoming" && r.Config.GetOpAMPConfig().Enabled && r.Config.GetOpAMPConfig().RecordUsage.Get() {
+		if span.Data["meta.signal_type"] == "log" {
+			r.Metrics.Count("bytes_received_logs", span.GetDataSize())
+		} else {
+			r.Metrics.Count("bytes_received_traces", span.GetDataSize())
+		}
+	}
+
 	// we know we're a span, but we need to check if we're in Stress Relief mode;
 	// if we are, then we want to make an immediate, deterministic trace decision
 	// and either drop or send the trace without even trying to cache or forward it.
diff --git a/tools/convert/configDataNames.txt b/tools/convert/configDataNames.txt
index 747508c..9f15b75 100644
--- a/tools/convert/configDataNames.txt
+++ b/tools/convert/configDataNames.txt
@@ -21,6 +21,14 @@ Network:
   - HoneycombAPI
 
 
+OpAMP:
+  - Enabled
+
+  - RecordUsage
+
+  - Endpoint
+
+
 AccessKeys:
   - ReceiveKeys (originally APIKeys)
 
diff --git a/tools/convert/metrics.go b/tools/convert/metrics.go
index a49a0ae..74f4ff3 100644
--- a/tools/convert/metrics.go
+++ b/tools/convert/metrics.go
@@ -111,7 +111,7 @@ func GenerateMetricsMetadata() error {
 
 		}
 		if !found {
-			return fmt.Errorf("Missing metrics.Metadata declaration in the package %s", pkg.Name)
+			fmt.Printf("Missing metrics.Metadata declaration in the package %s\n", pkg.Name)
 		}
 	}
 
diff --git a/tools/convert/metricsMeta.yaml b/tools/convert/metricsMeta.yaml
index 5f1cb1b..a61e406 100644
--- a/tools/convert/metricsMeta.yaml
+++ b/tools/convert/metricsMeta.yaml
@@ -1,4 +1,12 @@
 complete:
+    - name: is_ready
+      type: Gauge
+      unit: Dimensionless
+      description: Whether the system is ready to receive traffic
+    - name: is_alive
+      type: Gauge
+      unit: Dimensionless
+      description: Whether the system is alive and reporting in
     - name: collect_cache_entries
       type: Histogram
       unit: Dimensionless
@@ -31,14 +39,6 @@ complete:
       type: Histogram
       unit: Dimensionless
       description: Number of entries in the sent reasons cache
-    - name: is_ready
-      type: Gauge
-      unit: Dimensionless
-      description: Whether the system is ready to receive traffic
-    - name: is_alive
-      type: Gauge
-      unit: Dimensionless
-      description: Whether the system is alive and reporting in
     - name: redis_pubsub_published
       type: Counter
       unit: Dimensionless
@@ -344,6 +344,14 @@ hasprefix:
       type: Counter
       unit: Dimensionless
       description: the number of batches of otlp requests received
+    - name: bytes_received_traces
+      type: Counter
+      unit: Bytes
+      description: the number of bytes received in trace events
+    - name: bytes_received_logs
+      type: Counter
+      unit: Bytes
+      description: the number of bytes received in log events
     - name: queue_length
       type: Gauge
       unit: Dimensionless
diff --git a/tools/convert/minimal_config.yaml b/tools/convert/minimal_config.yaml
index ab33a26..a4870c4 100644
--- a/tools/convert/minimal_config.yaml
+++ b/tools/convert/minimal_config.yaml
@@ -10,6 +10,10 @@ Network:
   PeerListenAddr: "0.0.0.0:8081"
   HTTPIdleTimeout: 0s
   HoneycombAPI: "https://api.honeycomb.io"
+OpAMP:
+  Enabled: false
+  RecordUsage: true
+  Endpoint: "wss://127.0.0.1:4320/v1/opamp"
 AccessKeys:
   ReceiveKeys: 
     - "your-key-goes-here"
diff --git a/tools/convert/templates/configV2.tmpl b/tools/convert/templates/configV2.tmpl
index 9174466..d5af4ab 100644
--- a/tools/convert/templates/configV2.tmpl
+++ b/tools/convert/templates/configV2.tmpl
@@ -128,6 +128,44 @@ Network:
     ## Eligible for live reload.
     {{ nonDefaultOnly .Data "HoneycombAPI" "HoneycombAPI" "https://api.honeycomb.io" }}
 
+#########################
+## OpAMP Configuration ##
+#########################
+OpAMP:
+  ## OpAMP contains OpAMP configuration options. OpAMP support is an
+  ## experimental feature, subject to breaking changes at any time. Not
+  ## intended or supported for customer use. **FOR INTERNAL USE ONLY**
+  ##
+  ####
+    ## Enabled controls whether to enable OpAMP support.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## Not eligible for live reload.
+    {{ nonDefaultOnly .Data "Enabled" "Enabled" false }}
+
+    ## RecordUsage controls whether to record usage metrics.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## default: true
+    ## Eligible for live reload.
+    {{ nonDefaultOnly .Data "RecordUsage" "RecordUsage" true }}
+
+    ## Endpoint is the URL of the OpAMP server for this client.
+    ##
+    ## OpAMP support is an experimental feature, subject to breaking changes
+    ## at any time. Not intended or supported for customer use. **FOR
+    ## INTERNAL USE ONLY**
+    ##
+    ## default: wss://127.0.0.1:4320/v1/opamp
+    ## Not eligible for live reload.
+    Endpoint: wss://127.0.0.1:4320/v1/opamp
+
 ##############################
 ## Access Key Configuration ##
 ##############################


Test patch:
diff --git a/agent/agent_test.go b/agent/agent_test.go
new file mode 100644
index 0000000..657884f
--- /dev/null
+++ b/agent/agent_test.go
@@ -0,0 +1,229 @@
+package agent
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/honeycombio/refinery/config"
+	"github.com/honeycombio/refinery/internal/health"
+	"github.com/honeycombio/refinery/logger"
+	"github.com/honeycombio/refinery/metrics"
+	"github.com/jonboulle/clockwork"
+	"github.com/open-telemetry/opamp-go/client"
+	"github.com/open-telemetry/opamp-go/client/types"
+	"github.com/open-telemetry/opamp-go/protobufs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func TestAgentOnMessage_RemoteConfig(t *testing.T) {
+	var reloadCalled int
+	cfg := &config.MockConfig{
+		GetLoggerLevelVal:  config.InfoLevel,
+		GetSamplerTypeVal:  "FakeSamplerType",
+		GetSamplerTypeName: "FakeSamplerName",
+	}
+	cfg.Callbacks = []config.ConfigReloadCallback{
+		func(configHash, ruleCfgHash string) {
+			reloadCalled++
+		},
+	}
+	agent := NewAgent(Logger{Logger: &logger.NullLogger{}}, "1.0.0", cfg, &metrics.NullMetrics{}, &health.Health{})
+	defer agent.Stop(context.Background())
+
+	testcases := []struct {
+		name                string
+		configMap           map[string]*protobufs.AgentConfigFile
+		configHash          []byte
+		expectedReloadCount int
+		status              protobufs.RemoteConfigStatuses
+	}{
+		{
+			name:                "empty config map",
+			configMap:           map[string]*protobufs.AgentConfigFile{},
+			configHash:          []byte{0},
+			expectedReloadCount: 0,
+		},
+		{
+			name: "new refinery config from remote config",
+			configMap: map[string]*protobufs.AgentConfigFile{
+				"refinery_config": {
+					Body:        []byte(`{"Logger":{"Level":"debug"}}`),
+					ContentType: "text/yaml",
+				},
+			},
+			configHash:          []byte{1},
+			expectedReloadCount: 1,
+			status:              protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED,
+		},
+		{
+			name: "new refinery rules from remote config",
+			configMap: map[string]*protobufs.AgentConfigFile{
+				"refinery_config": {
+					Body:        []byte(`{"Logger":{"Level":"debug"}}`),
+					ContentType: "text/yaml",
+				},
+				"refinery_rules": {
+					Body: []byte(`{"rules":[{"name":"test","type":"fake"]}`),
+				},
+			},
+			configHash:          []byte{2},
+			expectedReloadCount: 2,
+			status:              protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED,
+		},
+		{
+			name: "same remote config should not cause reload",
+			configMap: map[string]*protobufs.AgentConfigFile{
+				"refinery_config": {
+					Body:        []byte(`{"Logger":{"Level":"debug"}}`),
+					ContentType: "text/yaml",
+				},
+				"refinery_rules": {
+					Body: []byte(`{"rules":[{"name":"test","type":"fake"]}`),
+				},
+			},
+			configHash:          []byte{2},
+			expectedReloadCount: 2,
+			status:              protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED,
+		},
+	}
+
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			agent.onMessage(context.Background(), &types.MessageData{
+				RemoteConfig: &protobufs.AgentRemoteConfig{
+					Config: &protobufs.AgentConfigMap{
+						ConfigMap: tc.configMap,
+					},
+					ConfigHash: tc.configHash,
+				},
+			})
+			require.Eventually(t, func() bool {
+				return tc.expectedReloadCount == reloadCalled
+			}, 1*time.Second, 100*time.Millisecond, fmt.Sprintf("unexpected reload count %d", reloadCalled))
+
+			require.Equal(t, tc.status, agent.remoteConfigStatus.GetStatus(), fmt.Sprintf("unexpected status %s", agent.remoteConfigStatus.GetStatus()))
+			if tc.status == protobufs.RemoteConfigStatuses_RemoteConfigStatuses_APPLIED {
+				assert.Equal(t, tc.configHash, agent.remoteConfigStatus.GetLastRemoteConfigHash())
+				assert.Equal(t, tc.configHash, agent.lastRemoteConfigReceived.GetConfigHash())
+				assert.Equal(t, tc.configMap, agent.lastRemoteConfigReceived.GetConfig().GetConfigMap())
+			}
+		})
+	}
+
+}
+
+func TestHealthCheck(t *testing.T) {
+	healthReporter := &health.MockHealthReporter{}
+	agent := NewAgent(Logger{Logger: &logger.NullLogger{}}, "1.0.0", &config.MockConfig{}, &metrics.NullMetrics{}, healthReporter)
+	defer agent.Stop(context.Background())
+
+	// health check should start with false
+	require.False(t, agent.calculateHealth().Healthy)
+
+	// health check should be false if Refinery is not ready
+	healthReporter.SetAlive(true)
+	require.False(t, agent.calculateHealth().Healthy)
+
+	// health check should be true if both alive and ready are true
+	healthReporter.SetReady(true)
+	require.True(t, agent.calculateHealth().Healthy)
+}
+
+func TestAgentUsageReport(t *testing.T) {
+	mockClient := &MockOpAMPClient{}
+	ctx, cancel := context.WithCancel(context.Background())
+	clock := clockwork.NewFakeClock()
+	agent := &Agent{
+		ctx:             ctx,
+		cancel:          cancel,
+		logger:          Logger{&logger.NullLogger{}},
+		agentType:       serviceName,
+		agentVersion:    "1.0.0",
+		opampClient:     mockClient,
+		effectiveConfig: &config.MockConfig{},
+		metrics:         &metrics.NullMetrics{},
+		health:          &health.MockHealthReporter{},
+		usageTracker:    newUsageTracker(),
+		clock:           clock,
+	}
+	agent.createAgentIdentity()
+	agent.hostname = "my-hostname"
+	defer cancel()
+
+	isSent := make(chan struct{})
+	close(isSent)
+	mockClient.On("SendCustomMessage", mock.Anything).Return(isSent, nil)
+
+	agent.usageTracker.Add(signal_traces, 1)
+	agent.usageTracker.Add(signal_logs, 2)
+	agent.usageTracker.Add(signal_traces, 3)
+	agent.usageTracker.Add(signal_logs, 4)
+
+	err := agent.sendUsageReport()
+	require.NoError(t, err)
+
+	// Format the timestamp to match the expected format in the payload
+	timeUnixNano := clock.Now()
+
+	metrics := pmetric.NewMetrics()
+	rm := metrics.ResourceMetrics().AppendEmpty()
+	resourceAttrs := rm.Resource().Attributes()
+	resourceAttrs.PutStr("service.name", "refinery")
+	resourceAttrs.PutStr("service.version", "1.0.0")
+	resourceAttrs.PutStr("host.name", "my-hostname")
+	sm := rm.ScopeMetrics().AppendEmpty()
+	ms := sm.Metrics().AppendEmpty()
+	ms.SetName("bytes_received")
+	sum := ms.SetEmptySum()
+	sum.SetAggregationTemporality(pmetric.AggregationTemporalityDelta)
+	d1 := sum.DataPoints().AppendEmpty()
+	d1.Attributes().PutStr("signal", "traces")
+	d1.SetIntValue(3)
+	d1.SetTimestamp(pcommon.NewTimestampFromTime(timeUnixNano))
+	d2 := sum.DataPoints().AppendEmpty()
+	d2.Attributes().PutStr("signal", "logs")
+	d2.SetIntValue(4)
+	d2.SetTimestamp(pcommon.NewTimestampFromTime(timeUnixNano))
+
+	matcher := mock.MatchedBy(func(payload *protobufs.CustomMessage) bool {
+		if payload.Capability != sendAgentTelemetryCapability {
+			return false
+		}
+
+		unmarshaler := &pmetric.JSONUnmarshaler{}
+		m, err := unmarshaler.UnmarshalMetrics(payload.Data)
+		require.NoError(t, err)
+
+		return pmetrictest.CompareMetrics(metrics, m, pmetrictest.IgnoreMetricDataPointsOrder()) == nil
+	})
+	// Assert that the mock client was called with the expected custom message
+	mockClient.AssertCalled(t, "SendCustomMessage", matcher)
+
+}
+
+// MockOpAMPClient is a mock implementation of the OpAMPClient interface
+type MockOpAMPClient struct {
+	mock.Mock
+	client.OpAMPClient
+}
+
+func (m *MockOpAMPClient) SendCustomMessage(msg *protobufs.CustomMessage) (chan struct{}, error) {
+	args := m.Called(msg)
+	return args.Get(0).(chan struct{}), args.Error(1)
+}
+
+func (m *MockOpAMPClient) SetHealth(health *protobufs.ComponentHealth) error {
+	return nil
+}
+
+func (m *MockOpAMPClient) Stop(ctx context.Context) error {
+	return nil
+}
diff --git a/agent/usage_report_test.go b/agent/usage_report_test.go
new file mode 100644
index 0000000..e80fead
--- /dev/null
+++ b/agent/usage_report_test.go
@@ -0,0 +1,175 @@
+package agent
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatatest/pmetrictest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+func TestUsageTracker_Add(t *testing.T) {
+	tests := []struct {
+		name           string
+		signalsAndVals []struct {
+			signal usageSignal
+			val    float64
+		}
+		expectedLength int
+		expectedData   map[usageSignal]float64
+	}{
+		{
+			name: "Add usage data",
+			signalsAndVals: []struct {
+				signal usageSignal
+				val    float64
+			}{
+				{signal: signal_traces, val: 1},
+				{signal: signal_traces, val: 2},
+				{signal: signal_logs, val: 3},
+				{signal: signal_logs, val: 4},
+			},
+			expectedLength: 2,
+			expectedData: map[usageSignal]float64{
+				signal_traces: 2,
+				signal_logs:   4,
+			},
+		},
+		{
+			name: "Add 0 usage data",
+			signalsAndVals: []struct {
+				signal usageSignal
+				val    float64
+			}{
+				{signal: signal_traces, val: 0},
+				{signal: signal_logs, val: 0},
+			},
+			expectedData:   make(map[usageSignal]float64),
+			expectedLength: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tracker := newUsageTracker()
+
+			for _, sv := range tt.signalsAndVals {
+				tracker.Add(sv.signal, sv.val)
+			}
+
+			assert.Equal(t, tt.expectedData, tracker.lastUsageData)
+			assert.Len(t, tracker.currentDataPoints, tt.expectedLength)
+			assert.Equal(t, tt.expectedData, tracker.currentDataPoints)
+		})
+	}
+}
+
+func TestUsageTracker_NewReport(t *testing.T) {
+	tests := []struct {
+		name           string
+		setup          func(tracker *usageTracker)
+		expectedError  error
+		expectedReport func(now time.Time) pmetric.Metrics
+	}{
+		{
+			name: "Generate usage report",
+			setup: func(tracker *usageTracker) {
+				tracker.Add(signal_traces, 1)
+				tracker.Add(signal_traces, 2)
+				tracker.Add(signal_logs, 2)
+			},
+			expectedError: nil,
+			expectedReport: func(now time.Time) pmetric.Metrics {
+				metrics := map[usageSignal]float64{
+					signal_traces: 2,
+					signal_logs:   2,
+				}
+				return newOTLPResourceMetricsPayload(metrics, now)
+			},
+		},
+		{
+			name: "Generate usage report with no data",
+			setup: func(tracker *usageTracker) {
+				// No data added
+			},
+			expectedError: errNoData,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tracker := newUsageTracker()
+			now := time.Now()
+
+			tt.setup(tracker)
+
+			report, err := tracker.NewReport("my-service", "1.0.0", "my-hostname", now)
+			if tt.expectedError != nil {
+				require.Error(t, err)
+				require.Nil(t, report)
+				assert.Equal(t, tt.expectedError, err)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, report)
+				if tt.expectedReport != nil {
+					unmarshaler := &pmetric.JSONUnmarshaler{}
+					m, err := unmarshaler.UnmarshalMetrics(report)
+					require.NoError(t, err)
+					require.NoError(t, pmetrictest.CompareMetrics(tt.expectedReport(now), m, pmetrictest.IgnoreMetricDataPointsOrder()))
+					assert.Empty(t, tracker.currentDataPoints)
+					assert.NotEmpty(t, tracker.lastDataPoints)
+				}
+			}
+
+			tracker.completeSend()
+			assert.Empty(t, tracker.lastDataPoints)
+		})
+	}
+}
+
+func newOTLPResourceMetricsPayload(metrics map[usageSignal]float64, now time.Time) pmetric.Metrics {
+	var dataPoints []map[string]interface{}
+	for signal, value := range metrics {
+		dataPoints = append(dataPoints, map[string]interface{}{
+			"attributes":   []map[string]interface{}{{"key": "signal", "value": map[string]string{"stringValue": string(signal)}}},
+			"timeUnixNano": fmt.Sprintf("%d", now.UnixNano()),
+			"asInt":        fmt.Sprintf("%d", int(value)),
+		})
+	}
+
+	payload := map[string]interface{}{
+		"resourceMetrics": []map[string]interface{}{
+			{
+				"resource": map[string]interface{}{
+					"attributes": []map[string]interface{}{
+						{"key": "service.name", "value": map[string]string{"stringValue": "my-service"}},
+						{"key": "service.version", "value": map[string]string{"stringValue": "1.0.0"}},
+						{"key": "host.name", "value": map[string]string{"stringValue": "my-hostname"}},
+					},
+				},
+				"scopeMetrics": []map[string]interface{}{
+					{
+						"metrics": []map[string]interface{}{
+							{
+								"name": "bytes_received",
+								"sum": map[string]interface{}{
+									"aggregationTemporality": 1,
+									"dataPoints":             dataPoints,
+								},
+							},
+						},
+						"scope": map[string]interface{}{},
+					},
+				},
+			},
+		},
+	}
+	bytes, _ := json.Marshal(payload)
+	unmarshaler := &pmetric.JSONUnmarshaler{}
+	m, _ := unmarshaler.UnmarshalMetrics(bytes)
+	return m
+}
diff --git a/config/config_test.go b/config/config_test.go
index 8caee4b..679de6a 100644
--- a/config/config_test.go
+++ b/config/config_test.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/honeycombio/refinery/config"
 	"github.com/honeycombio/refinery/internal/configwatcher"
+	"github.com/honeycombio/refinery/logger"
 	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -21,7 +22,7 @@ func getConfig(args []string) (config.Config, error) {
 	if err != nil {
 		return nil, err
 	}
-	return config.NewConfig(opts, func(err error) {})
+	return config.NewConfig(opts)
 }
 
 // creates two temporary yaml files from the strings passed in and returns their filenames
@@ -213,6 +214,7 @@ func TestReload(t *testing.T) {
 	watcher := &configwatcher.ConfigWatcher{
 		Config: c,
 		PubSub: pubsub,
+		Logger: &logger.NullLogger{},
 	}
 	watcher.Start()
 	defer watcher.Stop()
@@ -335,6 +337,10 @@ func TestReadDefaults(t *testing.T) {
 		t.Error("received", d, "expected", time.Hour)
 	}
 
+	if d := c.GetOpAMPConfig(); d.Enabled {
+		t.Error("received", d, "expected", time.Hour)
+	}
+
 	d, name := c.GetSamplerConfigForDestName("dataset-doesnt-exist")
 	assert.IsType(t, &config.DeterministicSamplerConfig{}, d)
 	assert.Equal(t, "DeterministicSampler", name)
diff --git a/config/config_test_reload_error_test.go b/config/config_test_reload_error_test.go
index bfd7023..d9afc92 100644
--- a/config/config_test_reload_error_test.go
+++ b/config/config_test_reload_error_test.go
@@ -4,14 +4,15 @@ package config_test
 
 import (
 	"os"
-	"sync"
 	"testing"
 	"time"
 
 	"github.com/honeycombio/refinery/config"
 	"github.com/honeycombio/refinery/internal/configwatcher"
+	"github.com/honeycombio/refinery/logger"
 	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestErrorReloading(t *testing.T) {
@@ -32,8 +33,7 @@ func TestErrorReloading(t *testing.T) {
 	opts, err := config.NewCmdEnvOptions([]string{"--config", cfg, "--rules_config", rules})
 	assert.NoError(t, err)
 
-	ch := make(chan interface{}, 1)
-	c, err := config.NewConfig(opts, func(err error) { ch <- 1 })
+	c, err := config.NewConfig(opts)
 	assert.NoError(t, err)
 
 	pubsub := &pubsub.LocalPubSub{
@@ -41,9 +41,11 @@ func TestErrorReloading(t *testing.T) {
 	}
 	pubsub.Start()
 	defer pubsub.Stop()
+	mockLogger := &logger.MockLogger{}
 	watcher := &configwatcher.ConfigWatcher{
 		Config: c,
 		PubSub: pubsub,
+		Logger: mockLogger,
 	}
 	watcher.Start()
 	defer watcher.Stop()
@@ -56,29 +58,18 @@ func TestErrorReloading(t *testing.T) {
 		t.Error("name received", d, "expected", "DeterministicSampler")
 	}
 
-	wg := &sync.WaitGroup{}
-
-	wg.Add(1)
-
-	go func() {
-		defer wg.Done()
-		select {
-		case <-ch:
-		case <-time.After(5 * time.Second):
-			t.Error("No error callback")
-		}
-	}()
-
 	// This is valid YAML, but invalid config
 	rm2 := makeYAML(
 		"RulesVersion", 2,
 		"Samplers.__default__.InvalidSampler.SampleRate", 50,
 	)
 	err = os.WriteFile(rules, []byte(rm2), 0644)
-
 	assert.NoError(t, err)
 
-	wg.Wait()
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		require.Len(c, mockLogger.Events, 1)
+		assert.Contains(c, mockLogger.Events[0].Fields["error"], "error reloading config")
+	}, 5*time.Second, 100*time.Millisecond)
 
 	// config should error and not update sampler to invalid type
 	d, _ = c.GetSamplerConfigForDestName("dataset5")
diff --git a/metrics/multi_metrics_test.go b/metrics/multi_metrics_test.go
index ef53044..b66f87e 100644
--- a/metrics/multi_metrics_test.go
+++ b/metrics/multi_metrics_test.go
@@ -91,10 +91,10 @@ func TestMultiMetrics_Register(t *testing.T) {
 	mm.Down("updown")
 	mm.Gauge("gauge", 42)
 
-	// counter should be 0 because it's not tracked by StoreMetrics
+	// counter should be 1 because it's tracked by StoreMetrics
 	val, ok := mm.Get("counter")
 	assert.True(t, ok)
-	assert.Equal(t, 0, int(val))
+	assert.Equal(t, 1, int(val))
 
 	// updown should be 2 because it's tracked by StoreMetrics
 	val, ok = mm.Get("updown")
diff --git a/route/route_test.go b/route/route_test.go
index 8ab45cf..479bfd5 100644
--- a/route/route_test.go
+++ b/route/route_test.go
@@ -800,3 +800,126 @@ func TestAddIncomingUserAgent(t *testing.T) {
 		require.Equal(t, "test-agent", event.Data["meta.refinery.incoming_user_agent"])
 	})
 }
+
+func TestProcessEventMetrics(t *testing.T) {
+
+	tests := []struct {
+		name           string
+		incomingOrPeer string
+		opampEnabled   bool
+		recordUsage    config.DefaultTrue
+		signalType     string
+		expectedCount  int
+		metricName     string
+	}{
+		{
+			name:           "log event with opamp enabled and record usage",
+			incomingOrPeer: "incoming",
+			opampEnabled:   true,
+			recordUsage:    config.DefaultTrue(true),
+			signalType:     "log",
+			expectedCount:  91,
+			metricName:     "bytes_received_logs",
+		},
+		{
+			name:           "trace event with opamp enabled and record usage",
+			incomingOrPeer: "incoming",
+			opampEnabled:   true,
+			recordUsage:    config.DefaultTrue(true),
+			signalType:     "trace",
+			expectedCount:  93,
+			metricName:     "bytes_received_traces",
+		},
+		{
+			name:           "log event with opamp disabled",
+			incomingOrPeer: "incoming",
+			opampEnabled:   false,
+			recordUsage:    config.DefaultTrue(true),
+			signalType:     "log",
+			expectedCount:  0,
+		},
+		{
+			name:           "log event with record usage disabled",
+			incomingOrPeer: "incoming",
+			opampEnabled:   true,
+			recordUsage:    config.DefaultTrue(false),
+			signalType:     "log",
+			expectedCount:  0,
+		},
+		{
+			name:           "log event from peer",
+			incomingOrPeer: "peer",
+			opampEnabled:   true,
+			recordUsage:    config.DefaultTrue(true),
+			signalType:     "log",
+			expectedCount:  0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mockMetrics := &metrics.MockMetrics{}
+			mockMetrics.Start()
+
+			mockConfig := &config.MockConfig{
+				GetOpAmpConfigVal: config.OpAMPConfig{
+					Enabled:     tt.opampEnabled,
+					RecordUsage: &tt.recordUsage,
+				},
+				TraceIdFieldNames: []string{"trace.trace_id"},
+			}
+
+			// Setup mock transmissions
+			mockUpstream := &transmit.MockTransmission{}
+			mockUpstream.Start()
+			mockPeer := &transmit.MockTransmission{}
+			mockPeer.Start()
+
+			mockSharder := &sharder.MockSharder{
+				Self: &sharder.TestShard{
+					Addr: "http://localhost:12345",
+				},
+			}
+
+			router := &Router{
+				Config:               mockConfig,
+				Logger:               &logger.NullLogger{},
+				Metrics:              mockMetrics,
+				UpstreamTransmission: mockUpstream,
+				PeerTransmission:     mockPeer,
+				Collector:            collect.NewMockCollector(),
+				Sharder:              mockSharder,
+				incomingOrPeer:       tt.incomingOrPeer,
+				iopLogger:            iopLogger{Logger: &logger.NullLogger{}, incomingOrPeer: tt.incomingOrPeer},
+			}
+
+			// Create test event with traceID and signal type
+			event := &types.Event{
+				Context:   context.Background(),
+				APIHost:   "test.honeycomb.io",
+				Dataset:   "test-dataset",
+				Timestamp: time.Now(),
+				Data: map[string]interface{}{
+					"trace.trace_id":    "trace-123",
+					"meta.signal_type":  tt.signalType,
+					"test_attribute":    "test_value",
+					"another_attribute": 123,
+				},
+			}
+			span := &types.Span{
+				Event:   *event,
+				TraceID: "trace-123",
+				IsRoot:  true,
+			}
+			size := span.GetDataSize()
+			if tt.expectedCount > 0 {
+				assert.Equal(t, tt.expectedCount, size)
+			}
+
+			// Call processEvent
+			err := router.processEvent(event, "request-123")
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expectedCount, mockMetrics.CounterIncrements[tt.metricName])
+		})
+	}
+}
diff --git a/sample/sample_test.go b/sample/sample_test.go
index 4f58583..4747fb1 100644
--- a/sample/sample_test.go
+++ b/sample/sample_test.go
@@ -20,7 +20,7 @@ func getConfig(args []string) (config.Config, error) {
 	if err != nil {
 		return nil, err
 	}
-	return config.NewConfig(opts, func(err error) {})
+	return config.NewConfig(opts)
 }
 
 // creates two temporary yaml files from the strings passed in and returns their filenames