exp/services/ledgerexporter: Initial draft

.github/workflows/horizon.yml

@@ -162,6 +162,7 @@ jobs:
   ledger-exporter:
     name: Test and push the Ledger Exporter images
     runs-on: ubuntu-latest
+    if: false # Disable the job
     steps:
       - uses: actions/checkout@v3
         with:

Makefile

@@ -14,8 +14,8 @@ xdr/Stellar-contract.x \
 xdr/Stellar-internal.x \
 xdr/Stellar-contract-config-setting.x
 
-XDRS = $(DOWNLOADABLE_XDRS) xdr/Stellar-lighthorizon.x
-
+XDRS = $(DOWNLOADABLE_XDRS) xdr/Stellar-lighthorizon.x \
+		xdr/Stellar-exporter.x
 
 
 XDRGEN_COMMIT=e2cac557162d99b12ae73b846cf3d5bfe16636de

exp/services/ledgerexporter/README.md

@@ -0,0 +1,80 @@
+# Ledger Exporter (Work in Progress)
+
+The Ledger Exporter is a tool designed to export ledger data from a Stellar network and upload it to a specified destination. It supports both bounded and unbounded modes, allowing users to export a specific range of ledgers or continuously export new ledgers as they arrive on the network.
+
+Ledger Exporter currently uses captive-core as the ledger backend and GCS as the destination data store.
+
+# Exported Data Format
+The tool allows for the export of multiple ledgers in a single exported file. The exported data is in XDR format and is compressed (gzip) before being uploaded.
+
+```go
+type LedgerCloseMetaBatch struct {
+    StartSequence uint32
+    EndSequence uint32
+    LedgerCloseMetas []LedgerCloseMeta
+}
+```
+
+## Getting Started
+
+### Installation (coming soon)
+
+### Command Line Options
+
+#### Bounded Mode:
+Exports a specific range of ledgers, defined by --start and --end.
+```bash
+ledgerexporter --start <start_ledger> --end <end_ledger> --config-file <config_file_path>
+```
+
+#### Unbounded Mode:
+Exports ledgers continuously starting from --start. In this mode, the end ledger is either not provided or set to 0.
+```bash
+ledgerexporter --start <start_ledger> --config-file <config_file_path>
+```
+
+
+Starts exporting from a specified number of ledgers before the latest ledger sequence number on the network.
+```bash
+ledgerexporter --from-last <number_of_ledgers> --config-file <config_file_path>
+```
+
+
+## Configuration File (config.toml)
+
+```toml
+network = "testnet"
+destination_url = "gcs://your-bucket-name"
+
+[exporter_config]
+ledgers_per_file = 64
+files_per_partition = 10
+```
+
+## Exported Files
+
+### File Organization:
+- Ledgers are grouped into files, with the number of ledgers per file set by `ledgers_per_file`.
+- Files are further organized into partitions, with the number of files per partition set by `files_per_partition`.
+
+### Filename Structure:
+- Filenames indicate the ledger range they contain, e.g., `0-63.xdr.gz` holds ledgers 0 to 63.
+- Partition directories group files, e.g., `/0-639/` holds files for ledgers 0 to 639.
+
+#### Example:
+with `ledgers_per_file = 64` and `files_per_partition = 10`:
+- Partition names: `/0-639`, `/640-1279`, ...
+- Filenames: `/0-639/0-63.xdr.gz`, `/0-639/64-127.xdr.gz`, ...
+
+#### Special Cases:
+
+- If `ledgers_per_file` is set to 1, filenames will only contain the ledger number.
+- If `files_per_partition` is set to 1, filenames will not contain the partition.
+
+#### Note:
+- Avoid changing `ledgers_per_file` and `files_per_partition` after configuration for consistency.
+
+#### Retrieving Data:
+- To locate a specific ledger sequence, calculate the partition name and ledger file name using `files_per_partition` and `ledgers_per_file`.
+- The `GetObjectKeyFromSequenceNumber` function automates this calculation.
+

exp/services/ledgerexporter/config.toml

@@ -0,0 +1,7 @@
+network = "testnet"
+destination_url = "gcs://exporter-test/ledgers"
+
+[exporter_config]
+  ledgers_per_file = 1
+  files_per_partition = 64000
+

