From 797d9874354643612519cdf1ff07d8e8140fdca0 Mon Sep 17 00:00:00 2001 From: YaroShkvorets Date: Fri, 9 Oct 2026 14:01:46 -0400 Subject: [PATCH 1/2] index-builder: serve gRPC health and HTTP /healthz The index builder never listened on --index-builder-grpc-listen-addr: its health handlers were never registered on a server, and App.Run only dialed that address for IsReady(). Start the gRPC server and register the health service, as the merger does. Add --index-builder-http-healthz-addr (default :10019), the same HTTP /healthz the merger and relayer expose since v1.14.4. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 ++ cmd/apps/index_builder.go | 13 +++--- cmd/main.go | 1 + cmd/shift_ports.go | 1 + constants.go | 1 + index-builder/app/index-builder/app.go | 42 ++++++++++++++++++ index-builder/app/index-builder/app_test.go | 47 +++++++++++++++++++++ index-builder/healthz.go | 11 +++++ index-builder/index-builder.go | 17 +++++--- index-builder/server.go | 20 +++++++++ index-builder/server_test.go | 45 ++++++++++++++++++++ shift_ports_test.go | 6 +++ 12 files changed, 197 insertions(+), 11 deletions(-) create mode 100644 index-builder/app/index-builder/app_test.go create mode 100644 index-builder/server.go create mode 100644 index-builder/server_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c2e61707..030a39a7 100644 --- a/CHANGELOG.md +++ b/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 the gRPC health check 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. @@ -30,6 +32,8 @@ If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you s ### Added +- `index-builder` now exposes HTTP `/healthz` on `--index-builder-http-healthz-addr` (default `:10019`), like the merger and relayer. + - Store URLs, such as `--common-merged-blocks-store-url`, accept `compression_config` to tune how files are written, matching the store's compression: a zstd level with an optional window in MiB (`best`, `better/32`), or a gzip level from `1` to `9`. For example `gs://bucket/merged-blocks?compression_config=best/32`. Files written with any setting are read back without configuration. An invalid value makes opening the store fail. - On zstd stores, `compression_config` also takes decoder settings, comma separated after the level or alone: `lowmem=false` gives each decoder a history buffer of twice the window, and `pool=` reuses decoders across reads and across every store using that name, in place of the `default` pool. `pool=none` gives each file a decoder of its own. For example `--common-merged-blocks-store-url=gs://bucket/merged-blocks?compression_config=best/32,lowmem=false,pool=blocks` and `--substreams-state-store-url=gs://bucket/states?compression_config=better/16,pool=cache`. Substreams tier2 gets them through the URLs tier1 sends. Reading 100 MiB `best/32` merged blocks files 10 at a time, adding `lowmem=false` to the `default` pool used 70% less CPU and 57% more peak heap. `lowmem=false` only helps files larger than their window plus about 1 MiB. diff --git a/cmd/apps/index_builder.go b/cmd/apps/index_builder.go index 2e0c8efb..2017349d 100644 --- a/cmd/apps/index_builder.go +++ b/cmd/apps/index_builder.go @@ -22,6 +22,7 @@ func RegisterIndexBuilderApp[B firecore.Block](chain *firecore.Chain[B], rootLog Description: "App the builds indexes out of Firehose blocks", RegisterFlags: func(cmd *cobra.Command) error { cmd.Flags().String("index-builder-grpc-listen-addr", firecore.IndexBuilderServiceAddr, "Address to listen for grpc-based healthz check") + cmd.Flags().String("index-builder-http-healthz-addr", firecore.IndexBuilderHTTPHealthzAddr, "Address to listen on for the HTTP /healthz endpoint. Set to an empty string to disable. Returns 200 when ready, 503 otherwise.") cmd.Flags().Uint64("index-builder-index-size", 10000, "Size of index bundles that will be created") cmd.Flags().Uint64("index-builder-start-block", 0, "Block number to start indexing") cmd.Flags().Uint64("index-builder-stop-block", 0, "Block number to stop indexing") @@ -81,11 +82,13 @@ func RegisterIndexBuilderApp[B firecore.Block](chain *firecore.Chain[B], rootLog }) app := index_builder.New(&index_builder.Config{ - BlockHandler: handler, - StartBlockResolver: startBlockResolver, - EndBlock: stopBlockNum, - MergedBlocksStoreURL: mergedBlocksStoreURL, - GRPCListenAddr: viper.GetString("index-builder-grpc-listen-addr"), + BlockHandler: handler, + StartBlockResolver: startBlockResolver, + EndBlock: stopBlockNum, + MergedBlocksStoreURL: mergedBlocksStoreURL, + GRPCListenAddr: viper.GetString("index-builder-grpc-listen-addr"), + HTTPHealthzListenAddr: viper.GetString("index-builder-http-healthz-addr"), + IsPendingShutdown: runtime.IsPendingShutdown, }) return app, nil diff --git a/cmd/main.go b/cmd/main.go index cefba624..dbe7024d 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -125,6 +125,7 @@ func Main[B firecore.Block](chain *firecore.Chain[B]) { --substreams-tier1-grpc-listen-addr :10016 --substreams-tier2-grpc-listen-addr :10017 --index-builder-grpc-listen-addr :10009 + --index-builder-http-healthz-addr :10019 Client connection addresses: --common-live-blocks-addr :10014 diff --git a/cmd/shift_ports.go b/cmd/shift_ports.go index fe35e063..afd9b406 100644 --- a/cmd/shift_ports.go +++ b/cmd/shift_ports.go @@ -25,6 +25,7 @@ var portFlags = []string{ "substreams-tier1-grpc-listen-addr", "substreams-tier2-grpc-listen-addr", "index-builder-grpc-listen-addr", + "index-builder-http-healthz-addr", // Client-side connection addresses (defaults point to services above) "common-live-blocks-addr", diff --git a/constants.go b/constants.go index deaf7d84..1df70e89 100644 --- a/constants.go +++ b/constants.go @@ -18,6 +18,7 @@ var ( SubstreamsTier1GRPCServingAddr string = ":10016" SubstreamsTier2GRPCServingAddr string = ":10017" RelayerHTTPHealthzAddr string = ":10018" + IndexBuilderHTTPHealthzAddr string = ":10019" // Data storage default locations BlocksCacheDirectory string = "file://{data-dir}/storage/blocks-cache" diff --git a/index-builder/app/index-builder/app.go b/index-builder/app/index-builder/app.go index dbfd6346..57baf3d8 100644 --- a/index-builder/app/index-builder/app.go +++ b/index-builder/app/index-builder/app.go @@ -3,6 +3,8 @@ package index_builder import ( "context" "fmt" + "net/http" + "time" "github.com/streamingfast/bstream" "github.com/streamingfast/dgrpc" @@ -22,6 +24,10 @@ type Config struct { MergedBlocksStoreURL string ForkedBlocksStoreURL string GRPCListenAddr string + + HTTPHealthzListenAddr string + + IsPendingShutdown func() bool `json:"-"` } type App struct { @@ -59,6 +65,7 @@ func (a *App) Run() error { startBlock, a.config.EndBlock, blockStore, + a.config.GRPCListenAddr, ) gs, err := dgrpc.NewInternalClient(a.config.GRPCListenAddr) @@ -72,12 +79,47 @@ func (a *App) Run() error { a.OnTerminating(indexBuilder.Shutdown) indexBuilder.OnTerminated(a.Shutdown) + if a.config.HTTPHealthzListenAddr != "" { + a.startHTTPHealthzServer(indexBuilder) + } + go indexBuilder.Launch() zlog.Info("index builder running") return nil } +func (a *App) startHTTPHealthzServer(indexBuilder *index_builder.IndexBuilder) { + mux := http.NewServeMux() + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + if a.config.IsPendingShutdown != nil && a.config.IsPendingShutdown() { + http.Error(w, "not ready: shutting down", http.StatusServiceUnavailable) + return + } + resp, err := indexBuilder.Check(context.Background(), &pbhealth.HealthCheckRequest{}) + if err != nil || resp.Status != pbhealth.HealthCheckResponse_SERVING { + http.Error(w, "not ready", http.StatusServiceUnavailable) + return + } + w.Write([]byte("ready\n")) + }) + + srv := &http.Server{Addr: a.config.HTTPHealthzListenAddr, Handler: mux} + a.OnTerminating(func(_ error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = srv.Shutdown(ctx) + }) + + zlog.Info("starting index builder http healthz server", zap.String("addr", a.config.HTTPHealthzListenAddr)) + go func() { + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + zlog.Error("index builder http healthz server failed", zap.Error(err)) + a.Shutdown(err) + } + }() +} + func (a *App) IsReady() bool { if a.readinessProbe == nil { return false diff --git a/index-builder/app/index-builder/app_test.go b/index-builder/app/index-builder/app_test.go new file mode 100644 index 00000000..8fa73ede --- /dev/null +++ b/index-builder/app/index-builder/app_test.go @@ -0,0 +1,47 @@ +package index_builder + +import ( + "context" + "io" + "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_ServesHTTPHealthz(t *testing.T) { + 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: freeAddr(t), + HTTPHealthzListenAddr: freeAddr(t), + }) + require.NoError(t, app.Run()) + defer app.Shutdown(nil) + + url := "http://" + app.config.HTTPHealthzListenAddr + "/healthz" + require.Eventually(t, func() bool { + resp, err := http.Get(url) + if err != nil { + return false + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + return resp.StatusCode == http.StatusOK && string(body) == "ready\n" + }, 10*time.Second, 100*time.Millisecond) + + require.Eventually(t, app.IsReady, 10*time.Second, 100*time.Millisecond) +} + +func freeAddr(t *testing.T) string { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + defer listener.Close() + return listener.Addr().String() +} diff --git a/index-builder/healthz.go b/index-builder/healthz.go index 63b69e0c..65b1a959 100644 --- a/index-builder/healthz.go +++ b/index-builder/healthz.go @@ -28,6 +28,17 @@ func (app *IndexBuilder) Check(ctx context.Context, in *pbhealth.HealthCheckRequ }, nil } +func (app *IndexBuilder) List(ctx context.Context, in *pbhealth.HealthListRequest) (*pbhealth.HealthListResponse, error) { + status := pbhealth.HealthCheckResponse_SERVING + return &pbhealth.HealthListResponse{ + Statuses: map[string]*pbhealth.HealthCheckResponse{ + "index-builder": &pbhealth.HealthCheckResponse{ + Status: status, + }, + }, + }, nil +} + // Watch is basic GRPC Healthcheck as a stream func (app *IndexBuilder) Watch(req *pbhealth.HealthCheckRequest, stream pbhealth.Health_WatchServer) error { err := stream.Send(&pbhealth.HealthCheckResponse{ diff --git a/index-builder/index-builder.go b/index-builder/index-builder.go index 6ecfc6fe..5de14fa2 100644 --- a/index-builder/index-builder.go +++ b/index-builder/index-builder.go @@ -27,21 +27,26 @@ type IndexBuilder struct { handler bstream.Handler blocksStore dstore.Store + + grpcListenAddr string } -func NewIndexBuilder(logger *zap.Logger, handler bstream.Handler, startBlockNum, stopBlockNum uint64, blockStore dstore.Store) *IndexBuilder { +func NewIndexBuilder(logger *zap.Logger, handler bstream.Handler, startBlockNum, stopBlockNum uint64, blockStore dstore.Store, grpcListenAddr string) *IndexBuilder { return &IndexBuilder{ - Shutter: shutter.New(), - startBlockNum: startBlockNum, - stopBlockNum: stopBlockNum, - handler: handler, - blocksStore: blockStore, + Shutter: shutter.New(), + startBlockNum: startBlockNum, + stopBlockNum: stopBlockNum, + handler: handler, + blocksStore: blockStore, + grpcListenAddr: grpcListenAddr, logger: logger, } } func (app *IndexBuilder) Launch() { + app.startGRPCServer() + err := app.launch() if errors.Is(err, stream.ErrStopBlockReached) { app.logger.Info("index builder reached stop block", zap.Uint64("stop_block_num", app.stopBlockNum)) diff --git a/index-builder/server.go b/index-builder/server.go new file mode 100644 index 00000000..4bf5972e --- /dev/null +++ b/index-builder/server.go @@ -0,0 +1,20 @@ +package index_builder + +import ( + dgrpcfactory "github.com/streamingfast/dgrpc/server/factory" + pbhealth "google.golang.org/grpc/health/grpc_health_v1" +) + +func (app *IndexBuilder) startGRPCServer() { + gs := dgrpcfactory.ServerFromOptions() + gs.OnTerminated(app.Shutdown) + app.logger.Info("grpc server created") + + app.OnTerminated(func(_ error) { + gs.Shutdown(0) + }) + pbhealth.RegisterHealthServer(gs.ServiceRegistrar(), app) + app.logger.Info("server registered") + + go gs.Launch(app.grpcListenAddr) +} diff --git a/index-builder/server_test.go b/index-builder/server_test.go new file mode 100644 index 00000000..8a8f18b5 --- /dev/null +++ b/index-builder/server_test.go @@ -0,0 +1,45 @@ +package index_builder + +import ( + "context" + "net" + "testing" + "time" + + "github.com/streamingfast/bstream" + pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" + "github.com/streamingfast/dstore" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + pbhealth "google.golang.org/grpc/health/grpc_health_v1" +) + +func TestIndexBuilder_ServesGRPCHealth(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()) + + store, err := dstore.NewDBinStore("file://" + t.TempDir()) + require.NoError(t, err) + + handler := bstream.HandlerFunc(func(*pbbstream.Block, interface{}) error { return nil }) + app := NewIndexBuilder(zap.NewNop(), handler, 0, 0, store, addr) + go app.Launch() + defer app.Shutdown(nil) + + conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + require.NoError(t, err) + defer conn.Close() + + client := pbhealth.NewHealthClient(conn) + require.Eventually(t, func() bool { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + resp, err := client.Check(ctx, &pbhealth.HealthCheckRequest{}) + return err == nil && resp.Status == pbhealth.HealthCheckResponse_SERVING + }, 10*time.Second, 100*time.Millisecond) +} diff --git a/shift_ports_test.go b/shift_ports_test.go index 1c8bf464..fb4201bf 100644 --- a/shift_ports_test.go +++ b/shift_ports_test.go @@ -62,6 +62,12 @@ func TestShiftAddressPort(t *testing.T) { offset: 100, expected: ":10109", }, + { + name: "index builder http healthz default port", + addr: ":10019", + offset: 100, + expected: ":10119", + }, { name: "reader node grpc default port", addr: ":10010", From 9e97688478631c079863011681055d2675156e01 Mon Sep 17 00:00:00 2001 From: YaroShkvorets Date: Fri, 9 Oct 2026 17:54:47 -0400 Subject: [PATCH 2/2] index-builder: serve health on the gRPC listen addr via dgrpc Per review: drop the dedicated --index-builder-http-healthz-addr and let dgrpc's WithHealthCheck serve both gRPC health and HTTP /healthz on --index-builder-grpc-listen-addr. Removes the unused hand-written health handlers and keeps NewIndexBuilder's signature unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 4 +- cmd/apps/index_builder.go | 14 +++--- cmd/main.go | 1 - cmd/shift_ports.go | 1 - constants.go | 1 - index-builder/app/index-builder/app.go | 53 ++++++++------------- index-builder/app/index-builder/app_test.go | 45 +++++++++-------- index-builder/healthz.go | 53 --------------------- index-builder/index-builder.go | 17 +++---- index-builder/server.go | 20 -------- index-builder/server_test.go | 45 ----------------- shift_ports_test.go | 6 --- 12 files changed, 54 insertions(+), 206 deletions(-) delete mode 100644 index-builder/healthz.go delete mode 100644 index-builder/server.go delete mode 100644 index-builder/server_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 030a39a7..7d51d966 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you s ### Fixed -- `index-builder` now serves the gRPC health check on `--index-builder-grpc-listen-addr`, which it never listened on. +- `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. @@ -32,8 +32,6 @@ If you were at `firehose-core` version `1.0.0` and are bumping to `1.1.0`, you s ### Added -- `index-builder` now exposes HTTP `/healthz` on `--index-builder-http-healthz-addr` (default `:10019`), like the merger and relayer. - - Store URLs, such as `--common-merged-blocks-store-url`, accept `compression_config` to tune how files are written, matching the store's compression: a zstd level with an optional window in MiB (`best`, `better/32`), or a gzip level from `1` to `9`. For example `gs://bucket/merged-blocks?compression_config=best/32`. Files written with any setting are read back without configuration. An invalid value makes opening the store fail. - On zstd stores, `compression_config` also takes decoder settings, comma separated after the level or alone: `lowmem=false` gives each decoder a history buffer of twice the window, and `pool=` reuses decoders across reads and across every store using that name, in place of the `default` pool. `pool=none` gives each file a decoder of its own. For example `--common-merged-blocks-store-url=gs://bucket/merged-blocks?compression_config=best/32,lowmem=false,pool=blocks` and `--substreams-state-store-url=gs://bucket/states?compression_config=better/16,pool=cache`. Substreams tier2 gets them through the URLs tier1 sends. Reading 100 MiB `best/32` merged blocks files 10 at a time, adding `lowmem=false` to the `default` pool used 70% less CPU and 57% more peak heap. `lowmem=false` only helps files larger than their window plus about 1 MiB. diff --git a/cmd/apps/index_builder.go b/cmd/apps/index_builder.go index 2017349d..7039c91a 100644 --- a/cmd/apps/index_builder.go +++ b/cmd/apps/index_builder.go @@ -22,7 +22,6 @@ func RegisterIndexBuilderApp[B firecore.Block](chain *firecore.Chain[B], rootLog Description: "App the builds indexes out of Firehose blocks", RegisterFlags: func(cmd *cobra.Command) error { cmd.Flags().String("index-builder-grpc-listen-addr", firecore.IndexBuilderServiceAddr, "Address to listen for grpc-based healthz check") - cmd.Flags().String("index-builder-http-healthz-addr", firecore.IndexBuilderHTTPHealthzAddr, "Address to listen on for the HTTP /healthz endpoint. Set to an empty string to disable. Returns 200 when ready, 503 otherwise.") cmd.Flags().Uint64("index-builder-index-size", 10000, "Size of index bundles that will be created") cmd.Flags().Uint64("index-builder-start-block", 0, "Block number to start indexing") cmd.Flags().Uint64("index-builder-stop-block", 0, "Block number to stop indexing") @@ -82,13 +81,12 @@ func RegisterIndexBuilderApp[B firecore.Block](chain *firecore.Chain[B], rootLog }) app := index_builder.New(&index_builder.Config{ - BlockHandler: handler, - StartBlockResolver: startBlockResolver, - EndBlock: stopBlockNum, - MergedBlocksStoreURL: mergedBlocksStoreURL, - GRPCListenAddr: viper.GetString("index-builder-grpc-listen-addr"), - HTTPHealthzListenAddr: viper.GetString("index-builder-http-healthz-addr"), - IsPendingShutdown: runtime.IsPendingShutdown, + BlockHandler: handler, + StartBlockResolver: startBlockResolver, + EndBlock: stopBlockNum, + MergedBlocksStoreURL: mergedBlocksStoreURL, + GRPCListenAddr: viper.GetString("index-builder-grpc-listen-addr"), + IsPendingShutdown: runtime.IsPendingShutdown, }) return app, nil diff --git a/cmd/main.go b/cmd/main.go index dbe7024d..cefba624 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -125,7 +125,6 @@ func Main[B firecore.Block](chain *firecore.Chain[B]) { --substreams-tier1-grpc-listen-addr :10016 --substreams-tier2-grpc-listen-addr :10017 --index-builder-grpc-listen-addr :10009 - --index-builder-http-healthz-addr :10019 Client connection addresses: --common-live-blocks-addr :10014 diff --git a/cmd/shift_ports.go b/cmd/shift_ports.go index afd9b406..fe35e063 100644 --- a/cmd/shift_ports.go +++ b/cmd/shift_ports.go @@ -25,7 +25,6 @@ var portFlags = []string{ "substreams-tier1-grpc-listen-addr", "substreams-tier2-grpc-listen-addr", "index-builder-grpc-listen-addr", - "index-builder-http-healthz-addr", // Client-side connection addresses (defaults point to services above) "common-live-blocks-addr", diff --git a/constants.go b/constants.go index 1df70e89..deaf7d84 100644 --- a/constants.go +++ b/constants.go @@ -18,7 +18,6 @@ var ( SubstreamsTier1GRPCServingAddr string = ":10016" SubstreamsTier2GRPCServingAddr string = ":10017" RelayerHTTPHealthzAddr string = ":10018" - IndexBuilderHTTPHealthzAddr string = ":10019" // Data storage default locations BlocksCacheDirectory string = "file://{data-dir}/storage/blocks-cache" diff --git a/index-builder/app/index-builder/app.go b/index-builder/app/index-builder/app.go index 57baf3d8..088803de 100644 --- a/index-builder/app/index-builder/app.go +++ b/index-builder/app/index-builder/app.go @@ -3,11 +3,11 @@ package index_builder import ( "context" "fmt" - "net/http" - "time" "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" @@ -25,8 +25,6 @@ type Config struct { ForkedBlocksStoreURL string GRPCListenAddr string - HTTPHealthzListenAddr string - IsPendingShutdown func() bool `json:"-"` } @@ -65,7 +63,6 @@ func (a *App) Run() error { startBlock, a.config.EndBlock, blockStore, - a.config.GRPCListenAddr, ) gs, err := dgrpc.NewInternalClient(a.config.GRPCListenAddr) @@ -79,9 +76,7 @@ func (a *App) Run() error { a.OnTerminating(indexBuilder.Shutdown) indexBuilder.OnTerminated(a.Shutdown) - if a.config.HTTPHealthzListenAddr != "" { - a.startHTTPHealthzServer(indexBuilder) - } + a.startGRPCServer() go indexBuilder.Launch() @@ -89,35 +84,25 @@ func (a *App) Run() error { return nil } -func (a *App) startHTTPHealthzServer(indexBuilder *index_builder.IndexBuilder) { - mux := http.NewServeMux() - mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { - if a.config.IsPendingShutdown != nil && a.config.IsPendingShutdown() { - http.Error(w, "not ready: shutting down", http.StatusServiceUnavailable) - return - } - resp, err := indexBuilder.Check(context.Background(), &pbhealth.HealthCheckRequest{}) - if err != nil || resp.Status != pbhealth.HealthCheckResponse_SERVING { - http.Error(w, "not ready", http.StatusServiceUnavailable) - return - } - w.Write([]byte("ready\n")) - }) - - srv := &http.Server{Addr: a.config.HTTPHealthzListenAddr, Handler: mux} +// 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) { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - _ = srv.Shutdown(ctx) + gs.Shutdown(0) }) - zlog.Info("starting index builder http healthz server", zap.String("addr", a.config.HTTPHealthzListenAddr)) - go func() { - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - zlog.Error("index builder http healthz server failed", zap.Error(err)) - a.Shutdown(err) - } - }() + 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 { diff --git a/index-builder/app/index-builder/app_test.go b/index-builder/app/index-builder/app_test.go index 8fa73ede..976683d9 100644 --- a/index-builder/app/index-builder/app_test.go +++ b/index-builder/app/index-builder/app_test.go @@ -2,7 +2,6 @@ package index_builder import ( "context" - "io" "net" "net/http" "testing" @@ -13,35 +12,35 @@ import ( "github.com/stretchr/testify/require" ) -func TestApp_ServesHTTPHealthz(t *testing.T) { +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: freeAddr(t), - HTTPHealthzListenAddr: freeAddr(t), + 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) - url := "http://" + app.config.HTTPHealthzListenAddr + "/healthz" - require.Eventually(t, func() bool { - resp, err := http.Get(url) - if err != nil { - return false - } - defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) - return resp.StatusCode == http.StatusOK && string(body) == "ready\n" - }, 10*time.Second, 100*time.Millisecond) - + // 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 freeAddr(t *testing.T) string { - t.Helper() - listener, err := net.Listen("tcp", "127.0.0.1:0") +func TestApp_HealthCheckNotReadyWhenPendingShutdown(t *testing.T) { + app := New(&Config{IsPendingShutdown: func() bool { return true }}) + + isReady, _, err := app.healthCheck(context.Background()) require.NoError(t, err) - defer listener.Close() - return listener.Addr().String() + require.False(t, isReady) } diff --git a/index-builder/healthz.go b/index-builder/healthz.go deleted file mode 100644 index 65b1a959..00000000 --- a/index-builder/healthz.go +++ /dev/null @@ -1,53 +0,0 @@ -package index_builder - -// Copyright 2019 dfuse Platform Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -import ( - "context" - - pbhealth "google.golang.org/grpc/health/grpc_health_v1" -) - -// Check is basic GRPC Healthcheck -func (app *IndexBuilder) Check(ctx context.Context, in *pbhealth.HealthCheckRequest) (*pbhealth.HealthCheckResponse, error) { - status := pbhealth.HealthCheckResponse_SERVING - return &pbhealth.HealthCheckResponse{ - Status: status, - }, nil -} - -func (app *IndexBuilder) List(ctx context.Context, in *pbhealth.HealthListRequest) (*pbhealth.HealthListResponse, error) { - status := pbhealth.HealthCheckResponse_SERVING - return &pbhealth.HealthListResponse{ - Statuses: map[string]*pbhealth.HealthCheckResponse{ - "index-builder": &pbhealth.HealthCheckResponse{ - Status: status, - }, - }, - }, nil -} - -// Watch is basic GRPC Healthcheck as a stream -func (app *IndexBuilder) Watch(req *pbhealth.HealthCheckRequest, stream pbhealth.Health_WatchServer) error { - err := stream.Send(&pbhealth.HealthCheckResponse{ - Status: pbhealth.HealthCheckResponse_SERVING, - }) - if err != nil { - return err - } - - <-stream.Context().Done() - return nil -} diff --git a/index-builder/index-builder.go b/index-builder/index-builder.go index 5de14fa2..6ecfc6fe 100644 --- a/index-builder/index-builder.go +++ b/index-builder/index-builder.go @@ -27,26 +27,21 @@ type IndexBuilder struct { handler bstream.Handler blocksStore dstore.Store - - grpcListenAddr string } -func NewIndexBuilder(logger *zap.Logger, handler bstream.Handler, startBlockNum, stopBlockNum uint64, blockStore dstore.Store, grpcListenAddr string) *IndexBuilder { +func NewIndexBuilder(logger *zap.Logger, handler bstream.Handler, startBlockNum, stopBlockNum uint64, blockStore dstore.Store) *IndexBuilder { return &IndexBuilder{ - Shutter: shutter.New(), - startBlockNum: startBlockNum, - stopBlockNum: stopBlockNum, - handler: handler, - blocksStore: blockStore, - grpcListenAddr: grpcListenAddr, + Shutter: shutter.New(), + startBlockNum: startBlockNum, + stopBlockNum: stopBlockNum, + handler: handler, + blocksStore: blockStore, logger: logger, } } func (app *IndexBuilder) Launch() { - app.startGRPCServer() - err := app.launch() if errors.Is(err, stream.ErrStopBlockReached) { app.logger.Info("index builder reached stop block", zap.Uint64("stop_block_num", app.stopBlockNum)) diff --git a/index-builder/server.go b/index-builder/server.go deleted file mode 100644 index 4bf5972e..00000000 --- a/index-builder/server.go +++ /dev/null @@ -1,20 +0,0 @@ -package index_builder - -import ( - dgrpcfactory "github.com/streamingfast/dgrpc/server/factory" - pbhealth "google.golang.org/grpc/health/grpc_health_v1" -) - -func (app *IndexBuilder) startGRPCServer() { - gs := dgrpcfactory.ServerFromOptions() - gs.OnTerminated(app.Shutdown) - app.logger.Info("grpc server created") - - app.OnTerminated(func(_ error) { - gs.Shutdown(0) - }) - pbhealth.RegisterHealthServer(gs.ServiceRegistrar(), app) - app.logger.Info("server registered") - - go gs.Launch(app.grpcListenAddr) -} diff --git a/index-builder/server_test.go b/index-builder/server_test.go deleted file mode 100644 index 8a8f18b5..00000000 --- a/index-builder/server_test.go +++ /dev/null @@ -1,45 +0,0 @@ -package index_builder - -import ( - "context" - "net" - "testing" - "time" - - "github.com/streamingfast/bstream" - pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" - "github.com/streamingfast/dstore" - "github.com/stretchr/testify/require" - "go.uber.org/zap" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - pbhealth "google.golang.org/grpc/health/grpc_health_v1" -) - -func TestIndexBuilder_ServesGRPCHealth(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()) - - store, err := dstore.NewDBinStore("file://" + t.TempDir()) - require.NoError(t, err) - - handler := bstream.HandlerFunc(func(*pbbstream.Block, interface{}) error { return nil }) - app := NewIndexBuilder(zap.NewNop(), handler, 0, 0, store, addr) - go app.Launch() - defer app.Shutdown(nil) - - conn, err := grpc.NewClient(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) - require.NoError(t, err) - defer conn.Close() - - client := pbhealth.NewHealthClient(conn) - require.Eventually(t, func() bool { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - - resp, err := client.Check(ctx, &pbhealth.HealthCheckRequest{}) - return err == nil && resp.Status == pbhealth.HealthCheckResponse_SERVING - }, 10*time.Second, 100*time.Millisecond) -} diff --git a/shift_ports_test.go b/shift_ports_test.go index fb4201bf..1c8bf464 100644 --- a/shift_ports_test.go +++ b/shift_ports_test.go @@ -62,12 +62,6 @@ func TestShiftAddressPort(t *testing.T) { offset: 100, expected: ":10109", }, - { - name: "index builder http healthz default port", - addr: ":10019", - offset: 100, - expected: ":10119", - }, { name: "reader node grpc default port", addr: ":10010",