enhancement: plugin interface

conduit/metrics/metrics.go

@@ -16,6 +16,7 @@ func RegisterPrometheusMetrics(subsystem string) {
 	instantiateCollectors(subsystem)
 
 	_ = prometheus.Register(BlockImportTimeSeconds)
+	// TODO: why is this registered twice?
 	_ = prometheus.Register(BlockImportTimeSeconds)
 	_ = prometheus.Register(ImportedTxnsPerBlock)
 	_ = prometheus.Register(ImportedRoundGauge)

conduit/pipeline/pipeline.go

@@ -267,10 +267,14 @@ func (p *pipelineImpl) Init() error {
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not make %s config: %w", p.cfg.Importer.Name, err)
 		}
-		genesis, err := (*p.importer).Init(p.ctx, *p.initProvider, pluginConfig, importerLogger)
+		err = (*p.importer).Init(p.ctx, *p.initProvider, pluginConfig, importerLogger)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize importer (%s): %w", p.cfg.Importer.Name, err)
 		}
+		genesis, err := (*p.importer).GetGenesis()
+		if err != nil {
+			return fmt.Errorf("Pipeline.GetGenesis(): could not obtain Genesis from the importer (%s): %w", p.cfg.Importer.Name, err)
+		}
 		(*p.initProvider).SetGenesis(genesis)
 
 		// write pipeline metadata
@@ -413,7 +417,7 @@ func (p *pipelineImpl) Start() {
 					}
 					metrics.ImporterTimeSeconds.Observe(time.Since(importStart).Seconds())
 
-					// TODO: Verify that the block was build with a known protocol version.
+					// TODO: Verify that the block was built with a known protocol version.
 
 					// Start time currently measures operations after block fetching is complete.
 					// This is for backwards compatibility w/ Indexer's metrics

conduit/pipeline/pipeline_test.go

@@ -52,8 +52,12 @@ type mockImporter struct {
 	rndReqErr       error
 }
 
