add(monitoring): use constants, add v2 exporter monitoring, monitor caller

Author: invis-bitfly

backend/cmd/exporter/main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/commons/version"
 	"github.com/gobitfly/beaconchain/pkg/exporter/modules"
 	"github.com/gobitfly/beaconchain/pkg/exporter/services"
+	"github.com/gobitfly/beaconchain/pkg/monitoring"
 )
 
 func Run() {
@@ -136,6 +137,10 @@ func Run() {
 
 	wg.Wait()
 
+	// enable light-weight db connection monitoring
+	monitoring.Init(false)
+	monitoring.Start()
+
 	if utils.Config.TieredCacheProvider != "redis" {
 		log.Fatal(fmt.Errorf("no cache provider set, please set TierdCacheProvider (example redis)"), "", 0)
 	}

backend/go.mod

@@ -52,7 +52,6 @@ require (
 	github.com/juliangruber/go-intersect v1.1.0
 	github.com/jung-kurt/gofpdf v1.16.2
 	github.com/k3a/html2text v1.2.1
-	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/klauspost/compress v1.17.6
 	github.com/klauspost/pgzip v1.2.6
 	github.com/lib/pq v1.10.9
@@ -66,6 +65,7 @@ require (
 	github.com/prysmaticlabs/go-ssz v0.0.0-20210121151755-f6208871c388
 	github.com/rocket-pool/rocketpool-go v1.8.4-0.20241009143357-7b6894d57365
 	github.com/rocket-pool/smartnode v1.14.1
+	github.com/sethvargo/go-envconfig v1.1.0
 	github.com/shopspring/decimal v1.4.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
@@ -234,7 +234,6 @@ require (
 	github.com/sanity-io/litter v1.5.5 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
 	github.com/sergi/go-diff v1.2.0 // indirect
-	github.com/sethvargo/go-envconfig v1.1.0 // indirect
 	github.com/sethvargo/go-retry v0.2.4 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect

backend/go.sum

@@ -601,8 +601,6 @@ github.com/k3a/html2text v1.2.1 h1:nvnKgBvBR/myqrwfLuiqecUtaK1lB9hGziIJKatNFVY=
 github.com/k3a/html2text v1.2.1/go.mod h1:ieEXykM67iT8lTvEWBh6fhpH4B23kB9OMKPdIBmgUqA=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
-github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8=
-github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=

backend/pkg/api/data_access/healthz.go

@@ -111,7 +111,7 @@ func (d *DataAccessService) GetHealthz(ctx context.Context, showAll bool) types.
 	response.Reports = make(map[string][]types.HealthzResult)
 	response.ReportingUUID = utils.GetUUID()
 	response.DeploymentType = utils.Config.DeploymentType
-	err := db.ClickHouseReader.SelectContext(ctx, &results, query, ch.Named("deployment_type", utils.Config.DeploymentType), ch.Named("clean_shutdown_event_id", constants.CleanShutdownEvent))
+	err := db.ClickHouseReader.SelectContext(ctx, &results, query, ch.Named("deployment_type", utils.Config.DeploymentType), ch.Named("clean_shutdown_event_id", string(constants.Event_MonitoringCleanShutdown)))
 	if err != nil {
 		response.Reports["response_error"] = []types.HealthzResult{
 			{
@@ -125,25 +125,16 @@ func (d *DataAccessService) GetHealthz(ctx context.Context, showAll bool) types.
 		return response
 	}
 
-	mustExist := []string{
-		"ch_rolling_1h",
-		"ch_rolling_24h",
-		"ch_rolling_7d",
-		"ch_rolling_30d",
-		"ch_rolling_90d",
-		"ch_rolling_total",
-		"ch_dashboard_epoch",
-		"api_service_avg_efficiency",
-		"api_service_validator_mapping",
-		"api_service_slot_viz",
-		"monitoring_timeouts",
-	}
 	for _, result := range results {
 		response.Reports[result.EventId] = append(response.Reports[result.EventId], result)
 	}
-	for _, id := range mustExist {
-		if _, ok := response.Reports[id]; !ok {
-			response.Reports[id] = []types.HealthzResult{
+	requiredEvents := constants.RequiredEvents
+	if utils.Config.DeploymentType == "production" {
+		requiredEvents = append(requiredEvents, constants.ProductionRequiredEvents...)
+	}
+	for _, id := range requiredEvents {
+		if _, ok := response.Reports[string(id)]; !ok {
+			response.Reports[string(id)] = []types.HealthzResult{
 				{
 					Status: constants.Failure,
 					Result: []map[string]string{

backend/pkg/api/services/service_average_network_efficiency.go

@@ -27,7 +27,7 @@ func (s *Services) startEfficiencyDataService(wg *sync.WaitGroup) {
 	for {
 		startTime := time.Now()
 		delay := time.Duration(utils.Config.Chain.ClConfig.SlotsPerEpoch*utils.Config.Chain.ClConfig.SecondsPerSlot) * time.Second
-		r := services.NewStatusReport("api_service_avg_efficiency", constants.Default, delay)
+		r := services.NewStatusReport(constants.Event_ApiServiceAvgEfficiency, constants.Default, delay)
 		r(constants.Running, nil)
 		err := s.updateEfficiencyData() // TODO: only update data if something has changed (new head epoch)
 		if err != nil {

backend/pkg/api/services/service_slot_viz.go

@@ -32,7 +32,7 @@ func (s *Services) startSlotVizDataService(wg *sync.WaitGroup) {
 	for {
 		startTime := time.Now()
 		delay := time.Duration(utils.Config.Chain.ClConfig.SecondsPerSlot) * time.Second
-		r := services.NewStatusReport("api_service_slot_viz", constants.Default, delay)
+		r := services.NewStatusReport(constants.Event_ApiServiceSlotViz, constants.Default, delay)
 		r(constants.Running, nil)
 		err := s.updateSlotVizData() // TODO: only update data if something has changed (new head slot or new head epoch)
 		if err != nil {

backend/pkg/api/services/service_validator_mapping.go

@@ -41,7 +41,7 @@ func (s *Services) startIndexMappingService(wg *sync.WaitGroup) {
 		startTime := time.Now()
 		delay := time.Duration(utils.Config.Chain.ClConfig.SecondsPerSlot) * time.Second
 		err = nil // clear error
-		r := services.NewStatusReport("api_service_validator_mapping", constants.Default, delay)
+		r := services.NewStatusReport(constants.Event_ApiServiceValidatorMapping, constants.Default, delay)
 		r(constants.Running, nil)
 		latestEpoch := cache.LatestEpoch.Get()
 		if currentValidatorMapping.Load() == nil || latestEpoch != lastEpochUpdate {

backend/pkg/exporter/modules/base.go

@@ -11,13 +11,16 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	"github.com/gobitfly/beaconchain/pkg/consapi"
 	"github.com/gobitfly/beaconchain/pkg/consapi/types"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/constants"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/services"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
 )
 
 type ModuleInterface interface {
 	Init() error
 	GetName() string // Used for logging
+	GetMonitoringEventId() constants.Event
 
 	OnHead(*types.StandardEventHeadResponse) error // !Do not block in this functions for an extended period of time!
 
@@ -63,7 +66,6 @@ func StartAll(context ModuleContext, modules []ModuleInterface, justV2 bool) {
 		log.Error(err, "beacon-node seems to be unavailable", 0)
 		time.Sleep(time.Second * 10)
 	}
-
 	// start subscription modules
 	startSubscriptionModules(&context, modules)
 }
@@ -146,15 +148,20 @@ func notifyAllModules(goPool *errgroup.Group, modules []ModuleInterface, f func(
 	for _, module := range modules {
 		module := module
 		goPool.Go(func() error {
+			start := time.Now()
+			r := services.NewStatusReport(module.GetMonitoringEventId(), constants.Default, constants.Default)
+			r(constants.Running, nil)
 			err := f(module)
 			if err != nil {
 				log.Error(err, fmt.Sprintf("error in module %s", module.GetName()), 0)
+				r(constants.Failure, map[string]string{"error": err.Error()})
+				return nil // return never gets caught anywhere? lets not risk a memory leak and instead return nil
 			}
+			r(constants.Success, map[string]string{"took_raw": fmt.Sprintf("%v", time.Since(start).Milliseconds())})
 			return nil
 		})
 	}
 }
-
 func GetModuleContext() (ModuleContext, error) {
 	cl := consapi.NewClient("http://" + utils.Config.Indexer.Node.Host + ":" + utils.Config.Indexer.Node.Port)
 

backend/pkg/exporter/modules/dashboard_data.go

@@ -19,6 +19,7 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/consapi/network"
 	constypes "github.com/gobitfly/beaconchain/pkg/consapi/types"
 	edb "github.com/gobitfly/beaconchain/pkg/exporter/db"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/constants"
 	"github.com/pkg/errors"
 	"github.com/prysmaticlabs/go-bitfield"
 	"golang.org/x/sync/errgroup"
@@ -992,6 +993,10 @@ func (d *dashboardData) GetName() string {
 	return "Dashboard-Data"
 }
 
+func (d *dashboardData) GetMonitoringEventId() constants.Event {
+	return constants.Event_ExporterModuleDashboardData
+}
+
 func (d *dashboardData) OnHead(event *constypes.StandardEventHeadResponse) error {
 	return nil
 }

backend/pkg/exporter/modules/execution_deposits_exporter.go

@@ -31,6 +31,7 @@ import (
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
 	constypes "github.com/gobitfly/beaconchain/pkg/consapi/types"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/constants"
 )
 
 // if we ever end up in a situation where we possibly have gaps in the data remember: the merkletree_index is unique.
@@ -155,6 +156,10 @@ func (d *executionDepositsExporter) GetName() string {
 	return "ExecutionDeposits-Exporter"
 }
 
+func (d *executionDepositsExporter) GetMonitoringEventId() constants.Event {
+	return constants.Event_ExporterModuleELDepositsExporter
+}
+
 func (d *executionDepositsExporter) OnChainReorg(event *constypes.StandardEventChainReorg) (err error) {
 	return nil // nop
 }