adding ledgerbackend datastore txmeta as a data source

cmd/export_account_signers.go

@@ -22,9 +22,9 @@ should be used in an initial data dump. In order to get account information with
 the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 
@@ -39,7 +39,7 @@ the export_ledger_entry_changes command.`,
 		numSigners := 0
 		var header xdr.LedgerHeaderHistoryEntry
 		for _, acc := range accounts {
-			if acc.AccountSignersChanged() {
+			if utils.AccountSignersChanged(acc) {
 				transformed, err := transform.TransformSigners(acc, header)
 				if err != nil {
 					cmdLogger.LogError(fmt.Errorf("could not json transform account signer: %v", err))

cmd/export_accounts.go

@@ -22,9 +22,9 @@ should be used in an initial data dump. In order to get account information with
 the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_all_history.go

@@ -20,13 +20,13 @@ This is a temporary command used to reduce the amount of requests to history arc
 in order to mitigate egress costs for the entity hosting history archives.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		allHistory, err := input.GetAllHistory(startNum, endNum, limit, env)
+		allHistory, err := input.GetAllHistory(startNum, endNum, limit, env, useCaptiveCore)
 		if err != nil {
 			cmdLogger.Fatal("could not read all history: ", err)
 		}

cmd/export_assets.go

@@ -16,14 +16,22 @@ var assetsCmd = &cobra.Command{
 	Long:  `Exports the assets that are created from payment operations over a specified ledger range`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
 		outFile := mustOutFile(path)
 
-		paymentOps, err := input.GetPaymentOperations(startNum, endNum, limit, isTest, isFuture)
+		var paymentOps []input.AssetTransformInput
+		var err error
+
+		if useCaptiveCore {
+			paymentOps, err = input.GetPaymentOperationsHistoryArchive(startNum, endNum, limit, env, useCaptiveCore)
+		} else {
+			paymentOps, err = input.GetPaymentOperations(startNum, endNum, limit, env, useCaptiveCore)
+		}
 		if err != nil {
 			cmdLogger.Fatal("could not read asset: ", err)
 		}

cmd/export_claimable_balances.go

@@ -22,9 +22,9 @@ var claimableBalancesCmd = &cobra.Command{
 	the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_config_setting.go

@@ -16,15 +16,15 @@ import (
 var configSettingCmd = &cobra.Command{
 	Use:   "export_config_setting",
 	Short: "Exports the config setting information.",
-	Long:  `Exports historical config settings data from the genesis ledger to the provided end-ledger to an output file. 
+	Long: `Exports historical config settings data from the genesis ledger to the provided end-ledger to an output file. 
 	The command reads from the bucket list, which includes the full history of the Stellar ledger. As a result, it 
 	should be used in an initial data dump. In order to get offer information within a specified ledger range, see 
 	the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_contract_code.go

@@ -16,15 +16,15 @@ import (
 var codeCmd = &cobra.Command{
 	Use:   "export_contract_code",
 	Short: "Exports the contract code information.",
-	Long:  `Exports historical contract code data from the genesis ledger to the provided end-ledger to an output file. 
+	Long: `Exports historical contract code data from the genesis ledger to the provided end-ledger to an output file. 
 	The command reads from the bucket list, which includes the full history of the Stellar ledger. As a result, it 
 	should be used in an initial data dump. In order to get offer information within a specified ledger range, see 
 	the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_contract_data.go

@@ -16,15 +16,15 @@ import (
 var dataCmd = &cobra.Command{
 	Use:   "export_contract_data",
 	Short: "Exports the contract data information made from the genesis ledger to a specified endpoint.",
-	Long:  `Exports historical contract data from the genesis ledger to the provided end-ledger to an output file. 
+	Long: `Exports historical contract data from the genesis ledger to the provided end-ledger to an output file. 
 	The command reads from the bucket list, which includes the full history of the Stellar ledger. As a result, it 
 	should be used in an initial data dump. In order to get offer information within a specified ledger range, see 
 	the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_diagnostic_events.go

@@ -16,13 +16,13 @@ var diagnosticEventsCmd = &cobra.Command{
 	Long:  `Exports the diagnostic events over a specified range to an output file.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		transactions, err := input.GetTransactions(startNum, endNum, limit, env)
