services/horizon: Add latest ledger closed_at metrics (#3281)

This commit adds:
* `history_latest_ledger_closed_at` to root resource that determines the
latest ingested ledger close time,
* `horizon_history_latest_ledger_closed_ago_seconds` to `/metrics` that
return time in seconds since the latest ingested ledger close time.

`core_latest_ledger - history_latest_ledger` doesn't always determine if
Horizon is fully sync with the network. It's possible that underlying
Core instance is behind or stuck.

protocols/horizon/main.go

@@ -309,15 +309,16 @@ type Root struct {
 		Transactions        hal.Link  `json:"transactions"`
 	} `json:"_links"`
 
-	HorizonVersion               string `json:"horizon_version"`
-	StellarCoreVersion           string `json:"core_version"`
-	IngestSequence               uint32 `json:"ingest_latest_ledger"`
-	HorizonSequence              int32  `json:"history_latest_ledger"`
-	HistoryElderSequence         int32  `json:"history_elder_ledger"`
-	CoreSequence                 int32  `json:"core_latest_ledger"`
-	NetworkPassphrase            string `json:"network_passphrase"`
-	CurrentProtocolVersion       int32  `json:"current_protocol_version"`
-	CoreSupportedProtocolVersion int32  `json:"core_supported_protocol_version"`
+	HorizonVersion               string    `json:"horizon_version"`
+	StellarCoreVersion           string    `json:"core_version"`
+	IngestSequence               uint32    `json:"ingest_latest_ledger"`
+	HorizonSequence              int32     `json:"history_latest_ledger"`
+	HorizonLatestClosedAt        time.Time `json:"history_latest_ledger_closed_at"`
+	HistoryElderSequence         int32     `json:"history_elder_ledger"`
+	CoreSequence                 int32     `json:"core_latest_ledger"`
+	NetworkPassphrase            string    `json:"network_passphrase"`
+	CurrentProtocolVersion       int32     `json:"current_protocol_version"`
+	CoreSupportedProtocolVersion int32     `json:"core_supported_protocol_version"`
 }
 
 // Signer represents one of an account's signers.

services/horizon/internal/app.go