exp/services/ledgerexporter/internal/app.go

@@ -0,0 +1,139 @@
+package exporter
+
+import (
+	"context"
+	_ "embed"
+	"fmt"
+	"os"
+	"os/signal"
+	"sync"
+	"syscall"
+
+	"github.com/pkg/errors"
+	"github.com/stellar/go/ingest/ledgerbackend"
+	_ "github.com/stellar/go/network"
+	supportlog "github.com/stellar/go/support/log"
+)
+
+var (
+	logger = supportlog.New().WithField("service", "ledger-exporter")
+)
+
+type App struct {
+	config             Config
+	backend            ledgerbackend.LedgerBackend
+	destinationStorage DataStore
+	exportManager      ExportManager
+	uploader           Uploader
+}
+
+func NewApp() *App {
+	logger.SetLevel(supportlog.DebugLevel)
+
+	config := Config{}
+	err := config.LoadConfig()
+	logFatalIf(err, "Could not load configuration")
+
+	app := &App{config: config}
+	return app
+}
+
+func (a *App) init(ctx context.Context) {
+	a.destinationStorage = mustNewDataStore(ctx, &a.config)
+	a.backend = mustNewLedgerBackend(ctx, a.config)
+	a.exportManager = NewExportManager(a.config.ExporterConfig, a.backend)
+	a.uploader = NewUploader(a.destinationStorage, a.exportManager.GetMetaArchiveChannel())
+}
+
+func (a *App) close() {
+	//TODO: error handling
+	a.destinationStorage.Close()
+	a.backend.Close()
+}
+
+func (a *App) Run() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	a.init(ctx)
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+
+		err := a.uploader.Run(ctx)
+		if err != nil && !errors.Is(err, context.Canceled) {
+			logger.Errorf("Error executing Uploader: %v", err)
+			cancel()
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+
+		err := a.exportManager.Run(ctx, a.config.StartLedger, a.config.EndLedger)
+		if err != nil && !errors.Is(err, context.Canceled) {
+			logger.Errorf("Error executing ExportManager: %v", err)
+			cancel()
+		}
+	}()
+
+	go func() {
+		// Handle OS signals to gracefully terminate the service
+		sigCh := make(chan os.Signal, 1)
+		signal.Notify(sigCh, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+		for {
+			select {
+			case <-ctx.Done():
+				logger.Infof("Received context done signal")
+				return
+			case sig := <-sigCh:
+				logger.Infof("Received signal: %v", sig)
+				cancel()
+				return
+			}
+		}
+	}()
+
+	wg.Wait()
+
+	a.close()
+
+	logger.Info("Shutting down ledger-exporter.")
+}
+
+func mustNewDataStore(ctx context.Context, config *Config) DataStore {
+	destinationStorage, err := NewDataStore(ctx, fmt.Sprintf("%s/%s", config.DestinationURL, config.Network))
+	logFatalIf(err, "Could not connect to destination storage")
+	return destinationStorage
+}
+
+// mustNewLedgerBackend Creates and initializes captive core ledger backend
+// Currently, only supports captive-core as ledger backend
+func mustNewLedgerBackend(ctx context.Context, config Config) ledgerbackend.LedgerBackend {
+	captiveConfig := config.GenerateCaptiveCoreConfig()
+
+	// Create a new captive core backend
+	backend, err := ledgerbackend.NewCaptive(captiveConfig)
+	logFatalIf(err, "Failed to create captive-core instance")
+
+	var ledgerRange ledgerbackend.Range
+	if config.EndLedger == 0 {
+		ledgerRange = ledgerbackend.UnboundedRange(config.StartLedger)
+	} else {
+		ledgerRange = ledgerbackend.BoundedRange(config.StartLedger, config.EndLedger)
+	}
+
+	err = backend.PrepareRange(ctx, ledgerRange)
+	logFatalIf(err, "Could not prepare captive core ledger backend")
+	return backend
+}
+
+func logFatalIf(err error, message string, args ...interface{}) {
+	if err != nil {
+		logger.WithError(err).Fatalf(message, args...)
+	}
+}