index-builder: serve health check on --index-builder-grpc-listen-addr

CHANGELOG.md

@@ -12,6 +12,8 @@ If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you s
 
 ### Fixed
 
+- `index-builder` now serves its health check, over gRPC and HTTP `/healthz`, on `--index-builder-grpc-listen-addr`, which it never listened on.
+
 - Merger no longer moves canonical one-block files to the forked-blocks store when readers write the same block with a different LIB, as happens on Polygon PoS / Amoy. Only blocks with another ID at the same height are moved.
 
 - The reader now always clamps a decoded block's `lib_num` down to its own block number when the node/plugin reports `lib_num` greater than the block number (invalid; `lib_num` equal to the block number is still valid), logs an error and increments `reader_node_invalid_libnum_clamped_count`, instead of letting the bad value reach the relayer where it moved LIB past head and silently stalled it forever.

cmd/apps/index_builder.go

@@ -86,6 +86,7 @@ func RegisterIndexBuilderApp[B firecore.Block](chain *firecore.Chain[B], rootLog
 				EndBlock:             stopBlockNum,
 				MergedBlocksStoreURL: mergedBlocksStoreURL,
 				GRPCListenAddr:       viper.GetString("index-builder-grpc-listen-addr"),
+				IsPendingShutdown:    runtime.IsPendingShutdown,
 			})
 
 			return app, nil

index-builder/app/index-builder/app.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/streamingfast/bstream"
 	"github.com/streamingfast/dgrpc"
+	dgrpcserver "github.com/streamingfast/dgrpc/server"
+	dgrpcfactory "github.com/streamingfast/dgrpc/server/factory"
 	"github.com/streamingfast/dmetrics"
 	"github.com/streamingfast/dstore"
 	index_builder "github.com/streamingfast/firehose-core/index-builder"
@@ -22,6 +24,8 @@ type Config struct {
 	MergedBlocksStoreURL string
 	ForkedBlocksStoreURL string
 	GRPCListenAddr       string
+
+	IsPendingShutdown func() bool `json:"-"`
 }
 
 type App struct {
@@ -72,12 +76,35 @@ func (a *App) Run() error {
 	a.OnTerminating(indexBuilder.Shutdown)
 	indexBuilder.OnTerminated(a.Shutdown)
 
+	a.startGRPCServer()
+
 	go indexBuilder.Launch()
 
 	zlog.Info("index builder running")
 	return nil
 }
 
+// startGRPCServer serves the health check over gRPC and HTTP (/healthz) on GRPCListenAddr.
+func (a *App) startGRPCServer() {
+	gs := dgrpcfactory.ServerFromOptions(
+		dgrpcserver.WithLogger(zlog),
+		dgrpcserver.WithHealthCheck(dgrpcserver.HealthCheckOverGRPC|dgrpcserver.HealthCheckOverHTTP, a.healthCheck),
+	)
+	gs.OnTerminated(a.Shutdown)
+	a.OnTerminating(func(_ error) {
+		gs.Shutdown(0)
+	})
+
+	go gs.Launch(a.config.GRPCListenAddr)
+}
+
+func (a *App) healthCheck(_ context.Context) (isReady bool, out interface{}, err error) {
+	if a.config.IsPendingShutdown != nil && a.config.IsPendingShutdown() {
+		return false, nil, nil
+	}
+	return true, nil, nil
+}
+
 func (a *App) IsReady() bool {
 	if a.readinessProbe == nil {
 		return false

index-builder/app/index-builder/app_test.go

@@ -0,0 +1,46 @@
+package index_builder
+
+import (
+	"context"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/streamingfast/bstream"
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApp_ServesHealthCheck(t *testing.T) {
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	addr := listener.Addr().String()
+	require.NoError(t, listener.Close())
+
+	app := New(&Config{
+		BlockHandler:         bstream.HandlerFunc(func(*pbbstream.Block, interface{}) error { return nil }),
+		StartBlockResolver:   func(context.Context) (uint64, error) { return 0, nil },
+		MergedBlocksStoreURL: "file://" + t.TempDir(),
+		GRPCListenAddr:       addr,
+	})
+	require.NoError(t, app.Run())
+	defer app.Shutdown(nil)
+
+	// gRPC health check, which IsReady queries
+	require.Eventually(t, app.IsReady, 10*time.Second, 100*time.Millisecond)
+
+	// HTTP /healthz on the same address
+	resp, err := http.Get("http://" + addr + "/healthz")
+	require.NoError(t, err)
+	defer resp.Body.Close()
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+}
+
+func TestApp_HealthCheckNotReadyWhenPendingShutdown(t *testing.T) {
+	app := New(&Config{IsPendingShutdown: func() bool { return true }})
+
+	isReady, _, err := app.healthCheck(context.Background())
+	require.NoError(t, err)
+	require.False(t, isReady)
+}