telemetry: Push telemetry event to OpenSearch on pipeline startup

conduit/data/block_export_data.go

@@ -2,6 +2,8 @@ package data
 
 import (
 	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+
+	"github.com/algorand/conduit/conduit/telemetry"
 )
 
 // RoundProvider is the interface which all data types sent to Exporters should implement
@@ -16,6 +18,7 @@ type InitProvider interface {
 	GetGenesis() *sdk.Genesis
 	SetGenesis(*sdk.Genesis)
 	NextDBRound() sdk.Round
+	GetTelemetryClient() *telemetry.OpenSearchClient
 }
 
 // BlockData is provided to the Exporter on each round.

conduit/data/config.go

@@ -41,6 +41,15 @@ type Metrics struct {
 	Prefix string `yaml:"prefix"`
 }
 
+// Telemetry configs for sending Telemetry to OpenSearch
+type Telemetry struct {
+	Enabled  bool   `yaml:"enabled"`
+	URI      string `yaml:"uri"`
+	Index    string `yaml:"index"`
+	UserName string `yaml:"username"`
+	Password string `yaml:"password"`
+}
+
 // Config stores configuration specific to the conduit pipeline
 type Config struct {
 	// ConduitArgs are the program inputs. Should not be serialized for config.
@@ -61,6 +70,8 @@ type Config struct {
 	RetryCount uint64 `yaml:"retry-count"`
 	// RetryDelay is a duration amount interpreted from a string
 	RetryDelay time.Duration `yaml:"retry-delay"`
+
+	Telemetry Telemetry `yaml:"telemetry"`
 }
 
 // Valid validates pipeline config

conduit/data/config_test.go

@@ -85,7 +85,6 @@ exporters:
   name: "noop"
   config:
     connectionstring: ""`, "field exporters not found"},
-
 		{"config not configs", `---
 log-level: info
 importer:

conduit/init_provider.go

@@ -2,19 +2,23 @@ package conduit
 
 import (
 	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+
+	"github.com/algorand/conduit/conduit/telemetry"
 )
 
 // PipelineInitProvider algod based init provider
 type PipelineInitProvider struct {
-	currentRound *sdk.Round
-	genesis      *sdk.Genesis
+	currentRound    *sdk.Round
+	genesis         *sdk.Genesis
+	telemetryClient *telemetry.OpenSearchClient
 }
 
 // MakePipelineInitProvider constructs an init provider.
-func MakePipelineInitProvider(currentRound *sdk.Round, genesis *sdk.Genesis) *PipelineInitProvider {
+func MakePipelineInitProvider(currentRound *sdk.Round, genesis *sdk.Genesis, client *telemetry.OpenSearchClient) *PipelineInitProvider {
 	return &PipelineInitProvider{
-		currentRound: currentRound,
-		genesis:      genesis,
+		currentRound:    currentRound,
+		genesis:         genesis,
+		telemetryClient: client,
 	}
 }
 
@@ -32,3 +36,8 @@ func (a *PipelineInitProvider) GetGenesis() *sdk.Genesis {
 func (a *PipelineInitProvider) NextDBRound() sdk.Round {
 	return *a.currentRound
 }
+
+// GetTelemetryClient gets the telemetry state in the init provider
+func (a *PipelineInitProvider) GetTelemetryClient() *telemetry.OpenSearchClient {
+	return a.telemetryClient
+}

conduit/pipeline/metadata.go

@@ -18,6 +18,7 @@ type state struct {
 	GenesisHash string `json:"genesis-hash"`
 	Network     string `json:"network"`
 	NextRound   uint64 `json:"next-round"`
+	TelemetryID string `json:"telemetry-id,omitempty"`
 }
 
 // encodeToFile writes the state object to the dataDir

conduit/pipeline/metadata_test.go

@@ -54,15 +54,18 @@ func TestBlockMetaDataFile(t *testing.T) {
 	assert.Equal(t, pipelineMetadata.GenesisHash, metaData.GenesisHash)
 	assert.Equal(t, pipelineMetadata.NextRound, metaData.NextRound)
 	assert.Equal(t, pipelineMetadata.Network, metaData.Network)
+	assert.Equal(t, pipelineMetadata.TelemetryID, metaData.TelemetryID)
 
 	// Test that file encodes correctly
 	pipelineMetadata.GenesisHash = "HASH"
 	pipelineMetadata.NextRound = 7
+	pipelineMetadata.TelemetryID = "SOME_ID"
 	err = pipelineMetadata.encodeToFile(datadir)
 	assert.NoError(t, err)
 	metaData, err = readBlockMetadata(datadir)
 	assert.NoError(t, err)
 	assert.Equal(t, "HASH", metaData.GenesisHash)
 	assert.Equal(t, uint64(7), metaData.NextRound)
 	assert.Equal(t, pipelineMetadata.Network, metaData.Network)
+	assert.Equal(t, pipelineMetadata.TelemetryID, metaData.TelemetryID)
 }

conduit/pipeline/pipeline.go

@@ -25,6 +25,7 @@ import (
 	"github.com/algorand/conduit/conduit/plugins/exporters"
 	"github.com/algorand/conduit/conduit/plugins/importers"
 	"github.com/algorand/conduit/conduit/plugins/processors"
+	"github.com/algorand/conduit/conduit/telemetry"
 )
 
 // Pipeline is a struct that orchestrates the entire
@@ -195,6 +196,25 @@ func (p *pipelineImpl) pluginRoundOverride() (uint64, error) {
 	return pluginOverride, nil
 }
 
+// initializeTelemetry initializes telemetry and reads or sets the GUID in the metadata.
+func (p *pipelineImpl) initializeTelemetry() (*telemetry.OpenSearchClient, error) {
+	telemetryConfig := telemetry.MakeTelemetryConfig(p.cfg.Telemetry.URI, p.cfg.Telemetry.Index, p.cfg.Telemetry.UserName, p.cfg.Telemetry.Password)
+	telemetryClient, err := telemetry.MakeOpenSearchClient(telemetryConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize telemetry: %w", err)
+	}
+	p.logger.Infof("Telemetry initialized with URI: %s", telemetryConfig.URI)
+
+	// If GUID is not in metadata, save it. Otherwise, use the GUID from metadata.
+	if p.pipelineMetadata.TelemetryID == "" {
+		p.pipelineMetadata.TelemetryID = telemetryClient.TelemetryConfig.GUID
+	} else {
+		telemetryClient.TelemetryConfig.GUID = p.pipelineMetadata.TelemetryID
+	}
+
+	return telemetryClient, nil
+}
+
 // Init prepares the pipeline for processing block data
 func (p *pipelineImpl) Init() error {
 	p.logger.Infof("Starting Pipeline Initialization")
@@ -254,8 +274,26 @@ func (p *pipelineImpl) Init() error {
 
 	// InitProvider
 	round := sdk.Round(p.pipelineMetadata.NextRound)
-	// Initial genesis object is nil--gets updated after importer.Init
-	var initProvider data.InitProvider = conduit.MakePipelineInitProvider(&round, nil)
+
+	// Initialize Telemetry
+	var telemetryClient *telemetry.OpenSearchClient
+	if p.cfg.Telemetry.Enabled {
+		// If telemetry cannot be initialized, log a warning and continue
+		// pipeline initialization.
+		var telemetryErr error
+		telemetryClient, telemetryErr = p.initializeTelemetry()
+		if telemetryErr != nil {
+			p.logger.Warn("Telemetry initialization failed. Continuing without telemetry.")
+		}
+		// Try sending a startup event. If it fails, log a warning and continue
+		event := telemetryClient.MakeTelemetryStartupEvent()
+		if telemetryErr = telemetryClient.SendEvent(event); telemetryErr != nil {
+			p.logger.Warnf("failed to send telemetry event: %s", telemetryErr)
+		}
+	}
+
+	// Initial genesis object is nil and gets updated after importer.Init
+	var initProvider data.InitProvider = conduit.MakePipelineInitProvider(&round, nil, telemetryClient)
 	p.initProvider = &initProvider
 
 	// Initialize Importer

conduit/pipeline/pipeline_test.go

@@ -523,6 +523,36 @@ func TestPipelineMetricsConfigs(t *testing.T) {
 	assert.Equal(t, pImpl.cfg.Metrics.Prefix, prefixOverride)
 }
 
+func TestPipelineTelemetryConfigs(t *testing.T) {
+	pImpl, _, _, _, _ := mockPipeline(t, "")
+
+	// telemetry OFF, check that client is nil
+	pImpl.cfg.Telemetry = data.Telemetry{
+		Enabled: false,
+	}
+	pImpl.Init()
+	client := (*pImpl.initProvider).GetTelemetryClient()
+	assert.Nil(t, client)
+
+	// telemetry ON
+	pImpl.cfg.Telemetry = data.Telemetry{
+		Enabled:  true,
+		URI:      "test-uri",
+		Index:    "test-index",
+		UserName: "test-username",
+		Password: "test-password",
+	}
+	pImpl.Init()
+	client = (*pImpl.initProvider).GetTelemetryClient()
+	assert.NotNil(t, client)
+	assert.NotNil(t, client.Client)
+	assert.Equal(t, true, client.TelemetryConfig.Enable)
+	assert.Equal(t, "test-uri", client.TelemetryConfig.URI)
+	assert.Equal(t, "test-index", client.TelemetryConfig.Index)
+	assert.Equal(t, "test-username", client.TelemetryConfig.UserName)
+	assert.Equal(t, "test-password", client.TelemetryConfig.Password)
+}
+
 func TestRoundOverrideValidConflict(t *testing.T) {
 	t.Run("processor_no_conflict", func(t *testing.T) {
 		pImpl, _, mImporter, mProcessor, _ := mockPipeline(t, "")

conduit/plugins/exporters/filewriter/file_exporter_test.go

@@ -80,7 +80,7 @@ func TestExporterInitDefaults(t *testing.T) {
 			defer fileExp.Close()
 			pcfg := plugins.MakePluginConfig(fmt.Sprintf("block-dir: %s", tc.blockdir))
 			pcfg.DataDir = tempdir
-			err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), pcfg, logger)
+			err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), pcfg, logger)
 			require.NoError(t, err)
 			pluginConfig := fileExp.Config()
 			assert.Contains(t, pluginConfig, fmt.Sprintf("block-dir: %s", tc.expected))
@@ -94,14 +94,14 @@ func TestExporterInit(t *testing.T) {
 	defer fileExp.Close()
 
 	// creates a new output file
-	err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), plugins.MakePluginConfig(config), logger)
+	err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), plugins.MakePluginConfig(config), logger)
 	pluginConfig := fileExp.Config()
 	configWithDefault := config + "filename-pattern: '%[1]d_block.json'\n" + "drop-certificate: false\n"
 	assert.Equal(t, configWithDefault, string(pluginConfig))
 	fileExp.Close()
 
 	// can open existing file
-	err = fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), plugins.MakePluginConfig(config), logger)
+	err = fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), plugins.MakePluginConfig(config), logger)
 	assert.NoError(t, err)
 	fileExp.Close()
 }