+		transactions, err := input.GetTransactions(startNum, endNum, limit, env, useCaptiveCore)
 		if err != nil {
 			cmdLogger.Fatal("could not read transactions: ", err)
 		}

cmd/export_effects.go

@@ -14,13 +14,13 @@ var effectsCmd = &cobra.Command{
 	Long:  "Exports the effects data over a specified range to an output file.",
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		transactions, err := input.GetTransactions(startNum, endNum, limit, env)
+		transactions, err := input.GetTransactions(startNum, endNum, limit, env, useCaptiveCore)
 		if err != nil {
 			cmdLogger.Fatalf("could not read transactions in [%d, %d] (limit=%d): %v", startNum, endNum, limit, err)
 		}

cmd/export_ledger_entry_changes.go

@@ -1,12 +1,14 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"os"
 	"path/filepath"
 
 	"github.com/spf13/cobra"
+	"github.com/stellar/go/ingest/ledgerbackend"
 	"github.com/stellar/go/xdr"
 	"github.com/stellar/stellar-etl/internal/input"
 	"github.com/stellar/stellar-etl/internal/transform"
@@ -26,14 +28,16 @@ confirmed by the Stellar network.
 If no data type flags are set, then by default all of them are exported. If any are set, it is assumed that the others should not
 be exported.`,
 	Run: func(cmd *cobra.Command, args []string) {
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		execPath, configPath, startNum, batchSize, outputFolder := utils.MustCoreFlags(cmd.Flags(), cmdLogger)
+		_, configPath, startNum, batchSize, outputFolder := utils.MustCoreFlags(cmd.Flags(), cmdLogger)
 		exports := utils.MustExportTypeFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 
+		cmd.Flags()
+
 		err := os.MkdirAll(outputFolder, os.ModePerm)
 		if err != nil {
 			cmdLogger.Fatalf("unable to mkdir %s: %v", outputFolder, err)
@@ -46,7 +50,7 @@ be exported.`,
 		// If none of the export flags are set, then we assume that everything should be exported
 		allFalse := true
 		for _, value := range exports {
-			if true == value {
+			if value {
 				allFalse = false
 				break
 			}
@@ -62,19 +66,15 @@ be exported.`,
 			cmdLogger.Fatal("stellar-core needs a config file path when exporting ledgers continuously (endNum = 0)")
 		}
 
-		execPath, err = filepath.Abs(execPath)
-		if err != nil {
-			cmdLogger.Fatal("could not get absolute filepath for stellar-core executable: ", err)
-		}
-
-		configPath, err = filepath.Abs(configPath)
+		ctx := context.Background()
+		backend, err := utils.CreateLedgerBackend(ctx, useCaptiveCore, env)
 		if err != nil {
-			cmdLogger.Fatal("could not get absolute filepath for the config file: ", err)
+			cmdLogger.Fatal("error creating a cloud storage backend: ", err)
 		}
 
-		core, err := input.PrepareCaptiveCore(execPath, configPath, startNum, endNum, env)
+		err = backend.PrepareRange(ctx, ledgerbackend.BoundedRange(startNum, endNum))
 		if err != nil {
-			cmdLogger.Fatal("error creating a prepared captive core instance: ", err)
+			cmdLogger.Fatal("error preparing ledger range for cloud storage backend: ", err)
 		}
 
 		if endNum == 0 {
@@ -83,7 +83,7 @@ be exported.`,
 
 		changeChan := make(chan input.ChangeBatch)
 		closeChan := make(chan int)
-		go input.StreamChanges(core, startNum, endNum, batchSize, changeChan, closeChan, env, cmdLogger)
+		go input.StreamChanges(&backend, startNum, endNum, batchSize, changeChan, closeChan, env, cmdLogger)
 
 		for {
 			select {
@@ -126,7 +126,7 @@ be exported.`,
 								}
 								transformedOutputs["accounts"] = append(transformedOutputs["accounts"], acc)
 							}
-							if change.AccountSignersChanged() {
+							if utils.AccountSignersChanged(change) {
 								signers, err := transform.TransformSigners(change, changes.LedgerHeaders[i])
 								if err != nil {
 									entry, _, _, _ := utils.ExtractEntryFromChange(change)
@@ -295,7 +295,6 @@ func init() {
 	utils.AddCloudStorageFlags(exportLedgerEntryChangesCmd.Flags())
 
 	exportLedgerEntryChangesCmd.MarkFlagRequired("start-ledger")
-	exportLedgerEntryChangesCmd.MarkFlagRequired("core-executable")
 	/*
 		Current flags:
 			start-ledger: the ledger sequence number for the beginning of the export period

cmd/export_ledger_transaction.go

@@ -16,13 +16,13 @@ var ledgerTransactionCmd = &cobra.Command{
 	Long:  `Exports the ledger_transaction transaction data over a specified range to an output file.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		ledgerTransaction, err := input.GetTransactions(startNum, endNum, limit, env)
+		ledgerTransaction, err := input.GetTransactions(startNum, endNum, limit, env, useCaptiveCore)
 		if err != nil {
 			cmdLogger.Fatal("could not read ledger_transaction: ", err)
 		}

cmd/export_ledgers.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	"github.com/stellar/go/historyarchive"
 	"github.com/stellar/stellar-etl/internal/input"
 	"github.com/stellar/stellar-etl/internal/transform"
 	"github.com/stellar/stellar-etl/internal/utils"
@@ -16,12 +17,20 @@ var ledgersCmd = &cobra.Command{
 	Long:  `Exports ledger data within the specified range to an output file. Encodes ledgers as JSON objects and exports them to the output file.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		ledgers, err := input.GetLedgers(startNum, endNum, limit, isTest, isFuture)
+		var ledgers []historyarchive.Ledger
+		var err error
+
+		if useCaptiveCore {
+			ledgers, err = input.GetLedgersHistoryArchive(startNum, endNum, limit, env, useCaptiveCore)
+		} else {
+			ledgers, err = input.GetLedgers(startNum, endNum, limit, env, useCaptiveCore)
+		}
 		if err != nil {
 			cmdLogger.Fatal("could not read ledgers: ", err)
 		}

cmd/export_liquidity_pools.go

@@ -22,9 +22,9 @@ should be used in an initial data dump. In order to get liqudity pools informati
 the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_offers.go

@@ -23,9 +23,9 @@ var offersCmd = &cobra.Command{
 	the export_ledger_entry_changes command.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, _, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 		path := utils.MustBucketFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
 

cmd/export_operations.go

@@ -16,13 +16,13 @@ var operationsCmd = &cobra.Command{
 	Long:  `Exports the operations data over a specified range. Each operation is an individual command that mutates the Stellar ledger.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		cmdLogger.SetLevel(logrus.InfoLevel)
-		endNum, strictExport, isTest, isFuture, extra := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
+		endNum, strictExport, isTest, isFuture, extra, useCaptiveCore, datastoreUrl := utils.MustCommonFlags(cmd.Flags(), cmdLogger)
 		cmdLogger.StrictExport = strictExport
 		startNum, path, limit := utils.MustArchiveFlags(cmd.Flags(), cmdLogger)
 		cloudStorageBucket, cloudCredentials, cloudProvider := utils.MustCloudStorageFlags(cmd.Flags(), cmdLogger)
-		env := utils.GetEnvironmentDetails(isTest, isFuture)
+		env := utils.GetEnvironmentDetails(isTest, isFuture, datastoreUrl)
 
-		operations, err := input.GetOperations(startNum, endNum, limit, env)
+		operations, err := input.GetOperations(startNum, endNum, limit, env, useCaptiveCore)
 		if err != nil {
 			cmdLogger.Fatal("could not read operations: ", err)
 		}