api: New API with health endpoint

api/api.go

@@ -0,0 +1,57 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/algorand/conduit/conduit/pipeline"
+)
+
+// StatusProvider is a subset of the Pipeline interface required by the health handler.
+type StatusProvider interface {
+	Status() (pipeline.Status, error)
+}
+
+func makeHealthHandler(p StatusProvider) func(w http.ResponseWriter, r *http.Request) {
+	return func(w http.ResponseWriter, r *http.Request) {
+		status, err := p.Status()
+		if err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			fmt.Fprintf(w, `{"error": "%s"}`, err)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		data, _ := json.Marshal(status)
+		fmt.Fprint(w, string(data))
+	}
+}
+
+// StartServer starts an http server that exposes a health check endpoint.
+// A callback is returned that can be used to gracefully shutdown the server.
+func StartServer(logger *log.Logger, p StatusProvider, address string) (func(ctx context.Context), error) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/health", makeHealthHandler(p))
+
+	srv := &http.Server{
+		Addr:    address,
+		Handler: mux,
+	}
+
+	go func() {
+		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Fatalf("failed to start API server: %s", err)
+		}
+	}()
+
+	shutdownCallback := func(ctx context.Context) {
+		if err := srv.Shutdown(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			logger.Fatalf("failed to shutdown API server: %s", err)
+		}
+	}
+	return shutdownCallback, nil
+}

api/api_test.go

@@ -0,0 +1,103 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/require"
+
+	"github.com/algorand/conduit/conduit/pipeline"
+)
+
+type mockStatusProvider struct {
+	s   pipeline.Status
+	err error
+}
+
+func (m mockStatusProvider) Status() (pipeline.Status, error) {
+	if m.err != nil {
+		return pipeline.Status{}, m.err
+	}
+	return m.s, nil
+}
+
+func TestStartServer_BadAddress(t *testing.T) {
+	l, h := test.NewNullLogger()
+	// capture the fatal log if any.
+	l.ExitFunc = func(int) {}
+	sp := mockStatusProvider{}
+
+	shutdown, err := StartServer(l, sp, "bad address")
+	defer shutdown(context.Background())
+	require.NoError(t, err)
+	time.Sleep(1 * time.Millisecond)
+
+	require.Len(t, h.Entries, 1)
+	require.Equal(t, h.LastEntry().Level, logrus.FatalLevel)
+}
+
+func TestStartServer_GracefulShutdown(t *testing.T) {
+	l, h := test.NewNullLogger()
+	// capture the fatal log if any.
+	l.ExitFunc = func(int) {}
+	sp := mockStatusProvider{}
+	shutdown, err := StartServer(l, sp, "bad address")
+	defer shutdown(context.Background())
+	require.NoError(t, err)
+	require.Len(t, h.Entries, 0)
+}
+
+func TestStartServer_HealthCheck(t *testing.T) {
+	l, _ := test.NewNullLogger()
+	// capture the fatal log if any.
+	l.ExitFunc = func(int) {}
+	sp := mockStatusProvider{
+		s: pipeline.Status{
+			Round: 999,
+		},
+	}
+
+	// Find an open port...
+	listener, err := net.Listen("tcp", ":0")
+	addr := listener.Addr().String()
+	listener.Close()
+	require.NoError(t, err)
+
+	// Start server.
+	shutdown, err := StartServer(l, sp, addr)
+	defer shutdown(context.Background())
+	require.NoError(t, err)
+
+	// Make request.
+	resp, err := http.Get("http://" + addr + "/health")
+	require.NoError(t, err)
+
+	// Make sure we got the right response.
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+	var respStatus pipeline.Status
+	json.NewDecoder(resp.Body).Decode(&respStatus)
+	require.NoError(t, err)
+	require.Equal(t, sp.s, respStatus)
+}
+
+func TestHealthHandlerError(t *testing.T) {
+	sp := mockStatusProvider{
+		err: fmt.Errorf("some error"),
+	}
+	handler := makeHealthHandler(sp)
+	rec := httptest.NewRecorder()
+	handler(rec, nil)
+
+	// validate response
+	resp := rec.Result()
+	require.Equal(t, http.StatusInternalServerError, resp.StatusCode)
+	require.Contains(t, rec.Body.String(), "some error")
+}

