enhancement: plugin interface

.markdownlint.yml

@@ -0,0 +1,7 @@
+default: true
+
+MD010:
+  code_blocks: false
+
+MD013:
+  line_length: 200

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

@@ -409,7 +409,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

@@ -669,9 +669,9 @@ func (e *errorImporter) Init(_ context.Context, _ data.InitProvider, _ plugins.P
 	return e.genesis, nil
 }
 
-func (e *errorImporter) Config() string {
-	return ""
-}
+// func (e *errorImporter) Config() string {
+// 	return ""
+// }
 
 func (e *errorImporter) Close() error {
 	return nil

conduit/plugins/exporters/example/example_exporter.go

@@ -37,9 +37,9 @@ func (exp *exampleExporter) Init(_ context.Context, _ data.InitProvider, _ plugi
 }
 
 // Config returns the unmarshaled config object
-func (exp *exampleExporter) Config() string {
-	panic("not implemented")
-}
+// 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 {

conduit/plugins/exporters/example/example_exporter_test.go

@@ -28,9 +28,9 @@ 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 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

@@ -11,24 +11,15 @@ import (
 
 // Exporter defines the interface for plugins
 type Exporter interface {
-	// PluginMetadata implement this interface.
-	plugins.PluginMetadata
+	// Plugin - implement this interface.
+	plugins.Plugin
 
 	// 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.
+	// The ExporterConfig passed to Init will contain the Unmarhsalled config file specific to this plugin.
+	// Should return an error if it fails--this will result in the Conduit 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
-
 	// Receive is called for each block to be processed by the exporter.
 	// Should return an error on failure--retries are configurable.
 	Receive(exportData data.BlockData) error

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,10 +67,10 @@ 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) 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)

conduit/plugins/exporters/filewriter/file_exporter_test.go

@@ -82,8 +82,8 @@ 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))
+			// pluginConfig := fileExp.Config()
+			// assert.Contains(t, pluginConfig, fmt.Sprintf("block-dir: %s", tc.expected))
 		})
 	}
 }
@@ -95,9 +95,10 @@ 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)
+	// 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

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,10 +46,10 @@ 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) 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,15 +38,15 @@ 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 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,10 +141,10 @@ 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) Config() string {
+// 	ret, _ := yaml.Marshal(exp.cfg)
+// 	return string(ret)
+// }
 
 func (exp *postgresqlExporter) Close() error {
 	if exp.db != nil {

conduit/plugins/exporters/postgresql/postgresql_exporter_test.go

@@ -57,15 +57,15 @@ 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 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()

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"
@@ -382,10 +381,10 @@ func (algodImp *algodImporter) Init(ctx context.Context, initProvider data.InitP
 	return &genesis, err
 }
 
-func (algodImp *algodImporter) Config() string {
-	s, _ := yaml.Marshal(algodImp.cfg)
-	return string(s)
-}
+// func (algodImp *algodImporter) Config() string {
+// 	s, _ := yaml.Marshal(algodImp.cfg)
+// 	return string(s)
+// }
 
 func (algodImp *algodImporter) Close() error {
 	if algodImp.cancel != nil {