dvovk/headers info

diagnostics/headers.go

@@ -0,0 +1,20 @@
+package diagnostics
+
+import (
+	"encoding/json"
+	"net/http"
+
+	diaglib "github.com/ledgerwatch/erigon-lib/diagnostics"
+)
+
+func SetupHeadersAccess(metricsMux *http.ServeMux, diag *diaglib.DiagnosticClient) {
+	metricsMux.HandleFunc("/headers", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+		w.Header().Set("Content-Type", "application/json")
+		writeHeaders(w, diag)
+	})
+}
+
+func writeHeaders(w http.ResponseWriter, diag *diaglib.DiagnosticClient) {
+	json.NewEncoder(w).Encode(diag.GetHeaders())
+}

diagnostics/setup.go

@@ -34,6 +34,7 @@ func Setup(ctx *cli.Context, metricsMux *http.ServeMux, node *node.ErigonNode) {
 	SetupBootnodesAccess(debugMux, node)
 	SetupStagesAccess(debugMux, diagnostic)
 	SetupMemAccess(debugMux)
+	SetupHeadersAccess(debugMux, diagnostic)
 	SetupBodiesAccess(debugMux, diagnostic)
 
 }

erigon-lib/diagnostics/block_execution.go

@@ -3,26 +3,22 @@ package diagnostics
 import (
 	"context"
 
-	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/log/v3"
 )
 
