Merge pull request #846 from gobitfly/BEDS-306/clean_shutdown_event

Beds 306/clean shutdown event

backend/cmd/api/main.go

@@ -52,7 +52,6 @@ func Run() {
 		// enable light-weight db connection monitoring
 		monitoring.Init(false)
 		monitoring.Start()
-		defer monitoring.Stop()
 	}
 
 	var dataAccessor dataaccess.DataAccessor
@@ -98,7 +97,7 @@ func Run() {
 	}()
 
 	utils.WaitForCtrlC()
-
+	monitoring.Stop() // this will emit a clean shutdown event
 	log.Info("shutting down server")
 	if srv != nil {
 		shutDownCtx, cancelShutDownCtx := context.WithTimeout(context.Background(), 10*time.Second)

backend/cmd/monitoring/main.go

@@ -55,8 +55,8 @@ func Run() {
 
 	monitoring.Init(true)
 	monitoring.Start()
-	defer monitoring.Stop()
 
 	// gotta wait forever
 	utils.WaitForCtrlC()
+	monitoring.Stop()
 }

backend/pkg/api/data_access/healthz.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"slices"
 
+	ch "github.com/ClickHouse/clickhouse-go/v2"
 	"github.com/gobitfly/beaconchain/pkg/api/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
 	"github.com/gobitfly/beaconchain/pkg/commons/log"
@@ -19,7 +20,17 @@ func (d *DataAccessService) GetHealthz(ctx context.Context, showAll bool) types.
 	var results []types.HealthzResult
 	var response types.HealthzData
 	query := `
-		with active_reports as (
+		with clean_shutdown_events as (
+			SELECT
+				emitter,
+				toNullable(inserted_at) as inserted_at
+			FROM
+				status_reports
+			WHERE
+				deployment_type = {deployment_type:String}
+				AND inserted_at >= now() - interval 1 days
+				AND event_id = {clean_shutdown_event_id:String}
+		), active_reports as (
 			SELECT
 				event_id,
 				emitter,
@@ -31,7 +42,8 @@ func (d *DataAccessService) GetHealthz(ctx context.Context, showAll bool) types.
 				status,
 				metadata
 			FROM status_reports
-			WHERE expires_at > now() and deployment_type = ?
+			LEFT JOIN clean_shutdown_events cse ON status_reports.emitter = clean_shutdown_events.emitter
+			WHERE expires_at > now() and deployment_type = {deployment_type:String} and (status_reports.inserted_at < cse.inserted_at or cse.inserted_at is null)
 			ORDER BY
 				event_id ASC,
 				emitter ASC,
@@ -99,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, 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))
 	if err != nil {
 		response.Reports["response_error"] = []types.HealthzResult{
 			{

backend/pkg/commons/log/log.go

@@ -36,12 +36,7 @@ func Infof(format string, args ...interface{}) {
 }
 
 func InfoWithFields(additionalInfos Fields, msg string) {
-	logFields := logrus.NewEntry(logrus.New())
-	for name, info := range additionalInfos {
-		logFields = logFields.WithField(name, info)
-	}
-
-	logFields.Info(msg)
+	logrus.WithFields(additionalInfos).Info(msg)
 }
 
 func Warn(args ...interface{}) {
@@ -53,34 +48,19 @@ func Warnf(format string, args ...interface{}) {
 }
 
 func WarnWithFields(additionalInfos Fields, msg string) {
-	logFields := logrus.NewEntry(logrus.New())
-	for name, info := range additionalInfos {
-		logFields = logFields.WithField(name, info)
-	}
-
-	logFields.Warn(msg)
+	logrus.WithFields(additionalInfos).Warn(msg)
 }
 
 func Tracef(format string, args ...interface{}) {
 	logrus.Tracef(format, args...)
 }
 
 func TraceWithFields(additionalInfos Fields, msg string) {
-	logFields := logrus.NewEntry(logrus.New())
-	for name, info := range additionalInfos {
-		logFields = logFields.WithField(name, info)
-	}
-
-	logFields.Trace(msg)
+	logrus.WithFields(additionalInfos).Trace(msg)
 }
 
 func DebugWithFields(additionalInfos Fields, msg string) {
-	logFields := logrus.NewEntry(logrus.New())
-	for name, info := range additionalInfos {
-		logFields = logFields.WithField(name, info)
-	}
-
-	logFields.Debug(msg)
+	logrus.WithFields(additionalInfos).Debug(msg)
 }
 
 func Debugf(format string, args ...interface{}) {
@@ -96,7 +76,7 @@ func logErrorInfo(err error, callerSkip int, additionalInfos ...Fields) *logrus.
 	}
 	pc, fullFilePath, line, ok := runtime.Caller(callerSkip + 2)
 	if ok {
-		logFields = logFields.WithFields(logrus.Fields{
+		logFields = logFields.WithFields(Fields{
 			"_file":     filepath.Base(fullFilePath),
 			"_function": runtime.FuncForPC(pc).Name(),
 			"_line":     line,
@@ -152,4 +132,4 @@ func logErrorInfo(err error, callerSkip int, additionalInfos ...Fields) *logrus.
 	return logFields
 }
 
-type Fields map[string]interface{}
+type Fields = logrus.Fields

backend/pkg/monitoring/constants/main.go

@@ -11,3 +11,5 @@ const (
 	Failure StatusType    = "failure"
 	Default time.Duration = -1 * time.Second
 )
+
+const CleanShutdownEvent = "clean_shutdown"

backend/pkg/monitoring/monitoring.go

@@ -2,9 +2,11 @@ package monitoring
 
 import (
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
+	"github.com/gobitfly/beaconchain/pkg/commons/log"
 	"github.com/gobitfly/beaconchain/pkg/commons/metrics"
 	"github.com/gobitfly/beaconchain/pkg/commons/types"
 	"github.com/gobitfly/beaconchain/pkg/commons/utils"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/constants"
 	"github.com/gobitfly/beaconchain/pkg/monitoring/services"
 )
 
@@ -33,6 +35,7 @@ func Init(full bool) {
 			&services.ServiceClickhouseRollings{},
 			&services.ServiceClickhouseEpoch{},
 			&services.ServiceTimeoutDetector{},
+			&services.CleanShutdownSpamDetector{},
 		)
 	}
 
@@ -42,13 +45,17 @@ func Init(full bool) {
 }
 
 func Start() {
+	log.Infof("starting monitoring services")
 	for _, service := range monitoredServices {
 		service.Start()
 	}
 }
 
 func Stop() {
+	log.Infof("stopping monitoring services")
 	for _, service := range monitoredServices {
 		service.Stop()
 	}
+	// this prevents status reports that werent shut down cleanly from triggering alerts
+	services.NewStatusReport(constants.CleanShutdownEvent, constants.Default, constants.Default)(constants.Success, nil)
 }

backend/pkg/monitoring/services/base.go

@@ -3,11 +3,8 @@ package services
 import (
 	"context"
 	"fmt"
-	"os"
-	"os/signal"
 	"sync"
 	"sync/atomic"
-	"syscall"
 	"time"
 
 	"github.com/gobitfly/beaconchain/pkg/commons/db"
@@ -45,31 +42,12 @@ func (s *ServiceBase) Stop() {
 	s.wg.Wait()
 }
 
-var isShuttingDown atomic.Bool
-var once sync.Once
-
-func autoGracefulStop() {
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
-	<-c
-	isShuttingDown.Store(true)
-}
-
 func NewStatusReport(id string, timeout time.Duration, check_interval time.Duration) func(status constants.StatusType, metadata map[string]string) {
 	runId := uuid.New().String()
-	// run if it hasnt started yet
-	once.Do(func() { go autoGracefulStop() })
 	return func(status constants.StatusType, metadata map[string]string) {
 		// acquire snowflake synchronously
 		flake := utils.GetSnowflake()
-
 		go func() {
-			// check if we are alive
-			if isShuttingDown.Load() {
-				log.Info("shutting down, not reporting status")
-				return
-			}
-
 			if metadata == nil {
 				metadata = make(map[string]string)
 			}
@@ -90,6 +68,15 @@ func NewStatusReport(id string, timeout time.Duration, check_interval time.Durat
 			if check_interval >= 5*time.Minute {
 				expires_at = timeouts_at.Add(check_interval)
 			}
+			log.TraceWithFields(log.Fields{
+				"emitter":         id,
+				"event_id":        utils.GetUUID(),
+				"deployment_type": utils.Config.DeploymentType,
+				"insert_id":       flake,
+				"expires_at":      expires_at,
+				"timeouts_at":     timeouts_at,
+				"metadata":        metadata,
+			}, "sending status report")
 			var err error
 			if db.ClickHouseNativeWriter != nil {
 				err = db.ClickHouseNativeWriter.AsyncInsert(

backend/pkg/monitoring/services/clean_shutdown_spam.go

@@ -0,0 +1,86 @@
+package services
+
+import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"github.com/gobitfly/beaconchain/pkg/commons/db"
+	"github.com/gobitfly/beaconchain/pkg/commons/log"
+	"github.com/gobitfly/beaconchain/pkg/commons/utils"
+	"github.com/gobitfly/beaconchain/pkg/monitoring/constants"
+)
+
+type CleanShutdownSpamDetector struct {
+	ServiceBase
+}
+
+func (s *CleanShutdownSpamDetector) Start() {
+	if !s.running.CompareAndSwap(false, true) {
+		// already running, return error
+		return
+	}
+	s.wg.Add(1)
+	go s.internalProcess()
+}
+
+func (s *CleanShutdownSpamDetector) internalProcess() {
+	defer s.wg.Done()
+	s.runChecks()
+	for {
+		select {
+		case <-s.ctx.Done():
+			return
+		case <-time.After(30 * time.Second):
+			s.runChecks()
+		}
+	}
+}
+
+func (s *CleanShutdownSpamDetector) runChecks() {
+	id := "monitoring_clean_shutdown_spam"
+	r := NewStatusReport(id, constants.Default, 30*time.Second)
+	r(constants.Running, nil)
+	if db.ClickHouseReader == nil {
+		r(constants.Failure, map[string]string{"error": "clickhouse reader is nil"})
+		// ignore
+		return
+	}
+	log.Tracef("checking clean shutdown spam")
+
+	query := `
+		SELECT
+			emitter
+		FROM
+			status_reports
+		WHERE
+			deployment_type = ?
+			AND inserted_at >= now() - interval 5 minutes
+			AND event_id = ?
+			`
+	ctx, cancel := context.WithTimeout(s.ctx, 15*time.Second)
+	defer cancel()
+	var emitters []string
+	err := db.ClickHouseReader.SelectContext(ctx, &emitters, query, utils.Config.DeploymentType, constants.CleanShutdownEvent)
+	if err != nil {
+		r(constants.Failure, map[string]string{"error": err.Error()})
+		return
+	}
+	threshold := 10
+	md := map[string]string{
+		"count":     strconv.Itoa(len(emitters)),
+		"threshold": strconv.Itoa(threshold),
+	}
+	if len(emitters) > threshold {
+		payload, err := json.Marshal(emitters)
+		if err != nil {
+			r(constants.Failure, map[string]string{"error": err.Error()})
+			return
+		}
+		md["emitters"] = string(payload)
+		r(constants.Failure, md)
+		return
+	}
+	r(constants.Success, md)
+}

backend/pkg/monitoring/services/clickhouse_rollings.go

@@ -71,7 +71,7 @@ func (s *ServiceClickhouseRollings) runChecks() {
 			err := db.ClickHouseReader.GetContext(ctx, &tsEpochTable, `
 					SELECT
 						max(epoch_timestamp)
-					FROM holesky.validator_dashboard_data_epoch`,
+					FROM validator_dashboard_data_epoch`,
 			)
 			if err != nil {
 				r(constants.Failure, map[string]string{"error": err.Error()})
@@ -81,7 +81,7 @@ func (s *ServiceClickhouseRollings) runChecks() {
 			err = db.ClickHouseReader.GetContext(ctx, &epochRollingTable, fmt.Sprintf(`
 					SELECT
 						max(epoch_end)
-					FROM holesky.validator_dashboard_data_rolling_%s`,
+					FROM validator_dashboard_data_rolling_%s`,
 				rolling,
 			),
 			)

backend/pkg/monitoring/services/timeout_detector.go

@@ -61,7 +61,7 @@ func (s *ServiceTimeoutDetector) runChecks() {
 				status,
 				metadata
 			FROM status_reports
-			WHERE expires_at > now() and deployment_type = ?
+			WHERE expires_at > now() and deployment_type = ? and emitter not in (select distinct emitter from status_reports where event_id = ? and inserted_at > now() - interval 1 days)
 			ORDER BY
 				event_id ASC,
 				emitter ASC,
@@ -87,6 +87,7 @@ func (s *ServiceTimeoutDetector) runChecks() {
 		)
 		SELECT
 			event_id,
+			emitter,
 			status,
 			inserted_at,
 			expires_at,
@@ -101,13 +102,14 @@ func (s *ServiceTimeoutDetector) runChecks() {
 	defer cancel()
 	var victims []struct {
 		EventID    string            `db:"event_id"`
+		Emitter    string            `db:"emitter"`
 		Status     string            `db:"status"`
 		InsertedAt time.Time         `db:"inserted_at"`
 		ExpiresAt  time.Time         `db:"expires_at"`
 		TimeoutsAt time.Time         `db:"timeouts_at"`
 		Metadata   map[string]string `db:"metadata"`
 	}
-	err := db.ClickHouseReader.SelectContext(ctx, &victims, query, utils.Config.DeploymentType)
+	err := db.ClickHouseReader.SelectContext(ctx, &victims, query, utils.Config.DeploymentType, constants.CleanShutdownEvent)
 	if err != nil {
 		r(constants.Failure, map[string]string{"error": err.Error()})
 		return