DS-1378 Telemetry Sampling config option

config_docs_test.go

@@ -19,8 +19,7 @@ var (
 func TestConfigDocs(t *testing.T) {
 	config, err := docs.GenerateConfig()
 	assert.NoError(t, err, "invalid config docs")
-	assert.Equal(t, configMD, config, "docs/CONFIG.md is out of date. Run '"+
-		"' to regenerate.")
+	assert.Equal(t, configMD, config, "docs/CONFIG.md is out of date. Run 'make config-docs' to regenerate.")
 
 	secrets, err := docs.GenerateSecrets()
 	assert.NoError(t, err, "invalid secrets docs")

core/config/docs/core.toml

@@ -403,6 +403,8 @@ DefaultTransactionQueueDepth = 1 # Default
 SimulateTransactions = false # Default
 # TraceLogging enables trace level logging.
 TraceLogging = false # Default
+# SampleTelemetry enables telemetry sampling.
+SampleTelemetry = false # Default
 # KeyValueStoreRootDir is the root directory for the key-value store used by OCR3.1.
 # This directory must be writable by the Chainlink node process and should support long-term persistence.
 KeyValueStoreRootDir = '~/.chainlink-data' # Default

core/config/ocr2_config.go

@@ -17,6 +17,7 @@ type OCR2 interface {
 	KeyBundleID() (string, error)
 	// OCR2 config, cannot override in jobs
 	TraceLogging() bool
+	SampleTelemetry() bool
 	CaptureEATelemetry() bool
 	DefaultTransactionQueueDepth() uint32
 	SimulateTransactions() bool

core/config/toml/types.go

@@ -1349,6 +1349,7 @@ type OCR2 struct {
 	DefaultTransactionQueueDepth       *uint32
 	SimulateTransactions               *bool
 	TraceLogging                       *bool
+	SampleTelemetry                    *bool
 	KeyValueStoreRootDir               *string
 }
 
@@ -1395,6 +1396,9 @@ func (o *OCR2) setFrom(f *OCR2) {
 	if v := f.TraceLogging; v != nil {
 		o.TraceLogging = v
 	}
+	if v := f.SampleTelemetry; v != nil {
+		o.SampleTelemetry = v
+	}
 	if v := f.KeyValueStoreRootDir; v != nil {
 		o.KeyValueStoreRootDir = v
 	}

core/services/chainlink/config_ocr2.go

@@ -53,6 +53,10 @@ func (o *ocr2Config) TraceLogging() bool {
 	return *o.c.TraceLogging
 }
 
+func (o *ocr2Config) SampleTelemetry() bool {
+	return *o.c.SampleTelemetry
+}
+
 func (o *ocr2Config) CaptureEATelemetry() bool {
 	return *o.c.CaptureEATelemetry
 }

core/services/chainlink/config_ocr2_test.go

@@ -36,6 +36,7 @@ func TestOCR2Config(t *testing.T) {
 	require.Equal(t, expectedContractSubscribeInterval, ocr2Cfg.ContractSubscribeInterval())
 	require.False(t, ocr2Cfg.SimulateTransactions())
 	require.False(t, ocr2Cfg.TraceLogging())
+	require.False(t, ocr2Cfg.SampleTelemetry())
 	require.Equal(t, uint32(1), ocr2Cfg.DefaultTransactionQueueDepth())
 	require.False(t, ocr2Cfg.CaptureEATelemetry())
 	require.True(t, ocr2Cfg.CaptureAutomationCustomTelemetry())

core/services/chainlink/config_test.go

@@ -438,6 +438,7 @@ func TestConfig_Marshal(t *testing.T) {
 		DefaultTransactionQueueDepth:       ptr[uint32](1),
 		SimulateTransactions:               ptr(false),
 		TraceLogging:                       ptr(false),
+		SampleTelemetry:                    ptr(false),
 		KeyValueStoreRootDir:               ptr("~/.chainlink-data"),
 	}
 	full.OCR = toml.OCR{
@@ -1129,6 +1130,7 @@ AllowNoBootstrappers = true
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 `},
 		{"JobDistributor", Config{Core: toml.Core{JobDistributor: full.JobDistributor}}, `[JobDistributor]

core/services/chainlink/testdata/config-empty-effective.toml

@@ -161,6 +161,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

core/services/chainlink/testdata/config-full.toml

@@ -167,6 +167,7 @@ AllowNoBootstrappers = true
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -161,6 +161,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

core/services/feeds/config.go

@@ -4,6 +4,7 @@ import (
 	"time"
 
 	commonconfig "github.com/smartcontractkit/chainlink-common/pkg/config"
+
 	coreconfig "github.com/smartcontractkit/chainlink/v2/core/config"
 )
 
@@ -42,4 +43,5 @@ type OCR2Config interface {
 	DefaultTransactionQueueDepth() uint32
 	SimulateTransactions() bool
 	TraceLogging() bool
+	SampleTelemetry() bool
 }

core/services/llo/delegate.go

@@ -72,6 +72,7 @@ type DelegateConfig struct {
 
 	// OCR3
 	TraceLogging                 bool
+	SampleTelemetry              bool
 	BinaryNetworkEndpointFactory ocr2types.BinaryNetworkEndpointFactory
 	V2Bootstrappers              []ocrcommontypes.BootstrapperLocator
 	// One Oracle will be started for each ContractConfigTracker
@@ -118,6 +119,7 @@ func NewDelegate(cfg DelegateConfig) (job.ServiceCtx, error) {
 		CaptureObservationTelemetry: cfg.CaptureObservationTelemetry,
 		CaptureOutcomeTelemetry:     cfg.CaptureOutcomeTelemetry,
 		CaptureReportTelemetry:      cfg.CaptureReportTelemetry,
+		SampleTelemetry:             cfg.SampleTelemetry,
 	})
 
 	ds := observation.NewDataSource(

core/services/llo/observation/observation_context.go

@@ -270,6 +270,9 @@ func toUint64(v any) (uint64, error) {
 	}
 }
 
+// TODO this is to return an array of values. maybe a map of streamid->streamval
+// map[stream][llo.Streamvalue]
+
 // extractFinalResultAsStreamValue extracts a final StreamValue from a TaskRunResults
 func extractFinalResultAsStreamValue(trrs pipeline.TaskRunResults) (llo.StreamValue, error) {
 	// pipeline.TaskRunResults comes ordered asc by index, this is guaranteed

core/services/llo/telem/sampling.go

@@ -34,20 +34,27 @@ type sampler struct {
 	samples   map[string]map[int32]any
 	samplesMu sync.Mutex
 
+	enabled     bool
 	prunePeriod time.Duration // exists, so we can test pruning
 	lggr        logger.Logger
 }
 
-func newSampler(lgger logger.SugaredLogger) *sampler {
+func newSampler(lgger logger.SugaredLogger, samplingEnabled bool) *sampler {
 	return &sampler{
 		samples:     make(map[string]map[int32]any),
+		enabled:     samplingEnabled,
 		prunePeriod: defaultPrunePeriod,
 		lggr:        lgger,
 	}
 }
 
 // Sample is the method which decides whether we're going to send the data downstream or not.
 func (s *sampler) Sample(typ synchronization.TelemetryType, msg proto.Message) bool {
+	// If sampling is not enabled we want to send each and every telemetry message, so always return true.
+	if !s.enabled {
+		return true
+	}
+
 	fp, ots, err := fingerprint(typ, msg)
 	if err != nil {
 		if !errors.Is(err, errUnsupportedTelemetryType) {
@@ -75,6 +82,11 @@ func (s *sampler) Sample(typ synchronization.TelemetryType, msg proto.Message) b
 //
 // This method is non-blocking. It starts a goroutine and returns.
 func (s *sampler) StartPruningLoop(ctx context.Context, wg *sync.WaitGroup) {
+	// We don't need pruning if sampling is not enabled.
+	if !s.enabled {
+		return
+	}
+
 	wg.Add(1)
 	go func() {
 		defer wg.Done()

core/services/llo/telem/sampling_test.go

@@ -117,7 +117,7 @@ func TestSample(t *testing.T) {
 	lggr := logger.TestSugared(t)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	samplr := newSampler(lggr)
+	samplr := newSampler(lggr, true)
 	samplr.StartPruningLoop(ctx, &sync.WaitGroup{})
 
 	t0 := time.Now()
@@ -148,7 +148,7 @@ func TestPruningLoop(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	samplr := newSampler(lggr)
+	samplr := newSampler(lggr, true)
 	// We need a prune time of at least one second in order to detect outdated entries.
 	// The entries have second-based timestamps and cannot distinguish shorter intervals.
 	samplr.prunePeriod = time.Second
@@ -199,7 +199,7 @@ func TestPruningLoop_Exits(t *testing.T) {
 	wg := &sync.WaitGroup{}
 
 	// Start the sampler and its loop. This increments the waitgroup's counter.
-	samplr := newSampler(lggr)
+	samplr := newSampler(lggr, true)
 	samplr.StartPruningLoop(ctx, wg)
 
 	// Create a channel which will be closed when the waitgroup is done, i.e. when the loop is closed.

core/services/llo/telem/telemetry.go

@@ -50,6 +50,7 @@ type TelemeterParams struct {
 	CaptureObservationTelemetry bool
 	CaptureOutcomeTelemetry     bool
 	CaptureReportTelemetry      bool
+	SampleTelemetry             bool
 }
 
 func NewTelemeterService(params TelemeterParams) TelemeterService {
@@ -78,7 +79,7 @@ func newTelemeter(params TelemeterParams) *telemeter {
 		}, 10),
 		currentSeqNr:    make(map[string]uint64),
 		telemetryBuffer: make(map[string]map[uint64][]telemetryEntry),
-		sampler:         newSampler(logger.Sugared(params.Logger)),
+		sampler:         newSampler(logger.Sugared(params.Logger), params.SampleTelemetry),
 	}
 	if params.CaptureOutcomeTelemetry {
 		t.chOutcomeTelemetry = make(chan *datastreamsllo.LLOOutcomeTelemetry, 100) // only one per round so 100 buffer should be more than enough even for very fast rounds

core/services/ocr2/delegate.go

@@ -34,6 +34,7 @@ import (
 	ocr2keepers21config "github.com/smartcontractkit/chainlink-automation/pkg/v3/config"
 	ocr2keepers21 "github.com/smartcontractkit/chainlink-automation/pkg/v3/plugin"
 	evmconfig "github.com/smartcontractkit/chainlink-evm/pkg/config"
+
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/vault/vaulttypes"
 	"github.com/smartcontractkit/chainlink/v2/core/config/env"
 	syncerV2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer/v2"
@@ -200,6 +201,7 @@ type ocr2Config interface {
 	KeyBundleID() (string, error)
 	SimulateTransactions() bool
 	TraceLogging() bool
+	SampleTelemetry() bool
 	CaptureAutomationCustomTelemetry() bool
 	AllowNoBootstrappers() bool
 	KeyValueStoreRootDir() string
@@ -1447,6 +1449,7 @@ func (d *Delegate) newServicesLLO(
 		ChainID:                  rid.ChainID,
 
 		TraceLogging:                 d.cfg.OCR2().TraceLogging(),
+		SampleTelemetry:              d.cfg.OCR2().SampleTelemetry(),
 		BinaryNetworkEndpointFactory: d.peerWrapper.Peer2,
 		V2Bootstrappers:              bootstrapPeers,
 		ContractTransmitter:          provider.ContractTransmitter(),

core/services/ocr2/validate/config.go

@@ -22,6 +22,7 @@ type OCR2Config interface {
 	DefaultTransactionQueueDepth() uint32
 	SimulateTransactions() bool
 	TraceLogging() bool
+	SampleTelemetry() bool
 }
 
 type InsecureConfig interface {

core/web/resolver/testdata/config-empty-effective.toml

@@ -161,6 +161,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

core/web/resolver/testdata/config-full.toml

@@ -167,6 +167,7 @@ AllowNoBootstrappers = true
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

core/web/resolver/testdata/config-multi-chain-effective.toml

@@ -161,6 +161,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

docs/CONFIG.md

@@ -1018,6 +1018,7 @@ AllowNoBootstrappers = false # Default
 DefaultTransactionQueueDepth = 1 # Default
 SimulateTransactions = false # Default
 TraceLogging = false # Default
+SampleTelemetry = false # Default
 KeyValueStoreRootDir = '~/.chainlink-data' # Default
 ```
 
@@ -1140,6 +1141,12 @@ TraceLogging = false # Default
 ```
 TraceLogging enables trace level logging.
 
+### SampleTelemetry
+```toml
+SampleTelemetry = false # Default
+```
+SampleTelemetry enables telemetry sampling.
+
 ### KeyValueStoreRootDir
 ```toml
 KeyValueStoreRootDir = '~/.chainlink-data' # Default

testdata/scripts/config/merge_raw_configs.txtar

@@ -308,6 +308,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/default.txtar

@@ -173,6 +173,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/defaults-override.txtar

@@ -234,6 +234,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/disk-based-logging-disabled.txtar

@@ -217,6 +217,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/disk-based-logging-no-dir.txtar

@@ -217,6 +217,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/disk-based-logging.txtar

@@ -217,6 +217,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/fallback-override.txtar

@@ -315,6 +315,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/invalid-ocr-p2p.txtar

@@ -202,6 +202,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/invalid.txtar

@@ -213,6 +213,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/valid.txtar

@@ -214,6 +214,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]

testdata/scripts/node/validate/warnings.txtar

@@ -196,6 +196,7 @@ AllowNoBootstrappers = false
 DefaultTransactionQueueDepth = 1
 SimulateTransactions = false
 TraceLogging = false
+SampleTelemetry = false
 KeyValueStoreRootDir = '~/.chainlink-data'
 
 [OCR]