From 87313887a4f50a4249e209ab5ebf9e529dd54686 Mon Sep 17 00:00:00 2001 From: Abdul Rabbani Date: Tue, 26 Apr 2022 13:57:01 -0400 Subject: [PATCH] Graceful Shutdown | Handle SSE | Requires Testing This Commit contains the following: * Graceful shutdowns. * Handling all incoming SSE events for reorgs, finalizations, and head. The structure of the `BeaconClient` has drastically changed and generics are used. --- .gitignore | 2 + cmd/capture.go | 18 +++--- cmd/head.go | 19 +++++- go.mod | 2 + go.sum | 5 ++ internal/boot/boot.go | 22 +++---- internal/boot/boot_test.go | 10 ++-- internal/shutdown/shutdown.go | 31 ++++++++++ pkg/beaconclient/beaconclient.go | 73 ++++++++++++++++++++++++ pkg/beaconclient/beaconclient_struct.go | 6 -- pkg/beaconclient/capturehead.go | 43 ++++++++++++++ pkg/beaconclient/head_topic.go | 7 --- pkg/beaconclient/healthcheck.go | 8 +-- pkg/beaconclient/healthcheck_test.go | 16 ++---- pkg/beaconclient/models.go | 37 ++++++++++++ pkg/beaconclient/ssehandler.go | 66 +++++++++++++++++++++ pkg/gracefulshutdown/gracefulshutdown.go | 64 +++++++++++++++++++++ pkg/loghelper/log_url.go | 12 ++++ 18 files changed, 388 insertions(+), 53 deletions(-) create mode 100644 internal/shutdown/shutdown.go create mode 100644 pkg/beaconclient/beaconclient.go delete mode 100644 pkg/beaconclient/beaconclient_struct.go create mode 100644 pkg/beaconclient/capturehead.go delete mode 100644 pkg/beaconclient/head_topic.go create mode 100644 pkg/beaconclient/models.go create mode 100644 pkg/beaconclient/ssehandler.go create mode 100644 pkg/gracefulshutdown/gracefulshutdown.go create mode 100644 pkg/loghelper/log_url.go diff --git a/.gitignore b/.gitignore index a0fdb30..d18d04f 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ ipld-ethcl-indexer ipld-ethcl-indexer.log +report.json +cover.profile diff --git a/cmd/capture.go b/cmd/capture.go index 54d3fb9..d81272d 100644 --- a/cmd/capture.go +++ b/cmd/capture.go @@ -6,20 +6,22 @@ package cmd import ( "os" + "time" "github.com/spf13/cobra" "github.com/spf13/viper" ) var ( - dbUsername string - dbPassword string - dbName string - dbAddress string - dbDriver string - dbPort int - bcAddress string - bcPort int + dbUsername string + dbPassword string + dbName string + dbAddress string + dbDriver string + dbPort int + bcAddress string + bcPort int + maxWaitSecondsShutdown time.Duration = time.Duration(5) * time.Second ) // captureCmd represents the capture command diff --git a/cmd/head.go b/cmd/head.go index 12b432a..10ee3b8 100644 --- a/cmd/head.go +++ b/cmd/head.go @@ -5,8 +5,13 @@ Copyright © 2022 NAME HERE package cmd import ( + "context" + "time" + + log "github.com/sirupsen/logrus" "github.com/spf13/cobra" "github.com/vulcanize/ipld-ethcl-indexer/internal/boot" + "github.com/vulcanize/ipld-ethcl-indexer/internal/shutdown" "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" ) @@ -22,10 +27,22 @@ var headCmd = &cobra.Command{ // Start the application to track at head. func startHeadTracking() { - _, err := boot.BootApplicationWithRetry(dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) + // Boot the application + log.Info("Starting the application in head tracking mode.") + ctx := context.Background() + + BC, DB, err := boot.BootApplicationWithRetry(ctx, dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) if err != nil { loghelper.LogError(err).Error("Unable to Start application") } + + // Capture head blocks + go BC.CaptureHead() + + // Shutdown when the time is right. + wait := shutdown.ShutdownServices(ctx, time.Duration(maxWaitSecondsShutdown), DB, BC) + + <-wait } func init() { diff --git a/go.mod b/go.mod index de583a9..e8829d7 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/pkg/errors v0.9.1 // indirect golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect golang.org/x/net v0.0.0-20220412020605-290c469a71a5 // indirect + gopkg.in/cenkalti/backoff.v1 v1.1.0 // indirect gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect ) @@ -35,6 +36,7 @@ require ( github.com/mitchellh/mapstructure v1.4.3 // indirect github.com/pelletier/go-toml v1.9.4 // indirect github.com/pelletier/go-toml/v2 v2.0.0-beta.8 // indirect + github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc github.com/spf13/afero v1.8.2 // indirect github.com/spf13/cast v1.4.1 // indirect github.com/spf13/cobra v1.4.0 diff --git a/go.sum b/go.sum index fa80980..0b56ef7 100644 --- a/go.sum +++ b/go.sum @@ -276,6 +276,8 @@ github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qR github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc h1:zAsgcP8MhzAbhMnB1QQ2O7ZhWYVGYSR2iVcjzQuPV+o= +github.com/r3labs/sse v0.0.0-20210224172625-26fe804710bc/go.mod h1:S8xSOnV3CgpNrWd0GQ/OoQfMtlg2uPRSuTzcSGrzwK8= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ= github.com/rs/zerolog v1.13.0/go.mod h1:YbFCdg8HfsridGWAh22vktObvhZbQsZXe4/zB0OKkWU= @@ -401,6 +403,7 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20190628185345-da137c7871d7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20191116160921-f9c825593386/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20191209160850-c0dbc17a3553/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -660,6 +663,8 @@ google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlba google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw= +gopkg.in/cenkalti/backoff.v1 v1.1.0 h1:Arh75ttbsvlpVA7WtVpH4u9h6Zl46xuptxqLxPiSo4Y= +gopkg.in/cenkalti/backoff.v1 v1.1.0/go.mod h1:J6Vskwqd+OMVJl8C33mmtxTBs2gyzfv7UDAkHu8BrjI= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= diff --git a/internal/boot/boot.go b/internal/boot/boot.go index 9d6c4a8..3c71b7b 100644 --- a/internal/boot/boot.go +++ b/internal/boot/boot.go @@ -1,6 +1,7 @@ package boot import ( + "context" "time" log "github.com/sirupsen/logrus" @@ -55,39 +56,38 @@ func SetupPostgresDb(dbHostname string, dbPort int, dbName string, dbUsername st // // 2. Connect to the database. // -func BootApplication(dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string, bcAddress string, bcPort int) (sql.Database, error) { +func BootApplication(ctx context.Context, dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string, bcAddress string, bcPort int) (*beaconclient.BeaconClient, sql.Database, error) { log.Info("Booting the Application") log.Debug("Creating the Beacon Client") - BC.Address = bcAddress - BC.Port = bcPort - log.Debug("Checking Beacon Client") + BC = beaconclient.CreateBeaconClient(ctx, bcAddress, bcPort) + log.Debug("Checking Beacon Client") err := BC.CheckBeaconClient() if err != nil { - return nil, err + return nil, nil, err } log.Debug("Setting up DB connection") DB, err := SetupPostgresDb(dbHostname, dbPort, dbName, dbUsername, dbPassword, driverName) if err != nil { - return nil, err + return nil, nil, err } - return DB, nil + return BC, DB, nil } // Add retry logic to ensure that we are give the Beacon Client and the DB time to start. -func BootApplicationWithRetry(dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string, bcAddress string, bcPort int) (sql.Database, error) { +func BootApplicationWithRetry(ctx context.Context, dbHostname string, dbPort int, dbName string, dbUsername string, dbPassword string, driverName string, bcAddress string, bcPort int) (*beaconclient.BeaconClient, sql.Database, error) { var err error for i := 0; i < maxRetry; i++ { - DB, err = BootApplication(dbHostname, dbPort, dbName, dbUsername, dbPassword, driverName, bcAddress, bcPort) + BC, DB, err = BootApplication(ctx, dbHostname, dbPort, dbName, dbUsername, dbPassword, driverName, bcAddress, bcPort) if err != nil { log.WithFields(log.Fields{ "retryNumber": i, }).Warn("Unable to boot application. Going to try again") time.Sleep(time.Duration(retryInterval) * time.Second) - continue } + break } - return DB, err + return BC, DB, err } diff --git a/internal/boot/boot_test.go b/internal/boot/boot_test.go index 729f30b..9db307e 100644 --- a/internal/boot/boot_test.go +++ b/internal/boot/boot_test.go @@ -1,6 +1,8 @@ package boot_test import ( + "context" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" "github.com/vulcanize/ipld-ethcl-indexer/internal/boot" @@ -20,26 +22,26 @@ var _ = Describe("Boot", func() { Describe("Booting the application", Label("integration"), func() { Context("When the DB and BC are both up and running", func() { It("Should connect successfully", func() { - db, err := boot.BootApplicationWithRetry(dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) + _, db, err := boot.BootApplicationWithRetry(context.Background(), dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) defer db.Close() Expect(err).To(BeNil()) }) }) Context("When the DB is running but not the BC", func() { It("Should not connect successfully", func() { - _, err := boot.BootApplication(dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, "hi", 100) + _, _, err := boot.BootApplication(context.Background(), dbAddress, dbPort, dbName, dbUsername, dbPassword, dbDriver, "hi", 100) Expect(err).ToNot(BeNil()) }) }) Context("When the BC is running but not the DB", func() { It("Should not connect successfully", func() { - _, err := boot.BootApplication("hi", 10, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) + _, _, err := boot.BootApplication(context.Background(), "hi", 10, dbName, dbUsername, dbPassword, dbDriver, bcAddress, bcPort) Expect(err).ToNot(BeNil()) }) }) Context("When neither the BC or DB are running", func() { It("Should not connect successfully", func() { - _, err := boot.BootApplication("hi", 10, dbName, dbUsername, dbPassword, dbDriver, "hi", 100) + _, _, err := boot.BootApplication(context.Background(), "hi", 10, dbName, dbUsername, dbPassword, dbDriver, "hi", 100) Expect(err).ToNot(BeNil()) }) }) diff --git a/internal/shutdown/shutdown.go b/internal/shutdown/shutdown.go new file mode 100644 index 0000000..9b1742f --- /dev/null +++ b/internal/shutdown/shutdown.go @@ -0,0 +1,31 @@ +package shutdown + +import ( + "context" + "time" + + "github.com/vulcanize/ipld-ethcl-indexer/pkg/beaconclient" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/database/sql" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/gracefulshutdown" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" +) + +// Shutdown all the internal services for the application. +func ShutdownServices(ctx context.Context, waitTime time.Duration, DB sql.Database, BC *beaconclient.BeaconClient) <-chan struct{} { + return gracefulshutdown.Shutdown(ctx, waitTime, map[string]gracefulshutdown.Operation{ + "database": func(ctx context.Context) error { + err := DB.Close() + if err != nil { + loghelper.LogError(err).Error("Unable to close the DB") + } + return err + }, + "beaconClient": func(ctx context.Context) error { + err := BC.StopHeadTracking() + if err != nil { + loghelper.LogError(err).Error("Unable to trigger shutdown of head tracking") + } + return err + }, + }) +} diff --git a/pkg/beaconclient/beaconclient.go b/pkg/beaconclient/beaconclient.go new file mode 100644 index 0000000..8704220 --- /dev/null +++ b/pkg/beaconclient/beaconclient.go @@ -0,0 +1,73 @@ +package beaconclient + +import ( + "context" + "fmt" + + "github.com/r3labs/sse" + log "github.com/sirupsen/logrus" +) + +var ( + bcHealthEndpoint = "/eth/v1/node/health" // Endpoint used for the healthcheck + bcHeadTopicEndpoint = "/eth/v1/events?topics=head" // Endpoint used to subscribe to the head of the chain + bcReorgTopicEndpoint = "/eth/v1/events?topics=chain_reorg" // Endpoint used to subscribe to the head of the chain + bcFinalizedTopicEndpoint = "/eth/v1/events?topics=finalized_checkpoint" // Endpoint used to subscribe to the head of the chain + connectionProtocol = "http" +) + +// A struct that capture the Beacon Server that the Beacon Client will be interacting with and querying. +type BeaconClient struct { + Context context.Context // A context generic context with multiple uses. + ServerAddress string // Address of the Beacon Server + ServerPort int // Port of the Beacon Server + PerformHeadTracking bool // Should we track head? + PerformHistoricalProcessing bool // Should we perform historical processing? + HeadTracking *SseEvents[Head] // Track the head block + ReOrgTracking *SseEvents[ChainReorg] // Track all Reorgs + FinalizationTracking *SseEvents[FinalizedCheckpoint] // Track all finalization checkpoints +} + +// A struct to keep track of relevant the head event topic. +type SseEvents[P ProcessedEvents] struct { + Url string // The url for the subscription. Primarily used for logging + MessagesCh chan *sse.Event // Contains all the messages from the SSE Channel + ErrorCh chan *SseError // Contains any errors while SSE streaming occurred + ProcessCh chan *P // Used to capture processed data in its proper struct. + SseClient *sse.Client // sse.Client object that is used to interact with the SSE stream +} + +// An object to capture any errors when turning an SSE message to JSON. +type SseError struct { + err error + msg []byte +} + +// A Function to create the BeaconClient. +func CreateBeaconClient(ctx context.Context, bcAddress string, bcPort int) *BeaconClient { + log.Info("Creating the BeaconClient") + return &BeaconClient{ + Context: ctx, + ServerAddress: bcAddress, + ServerPort: bcPort, + HeadTracking: createSseEvent[Head](connectionProtocol, bcAddress, bcPort, bcHeadTopicEndpoint), + ReOrgTracking: createSseEvent[ChainReorg](connectionProtocol, bcAddress, bcPort, bcReorgTopicEndpoint), + FinalizationTracking: createSseEvent[FinalizedCheckpoint](connectionProtocol, bcAddress, bcPort, bcFinalizedTopicEndpoint), + } +} + +// Create all the channels to handle a SSE events +func createSseEvent[P ProcessedEvents](connectionProtocol, bcAddress string, bcPort int, endpoint string) *SseEvents[P] { + url := fmt.Sprintf("%s://%s:%d%s", connectionProtocol, bcAddress, bcPort, endpoint) + sseEvents := &SseEvents[P]{ + Url: url, + MessagesCh: make(chan *sse.Event), + ErrorCh: make(chan *SseError), + ProcessCh: make(chan *P), + SseClient: func(url string) *sse.Client { + log.WithFields(log.Fields{"url": url}).Info("Creating SSE client") + return sse.NewClient(url) + }(url), + } + return sseEvents +} diff --git a/pkg/beaconclient/beaconclient_struct.go b/pkg/beaconclient/beaconclient_struct.go deleted file mode 100644 index 0e9b00d..0000000 --- a/pkg/beaconclient/beaconclient_struct.go +++ /dev/null @@ -1,6 +0,0 @@ -package beaconclient - -type BeaconClient struct { - Address string - Port int -} diff --git a/pkg/beaconclient/capturehead.go b/pkg/beaconclient/capturehead.go new file mode 100644 index 0000000..6a20c8f --- /dev/null +++ b/pkg/beaconclient/capturehead.go @@ -0,0 +1,43 @@ +package beaconclient + +import ( + "time" + + log "github.com/sirupsen/logrus" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" +) + +// This function will perform all the heavy lifting for tracking the head of the chain. +func (bc *BeaconClient) CaptureHead() { + log.Info("We are tracking the head of the chain.") + //go readProcessedEvents(bc.HeadTracking.ProcessCh) + bc.CaptureHeadTopic() +} + +// Stop the head tracking service. +func (bc *BeaconClient) StopHeadTracking() error { + log.Info("We are going to stop tracking the head of chain because of the shutdown signal.") + chHead := make(chan bool) + chReorg := make(chan bool) + chFinal := make(chan bool) + + go bc.HeadTracking.finishProcessingChannel(chHead) + go bc.ReOrgTracking.finishProcessingChannel(chReorg) + go bc.FinalizationTracking.finishProcessingChannel(chFinal) + + <-chHead + <-chFinal + <-chReorg + log.Info("Successfully stopped the head tracking service.") + return nil +} + +func (se *SseEvents[ProcessedEvents]) finishProcessingChannel(finish chan<- bool) { + loghelper.LogUrl(se.Url).Info("Received a close event.") + se.SseClient.Unsubscribe(se.MessagesCh) + for len(se.MessagesCh) != 0 { + time.Sleep(time.Duration(shutdownWaitInterval) * time.Millisecond) + } + loghelper.LogUrl(se.Url).Info("Done processing all messages, ready for shutdown") + finish <- true +} diff --git a/pkg/beaconclient/head_topic.go b/pkg/beaconclient/head_topic.go deleted file mode 100644 index 1db229d..0000000 --- a/pkg/beaconclient/head_topic.go +++ /dev/null @@ -1,7 +0,0 @@ -// This package will handle all event subscriptions that utilize SSE. - -package beaconclient - -func ListenToHead() { - -} diff --git a/pkg/beaconclient/healthcheck.go b/pkg/beaconclient/healthcheck.go index bf6b562..342ccd1 100644 --- a/pkg/beaconclient/healthcheck.go +++ b/pkg/beaconclient/healthcheck.go @@ -9,17 +9,13 @@ import ( "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" ) -var ( - bcHealthEndpoint = "/eth/v1/node/health" -) - // This function will ensure that we can connect to the beacon client. // Keep in mind, the beacon client will allow you to connect to it but it might // Not allow you to make http requests. This is part of its built in logic, and you will have // to follow their provided guidelines. https://lighthouse-book.sigmaprime.io/api-bn.html#security -func (bc *BeaconClient) CheckBeaconClient() error { +func (bc BeaconClient) CheckBeaconClient() error { log.Debug("Attempting to connect to the beacon client") - bcEndpoint := "http://" + bc.Address + ":" + strconv.Itoa(bc.Port) + bcHealthEndpoint + bcEndpoint := "http://" + bc.ServerAddress + ":" + strconv.Itoa(bc.ServerPort) + bcHealthEndpoint resp, err := http.Get(bcEndpoint) if err != nil { loghelper.LogError(err).Error("Unable to get bc endpoint: ", bcEndpoint) diff --git a/pkg/beaconclient/healthcheck_test.go b/pkg/beaconclient/healthcheck_test.go index ebebd0c..cc255dc 100644 --- a/pkg/beaconclient/healthcheck_test.go +++ b/pkg/beaconclient/healthcheck_test.go @@ -1,6 +1,8 @@ package beaconclient_test import ( + "context" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" beaconclient "github.com/vulcanize/ipld-ethcl-indexer/pkg/beaconclient" @@ -8,14 +10,8 @@ import ( var _ = Describe("Healthcheck", func() { var ( - BC = beaconclient.BeaconClient{ - Address: "localhost", - Port: 5052, - } - errBc = beaconclient.BeaconClient{ - Address: "blah", - Port: 10, - } + BC = beaconclient.CreateBeaconClient(context.Background(), "localhost", 5052) + errBc = beaconclient.CreateBeaconClient(context.Background(), "blah-blah", 1010) ) Describe("Connecting to the lighthouse client", Label("integration"), func() { Context("When the client is running", func() { @@ -24,8 +20,8 @@ var _ = Describe("Healthcheck", func() { Expect(err).To(BeNil()) }) }) - Context("When the client is running", func() { - It("We should connect successfully", func() { + Context("When the client is not running", func() { + It("We not should connect successfully", func() { err := errBc.CheckBeaconClient() Expect(err).ToNot(BeNil()) }) diff --git a/pkg/beaconclient/models.go b/pkg/beaconclient/models.go new file mode 100644 index 0000000..e047845 --- /dev/null +++ b/pkg/beaconclient/models.go @@ -0,0 +1,37 @@ +package beaconclient + +// This interface captured what the events can be for processed event streams. +type ProcessedEvents interface { + Head | FinalizedCheckpoint | ChainReorg +} + +// This struct captures the JSON representation of the head topic +type Head struct { + Slot string `json:"slot"` + Block string `json:"block"` + State string `json:"state"` + CurrentDutyDependentRoot string `json:"current_duty_dependent_root"` + PreviousDutyDependentRoot string `json:"previous_duty_dependent_root"` + EpochTransition bool `json:"epoch_transition"` + ExecutionOptimistic bool `json:"execution_optimistic"` +} + +// This struct captures the JSON representation of the finalized_checkpoint topic. +type FinalizedCheckpoint struct { + Block string `json:"block"` + State string `json:"state"` + Epoch string `json:"epoch"` + ExecutionOptimistic bool `json:"execution_optimistic"` +} + +// This struct captures the JSON representation of the chain_reorg topic. +type ChainReorg struct { + Slot string `json:"slot"` + Depth string `json:"depth"` + OldHeadBlock string `json:"old_head_block"` + NewHeadBlock string `json:"new_head_block"` + OldHeadState string `json:"old_head_state"` + NewHeadState string `json:"new_head_state"` + Epoch string `json:"epoch"` + ExecutionOptimistic bool `json:"execution_optimistic"` +} diff --git a/pkg/beaconclient/ssehandler.go b/pkg/beaconclient/ssehandler.go new file mode 100644 index 0000000..6e3c003 --- /dev/null +++ b/pkg/beaconclient/ssehandler.go @@ -0,0 +1,66 @@ +// This package will handle all event subscriptions that utilize SSE. + +package beaconclient + +import ( + "encoding/json" + "time" + + log "github.com/sirupsen/logrus" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" +) + +var ( + shutdownWaitInterval = time.Duration(5) * time.Second +) + +// This function will capture all the SSE events for a given SseEvents object. +// When new messages come in, it will ensure that they are decoded into JSON. +// If any errors occur, it log the error information. +func handleSseEvent[P ProcessedEvents](eventHandler *SseEvents[P]) { + loghelper.LogUrl(eventHandler.Url).Info("Subscribing to Messages") + go eventHandler.SseClient.SubscribeChanRaw(eventHandler.MessagesCh) + for { + select { + case message := <-eventHandler.MessagesCh: + // Message can be nil if its a keep-alive message + if len(message.Data) != 0 { + go processMsg(message.Data, eventHandler.ProcessCh, eventHandler.ErrorCh) + } + + case headErr := <-eventHandler.ErrorCh: + log.WithFields(log.Fields{ + "url": eventHandler.Url, + "err": headErr.err, + "msg": headErr.msg, + }, + ).Error("Unable to handle event.") + + case process := <-eventHandler.ProcessCh: + log.WithFields(log.Fields{"processed": process}).Debug("Processesing a Message") + } + } +} + +// Capture all of the head topics. +func (bc *BeaconClient) CaptureHeadTopic() { + log.Info("We are capturing all SSE events") + go handleSseEvent(bc.HeadTracking) + go handleSseEvent(bc.ReOrgTracking) + go handleSseEvent(bc.FinalizationTracking) +} + +// Turn the data object into a Struct. +func processMsg[P ProcessedEvents](msg []byte, processCh chan<- *P, errorCh chan<- *SseError) { + log.WithFields(log.Fields{"msg": msg}).Debug("Processing a Message") + var msgMarshaled P + err := json.Unmarshal(msg, &msgMarshaled) + if err != nil { + errorCh <- &SseError{ + err: err, + msg: msg, + } + return + } + processCh <- &msgMarshaled +} diff --git a/pkg/gracefulshutdown/gracefulshutdown.go b/pkg/gracefulshutdown/gracefulshutdown.go new file mode 100644 index 0000000..36510db --- /dev/null +++ b/pkg/gracefulshutdown/gracefulshutdown.go @@ -0,0 +1,64 @@ +package gracefulshutdown + +import ( + "context" + "os" + "os/signal" + "sync" + "syscall" + "time" + + log "github.com/sirupsen/logrus" + "github.com/vulcanize/ipld-ethcl-indexer/pkg/loghelper" +) + +// operation is a clean up function on shutting down +type Operation func(ctx context.Context) error + +// gracefulShutdown waits for termination syscalls and doing clean up operations after received it +func Shutdown(ctx context.Context, timeout time.Duration, ops map[string]Operation) <-chan struct{} { + wait := make(chan struct{}) + go func() { + s := make(chan os.Signal, 1) + + // add any other syscalls that you want to be notified with + signal.Notify(s, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + <-s + + log.Info("Shutting Down your application") + + // set timeout for the ops to be done to prevent system hang + timeoutFunc := time.AfterFunc(timeout, func() { + log.Warnf("timeout %d ms has been elapsed, force exit", timeout.Milliseconds()) + os.Exit(0) + }) + + defer timeoutFunc.Stop() + + var wg sync.WaitGroup + + // Do the operations asynchronously to save time + for key, op := range ops { + wg.Add(1) + innerOp := op + innerKey := key + go func() { + defer wg.Done() + + log.Infof("cleaning up: %s", innerKey) + if err := innerOp(ctx); err != nil { + loghelper.LogError(err).Errorf("%s: clean up failed: %s", innerKey, err.Error()) + return + } + + log.Infof("%s was shutdown gracefully", innerKey) + }() + } + + wg.Wait() + + close(wait) + }() + + return wait +} diff --git a/pkg/loghelper/log_url.go b/pkg/loghelper/log_url.go new file mode 100644 index 0000000..4a7d781 --- /dev/null +++ b/pkg/loghelper/log_url.go @@ -0,0 +1,12 @@ +package loghelper + +import ( + log "github.com/sirupsen/logrus" +) + +// A simple helper function that will help wrap the error message. +func LogUrl(url string) *log.Entry { + return log.WithFields(log.Fields{ + "url": url, + }) +}