-func (d *DiagnosticClient) setupBlockExecutionDiagnostics() {
-	d.runBlockExecutionListener()
+func (d *DiagnosticClient) setupBlockExecutionDiagnostics(rootCtx context.Context) {
+	d.runBlockExecutionListener(rootCtx)
 }
 
-func (d *DiagnosticClient) runBlockExecutionListener() {
+func (d *DiagnosticClient) runBlockExecutionListener(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[BlockExecutionStatistics](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[BlockExecutionStatistics](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(BlockExecutionStatistics{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				d.mu.Lock()

erigon-lib/diagnostics/bodies.go

@@ -3,29 +3,25 @@ package diagnostics
 import (
 	"context"
 
-	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/log/v3"
 )
 
-func (d *DiagnosticClient) setupBodiesDiagnostics() {
-	d.runBodiesBlockDownloadListener()
-	d.runBodiesBlockWriteListener()
-	d.runBodiesProcessingListener()
-	d.runBodiesProcessedListener()
+func (d *DiagnosticClient) setupBodiesDiagnostics(rootCtx context.Context) {
+	d.runBodiesBlockDownloadListener(rootCtx)
+	d.runBodiesBlockWriteListener(rootCtx)
+	d.runBodiesProcessingListener(rootCtx)
+	d.runBodiesProcessedListener(rootCtx)
 }
 
-func (d *DiagnosticClient) runBodiesBlockDownloadListener() {
+func (d *DiagnosticClient) runBodiesBlockDownloadListener(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[BodiesDownloadBlockUpdate](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[BodiesDownloadBlockUpdate](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(BodiesDownloadBlockUpdate{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				d.bodiesMutex.Lock()
@@ -37,18 +33,15 @@ func (d *DiagnosticClient) runBodiesBlockDownloadListener() {
 	}()
 }
 
-func (d *DiagnosticClient) runBodiesBlockWriteListener() {
+func (d *DiagnosticClient) runBodiesBlockWriteListener(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[BodiesWriteBlockUpdate](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[BodiesWriteBlockUpdate](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(BodiesWriteBlockUpdate{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				d.bodiesMutex.Lock()
@@ -60,18 +53,15 @@ func (d *DiagnosticClient) runBodiesBlockWriteListener() {
 	}()
 }
 
-func (d *DiagnosticClient) runBodiesProcessedListener() {
+func (d *DiagnosticClient) runBodiesProcessedListener(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[BodiesProcessedUpdate](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[BodiesProcessedUpdate](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(BodiesProcessedUpdate{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				d.bodiesMutex.Lock()
@@ -83,18 +73,15 @@ func (d *DiagnosticClient) runBodiesProcessedListener() {
 	}()
 }
 
-func (d *DiagnosticClient) runBodiesProcessingListener() {
+func (d *DiagnosticClient) runBodiesProcessingListener(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[BodiesProcessingUpdate](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[BodiesProcessingUpdate](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(BodiesProcessingUpdate{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				d.bodiesMutex.Lock()

erigon-lib/diagnostics/client.go

@@ -3,6 +3,8 @@ package diagnostics
 import (
 	"net/http"
 	"sync"
+
+	"github.com/ledgerwatch/erigon-lib/common"
 )
 
 type DiagnosticClient struct {
@@ -12,8 +14,10 @@ type DiagnosticClient struct {
 	syncStats           SyncStatistics
 	snapshotFileList    SnapshoFilesList
 	mu                  sync.Mutex
+	headerMutex         sync.Mutex
 	hardwareInfo        HardwareInfo
 	peersSyncMap        sync.Map
+	headers             Headers
 	bodies              BodiesInfo
 	bodiesMutex         sync.Mutex
 	resourcesUsage      ResourcesUsage
@@ -35,13 +39,17 @@ func NewDiagnosticClient(metricsMux *http.ServeMux, dataDirPath string) *Diagnos
 }
 
 func (d *DiagnosticClient) Setup() {
-	d.setupSnapshotDiagnostics()
-	d.setupStagesDiagnostics()
+
+	rootCtx, _ := common.RootContext()
+
+	d.setupSnapshotDiagnostics(rootCtx)
+	d.setupStagesDiagnostics(rootCtx)
 	d.setupSysInfoDiagnostics()
-	d.setupNetworkDiagnostics()
-	d.setupBlockExecutionDiagnostics()
-	d.setupBodiesDiagnostics()
-	d.setupResourcesUsageDiagnostics()
+	d.setupNetworkDiagnostics(rootCtx)
+	d.setupBlockExecutionDiagnostics(rootCtx)
+	d.setupHeadersDiagnostics(rootCtx)
+	d.setupBodiesDiagnostics(rootCtx)
+	d.setupResourcesUsageDiagnostics(rootCtx)
 
 	//d.logDiagMsgs()
 }

erigon-lib/diagnostics/entities.go

@@ -167,6 +167,28 @@ type BlockHeadersUpdate struct {
 	RejectedBadHeaders  int     `json:"rejectedBadHeaders"`
 }
 
+type HeadersWaitingUpdate struct {
+	From uint64 `json:"from"`
+}
+
+type HeaderCanonicalMarkerUpdate struct {
+	AncestorHeight uint64 `json:"ancestorHeight"`
+	AncestorHash   string `json:"ancestorHash"`
+}
+type HeadersProcessedUpdate struct {
+	Highest   uint64  `json:"highest"`
+	Age       int     `json:"age"`
+	Headers   uint64  `json:"headers"`
+	In        float64 `json:"in"`
+	BlkPerSec uint64  `json:"blkPerSec"`
+}
+
+type Headers struct {
+	WaitingForHeaders uint64                      `json:"waitingForHeaders"`
+	WriteHeaders      BlockHeadersUpdate          `json:"writeHeaders"`
+	CanonicalMarker   HeaderCanonicalMarkerUpdate `json:"canonicalMarker"`
+	Processed         HeadersProcessedUpdate      `json:"processed"`
+}
 type BodiesInfo struct {
 	BlockDownload BodiesDownloadBlockUpdate `json:"blockDownload"`
 	BlockWrite    BodiesWriteBlockUpdate    `json:"blockWrite"`
@@ -276,3 +298,15 @@ func (ti CurrentSyncStage) Type() Type {
 func (ti PeerStatisticMsgUpdate) Type() Type {
 	return TypeOf(ti)
 }
+
+func (ti HeadersWaitingUpdate) Type() Type {
+	return TypeOf(ti)
+}
+
+func (ti HeaderCanonicalMarkerUpdate) Type() Type {
+	return TypeOf(ti)
+}
+
+func (ti HeadersProcessedUpdate) Type() Type {
+	return TypeOf(ti)
+}

erigon-lib/diagnostics/headers.go

@@ -0,0 +1,102 @@
+package diagnostics
+
+import (
+	"context"
+
+	"github.com/ledgerwatch/log/v3"
+)
+
+func (d *DiagnosticClient) setupHeadersDiagnostics(rootCtx context.Context) {
+	d.runHeadersWaitingListener(rootCtx)
+	d.runWriteHeadersListener(rootCtx)
+	d.runCanonicalMarkerListener(rootCtx)
+	d.runProcessedListener(rootCtx)
+}
+
+func (d *DiagnosticClient) GetHeaders() Headers {
+	return d.headers
+}
+
+func (d *DiagnosticClient) runHeadersWaitingListener(rootCtx context.Context) {
+	go func() {
+		ctx, ch, closeChannel := Context[HeadersWaitingUpdate](rootCtx, 1)
+		defer closeChannel()
+
+		StartProviders(ctx, TypeOf(HeadersWaitingUpdate{}), log.Root())
+		for {
+			select {
+			case <-rootCtx.Done():
+				return
+			case info := <-ch:
+				d.headerMutex.Lock()
+				d.headers.WaitingForHeaders = info.From
+				d.headerMutex.Unlock()
+
+				return
+			}
+		}
+	}()
+}
+
+func (d *DiagnosticClient) runWriteHeadersListener(rootCtx context.Context) {
+	go func() {
+		ctx, ch, closeChannel := Context[BlockHeadersUpdate](rootCtx, 1)
+		defer closeChannel()
+
+		StartProviders(ctx, TypeOf(BlockHeadersUpdate{}), log.Root())
+		for {
+			select {
+			case <-rootCtx.Done():
+				return
+			case info := <-ch:
+				d.headerMutex.Lock()
+				d.headers.WriteHeaders = info
+				d.headerMutex.Unlock()
+
+				return
+			}
+		}
+	}()
+}
+
+func (d *DiagnosticClient) runCanonicalMarkerListener(rootCtx context.Context) {
+	go func() {
+		ctx, ch, closeChannel := Context[HeaderCanonicalMarkerUpdate](rootCtx, 1)
+		defer closeChannel()
+
+		StartProviders(ctx, TypeOf(HeaderCanonicalMarkerUpdate{}), log.Root())
+		for {
+			select {
+			case <-rootCtx.Done():
+				return
+			case info := <-ch:
+				d.headerMutex.Lock()
+				d.headers.CanonicalMarker = info
+				d.headerMutex.Unlock()
+
+				return
+			}
+		}
+	}()
+}
+
+func (d *DiagnosticClient) runProcessedListener(rootCtx context.Context) {
+	go func() {
+		ctx, ch, closeChannel := Context[HeadersProcessedUpdate](rootCtx, 1)
+		defer closeChannel()
+
+		StartProviders(ctx, TypeOf(HeadersProcessedUpdate{}), log.Root())
+		for {
+			select {
+			case <-rootCtx.Done():
+				return
+			case info := <-ch:
+				d.headerMutex.Lock()
+				d.headers.Processed = info
+				d.headerMutex.Unlock()
+
+				return
+			}
+		}
+	}()
+}

erigon-lib/diagnostics/network.go

@@ -3,26 +3,22 @@ package diagnostics
 import (
 	"context"
 
-	"github.com/ledgerwatch/erigon-lib/common"
 	"github.com/ledgerwatch/log/v3"
 )
 
-func (d *DiagnosticClient) setupNetworkDiagnostics() {
-	d.runCollectPeersStatistics()
+func (d *DiagnosticClient) setupNetworkDiagnostics(rootCtx context.Context) {
+	d.runCollectPeersStatistics(rootCtx)
 }
 
-func (d *DiagnosticClient) runCollectPeersStatistics() {
+func (d *DiagnosticClient) runCollectPeersStatistics(rootCtx context.Context) {
 	go func() {
-		ctx, ch, cancel := Context[PeerStatisticMsgUpdate](context.Background(), 1)
-		defer cancel()
-
-		rootCtx, _ := common.RootContext()
+		ctx, ch, closeChannel := Context[PeerStatisticMsgUpdate](rootCtx, 1)
+		defer closeChannel()
 
 		StartProviders(ctx, TypeOf(PeerStatisticMsgUpdate{}), log.Root())
 		for {
 			select {
 			case <-rootCtx.Done():
-				cancel()
 				return
 			case info := <-ch:
 				if value, ok := d.peersSyncMap.Load(info.PeerID); ok {