usability: Better error messages and documentation. (#20)

cmd/conduit/internal/list/list.go

@@ -16,8 +16,16 @@ import (
 // Command is the list command to embed in a root cobra command.
 var Command = &cobra.Command{
 	Use:   "list",
-	Short: "lists all plugins available to conduit",
-	Args:  cobra.NoArgs,
+	Short: "List all available Conduit plugins",
+	Long: `List all available Conduit plugins by type and a short description.
+
+Use this utility to explore the plugins. Drill into each plugin to get a
+sample configuration.
+
+Example:
+  conduit list importers algod
+  conduit list processors filter_processor`,
+	Args: cobra.NoArgs,
 	RunE: func(cmd *cobra.Command, args []string) error {
 		printAll()
 		return nil
@@ -26,11 +34,12 @@ var Command = &cobra.Command{
 	SilenceErrors: true,
 }
 
-func makeDetailsCommand(use string, data func() []conduit.Metadata) *cobra.Command {
+func makeDetailsCommand(pluginType string, data func() []conduit.Metadata) *cobra.Command {
 	return &cobra.Command{
-		Use:     use + "s",
-		Aliases: []string{use},
-		Short:   fmt.Sprintf("usage detail for %s plugins", use),
+		Use:     pluginType + "s",
+		Aliases: []string{pluginType},
+		Short:   fmt.Sprintf("Usage details for %s plugins.", pluginType),
+		Long:    fmt.Sprintf(`Usage details for %s plugins. Pass in a specific plugin as a positional argument for a sample configuration file.`, pluginType),
 		Args:    cobra.MaximumNArgs(1),
 		Run: func(cmd *cobra.Command, args []string) {
 			if len(args) == 0 {

cmd/conduit/main.go

@@ -5,6 +5,7 @@ import (
 	_ "embed"
 	"fmt"
 	"os"
+	"strings"
 
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -29,6 +30,10 @@ var (
 	banner string
 )
 
+const (
+	conduitEnvVar = "CONDUIT_DATA_DIR"
+)
+
 // init() function for main package
 func init() {
 	conduitCmd.AddCommand(initialize.InitCommand)
@@ -40,7 +45,11 @@ func runConduitCmdWithConfig(args *conduit.Args) error {
 	defer pipeline.HandlePanic(logger)
 
 	if args.ConduitDataDir == "" {
-		args.ConduitDataDir = os.Getenv("CONDUIT_DATA_DIR")
+		args.ConduitDataDir = os.Getenv(conduitEnvVar)
+	}
+
+	if args.ConduitDataDir == "" {
+		return fmt.Errorf("the data directory is required and must be provided with a command line option or the '%s' environment variable", conduitEnvVar)
 	}
 
 	pCfg, err := pipeline.MakePipelineConfig(args)
@@ -49,14 +58,18 @@ func runConduitCmdWithConfig(args *conduit.Args) error {
 	}
 
 	// Initialize logger
-	level, err := log.ParseLevel(pCfg.PipelineLogLevel)
+	level, err := log.ParseLevel(pCfg.LogLevel)
 	if err != nil {
-		return fmt.Errorf("runConduitCmdWithConfig(): invalid log level: %s", err)
+		var levels []string
+		for _, l := range log.AllLevels {
+			levels = append(levels, l.String())
+		}
+		return fmt.Errorf("invalid configuration: '%s' is not a valid log level, valid levels: %s", pCfg.LogLevel, strings.Join(levels, ", "))
 	}
 
 	logger, err = loggers.MakeThreadSafeLogger(level, pCfg.LogFile)
 	if err != nil {
-		return fmt.Errorf("runConduitCmdWithConfig(): failed to create logger: %w", err)
+		return fmt.Errorf("failed to create logger: %w", err)
 	}
 
 	logger.Infof("Using data directory: %s", args.ConduitDataDir)
@@ -77,8 +90,7 @@ func runConduitCmdWithConfig(args *conduit.Args) error {
 	if err != nil {
 		err = fmt.Errorf("pipeline creation error: %w", err)
 
-		// Make sure the error is written to stdout once.
-		fmt.Println(err)
+		// Suppress log, it is about to be printed to stderr.
 		if pCfg.LogFile != "" {
 			logger.Error(err)
 		}
@@ -87,6 +99,10 @@ func runConduitCmdWithConfig(args *conduit.Args) error {
 
 	err = pipeline.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)
 	}
 	pipeline.Start()
@@ -101,11 +117,24 @@ func makeConduitCmd() *cobra.Command {
 	var vFlag bool
 	cmd := &cobra.Command{
 		Use:   "conduit",
-		Short: "run the conduit framework",
-		Long:  "run the conduit framework",
-		Args:  cobra.NoArgs,
-		RunE: func(cmd *cobra.Command, args []string) error {
-			return runConduitCmdWithConfig(cfg)
+		Short: "Run the Conduit framework.",
+		Long: `Conduit is a framework for ingesting blocks from the Algorand blockchain
+into external applications. It is designed as a modular plugin system that
+allows users to configure their own data pipelines.
+
+You must provide a data directory containing a file named conduit.yml. The
+file configures pipeline and all enabled plugins.
+
+See other subcommands for further built in utilities and information.
+
+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:\n%s.\n", err)
+				os.Exit(1)
+			}
 		},
 		PersistentPreRun: func(cmd *cobra.Command, args []string) {
 			if vFlag {
@@ -114,13 +143,12 @@ func makeConduitCmd() *cobra.Command {
 				os.Exit(0)
 			}
 		},
-		SilenceUsage: true,
-		// Silence errors because our logger will catch and print any errors
-		SilenceErrors: true,
 	}
-	cmd.Flags().StringVarP(&cfg.ConduitDataDir, "data-dir", "d", "", "set the data directory for the conduit binary")
-	cmd.Flags().Uint64VarP(&cfg.NextRoundOverride, "next-round-override", "r", 0, "set the starting round. Overrides next-round in metadata.json")
-	cmd.Flags().BoolVarP(&vFlag, "version", "v", false, "print the conduit version")
+	cmd.Flags().StringVarP(&cfg.ConduitDataDir, "data-dir", "d", "", "Set the Conduit data directory. If not set the CONDUIT_DATA_DIR environment variable is used.")
+	cmd.Flags().Uint64VarP(&cfg.NextRoundOverride, "next-round-override", "r", 0, "Set the starting round. Overrides next-round in metadata.json. Some exporters do not support overriding the starting round.")
+	cmd.Flags().BoolVarP(&vFlag, "version", "v", false, "Print the Conduit version.")
+	// No need for shell completions.
+	cmd.CompletionOptions.DisableDefaultCmd = true
 
 	return cmd
 }

cmd/conduit/main_test.go

@@ -14,6 +14,18 @@ import (
 	"github.com/algorand/conduit/conduit/pipeline"
 )
 
+// Fills in a temp data dir and creates files
+// TODO: Refactor the code so that testing can be done without creating files and directories.
+func setupDataDir(t *testing.T, cfg pipeline.Config) *conduit.Args {
+	conduitArgs := &conduit.Args{ConduitDataDir: t.TempDir()}
+	data, err := yaml.Marshal(&cfg)
+	require.NoError(t, err)
+	configFile := path.Join(conduitArgs.ConduitDataDir, conduit.DefaultConfigName)
+	os.WriteFile(configFile, data, 0755)
+	require.FileExists(t, configFile)
+	return conduitArgs
+}
+
 func TestBanner(t *testing.T) {
 	test := func(t *testing.T, hideBanner bool) {
 		// Capture stdout.
@@ -34,14 +46,10 @@ func TestBanner(t *testing.T) {
 			Processors:  nil,
 			Exporter:    pipeline.NameConfigPair{Name: "test", Config: map[string]interface{}{"a": "a"}},
 		}
-		data, err := yaml.Marshal(&cfg)
-		require.NoError(t, err)
-		configFile := path.Join(cfg.ConduitArgs.ConduitDataDir, conduit.DefaultConfigName)
-		os.WriteFile(configFile, data, 0755)
-		require.FileExists(t, configFile)
+		args := setupDataDir(t, cfg)
 
-		err = runConduitCmdWithConfig(cfg.ConduitArgs)
-		data, err = os.ReadFile(stdoutFilePath)
+		err = runConduitCmdWithConfig(args)
+		data, err := os.ReadFile(stdoutFilePath)
 		require.NoError(t, err)
 
 		if hideBanner {
@@ -60,6 +68,21 @@ func TestBanner(t *testing.T) {
 	})
 }
 
+func TestEmptyDataDir(t *testing.T) {
+	args := conduit.Args{}
+	err := runConduitCmdWithConfig(&args)
+	require.ErrorContains(t, err, conduitEnvVar)
+}
+
+func TestInvalidLogLevel(t *testing.T) {
+	cfg := pipeline.Config{
+		LogLevel: "invalid",
+	}
+	args := setupDataDir(t, cfg)
+	err := runConduitCmdWithConfig(args)
+	require.ErrorContains(t, err, "not a valid log level")
+}
+
 func TestLogFile(t *testing.T) {
 	// returns stdout
 	test := func(t *testing.T, logfile string) ([]byte, error) {
@@ -75,19 +98,15 @@ func TestLogFile(t *testing.T) {
 		os.Stdout = f
 
 		cfg := pipeline.Config{
-			LogFile:     logfile,
-			ConduitArgs: &conduit.Args{ConduitDataDir: t.TempDir()},
-			Importer:    pipeline.NameConfigPair{Name: "test", Config: map[string]interface{}{"a": "a"}},
-			Processors:  nil,
-			Exporter:    pipeline.NameConfigPair{Name: "test", Config: map[string]interface{}{"a": "a"}},
+			LogFile:    logfile,
+			Importer:   pipeline.NameConfigPair{Name: "test", Config: map[string]interface{}{"a": "a"}},
+			Processors: nil,
+			Exporter:   pipeline.NameConfigPair{Name: "test", Config: map[string]interface{}{"a": "a"}},
 		}
-		data, err := yaml.Marshal(&cfg)
-		require.NoError(t, err)
-		configFile := path.Join(cfg.ConduitArgs.ConduitDataDir, conduit.DefaultConfigName)
-		os.WriteFile(configFile, data, 0755)
-		require.FileExists(t, configFile)
+		args := setupDataDir(t, cfg)
 
-		err = runConduitCmdWithConfig(cfg.ConduitArgs)
+		err = runConduitCmdWithConfig(args)
+		require.ErrorContains(t, err, "pipeline creation error")
 		return os.ReadFile(stdoutFilePath)
 	}
 
@@ -98,7 +117,6 @@ func TestLogFile(t *testing.T) {
 		dataStr := string(data)
 		require.Contains(t, dataStr, "{")
 		require.Contains(t, dataStr, "\nWriting logs to console.")
-		require.Contains(t, dataStr, "\npipeline creation error")
 	})
 
 	// logging to file
@@ -112,9 +130,9 @@ func TestLogFile(t *testing.T) {
 		require.NoError(t, err)
 		logdataStr := string(logdata)
 		require.Contains(t, logdataStr, "{")
+		// pipeline error is not suppressed from log file.
+		require.Contains(t, logdataStr, "pipeline creation error")
 		// written to stdout and logfile
-		require.Contains(t, dataStr, "\npipeline creation error")
 		require.Contains(t, dataStr, "\nWriting logs to file:")
-		require.Contains(t, logdataStr, `"msg":"pipeline creation error`)
 	})
 }

conduit/config.go

@@ -19,7 +19,6 @@ var DefaultLogLevel = log.InfoLevel
 var DefaultMetricsPrefix = "conduit"
 
 // Args configuration for conduit running.
-// This is needed to support a CONDUIT_DATA_DIR environment variable.
 type Args struct {
 	ConduitDataDir    string `yaml:"data-dir"`
 	NextRoundOverride uint64 `yaml:"next-round-override"`

conduit/pipeline/pipeline.go

@@ -52,8 +52,8 @@ type Config struct {
 	PIDFilePath string `yaml:"pid-filepath"`
 	HideBanner  bool   `yaml:"hide-banner"`
 
-	LogFile          string `yaml:"log-file"`
-	PipelineLogLevel string `yaml:"log-level"`
+	LogFile  string `yaml:"log-file"`
+	LogLevel string `yaml:"log-level"`
 	// Store a local copy to access parent variables
 	Importer   NameConfigPair   `yaml:"importer"`
 	Processors []NameConfigPair `yaml:"processors"`
@@ -70,11 +70,6 @@ func (cfg *Config) Valid() error {
 	if cfg.ConduitArgs == nil {
 		return fmt.Errorf("Args.Valid(): conduit args were nil")
 	}
-	if cfg.PipelineLogLevel != "" {
-		if _, err := log.ParseLevel(cfg.PipelineLogLevel); err != nil {
-			return fmt.Errorf("Args.Valid(): pipeline log level (%s) was invalid: %w", cfg.PipelineLogLevel, err)
-		}
-	}
 
 	// If it is a negative time, it is an error
 	if cfg.RetryDelay < 0 {
@@ -122,8 +117,8 @@ func MakePipelineConfig(args *conduit.Args) (*Config, error) {
 	pCfg.ConduitArgs = args
 
 	// Default log level.
-	if pCfg.PipelineLogLevel == "" {
-		pCfg.PipelineLogLevel = conduit.DefaultLogLevel.String()
+	if pCfg.LogLevel == "" {
+		pCfg.LogLevel = conduit.DefaultLogLevel.String()
 	}
 
 	if err := pCfg.Valid(); err != nil {

conduit/pipeline/pipeline_test.go

@@ -37,23 +37,22 @@ func TestPipelineConfigValidity(t *testing.T) {
 		errContains string
 	}{
 		{"valid", Config{
-			ConduitArgs:      &conduit.Args{ConduitDataDir: ""},
-			PipelineLogLevel: "info",
-			Importer:         NameConfigPair{"test", map[string]interface{}{"a": "a"}},
-			Processors:       nil,
-			Exporter:         NameConfigPair{"test", map[string]interface{}{"a": "a"}},
+			ConduitArgs: &conduit.Args{ConduitDataDir: ""},
+			LogLevel:    "info",
+			Importer:    NameConfigPair{"test", map[string]interface{}{"a": "a"}},
+			Processors:  nil,
+			Exporter:    NameConfigPair{"test", map[string]interface{}{"a": "a"}},
 		}, ""},
 
 		{"valid 2", Config{
-			ConduitArgs:      &conduit.Args{ConduitDataDir: ""},
-			PipelineLogLevel: "info",
-			Importer:         NameConfigPair{"test", map[string]interface{}{"a": "a"}},
-			Processors:       []NameConfigPair{{"test", map[string]interface{}{"a": "a"}}},
-			Exporter:         NameConfigPair{"test", map[string]interface{}{"a": "a"}},
+			ConduitArgs: &conduit.Args{ConduitDataDir: ""},
+			LogLevel:    "info",
+			Importer:    NameConfigPair{"test", map[string]interface{}{"a": "a"}},
+			Processors:  []NameConfigPair{{"test", map[string]interface{}{"a": "a"}}},
+			Exporter:    NameConfigPair{"test", map[string]interface{}{"a": "a"}},
 		}, ""},
 
 		{"empty config", Config{ConduitArgs: nil}, "Args.Valid(): conduit args were nil"},
-		{"invalid log level", Config{ConduitArgs: &conduit.Args{ConduitDataDir: ""}, PipelineLogLevel: "asdf"}, "Args.Valid(): pipeline log level (asdf) was invalid:"},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -170,7 +169,7 @@ exporter:
 
 	pCfg, err := MakePipelineConfig(cfg)
 	assert.Nil(t, err)
-	assert.Equal(t, pCfg.PipelineLogLevel, "info")
+	assert.Equal(t, pCfg.LogLevel, "info")
 	assert.Equal(t, pCfg.Valid(), nil)
 	assert.Equal(t, pCfg.Importer.Name, "algod")
 	assert.Equal(t, pCfg.Importer.Config["token"], "e36c01fc77e490f23e61899c0c22c6390d0fff1443af2c95d056dc5ce4e61302")