conduit/data/config.go

@@ -56,6 +56,11 @@
 	Password string `yaml:"password"`
 }
 
+// API defines parameters for the Conduit API server.
+type API struct {
+	Address string `yaml:"addr"`
+}
+
 // Config stores configuration specific to the conduit pipeline
 type Config struct {
 	// ConduitArgs are the program inputs. Should not be serialized for config.
@@ -72,6 +77,7 @@
 	Processors []NameConfigPair `yaml:"processors"`
 	Exporter   NameConfigPair   `yaml:"exporter"`
 	Metrics    Metrics          `yaml:"metrics"`
+	API        API              `yaml:"api"`
 	// RetryCount is the number of retries to perform for an error in the pipeline
 	RetryCount uint64 `yaml:"retry-count"`
 	// RetryDelay is a duration amount interpreted from a string
@@ -88,8 +94,8 @@
 
 	// If it is a negative time, it is an error
 	if cfg.RetryDelay < 0 {
 		return fmt.Errorf("Args.Valid(): invalid retry delay - time duration was negative (%s)", cfg.RetryDelay.String())
 	}
 
 	return nil
 }
@@ -101,8 +107,8 @@
 	}
 
 	if !isDir(args.ConduitDataDir) {
 		return nil, fmt.Errorf("MakePipelineConfig(): invalid data dir '%s'", args.ConduitDataDir)
 	}
 
 	// Search for pipeline configuration in data directory
 	autoloadParamConfigPath, err := getConfigFromDataDir(args.ConduitDataDir, DefaultConfigBaseName, []string{"yml", "yaml"})
@@ -112,8 +118,8 @@
 
 	file, err := os.Open(autoloadParamConfigPath)
 	if err != nil {
 		return nil, fmt.Errorf("MakePipelineConfig(): reading config error: %w", err)
 	}
 
 	pCfgDecoder := yaml.NewDecoder(file)
 	// Make sure we are strict about only unmarshalling known fields
@@ -137,8 +143,8 @@
 	}
 
 	if err := pCfg.Valid(); err != nil {
 		return nil, fmt.Errorf("MakePipelineConfig(): config file (%s) had mal-formed schema: %w", autoloadParamConfigPath, err)
 	}
 
 	return &pCfg, nil
 }
@@ -174,9 +180,9 @@
 	}
 
 	if count > 1 {
 		return "", fmt.Errorf("config filename (%s) in data directory (%s) matched more than one filetype: %v",
 			configFilename, dataDirectory, configFileTypes)
 	}
 
 	// if count == 0 then the fullpath will be set to "" and error will be nil
 	// if count == 1 then it fullpath will be correct

conduit/pipeline/pipeline.go

@@ -38,6 +38,12 @@
 	Stop()
 	Error() error
 	Wait()
+	Status() (Status, error)
+}
+
+// Status is a struct that contains the current pipeline status.
+type Status struct {
+	Round uint64 `json:"round"`
 }
 
 type pipelineImpl struct {
@@ -58,6 +64,9 @@
 	completeCallback []conduit.OnCompleteFunc
 
 	pipelineMetadata state
+
+	statusMu sync.Mutex
+	status   Status
 }
 
 type pluginChannel chan data.BlockData
@@ -122,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)
@@ -136,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)
 		}
 	}
 
@@ -187,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
@@ -218,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
 }
@@ -246,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)
 		}
 	}
 
@@ -300,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()
@@ -318,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
@@ -340,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)
 	}
@@ -351,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)
 	}
@@ -364,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)
 	}
 
@@ -382,11 +391,15 @@
 		go p.startMetricsServer()
 	}
 