@@ -125,7 +125,7 @@ func sendData(t *testing.T, fileExp exporters.Exporter, config string, numRounds
 
 	// initialize
 	rnd := sdk.Round(0)
-	err = fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&rnd, nil), plugins.MakePluginConfig(config), logger)
+	err = fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&rnd, nil, nil), plugins.MakePluginConfig(config), logger)
 	require.NoError(t, err)
 
 	// incorrect round
@@ -186,7 +186,7 @@ func TestExporterClose(t *testing.T) {
 	config, _ := getConfig(t)
 	fileExp := fileCons.New()
 	rnd := sdk.Round(0)
-	fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&rnd, nil), plugins.MakePluginConfig(config), logger)
+	fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&rnd, nil, nil), plugins.MakePluginConfig(config), logger)
 	require.NoError(t, fileExp.Close())
 }
 

conduit/plugins/exporters/postgresql/postgresql_exporter_test.go

@@ -41,20 +41,20 @@ func TestExporterMetadata(t *testing.T) {
 func TestConnectDisconnectSuccess(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("test: true\nconnection-string: ''")
-	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}), cfg, logger))
+	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}, nil), cfg, logger))
 	assert.NoError(t, pgsqlExp.Close())
 }
 
 func TestConnectUnmarshalFailure(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("'")
-	assert.ErrorContains(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), cfg, logger), "connect failure in unmarshalConfig")
+	assert.ErrorContains(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), cfg, logger), "connect failure in unmarshalConfig")
 }
 
 func TestConnectDbFailure(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("")
-	assert.ErrorContains(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), cfg, logger), "connection string is empty for postgres")
+	assert.ErrorContains(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil, nil), cfg, logger), "connection string is empty for postgres")
 }
 
 func TestConfigDefault(t *testing.T) {
@@ -70,7 +70,7 @@ func TestConfigDefault(t *testing.T) {
 func TestReceiveInvalidBlock(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("test: true")
-	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}), cfg, logger))
+	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}, nil), cfg, logger))
 	invalidBlock := data.BlockData{
 		BlockHeader: sdk.BlockHeader{
 			Round: 1,
@@ -86,7 +86,7 @@ func TestReceiveInvalidBlock(t *testing.T) {
 func TestReceiveAddBlockSuccess(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("test: true")
-	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}), cfg, logger))
+	assert.NoError(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}, nil), cfg, logger))
 
 	block := data.BlockData{
 		BlockHeader: sdk.BlockHeader{},
@@ -102,7 +102,7 @@ func TestPostgresqlExporterInit(t *testing.T) {
 	cfg := plugins.MakePluginConfig("test: true")
 
 	// genesis hash mismatch
-	initProvider := conduit.MakePipelineInitProvider(&round, &sdk.Genesis{})
+	initProvider := conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}, nil)
 	initProvider.SetGenesis(&sdk.Genesis{
 		Network: "test",
 	})
@@ -111,7 +111,7 @@ func TestPostgresqlExporterInit(t *testing.T) {
 
 	// incorrect round
 	round = 1
-	err = pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}), cfg, logger)
+	err = pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, &sdk.Genesis{}, nil), cfg, logger)
 	assert.Contains(t, err.Error(), "initializing block round 1 but next round to account is 0")
 }