diff --git a/docs/release-notes/change-log.md b/docs/release-notes/change-log.md index ae395b060..fcd94b524 100644 --- a/docs/release-notes/change-log.md +++ b/docs/release-notes/change-log.md @@ -16,9 +16,19 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), - Fix tier1 requests rejected before their body is read (authentication, compression enforcement) sometimes failing at a load balancer with HTTP 502 or `INTERNAL` instead of returning their error. +- Demoted the tier2 "refusing Substreams ProcessRange request" log from `Info` to `Debug`. It was the + single largest source of Cloud Logging volume on `substreams-workers` under backpressure; the + `substreams_tier2_rejected_request_counter` metric (by reason) already tracks refusal rate without it. + +- Demoted tier2's gRPC "finished streaming call with code ResourceExhausted" completion log from `Info` + to `Debug`, using the newly added `dgrpc` `server.WithCodeLevelFunc` override. It was the second largest + source of Cloud Logging volume on `substreams-workers`, logged on every backpressure refusal alongside + the line above. + ### Dependencies - Bumped `github.com/streamingfast/firehose-networks` to v0.3.0. +- Bumped `github.com/streamingfast/dgrpc` to pick up `server.WithCodeLevelFunc`. ## v1.23.0 diff --git a/go.mod b/go.mod index 05574e2bf..044b465ef 100644 --- a/go.mod +++ b/go.mod @@ -12,7 +12,7 @@ require ( github.com/streamingfast/dauth v0.0.0-20260304175046-02898e30442d github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c github.com/streamingfast/derr v0.0.0-20250814163534-bd7407bd89d7 - github.com/streamingfast/dgrpc v0.0.0-20260801042045-4c4ac6a8e41a + github.com/streamingfast/dgrpc v0.0.0-20260929191947-99e20897ecc7 github.com/streamingfast/dhttp v0.1.3-0.20251218140957-6d46b8f12eb1 github.com/streamingfast/dstore v0.2.4-0.20260825160629-56e87480522c github.com/streamingfast/logging v1.2.3-0.20260810132752-360563ac68a9 diff --git a/go.sum b/go.sum index 73a2f978a..5564e98ee 100644 --- a/go.sum +++ b/go.sum @@ -661,8 +661,8 @@ github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c h1:6WjE2yInE+ github.com/streamingfast/dbin v0.9.1-0.20231117225723-59790c798e2c/go.mod h1:dbfiy9ORrL8c6ldSq+L0H9pg8TOqqu/FsghsgUEWK54= github.com/streamingfast/derr v0.0.0-20250814163534-bd7407bd89d7 h1:KeyhmbdPhEzot+On1XrHSAOmRoeyIRiM0EkGQAqnVxI= github.com/streamingfast/derr v0.0.0-20250814163534-bd7407bd89d7/go.mod h1:QSm/AfaDsE0k1xBYi0lW580YJ/WDV/FKZI628tkZR0Y= -github.com/streamingfast/dgrpc v0.0.0-20260801042045-4c4ac6a8e41a h1:OAh7uS72Z4zBSd8zuR6HoNtFKhR+EaWbJEsVMOgu6VE= -github.com/streamingfast/dgrpc v0.0.0-20260801042045-4c4ac6a8e41a/go.mod h1:bY6GwaM+J+IonNtsYFI9vY214hypoP/3JWI0YIgasOY= +github.com/streamingfast/dgrpc v0.0.0-20260929191947-99e20897ecc7 h1:66e3kGlDJco8hZ7x+9fblUeorismkWpcOsWgvCe7Y0M= +github.com/streamingfast/dgrpc v0.0.0-20260929191947-99e20897ecc7/go.mod h1:bY6GwaM+J+IonNtsYFI9vY214hypoP/3JWI0YIgasOY= github.com/streamingfast/dhammer v0.0.0-20220506192416-3797a7906da2 h1:/mcLVdwy6NeHWfJwuh2GD4+FMfPa59fkfM15sl8Jejk= github.com/streamingfast/dhammer v0.0.0-20220506192416-3797a7906da2/go.mod h1:MyG3U4ABuf7ANS8tix+e8UUevN7B9juhEnAbslS/X3M= github.com/streamingfast/dhttp v0.1.3-0.20251218140957-6d46b8f12eb1 h1:V9oSpY69/OZKTptZEPzJFaC8RS0Y8kffknNTDSZwpME= diff --git a/service/server.go b/service/server.go index cbac3a268..957c82788 100644 --- a/service/server.go +++ b/service/server.go @@ -23,9 +23,11 @@ import ( "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "go.opentelemetry.io/otel" "go.uber.org/zap" + "go.uber.org/zap/zapcore" "golang.org/x/net/http2" "golang.org/x/net/http2/h2c" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/encoding" _ "google.golang.org/grpc/experimental" ) @@ -159,6 +161,9 @@ func ListenTier2( regexp.MustCompile(`TLS handshake error.*first record does not look like a TLS handshake`), regexp.MustCompile(`http2: server: error reading preface.*connection reset by peer`), ), + // ResourceExhausted is routine admission-control backpressure on tier2, not worth Info + // on every refusal; everything else keeps dgrpc's normal-verbosity mapping. + dgrpcserver.WithCodeLevelFunc(tier2CodeLevel), } if enforceCompression { options = append(options, dgrpcserver.WithEnforceCompression()) @@ -197,3 +202,19 @@ func ListenTier2( return } + +// tier2CodeLevel mirrors dgrpc's normal-verbosity gRPC-code-to-zap-level mapping, except +// codes.ResourceExhausted is demoted to Debug: on tier2 it signals routine admission-control +// backpressure, not an abnormal condition worth Info on every refusal. +func tier2CodeLevel(code codes.Code) zapcore.Level { + switch code { + case codes.ResourceExhausted: + return zap.DebugLevel + case codes.OK, codes.Canceled, codes.InvalidArgument, codes.NotFound, codes.PermissionDenied, codes.Unauthenticated: + return zap.DebugLevel + case codes.Unknown, codes.Unimplemented, codes.Internal, codes.DataLoss: + return zap.ErrorLevel + default: + return zap.InfoLevel + } +} diff --git a/service/tier2.go b/service/tier2.go index 1f78f1e73..ac201892f 100644 --- a/service/tier2.go +++ b/service/tier2.go @@ -273,7 +273,7 @@ func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, s if s.isOverloaded() { err := status.Error(codes.ResourceExhausted, "service currently overloaded") fields = append(fields, zap.Error(err)) - logger.Info("refusing Substreams ProcessRange request", fields...) + logger.Debug("refusing Substreams ProcessRange request", fields...) return err } @@ -297,7 +297,7 @@ func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, s if request.Modules == nil { err := status.Error(codes.InvalidArgument, "missing modules in request") fields = append(fields, zap.Error(err)) - logger.Info("refusing Substreams ProcessRange request", fields...) + logger.Debug("refusing Substreams ProcessRange request", fields...) return err } moduleNames := make([]string, len(request.Modules.Modules)) @@ -309,7 +309,7 @@ func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, s if err := ValidateTier2Request(request); err != nil { err = status.Errorf(codes.InvalidArgument, "validate request: %s", err) fields = append(fields, zap.Error(err)) - logger.Info("refusing Substreams ProcessRange request", fields...) + logger.Debug("refusing Substreams ProcessRange request", fields...) return err } @@ -317,7 +317,7 @@ func (s *Tier2Service) ProcessRange(request *pbssinternal.ProcessRangeRequest, s if err != nil { err = bsstream.NewErrInvalidArg("%s", err.Error()) fields = append(fields, zap.Error(err)) - logger.Info("refusing Substreams ProcessRange request", fields...) + logger.Debug("refusing Substreams ProcessRange request", fields...) return err } outputModuleHash := execGraph.ModuleHashes()[request.OutputModule]