fix: logging for subs error, added metrics for latest block fetched

Author: blockchainluffy

rolling-shutter/keyperimpl/shutterservice/metrics.go

@@ -0,0 +1,16 @@
+package shutterservice
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var metricsRegistryEventsSyncedUntil = prometheus.NewGauge(
+	prometheus.GaugeOpts{
+		Namespace: "shutter",
+		Subsystem: "shutter_api",
+		Name:      "registry_events_syned_until",
+		Help:      "Current value of the latest block fetched",
+	},
+)
+
+func init() {
+	prometheus.MustRegister(metricsRegistryEventsSyncedUntil)
+}

rolling-shutter/keyperimpl/shutterservice/registrysyncer.go

@@ -182,6 +182,8 @@ func (s *RegistrySyncer) syncRange(
 		Int("num-inserted-events", len(filteredEvents)).
 		Int("num-discarded-events", len(events)-len(filteredEvents)).
 		Msg("synced registry contract")
+
+	metricsRegistryEventsSyncedUntil.Set(float64(end))
 	return nil
 }
 

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -62,7 +62,7 @@ func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) erro
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewEonPubkey(ctx)
+		return s.watchNewEonPubkey(ctx, subs.Err())
 	})
 	return nil
 }
@@ -126,7 +126,7 @@ func (s *EonPubKeySyncer) GetEonPubKeyForEon(ctx context.Context, opts *bind.Cal
 	}, nil
 }
 
-func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
+func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newEonKey, ok := <-s.keyBroadcastCh:
@@ -147,6 +147,10 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			if err != nil {
+				s.Log.Error("subscription error for watchNewEonPubkey", err)
+			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}

rolling-shutter/medley/chainsync/syncer/keyperset.go

@@ -72,7 +72,7 @@ func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) erro
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchNewKeypersService(ctx)
+		return s.watchNewKeypersService(ctx, subs.Err())
 	})
 	return nil
 }
@@ -204,7 +204,7 @@ func (s *KeyperSetSyncer) newEvent(
 	}, nil
 }
 
-func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context) error {
+func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newKeypers, ok := <-s.keyperAddedCh:
@@ -234,6 +234,10 @@ func (s *KeyperSetSyncer) watchNewKeypersService(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			if err != nil {
+				s.Log.Error("subscription error for watchNewKeypersService", err)
+			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}

rolling-shutter/medley/chainsync/syncer/shutterstate.go

@@ -63,7 +63,7 @@ func (s *ShutterStateSyncer) Start(ctx context.Context, runner service.Runner) e
 	runner.Defer(subsUnpaused.Unsubscribe)
 
 	runner.Go(func() error {
-		return s.watchPaused(ctx)
+		return s.watchPaused(ctx, subs.Err(), subsUnpaused.Err())
 	})
 	return nil
 }
@@ -87,7 +87,7 @@ func (s *ShutterStateSyncer) handle(ctx context.Context, ev *event.ShutterState)
 	}
 }
 
-func (s *ShutterStateSyncer) watchPaused(ctx context.Context) error {
+func (s *ShutterStateSyncer) watchPaused(ctx context.Context, subsErr <-chan error, subsErrUnpaused <-chan error) error {
 	isActive, err := s.pollIsActive(ctx)
 	if err != nil {
 		// XXX: this will fail everything, do we want that?
@@ -123,6 +123,14 @@ func (s *ShutterStateSyncer) watchPaused(ctx context.Context) error {
 			}
 			isActive = ev.Active
 			s.handle(ctx, ev)
+		case err := <-subsErr:
+			if err != nil {
+				s.Log.Error("subscription error for watchPaused")
+			}
+		case err := <-subsErrUnpaused:
+			if err != nil {
+				s.Log.Error("subscription error for watchUnpaused", err)
+			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}

rolling-shutter/medley/chainsync/syncer/unsafehead.go

@@ -33,12 +33,12 @@ func (s *UnsafeHeadSyncer) Start(ctx context.Context, runner service.Runner) err
 	}
 	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
-		return s.watchLatestUnsafeHead(ctx)
+		return s.watchLatestUnsafeHead(ctx, subs.Err())
 	})
 	return nil
 }
 
-func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context) error {
+func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context, subsErr <-chan error) error {
 	for {
 		select {
 		case newHeader, ok := <-s.newLatestHeadCh:
@@ -58,6 +58,10 @@ func (s *UnsafeHeadSyncer) watchLatestUnsafeHead(ctx context.Context) error {
 					err.Error(),
 				)
 			}
+		case err := <-subsErr:
+			if err != nil {
+				s.Log.Error("subscription error for watchLatestUnsafeHead", err)
+			}
 		case <-ctx.Done():
 			return ctx.Err()
 		}