Remove code related to legacy ingestion support

services/horizon/CHANGELOG.md

@@ -12,11 +12,12 @@ file. This project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 - Added new command-line flag `--network` to specify the Stellar network (pubnet or testnet), aiming at simplifying the configuration process by automatically configuring the following parameters based on the chosen network: `--history-archive-urls`, `--network-passphrase`, and `--captive-core-config-path` ([4949](https://github.com/stellar/go/pull/4949)).
+- Add a deprecation warning for using command-line flags when running Horizon ([5051](https://github.com/stellar/go/pull/5051))
+- Deprecate configuration flags related to legacy non-captive core ingestion ([5100](https://github.com/stellar/go/pull/5100))
 
 ### Fixed
 - The same slippage calculation from the [`v2.26.1`](#2261) hotfix now properly excludes spikes for smoother trade aggregation plots ([4999](https://github.com/stellar/go/pull/4999)).
 - Limit the display of global flags on command line help `-h` output ([5077](https://github.com/stellar/go/pull/5077)).
-- Add a deprecation warning for using command-line flags when running Horizon ([5051](https://github.com/stellar/go/pull/5051))
 
 ### DB Schema Migration
 - Drop unused indices from the Horizon database. For the database with full history, the migration is anticipated to take up to an hour and is expected to free up approximately 1.3TB of storage ([5081](https://github.com/stellar/go/pull/5081)).

services/horizon/cmd/db.go

@@ -406,7 +406,6 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
 		ReingestEnabled:             true,
 		MaxReingestRetries:          int(retries),
 		ReingestRetryBackoffSeconds: int(retryBackoffSeconds),
-		EnableCaptiveCore:           config.EnableCaptiveCoreIngestion,
 		CaptiveCoreBinaryPath:       config.CaptiveCoreBinaryPath,
 		CaptiveCoreConfigUseDB:      config.CaptiveCoreConfigUseDB,
 		RemoteCaptiveCoreURL:        config.RemoteCaptiveCoreURL,
@@ -422,16 +421,6 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
 		return fmt.Errorf("cannot open Horizon DB: %v", err)
 	}
 
-	if !config.EnableCaptiveCoreIngestion {
-		if config.StellarCoreDatabaseURL == "" {
-			return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
-		}
-		if ingestConfig.CoreSession, err = db.Open("postgres", config.StellarCoreDatabaseURL); err != nil {
-			ingestConfig.HistorySession.Close()
-			return fmt.Errorf("cannot open Core DB: %v", err)
-		}
-	}
-
 	if parallelWorkers > 1 {
 		system, systemErr := ingest.NewParallelSystems(ingestConfig, parallelWorkers)
 		if systemErr != nil {

services/horizon/cmd/ingest.go

@@ -128,7 +128,6 @@ var ingestVerifyRangeCmd = &cobra.Command{
 			NetworkPassphrase:        globalConfig.NetworkPassphrase,
 			HistorySession:           horizonSession,
 			HistoryArchiveURLs:       globalConfig.HistoryArchiveURLs,
-			EnableCaptiveCore:        globalConfig.EnableCaptiveCoreIngestion,
 			CaptiveCoreBinaryPath:    globalConfig.CaptiveCoreBinaryPath,
 			CaptiveCoreConfigUseDB:   globalConfig.CaptiveCoreConfigUseDB,
 			RemoteCaptiveCoreURL:     globalConfig.RemoteCaptiveCoreURL,
@@ -139,18 +138,6 @@ var ingestVerifyRangeCmd = &cobra.Command{
 			EnableIngestionFiltering: globalConfig.EnableIngestionFiltering,
 		}
 
-		if !ingestConfig.EnableCaptiveCore {
-			if globalConfig.StellarCoreDatabaseURL == "" {
-				return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
-			}
-
-			coreSession, dbErr := db.Open("postgres", globalConfig.StellarCoreDatabaseURL)
-			if dbErr != nil {
-				return fmt.Errorf("cannot open Core DB: %v", dbErr)
-			}
-			ingestConfig.CoreSession = coreSession
-		}
-
 		system, err := ingest.NewSystem(ingestConfig)
 		if err != nil {
 			return err
@@ -224,24 +211,10 @@ var ingestStressTestCmd = &cobra.Command{
 			NetworkPassphrase:      globalConfig.NetworkPassphrase,
 			HistorySession:         horizonSession,
 			HistoryArchiveURLs:     globalConfig.HistoryArchiveURLs,
-			EnableCaptiveCore:      globalConfig.EnableCaptiveCoreIngestion,
 			RoundingSlippageFilter: globalConfig.RoundingSlippageFilter,
-		}
-
-		if globalConfig.EnableCaptiveCoreIngestion {
-			ingestConfig.CaptiveCoreBinaryPath = globalConfig.CaptiveCoreBinaryPath
-			ingestConfig.RemoteCaptiveCoreURL = globalConfig.RemoteCaptiveCoreURL
-			ingestConfig.CaptiveCoreConfigUseDB = globalConfig.CaptiveCoreConfigUseDB
-		} else {
-			if globalConfig.StellarCoreDatabaseURL == "" {
-				return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
-			}
-
-			coreSession, dbErr := db.Open("postgres", globalConfig.StellarCoreDatabaseURL)
-			if dbErr != nil {
-				return fmt.Errorf("cannot open Core DB: %v", dbErr)
-			}
-			ingestConfig.CoreSession = coreSession
+			CaptiveCoreBinaryPath:  globalConfig.CaptiveCoreBinaryPath,
+			RemoteCaptiveCoreURL:   globalConfig.RemoteCaptiveCoreURL,
+			CaptiveCoreConfigUseDB: globalConfig.CaptiveCoreConfigUseDB,
 		}
 
 		system, err := ingest.NewSystem(ingestConfig)
@@ -315,25 +288,11 @@ var ingestInitGenesisStateCmd = &cobra.Command{
 			NetworkPassphrase:        globalConfig.NetworkPassphrase,
 			HistorySession:           horizonSession,
 			HistoryArchiveURLs:       globalConfig.HistoryArchiveURLs,
-			EnableCaptiveCore:        globalConfig.EnableCaptiveCoreIngestion,
 			CheckpointFrequency:      globalConfig.CheckpointFrequency,
 			RoundingSlippageFilter:   globalConfig.RoundingSlippageFilter,
 			EnableIngestionFiltering: globalConfig.EnableIngestionFiltering,
-		}
-
-		if globalConfig.EnableCaptiveCoreIngestion {
-			ingestConfig.CaptiveCoreBinaryPath = globalConfig.CaptiveCoreBinaryPath
-			ingestConfig.CaptiveCoreConfigUseDB = globalConfig.CaptiveCoreConfigUseDB
-		} else {
-			if globalConfig.StellarCoreDatabaseURL == "" {
-				return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
-			}
-
-			coreSession, dbErr := db.Open("postgres", globalConfig.StellarCoreDatabaseURL)
-			if dbErr != nil {
-				return fmt.Errorf("cannot open Core DB: %v", dbErr)
-			}
-			ingestConfig.CoreSession = coreSession
+			CaptiveCoreBinaryPath:    globalConfig.CaptiveCoreBinaryPath,
+			CaptiveCoreConfigUseDB:   globalConfig.CaptiveCoreConfigUseDB,
 		}
 
 		system, err := ingest.NewSystem(ingestConfig)
@@ -392,7 +351,6 @@ var ingestBuildStateCmd = &cobra.Command{
 			NetworkPassphrase:        globalConfig.NetworkPassphrase,
 			HistorySession:           horizonSession,
 			HistoryArchiveURLs:       globalConfig.HistoryArchiveURLs,
-			EnableCaptiveCore:        globalConfig.EnableCaptiveCoreIngestion,
 			CaptiveCoreBinaryPath:    globalConfig.CaptiveCoreBinaryPath,
 			CaptiveCoreConfigUseDB:   globalConfig.CaptiveCoreConfigUseDB,
 			RemoteCaptiveCoreURL:     globalConfig.RemoteCaptiveCoreURL,
@@ -403,20 +361,6 @@ var ingestBuildStateCmd = &cobra.Command{
 			EnableIngestionFiltering: globalConfig.EnableIngestionFiltering,
 		}
 
-		if !ingestBuildStateSkipChecks {
-			if !ingestConfig.EnableCaptiveCore {
-				if globalConfig.StellarCoreDatabaseURL == "" {
-					return fmt.Errorf("flag --%s cannot be empty", horizon.StellarCoreDBURLFlagName)
-				}
-
-				coreSession, dbErr := db.Open("postgres", globalConfig.StellarCoreDatabaseURL)
-				if dbErr != nil {
-					return fmt.Errorf("cannot open Core DB: %v", dbErr)
-				}
-				ingestConfig.CoreSession = coreSession
-			}
-		}
-
 		system, err := ingest.NewSystem(ingestConfig)
 		if err != nil {
 			return err

services/horizon/docker/verify-range/start

@@ -72,9 +72,8 @@ dump_horizon_db() {
 export NETWORK_PASSPHRASE="Public Global Stellar Network ; September 2015"
 export HISTORY_ARCHIVE_URLS="https://s3-eu-west-1.amazonaws.com/history.stellar.org/prd/core-live/core_live_001"
 export DATABASE_URL="postgres://postgres:postgres@localhost:5432/horizon?sslmode=disable"
-export CAPTIVE_CORE_CONFIG_APPEND_PATH="/captive-core-pubnet.cfg" 
+export CAPTIVE_CORE_CONFIG_APPEND_PATH="/captive-core-pubnet.cfg"
 export STELLAR_CORE_BINARY_PATH="/usr/bin/stellar-core"
-export ENABLE_CAPTIVE_CORE_INGESTION="true"
 
 cd stellar-go
 git pull origin

services/horizon/internal/config.go

@@ -19,7 +19,6 @@ type Config struct {
 	Port               uint
 	AdminPort          uint
 
-	EnableCaptiveCoreIngestion  bool
 	EnableIngestionFiltering    bool
 	CaptiveCoreBinaryPath       string
 	RemoteCaptiveCoreURL        string
@@ -30,8 +29,7 @@ type Config struct {
 	CaptiveCoreReuseStoragePath bool
 	CaptiveCoreConfigUseDB      bool
 
-	StellarCoreDatabaseURL string
-	StellarCoreURL         string
+	StellarCoreURL string
 
 	// MaxDBConnections has a priority over all 4 values below.
 	MaxDBConnections            int

services/horizon/internal/flags.go

@@ -58,7 +58,6 @@ const (
 	// DisableTxSubFlagName is the command line flag for disabling transaction submission feature of Horizon
 	DisableTxSubFlagName = "disable-tx-sub"
 
-	captiveCoreMigrationHint = "If you are migrating from Horizon 1.x.y, start with the Migration Guide here: https://developers.stellar.org/docs/run-api-server/migrating/"
 	// StellarPubnet is a constant representing the Stellar public network
 	StellarPubnet = "pubnet"
 	// StellarTestnet is a constant representing the Stellar test network
@@ -251,12 +250,21 @@ func Flags() (*Config, support.ConfigOptions) {
 			UsedInCommands: IngestionCommands,
 		},
 		&support.ConfigOption{
-			Name:           "enable-captive-core-ingestion",
-			OptType:        types.Bool,
-			FlagDefault:    true,
-			Required:       false,
-			Usage:          "causes Horizon to ingest from a Captive Stellar Core process instead of a persistent Stellar Core database",
-			ConfigKey:      &config.EnableCaptiveCoreIngestion,
+			Name:        EnableCaptiveCoreIngestionFlagName,
+			OptType:     types.String,
+			FlagDefault: "",
+			Required:    false,
+			Hidden:      true,
+			CustomSetValue: func(opt *support.ConfigOption) error {
+				if val := viper.GetString(opt.Name); val != "" {
+					stdLog.Printf(
+						"DEPRECATED - The usage of the flag --enable-captive-core-ingestion has been deprecated. " +
+							"Horizon now uses Captive-Core ingestion by default and this flag will soon be removed in " +
+							"the future.",
+					)
+				}
+				return nil
+			},
 			UsedInCommands: IngestionCommands,
 		},
 		&support.ConfigOption{
@@ -323,12 +331,21 @@ func Flags() (*Config, support.ConfigOptions) {
 			UsedInCommands: IngestionCommands,
 		},
 		&support.ConfigOption{
-			Name:           StellarCoreDBURLFlagName,
-			EnvVar:         "STELLAR_CORE_DATABASE_URL",
-			ConfigKey:      &config.StellarCoreDatabaseURL,
-			OptType:        types.String,
-			Required:       false,
-			Usage:          "stellar-core postgres database to connect with",
+			Name:     StellarCoreDBURLFlagName,
+			EnvVar:   "STELLAR_CORE_DATABASE_URL",
+			OptType:  types.String,
+			Required: false,
+			Hidden:   true,
+			CustomSetValue: func(opt *support.ConfigOption) error {
+				if val := viper.GetString(opt.Name); val != "" {
+					stdLog.Printf(
+						"DEPRECATED - The usage of the flag --stellar-core-db-url has been deprecated. " +
+							"Horizon now uses Captive-Core ingestion by default and this flag will soon be removed in " +
+							"the future.",
+					)
+				}
+				return nil
+			},
 			UsedInCommands: IngestionCommands,
 		},
 		&support.ConfigOption{
@@ -724,9 +741,6 @@ func NewAppFromFlags(config *Config, flags support.ConfigOptions) (*App, error)
 	if config.StellarCoreURL == "" {
 		return nil, fmt.Errorf("flag --%s cannot be empty", StellarCoreURLFlagName)
 	}
-	if config.Ingest && !config.EnableCaptiveCoreIngestion && config.StellarCoreDatabaseURL == "" {
-		return nil, fmt.Errorf("flag --%s cannot be empty", StellarCoreDBURLFlagName)
-	}
 
 	log.Infof("Initializing horizon...")
 	app, err := NewApp(*config)
@@ -916,24 +930,9 @@ func ApplyFlags(config *Config, flags support.ConfigOptions, options ApplyOption
 			return err
 		}
 
-		if config.EnableCaptiveCoreIngestion {
-			err := setCaptiveCoreConfiguration(config, options)
-			if err != nil {
-				return errors.Wrap(err, "error generating captive core configuration")
-			}
-		}
-	} else {
-		if config.EnableCaptiveCoreIngestion && (config.CaptiveCoreBinaryPath != "" || config.CaptiveCoreConfigPath != "") {
-			captiveCoreConfigFlag := captiveCoreConfigAppendPathName
-			if viper.GetString(CaptiveCoreConfigPathName) != "" {
-				captiveCoreConfigFlag = CaptiveCoreConfigPathName
-			}
-			return fmt.Errorf("invalid config: one or more captive core params passed (--%s or --%s) but --ingest not set"+captiveCoreMigrationHint,
-				StellarCoreBinaryPathName, captiveCoreConfigFlag)
-		}
-		if config.StellarCoreDatabaseURL != "" {
-			return fmt.Errorf("invalid config: --%s passed but --ingest not set"+
-				"", StellarCoreDBURLFlagName)
+		err := setCaptiveCoreConfiguration(config, options)
+		if err != nil {
+			return errors.Wrap(err, "error generating captive core configuration")
 		}
 	}
 

services/horizon/internal/flags_test.go

@@ -208,20 +208,19 @@ func Test_createCaptiveCoreConfig(t *testing.T) {
 
 func TestEnvironmentVariables(t *testing.T) {
 	environmentVars := map[string]string{
-		"INGEST":                        "false",
-		"HISTORY_ARCHIVE_URLS":          "http://localhost:1570",
-		"DATABASE_URL":                  "postgres://postgres@localhost/test_332cb65e6b00?sslmode=disable&timezone=UTC",
-		"STELLAR_CORE_URL":              "http://localhost:11626",
-		"NETWORK_PASSPHRASE":            "Standalone Network ; February 2017",
-		"APPLY_MIGRATIONS":              "true",
-		"ENABLE_CAPTIVE_CORE_INGESTION": "false",
-		"CHECKPOINT_FREQUENCY":          "8",
-		"MAX_DB_CONNECTIONS":            "50",
-		"ADMIN_PORT":                    "6060",
-		"PORT":                          "8001",
-		"CAPTIVE_CORE_BINARY_PATH":      os.Getenv("HORIZON_INTEGRATION_TESTS_CAPTIVE_CORE_BIN"),
-		"CAPTIVE_CORE_CONFIG_PATH":      "../docker/captive-core-classic-integration-tests.cfg",
-		"CAPTIVE_CORE_USE_DB":           "true",
+		"INGEST":                   "false",
+		"HISTORY_ARCHIVE_URLS":     "http://localhost:1570",
+		"DATABASE_URL":             "postgres://postgres@localhost/test_332cb65e6b00?sslmode=disable&timezone=UTC",
+		"STELLAR_CORE_URL":         "http://localhost:11626",
+		"NETWORK_PASSPHRASE":       "Standalone Network ; February 2017",
+		"APPLY_MIGRATIONS":         "true",
+		"CHECKPOINT_FREQUENCY":     "8",
+		"MAX_DB_CONNECTIONS":       "50",
+		"ADMIN_PORT":               "6060",
+		"PORT":                     "8001",
+		"CAPTIVE_CORE_BINARY_PATH": os.Getenv("HORIZON_INTEGRATION_TESTS_CAPTIVE_CORE_BIN"),
+		"CAPTIVE_CORE_CONFIG_PATH": "../docker/captive-core-classic-integration-tests.cfg",
+		"CAPTIVE_CORE_USE_DB":      "true",
 	}
 
 	envManager := test.NewEnvironmentManager()
@@ -252,7 +251,6 @@ func TestEnvironmentVariables(t *testing.T) {
 	assert.Equal(t, config.StellarCoreURL, "http://localhost:11626")
 	assert.Equal(t, config.NetworkPassphrase, "Standalone Network ; February 2017")
 	assert.Equal(t, config.ApplyMigrations, true)
-	assert.Equal(t, config.EnableCaptiveCoreIngestion, false)
 	assert.Equal(t, config.CheckpointFrequency, uint32(8))
 	assert.Equal(t, config.MaxDBConnections, 50)
 	assert.Equal(t, config.AdminPort, uint(6060))

services/horizon/internal/helpers_test.go

@@ -8,7 +8,6 @@ import (
 
 	"github.com/stellar/go/network"
 	"github.com/stellar/go/services/horizon/internal/test"
-	tdb "github.com/stellar/go/services/horizon/internal/test/db"
 	supportLog "github.com/stellar/go/support/log"
 )
 
@@ -22,8 +21,7 @@ func NewTestApp(dsn string) *App {
 
 func NewTestConfig(dsn string) Config {
 	return Config{
-		DatabaseURL:            dsn,
-		StellarCoreDatabaseURL: tdb.StellarCoreURL(),
+		DatabaseURL: dsn,
 		RateQuota: &throttled.RateQuota{
 			MaxRate:  throttled.PerHour(1000),
 			MaxBurst: 100,

services/horizon/internal/ingest/main.go

@@ -76,7 +76,6 @@ type Config struct {
 	CoreSession            db.SessionInterface
 	StellarCoreURL         string
 	StellarCoreCursor      string
-	EnableCaptiveCore      bool
 	CaptiveCoreBinaryPath  string
 	CaptiveCoreStoragePath string
 	CaptiveCoreToml        *ledgerbackend.CaptiveCoreToml
@@ -108,16 +107,14 @@ type Config struct {
 // LocalCaptiveCoreEnabled returns true if configured to run
 // a local captive core instance for ingestion.
 func (c Config) LocalCaptiveCoreEnabled() bool {
-	// c.EnableCaptiveCore is true for both local and remote captive core
-	// and c.RemoteCaptiveCoreURL is always empty when running
-	// local captive core.
-	return c.EnableCaptiveCore && c.RemoteCaptiveCoreURL == ""
+	// c.RemoteCaptiveCoreURL is always empty when running local captive core.
+	return c.RemoteCaptiveCoreURL == ""
 }
 
 // RemoteCaptiveCoreEnabled returns true if configured to run
 // a remote captive core instance for ingestion.
 func (c Config) RemoteCaptiveCoreEnabled() bool {
-	return c.EnableCaptiveCore && c.RemoteCaptiveCoreURL != ""
+	return c.RemoteCaptiveCoreURL != ""
 }
 
 const (
@@ -742,7 +739,7 @@ func (s *system) resetStateVerificationErrors() {
 }
 
 func (s *system) updateCursor(ledgerSequence uint32) error {
-	if s.stellarCoreClient == nil || s.config.EnableCaptiveCore {
+	if s.stellarCoreClient == nil {
 		return nil
 	}
 

services/horizon/internal/init.go

@@ -92,10 +92,6 @@ func mustInitHorizonDB(app *App) {
 func initIngester(app *App) {
 	var err error
 	var coreSession db.SessionInterface
-	if !app.config.EnableCaptiveCoreIngestion {
-		coreSession = mustNewDBSession(
-			db.CoreSubservice, app.config.StellarCoreDatabaseURL, ingest.MaxDBConnections, ingest.MaxDBConnections, app.prometheusRegistry)
-	}
 	app.ingester, err = ingest.NewSystem(ingest.Config{
 		CoreSession: coreSession,
 		HistorySession: mustNewDBSession(
@@ -111,7 +107,6 @@ func initIngester(app *App) {
 		CaptiveCoreConfigUseDB:               app.config.CaptiveCoreConfigUseDB,
 		CaptiveCoreToml:                      app.config.CaptiveCoreToml,
 		RemoteCaptiveCoreURL:                 app.config.RemoteCaptiveCoreURL,
-		EnableCaptiveCore:                    app.config.EnableCaptiveCoreIngestion,
 		DisableStateVerification:             app.config.IngestDisableStateVerification,
 		StateVerificationCheckpointFrequency: uint32(app.config.IngestStateVerificationCheckpointFrequency),
 		StateVerificationTimeout:             app.config.IngestStateVerificationTimeout,