+	p.statusMu.Lock()
+	defer p.statusMu.Unlock()
+	p.status.Round = p.pipelineMetadata.NextRound
+
 	return err
 }
 
 func (p *pipelineImpl) Stop() {
 	p.ccf(nil)
 	p.wg.Wait()
 
 	if p.profFile != nil {
@@ -402,21 +415,21 @@
 		}
 	}
 
 	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 {
@@ -459,8 +472,8 @@
 		for {
 			startRound := time.Now()
 			select {
 			case <-p.ctx.Done():
 				p.logger.Infof("importer handler exiting. lastRnd=%d", lastRnd)
 				return
 			case rnd := <-roundChan:
 				lastRnd = rnd
@@ -580,9 +593,9 @@
 				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)
@@ -602,11 +615,14 @@
 				// Increment Round, update metadata
 				nextRound := lastRound + 1
 				p.pipelineMetadata.NextRound = nextRound
+				p.statusMu.Lock()
+				p.status.Round = nextRound
+				p.statusMu.Unlock()
 				lastError = p.pipelineMetadata.encodeToFile(p.cfg.ConduitArgs.ConduitDataDir)
 				if lastError != nil {
 					lastError = fmt.Errorf("aborting after updating NextRound=%d BUT failing to save metadata: %w", nextRound, lastError)
 					return
 				}
 				p.logger.Tracef("exporter %s incremented pipeline metadata NextRound to %d", eName, nextRound)
 
 				for i, cb := range p.completeCallback {
@@ -688,6 +704,13 @@
 	p.wg.Wait()
 }
 
+func (p *pipelineImpl) Status() (Status, error) {
+	p.statusMu.Lock()
+	ret := p.status
+	p.statusMu.Unlock()
+	return ret, nil
+}
+
 // start a http server serving /metrics
 func (p *pipelineImpl) startMetricsServer() {
 	http.Handle("/metrics", promhttp.Handler())
@@ -730,7 +753,7 @@
 		return nil, fmt.Errorf("MakePipeline(): could not build importer '%s': %w", importerName, err)
 	}
 
 	pipeline.importer = importerConstructor.New()
 	logger.Infof("Found Importer: %s", importerName)
 
 	// ---
@@ -738,12 +761,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)
 	}
 
@@ -751,12 +774,12 @@
 
 	exporterName := cfg.Exporter.Name
 
 	exporterConstructor, err := exporters.ExporterConstructorByName(exporterName)
 	if err != nil {
 		return nil, fmt.Errorf("MakePipeline(): could not build exporter '%s': %w", exporterName, err)
 	}
 
 	pipeline.exporter = exporterConstructor.New()
 	logger.Infof("Found Exporter: %s", exporterName)
 
 	return pipeline, nil
@@ -768,22 +791,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
 }

pkg/cli/cli.go

@@ -10,6 +10,7 @@
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
+	"github.com/algorand/conduit/api"
 	"github.com/algorand/conduit/conduit/data"
 	"github.com/algorand/conduit/conduit/loggers"
 	"github.com/algorand/conduit/conduit/pipeline"
@@ -47,8 +48,8 @@
 
 	pCfg, err := data.MakePipelineConfig(args)
 	if err != nil {
 		return err
 	}
 
 	// Initialize logger
 	level, err := log.ParseLevel(pCfg.LogLevel)
@@ -62,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)
@@ -91,26 +92,40 @@
 		return err
 	}
 
 	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 != "" {
+		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())
+	}
+
 	pline.Wait()
 	return pline.Error()
 }
 
 // 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
@@ -132,17 +147,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.")

pkg/cli/internal/initialize/conduit.yml.example

@@ -17,6 +17,10 @@ retry-delay: "1s"
 # Whether or not to print the conduit banner on startup.
 hide-banner: false
 
+# When the address is not empty information is available on '/health'
+api:
+    addr: ":8981"
+
 # When enabled prometheus metrics are available on '/metrics'
 metrics:
     mode: OFF
@@ -33,14 +37,3 @@ processors:
 # An exporter is defined to do something with the data.
 exporter:
 %s
-
-# Enable telemetry for conduit
-telemetry:
-  enabled: false
-
-  # By default the following fields will be configured to send data to Algorand.
-  # To store your own telemetry events, they can be overridden.
-  # uri: ""
-  # index: ""
-  # username: ""
-  # password: ""