Pipelining followups

conduit/pipeline/common.go

@@ -7,6 +7,8 @@ import (
 
 	log "github.com/sirupsen/logrus"
 
+	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+
 	"github.com/algorand/conduit/conduit/data"
 )
 
@@ -20,14 +22,14 @@ func HandlePanic(logger *log.Logger) {
 type empty struct{}
 
 type pluginInput interface {
-	uint64 | data.BlockData | string
+	uint64 | data.BlockData | string | empty
 }
 
 type pluginOutput interface {
-	pluginInput | empty
+	pluginInput | *sdk.Genesis
 }
 
-// Retries is a wrapper for retrying a function call f() with a cancellation context,
+// retries is a wrapper for retrying a function call f() with a cancellation context,
 // a delay and a max retry count. It attempts to call the wrapped function at least once
 // and only after the first attempt will pay attention to a context cancellation.
 // This can allow the pipeline to receive a cancellation and guarantee attempting to finish
@@ -45,7 +47,7 @@ type pluginOutput interface {
 //   - when p.cfg.retryCount > 0, the error will be a join of all the errors encountered during the retries
 //   - when p.cfg.retryCount == 0, the error will be the last error encountered
 //   - the returned duration dur is the total time spent in the function, including retries
-func Retries[X pluginInput, Y pluginOutput](f func(x X) (Y, error), x X, p *pipelineImpl, msg string) (y Y, dur time.Duration, err error) {
+func retries[X pluginInput, Y pluginOutput](f func(x X) (Y, error), x X, p *pipelineImpl, msg string) (y Y, dur time.Duration, err error) {
 	start := time.Now()
 
 	for i := uint64(0); p.cfg.RetryCount == 0 || i <= p.cfg.RetryCount; i++ {
@@ -74,9 +76,19 @@ func Retries[X pluginInput, Y pluginOutput](f func(x X) (Y, error), x X, p *pipe
 	return
 }
 
-// RetriesNoOutput applies the same logic as Retries, but for functions that return no output.
-func RetriesNoOutput[X pluginInput](f func(x X) error, a X, p *pipelineImpl, msg string) (time.Duration, error) {
-	_, d, err := Retries(func(x X) (empty, error) {
+// TODO: probly the following function and its unit test should be axed
+// retriesNoInput applies the same logic as Retries, but for functions that take no input.
+
+//nolint:unused
+func retriesNoInput[Y pluginOutput](f func() (Y, error), p *pipelineImpl, msg string) (Y, time.Duration, error) {
+	return retries(func(x empty) (Y, error) {
+		return f()
+	}, empty{}, p, msg)
+}
+
+// retriesNoOutput applies the same logic as Retries, but for functions that return no output.
+func retriesNoOutput[X pluginInput](f func(x X) error, a X, p *pipelineImpl, msg string) (time.Duration, error) {
+	_, d, err := retries(func(x X) (empty, error) {
 		return empty{}, f(x)
 	}, a, p, msg)
 	return d, err

conduit/pipeline/common_test.go

@@ -41,6 +41,18 @@ func TestRetries(t *testing.T) {
 		}
 	}
 
+	succeedAfterFactoryNoInput := func(succeedAfter uint64, never bool) func() (uint64, error) {
+		tries := uint64(0)
+
+		return func() (uint64, error) {
+			if tries >= succeedAfter && !never {
+				return tries + 1, nil
+			}
+			tries++
+			return 0, fmt.Errorf("%w: tries=%d", errSentinelCause, tries-1)
+		}
+	}
+
 	cases := []struct {
 		name         string
 		retryCount   uint64
@@ -106,8 +118,8 @@ func TestRetries(t *testing.T) {
 	for _, tc := range cases {
 		tc := tc
 
-		// run cases for Retries()
-		t.Run("Retries() "+tc.name, func(t *testing.T) {
+		// run cases for retries()
+		t.Run("retries() "+tc.name, func(t *testing.T) {
 			t.Parallel()
 			ctx, ccf := context.WithCancelCause(context.Background())
 			p := &pipelineImpl{
@@ -127,7 +139,7 @@ func TestRetries(t *testing.T) {
 				yChan := make(chan uint64)
 				errChan := make(chan error)
 				go func() {
-					y, _, err := Retries(succeedAfter, 0, p, "test")
+					y, _, err := retries(succeedAfter, 0, p, "test")
 					yChan <- y
 					errChan <- err
 				}()
@@ -144,7 +156,7 @@ func TestRetries(t *testing.T) {
 				return
 			}
 
-			y, _, err := Retries(succeedAfter, 0, p, "test")
+			y, _, err := retries(succeedAfter, 0, p, "test")
 			if tc.retryCount == 0 { // WLOG tc.neverSucceed == false
 				require.NoError(t, err, tc.name)
 
@@ -163,8 +175,8 @@ func TestRetries(t *testing.T) {
 			}
 		})
 
-		// run cases for RetriesNoOutput()
-		t.Run("RetriesNoOutput() "+tc.name, func(t *testing.T) {
+		// run cases for retriesNoOutput()
+		t.Run("retriesNoOutput() "+tc.name, func(t *testing.T) {
 			t.Parallel()
 			ctx, ccf := context.WithCancelCause(context.Background())
 			p := &pipelineImpl{
@@ -183,7 +195,7 @@ func TestRetries(t *testing.T) {
 
 				errChan := make(chan error)
 				go func() {
-					_, err := RetriesNoOutput(succeedAfterNoOutput, 0, p, "test")
+					_, err := retriesNoOutput(succeedAfterNoOutput, 0, p, "test")
 					errChan <- err
 				}()
 				time.Sleep(5 * time.Millisecond)
@@ -197,7 +209,7 @@ func TestRetries(t *testing.T) {
 				return
 			}
 
-			_, err := RetriesNoOutput(succeedAfterNoOutput, 0, p, "test")
+			_, err := retriesNoOutput(succeedAfterNoOutput, 0, p, "test")
 			if tc.retryCount == 0 { // WLOG tc.neverSucceed == false
 				require.NoError(t, err, tc.name)
 			} else { // retryCount > 0 so doesn't retry forever
@@ -209,5 +221,62 @@ func TestRetries(t *testing.T) {
 				}
 			}
 		})
+
+		// run case for retriesNoInput()
+		t.Run("retriesNoInput() "+tc.name, func(t *testing.T) {
+			t.Parallel()
+			ctx, ccf := context.WithCancelCause(context.Background())
+			p := &pipelineImpl{
+				ctx:    ctx,
+				ccf:    ccf,
+				logger: log.New(),
+				cfg: &data.Config{
+					RetryCount: tc.retryCount,
+					RetryDelay: 1 * time.Millisecond,
+				},
+			}
+			succeedAfterNoInput := succeedAfterFactoryNoInput(tc.succeedAfter, tc.neverSucceed)
+
+			if tc.retryCount == 0 && tc.neverSucceed {
+				// avoid infinite loop by cancelling the context
+
+				errChan := make(chan error)
+				yChan := make(chan uint64)
+				go func() {
+					out, _, err := retriesNoInput(succeedAfterNoInput, p, "test")
+					yChan <- out
+					errChan <- err
+				}()
+				time.Sleep(5 * time.Millisecond)
+				errTestCancelled := errors.New("test cancelled")
+				go func() {
+					ccf(errTestCancelled)
+				}()
+				y := <-yChan
+				err := <-errChan
+				require.ErrorIs(t, err, errTestCancelled, tc.name)
+				require.ErrorIs(t, err, errSentinelCause, tc.name)
+				require.Zero(t, y, tc.name)
+				return
+			}
+
+			y, _, err := retriesNoInput(succeedAfterNoInput, p, "test")
+			if tc.retryCount == 0 { // WLOG tc.neverSucceed == false
+				require.NoError(t, err, tc.name)
+
+				// note we subtract 1 from y below because succeedAfter has added 1 to its output
+				require.Equal(t, tc.succeedAfter, y-1, tc.name)
+			} else { // retryCount > 0 so doesn't retry forever
+				if tc.neverSucceed || tc.succeedAfter > tc.retryCount {
+					require.ErrorContains(t, err, fmt.Sprintf("%d", tc.retryCount), tc.name)
+					require.ErrorIs(t, err, errSentinelCause, tc.name)
+					require.Zero(t, y, tc.name)
+				} else { // !tc.neverSucceed && succeedAfter <= retryCount
+					require.NoError(t, err, tc.name)
+					require.Equal(t, tc.succeedAfter, y-1, tc.name)
+				}
+			}
+		})
+
 	}
 }

conduit/pipeline/pipeline.go

@@ -131,8 +131,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)
@@ -145,7 +145,7 @@
 		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)
 		}
 	}
 
@@ -196,8 +196,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
@@ -227,16 +227,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
 }
@@ -255,12 +255,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)
 		}
 	}
 
@@ -309,7 +309,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()
@@ -327,16 +327,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)
 		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
@@ -349,8 +349,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)
 	}
@@ -360,11 +360,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)
 		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)
 	}
@@ -373,12 +373,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)
 		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)
 	}
 
@@ -417,19 +417,19 @@
 
 	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)
 	}
 
 	for _, processor := range p.processors {
 		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)
 		}
 	}
 
 	if err := p.exporter.Close(); err != nil {
 		p.logger.Errorf("Pipeline.Stop(): Exporter (%s) error on close: %v", p.exporter.Metadata().Name, err)
 	}
 }
 
 func numInnerTxn(txn sdk.SignedTxnWithAD) int {
@@ -472,16 +472,16 @@
 		for {
 			startRound := time.Now()
 			select {
 			case <-p.ctx.Done():
 				p.logger.Infof("importer handler exiting. lastRnd=%d", lastRnd)
 				return
 			case rnd := <-roundChan:
 				lastRnd = rnd
 				waitTime := time.Since(startRound)
 				totalSelectWait += waitTime
 				p.logger.Tracef("importer handler waited %dms to receive round %d", waitTime.Milliseconds(), rnd)
 
-				blkData, importTime, lastError := Retries(importer.GetBlock, rnd, p, importer.Metadata().Name)
+				blkData, importTime, lastError := retries(importer.GetBlock, rnd, p, importer.Metadata().Name)
 				if lastError != nil {
 					p.cancelWithProblem(fmt.Errorf("importer %s handler (%w): failed to import round %d after %dms: %w", importer.Metadata().Name, errImporterCause, rnd, importTime.Milliseconds(), lastError))
 					return
@@ -533,7 +533,7 @@
 
 				var procTime time.Duration
 				var lastError error
-				blk, procTime, lastError = Retries(proc.Process, blk, p, proc.Metadata().Name)
+				blk, procTime, lastError = retries(proc.Process, blk, p, proc.Metadata().Name)
 				if lastError != nil {
 					p.cancelWithProblem(fmt.Errorf("processor[%d] %s handler (%w): failed to process round %d after %dms: %w", idx, proc.Metadata().Name, errProcessorCause, lastRnd, procTime.Milliseconds(), lastError))
 					return
@@ -543,8 +543,8 @@
 				selectStart := time.Now()
 				// check context in case out channel is full.
 				select {
 				case <-p.ctx.Done():
 					return
 				case blkOutChan <- blk:
 					waitTime := time.Since(selectStart)
 					totalFeedWait += waitTime
@@ -593,12 +593,12 @@
 				lastRound = blk.Round()
 
 				if p.pipelineMetadata.NextRound != lastRound {
 					lastError = fmt.Errorf("aborting after out of order block data. %d != %d", p.pipelineMetadata.NextRound, lastRound)
 					return
 				}
 
 				var exportTime time.Duration
-				exportTime, lastError = RetriesNoOutput(exporter.Receive, blk, p, eName)
+				exportTime, lastError = retriesNoOutput(exporter.Receive, blk, p, eName)
 				if lastError != nil {
 					lastError = fmt.Errorf("aborting after failing to export round %d: %w", lastRound, lastError)
 					return
@@ -640,16 +640,15 @@
 				// WARNING: removing/re-log-levelling the following will BREAK:
 				// - the E2E test (Search for "Pipeline round" in subslurp.py)
 				// - the internal tools logstats collector (See func ConduitCollector in logstats.go of internal-tools repo)
-				p.logger.Infof(logstatsE2Elog(nextRound, lastRound, len(blk.Payset), exportTime))
+				p.logger.Infof(logstatsE2Elog(lastRound, len(blk.Payset), exportTime))
 			}
 		}
 	}()
 }
 
-func logstatsE2Elog(nextRound, lastRound uint64, topLevelTxnCount int, exportTime time.Duration) string {
+func logstatsE2Elog(lastRound uint64, topLevelTxnCount int, exportTime time.Duration) string {
 	return fmt.Sprintf(
-		"UPDATED Pipeline NextRound=%d. FINISHED Pipeline round r=%d (%d txn) exported in %s",
-		nextRound,
+		"FINISHED Pipeline round r=%d (%d txn) exported in %s",
 		lastRound,
 		topLevelTxnCount,
 		exportTime,
@@ -696,8 +695,6 @@
 			}
 		}
 	}(p.pipelineMetadata.NextRound)
-
-	<-p.ctx.Done()
 }
 
 func (p *pipelineImpl) Wait() {
@@ -761,12 +758,12 @@
 	for _, processorConfig := range cfg.Processors {
 		processorName := processorConfig.Name
 
 		processorConstructor, err := processors.ProcessorConstructorByName(processorName)
 		if err != nil {
 			return nil, fmt.Errorf("MakePipeline(): could not build processor '%s': %w", processorName, err)
 		}
 
 		pipeline.processors = append(pipeline.processors, processorConstructor.New())
 		logger.Infof("Found Processor: %s", processorName)
 	}
 
@@ -791,22 +788,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

@@ -990,13 +990,12 @@ func TestMetrics(t *testing.T) {
 }
 
 func TestLogStatsE2Elog(t *testing.T) {
-	nextRound := uint64(1337)
 	round := uint64(42)
 	numTxns := 13
 	duration := 12345600 * time.Microsecond
 
-	expectedLog := "UPDATED Pipeline NextRound=1337. FINISHED Pipeline round r=42 (13 txn) exported in 12.3456s"
-	log := logstatsE2Elog(nextRound, round, numTxns, duration)
+	expectedLog := "FINISHED Pipeline round r=42 (13 txn) exported in 12.3456s"
+	log := logstatsE2Elog(round, numTxns, duration)
 	require.Equal(t, expectedLog, log)
 
 	logstatsRex, err := regexp.Compile(`round r=(\d+) \((\d+) txn\) exported in (.*)`)

conduit/plugins/importers/all/all.go

@@ -4,4 +4,5 @@ import (
 	// Call package wide init function
 	_ "github.com/algorand/conduit/conduit/plugins/importers/algod"
 	_ "github.com/algorand/conduit/conduit/plugins/importers/filereader"
+	_ "github.com/algorand/conduit/conduit/plugins/importers/noop"
 )

conduit/plugins/importers/noop/noop_importer.go

@@ -0,0 +1,81 @@
+package noop
+
+import (
+	"context"
+	_ "embed" // used to embed config
+	"fmt"
+	"time"
+
+	"github.com/sirupsen/logrus"
+
+	sdk "github.com/algorand/go-algorand-sdk/v2/types"
+
+	"github.com/algorand/conduit/conduit/data"
+	"github.com/algorand/conduit/conduit/plugins"
+	"github.com/algorand/conduit/conduit/plugins/importers"
+)
+
+// PluginName to use when configuring.
+var PluginName = "noop"
+
+const sleepForGetBlock = 100 * time.Millisecond
+
+// `noopImporter`s will function without ever erroring. This means they will also process out of order blocks
+// which may or may not be desirable for different use cases--it can hide errors in actual importers expecting in order
+// block processing.
+// The `noopImporter` will maintain `Round` state according to the round of the last block it processed.
+// It also sleeps 100 milliseconds between blocks to slow down the pipeline.
+type noopImporter struct {
+	round uint64
+	cfg   ImporterConfig
+}
+
+//go:embed sample.yaml
+var sampleConfig string
+
+var metadata = plugins.Metadata{
+	Name:         PluginName,
+	Description:  "noop importer",
+	Deprecated:   false,
+	SampleConfig: sampleConfig,
+}
+
+func (imp *noopImporter) Metadata() plugins.Metadata {
+	return metadata
+}
+
+func (imp *noopImporter) Init(_ context.Context, _ data.InitProvider, cfg plugins.PluginConfig, _ *logrus.Logger) error {
+	if err := cfg.UnmarshalConfig(&imp.cfg); err != nil {
+		return fmt.Errorf("init failure in unmarshalConfig: %v", err)
+	}
+	imp.round = imp.cfg.Round
+	return nil
+}
+
+func (imp *noopImporter) Close() error {
+	return nil
+}
+
+func (imp *noopImporter) GetGenesis() (*sdk.Genesis, error) {
+	return &sdk.Genesis{}, nil
+}
+
+func (imp *noopImporter) GetBlock(rnd uint64) (data.BlockData, error) {
+	time.Sleep(sleepForGetBlock)
+	imp.round = rnd
+	return data.BlockData{
+		BlockHeader: sdk.BlockHeader{
+			Round: sdk.Round(rnd),
+		},
+	}, nil
+}
+
+func (imp *noopImporter) Round() uint64 {
+	return imp.round
+}
+
+func init() {
+	importers.Register(PluginName, importers.ImporterConstructorFunc(func() importers.Importer {
+		return &noopImporter{}
+	}))
+}

conduit/plugins/importers/noop/noop_importer_config.go

@@ -0,0 +1,7 @@
+package noop
+
+// ImporterConfig specific to the noop importer
+type ImporterConfig struct {
+	// Optionally specify the round to start on
+	Round uint64 `yaml:"round"`
+}

conduit/plugins/importers/noop/sample.yaml

@@ -0,0 +1,3 @@
+name: noop
+# noop has no config
+config:

e2e_tests/src/e2e_conduit/subslurp.py

@@ -7,7 +7,7 @@
 logger = logging.getLogger(__name__)
 
 # Matches conduit log output:
-# "UPDATED Pipeline NextRound=1337. FINISHED Pipeline round r=42 (13 txn) exported in 12.3456s"
+# "FINISHED Pipeline round r=42 (13 txn) exported in 12.3456s"
 FINISH_ROUND: re.Pattern = re.compile(b"FINISHED Pipeline round r=(\d+)")
 
 

pkg/cli/cli.go

@@ -48,8 +48,8 @@
 
 	pCfg, err := data.MakePipelineConfig(args)
 	if err != nil {
 		return err
 	}
 
 	// Initialize logger
 	level, err := log.ParseLevel(pCfg.LogLevel)
@@ -63,8 +63,8 @@
 
 	logger, err = loggers.MakeThreadSafeLogger(level, pCfg.LogFile)
 	if err != nil {
 		return fmt.Errorf("failed to create logger: %w", err)
 	}
 
 	logger.Infof("Starting Conduit %s", version.LongVersion())
 	logger.Infof("Using data directory: %s", args.ConduitDataDir)
@@ -94,26 +94,29 @@
 
 	err = pline.Init()
 	if err != nil {
 		// Suppress log, it is about to be printed to stderr.
 		if pCfg.LogFile != "" {
 			logger.Error(err)
 		}
 		return fmt.Errorf("pipeline init error: %w", err)
 	}
 	pline.Start()
 	defer pline.Stop()
 
 	// Start server
 	if pCfg.API.Address != "" {
+		logger.Infof("starting API server on %s", pCfg.API.Address)
 		shutdown, err := api.StartServer(logger, pline, pCfg.API.Address)
 		if err != nil {
 			// Suppress log, it is about to be printed to stderr.
 			if pCfg.LogFile != "" {
 				logger.Error(err)
 			}
 			return fmt.Errorf("failed to start API server: %w", err)
 		}
 		defer shutdown(context.Background())
+	} else {
+		logger.Info("API server is disabled")
 	}
 
 	pline.Wait()
@@ -121,11 +124,11 @@
 }
 
 // MakeConduitCmdWithUtilities creates the main cobra command with all utilities
 func MakeConduitCmdWithUtilities() *cobra.Command {
 	cmd := MakeConduitCmd()
 	cmd.AddCommand(initialize.InitCommand)
 	cmd.AddCommand(list.Command)
 	return cmd
 }
 
 // MakeConduitCmd creates the main cobra command, initializes flags
@@ -147,17 +150,17 @@
 Detailed documentation is online: https://github.com/algorand/conduit`,
 		Args: cobra.NoArgs,
 		Run: func(cmd *cobra.Command, args []string) {
 			err := runConduitCmdWithConfig(cfg)
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "\nExiting with error:\t%s.\n", err)
 				os.Exit(1)
 			}
 		},
 		PersistentPreRun: func(cmd *cobra.Command, args []string) {
 			if vFlag {
 				fmt.Printf("%s\n", version.LongVersion())
 				os.Exit(0)
 			}
 		},
 	}
 	cmd.Flags().StringVarP(&cfg.ConduitDataDir, "data-dir", "d", "", "Set the Conduit data directory. If not set the CONDUIT_DATA_DIR environment variable is used.")