@@ -68,17 +68,18 @@ type App struct {
 	ledgerState     *ledger.State
 
 	// metrics
-	prometheusRegistry         *prometheus.Registry
-	buildInfoGauge             *prometheus.GaugeVec
-	ingestingGauge             prometheus.Gauge
-	historyLatestLedgerCounter prometheus.CounterFunc
-	historyElderLedgerCounter  prometheus.CounterFunc
-	dbMaxOpenConnectionsGauge  prometheus.GaugeFunc
-	dbOpenConnectionsGauge     prometheus.GaugeFunc
-	dbInUseConnectionsGauge    prometheus.GaugeFunc
-	dbWaitCountCounter         prometheus.CounterFunc
-	dbWaitDurationCounter      prometheus.CounterFunc
-	coreLatestLedgerCounter    prometheus.CounterFunc
+	prometheusRegistry                *prometheus.Registry
+	buildInfoGauge                    *prometheus.GaugeVec
+	ingestingGauge                    prometheus.Gauge
+	historyLatestLedgerCounter        prometheus.CounterFunc
+	historyLatestLedgerClosedAgoGauge prometheus.GaugeFunc
+	historyElderLedgerCounter         prometheus.CounterFunc
+	dbMaxOpenConnectionsGauge         prometheus.GaugeFunc
+	dbOpenConnectionsGauge            prometheus.GaugeFunc
+	dbInUseConnectionsGauge           prometheus.GaugeFunc
+	dbWaitCountCounter                prometheus.CounterFunc
+	dbWaitDurationCounter             prometheus.CounterFunc
+	coreLatestLedgerCounter           prometheus.CounterFunc
 }
 
 func (a *App) GetCoreSettings() actions.CoreSettings {
@@ -208,7 +209,8 @@ func (a *App) UpdateLedgerState() {
 	}
 	next.CoreLatest = int32(coreInfo.Info.Ledger.Num)
 
-	err = a.HistoryQ().LatestLedger(&next.HistoryLatest)
+	next.HistoryLatest, next.HistoryLatestClosedAt, err =
+		a.HistoryQ().LatestLedgerSequenceClosedAt()
 	if err != nil {
 		logErr(err, "failed to load the latest known ledger state from history DB")
 		return

services/horizon/internal/app_test.go

@@ -45,10 +45,12 @@ func TestMetrics(t *testing.T) {
 	ht.Assert.Equal(200, w.Code)
 
 	hl := ht.App.historyLatestLedgerCounter
+	hlc := ht.App.historyLatestLedgerClosedAgoGauge
 	he := ht.App.historyElderLedgerCounter
 	cl := ht.App.coreLatestLedgerCounter
 
 	ht.Require.EqualValues(3, getMetricValue(hl).GetCounter().GetValue())
+	ht.Require.Less(float64(1000), getMetricValue(hlc).GetGauge().GetValue())
 	ht.Require.EqualValues(1, getMetricValue(he).GetCounter().GetValue())
 	ht.Require.EqualValues(64, getMetricValue(cl).GetCounter().GetValue())
 }

services/horizon/internal/db2/history/main.go

@@ -719,6 +719,21 @@ func (q *Q) LatestLedger(dest interface{}) error {
 	return q.GetRaw(dest, `SELECT COALESCE(MAX(sequence), 0) FROM history_ledgers`)
 }
 
+// LatestLedgerSequenceClosedAt loads the latest known ledger sequence and close time,
+// returns empty values if no ledgers in a DB.
+func (q *Q) LatestLedgerSequenceClosedAt() (int32, time.Time, error) {
+	ledger := struct {
+		Sequence int32     `db:"sequence"`
+		ClosedAt time.Time `db:"closed_at"`
+	}{}
+	err := q.GetRaw(&ledger, `SELECT sequence, closed_at FROM history_ledgers ORDER BY sequence DESC LIMIT 1`)
+	if err == sql.ErrNoRows {
+		// Will return empty values
+		return ledger.Sequence, ledger.ClosedAt, nil
+	}
+	return ledger.Sequence, ledger.ClosedAt, err
+}
+
 // LatestLedgerBaseFeeAndSequence loads the latest known ledger's base fee and
 // sequence number.
 func (q *Q) LatestLedgerBaseFeeAndSequence(dest interface{}) error {

services/horizon/internal/db2/history/main_test.go

@@ -2,6 +2,7 @@ package history
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stellar/go/services/horizon/internal/test"
 )
@@ -20,6 +21,27 @@ func TestLatestLedger(t *testing.T) {
 	}
 }
 
+func TestLatestLedgerSequenceClosedAt(t *testing.T) {
+	tt := test.Start(t)
+	tt.Scenario("base")
+	defer tt.Finish()
+	q := &Q{tt.HorizonSession()}
+
+	sequence, closedAt, err := q.LatestLedgerSequenceClosedAt()
+	if tt.Assert.NoError(err) {
+		tt.Assert.Equal(int32(3), sequence)
+		tt.Assert.Equal("2019-10-31T13:19:46Z", closedAt.Format(time.RFC3339))
+	}
+
+	test.ResetHorizonDB(t, tt.HorizonDB)
+
+	sequence, closedAt, err = q.LatestLedgerSequenceClosedAt()
+	if tt.Assert.NoError(err) {
+		tt.Assert.Equal(int32(0), sequence)
+		tt.Assert.Equal("0001-01-01T00:00:00Z", closedAt.Format(time.RFC3339))
+	}
+}
+
 func TestGetLatestLedgerEmptyDB(t *testing.T) {
 	tt := test.Start(t)
 	defer tt.Finish()

services/horizon/internal/init.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/http"
 	"runtime"
+	"time"
 
 	"github.com/getsentry/raven-go"
 	"github.com/prometheus/client_golang/prometheus"
@@ -149,6 +150,18 @@ func initDbMetrics(app *App) {
 	)
 	app.prometheusRegistry.MustRegister(app.historyLatestLedgerCounter)
 
+	app.historyLatestLedgerClosedAgoGauge = prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Namespace: "horizon", Subsystem: "history", Name: "latest_ledger_closed_ago_seconds",
+			Help: "seconds since the close of the last ingested ledger",
+		},
+		func() float64 {
+			ls := app.ledgerState.CurrentStatus()
+			return time.Since(ls.HistoryLatestClosedAt).Seconds()
+		},
+	)
+	app.prometheusRegistry.MustRegister(app.historyLatestLedgerClosedAgoGauge)
+
 	app.historyElderLedgerCounter = prometheus.NewCounterFunc(
 		prometheus.CounterOpts{Namespace: "horizon", Subsystem: "history", Name: "elder_ledger"},
 		func() float64 {

services/horizon/internal/ledger/main.go

@@ -7,15 +7,17 @@ package ledger
 
 import (
 	"sync"
+	"time"
 )
 
 // Status represents a snapshot of both horizon's and stellar-core's view of the
 // ledger.
 type Status struct {
-	CoreLatest       int32  `db:"core_latest"`
-	HistoryLatest    int32  `db:"history_latest"`
-	HistoryElder     int32  `db:"history_elder"`
-	ExpHistoryLatest uint32 `db:"exp_history_latest"`
+	CoreLatest            int32     `db:"core_latest"`
+	HistoryLatest         int32     `db:"history_latest"`
+	HistoryLatestClosedAt time.Time `db:"history_latest_closed_at"`
+	HistoryElder          int32     `db:"history_elder"`
+	ExpHistoryLatest      uint32    `db:"exp_history_latest"`
 }
 
 // State is an in-memory data structure which holds a snapshot of both

services/horizon/internal/resourceadapter/root.go

@@ -24,6 +24,7 @@ func PopulateRoot(
 ) {
 	dest.IngestSequence = ledgerState.ExpHistoryLatest
 	dest.HorizonSequence = ledgerState.HistoryLatest
+	dest.HorizonLatestClosedAt = ledgerState.HistoryLatestClosedAt
 	dest.HistoryElderSequence = ledgerState.HistoryElder
 	dest.CoreSequence = ledgerState.CoreLatest
 	dest.HorizonVersion = hVersion