services/horizon/cmd: Run metrics server when invoking horizon ingest commands (#5816)

Author: tamirms

services/horizon/cmd/db.go

@@ -211,21 +211,23 @@ func runDBReingestRange(ledgerRanges []history.LedgerRange, reingestForce bool,
 	}
 	defer system.Shutdown()
 
-	err = system.ReingestRange(ledgerRanges, reingestForce, true)
-	if err != nil {
-		if _, ok := errors.Cause(err).(ingest.ErrReingestRangeConflict); ok {
-			return fmt.Errorf(`The range you have provided overlaps with Horizon's most recently ingested ledger.
+	return runWithMetrics(config.AdminPort, system, func() error {
+		err = system.ReingestRange(ledgerRanges, reingestForce, true)
+		if err != nil {
+			if _, ok := errors.Cause(err).(ingest.ErrReingestRangeConflict); ok {
+				return fmt.Errorf(`The range you have provided overlaps with Horizon's most recently ingested ledger.
 It is not possible to run the reingest command on this range in parallel with
 Horizon's ingestion system.
 Either reduce the range so that it doesn't overlap with Horizon's ingestion system,
 or, use the force flag to ensure that Horizon's ingestion system is blocked until
 the reingest command completes.`)
-		}
+			}
 
-		return err
-	}
-	hlog.Info("Range run successfully!")
-	return nil
+			return err
+		}
+		hlog.Info("Range run successfully!")
+		return nil
+	})
 }
 
 func runDBDetectGaps(config horizon.Config) ([]history.LedgerRange, error) {

services/horizon/cmd/ingest.go

@@ -8,12 +8,16 @@ import (
 	_ "net/http/pprof"
 	"time"
 
+	"github.com/go-chi/chi"
+	chimiddleware "github.com/go-chi/chi/middleware"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 
 	"github.com/stellar/go/historyarchive"
 	horizon "github.com/stellar/go/services/horizon/internal"
 	"github.com/stellar/go/services/horizon/internal/db2/history"
+	"github.com/stellar/go/services/horizon/internal/httpx"
 	"github.com/stellar/go/services/horizon/internal/ingest"
 	"github.com/stellar/go/support/config"
 	support "github.com/stellar/go/support/config"
@@ -23,7 +27,7 @@ import (
 
 var ingestBuildStateSequence uint32
 var ingestBuildStateSkipChecks bool
-var ingestVerifyFrom, ingestVerifyTo, ingestVerifyDebugServerPort uint32
+var ingestVerifyFrom, ingestVerifyTo uint32
 var ingestVerifyState bool
 var ingestVerifyStorageBackendConfigPath string
 var ingestVerifyLedgerBackendType ingest.LedgerBackendType
@@ -73,14 +77,6 @@ var ingestVerifyRangeCmdOpts = support.ConfigOptions{
 		FlagDefault: false,
 		Usage:       "[optional] verifies state at the last ledger of the range when true",
 	},
-	{
-		Name:        "debug-server-port",
-		ConfigKey:   &ingestVerifyDebugServerPort,
-		OptType:     types.Uint32,
-		Required:    false,
-		FlagDefault: uint32(0),
-		Usage:       "[optional] opens a net/http/pprof server at given port",
-	},
 	generateLedgerBackendOpt(&ingestVerifyLedgerBackendType),
 	generateDatastoreConfigOpt(&ingestVerifyStorageBackendConfigPath),
 }
@@ -159,19 +155,6 @@ func DefineIngestCommands(rootCmd *cobra.Command, horizonConfig *horizon.Config,
 				return err
 			}
 
-			if ingestVerifyDebugServerPort != 0 {
-				go func() {
-					log.Infof("Starting debug server at: %d", ingestVerifyDebugServerPort)
-					err := http.ListenAndServe(
-						fmt.Sprintf("localhost:%d", ingestVerifyDebugServerPort),
-						nil,
-					)
-					if err != nil {
-						log.Error(err)
-					}
-				}()
-			}
-
 			mngr := historyarchive.NewCheckpointManager(horizonConfig.CheckpointFrequency)
 			if !mngr.IsCheckpoint(ingestVerifyFrom) && ingestVerifyFrom != 1 {
 				return fmt.Errorf("`--from` must be a checkpoint ledger")
@@ -249,17 +232,12 @@ func DefineIngestCommands(rootCmd *cobra.Command, horizonConfig *horizon.Config,
 			if err != nil {
 				return err
 			}
-
-			err = system.StressTest(
-				stressTestNumTransactions,
-				stressTestChangesPerTransaction,
-			)
-			if err != nil {
-				return err
-			}
-
-			log.Info("Stress test completed successfully!")
-			return nil
+			return runWithMetrics(horizonConfig.AdminPort, system, func() error {
+				return system.StressTest(
+					stressTestNumTransactions,
+					stressTestChangesPerTransaction,
+				)
+			})
 		},
 	}
 
@@ -365,16 +343,12 @@ func DefineIngestCommands(rootCmd *cobra.Command, horizonConfig *horizon.Config,
 				return err
 			}
 
-			err = system.BuildState(
-				ingestBuildStateSequence,
-				ingestBuildStateSkipChecks,
-			)
-			if err != nil {
-				return err
-			}
-
-			log.Info("State built successfully!")
-			return nil
+			return runWithMetrics(horizonConfig.AdminPort, system, func() error {
+				return system.BuildState(
+					ingestBuildStateSequence,
+					ingestBuildStateSkipChecks,
+				)
+			})
 		},
 	}
 
@@ -444,11 +418,13 @@ func DefineIngestCommands(rootCmd *cobra.Command, horizonConfig *horizon.Config,
 				return err
 			}
 
-			return system.LoadTest(
-				ingestionLoadTestLedgersPath,
-				ingestionLoadTestCloseDuration,
-				ingestionLoadTestFixturesPath,
-			)
+			return runWithMetrics(horizonConfig.AdminPort, system, func() error {
+				return system.LoadTest(
+					ingestionLoadTestLedgersPath,
+					ingestionLoadTestCloseDuration,
+					ingestionLoadTestFixturesPath,
+				)
+			})
 		},
 	}
 
@@ -496,6 +472,46 @@ func DefineIngestCommands(rootCmd *cobra.Command, horizonConfig *horizon.Config,
 	)
 }
 
+func runWithMetrics(metricsPort uint, system ingest.System, f func() error) error {
+	if metricsPort != 0 {
+		log.Infof("Starting metrics server at: %d", metricsPort)
+		mux := chi.NewMux()
+		mux.Use(chimiddleware.StripSlashes)
+		mux.Use(chimiddleware.RequestID)
+		mux.Use(chimiddleware.RequestLogger(&chimiddleware.DefaultLogFormatter{
+			Logger:  log.DefaultLogger,
+			NoColor: true,
+		}))
+		registry := prometheus.NewRegistry()
+		system.RegisterMetrics(registry)
+		httpx.AddMetricRoutes(mux, registry)
+		metricsServer := &http.Server{
+			Addr:        fmt.Sprintf(":%d", metricsPort),
+			Handler:     mux,
+			ReadTimeout: 5 * time.Second,
+		}
+		go func() {
+			if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+				log.Fatalf("error running metrics server: %v", err)
+			}
+		}()
+		defer func() {
+			log.Info("Waiting for metrics to be flushed")
+			// by default, the scrape_interval for prometheus is 1 minute
+			// so if we sleep for 1.5 minutes we ensure that all remaining metrics
+			// will be picked up by the prometheus scraper
+			time.Sleep(time.Minute + time.Second*30)
+			log.Info("Shutting down metrics server...")
+			if err := metricsServer.Shutdown(context.Background()); err != nil {
+				log.Warnf("error shutting down metrics server: %v", err)
+			}
+		}()
+	} else {
+		log.Info("Metrics server disabled")
+	}
+	return f()
+}
+
 func init() {
 	DefineIngestCommands(RootCmd, globalConfig, globalFlags)
 }
@@ -525,11 +541,13 @@ func processVerifyRange(horizonConfig *horizon.Config, horizonFlags config.Confi
 		return err
 	}
 
-	return system.VerifyRange(
-		ingestVerifyFrom,
-		ingestVerifyTo,
-		ingestVerifyState,
-	)
+	return runWithMetrics(horizonConfig.AdminPort, system, func() error {
+		return system.VerifyRange(
+			ingestVerifyFrom,
+			ingestVerifyTo,
+			ingestVerifyState,
+		)
+	})
 }
 
 // generateDatastoreConfigOpt returns a *support.ConfigOption for the datastore-config flag

services/horizon/internal/httpx/router.go

@@ -394,9 +394,7 @@ func (r *Router) addRoutes(config *RouterConfig, rateLimiter *throttled.HTTPRate
 		w.Header().Set("Content-Type", "application/openapi+yaml")
 		w.Write(p)
 	})
-	r.Internal.Get("/metrics", promhttp.HandlerFor(config.PrometheusRegistry, promhttp.HandlerOpts{}).ServeHTTP)
-	r.Internal.Get("/debug/pprof/heap", pprof.Index)
-	r.Internal.Get("/debug/pprof/profile", pprof.Profile)
+	AddMetricRoutes(r.Internal, config.PrometheusRegistry)
 	r.Internal.Route("/ingestion/filters", func(r chi.Router) {
 		handler := actions.FilterConfigHandler{}
 		r.With(historyMiddleware).Put("/asset", handler.UpdateAssetConfig)
@@ -405,3 +403,9 @@ func (r *Router) addRoutes(config *RouterConfig, rateLimiter *throttled.HTTPRate
 		r.With(historyMiddleware).Get("/account", handler.GetAccountConfig)
 	})
 }
+
+func AddMetricRoutes(mux *chi.Mux, metrics *prometheus.Registry) {
+	mux.Get("/metrics", promhttp.HandlerFor(metrics, promhttp.HandlerOpts{}).ServeHTTP)
+	mux.Get("/debug/pprof/heap", pprof.Index)
+	mux.Get("/debug/pprof/profile", pprof.Profile)
+}