Repository: stellar/go
Pull request #5322 (2024-05-17 10:14:20)

exp/services/ledgerexporter: Add latest uploaded ledger metric
<!-- If you're making a doc PR or something tiny where the below is irrelevant, delete this
template and use a short description, but in your description aim to include both what the
change is, and why it is being made, with enough context for anyone to understand. -->

<details>
  <summary>PR Checklist</summary>
  
### PR Structure

* [ ] This PR has reasonably narrow scope (if not, break it down into smaller PRs).
* [ ] This PR avoids mixing refactoring changes with feature changes (split into two PRs
  otherwise).
* [ ] This PR's title starts with name of package that is most changed in the PR, ex.
  `services/friendbot`, or `all` or `doc` if the changes are broad or impact many
  packages.

### Thoroughness

* [ ] This PR adds tests for the most critical parts of the new functionality or fixes.
* [ ] I've updated any docs ([developer docs](https://developers.stellar.org/api/), `.md`
  files, etc... affected by this change). Take a look in the `docs` folder for a given service,
  like [this one](https://github.com/stellar/go/tree/master/services/horizon/internal/docs).

### Release planning

* [ ] I've updated the relevant CHANGELOG ([here](services/horizon/CHANGELOG.md) for Horizon) if
  needed with deprecations, added features, breaking changes, and DB schema changes.
* [ ] I've decided if this PR requires a new major/minor version according to
  [semver](https://semver.org/), or if it's mainly a patch change. The PR is targeted at the next
  release branch if it's not a patch change.
</details>

### What

Add latest ledger uploaded metric. @urvisavla suggested including this metric in https://github.com/stellar/go/pull/5265 but but I decided to defer until we tried implementing concurrent uploads. After [prototyping](https://github.com/stellar/go/pull/5315) concurrent uploads, it turns out that there are no significant benefits. So, in this PR I have gone back to implement the latest ledger uploaded metric.

### Known limitations

[N/A]



Patch:
diff --git a/exp/services/ledgerexporter/internal/app.go b/exp/services/ledgerexporter/internal/app.go
index fe84022d..e6dde872 100644
--- a/exp/services/ledgerexporter/internal/app.go
+++ b/exp/services/ledgerexporter/internal/app.go
@@ -25,6 +25,15 @@ import (
 const (
 	adminServerReadTimeout     = 5 * time.Second
 	adminServerShutdownTimeout = time.Second * 5
+	// TODO: make this timeout configurable
+	uploadShutdownTimeout = 10 * time.Second
+	// We expect the queue size to rarely exceed 1 or 2 because
+	// upload speeds are expected to be much faster than the rate at which
+	// captive core emits ledgers. However, configuring a higher capacity
+	// than our expectation is useful because if we observe a large queue
+	// size in our metrics that is an indication that uploads to the
+	// data store have degraded
+	uploadQueueCapacity = 128
 )
 
 var (
@@ -116,8 +125,7 @@ func (a *App) init(ctx context.Context) error {
 		return err
 	}
 
-	// TODO: make queue size configurable instead of hard coding it to 1
-	queue := NewUploadQueue(1, registry)
+	queue := NewUploadQueue(uploadQueueCapacity, registry)
 	if a.exportManager, err = NewExportManager(a.config.LedgerBatchConfig, a.ledgerBackend, queue, registry); err != nil {
 		return err
 	}
@@ -190,7 +198,7 @@ func (a *App) Run() {
 	go func() {
 		defer wg.Done()
 
-		err := a.uploader.Run(ctx)
+		err := a.uploader.Run(ctx, uploadShutdownTimeout)
 		if err != nil && !errors.Is(err, context.Canceled) {
 			logger.WithError(err).Error("Error executing Uploader")
 			cancel()
diff --git a/exp/services/ledgerexporter/internal/uploader.go b/exp/services/ledgerexporter/internal/uploader.go
index 94344ee1..78296cde 100644
--- a/exp/services/ledgerexporter/internal/uploader.go
+++ b/exp/services/ledgerexporter/internal/uploader.go
@@ -19,6 +19,7 @@ type Uploader struct {
 	queue                UploadQueue
 	uploadDurationMetric *prometheus.SummaryVec
 	objectSizeMetrics    *prometheus.SummaryVec
+	latestLedgerMetric   prometheus.Gauge
 }
 
 // NewUploader constructs a new Uploader instance
@@ -43,12 +44,17 @@ func NewUploader(
 		},
 		[]string{"ledgers", "already_exists", "compression"},
 	)
-	prometheusRegistry.MustRegister(uploadDurationMetric, objectSizeMetrics)
+	latestLedgerMetric := prometheus.NewGauge(prometheus.GaugeOpts{
+		Namespace: "ledger_exporter", Subsystem: "uploader", Name: "latest_ledger",
+		Help: "sequence number of the latest ledger uploaded",
+	})
+	prometheusRegistry.MustRegister(uploadDurationMetric, objectSizeMetrics, latestLedgerMetric)
 	return Uploader{
 		dataStore:            destination,
 		queue:                queue,
 		uploadDurationMetric: uploadDurationMetric,
 		objectSizeMetrics:    objectSizeMetrics,
+		latestLedgerMetric:   latestLedgerMetric,
 	}
 }
 
@@ -93,6 +99,8 @@ func (u Uploader) Upload(ctx context.Context, metaArchive *LedgerMetaArchive) er
 	if err != nil {
 		return errors.Wrapf(err, "error uploading %s", metaArchive.ObjectKey)
 	}
+
+	logger.Infof("Uploaded %s successfully", metaArchive.ObjectKey)
 	alreadyExists := strconv.FormatBool(!ok)
 
 	u.uploadDurationMetric.With(prometheus.Labels{
@@ -109,22 +117,36 @@ func (u Uploader) Upload(ctx context.Context, metaArchive *LedgerMetaArchive) er
 		"ledgers":        numLedgers,
 		"already_exists": alreadyExists,
 	}).Observe(float64(writerTo.totalCompressed))
+	u.latestLedgerMetric.Set(float64(metaArchive.Data.EndSequence))
 	return nil
 }
 
-// TODO: make it configurable
-var uploaderShutdownWaitTime = 10 * time.Second
-
 // Run starts the uploader, continuously listening for LedgerMetaArchive objects to upload.
-func (u Uploader) Run(ctx context.Context) error {
+func (u Uploader) Run(ctx context.Context, shutdownDelayTime time.Duration) error {
 	uploadCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	go func() {
-		<-ctx.Done()
-		logger.Info("Context done, waiting for remaining uploads to complete...")
-		// wait for a few seconds to upload remaining objects from metaArchiveCh
-		<-time.After(uploaderShutdownWaitTime)
-		logger.Info("Timeout reached, canceling remaining uploads...")
-		cancel()
+		select {
+		case <-uploadCtx.Done():
+			// if uploadCtx is cancelled that means we have exited Run()
+			// and therefore there are no remaining uploads
+			return
+		case <-ctx.Done():
+			logger.Info("Received shutdown signal, waiting for remaining uploads to complete...")
+		}
+
+		select {
+		case <-time.After(shutdownDelayTime):
+			// wait for some time to upload remaining objects from
+			// the upload queue
+			logger.Info("Timeout reached, canceling remaining uploads...")
+			cancel()
+		case <-uploadCtx.Done():
+			// if uploadCtx is cancelled that means we have exited Run()
+			// and therefore there are no remaining uploads
+			return
+		}
 	}()
 
 	for {
@@ -141,6 +163,5 @@ func (u Uploader) Run(ctx context.Context) error {
 		if err = u.Upload(uploadCtx, metaObject); err != nil {
 			return err
 		}
-		logger.Infof("Uploaded %s successfully", metaObject.ObjectKey)
 	}
 }


Test patch:
diff --git a/exp/services/ledgerexporter/internal/uploader_test.go b/exp/services/ledgerexporter/internal/uploader_test.go
index b31397c1..5fbf5e22 100644
--- a/exp/services/ledgerexporter/internal/uploader_test.go
+++ b/exp/services/ledgerexporter/internal/uploader_test.go
@@ -10,14 +10,16 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/suite"
+
 	"github.com/stellar/go/support/compressxdr"
 	"github.com/stellar/go/support/datastore"
 	"github.com/stellar/go/support/errors"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
 )
 
+var testShutdownDelayTime = 300 * time.Millisecond
+
 func TestUploaderSuite(t *testing.T) {
 	suite.Run(t, new(UploaderSuite))
 }
@@ -34,6 +36,10 @@ func (s *UploaderSuite) SetupTest() {
 	s.mockDataStore = datastore.MockDataStore{}
 }
 
+func (s *UploaderSuite) TearDownTest() {
+	s.mockDataStore.AssertExpectations(s.T())
+}
+
 func (s *UploaderSuite) TestUpload() {
 	s.testUpload(false)
 	s.testUpload(true)
@@ -52,13 +58,13 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 		Run(func(args mock.Arguments) {
 			capturedKey = args.Get(1).(string)
 			_, err := args.Get(2).(io.WriterTo).WriteTo(&capturedBuf)
-			require.NoError(s.T(), err)
+			s.Require().NoError(err)
 		}).Return(putOkReturnVal, nil).Once()
 
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
-	require.NoError(s.T(), dataUploader.Upload(context.Background(), archive))
+	s.Require().NoError(dataUploader.Upload(context.Background(), archive))
 
 	expectedCompressedLength := capturedBuf.Len()
 	var decodedArchive LedgerMetaArchive
@@ -66,31 +72,29 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 
 	decoder := xdrDecoder
 	_, err := decoder.ReadFrom(&capturedBuf)
-	require.NoError(s.T(), err)
+	s.Require().NoError(err)
 
 	// require that the decoded data matches the original test data
-	require.Equal(s.T(), key, capturedKey)
-	require.Equal(s.T(), archive.Data, decodedArchive.Data)
+	s.Require().Equal(key, capturedKey)
+	s.Require().Equal(archive.Data, decodedArchive.Data)
 
 	alreadyExists := !putOkReturnVal
 	metric, err := dataUploader.uploadDurationMetric.MetricVec.GetMetricWith(prometheus.Labels{
 		"ledgers":        "100",
 		"already_exists": strconv.FormatBool(alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(1),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
-	require.Positive(s.T(), getMetricValue(metric).GetSummary().GetSampleSum())
+	s.Require().Positive(getMetricValue(metric).GetSummary().GetSampleSum())
 	metric, err = dataUploader.uploadDurationMetric.MetricVec.GetMetricWith(prometheus.Labels{
 		"ledgers":        "100",
 		"already_exists": strconv.FormatBool(!alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(0),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
@@ -100,14 +104,12 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 		"compression":    decoder.Compressor.Name(),
 		"already_exists": strconv.FormatBool(alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(1),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
-	require.Equal(
-		s.T(),
+	s.Require().Equal(
 		float64(expectedCompressedLength),
 		getMetricValue(metric).GetSummary().GetSampleSum(),
 	)
@@ -116,9 +118,8 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 		"compression":    decoder.Compressor.Name(),
 		"already_exists": strconv.FormatBool(!alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(0),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
@@ -128,16 +129,14 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 		"compression":    "none",
 		"already_exists": strconv.FormatBool(alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(1),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
 	uncompressedPayload, err := decodedArchive.Data.MarshalBinary()
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		float64(len(uncompressedPayload)),
 		getMetricValue(metric).GetSummary().GetSampleSum(),
 	)
@@ -146,12 +145,16 @@ func (s *UploaderSuite) testUpload(putOkReturnVal bool) {
 		"compression":    "none",
 		"already_exists": strconv.FormatBool(!alreadyExists),
 	})
-	require.NoError(s.T(), err)
-	require.Equal(
-		s.T(),
+	s.Require().NoError(err)
+	s.Require().Equal(
 		uint64(0),
 		getMetricValue(metric).GetSummary().GetSampleCount(),
 	)
+
+	s.Require().Equal(
+		float64(100),
+		getMetricValue(dataUploader.latestLedgerMetric).GetGauge().GetValue(),
+	)
 }
 
 func (s *UploaderSuite) TestUploadPutError() {
@@ -164,22 +167,21 @@ func (s *UploaderSuite) testUploadPutError(putOkReturnVal bool) {
 	archive := NewLedgerMetaArchive(key, start, end)
 
 	s.mockDataStore.On("PutFileIfNotExists", context.Background(), key,
-		mock.Anything).Return(putOkReturnVal, errors.New("error in PutFileIfNotExists"))
+		mock.Anything).Return(putOkReturnVal, errors.New("error in PutFileIfNotExists")).Once()
 
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
 	err := dataUploader.Upload(context.Background(), archive)
-	require.Equal(s.T(), fmt.Sprintf("error uploading %s: error in PutFileIfNotExists", key), err.Error())
+	s.Require().Equal(fmt.Sprintf("error uploading %s: error in PutFileIfNotExists", key), err.Error())
 
 	for _, alreadyExists := range []string{"true", "false"} {
 		metric, err := dataUploader.uploadDurationMetric.MetricVec.GetMetricWith(prometheus.Labels{
 			"ledgers":        "100",
 			"already_exists": alreadyExists,
 		})
-		require.NoError(s.T(), err)
-		require.Equal(
-			s.T(),
+		s.Require().NoError(err)
+		s.Require().Equal(
 			uint64(0),
 			getMetricValue(metric).GetSummary().GetSampleCount(),
 		)
@@ -190,61 +192,90 @@ func (s *UploaderSuite) testUploadPutError(putOkReturnVal bool) {
 				"compression":    compression,
 				"already_exists": alreadyExists,
 			})
-			require.NoError(s.T(), err)
-			require.Equal(
-				s.T(),
+			s.Require().NoError(err)
+			s.Require().Equal(
 				uint64(0),
 				getMetricValue(metric).GetSummary().GetSampleCount(),
 			)
 		}
+
+		s.Require().Equal(
+			float64(0),
+			getMetricValue(dataUploader.latestLedgerMetric).GetGauge().GetValue(),
+		)
 	}
 }
 
-func (s *UploaderSuite) TestRunChannelClose() {
-	s.mockDataStore.On("PutFileIfNotExists", mock.Anything,
-		mock.Anything, mock.Anything).Return(true, nil)
+func (s *UploaderSuite) TestRunUntilQueueClose() {
+	var prev *mock.Call
+	for i := 1; i <= 100; i++ {
+		key := fmt.Sprintf("test-%d", i)
+		cur := s.mockDataStore.On("PutFileIfNotExists", mock.Anything,
+			key, mock.Anything).Return(true, nil).Once()
+		if prev != nil {
+			cur.NotBefore(prev)
+		}
+		prev = cur
+	}
 
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 	go func() {
-		key, start, end := "test", uint32(1), uint32(100)
-		for i := start; i <= end; i++ {
-			s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive(key, i, i)))
+		for i := uint32(1); i <= uint32(100); i++ {
+			key := fmt.Sprintf("test-%d", i)
+			s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive(key, i, i)))
 		}
-		<-time.After(time.Second * 2)
 		queue.Close()
 	}()
 
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
-	require.NoError(s.T(), dataUploader.Run(context.Background()))
+	s.Require().NoError(dataUploader.Run(context.Background(), testShutdownDelayTime))
+
+	s.Require().Equal(
+		float64(100),
+		getMetricValue(dataUploader.latestLedgerMetric).GetGauge().GetValue(),
+	)
 }
 
 func (s *UploaderSuite) TestRunContextCancel() {
-	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 	registry := prometheus.NewRegistry()
 	queue := NewUploadQueue(1, registry)
 
-	s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
+	first := s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test", mock.Anything).
+		Return(true, nil).Once().Run(func(args mock.Arguments) {
+		cancel()
+	})
+	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test1", mock.Anything).
+		Return(true, nil).Once().NotBefore(first).Run(func(args mock.Arguments) {
+		ctxArg := args.Get(0).(context.Context)
+		s.Require().NoError(ctxArg.Err())
+	})
 
 	go func() {
-		<-time.After(time.Second * 2)
-		cancel()
+		s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
+		s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test1", 2, 2)))
 	}()
 
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
-	require.EqualError(s.T(), dataUploader.Run(ctx), "context canceled")
+	s.Require().EqualError(dataUploader.Run(ctx, testShutdownDelayTime), "context canceled")
+	s.Require().Equal(
+		float64(2),
+		getMetricValue(dataUploader.latestLedgerMetric).GetGauge().GetValue(),
+	)
 }
 
 func (s *UploaderSuite) TestRunUploadError() {
 	registry := prometheus.NewRegistry()
-	queue := NewUploadQueue(1, registry)
+	queue := NewUploadQueue(2, registry)
+
+	s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
+	s.Require().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test1", 2, 2)))
 
-	s.Assert().NoError(queue.Enqueue(s.ctx, NewLedgerMetaArchive("test", 1, 1)))
 	s.mockDataStore.On("PutFileIfNotExists", mock.Anything, "test",
-		mock.Anything).Return(false, errors.New("Put error"))
+		mock.Anything).Return(false, errors.New("Put error")).Once()
 
 	dataUploader := NewUploader(&s.mockDataStore, queue, registry)
-	err := dataUploader.Run(context.Background())
-	require.Equal(s.T(), "error uploading test: Put error", err.Error())
+	err := dataUploader.Run(context.Background(), testShutdownDelayTime)
+	s.Require().Equal("error uploading test: Put error", err.Error())
 }