pipelineImpl: de-pointerize plugin interface fields

conduit/pipeline/pipeline.go

@@ -51,9 +51,9 @@
 
 	initProvider *data.InitProvider
 
-	importer         *importers.Importer
-	processors       []*processors.Processor
-	exporter         *exporters.Exporter
+	importer         importers.Importer
+	processors       []processors.Processor
+	exporter         exporters.Exporter
 	completeCallback []conduit.OnCompleteFunc
 
 	pipelineMetadata state
@@ -72,30 +72,30 @@
 }
 
 func (p *pipelineImpl) registerLifecycleCallbacks() {
-	if v, ok := (*p.importer).(conduit.Completed); ok {
+	if v, ok := p.importer.(conduit.Completed); ok {
 		p.completeCallback = append(p.completeCallback, v.OnComplete)
 	}
 	for _, processor := range p.processors {
-		if v, ok := (*processor).(conduit.Completed); ok {
+		if v, ok := processor.(conduit.Completed); ok {
 			p.completeCallback = append(p.completeCallback, v.OnComplete)
 		}
 	}
-	if v, ok := (*p.exporter).(conduit.Completed); ok {
+	if v, ok := p.exporter.(conduit.Completed); ok {
 		p.completeCallback = append(p.completeCallback, v.OnComplete)
 	}
 }
 
 func (p *pipelineImpl) registerPluginMetricsCallbacks() {
 	var collectors []prometheus.Collector
-	if v, ok := (*p.importer).(conduit.PluginMetrics); ok {
+	if v, ok := p.importer.(conduit.PluginMetrics); ok {
 		collectors = append(collectors, v.ProvideMetrics(p.cfg.Metrics.Prefix)...)
 	}
 	for _, processor := range p.processors {
-		if v, ok := (*processor).(conduit.PluginMetrics); ok {
+		if v, ok := processor.(conduit.PluginMetrics); ok {
 			collectors = append(collectors, v.ProvideMetrics(p.cfg.Metrics.Prefix)...)
 		}
 	}
-	if v, ok := (*p.exporter).(conduit.PluginMetrics); ok {
+	if v, ok := p.exporter.(conduit.PluginMetrics); ok {
 		collectors = append(collectors, v.ProvideMetrics(p.cfg.Metrics.Prefix)...)
 	}
 	for _, c := range collectors {
@@ -108,8 +108,8 @@
 func (p *pipelineImpl) makeConfig(cfg data.NameConfigPair, pluginType plugins.PluginType) (*log.Logger, plugins.PluginConfig, error) {
 	configs, err := yaml.Marshal(cfg.Config)
 	if err != nil {
 		return nil, plugins.PluginConfig{}, fmt.Errorf("makeConfig(): could not serialize config: %w", err)
 	}
 
 	lgr := log.New()
 	lgr.SetOutput(p.logger.Out)
@@ -122,8 +122,8 @@
 		config.DataDir = path.Join(p.cfg.ConduitArgs.ConduitDataDir, fmt.Sprintf("%s_%s", pluginType, cfg.Name))
 		err = os.MkdirAll(config.DataDir, os.ModePerm)
 		if err != nil {
 			return nil, plugins.PluginConfig{}, fmt.Errorf("makeConfig: unable to create plugin data directory: %w", err)
 		}
 	}
 
 	return lgr, config, nil
@@ -143,23 +143,23 @@
 	}
 	var parts []overridePart
 
-	if v, ok := (*p.importer).(conduit.RoundRequestor); ok {
+	if v, ok := p.importer.(conduit.RoundRequestor); ok {
 		parts = append(parts, overridePart{
 			RoundRequest: v.RoundRequest,
 			cfg:          p.cfg.Importer,
 			t:            plugins.Importer,
 		})
 	}
 	for idx, processor := range p.processors {
-		if v, ok := (*processor).(conduit.RoundRequestor); ok {
+		if v, ok := processor.(conduit.RoundRequestor); ok {
 			parts = append(parts, overridePart{
 				RoundRequest: v.RoundRequest,
 				cfg:          p.cfg.Processors[idx],
 				t:            plugins.Processor,
 			})
 		}
 	}
-	if v, ok := (*p.exporter).(conduit.RoundRequestor); ok {
+	if v, ok := p.exporter.(conduit.RoundRequestor); ok {
 		parts = append(parts, overridePart{
 			RoundRequest: v.RoundRequest,
 			cfg:          p.cfg.Exporter,
@@ -173,8 +173,8 @@
 	for _, part := range parts {
 		_, config, err := p.makeConfig(part.cfg, part.t)
 		if err != nil {
 			return 0, err
 		}
 		rnd, err := part.RoundRequest(config)
 		if err != nil {
 			return 0, err
@@ -204,16 +204,16 @@
 	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
 }
@@ -232,12 +232,12 @@
 		var err error
 		profFile, err := os.Create(p.cfg.CPUProfile)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): unable to create profile: %w", err)
 		}
 		p.profFile = profFile
 		err = pprof.StartCPUProfile(profFile)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): unable to start pprof: %w", err)
 		}
 	}
 
@@ -286,7 +286,7 @@
 		var telemetryErr error
 		telemetryClient, telemetryErr = p.initializeTelemetry()
 		if telemetryErr != nil {
 			p.logger.Warnf("Telemetry initialization failed, continuing without telemetry: %s", telemetryErr)
 		} else {
 			// Try sending a startup event. If it fails, log a warning and continue
 			event := telemetryClient.MakeTelemetryStartupEvent()
@@ -304,16 +304,16 @@
 	{
 		importerLogger, pluginConfig, err := p.makeConfig(p.cfg.Importer, plugins.Importer)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not make %s config: %w", p.cfg.Importer.Name, err)
 		}
-		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()
+		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
@@ -326,8 +326,8 @@
 		p.pipelineMetadata.Network = genesis.Network
 		err = p.pipelineMetadata.encodeToFile(p.cfg.ConduitArgs.ConduitDataDir)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init() failed to write metadata to file: %w", err)
 		}
 
 		p.logger.Infof("Initialized Importer: %s", p.cfg.Importer.Name)
 	}
@@ -337,11 +337,11 @@
 		ncPair := p.cfg.Processors[idx]
 		logger, config, err := p.makeConfig(ncPair, plugins.Processor)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", ncPair, err)
 		}
-		err = (*processor).Init(p.ctx, *p.initProvider, config, logger)
+		err = processor.Init(p.ctx, *p.initProvider, config, logger)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", ncPair.Name, err)
 		}
 		p.logger.Infof("Initialized Processor: %s", ncPair.Name)
 	}
@@ -350,12 +350,12 @@
 	{
 		logger, config, err := p.makeConfig(p.cfg.Exporter, plugins.Exporter)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize processor (%s): %w", p.cfg.Exporter.Name, err)
 		}
-		err = (*p.exporter).Init(p.ctx, *p.initProvider, config, logger)
+		err = p.exporter.Init(p.ctx, *p.initProvider, config, logger)
 		if err != nil {
 			return fmt.Errorf("Pipeline.Init(): could not initialize Exporter (%s): %w", p.cfg.Exporter.Name, err)
 		}
 		p.logger.Infof("Initialized Exporter: %s", p.cfg.Exporter.Name)
 	}
 
@@ -388,20 +388,20 @@
 		}
 	}
 
-	if err := (*p.importer).Close(); err != nil {
+	if err := p.importer.Close(); err != nil {
 		// Log and continue on closing the rest of the pipeline
-		p.logger.Errorf("Pipeline.Stop(): Importer (%s) error on close: %v", (*p.importer).Metadata().Name, err)
+		p.logger.Errorf("Pipeline.Stop(): Importer (%s) error on close: %v", p.importer.Metadata().Name, err)
 	}
 
 	for _, processor := range p.processors {
-		if err := (*processor).Close(); err != nil {
+		if err := processor.Close(); err != nil {
 			// Log and continue on closing the rest of the pipeline
-			p.logger.Errorf("Pipeline.Stop(): Processor (%s) error on close: %v", (*processor).Metadata().Name, err)
+			p.logger.Errorf("Pipeline.Stop(): Processor (%s) error on close: %v", processor.Metadata().Name, err)
 		}
 	}
 
-	if err := (*p.exporter).Close(); err != nil {
-		p.logger.Errorf("Pipeline.Stop(): Exporter (%s) error on close: %v", (*p.exporter).Metadata().Name, err)
+	if err := p.exporter.Close(); err != nil {
+		p.logger.Errorf("Pipeline.Stop(): Exporter (%s) error on close: %v", p.exporter.Metadata().Name, err)
 	}
 }
 
@@ -460,7 +460,7 @@
 					p.logger.Infof("Pipeline round: %v", p.pipelineMetadata.NextRound)
 					// fetch block
 					importStart := time.Now()
-					blkData, err := (*p.importer).GetBlock(p.pipelineMetadata.NextRound)
+					blkData, err := p.importer.GetBlock(p.pipelineMetadata.NextRound)
 					if err != nil {
 						p.logger.Errorf("%v", err)
 						p.setError(err)
@@ -477,18 +477,18 @@
 					start := time.Now()
 					for _, proc := range p.processors {
 						processorStart := time.Now()
-						blkData, err = (*proc).Process(blkData)
+						blkData, err = proc.Process(blkData)
 						if err != nil {
 							p.logger.Errorf("%v", err)
 							p.setError(err)
 							retry++
 							goto pipelineRun
 						}
-						metrics.ProcessorTimeSeconds.WithLabelValues((*proc).Metadata().Name).Observe(time.Since(processorStart).Seconds())
+						metrics.ProcessorTimeSeconds.WithLabelValues(proc.Metadata().Name).Observe(time.Since(processorStart).Seconds())
 					}
 					// run through exporter
 					exporterStart := time.Now()
-					err = (*p.exporter).Receive(blkData)
+					err = p.exporter.Receive(blkData)
 					if err != nil {
 						p.logger.Errorf("%v", err)
 						p.setError(err)
@@ -563,47 +563,44 @@
 		logger:       logger,
 		initProvider: nil,
 		importer:     nil,
-		processors:   []*processors.Processor{},
+		processors:   []processors.Processor{},
 		exporter:     nil,
 	}
 
 	importerName := cfg.Importer.Name
 
-	importerBuilder, err := importers.ImporterBuilderByName(importerName)
+	importerConstructor, err := importers.ImporterConstructorByName(importerName)
 	if err != nil {
 		return nil, fmt.Errorf("MakePipeline(): could not build importer '%s': %w", importerName, err)
 	}
 
-	importer := importerBuilder.New()
-	pipeline.importer = &importer
+	pipeline.importer = importerConstructor.New()
 	logger.Infof("Found Importer: %s", importerName)
 
 	// ---
 
 	for _, processorConfig := range cfg.Processors {
 		processorName := processorConfig.Name
 
-		processorBuilder, err := processors.ProcessorBuilderByName(processorName)
+		processorConstructor, err := processors.ProcessorConstructorByName(processorName)
 		if err != nil {
 			return nil, fmt.Errorf("MakePipeline(): could not build processor '%s': %w", processorName, err)
 		}
 
-		processor := processorBuilder.New()
-		pipeline.processors = append(pipeline.processors, &processor)
+		pipeline.processors = append(pipeline.processors, processorConstructor.New())
 		logger.Infof("Found Processor: %s", processorName)
 	}
 
 	// ---
 
 	exporterName := cfg.Exporter.Name
 
-	exporterBuilder, err := exporters.ExporterBuilderByName(exporterName)
+	exporterConstructor, err := exporters.ExporterConstructorByName(exporterName)
 	if err != nil {
 		return nil, fmt.Errorf("MakePipeline(): could not build exporter '%s': %w", exporterName, err)
 	}
 
-	exporter := exporterBuilder.New()
-	pipeline.exporter = &exporter
+	pipeline.exporter = exporterConstructor.New()
 	logger.Infof("Found Exporter: %s", exporterName)
 
 	return pipeline, nil
@@ -615,22 +612,22 @@
 	logger.Infof("Creating PID file at: %s", pidFilePath)
 	fout, err := os.Create(pidFilePath)
 	if err != nil {
 		err = fmt.Errorf("%s: could not create pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 
 	if _, err = fmt.Fprintf(fout, "%d", os.Getpid()); err != nil {
 		err = fmt.Errorf("%s: could not write pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 
 	err = fout.Close()
 	if err != nil {
 		err = fmt.Errorf("%s: could not close pid file, %v", pidFilePath, err)
 		logger.Error(err)
 		return err
 	}
 	return err
 }

conduit/pipeline/pipeline_test.go

@@ -233,9 +233,9 @@ func mockPipeline(t *testing.T, dataDir string) (*pipelineImpl, *test.Hook, *moc
 		},
 		logger:       l,
 		initProvider: nil,
-		importer:     &pImporter,
-		processors:   []*processors.Processor{&pProcessor},
-		exporter:     &pExporter,
+		importer:     pImporter,
+		processors:   []processors.Processor{pProcessor},
+		exporter:     pExporter,
 		pipelineMetadata: state{
 			GenesisHash: "",
 			Network:     "",
@@ -271,9 +271,9 @@ func TestPipelineRun(t *testing.T) {
 		cf:               cf,
 		logger:           l,
 		initProvider:     nil,
-		importer:         &pImporter,
-		processors:       []*processors.Processor{&pProcessor},
-		exporter:         &pExporter,
+		importer:         pImporter,
+		processors:       []processors.Processor{pProcessor},
+		exporter:         pExporter,
 		completeCallback: []conduit.OnCompleteFunc{cbComplete.OnComplete},
 		pipelineMetadata: state{
 			NextRound:   0,
@@ -371,9 +371,9 @@ func TestPipelineErrors(t *testing.T) {
 		},
 		logger:           l,
 		initProvider:     nil,
-		importer:         &pImporter,
-		processors:       []*processors.Processor{&pProcessor},
-		exporter:         &pExporter,
+		importer:         pImporter,
+		processors:       []processors.Processor{pProcessor},
+		exporter:         pExporter,
 		completeCallback: []conduit.OnCompleteFunc{cbComplete.OnComplete},
 		pipelineMetadata: state{},
 	}
@@ -440,9 +440,9 @@ func Test_pipelineImpl_registerLifecycleCallbacks(t *testing.T) {
 		cfg:          &data.Config{},
 		logger:       l,
 		initProvider: nil,
-		importer:     &pImporter,
-		processors:   []*processors.Processor{&pProcessor, &pProcessor},
-		exporter:     &pExporter,
+		importer:     pImporter,
+		processors:   []processors.Processor{pProcessor, pProcessor},
+		exporter:     pExporter,
 	}
 
 	// Each plugin implements the Completed interface, so there should be 4
@@ -485,7 +485,7 @@ func TestGenesisHash(t *testing.T) {
 
 	// mock a different genesis hash
 	var pImporter importers.Importer = &mockImporter{genesis: sdk.Genesis{Network: "dev"}}
-	pImpl.importer = &pImporter
+	pImpl.importer = pImporter
 	err = pImpl.Init()
 	assert.Contains(t, err.Error(), "genesis hash in metadata does not match")
 }
@@ -797,9 +797,9 @@ func TestPipelineRetryVariables(t *testing.T) {
 				},
 				logger:       l,
 				initProvider: nil,
-				importer:     &pImporter,
-				processors:   []*processors.Processor{&pProcessor},
-				exporter:     &pExporter,
+				importer:     pImporter,
+				processors:   []processors.Processor{pProcessor},
+				exporter:     pExporter,
 				pipelineMetadata: state{
 					GenesisHash: "",
 					Network:     "",

conduit/plugins/exporters/exporter_factory.go

@@ -34,8 +34,8 @@ func Register(name string, constructor ExporterConstructor) {
 	Exporters[name] = constructor
 }
 
-// ExporterBuilderByName returns a Processor constructor for the name provided
-func ExporterBuilderByName(name string) (ExporterConstructor, error) {
+// ExporterConstructorByName returns a Processor constructor for the name provided
+func ExporterConstructorByName(name string) (ExporterConstructor, error) {
 	constructor, ok := Exporters[name]
 	if !ok {
 		return nil, fmt.Errorf("no Exporter Constructor for %s", name)

conduit/plugins/exporters/exporter_factory_test.go

@@ -37,14 +37,14 @@ func TestExporterByNameSuccess(t *testing.T) {
 	me := mockExporter{}
 	Register("foobar", &mockExporterConstructor{&me})
 
-	expC, err := ExporterBuilderByName("foobar")
+	expC, err := ExporterConstructorByName("foobar")
 	assert.NoError(t, err)
 	exp := expC.New()
 	assert.Implements(t, (*Exporter)(nil), exp)
 }
 
 func TestExporterByNameNotFound(t *testing.T) {
-	_, err := ExporterBuilderByName("barfoo")
+	_, err := ExporterConstructorByName("barfoo")
 	expectedErr := "no Exporter Constructor for barfoo"
 	assert.EqualError(t, err, expectedErr)
 }

conduit/plugins/exporters/noop/noop_exporter_test.go

@@ -21,7 +21,7 @@ var ne = nc.New()
 func TestExporterBuilderByName(t *testing.T) {
 	// init() has already registered the noop exporter
 	assert.Contains(t, exporters.Exporters, metadata.Name)
-	neBuilder, err := exporters.ExporterBuilderByName(metadata.Name)
+	neBuilder, err := exporters.ExporterConstructorByName(metadata.Name)
 	assert.NoError(t, err)
 	ne := neBuilder.New()
 	assert.Implements(t, (*exporters.Exporter)(nil), ne)

conduit/plugins/importers/importer_factory.go

@@ -4,9 +4,9 @@ import (
 	"fmt"
 )
 
-// Constructor must be implemented by each Importer.
+// ImporterConstructor must be implemented by each Importer.
 // It provides a basic no-arg constructor for instances of an ImporterImpl.
-type Constructor interface {
+type ImporterConstructor interface {
 	// New should return an instantiation of a Importer.
 	// Configuration values should be passed and can be processed during `Init()`.
 	New() Importer
@@ -21,20 +21,20 @@ func (f ImporterConstructorFunc) New() Importer {
 }
 
 // Importers are the constructors to build importer plugins.
-var Importers = make(map[string]Constructor)
+var Importers = make(map[string]ImporterConstructor)
 
 // Register is used to register Constructor implementations. This mechanism allows
 // for loose coupling between the configuration and the implementation. It is extremely similar to the way sql.DB
 // drivers are configured and used.
-func Register(name string, constructor Constructor) {
+func Register(name string, constructor ImporterConstructor) {
 	if _, ok := Importers[name]; ok {
 		panic(fmt.Errorf("importer %s already registered", name))
 	}
 	Importers[name] = constructor
 }
 
-// ImporterBuilderByName returns a Importer constructor for the name provided
-func ImporterBuilderByName(name string) (Constructor, error) {
+// ImporterConstructorByName returns a Importer constructor for the name provided
+func ImporterConstructorByName(name string) (ImporterConstructor, error) {
 	constructor, ok := Importers[name]
 	if !ok {
 		return nil, fmt.Errorf("no Importer Constructor for %s", name)