[+] allow sink schema upgrades without --metrics/--sources, fixes #1114 (#1175)

* [+] allow sink schema upgrades without --metrics/--sources, fixes #1114

Removed the requirement that both sources and metrics must be PostgreSQL
connections. Added support for upgrading only sink when sources/metrics
are YAML files. Changed error messages to warnings ([WARN]) for
non-upgradeable configurations. Only fails if no components support
upgrade (previously failed if any didn't support upgrade).

* make linter happy

* upgrade all supported storages one by one regardless of specified

Author: pashagolub

cmd/pgwatch/version.go

@@ -4,19 +4,21 @@ import "fmt"
 
 // version output variables
 var (
-	commit  = "unknown"
-	version = "unknown"
-	date    = "unknown"
-	dbapi   = "00824"
+	commit       = "unknown"
+	version      = "unknown"
+	date         = "unknown"
+	configSchema = "00824"
+	sinkSchema   = "01110"
 )
 
 func printVersion() {
 	fmt.Printf(`
 Version info:
-  Version:      %s
-  DB Schema:    %s
-  Git Commit:   %s
-  Built:        %s
+  Version:       %s
+  Config Schema: %s
+  Sink Schema:   %s
+  Git Commit:    %s
+  Built:         %s
 
-`, version, dbapi, commit, date)
+`, version, configSchema, sinkSchema, commit, date)
 }

internal/cmdopts/cmdconfig.go

@@ -3,6 +3,7 @@ package cmdopts
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"github.com/cybertec-postgresql/pgwatch/v5/internal/db"
 	"github.com/cybertec-postgresql/pgwatch/v5/internal/metrics"
@@ -87,46 +88,62 @@ type ConfigUpgradeCommand struct {
 // Execute upgrades the configuration schema.
 func (cmd *ConfigUpgradeCommand) Execute([]string) (err error) {
 	opts := cmd.owner
-	if err = opts.ValidateConfig(); err != nil {
-		return
+	// For upgrade command, validate that at least one component is specified
+	if len(opts.Sources.Sources)+len(opts.Metrics.Metrics)+len(opts.Sinks.Sinks) == 0 {
+		opts.CompleteCommand(ExitCodeConfigError)
+		return errors.New("at least one of --sources, --metrics, or --sink must be specified")
 	}
+
 	ctx := context.Background()
-	// Upgrade metrics/sources configuration if it's postgres
-	if opts.IsPgConnStr(opts.Metrics.Metrics) && opts.IsPgConnStr(opts.Sources.Sources) {
-		err = opts.InitMetricReader(ctx)
-		if err != nil {
-			opts.CompleteCommand(ExitCodeConfigError)
-			return
+
+	f := func(uri string, newMigratorFunc func() (any, error)) error {
+		if uri == "" {
+			return nil
 		}
-		if m, ok := opts.MetricsReaderWriter.(db.Migrator); ok {
-			err = m.Migrate()
-			if err != nil {
-				opts.CompleteCommand(ExitCodeConfigError)
-				return
-			}
+		if !opts.IsPgConnStr(uri) {
+			return fmt.Errorf("cannot upgrade storage %s: %w", uri, errors.ErrUnsupported)
 		}
-	} else {
-		opts.CompleteCommand(ExitCodeConfigError)
-		return errors.New("configuration storage does not support upgrade")
-	}
-	// Upgrade sinks configuration if it's postgres
-	if len(opts.Sinks.Sinks) > 0 {
-		opts.SinksWriter, err = sinks.NewSinkWriter(ctx, &opts.Sinks)
-		if err != nil {
-			opts.CompleteCommand(ExitCodeConfigError)
-			return
+		m, initErr := newMigratorFunc()
+		if initErr != nil {
+			return initErr
 		}
-		if m, ok := opts.SinksWriter.(db.Migrator); ok {
-			err = m.Migrate()
-			if err != nil {
-				opts.CompleteCommand(ExitCodeConfigError)
-				return
-			}
-		} else {
-			opts.CompleteCommand(ExitCodeConfigError)
-			return errors.New("sink storage does not support upgrade")
+		return m.(db.Migrator).Migrate()
+
+	}
+
+	err = f(opts.Sources.Sources, func() (any, error) {
+		return sources.NewPostgresSourcesReaderWriter(ctx, opts.Sources.Sources)
+	})
+
+	err = errors.Join(err, f(opts.Metrics.Metrics, func() (any, error) {
+		return metrics.NewPostgresMetricReaderWriter(ctx, opts.Metrics.Metrics)
+	}))
+
+	for _, uri := range opts.Sinks.Sinks {
+		err = errors.Join(err, f(uri, func() (any, error) {
+			return sinks.NewPostgresSinkMigrator(ctx, uri)
+		}))
+	}
+
+	if err == nil {
+		opts.CompleteCommand(ExitCodeOK)
+		return nil
+	}
+
+	// Check if all errors are ErrUnsupported
+	allUnsupported := true
+	for _, e := range err.(interface{ Unwrap() []error }).Unwrap() {
+		if !errors.Is(e, errors.ErrUnsupported) {
+			allUnsupported = false
+			break
 		}
+		fmt.Fprintln(opts.OutputWriter, e)
 	}
-	opts.CompleteCommand(ExitCodeOK)
-	return
+
+	if allUnsupported {
+		opts.CompleteCommand(ExitCodeOK)
+		return nil
+	}
+	opts.CompleteCommand(ExitCodeConfigError)
+	return err
 }

internal/cmdopts/cmdconfig_integration_test.go

@@ -2,6 +2,7 @@ package cmdopts
 
 import (
 	"context"
+	"io"
 	"testing"
 	"time"
 
@@ -29,16 +30,15 @@ func TestConfigUpgrade_VerifyNoCircularDependency(t *testing.T) {
 	require.NoError(t, err)
 
 	ctx := context.Background()
+
+	// Test with only metrics and sinks (sources reader doesn't implement Migrator)
 	opts := &Options{
-		Metrics: metrics.CmdOpts{Metrics: connStr},
-		Sources: sources.CmdOpts{Sources: connStr, Refresh: 120, MaxParallelConnectionsPerDb: 1},
-		Sinks:   sinks.CmdOpts{BatchingDelay: time.Second},
+		Metrics:      metrics.CmdOpts{Metrics: connStr},
+		Sinks:        sinks.CmdOpts{Sinks: []string{connStr}, RetentionInterval: "30 days", BatchingDelay: time.Second, PartitionInterval: "1 week", MaintenanceInterval: "12 hours"},
+		OutputWriter: io.Discard,
 	}
 
-	// This is the key test: config upgrade should work even on empty database
-	// (or database needing migrations). Before the fix, this would fail with
-	// circular dependency because InitConfigReaders would fail, preventing
-	// the upgrade from running
+	// Config upgrade should work on metrics and sinks
 	cmd := ConfigUpgradeCommand{owner: opts}
 	err = cmd.Execute(nil)
 	assert.NoError(t, err)
@@ -48,7 +48,7 @@ func TestConfigUpgrade_VerifyNoCircularDependency(t *testing.T) {
 	opts2 := &Options{
 		Metrics: metrics.CmdOpts{Metrics: connStr},
 		Sources: sources.CmdOpts{Sources: connStr, Refresh: 120, MaxParallelConnectionsPerDb: 1},
-		Sinks:   sinks.CmdOpts{BatchingDelay: time.Second},
+		Sinks:   sinks.CmdOpts{Sinks: []string{connStr}},
 	}
 	err = opts2.InitConfigReaders(ctx)
 	assert.NoError(t, err)