-func (m *mockImporter) Init(_ context.Context, _ data.InitProvider, cfg plugins.PluginConfig, _ *log.Logger) (*sdk.Genesis, error) {
+func (m *mockImporter) Init(_ context.Context, _ data.InitProvider, cfg plugins.PluginConfig, _ *log.Logger) error {
 	m.cfg = cfg
+	return nil
+}
+
+func (m *mockImporter) GetGenesis() (*sdk.Genesis, error) {
 	return &m.genesis, nil
 }
 
@@ -668,12 +672,12 @@ func (e *errorImporter) Metadata() plugins.Metadata {
 	return errorImporterMetadata
 }
 
-func (e *errorImporter) Init(_ context.Context, _ data.InitProvider, _ plugins.PluginConfig, _ *log.Logger) (*sdk.Genesis, error) {
-	return e.genesis, nil
+func (e *errorImporter) Init(_ context.Context, _ data.InitProvider, _ plugins.PluginConfig, _ *log.Logger) error {
+	return nil
 }
 
-func (e *errorImporter) Config() string {
-	return ""
+func (e *errorImporter) GetGenesis() (*sdk.Genesis, error) {
+	return e.genesis, nil
 }
 
 func (e *errorImporter) Close() error {

conduit/plugins/exporters/example/example_exporter.go

@@ -36,11 +36,6 @@ func (exp *exampleExporter) Init(_ context.Context, _ data.InitProvider, _ plugi
 	panic("not implemented")
 }
 
-// Config returns the unmarshaled config object
-func (exp *exampleExporter) Config() string {
-	panic("not implemented")
-}
-
 // Close provides the opportunity to close connections, flush buffers, etc. when the process is terminating
 func (exp *exampleExporter) Close() error {
 	panic("not implemented")

conduit/plugins/exporters/example/example_exporter_test.go

@@ -28,10 +28,6 @@ func TestExporterInit(t *testing.T) {
 	assert.Panics(t, func() { exExp.Init(context.Background(), nil, plugins.MakePluginConfig(""), nil) })
 }
 
-func TestExporterConfig(t *testing.T) {
-	assert.Panics(t, func() { exExp.Config() })
-}
-
 func TestExporterClose(t *testing.T) {
 	assert.Panics(t, func() { exExp.Close() })
 }

conduit/plugins/exporters/exporter.go

@@ -1,33 +1,14 @@
 package exporters
 
 import (
-	"context"
-
-	"github.com/sirupsen/logrus"
-
 	"github.com/algorand/conduit/conduit/data"
 	"github.com/algorand/conduit/conduit/plugins"
 )
 
 // Exporter defines the interface for plugins
 type Exporter interface {
-	// PluginMetadata implement this interface.
-	plugins.PluginMetadata
-
-	// Init will be called during initialization, before block data starts going through the pipeline.
-	// Typically used for things like initializing network connections.
-	// The ExporterConfig passed to Connect will contain the Unmarhsalled config file specific to this plugin.
-	// Should return an error if it fails--this will result in the Indexer process terminating.
-	Init(ctx context.Context, initProvider data.InitProvider, cfg plugins.PluginConfig, logger *logrus.Logger) error
-
-	// Config returns the configuration options used to create an Exporter.
-	// Initialized during Connect, it should return nil until the Exporter has been Connected.
-	Config() string
-
-	// Close will be called during termination of the Indexer process.
-	// There is no guarantee that plugin lifecycle hooks will be invoked in any specific order in relation to one another.
-	// Returns an error if it fails which will be surfaced in the logs, but the process is already terminating.
-	Close() error
+	// Plugin - implement this interface.
+	plugins.Plugin
 
 	// Receive is called for each block to be processed by the exporter.
 	// Should return an error on failure--retries are configurable.

conduit/plugins/exporters/filewriter/file_exporter.go

@@ -9,7 +9,6 @@ import (
 	"path"
 
 	"github.com/sirupsen/logrus"
-	"gopkg.in/yaml.v3"
 
 	"github.com/algorand/conduit/conduit/data"
 	"github.com/algorand/conduit/conduit/plugins"
@@ -68,11 +67,6 @@ func (exp *fileExporter) Init(_ context.Context, initProvider data.InitProvider,
 	return err
 }
 
-func (exp *fileExporter) Config() string {
-	ret, _ := yaml.Marshal(exp.cfg)
-	return string(ret)
-}
-
 func (exp *fileExporter) Close() error {
 	exp.logger.Infof("latest round on file: %d", exp.round)
 	return nil

conduit/plugins/exporters/filewriter/file_exporter_test.go

@@ -82,8 +82,6 @@ func TestExporterInitDefaults(t *testing.T) {
 			pcfg.DataDir = tempdir
 			err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), pcfg, logger)
 			require.NoError(t, err)
-			pluginConfig := fileExp.Config()
-			assert.Contains(t, pluginConfig, fmt.Sprintf("block-dir: %s", tc.expected))
 		})
 	}
 }
@@ -95,9 +93,7 @@ func TestExporterInit(t *testing.T) {
 
 	// creates a new output file
 	err := fileExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, 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))
+	assert.NoError(t, err)
 	fileExp.Close()
 
 	// can open existing file

conduit/plugins/exporters/noop/noop_exporter.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 
 	"github.com/sirupsen/logrus"
-	"gopkg.in/yaml.v3"
 
 	"github.com/algorand/conduit/conduit/data"
 	"github.com/algorand/conduit/conduit/plugins"
@@ -47,11 +46,6 @@ func (exp *noopExporter) Init(_ context.Context, initProvider data.InitProvider,
 	return nil
 }
 
-func (exp *noopExporter) Config() string {
-	ret, _ := yaml.Marshal(exp.cfg)
-	return string(ret)
-}
-
 func (exp *noopExporter) Close() error {
 	return nil
 }

conduit/plugins/exporters/noop/noop_exporter_test.go

@@ -5,7 +5,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"gopkg.in/yaml.v3"
 
 	"github.com/algorand/conduit/conduit"
 	"github.com/algorand/conduit/conduit/data"
@@ -39,16 +38,6 @@ func TestExporterInit(t *testing.T) {
 	assert.NoError(t, ne.Init(context.Background(), &conduit.PipelineInitProvider{}, plugins.MakePluginConfig(""), nil))
 }
 
-func TestExporterConfig(t *testing.T) {
-	defaultConfig := &ExporterConfig{}
-	expected, err := yaml.Marshal(defaultConfig)
-	if err != nil {
-		t.Fatalf("unable to Marshal default noop.ExporterConfig: %v", err)
-	}
-	assert.NoError(t, ne.Init(context.Background(), &conduit.PipelineInitProvider{}, plugins.MakePluginConfig(""), nil))
-	assert.Equal(t, string(expected), ne.Config())
-}
-
 func TestExporterClose(t *testing.T) {
 	assert.NoError(t, ne.Close())
 }

conduit/plugins/exporters/postgresql/postgresql_exporter.go

@@ -10,7 +10,6 @@ import (
 	"sync/atomic"
 
 	"github.com/sirupsen/logrus"
-	"gopkg.in/yaml.v3"
 
 	sdk "github.com/algorand/go-algorand-sdk/v2/types"
 	"github.com/algorand/indexer/idb"
@@ -142,11 +141,6 @@ func (exp *postgresqlExporter) Init(ctx context.Context, initProvider data.InitP
 	return nil
 }
 
-func (exp *postgresqlExporter) Config() string {
-	ret, _ := yaml.Marshal(exp.cfg)
-	return string(ret)
-}
-
 func (exp *postgresqlExporter) Close() error {
 	if exp.db != nil {
 		exp.db.Close()

conduit/plugins/exporters/postgresql/postgresql_exporter_test.go

@@ -57,16 +57,6 @@ func TestConnectDbFailure(t *testing.T) {
 	assert.ErrorContains(t, pgsqlExp.Init(context.Background(), conduit.MakePipelineInitProvider(&round, nil), cfg, logger), "connection string is empty for postgres")
 }
 
-func TestConfigDefault(t *testing.T) {
-	pgsqlExp := pgsqlConstructor.New()
-	defaultConfig := &ExporterConfig{}
-	expected, err := yaml.Marshal(defaultConfig)
-	if err != nil {
-		t.Fatalf("unable to Marshal default postgresql.ExporterConfig: %v", err)
-	}
-	assert.Equal(t, string(expected), pgsqlExp.Config())
-}
-
 func TestReceiveInvalidBlock(t *testing.T) {
 	pgsqlExp := pgsqlConstructor.New()
 	cfg := plugins.MakePluginConfig("test: true")

conduit/plugins/importers/algod/algod_importer.go

@@ -15,7 +15,6 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
-	"gopkg.in/yaml.v3"
 
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/algod"
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/common"
@@ -55,6 +54,7 @@ type algodImporter struct {
 	ctx     context.Context
 	cancel  context.CancelFunc
 	mode    int
+	genesis *sdk.Genesis
 }
 
 //go:embed sample.yaml
@@ -328,12 +328,12 @@ func (algodImp *algodImporter) catchupNode(network string, targetRound uint64) e
 	return err
 }
 
-func (algodImp *algodImporter) Init(ctx context.Context, initProvider data.InitProvider, cfg plugins.PluginConfig, logger *logrus.Logger) (*sdk.Genesis, error) {
+func (algodImp *algodImporter) Init(ctx context.Context, initProvider data.InitProvider, cfg plugins.PluginConfig, logger *logrus.Logger) error {
 	algodImp.ctx, algodImp.cancel = context.WithCancel(ctx)
 	algodImp.logger = logger
 	err := cfg.UnmarshalConfig(&algodImp.cfg)
 	if err != nil {
-		return nil, fmt.Errorf("connect failure in unmarshalConfig: %v", err)
+		return fmt.Errorf("connect failure in unmarshalConfig: %v", err)
 	}
 
 	// To support backwards compatibility with the daemon we default to archival mode
@@ -347,13 +347,13 @@ func (algodImp *algodImporter) Init(ctx context.Context, initProvider data.InitP
 	case followerModeStr:
 		algodImp.mode = followerMode
 	default:
-		return nil, fmt.Errorf("algod importer was set to a mode (%s) that wasn't supported", algodImp.cfg.Mode)
+		return fmt.Errorf("algod importer was set to a mode (%s) that wasn't supported", algodImp.cfg.Mode)
 	}
 
 	var client *algod.Client
 	u, err := url.Parse(algodImp.cfg.NetAddr)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	if u.Scheme != "http" && u.Scheme != "https" {
@@ -362,34 +362,34 @@ func (algodImp *algodImporter) Init(ctx context.Context, initProvider data.InitP
 	}
 	client, err = algod.MakeClient(algodImp.cfg.NetAddr, algodImp.cfg.Token)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	algodImp.aclient = client
-
-	genesisResponse, err := client.GetGenesis().Do(ctx)
+	genesisResponse, err := algodImp.aclient.GetGenesis().Do(algodImp.ctx)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	genesis := sdk.Genesis{}
 
 	// Don't fail on unknown properties here since the go-algorand and SDK genesis types differ slightly
 	err = json.LenientDecode([]byte(genesisResponse), &genesis)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	if reflect.DeepEqual(genesis, sdk.Genesis{}) {
-		return nil, fmt.Errorf("unable to fetch genesis file from API at %s", algodImp.cfg.NetAddr)
+		return fmt.Errorf("unable to fetch genesis file from API at %s", algodImp.cfg.NetAddr)
 	}
+	algodImp.genesis = &genesis
 
-	err = algodImp.catchupNode(genesis.Network, uint64(initProvider.NextDBRound()))
-
-	return &genesis, err
+	return algodImp.catchupNode(genesis.Network, uint64(initProvider.NextDBRound()))
 }
 
-func (algodImp *algodImporter) Config() string {
-	s, _ := yaml.Marshal(algodImp.cfg)
-	return string(s)
+func (algodImp *algodImporter) GetGenesis() (*sdk.Genesis, error) {
+	if algodImp.genesis != nil {
+		return algodImp.genesis, nil
+	}
+	return nil, fmt.Errorf("algod importer is missing its genesis: GetGenesis() should be called only after Init()")
 }
 
 func (algodImp *algodImporter) Close() error {