diff --git a/CHANGELOG.md b/CHANGELOG.md index c2e61707..7d51d966 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 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. diff --git a/cmd/apps/index_builder.go b/cmd/apps/index_builder.go index 2e0c8efb..7039c91a 100644 --- a/cmd/apps/index_builder.go +++ b/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 diff --git a/index-builder/app/index-builder/app.go b/index-builder/app/index-builder/app.go index dbfd6346..088803de 100644 --- a/index-builder/app/index-builder/app.go +++ b/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 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..976683d9 --- /dev/null +++ b/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) +} diff --git a/index-builder/healthz.go b/index-builder/healthz.go deleted file mode 100644 index 63b69e0c..00000000 --- a/index-builder/healthz.go +++ /dev/null @@ -1,42 +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 -} - -// 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 -}