diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go index 137dd2fb33..4c82864c92 100644 --- a/common/dynamicconfig/constants.go +++ b/common/dynamicconfig/constants.go @@ -3225,6 +3225,12 @@ WorkerActivitiesPerSecond, MaxConcurrentActivityTaskPollers. `MaxUserMetadataDetailsSize is the maximum size of user metadata details payloads in bytes.`, ) + MaxServiceErrorMessageLength = NewGlobalIntSetting( + "system.maxServiceErrorMessageLength", + 4000, + "MaxServiceErrorMessageLength is the max length of service error message. If it's longer, it will be truncated.", + ) + LogAllReqErrors = NewNamespaceBoolSetting( "system.logAllReqErrors", false, diff --git a/common/rpc/interceptor/service_error_interceptor.go b/common/rpc/interceptor/service_error_interceptor.go index ebd8957f4a..8a6f014d73 100644 --- a/common/rpc/interceptor/service_error_interceptor.go +++ b/common/rpc/interceptor/service_error_interceptor.go @@ -5,24 +5,33 @@ import ( "errors" "go.temporal.io/api/serviceerror" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/persistence/serialization" "go.temporal.io/server/common/util" "google.golang.org/grpc" "google.golang.org/grpc/status" ) -const ( - maxMessageLength = 4000 - truncatedSuffix = "... " -) +const truncatedSuffix = "... " -func ServiceErrorInterceptor( +type ServiceErrorInterceptor struct { + maxMessageLength dynamicconfig.IntPropertyFn +} + +func NewServiceErrorInterceptor( + maxMessageLength dynamicconfig.IntPropertyFn, +) *ServiceErrorInterceptor { + return &ServiceErrorInterceptor{ + maxMessageLength: maxMessageLength, + } +} + +func (i *ServiceErrorInterceptor) Intercept( ctx context.Context, req any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler, ) (any, error) { - resp, err := handler(ctx, req) var deserializationError *serialization.DeserializationError @@ -33,10 +42,11 @@ func ServiceErrorInterceptor( } // truncate message length if needed + maxLength := i.maxMessageLength() st := serviceerror.ToStatus(err) - if len(st.Message()) > maxMessageLength { + if len(st.Message()) > maxLength { p := st.Proto() - p.Message = util.TruncateUTF8(p.Message, maxMessageLength-len(truncatedSuffix)) + truncatedSuffix + p.Message = util.TruncateUTF8(p.Message, maxLength-len(truncatedSuffix)) + truncatedSuffix st = status.FromProto(p) } diff --git a/common/rpc/interceptor/service_error_interceptor_test.go b/common/rpc/interceptor/service_error_interceptor_test.go index f905277259..4b86d04d23 100644 --- a/common/rpc/interceptor/service_error_interceptor_test.go +++ b/common/rpc/interceptor/service_error_interceptor_test.go @@ -8,11 +8,14 @@ import ( "github.com/stretchr/testify/require" enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/persistence/serialization" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) +const testMaxMessageLength = 4000 + type UnaryHandler func(ctx context.Context, req any) (any, error) type ( @@ -28,8 +31,9 @@ func (e *ErrorWithoutStatus) Error() string { // Error returns string message. func TestServiceErrorInterceptorUnknown(t *testing.T) { + interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + _, err := interceptor.Intercept(t.Context(), nil, nil, func(ctx context.Context, req any) (any, error) { return nil, status.Error(codes.InvalidArgument, "invalid argument") }) @@ -37,7 +41,7 @@ func TestServiceErrorInterceptorUnknown(t *testing.T) { require.Error(t, err) require.Equal(t, codes.InvalidArgument, status.Code(err)) - _, err = ServiceErrorInterceptor(t.Context(), nil, nil, + _, err = interceptor.Intercept(t.Context(), nil, nil, func(ctx context.Context, req any) (any, error) { errWithoutStatus := &ErrorWithoutStatus{ Message: "unknown error without status", @@ -50,12 +54,13 @@ func TestServiceErrorInterceptorUnknown(t *testing.T) { } func TestServiceErrorInterceptorSer(t *testing.T) { + interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) serErrors := []error{ serialization.NewDeserializationError(enumspb.ENCODING_TYPE_PROTO3, nil), serialization.NewSerializationError(enumspb.ENCODING_TYPE_PROTO3, nil), } for _, inErr := range serErrors { - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, inErr }) @@ -64,8 +69,10 @@ func TestServiceErrorInterceptorSer(t *testing.T) { } func TestServiceErrorInterceptorTruncation(t *testing.T) { + interceptor := NewServiceErrorInterceptor(dynamicconfig.GetIntPropertyFn(testMaxMessageLength)) + t.Run("nil error is not affected", func(t *testing.T) { - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return "ok", nil }) @@ -74,7 +81,7 @@ func TestServiceErrorInterceptorTruncation(t *testing.T) { t.Run("short message is not truncated", func(t *testing.T) { msg := "short error" - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, serviceerror.NewInternal(msg) }) @@ -84,8 +91,8 @@ func TestServiceErrorInterceptorTruncation(t *testing.T) { }) t.Run("message at exact limit is not truncated", func(t *testing.T) { - msg := strings.Repeat("a", maxMessageLength) - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + msg := strings.Repeat("a", testMaxMessageLength) + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, serviceerror.NewInternal(msg) }) @@ -95,20 +102,20 @@ func TestServiceErrorInterceptorTruncation(t *testing.T) { }) t.Run("message over limit is truncated", func(t *testing.T) { - msg := strings.Repeat("a", maxMessageLength+100) - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + msg := strings.Repeat("a", testMaxMessageLength+100) + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, serviceerror.NewInternal(msg) }) require.Error(t, err) st := status.Convert(err) - require.LessOrEqual(t, len(st.Message()), maxMessageLength) + require.LessOrEqual(t, len(st.Message()), testMaxMessageLength) require.True(t, strings.HasSuffix(st.Message(), truncatedSuffix)) }) t.Run("truncation preserves error code", func(t *testing.T) { - msg := strings.Repeat("x", maxMessageLength+500) - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + msg := strings.Repeat("x", testMaxMessageLength+500) + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, serviceerror.NewNotFound(msg) }) @@ -119,15 +126,15 @@ func TestServiceErrorInterceptorTruncation(t *testing.T) { t.Run("truncation respects multi-byte UTF-8 boundary", func(t *testing.T) { // Fill up to near the limit with multi-byte characters (3 bytes each for '€') // then push over the limit so truncation must split within the repeated chars. - euroCount := maxMessageLength / len("€") // each '€' is 3 bytes + euroCount := testMaxMessageLength / len("€") // each '€' is 3 bytes msg := strings.Repeat("€", euroCount+100) - _, err := ServiceErrorInterceptor(t.Context(), nil, nil, + _, err := interceptor.Intercept(t.Context(), nil, nil, func(_ context.Context, _ any) (any, error) { return nil, serviceerror.NewInternal(msg) }) require.Error(t, err) st := status.Convert(err) - require.LessOrEqual(t, len(st.Message()), maxMessageLength) + require.LessOrEqual(t, len(st.Message()), testMaxMessageLength) require.True(t, strings.HasSuffix(st.Message(), truncatedSuffix)) // Verify the truncated body (without suffix) is valid UTF-8 by checking // that no partial rune was left behind — the full message should be valid. diff --git a/service/frontend/fx.go b/service/frontend/fx.go index da21a09d13..a40077a8aa 100644 --- a/service/frontend/fx.go +++ b/service/frontend/fx.go @@ -77,6 +77,7 @@ var Module = fx.Options( // A more robust approach would require using fx groups but we shouldn't overcomplicate until this becomes an issue. fx.Provide(MuxRouterProvider), fx.Provide(ConfigProvider), + fx.Provide(ServiceErrorInterceptorProvider), fx.Provide(NamespaceLogInterceptorProvider), fx.Provide(NamespaceHandoverInterceptorProvider), fx.Provide(interceptor.NewRoutingKeyExtractor), @@ -218,6 +219,7 @@ func GrpcServerOptionsProvider( serviceConfig *Config, serviceName primitives.ServiceName, rpcFactory common.RPCFactory, + serviceErrorInterceptor *interceptor.ServiceErrorInterceptor, namespaceLogInterceptor *interceptor.NamespaceLogInterceptor, namespaceRateLimiterInterceptor interceptor.NamespaceRateLimitInterceptor, namespaceCountLimiterInterceptor *interceptor.ConcurrentRequestLimitInterceptor, @@ -271,7 +273,7 @@ func GrpcServerOptionsProvider( // Mask error interceptor should be the most outer interceptor since it handle the errors format // Service Error Interceptor should be the next most outer interceptor on error handling maskInternalErrorDetailsInterceptor.Intercept, - interceptor.ServiceErrorInterceptor, + serviceErrorInterceptor.Intercept, interceptor.NewFrontendServiceErrorInterceptor(logger), // BusinessID interceptor extracts business ID and adds it to context for use, must be before any interceptor that touches namespaces (namespaceValidator, handoverInterceptor) businessIDInterceptor.Intercept, @@ -342,6 +344,14 @@ func ConfigProvider( ) } +func ServiceErrorInterceptorProvider( + dc *dynamicconfig.Collection, +) *interceptor.ServiceErrorInterceptor { + return interceptor.NewServiceErrorInterceptor( + dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + ) +} + func ThrottledLoggerRpsFnProvider(serviceConfig *Config) resource.ThrottledLoggerRpsFn { return func() float64 { return float64(serviceConfig.ThrottledLogRPS()) } } diff --git a/service/frontend/fx_test.go b/service/frontend/fx_test.go index 566565350b..969eba47f6 100644 --- a/service/frontend/fx_test.go +++ b/service/frontend/fx_test.go @@ -12,6 +12,7 @@ import ( enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/membership" "go.temporal.io/server/common/metrics" @@ -205,6 +206,10 @@ func TestRateLimitInterceptorProvider(t *testing.T) { } tc.configure(&tc) + serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(4000), + ) + // Create a rate limit interceptor which uses the per-instance and global RPS limits from the test case. rateLimitInterceptor := RateLimitInterceptorProvider(&Config{ RPS: func() int { @@ -225,7 +230,7 @@ func TestRateLimitInterceptorProvider(t *testing.T) { // Create a gRPC server for the fake workflow service. svc := &testSvc{} server := grpc.NewServer(grpc.ChainUnaryInterceptor( - interceptor.ServiceErrorInterceptor, + serviceErrorInterceptor.Intercept, interceptor.NewFrontendServiceErrorInterceptor(log.NewTestLogger()), rateLimitInterceptor.Intercept, )) @@ -569,6 +574,10 @@ func TestNamespaceRateLimitInterceptorProvider(t *testing.T) { config := getTestConfig(tc) + serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(4000), + ) + // Create a rate limit interceptor. rateLimitInterceptor := NamespaceRateLimitInterceptorProvider( primitives.FrontendService, @@ -582,7 +591,7 @@ func TestNamespaceRateLimitInterceptorProvider(t *testing.T) { // Create a gRPC server for the fake workflow service. svc := &testSvc{} server := grpc.NewServer(grpc.ChainUnaryInterceptor( - interceptor.ServiceErrorInterceptor, + serviceErrorInterceptor.Intercept, interceptor.NewFrontendServiceErrorInterceptor(log.NewTestLogger()), rateLimitInterceptor.Intercept, )) @@ -764,13 +773,17 @@ func TestNamespaceRateLimitMetrics(t *testing.T) { }, } + serviceErrorInterceptor := interceptor.NewServiceErrorInterceptor( + dynamicconfig.GetIntPropertyFn(4000), + ) + // Create a rate limit interceptor which uses the per-instance and global RPS limits from the test case. rateLimitInterceptor := RateLimitInterceptorProvider(config, serviceResolver, metricsHandler, log.NewTestLogger()) // Create a gRPC server for the fake workflow service. svc := &testSvc{} server := grpc.NewServer(grpc.ChainUnaryInterceptor( - interceptor.ServiceErrorInterceptor, + serviceErrorInterceptor.Intercept, interceptor.NewFrontendServiceErrorInterceptor(log.NewTestLogger()), rateLimitInterceptor.Intercept, )) diff --git a/service/fx.go b/service/fx.go index e9d412ba8c..5368c261ff 100644 --- a/service/fx.go +++ b/service/fx.go @@ -40,6 +40,7 @@ type ( Logger log.Logger RPCFactory common.RPCFactory + ServiceErrorInterceptor *interceptor.ServiceErrorInterceptor RetryableInterceptor *interceptor.RetryableInterceptor TelemetryInterceptor *interceptor.TelemetryInterceptor NamespaceRateLimitInterceptor interceptor.NamespaceRateLimitInterceptor `optional:"true"` @@ -157,7 +158,7 @@ func GrpcServerOptionsProvider( func getUnaryInterceptors(params GrpcServerOptionsParams) []grpc.UnaryServerInterceptor { interceptors := []grpc.UnaryServerInterceptor{ - interceptor.ServiceErrorInterceptor, + params.ServiceErrorInterceptor.Intercept, metrics.NewServerMetricsContextInjectorInterceptor(), metrics.NewServerMetricsTrailerPropagatorInterceptor(params.Logger), params.TelemetryInterceptor.UnaryIntercept, diff --git a/service/history/fx.go b/service/history/fx.go index 58d54854be..bc34c37ee2 100644 --- a/service/history/fx.go +++ b/service/history/fx.go @@ -64,6 +64,7 @@ var Module = fx.Options( ChasmEngineModule, fx.Provide(ConfigProvider), // might be worth just using provider for configs.Config directly fx.Provide(workflow.NewCommandHandlerRegistry), + fx.Provide(ServiceErrorInterceptorProvider), fx.Provide(RetryableInterceptorProvider), fx.Provide(ErrorHandlerProvider), fx.Provide(TelemetryInterceptorProvider), @@ -196,6 +197,14 @@ func ConfigProvider( ) } +func ServiceErrorInterceptorProvider( + dc *dynamicconfig.Collection, +) *interceptor.ServiceErrorInterceptor { + return interceptor.NewServiceErrorInterceptor( + dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + ) +} + func ThrottledLoggerRpsFnProvider(serviceConfig *configs.Config) resource.ThrottledLoggerRpsFn { return func() float64 { return float64(serviceConfig.ThrottledLogRPS()) } } diff --git a/service/matching/fx.go b/service/matching/fx.go index 665e359004..2332c832e4 100644 --- a/service/matching/fx.go +++ b/service/matching/fx.go @@ -35,6 +35,7 @@ var Module = fx.Options( fx.Provide(PersistenceRateLimitingParamsProvider), service.PersistenceLazyLoadedServiceResolverModule, fx.Provide(ThrottledLoggerRpsFnProvider), + fx.Provide(ServiceErrorInterceptorProvider), fx.Provide(ContextMetadataInterceptorProvider), fx.Provide(RetryableInterceptorProvider), fx.Provide(ErrorHandlerProvider), @@ -62,6 +63,14 @@ func ConfigProvider( return NewConfig(dc) } +func ServiceErrorInterceptorProvider( + dc *dynamicconfig.Collection, +) *interceptor.ServiceErrorInterceptor { + return interceptor.NewServiceErrorInterceptor( + dynamicconfig.MaxServiceErrorMessageLength.Get(dc), + ) +} + func RetryableInterceptorProvider() *interceptor.RetryableInterceptor { return interceptor.NewRetryableInterceptor( common.CreateMatchingHandlerRetryPolicy(),