This is an automated email from the ASF dual-hosted git repository.
AlexStocks pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go-pixiu.git
The following commit(s) were added to refs/heads/develop by this push:
new 6a498838d perf: cut gRPC/Triple/Dubbo proxy latency via connection
reuse and descriptor caching (fix #820) (#1017)
6a498838d is described below
commit 6a498838d50d88fe978748e302c25d8d67e9f98a
Author: Tsukikage <[email protected]>
AuthorDate: Wed Aug 26 21:20:31 2026 -0700
perf: cut gRPC/Triple/Dubbo proxy latency via connection reuse and
descriptor caching (fix #820) (#1017)
* perf: reduce gRPC, Triple, and Dubbo proxy latency
---
pkg/adapter/dubboregistry/registry/registry.go | 4 +
.../dubboregistry/registry/registry_test.go | 20 +
pkg/client/dubbo/call_test.go | 24 +-
pkg/client/dubbo/dubbo.go | 171 ++++---
pkg/client/dubbo/typeconv.go | 30 +-
pkg/client/dubbo/typeconv_test.go | 6 +
pkg/cmd/gateway.go | 27 +-
pkg/cmd/gateway_test.go | 23 +
pkg/common/constant/key.go | 1 +
pkg/common/constant/pixiu.go | 1 +
pkg/common/extension/filter/filter_chain.go | 19 +
pkg/common/extension/filter/filter_manager.go | 146 +++++-
pkg/common/extension/filter/filter_manager_test.go | 30 ++
pkg/common/http/manager.go | 11 +
pkg/filter/http/grpcproxy/connection_manager.go | 456 +++++++++++++++++
.../http/grpcproxy/connection_manager_test.go | 553 +++++++++++++++++++++
pkg/filter/http/grpcproxy/descriptor.go | 150 +++++-
pkg/filter/http/grpcproxy/grpc.go | 158 ++++--
pkg/listener/http/http_listener.go | 145 +++++-
pkg/listener/http/http_listener_test.go | 94 ++++
pkg/server/cluster_manager.go | 282 ++++++++++-
pkg/server/cluster_manager_test.go | 17 +
pkg/server/pixiu_start.go | 3 +
pkg/tracing/driver.go | 2 +
tools/benchmark/README.md | 58 ++-
tools/benchmark/README_CN.md | 90 ++--
.../benchmark/protocol/grpc/pixiu/conf/config.yaml | 2 +-
.../protocol/triple/go-server/cmd/server.go | 4 +-
.../protocol/triple/pixiu/conf/config.yaml | 4 +-
tools/benchmark/test/dubbo_suite/dubbo_test.go | 2 +-
tools/benchmark/test/grpc_suite/grpc_test.go | 4 +-
tools/benchmark/test/test_tool.go | 5 +-
.../test/triple_suite/proto_suite/proto_test.go | 10 +-
33 files changed, 2265 insertions(+), 287 deletions(-)
diff --git a/pkg/adapter/dubboregistry/registry/registry.go
b/pkg/adapter/dubboregistry/registry/registry.go
index b6d79355c..0ab736d16 100644
--- a/pkg/adapter/dubboregistry/registry/registry.go
+++ b/pkg/adapter/dubboregistry/registry/registry.go
@@ -24,6 +24,7 @@ import (
import (
"dubbo.apache.org/dubbo-go/v3/common"
+ dubboConstant "dubbo.apache.org/dubbo-go/v3/common/constant"
"github.com/pkg/errors"
)
@@ -95,6 +96,9 @@ func GetRegistry(name string, regConfig model.Registry,
listener common2.Registr
// CreateAPIConfig returns router.API struct base on the input
func CreateAPIConfig(urlPattern, location string, dboBackendConfig
config.DubboBackendConfig, methodString string, mappingParams
[]config.MappingParam) router.API {
dboBackendConfig.Method = methodString
+ if strings.TrimSpace(dboBackendConfig.Serialization) == "" {
+ dboBackendConfig.Serialization =
dubboConstant.Hessian2Serialization
+ }
url := strings.Join([]string{urlPattern, methodString},
constant.PathSlash)
var requestType string
switch dboBackendConfig.Protocol {
diff --git a/pkg/adapter/dubboregistry/registry/registry_test.go
b/pkg/adapter/dubboregistry/registry/registry_test.go
index c49fd9d61..867fbfb92 100644
--- a/pkg/adapter/dubboregistry/registry/registry_test.go
+++ b/pkg/adapter/dubboregistry/registry/registry_test.go
@@ -26,6 +26,10 @@ import (
"github.com/stretchr/testify/require"
)
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/config"
+)
+
func TestParseDubboStringPreservesSerialization(t *testing.T) {
backend, methods, location, err :=
ParseDubboString("tri://127.0.0.1:20001/org.apache.dubbogo.samples.api.Greeter?application=BDTService&interface=org.apache.dubbogo.samples.api.Greeter&methods=SayHello&serialization=hessian2")
require.NoError(t, err)
@@ -35,3 +39,19 @@ func TestParseDubboStringPreservesSerialization(t
*testing.T) {
assert.Equal(t, []string{"SayHello"}, methods)
assert.Equal(t, "127.0.0.1:20001", location)
}
+
+func TestCreateAPIConfigDefaultsGenericSerialization(t *testing.T) {
+ api := CreateAPIConfig(
+ "/dubbo.io/org.apache.dubbo.sample.UserProvider",
+ "127.0.0.1:20000",
+ config.DubboBackendConfig{
+ ApplicationName: "dubbo.io",
+ Protocol: "dubbo",
+ Interface: "org.apache.dubbo.sample.UserProvider",
+ },
+ "GetUser",
+ nil,
+ )
+
+ assert.Equal(t, "hessian2", api.Method.IntegrationRequest.Serialization)
+}
diff --git a/pkg/client/dubbo/call_test.go b/pkg/client/dubbo/call_test.go
index 668ff2e85..d06782248 100644
--- a/pkg/client/dubbo/call_test.go
+++ b/pkg/client/dubbo/call_test.go
@@ -35,6 +35,7 @@ import (
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/propagation"
+ "go.opentelemetry.io/otel/trace"
)
import (
@@ -145,14 +146,13 @@ func TestResolveFromOutboundRegistryMode(t *testing.T) {
func TestPreparePayloadRejectsLengthMismatch(t *testing.T) {
dc := NewDubboClient()
- types, vals, finalValues, err :=
dc.preparePayload(&DubboOutboundRequest{
+ types, vals, err := dc.preparePayload(&DubboOutboundRequest{
Arguments: []any{"only-one"},
ParamTypes: []string{"java.lang.String", "int"},
})
assert.Nil(t, types)
assert.Nil(t, vals)
- assert.Nil(t, finalValues)
assert.EqualError(t, err, "arguments/paramTypes length mismatch: 1 vs
2")
}
@@ -230,6 +230,26 @@ func TestCallUsesOutboundOnly(t *testing.T) {
assert.Equal(t, "ok", res)
}
+func TestWithAttachmentsPropagatesExternalSpanWhenTracingDisabled(t
*testing.T) {
+ restorePropagator(t, propagation.TraceContext{})
+ previousTracingEnabled := tracingEnabled.Load()
+ t.Cleanup(func() { SetTracingEnabled(previousTracingEnabled) })
+ SetTracingEnabled(false)
+
+ spanContext := trace.NewSpanContext(trace.SpanContextConfig{
+ TraceID: [16]byte{1},
+ SpanID: [8]byte{2},
+ TraceFlags: trace.FlagsSampled,
+ Remote: true,
+ })
+ ctx := trace.ContextWithRemoteSpanContext(context.Background(),
spanContext)
+ ctx = withAttachments(ctx, nil)
+
+ attachments, ok :=
ctx.Value(dubboConstant.AttachmentKey).(map[string]any)
+ require.True(t, ok)
+ require.Equal(t,
"00-01000000000000000000000000000000-0200000000000000-01",
attachments["traceparent"])
+}
+
func TestCallAppliesTimeout(t *testing.T) {
dc := NewDubboClient()
restorePropagator(t, fixedPropagator{})
diff --git a/pkg/client/dubbo/dubbo.go b/pkg/client/dubbo/dubbo.go
index 8f06db3fb..ce32a1e4b 100644
--- a/pkg/client/dubbo/dubbo.go
+++ b/pkg/client/dubbo/dubbo.go
@@ -24,6 +24,7 @@ import (
"strconv"
"strings"
"sync"
+ "sync/atomic"
"time"
)
@@ -42,6 +43,7 @@ import (
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/propagation"
+ "go.opentelemetry.io/otel/trace"
)
import (
@@ -70,8 +72,17 @@ var (
Owner: "Dubbogo Pixiu",
Environment: "dev",
}
+ // tracingEnabled records whether a tracer provider was configured at
startup.
+ // When tracing is disabled, per-request span creation and JSON payload
+ // marshaling for span attributes are skipped.
+ tracingEnabled atomic.Bool
)
+// SetTracingEnabled records whether a tracer provider has been configured.
+func SetTracingEnabled(enabled bool) {
+ tracingEnabled.Store(enabled)
+}
+
// Client client to generic invoke dubbo
type Client struct {
lock sync.RWMutex
@@ -79,6 +90,10 @@ type Client struct {
dubboProxyConfig *DubboProxyConfig
registries map[string]*global.RegistryConfig
dubboClient *dclient.Client
+ // immutable after Apply(): cached to avoid re-computing per request
+ registryIDs []string
+ useNacosWarmup bool
+ consumerDefaults resolvedConsumerDefaults
}
type resolvedConsumerDefaults struct {
@@ -173,6 +188,19 @@ func (dc *Client) Apply() error {
}
dc.registries = registries
+ // Cache immutable per-request resolution data
+ dc.registryIDs = make([]string, 0, len(registries))
+ useNacosWarmup := false
+ for id, registry := range registries {
+ dc.registryIDs = append(dc.registryIDs, id)
+ if registry != nil && registry.Protocol == "nacos" {
+ useNacosWarmup = true
+ }
+ }
+ sort.Strings(dc.registryIDs)
+ dc.useNacosWarmup = useNacosWarmup
+ dc.consumerDefaults = dc.resolveGlobalConsumerDefaults()
+
// Create dubbo client with registries and application config
var err error
dc.dubboClient, err = dclient.NewClient(
@@ -203,7 +231,7 @@ func (dc *Client) Call(ctx context.Context, req
*DubboOutboundRequest) (any, err
}
spec := dc.resolveFromOutbound(req)
- types, vals, finalValues, err := dc.preparePayload(req)
+ types, vals, err := dc.preparePayload(req)
if err != nil {
return nil, err
}
@@ -221,19 +249,26 @@ func (dc *Client) Call(ctx context.Context, req
*DubboOutboundRequest) (any, err
defer cancel()
}
- spanCtx, span := otel.Tracer(traceNameDubbogoClient).Start(invokeCtx,
spanNameDubbogoClient)
- defer span.End()
- span.SetAttributes(
- attribute.String(spanTagMethod, req.Method),
- attribute.StringSlice(spanTagType, types),
- attribute.String(spanTagValues, string(finalValues)),
- )
+ invokeCtx = withAttachments(invokeCtx, req.Attachments)
+ var span trace.Span
+ if tracingEnabled.Load() {
+ var spanCtx context.Context
+ spanCtx, span =
otel.Tracer(traceNameDubbogoClient).Start(invokeCtx, spanNameDubbogoClient)
+ defer span.End()
+ span.SetAttributes(
+ attribute.String(spanTagMethod, req.Method),
+ attribute.StringSlice(spanTagType, types),
+ attribute.String(spanTagValues,
string(spanValues(vals))),
+ )
+ // carry the trace context as upstream attachments
+ invokeCtx = withAttachments(spanCtx, nil)
+ }
- spanCtx = context.WithValue(spanCtx, constant.AttachmentKey,
mergeOutboundAttachments(spanCtx, req.Attachments))
- ctxWithAttachment := withAttachments(spanCtx)
- rst, err := gs.Invoke(ctxWithAttachment, req.Method, types, vals)
+ rst, err := gs.Invoke(invokeCtx, req.Method, types, vals)
if err != nil {
- span.RecordError(err)
+ if span != nil {
+ span.RecordError(err)
+ }
return nil, err
}
@@ -241,6 +276,14 @@ func (dc *Client) Call(ctx context.Context, req
*DubboOutboundRequest) (any, err
return rst, nil
}
+func spanValues(vals []hessian.Object) []byte {
+ finalValues, err := json.Marshal(vals)
+ if err != nil {
+ return nil
+ }
+ return finalValues
+}
+
func (dc *Client) resolveFromOutbound(req *DubboOutboundRequest)
resolvedReferSpec {
spec := resolvedReferSpec{
Interface: req.Service,
@@ -248,7 +291,11 @@ func (dc *Client) resolveFromOutbound(req
*DubboOutboundRequest) resolvedReferSp
Version: req.Version,
EffectiveProtocol: req.Protocol,
EffectiveSerialization: req.Serialization,
- ConsumerDefaults: dc.resolveGlobalConsumerDefaults(),
+ ConsumerDefaults: dc.consumerDefaults,
+ }
+ if spec.ConsumerDefaults.Cluster == "" &&
spec.ConsumerDefaults.RequestTimeout == 0 {
+ // fallback for clients used before Apply()
+ spec.ConsumerDefaults = dc.resolveGlobalConsumerDefaults()
}
if strings.TrimSpace(req.Address) != "" {
@@ -257,19 +304,25 @@ func (dc *Client) resolveFromOutbound(req
*DubboOutboundRequest) resolvedReferSp
return spec
}
- registryIDs := make([]string, 0, len(dc.registries))
- useNacosWarmup := false
- for id, registry := range dc.registries {
- registryIDs = append(registryIDs, id)
- if registry != nil && registry.Protocol == "nacos" {
- useNacosWarmup = true
+ spec.Mode = "registry"
+ if dc.registryIDs == nil && len(dc.registries) > 0 {
+ // fallback for clients used before Apply()
+ registryIDs := make([]string, 0, len(dc.registries))
+ useNacosWarmup := false
+ for id, registry := range dc.registries {
+ registryIDs = append(registryIDs, id)
+ if registry != nil && registry.Protocol == "nacos" {
+ useNacosWarmup = true
+ }
}
+ sort.Strings(registryIDs)
+ spec.RegistryIDs = registryIDs
+ spec.UseNacosWarmup = useNacosWarmup
+ return spec
}
- sort.Strings(registryIDs)
- spec.Mode = "registry"
- spec.RegistryIDs = registryIDs
- spec.UseNacosWarmup = useNacosWarmup
+ spec.RegistryIDs = dc.registryIDs
+ spec.UseNacosWarmup = dc.useNacosWarmup
return spec
}
@@ -297,12 +350,12 @@ func (dc *Client) resolveGlobalConsumerDefaults()
resolvedConsumerDefaults {
return defaults
}
-func (dc *Client) preparePayload(req *DubboOutboundRequest) ([]string,
[]hessian.Object, []byte, error) {
+func (dc *Client) preparePayload(req *DubboOutboundRequest) ([]string,
[]hessian.Object, error) {
if len(req.Arguments) == 0 && len(req.ParamTypes) == 0 {
- return []string{}, []hessian.Object{}, []byte("[]"), nil
+ return []string{}, []hessian.Object{}, nil
}
if len(req.Arguments) != len(req.ParamTypes) {
- return nil, nil, nil, errors.Errorf("arguments/paramTypes
length mismatch: %d vs %d", len(req.Arguments), len(req.ParamTypes))
+ return nil, nil, errors.Errorf("arguments/paramTypes length
mismatch: %d vs %d", len(req.Arguments), len(req.ParamTypes))
}
types := append([]string(nil), req.ParamTypes...)
@@ -311,15 +364,19 @@ func (dc *Client) preparePayload(req
*DubboOutboundRequest) ([]string, []hessian
vals[i] = arg
}
- finalValues, err := json.Marshal(vals)
- if err != nil {
- return nil, nil, nil, errors.Wrap(err, "marshal dubbo
arguments")
- }
-
- return types, vals, finalValues, nil
+ return types, vals, nil
}
-func mergeOutboundAttachments(ctx context.Context, outbound map[string]any)
map[string]any {
+func withAttachments(ctx context.Context, outbound map[string]any)
context.Context {
+ // The internal tracing switch does not describe the caller's context.
Keep
+ // the fast path only when there is no outbound state and no externally
+ // supplied span context that the global propagator must inject.
+ if !tracingEnabled.Load() &&
+ len(outbound) == 0 &&
+ ctx.Value(constant.AttachmentKey) == nil &&
+ !trace.SpanContextFromContext(ctx).IsValid() {
+ return ctx
+ }
attachments := make(map[string]any, len(outbound))
if attaRaw := ctx.Value(constant.AttachmentKey); attaRaw != nil {
switch userAtta := attaRaw.(type) {
@@ -336,7 +393,15 @@ func mergeOutboundAttachments(ctx context.Context,
outbound map[string]any) map[
for key, val := range outbound {
attachments[key] = val
}
- return attachments
+
+ carrier := propagation.MapCarrier{}
+ // Carry tracing headers as Dubbo attachments for the upstream
invocation.
+ otel.GetTextMapPropagator().Inject(ctx, carrier)
+ for key, val := range carrier {
+ attachments[key] = val
+ }
+
+ return context.WithValue(ctx, constant.AttachmentKey, attachments)
}
func prepareInvokeContext(ctx context.Context, timeout time.Duration)
(context.Context, context.CancelFunc) {
@@ -355,15 +420,6 @@ func (dc *Client) get(key string) *generic.GenericService {
return dc.GenericServicePool[key]
}
-func (dc *Client) check(key string) bool {
- dc.lock.RLock()
- defer dc.lock.RUnlock()
- if _, ok := dc.GenericServicePool[key]; ok {
- return true
- }
- return false
-}
-
func (spec resolvedReferSpec) validate() error {
switch spec.Mode {
case "registry":
@@ -420,8 +476,8 @@ func (dc *Client) Get(spec resolvedReferSpec)
(*generic.GenericService, error) {
if err != nil {
return nil, err
}
- if dc.check(key) {
- return dc.get(key), nil
+ if service := dc.get(key); service != nil {
+ return service, nil
}
return dc.create(spec)
@@ -587,28 +643,3 @@ func loadBalanceReferenceOption(loadBalance string)
dclient.ReferenceOption {
return dclient.WithLoadBalance(loadBalance)
}
}
-
-func withAttachments(ctx context.Context) context.Context {
- attachments := make(map[string]any)
- if attaRaw := ctx.Value(constant.AttachmentKey); attaRaw != nil {
- switch userAtta := attaRaw.(type) {
- case map[string]any:
- for key, val := range userAtta {
- attachments[key] = val
- }
- case map[string]string:
- for key, val := range userAtta {
- attachments[key] = val
- }
- }
- }
-
- carrier := propagation.MapCarrier{}
- // Carry tracing headers as Dubbo attachments for the upstream
invocation.
- otel.GetTextMapPropagator().Inject(ctx, carrier)
- for key, val := range carrier {
- attachments[key] = val
- }
-
- return context.WithValue(ctx, constant.AttachmentKey, attachments)
-}
diff --git a/pkg/client/dubbo/typeconv.go b/pkg/client/dubbo/typeconv.go
index a0a3934fa..276a7cf0e 100644
--- a/pkg/client/dubbo/typeconv.go
+++ b/pkg/client/dubbo/typeconv.go
@@ -98,25 +98,35 @@ func CoerceDirectInvokeValue(parameterType string, value
any) (any, error) {
if !ok {
return value, nil
}
-
- result := make([]any, len(items))
- for i, item := range items {
- mapped, err := MapTypes(elementType, item)
- if err != nil {
- return nil, err
- }
- result[i] = mapped
- }
- return result, nil
+ return coerceScalarSlice(elementType, items)
}
if _, ok := cst.JTypeMapper[normalizeJavaTypeName(trimmed)]; ok {
+ // The legacy generic HTTP contract uses the element type for a
+ // collection argument, for example types="string" with
+ // values=[["003", "002"]]. Preserve that contract while still
+ // coercing each scalar element to the declared type.
+ if items, ok := value.([]any); ok {
+ return coerceScalarSlice(trimmed, items)
+ }
return MapTypes(trimmed, value)
}
return value, nil
}
+func coerceScalarSlice(parameterType string, items []any) ([]any, error) {
+ result := make([]any, len(items))
+ for i, item := range items {
+ mapped, err := MapTypes(parameterType, item)
+ if err != nil {
+ return nil, err
+ }
+ result[i] = mapped
+ }
+ return result, nil
+}
+
// NormalizeReferenceProtocol canonicalizes dubbo reference protocols.
func NormalizeReferenceProtocol(protocol string) string {
normalized := strings.ToLower(strings.TrimSpace(protocol))
diff --git a/pkg/client/dubbo/typeconv_test.go
b/pkg/client/dubbo/typeconv_test.go
index d1199c1d1..c091f85a4 100644
--- a/pkg/client/dubbo/typeconv_test.go
+++ b/pkg/client/dubbo/typeconv_test.go
@@ -68,6 +68,12 @@ func TestCoerceDirectInvokeValueWrapperFQNArray(t
*testing.T) {
assert.Equal(t, []any{1, 2}, val)
}
+func TestCoerceDirectInvokeValueLegacyGenericArrayWithScalarType(t *testing.T)
{
+ val, err := CoerceDirectInvokeValue("string", []any{"003", "002"})
+ require.NoError(t, err)
+ assert.Equal(t, []any{"003", "002"}, val)
+}
+
func TestNormalizeReferenceProtocol(t *testing.T) {
assert.Equal(t, "tri", NormalizeReferenceProtocol("triple"))
assert.Equal(t, "tri", NormalizeReferenceProtocol(" TRI "))
diff --git a/pkg/cmd/gateway.go b/pkg/cmd/gateway.go
index 2be1a3b17..7f203ca26 100644
--- a/pkg/cmd/gateway.go
+++ b/pkg/cmd/gateway.go
@@ -50,6 +50,9 @@ var (
logFormat string
limitCpus string
+ // port for the reload HTTP server, configurable so multiple
+ // gateway instances can run side by side without port conflicts
+ reloadPort int
)
var (
@@ -92,6 +95,7 @@ func init() {
startGatewayCmd.PersistentFlags().StringVarP(&logLevel,
constant.LogLevelKey, "l", os.Getenv(constant.EnvDubbogoPixiuLogLevel),
"dubbogo pixiu log level, trace|debug|info|warning|error|critical")
startGatewayCmd.PersistentFlags().StringVarP(&limitCpus,
constant.LimitCpusKey, "m", os.Getenv(constant.EnvDubbogoPixiuLimitCpus),
"dubbogo pixiu schedule threads count")
startGatewayCmd.PersistentFlags().StringVarP(&logFormat,
constant.LogFormatKey, "f", os.Getenv(constant.EnvDubbogoPixiuLogFormat),
"dubbogo pixiu log format, currently useless")
+ startGatewayCmd.PersistentFlags().IntVarP(&reloadPort,
constant.ReloadPortKey, "r", constant.DefaultReloadPort, "dubbogo pixiu reload
HTTP server port")
GatewayCmd.AddCommand(startGatewayCmd)
}
@@ -122,8 +126,8 @@ func (d *DefaultDeployer) initialize() error {
// Set config path for hot reload
hotreload.SetConfigPath(configPath)
- // Start HTTP reload endpoint on port 18380
- if err := hotreload.StartReloadServer(18380, ""); err != nil {
+ // Start HTTP reload endpoint, port configurable via --reload-port
+ if err := hotreload.StartReloadServer(reloadPort, ""); err != nil {
logger.Warnf("[startGatewayCmd] failed to start reload server:
%s", err.Error())
}
@@ -169,17 +173,16 @@ func initDefaultValue() {
// initLog initializes logger according to log config file and log level
func initLog() error {
- err := logger.InitLog(logConfigPath)
- if err != nil {
- // cause `logger.InitLog` already handle init failed, so just
use logger to log
- return err
- }
-
+ initErr := logger.InitLog(logConfigPath)
lvl := logger.ParseLogLevel(logLevel)
- if ok := logger.SetLoggerLevel(lvl); !ok {
- err = fmt.Errorf("set logLevel failed")
- }
- return err
+ if ok := logger.SetLoggerLevel(lvl); !ok && initErr == nil {
+ return fmt.Errorf("set logLevel failed")
+ }
+ // Apply the requested level even when the optional log config file is
+ // missing or invalid. InitLog already installs the default logger in
that
+ // case, and leaving it at development/debug level adds avoidable
hot-path
+ // logging during gateway startup and request handling.
+ return initErr
}
func initLogWithConfig(boot *model.Bootstrap) {
diff --git a/pkg/cmd/gateway_test.go b/pkg/cmd/gateway_test.go
index 8e1821bd3..2a5d31c92 100644
--- a/pkg/cmd/gateway_test.go
+++ b/pkg/cmd/gateway_test.go
@@ -26,6 +26,12 @@ import (
"github.com/spf13/cobra"
"github.com/stretchr/testify/assert"
+
+ "go.uber.org/zap"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
)
// MockDeployer for testing
@@ -198,6 +204,23 @@ func TestInitDefaultValue(t *testing.T) {
// logFormat can be empty as DefaultLogFormat is ""
}
+func TestInitLogAppliesLevelWhenConfigIsMissing(t *testing.T) {
+ previousPath := logConfigPath
+ previousLevel := logLevel
+ defer func() {
+ logConfigPath = previousPath
+ logLevel = previousLevel
+ logger.InitLogger(nil)
+ }()
+
+ logConfigPath = "/tmp/dubbo-go-pixiu-missing-log-config.yml"
+ logLevel = "info"
+
+ err := initLog()
+ assert.Error(t, err)
+ assert.False(t,
logger.GetLogger().Desugar().Core().Enabled(zap.DebugLevel))
+}
+
func TestGatewayCmdAddedToRootCmd(t *testing.T) {
// Check that GatewayCmd is properly initialized
assert.NotNil(t, GatewayCmd)
diff --git a/pkg/common/constant/key.go b/pkg/common/constant/key.go
index 721361481..0e8dac662 100644
--- a/pkg/common/constant/key.go
+++ b/pkg/common/constant/key.go
@@ -80,6 +80,7 @@ const (
LogLevelKey = "log-level"
LimitCpusKey = "limit-cpus"
LogFormatKey = "log-format"
+ ReloadPortKey = "reload-port"
)
const (
diff --git a/pkg/common/constant/pixiu.go b/pkg/common/constant/pixiu.go
index 464e5ed8d..f8a35107d 100644
--- a/pkg/common/constant/pixiu.go
+++ b/pkg/common/constant/pixiu.go
@@ -73,4 +73,5 @@ const (
DefaultLogLevel = "info"
DefaultLimitCpus = "0"
DefaultLogFormat = ""
+ DefaultReloadPort = 18380
)
diff --git a/pkg/common/extension/filter/filter_chain.go
b/pkg/common/extension/filter/filter_chain.go
index 07d8108da..c4fe08daf 100644
--- a/pkg/common/extension/filter/filter_chain.go
+++ b/pkg/common/extension/filter/filter_chain.go
@@ -17,6 +17,10 @@
package filter
+import (
+ "sync"
+)
+
import (
"github.com/apache/dubbo-go-pixiu/pkg/context/http"
)
@@ -36,6 +40,21 @@ type defaultFilterChain struct {
encodeFilters []HttpEncodeFilter
encodeFiltersIndex int
+ release func()
+ releaseOnce sync.Once
+}
+
+func (c *defaultFilterChain) setRelease(release func()) {
+ c.release = release
+}
+
+// Release releases resources leased by this request's filter chain.
+func (c *defaultFilterChain) Release() {
+ c.releaseOnce.Do(func() {
+ if c.release != nil {
+ c.release()
+ }
+ })
}
func NewDefaultFilterChain() FilterChain {
diff --git a/pkg/common/extension/filter/filter_manager.go
b/pkg/common/extension/filter/filter_manager.go
index 3a87b2dd4..4129b5a1f 100644
--- a/pkg/common/extension/filter/filter_manager.go
+++ b/pkg/common/extension/filter/filter_manager.go
@@ -40,7 +40,11 @@ type FilterManager struct {
filtersArray []*HttpFilterFactory
filterConfigs []*model.HTTPFilter
- mu sync.RWMutex
+ mu sync.RWMutex
+ lifecycleMu sync.Mutex
+ factoryRefs map[*HttpFilterFactory]int
+ retired map[*HttpFilterFactory]bool
+ closed bool
}
// NewFilterManager create filter manager
@@ -57,7 +61,14 @@ func NewEmptyFilterManager() *FilterManager {
func (fm *FilterManager) CreateFilterChain(ctx *http.HttpContext) FilterChain {
chain := NewDefaultFilterChain()
- for _, f := range fm.GetFactory() {
+ fm.mu.RLock()
+ defer fm.mu.RUnlock()
+ factories := append([]*HttpFilterFactory(nil), fm.filtersArray...)
+ fm.leaseFactories(factories)
+ chain.(*defaultFilterChain).setRelease(func() {
+ fm.releaseFactories(factories)
+ })
+ for _, f := range factories {
_ = (*f).PrepareFilterChain(ctx, chain)
}
return chain
@@ -68,7 +79,7 @@ func (fm *FilterManager) GetFactory() []*HttpFilterFactory {
fm.mu.RLock()
defer fm.mu.RUnlock()
- return fm.filtersArray
+ return append([]*HttpFilterFactory(nil), fm.filtersArray...)
}
// Load the filter from config
@@ -78,22 +89,130 @@ func (fm *FilterManager) Load() {
// ReLoad filter configs
func (fm *FilterManager) ReLoad(filters []*model.HTTPFilter) {
+ fm.mu.RLock()
+ closed := fm.closed
+ fm.mu.RUnlock()
+ if closed {
+ return
+ }
+
tmp := make(map[string]HttpFilterFactory)
- filtersArray := make([]*HttpFilterFactory, len(filters))
- for i, f := range filters {
+ filtersArray := make([]*HttpFilterFactory, 0, len(filters))
+ for _, f := range filters {
apply, err := fm.Apply(f.Name, f.Config)
if err != nil {
logger.Errorf("apply [%s] init fail, %s", f.Name,
err.Error())
+ continue
}
tmp[f.Name] = apply
- filtersArray[i] = &apply
+ filtersArray = append(filtersArray, &apply)
}
// avoid filter inconsistency
fm.mu.Lock()
- defer fm.mu.Unlock()
-
+ if fm.closed {
+ fm.mu.Unlock()
+ fm.closeAndLog(filtersArray)
+ return
+ }
+ oldFilters := fm.filtersArray
fm.filters = tmp
fm.filtersArray = filtersArray
+ ready := fm.retireFactories(oldFilters)
+ fm.mu.Unlock()
+ fm.closeAndLog(ready)
+}
+
+func (fm *FilterManager) leaseFactories(factories []*HttpFilterFactory) {
+ fm.lifecycleMu.Lock()
+ defer fm.lifecycleMu.Unlock()
+ if fm.factoryRefs == nil {
+ fm.factoryRefs = make(map[*HttpFilterFactory]int)
+ }
+ for _, factory := range factories {
+ if factory != nil {
+ fm.factoryRefs[factory]++
+ }
+ }
+}
+
+func (fm *FilterManager) retireFactories(factories []*HttpFilterFactory)
[]*HttpFilterFactory {
+ fm.lifecycleMu.Lock()
+ defer fm.lifecycleMu.Unlock()
+ if fm.retired == nil {
+ fm.retired = make(map[*HttpFilterFactory]bool)
+ }
+ var ready []*HttpFilterFactory
+ for _, factory := range factories {
+ if factory == nil {
+ continue
+ }
+ fm.retired[factory] = true
+ if fm.factoryRefs[factory] == 0 {
+ delete(fm.retired, factory)
+ ready = append(ready, factory)
+ }
+ }
+ return ready
+}
+
+func (fm *FilterManager) releaseFactories(factories []*HttpFilterFactory) {
+ fm.lifecycleMu.Lock()
+ var ready []*HttpFilterFactory
+ for _, factory := range factories {
+ if factory == nil {
+ continue
+ }
+ fm.factoryRefs[factory]--
+ if fm.factoryRefs[factory] == 0 {
+ delete(fm.factoryRefs, factory)
+ if fm.retired[factory] {
+ delete(fm.retired, factory)
+ ready = append(ready, factory)
+ }
+ }
+ }
+ fm.lifecycleMu.Unlock()
+ fm.closeAndLog(ready)
+}
+
+func (fm *FilterManager) closeAndLog(factories []*HttpFilterFactory) {
+ if err := fm.closeFactories(factories); err != nil {
+ logger.Warnf("failed to close retired HTTP filter factory: %v",
err)
+ }
+}
+
+func (fm *FilterManager) closeFactories(factories []*HttpFilterFactory) error {
+ var firstErr error
+ for _, factory := range factories {
+ if factory == nil || *factory == nil {
+ continue
+ }
+ closer, ok := (*factory).(interface{ Close() error })
+ if !ok {
+ continue
+ }
+ if err := closer.Close(); err != nil && firstErr == nil {
+ firstErr = err
+ }
+ }
+ return firstErr
+}
+
+// Close releases resources held by HTTP filter factories that expose an
+// optional Close method. Keeping this optional preserves compatibility with
+// existing HTTP filter implementations.
+func (fm *FilterManager) Close() error {
+ fm.mu.Lock()
+ if fm.closed {
+ fm.mu.Unlock()
+ return nil
+ }
+ fm.closed = true
+ factories := fm.filtersArray
+ fm.filtersArray = nil
+ ready := fm.retireFactories(factories)
+ fm.mu.Unlock()
+ return fm.closeFactories(ready)
}
// Apply return a new filter factory by name & conf
@@ -108,16 +227,27 @@ func (fm *FilterManager) Apply(name string, conf
map[string]any) (HttpFilterFact
if err != nil {
return nil, errors.New("plugin create filter error")
}
+ if filter == nil {
+ return nil, errors.New("plugin returned nil filter factory")
+ }
+ closeFilter := func() {
+ if closer, ok := filter.(interface{ Close() error }); ok {
+ _ = closer.Close()
+ }
+ }
factoryConf := filter.Config()
if err := yaml.ParseConfig(factoryConf, conf); err != nil {
+ closeFilter()
return nil, errors.Wrap(err, "config error")
}
if err = defaults.Set(factoryConf); err != nil {
+ closeFilter()
return nil, err
}
err = filter.Apply()
if err != nil {
+ closeFilter()
return nil, errors.Wrap(err, "create fail")
}
return filter, nil
diff --git a/pkg/common/extension/filter/filter_manager_test.go
b/pkg/common/extension/filter/filter_manager_test.go
index e0bce7020..d729fc685 100644
--- a/pkg/common/extension/filter/filter_manager_test.go
+++ b/pkg/common/extension/filter/filter_manager_test.go
@@ -19,6 +19,7 @@ package filter
import (
"fmt"
+ "sync"
"testing"
)
@@ -144,6 +145,35 @@ func TestLoad(t *testing.T) {
runFilter(t, fm, filtersConf)
}
+func TestReloadSkipsFailedFactoriesAndDoesNotReopenAfterClose(t *testing.T) {
+ fm := NewEmptyFilterManager()
+ fm.ReLoad([]*model.HTTPFilter{{Name: "missing-filter"}})
+ assert.Empty(t, fm.GetFactory())
+ assert.NotPanics(t, func() {
fm.CreateFilterChain(&contexthttp.HttpContext{}) })
+
+ fm.ReLoad([]*model.HTTPFilter{{Name: DEMO}})
+ assert.Len(t, fm.GetFactory(), 1)
+ assert.NoError(t, fm.Close())
+ fm.ReLoad([]*model.HTTPFilter{{Name: DEMO}})
+ assert.Empty(t, fm.GetFactory())
+}
+
+func TestReloadAndCloseDoNotReopenManager(t *testing.T) {
+ fm := NewEmptyFilterManager()
+ filters := []*model.HTTPFilter{{Name: DEMO}}
+ var wg sync.WaitGroup
+ wg.Go(func() {
+ for range 20 {
+ fm.ReLoad(filters)
+ }
+ })
+ wg.Go(func() {
+ assert.NoError(t, fm.Close())
+ })
+ wg.Wait()
+ assert.Empty(t, fm.GetFactory())
+}
+
func runFilter(t *testing.T, fm *FilterManager, filtersConf
[]*model.HTTPFilter) {
fm.ReLoad(filtersConf)
diff --git a/pkg/common/http/manager.go b/pkg/common/http/manager.go
index 5177bcad4..7eaf5c651 100644
--- a/pkg/common/http/manager.go
+++ b/pkg/common/http/manager.go
@@ -92,6 +92,14 @@ func (hcm *HttpConnectionManager) Handle(hc
*pch.HttpContext) error {
return nil
}
+// Close releases resources held by the HTTP filter chain.
+func (hcm *HttpConnectionManager) Close() error {
+ if hcm.filterManager == nil {
+ return nil
+ }
+ return hcm.filterManager.Close()
+}
+
func (hcm *HttpConnectionManager) ServeHTTP(w stdHttp.ResponseWriter, r
*stdHttp.Request) {
hc := hcm.pool.Get().(*pch.HttpContext)
defer hcm.pool.Put(hc)
@@ -109,6 +117,9 @@ func (hcm *HttpConnectionManager) ServeHTTP(w
stdHttp.ResponseWriter, r *stdHttp
// handleHTTPRequest handle http request
func (hcm *HttpConnectionManager) handleHTTPRequest(c *pch.HttpContext) {
filterChain := hcm.filterManager.CreateFilterChain(c)
+ if releaser, ok := filterChain.(interface{ Release() }); ok {
+ defer releaser.Release()
+ }
// recover any err when filterChain run
defer func() {
diff --git a/pkg/filter/http/grpcproxy/connection_manager.go
b/pkg/filter/http/grpcproxy/connection_manager.go
new file mode 100644
index 000000000..e6c7796ee
--- /dev/null
+++ b/pkg/filter/http/grpcproxy/connection_manager.go
@@ -0,0 +1,456 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package grpcproxy
+
+import (
+ "container/list"
+ "context"
+ "fmt"
+ "sync"
+ "time"
+)
+
+import (
+ "golang.org/x/sync/singleflight"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/connectivity"
+ "google.golang.org/grpc/credentials/insecure"
+)
+
+const (
+ defaultGRPCDialTimeout = 5 * time.Second
+ // Endpoint tombstones only need to cover recently delivered lifecycle
+ // events. Active connections and in-flight dials are pinned separately
and
+ // are never evicted by this bound.
+ maxEndpointTombstones = 1024
+)
+
+type grpcConnectionDialer func(context.Context, string) (*grpc.ClientConn,
error)
+
+// grpcConnectionManager owns long-lived backend connections for the HTTP gRPC
+// proxy. A grpc.ClientConn is safe for concurrent use and multiplexes calls
+// over HTTP/2, so a sync.Pool is both unnecessary and incorrect here.
+type grpcConnectionManager struct {
+ connections sync.Map
+ creates singleflight.Group
+ dial grpcConnectionDialer
+ dialTimeout time.Duration
+ onRemove func(*grpc.ClientConn)
+
+ mu sync.Mutex
+ closed bool
+ endpointGenerations map[string]uint64
+ endpointEventVers map[string]uint64
+ endpointRefs map[string]int
+ endpointRemoved map[string]bool
+ endpointTombstones map[string]*list.Element
+ tombstoneOrder *list.List
+ // endpointPresent is an optional authoritative snapshot check used
after
+ // bounded tombstone metadata has been evicted. It is intentionally kept
+ // outside the manager so direct-endpoint users do not need a cluster
store.
+ endpointPresent func(key, endpoint string) bool
+}
+
+func newGRPCConnectionManager() *grpcConnectionManager {
+ return &grpcConnectionManager{
+ dial: dialGRPCConnection,
+ dialTimeout: defaultGRPCDialTimeout,
+ endpointGenerations: make(map[string]uint64),
+ endpointEventVers: make(map[string]uint64),
+ endpointRefs: make(map[string]int),
+ endpointRemoved: make(map[string]bool),
+ endpointTombstones: make(map[string]*list.Element),
+ tombstoneOrder: list.New(),
+ }
+}
+
+func (m *grpcConnectionManager) initEndpointStateLocked() {
+ if m.endpointGenerations == nil {
+ m.endpointGenerations = make(map[string]uint64)
+ }
+ if m.endpointEventVers == nil {
+ m.endpointEventVers = make(map[string]uint64)
+ }
+ if m.endpointRefs == nil {
+ m.endpointRefs = make(map[string]int)
+ }
+ if m.endpointRemoved == nil {
+ m.endpointRemoved = make(map[string]bool)
+ }
+ if m.endpointTombstones == nil {
+ m.endpointTombstones = make(map[string]*list.Element)
+ }
+ if m.tombstoneOrder == nil {
+ m.tombstoneOrder = list.New()
+ }
+}
+
+func (m *grpcConnectionManager) discardEndpointTombstoneLocked(key string) {
+ if element, ok := m.endpointTombstones[key]; ok {
+ m.tombstoneOrder.Remove(element)
+ delete(m.endpointTombstones, key)
+ }
+}
+
+func (m *grpcConnectionManager) discardEndpointStateLocked(key string) {
+ m.discardEndpointTombstoneLocked(key)
+ delete(m.endpointGenerations, key)
+ delete(m.endpointEventVers, key)
+ delete(m.endpointRemoved, key)
+}
+
+func (m *grpcConnectionManager) rememberEndpointTombstoneLocked(key string) {
+ m.initEndpointStateLocked()
+ if element, ok := m.endpointTombstones[key]; ok {
+ m.tombstoneOrder.MoveToFront(element)
+ return
+ }
+ element := m.tombstoneOrder.PushFront(key)
+ m.endpointTombstones[key] = element
+
+ for len(m.endpointTombstones) > maxEndpointTombstones {
+ var evict *list.Element
+ for element := m.tombstoneOrder.Back(); element != nil; element
= element.Prev() {
+ candidate := element.Value.(string)
+ if m.endpointRefs[candidate] == 0 &&
!m.hasConnection(candidate) {
+ evict = element
+ break
+ }
+ }
+ if evict == nil {
+ return
+ }
+ candidate := evict.Value.(string)
+ m.tombstoneOrder.Remove(evict)
+ delete(m.endpointTombstones, candidate)
+ delete(m.endpointGenerations, candidate)
+ delete(m.endpointEventVers, candidate)
+ delete(m.endpointRemoved, candidate)
+ }
+}
+
+func (m *grpcConnectionManager) hasConnection(key string) bool {
+ _, ok := m.connections.Load(key)
+ return ok
+}
+
+func (m *grpcConnectionManager) pinEndpoint(key string) (uint64, error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.initEndpointStateLocked()
+ if m.closed {
+ return 0, fmt.Errorf("grpc connection manager is closed")
+ }
+ if m.endpointRemoved[key] {
+ return 0, fmt.Errorf("grpc endpoint was removed")
+ }
+ m.endpointRefs[key]++
+ return m.endpointGenerations[key], nil
+}
+
+func (m *grpcConnectionManager) isEndpointPresent(key, endpoint string) bool {
+ if m.endpointPresent == nil {
+ return true
+ }
+ return m.endpointPresent(key, endpoint)
+}
+
+func (m *grpcConnectionManager) unpinEndpoint(key string) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if m.endpointRefs[key] > 1 {
+ m.endpointRefs[key]--
+ return
+ }
+ delete(m.endpointRefs, key)
+ if m.endpointRemoved[key] && !m.hasConnection(key) {
+ m.rememberEndpointTombstoneLocked(key)
+ } else if !m.endpointRemoved[key] && m.endpointEventVers[key] == 0 &&
!m.hasConnection(key) {
+ // Get may create a temporary generation entry for a direct
endpoint
+ // before the cluster manager has delivered a lifecycle event.
Do not
+ // retain that request-only state after a failed or canceled
dial.
+ m.discardEndpointStateLocked(key)
+ }
+}
+
+func (m *grpcConnectionManager) finalizeRemovedEndpoint(key string) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if m.endpointRemoved[key] && m.endpointRefs[key] == 0 &&
!m.hasConnection(key) {
+ m.rememberEndpointTombstoneLocked(key)
+ }
+}
+
+func dialGRPCConnection(ctx context.Context, endpoint string)
(*grpc.ClientConn, error) {
+ return grpc.DialContext( //nolint:staticcheck // SA1019: the context is
required to enforce the dial timeout.
+ ctx,
+ endpoint,
+ grpc.WithTransportCredentials(insecure.NewCredentials()),
+ )
+}
+
+func (m *grpcConnectionManager) Get(ctx context.Context, key, endpoint string)
(*grpc.ClientConn, error) {
+ if key == "" || endpoint == "" {
+ return nil, fmt.Errorf("grpc connection key and endpoint must
not be empty")
+ }
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ if !m.isEndpointPresent(key, endpoint) {
+ return nil, fmt.Errorf("grpc endpoint was removed")
+ }
+
+ if conn, ok := m.loadHealthy(key); ok {
+ if !m.isEndpointPresent(key, endpoint) {
+ m.remove(key, conn)
+ return nil, fmt.Errorf("grpc endpoint was removed")
+ }
+ return conn, nil
+ }
+
+ endpointGeneration, err := m.pinEndpoint(key)
+ if err != nil {
+ return nil, err
+ }
+ createKey := fmt.Sprintf("%s\x00%d", key, endpointGeneration)
+ result := m.creates.DoChan(createKey, func() (any, error) {
+ if !m.isEndpointPresent(key, endpoint) {
+ return nil, fmt.Errorf("grpc endpoint was removed")
+ }
+ m.mu.Lock()
+ currentGeneration := m.endpointGenerations[key]
+ removed := m.endpointRemoved[key]
+ m.mu.Unlock()
+ if removed || currentGeneration != endpointGeneration {
+ return nil, fmt.Errorf("grpc endpoint was removed while
connecting")
+ }
+ if conn, ok := m.loadHealthy(key); ok {
+ if !m.isEndpointPresent(key, endpoint) {
+ m.remove(key, conn)
+ return nil, fmt.Errorf("grpc endpoint was
removed")
+ }
+ return conn, nil
+ }
+
+ m.mu.Lock()
+ if m.closed {
+ m.mu.Unlock()
+ return nil, fmt.Errorf("grpc connection manager is
closed")
+ }
+ dial := m.dial
+ dialTimeout := m.dialTimeout
+ m.mu.Unlock()
+
+ dialCtx, cancel := context.WithTimeout(context.Background(),
dialTimeout)
+ defer cancel()
+ conn, err := dial(dialCtx, endpoint)
+ if err != nil {
+ return nil, err
+ }
+
+ m.mu.Lock()
+ closed := m.closed
+ removed = m.endpointRemoved[key] || m.endpointGenerations[key]
!= endpointGeneration
+ if !closed && !removed {
+ m.connections.Store(key, conn)
+ }
+ m.mu.Unlock()
+ if closed || removed || !m.isEndpointPresent(key, endpoint) {
+ removedFromManager := false
+ if !closed && !removed {
+ removedFromManager = m.remove(key, conn)
+ }
+ if !removedFromManager {
+ _ = conn.Close()
+ }
+ if closed {
+ return nil, fmt.Errorf("grpc connection manager
closed while dialing")
+ }
+ return nil, fmt.Errorf("grpc endpoint was removed while
connecting")
+ }
+
+ return conn, nil
+ })
+
+ select {
+ case <-ctx.Done():
+ go func() {
+ <-result
+ m.unpinEndpoint(key)
+ }()
+ return nil, ctx.Err()
+ case result := <-result:
+ m.unpinEndpoint(key)
+ if result.Err != nil {
+ return nil, result.Err
+ }
+ return result.Val.(*grpc.ClientConn), nil
+ }
+}
+
+func (m *grpcConnectionManager) loadHealthy(key string) (*grpc.ClientConn,
bool) {
+ m.mu.Lock()
+ removed := m.endpointRemoved[key]
+ m.mu.Unlock()
+ if removed {
+ return nil, false
+ }
+ value, ok := m.connections.Load(key)
+ if !ok {
+ return nil, false
+ }
+
+ conn, ok := value.(*grpc.ClientConn)
+ if ok && m.isHealthy(conn) {
+ return conn, true
+ }
+ if ok {
+ m.remove(key, conn)
+ }
+ return nil, false
+}
+
+func (m *grpcConnectionManager) isHealthy(conn *grpc.ClientConn) bool {
+ if conn == nil {
+ return false
+ }
+ state := conn.GetState()
+ return state != connectivity.Shutdown
+}
+
+// Invalidate removes a connection only when its transport is known to be
+// unusable. Application-level RPC errors must not cause healthy connections to
+// churn.
+func (m *grpcConnectionManager) Invalidate(key string, conn *grpc.ClientConn) {
+ if conn == nil || m.isHealthy(conn) {
+ return
+ }
+ m.remove(key, conn)
+}
+
+// RemoveEndpoint closes and removes the connection for a deleted endpoint.
+func (m *grpcConnectionManager) RemoveEndpoint(clusterName, endpoint string) {
+ key := grpcConnectionKey(clusterName, endpoint)
+ m.mu.Lock()
+ m.initEndpointStateLocked()
+ if m.closed {
+ m.mu.Unlock()
+ return
+ }
+ m.endpointGenerations[key]++
+ m.endpointRemoved[key] = true
+ m.mu.Unlock()
+ value, ok := m.connections.Load(key)
+ if !ok {
+ m.finalizeRemovedEndpoint(key)
+ return
+ }
+ conn, ok := value.(*grpc.ClientConn)
+ if ok {
+ m.remove(key, conn)
+ }
+ m.finalizeRemovedEndpoint(key)
+}
+
+// UpdateEndpointState applies an ordered endpoint lifecycle event. Additions
+// advance the same generation as removals so a delayed removal cannot delete
+// a connection created after the endpoint was re-added.
+func (m *grpcConnectionManager) UpdateEndpointState(clusterName, endpoint
string, present bool, eventVersion uint64) {
+ key := grpcConnectionKey(clusterName, endpoint)
+ m.mu.Lock()
+ m.initEndpointStateLocked()
+ if m.closed {
+ m.mu.Unlock()
+ return
+ }
+ if eventVersion <= m.endpointEventVers[key] {
+ m.mu.Unlock()
+ return
+ }
+ m.endpointEventVers[key] = eventVersion
+ m.endpointGenerations[key]++
+ m.endpointRemoved[key] = !present
+ if present {
+ m.discardEndpointTombstoneLocked(key)
+ }
+ var conn *grpc.ClientConn
+ if value, ok := m.connections.Load(key); ok {
+ conn, _ = value.(*grpc.ClientConn)
+ }
+ m.mu.Unlock()
+ if conn != nil {
+ // A present event starts a new endpoint incarnation. Remove any
+ // connection left over from the previous incarnation before a
new Get
+ // can reuse it.
+ m.remove(key, conn)
+ }
+ if !present {
+ m.finalizeRemovedEndpoint(key)
+ }
+}
+
+func (m *grpcConnectionManager) remove(key string, expected *grpc.ClientConn)
bool {
+ value, ok := m.connections.Load(key)
+ if !ok || value != expected {
+ return false
+ }
+ if m.connections.CompareAndDelete(key, expected) {
+ _ = expected.Close()
+ if m.onRemove != nil {
+ m.onRemove(expected)
+ }
+ return true
+ }
+ return false
+}
+
+func (m *grpcConnectionManager) Close() error {
+ m.mu.Lock()
+ if m.closed {
+ m.mu.Unlock()
+ return nil
+ }
+ m.closed = true
+ m.endpointGenerations = nil
+ m.endpointEventVers = nil
+ m.endpointRefs = nil
+ m.endpointRemoved = nil
+ m.endpointTombstones = nil
+ m.tombstoneOrder = nil
+ var connections []*grpc.ClientConn
+ m.connections.Range(func(key, value any) bool {
+ m.connections.Delete(key)
+ if conn, ok := value.(*grpc.ClientConn); ok {
+ connections = append(connections, conn)
+ }
+ return true
+ })
+ m.mu.Unlock()
+
+ var firstErr error
+ for _, conn := range connections {
+ if err := conn.Close(); err != nil && firstErr == nil {
+ firstErr = err
+ }
+ if m.onRemove != nil {
+ m.onRemove(conn)
+ }
+ }
+ return firstErr
+}
diff --git a/pkg/filter/http/grpcproxy/connection_manager_test.go
b/pkg/filter/http/grpcproxy/connection_manager_test.go
new file mode 100644
index 000000000..905fc067c
--- /dev/null
+++ b/pkg/filter/http/grpcproxy/connection_manager_test.go
@@ -0,0 +1,553 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package grpcproxy
+
+import (
+ "context"
+ "fmt"
+ "net"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/jhump/protoreflect/desc" //nolint:staticcheck //
legacy descriptor API used by grpcproxy.
+ "github.com/jhump/protoreflect/desc/protoparse" //nolint:staticcheck //
legacy parser used by grpcproxy.
+
+ "github.com/stretchr/testify/require"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/connectivity"
+ "google.golang.org/grpc/credentials/insecure"
+ reflectpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"
+ "google.golang.org/grpc/status"
+)
+
+import (
+ ct "github.com/apache/dubbo-go-pixiu/pkg/context"
+)
+
+func startTestGRPCServer(t *testing.T) string {
+ t.Helper()
+
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+
+ server := grpc.NewServer()
+ go func() {
+ _ = server.Serve(listener)
+ }()
+ t.Cleanup(func() {
+ server.Stop()
+ _ = listener.Close()
+ })
+ return listener.Addr().String()
+}
+
+func testConnectionManager(t *testing.T, calls *atomic.Int32)
*grpcConnectionManager {
+ t.Helper()
+ return &grpcConnectionManager{
+ dial: func(ctx context.Context, endpoint string)
(*grpc.ClientConn, error) {
+ calls.Add(1)
+ return grpc.DialContext(ctx, endpoint,
grpc.WithTransportCredentials(insecure.NewCredentials())) //nolint:staticcheck
// the test verifies context-bounded dialing.
+ },
+ dialTimeout: time.Second,
+ }
+}
+
+func TestGRPCConnectionManagerSharesConcurrentConnection(t *testing.T) {
+ endpoint := startTestGRPCServer(t)
+ var dialCalls atomic.Int32
+ manager := testConnectionManager(t, &dialCalls)
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ const requests = 64
+ connections := make([]*grpc.ClientConn, requests)
+ errs := make(chan error, requests)
+ var wg sync.WaitGroup
+ for i := 0; i < requests; i++ {
+ wg.Add(1)
+ go func(index int) {
+ defer wg.Done()
+ conn, err := manager.Get(context.Background(),
grpcConnectionKey("cluster", endpoint), endpoint)
+ if err != nil {
+ errs <- err
+ return
+ }
+ connections[index] = conn
+ }(i)
+ }
+ wg.Wait()
+ close(errs)
+ for err := range errs {
+ require.NoError(t, err)
+ }
+
+ require.Equal(t, int32(1), dialCalls.Load())
+ for _, conn := range connections[1:] {
+ require.Same(t, connections[0], conn)
+ }
+}
+
+func TestGRPCConnectionManagerRecreatesUnhealthyConnection(t *testing.T) {
+ endpoint := startTestGRPCServer(t)
+ var dialCalls atomic.Int32
+ manager := testConnectionManager(t, &dialCalls)
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ key := grpcConnectionKey("cluster", endpoint)
+ first, err := manager.Get(context.Background(), key, endpoint)
+ require.NoError(t, err)
+ require.NoError(t, first.Close())
+ manager.Invalidate(key, first)
+
+ second, err := manager.Get(context.Background(), key, endpoint)
+ require.NoError(t, err)
+ require.NotSame(t, first, second)
+ require.Equal(t, int32(2), dialCalls.Load())
+}
+
+func TestGRPCConnectionManagerRetainsTransientFailureConnection(t *testing.T) {
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ endpoint := listener.Addr().String()
+ require.NoError(t, listener.Close())
+
+ conn, err := grpc.NewClient("passthrough:///"+endpoint,
grpc.WithTransportCredentials(insecure.NewCredentials()))
+ require.NoError(t, err)
+ conn.Connect()
+ require.Eventually(t, func() bool {
+ return conn.GetState() == connectivity.TransientFailure
+ }, time.Second, time.Millisecond)
+
+ manager := newGRPCConnectionManager()
+ key := grpcConnectionKey("cluster", endpoint)
+ manager.connections.Store(key, conn)
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ got, ok := manager.loadHealthy(key)
+ require.True(t, ok)
+ require.Same(t, conn, got)
+ manager.Invalidate(key, conn)
+ _, stillCached := manager.connections.Load(key)
+ require.True(t, stillCached)
+}
+
+func TestGRPCConnectionManagerHonorsCallerTimeoutWhileCreating(t *testing.T) {
+ var dialCalls atomic.Int32
+ manager := &grpcConnectionManager{
+ dial: func(ctx context.Context, _ string) (*grpc.ClientConn,
error) {
+ dialCalls.Add(1)
+ <-ctx.Done()
+ return nil, ctx.Err()
+ },
+ dialTimeout: time.Second,
+ }
+
+ ctx, cancel := context.WithTimeout(context.Background(),
10*time.Millisecond)
+ defer cancel()
+ _, err := manager.Get(ctx, "cluster\x00endpoint", "endpoint")
+ require.ErrorIs(t, err, context.DeadlineExceeded)
+ require.Equal(t, int32(1), dialCalls.Load())
+ require.NoError(t, manager.Close())
+}
+
+func TestGRPCConnectionManagerDoesNotPublishRemovedEndpointAfterDial(t
*testing.T) {
+ endpoint := startTestGRPCServer(t)
+ dialStarted := make(chan struct{})
+ releaseDial := make(chan struct{})
+ manager := &grpcConnectionManager{
+ dial: func(ctx context.Context, endpoint string)
(*grpc.ClientConn, error) {
+ close(dialStarted)
+ <-releaseDial
+ return grpc.DialContext(ctx, endpoint,
grpc.WithTransportCredentials(insecure.NewCredentials())) //nolint:staticcheck
// the test verifies endpoint lifecycle.
+ },
+ dialTimeout: time.Second,
+ }
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ key := grpcConnectionKey("cluster", endpoint)
+ result := make(chan error, 1)
+ go func() {
+ _, err := manager.Get(context.Background(), key, endpoint)
+ result <- err
+ }()
+ <-dialStarted
+ manager.RemoveEndpoint("cluster", endpoint)
+ close(releaseDial)
+
+ require.Error(t, <-result)
+ _, ok := manager.connections.Load(key)
+ require.False(t, ok, "a removed endpoint must not be published after
dialing")
+}
+
+func TestGRPCConnectionManagerBoundsEndpointTombstones(t *testing.T) {
+ manager := newGRPCConnectionManager()
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ const churn = maxEndpointTombstones * 4
+ for i := 0; i < churn; i++ {
+ manager.UpdateEndpointState(
+ "cluster",
+ fmt.Sprintf("127.0.0.1:%d", 20000+i),
+ false,
+ uint64(i+1),
+ )
+ }
+
+ manager.mu.Lock()
+ defer manager.mu.Unlock()
+ require.LessOrEqual(t, len(manager.endpointGenerations),
maxEndpointTombstones)
+ require.LessOrEqual(t, len(manager.endpointEventVers),
maxEndpointTombstones)
+ require.LessOrEqual(t, len(manager.endpointTombstones),
maxEndpointTombstones)
+}
+
+func TestGRPCConnectionManagerRejectsEvictedRemovedEndpointFromSnapshot(t
*testing.T) {
+ endpoint := "127.0.0.1:20000"
+ current := make(map[string]bool)
+ var currentMu sync.Mutex
+ var dialCalls atomic.Int32
+ manager := &grpcConnectionManager{
+ dial: func(context.Context, string) (*grpc.ClientConn, error) {
+ dialCalls.Add(1)
+ return nil, fmt.Errorf("unexpected dial")
+ },
+ dialTimeout: time.Second,
+ endpointPresent: func(_, address string) bool {
+ currentMu.Lock()
+ defer currentMu.Unlock()
+ return current[address]
+ },
+ }
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ current[endpoint] = true
+ manager.UpdateEndpointState("cluster", endpoint, true, 1)
+ current[endpoint] = false
+ manager.UpdateEndpointState("cluster", endpoint, false, 2)
+ for i := 0; i < maxEndpointTombstones+1; i++ {
+ address := fmt.Sprintf("127.0.0.1:%d", 21000+i)
+ manager.UpdateEndpointState("cluster", address, false,
uint64(i+3))
+ }
+
+ _, err := manager.Get(context.Background(),
grpcConnectionKey("cluster", endpoint), endpoint)
+ require.EqualError(t, err, "grpc endpoint was removed")
+ require.Zero(t, dialCalls.Load())
+}
+
+func TestGRPCConnectionManagerReclaimsRequestOnlyEndpointState(t *testing.T) {
+ manager := &grpcConnectionManager{
+ dial: func(context.Context, string) (*grpc.ClientConn, error) {
+ return nil, fmt.Errorf("dial failed")
+ },
+ dialTimeout: time.Second,
+ }
+ t.Cleanup(func() { require.NoError(t, manager.Close()) })
+
+ _, err := manager.Get(context.Background(), "cluster\x00endpoint",
"endpoint")
+ require.EqualError(t, err, "dial failed")
+
+ manager.mu.Lock()
+ defer manager.mu.Unlock()
+ require.Empty(t, manager.endpointGenerations)
+ require.Empty(t, manager.endpointEventVers)
+}
+
+func TestGRPCConnectionManagerClosePreventsNewConnections(t *testing.T) {
+ var dialCalls atomic.Int32
+ manager := testConnectionManager(t, &dialCalls)
+ require.NoError(t, manager.Close())
+
+ _, err := manager.Get(context.Background(),
grpcConnectionKey("cluster", "endpoint"), "endpoint")
+ require.Error(t, err)
+ require.Equal(t, int32(0), dialCalls.Load())
+}
+
+func TestDescriptorSourceHonorsRequestTimeout(t *testing.T) {
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+
+ server := grpc.NewServer()
+ reflectpb.RegisterServerReflectionServer(server,
&blockingReflectionServer{})
+ go func() {
+ _ = server.Serve(listener)
+ }()
+ t.Cleanup(func() {
+ server.Stop()
+ _ = listener.Close()
+ })
+
+ conn, err := grpc.NewClient(listener.Addr().String(),
grpc.WithTransportCredentials(insecure.NewCredentials()))
+ require.NoError(t, err)
+ t.Cleanup(func() { require.NoError(t, conn.Close()) })
+
+ descriptor := &Descriptor{}
+ ctx, cancel := context.WithTimeout(context.Background(),
100*time.Millisecond)
+ defer cancel()
+ source, err := descriptor.getServerDescriptorSourceCtx(
+ context.WithValue(ctx, ct.ContextKey(GrpcClientConnKey), conn),
+ &Config{},
+ )
+ require.NoError(t, err)
+
+ _, err = source.FindSymbol("test.SlowService")
+ require.Error(t, err)
+ if st, ok := status.FromError(err); ok {
+ require.Equal(t, codes.DeadlineExceeded, st.Code())
+ } else {
+ require.ErrorIs(t, err, context.DeadlineExceeded)
+ }
+}
+
+type blockingReflectionServer struct{}
+
+func (blockingReflectionServer) ServerReflectionInfo(stream
reflectpb.ServerReflection_ServerReflectionInfoServer) error {
+ <-stream.Context().Done()
+ return stream.Context().Err()
+}
+
+type countingDescriptorSource struct {
+ descriptor desc.Descriptor
+ findCalls atomic.Int32
+}
+
+func (s *countingDescriptorSource) ListServices() ([]string, error) {
+ return []string{"test.Greeter"}, nil
+}
+
+func (s *countingDescriptorSource) FindSymbol(string) (desc.Descriptor, error)
{
+ s.findCalls.Add(1)
+ return s.descriptor, nil
+}
+
+func (s *countingDescriptorSource) AllExtensionsForType(string)
([]*desc.FieldDescriptor, error) {
+ return nil, nil
+}
+
+func TestDescriptorCachesMethodLookupPerConnection(t *testing.T) {
+ files, err := (protoparse.Parser{
+ Accessor: protoparse.FileContentsFromMap(map[string]string{
+ "test.proto": `syntax = "proto3";
+package test;
+
+service Greeter {
+ rpc Hello(Request) returns (Response);
+}
+
+message Request {}
+message Response {}
+`,
+ }),
+ }).ParseFiles("test.proto")
+ require.NoError(t, err)
+
+ source := &countingDescriptorSource{descriptor:
files[0].FindSymbol("test.Greeter")}
+ descriptor := &Descriptor{}
+ conn := &grpc.ClientConn{}
+ _, err = descriptor.getMethodDescriptor(source, conn, "test.Greeter",
"Hello")
+ require.NoError(t, err)
+
+ const requests = 32
+ methods := make([]*desc.MethodDescriptor, requests)
+ errs := make([]error, requests)
+ var wg sync.WaitGroup
+ for i := 0; i < requests; i++ {
+ wg.Add(1)
+ go func(index int) {
+ defer wg.Done()
+ methods[index], errs[index] =
descriptor.getMethodDescriptor(source, conn, "test.Greeter", "Hello")
+ }(i)
+ }
+ wg.Wait()
+
+ for index, method := range methods {
+ require.NoError(t, errs[index])
+ require.NotNil(t, method)
+ require.Equal(t, "Hello", method.GetName())
+ }
+ require.Equal(t, int32(1), source.findCalls.Load())
+}
+
+func TestDescriptorLookupDoesNotShareRequestBoundSource(t *testing.T) {
+ files, err := (protoparse.Parser{
+ Accessor: protoparse.FileContentsFromMap(map[string]string{
+ "test.proto": `syntax = "proto3";
+package test;
+
+service Greeter {
+ rpc Hello(Request) returns (Response);
+}
+
+message Request {}
+message Response {}
+`,
+ }),
+ }).ParseFiles("test.proto")
+ require.NoError(t, err)
+
+ firstSource := &blockingDescriptorSource{
+ descriptor: files[0].FindSymbol("test.Greeter"),
+ started: make(chan struct{}),
+ release: make(chan struct{}),
+ }
+ secondSource := &countingDescriptorSource{descriptor:
files[0].FindSymbol("test.Greeter")}
+ descriptor := &Descriptor{}
+ conn := &grpc.ClientConn{}
+
+ firstResult := make(chan error, 1)
+ go func() {
+ _, lookupErr := descriptor.getMethodDescriptor(firstSource,
conn, "test.Greeter", "Hello")
+ firstResult <- lookupErr
+ }()
+ <-firstSource.started
+
+ secondResult := make(chan error, 1)
+ go func() {
+ _, lookupErr := descriptor.getMethodDescriptor(secondSource,
conn, "test.Greeter", "Hello")
+ secondResult <- lookupErr
+ }()
+
+ select {
+ case lookupErr := <-secondResult:
+ require.NoError(t, lookupErr)
+ case <-time.After(200 * time.Millisecond):
+ close(firstSource.release)
+ t.Fatal("request-bound descriptor lookup was shared with a
canceled/slow caller")
+ }
+
+ close(firstSource.release)
+ require.NoError(t, <-firstResult)
+}
+
+type blockingDescriptorSource struct {
+ descriptor desc.Descriptor
+ started chan struct{}
+ release chan struct{}
+}
+
+func (s *blockingDescriptorSource) ListServices() ([]string, error) {
+ return []string{"test.Greeter"}, nil
+}
+
+func (s *blockingDescriptorSource) FindSymbol(string) (desc.Descriptor, error)
{
+ close(s.started)
+ <-s.release
+ return s.descriptor, nil
+}
+
+func (s *blockingDescriptorSource) AllExtensionsForType(string)
([]*desc.FieldDescriptor, error) {
+ return nil, nil
+}
+
+func TestDescriptorRemovalForOneConnectionDoesNotInvalidateAnother(t
*testing.T) {
+ files, err := (protoparse.Parser{
+ Accessor: protoparse.FileContentsFromMap(map[string]string{
+ "test.proto": `syntax = "proto3";
+package test;
+
+service Greeter {
+ rpc Hello(Request) returns (Response);
+}
+
+message Request {}
+message Response {}
+`,
+ }),
+ }).ParseFiles("test.proto")
+ require.NoError(t, err)
+
+ source := &blockingDescriptorSource{
+ descriptor: files[0].FindSymbol("test.Greeter"),
+ started: make(chan struct{}),
+ release: make(chan struct{}),
+ }
+ descriptor := &Descriptor{}
+ connA := &grpc.ClientConn{}
+ connB := &grpc.ClientConn{}
+ _, err = descriptor.getMethodDescriptor(
+ &countingDescriptorSource{descriptor:
files[0].FindSymbol("test.Greeter")},
+ connB, "test.Greeter", "Hello",
+ )
+ require.NoError(t, err)
+
+ result := make(chan error, 1)
+ go func() {
+ _, lookupErr := descriptor.getMethodDescriptor(source, connA,
"test.Greeter", "Hello")
+ result <- lookupErr
+ }()
+ <-source.started
+
+ descriptor.removeConnection(connB)
+ close(source.release)
+
+ require.NoError(t, <-result)
+}
+
+func TestDescriptorRemovalSeparatesNewLookupOnSameConnection(t *testing.T) {
+ files, err := (protoparse.Parser{
+ Accessor: protoparse.FileContentsFromMap(map[string]string{
+ "test.proto": `syntax = "proto3";
+package test;
+
+service Greeter {
+ rpc Hello(Request) returns (Response);
+}
+
+message Request {}
+message Response {}
+`,
+ }),
+ }).ParseFiles("test.proto")
+ require.NoError(t, err)
+
+ oldSource := &blockingDescriptorSource{
+ descriptor: files[0].FindSymbol("test.Greeter"),
+ started: make(chan struct{}),
+ release: make(chan struct{}),
+ }
+ newSource := &countingDescriptorSource{descriptor:
files[0].FindSymbol("test.Greeter")}
+ descriptor := &Descriptor{}
+ conn := &grpc.ClientConn{}
+
+ oldResult := make(chan error, 1)
+ go func() {
+ _, lookupErr := descriptor.getMethodDescriptor(oldSource, conn,
"test.Greeter", "Hello")
+ oldResult <- lookupErr
+ }()
+ <-oldSource.started
+ descriptor.removeConnection(conn)
+
+ newResult := make(chan error, 1)
+ go func() {
+ _, lookupErr := descriptor.getMethodDescriptor(newSource, conn,
"test.Greeter", "Hello")
+ newResult <- lookupErr
+ }()
+ select {
+ case lookupErr := <-newResult:
+ require.NoError(t, lookupErr)
+ case <-time.After(200 * time.Millisecond):
+ close(oldSource.release)
+ t.Fatal("new lookup joined the removed connection's in-flight
lookup")
+ }
+
+ close(oldSource.release)
+ require.Error(t, <-oldResult)
+}
diff --git a/pkg/filter/http/grpcproxy/descriptor.go
b/pkg/filter/http/grpcproxy/descriptor.go
index 40cc8e7aa..495b371d6 100644
--- a/pkg/filter/http/grpcproxy/descriptor.go
+++ b/pkg/filter/http/grpcproxy/descriptor.go
@@ -23,6 +23,7 @@ import (
"os"
"path/filepath"
"strings"
+ "sync"
)
import (
@@ -42,7 +43,27 @@ import (
)
type Descriptor struct {
- fileSource *fileSource
+ fileSource *fileSource
+ methodMu sync.RWMutex
+ methodDescs map[*grpc.ClientConn]map[string]*desc.MethodDescriptor
+ // connectionStates invalidates only lookups for the connection that
+ // was removed. closeGeneration invalidates all in-flight lookups on
close.
+ connectionStates map[*grpc.ClientConn]*descriptorConnectionState
+ closeGeneration uint64
+ nextStateSequence uint64
+ closed bool
+}
+
+type descriptorConnectionState struct {
+ generation uint64
+}
+
+type serviceNotExposedError struct {
+ service string
+}
+
+func (e *serviceNotExposedError) Error() string {
+ return fmt.Sprintf("service not exposed: %s", e.service)
}
func (dr *Descriptor) GetCurrentDescriptorSource(ctx context.Context)
(DescriptorSource, error) {
@@ -90,7 +111,7 @@ func (dr *Descriptor) getDescriptorCompose(ctx
context.Context, cfg *Config) (De
cs := &compositeSource{}
cs.reflection, err = dr.getServerDescriptorSourceCtx(ctx, cfg)
- cs.file = dr.fileSource
+ cs.file = dr.getFileSource()
return cs, err
}
@@ -117,24 +138,121 @@ func (dr *Descriptor)
getServerDescriptorSourceCtx(refCtx context.Context, cfg *
default:
err = errors.Errorf("found a value of type %s, which is not
*grpc.ClientConn, ", t)
}
- return &serverSource{client: grpcreflect.NewClient(refCtx,
reflectpb.NewServerReflectionClient(cc))}, err
+ if err != nil {
+ return nil, err
+ }
+
+ // The reflection client is created per lookup and bound to the request
+ // context so every remote reflection RPC honors the request timeout.
+ // It must not be cached connection-scoped: grpcreflect reuses the root
+ // context for every RPC, and a cached client would lose the deadline
and
+ // keep the per-request timeout from applying. The method descriptor
+ // cache in getMethodDescriptor below is what avoids repeating the
+ // reflection RPC after the first lookup.
+ return &serverSource{client: grpcreflect.NewClientV1Alpha(refCtx,
reflectpb.NewServerReflectionClient(cc))}, nil
}
// nolint
func (dr *Descriptor) getServerDescriptorSource(refCtx context.Context, cc
*grpc.ClientConn) DescriptorSource {
- return &serverSource{client: grpcreflect.NewClient(refCtx,
reflectpb.NewServerReflectionClient(cc))}
+ return &serverSource{client: grpcreflect.NewClientV1Alpha(refCtx,
reflectpb.NewServerReflectionClient(cc))}
}
-func (dr *Descriptor) getFileDescriptorCompose(ctx context.Context, cfg
*Config) (DescriptorSource, error) {
- if dr.fileSource == nil {
- dr.initFileDescriptorSource(cfg)
+func (dr *Descriptor) removeConnection(cc *grpc.ClientConn) {
+ if cc == nil {
+ return
+ }
+ dr.methodMu.Lock()
+ delete(dr.methodDescs, cc)
+ state := dr.connectionStates[cc]
+ if state != nil {
+ dr.nextStateSequence++
+ state.generation = dr.nextStateSequence
+ // Keep the state alive for any in-flight lookup, but do not
retain the
+ // removed connection in the descriptor's long-lived map.
+ delete(dr.connectionStates, cc)
}
- return dr.fileSource, nil
+ dr.methodMu.Unlock()
}
-func (dr *Descriptor) initFileDescriptorSource(cfg *Config) *Descriptor {
+func (dr *Descriptor) Close() {
+ dr.methodMu.Lock()
+ dr.closed = true
+ dr.methodDescs = nil
+ dr.connectionStates = nil
+ dr.closeGeneration++
+ dr.methodMu.Unlock()
+}
+
+func (dr *Descriptor) getMethodDescriptor(source DescriptorSource, cc
*grpc.ClientConn, service, method string) (*desc.MethodDescriptor, error) {
+ key := service + "\x00" + method
+ dr.methodMu.Lock()
+ if dr.closed {
+ dr.methodMu.Unlock()
+ return nil, errors.New("descriptor is closed")
+ }
+ if methods := dr.methodDescs[cc]; methods != nil {
+ if descriptor, ok := methods[key]; ok {
+ dr.methodMu.Unlock()
+ return descriptor, nil
+ }
+ }
+ state := dr.connectionStates[cc]
+ if state == nil {
+ if dr.connectionStates == nil {
+ dr.connectionStates =
make(map[*grpc.ClientConn]*descriptorConnectionState)
+ }
+ dr.nextStateSequence++
+ state = &descriptorConnectionState{generation:
dr.nextStateSequence}
+ dr.connectionStates[cc] = state
+ }
+ connectionGeneration := state.generation
+ closeGeneration := dr.closeGeneration
+ dr.methodMu.Unlock()
+ // Do not singleflight this lookup across requests. DescriptorSource
carries
+ // the request context used by server reflection, so sharing the first
+ // caller's source would let its timeout or cancellation fail other
callers.
+ // The per-connection method cache below still removes repeated
reflection
+ // calls after the first successful lookup.
+ dscp, err := source.FindSymbol(service)
+ if err != nil {
+ return nil, err
+ }
+ svcDesc, ok := dscp.(*desc.ServiceDescriptor)
+ if !ok {
+ return nil, &serviceNotExposedError{service: service}
+ }
+ descriptor := svcDesc.FindMethodByName(method)
+ if descriptor == nil {
+ return nil, fmt.Errorf("method not found: %s/%s", service,
method)
+ }
- if dr.fileSource != nil {
+ dr.methodMu.Lock()
+ defer dr.methodMu.Unlock()
+ if dr.closed || dr.closeGeneration != closeGeneration ||
state.generation != connectionGeneration {
+ return nil, errors.New("descriptor cache invalidated")
+ }
+ if methods := dr.methodDescs[cc]; methods != nil {
+ if cached, ok := methods[key]; ok {
+ return cached, nil
+ }
+ }
+ if dr.methodDescs == nil {
+ dr.methodDescs =
make(map[*grpc.ClientConn]map[string]*desc.MethodDescriptor)
+ }
+ if dr.methodDescs[cc] == nil {
+ dr.methodDescs[cc] = make(map[string]*desc.MethodDescriptor)
+ }
+ dr.methodDescs[cc][key] = descriptor
+ return descriptor, nil
+}
+
+func (dr *Descriptor) getFileDescriptorCompose(ctx context.Context, cfg
*Config) (DescriptorSource, error) {
+ dr.initFileDescriptorSource(cfg)
+ return dr.getFileSource(), nil
+}
+
+func (dr *Descriptor) initFileDescriptorSource(cfg *Config) *Descriptor {
+ if dr.getFileSource() != nil {
return dr
}
@@ -145,11 +263,21 @@ func (dr *Descriptor) initFileDescriptorSource(cfg
*Config) *Descriptor {
return dr
}
- dr.fileSource = descriptor
+ dr.methodMu.Lock()
+ if dr.fileSource == nil {
+ dr.fileSource = descriptor
+ }
+ dr.methodMu.Unlock()
return dr
}
+func (dr *Descriptor) getFileSource() *fileSource {
+ dr.methodMu.RLock()
+ defer dr.methodMu.RUnlock()
+ return dr.fileSource
+}
+
func loadFileSource(gc *Config) (*fileSource, error) {
var fsrc fileSource
diff --git a/pkg/filter/http/grpcproxy/grpc.go
b/pkg/filter/http/grpcproxy/grpc.go
index ca43e3177..9148dc862 100644
--- a/pkg/filter/http/grpcproxy/grpc.go
+++ b/pkg/filter/http/grpcproxy/grpc.go
@@ -40,7 +40,6 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
- "google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
@@ -90,21 +89,24 @@ type (
cfg *Config
// grpc descriptor source factory
descriptor *Descriptor
- // hold grpc.ClientConns, key format: cluster name + "." +
endpoint
- pools map[string]*sync.Pool
+ // hold grpc.ClientConns, key format: cluster name + NUL +
endpoint
+ connections *grpcConnectionManager
+ removeEndpointHandler func()
- extReg *dynamic.ExtensionRegistry
- registered map[string]bool
+ extReg *dynamic.ExtensionRegistry
+ registered map[string]bool
+ extensionMu *sync.RWMutex
}
Filter struct {
cfg *Config
// grpc descriptor source factory
descriptor *Descriptor
- // hold grpc.ClientConns, key format: cluster name + "." +
endpoint
- pools map[string]*sync.Pool
+ // hold grpc.ClientConns, key format: cluster name + NUL +
endpoint
+ connections *grpcConnectionManager
- extReg *dynamic.ExtensionRegistry
- registered map[string]bool
+ extReg *dynamic.ExtensionRegistry
+ registered map[string]bool
+ extensionMu *sync.RWMutex
}
// Config describe the config of AccessFilter
@@ -148,12 +150,40 @@ func (p *Plugin) Kind() string {
}
func (p *Plugin) CreateFilterFactory() (filter.HttpFilterFactory, error) {
- return &FilterFactory{cfg: &Config{DescriptorSourceStrategy: AUTO},
descriptor: &Descriptor{}}, nil
+ descriptor := &Descriptor{}
+ connections := newGRPCConnectionManager()
+ connections.onRemove = descriptor.removeConnection
+ var removeEndpointHandler func()
+ if clusterManager := server.GetClusterManager(); clusterManager != nil {
+ connections.endpointPresent = func(key, endpoint string) bool {
+ clusterName, _, ok := strings.Cut(key, "\x00")
+ return ok &&
clusterManager.HasEndpointAddress(clusterName, endpoint)
+ }
+ removeEndpointHandler =
clusterManager.AddEndpointStateHandler(func(clusterName, endpoint string,
present bool, version uint64) {
+ connections.UpdateEndpointState(clusterName, endpoint,
present, version)
+ })
+ }
+ return &FilterFactory{
+ cfg: &Config{DescriptorSourceStrategy: AUTO},
+ descriptor: descriptor,
+ connections: connections,
+ removeEndpointHandler: removeEndpointHandler,
+ extReg: &dynamic.ExtensionRegistry{},
+ registered: make(map[string]bool),
+ extensionMu: &sync.RWMutex{},
+ }, nil
}
func (factory *FilterFactory) PrepareFilterChain(ctx *http.HttpContext, chain
filter.FilterChain) error {
// Deep copy config to avoid pointer sharing (factory.cfg may change at
runtime)
- f := &Filter{cfg: factory.cfg.DeepCopy(), descriptor:
factory.descriptor, pools: factory.pools, extReg: factory.extReg, registered:
factory.registered}
+ f := &Filter{
+ cfg: factory.cfg.DeepCopy(),
+ descriptor: factory.descriptor,
+ connections: factory.connections,
+ extReg: factory.extReg,
+ registered: factory.registered,
+ extensionMu: factory.extensionMu,
+ }
chain.AppendDecodeFilters(f)
return nil
}
@@ -199,21 +229,13 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
defer cancel()
ep := e.Address.GetAddress()
- p, ok := f.pools[strings.Join([]string{re.Cluster, ep}, ".")]
- if !ok {
- p = &sync.Pool{}
- }
-
- clientConn, ok = p.Get().(*grpc.ClientConn)
- if !ok || clientConn == nil {
- // TODO(Kenway): Support Credential and TLS
- clientConn, err = grpc.DialContext(ctx, ep,
grpc.WithTransportCredentials(insecure.NewCredentials()))
- if err != nil || clientConn == nil {
- logger.Errorf("%s err {failed to connect to grpc
service provider}", loggerHeader)
- errResp :=
http.ServiceUnavailable.WithError(fmt.Errorf("endpoint not found: %w", err))
- c.SendLocalReply(errResp.Status, errResp.ToJSON())
- return filter.Stop
- }
+ connectionKey := grpcConnectionKey(re.Cluster, ep)
+ clientConn, err = f.connections.Get(ctx, connectionKey, ep)
+ if err != nil || clientConn == nil {
+ logger.Errorf("%s err {failed to connect to grpc service
provider}: %v", loggerHeader, err)
+ errResp :=
http.ServiceUnavailable.WithError(fmt.Errorf("endpoint not found: %w", err))
+ c.SendLocalReply(errResp.Status, errResp.ToJSON())
+ return filter.Stop
}
// get DescriptorSource, contain file and reflection
@@ -227,24 +249,20 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
//put DescriptorSource concurrent, del if no need
ctx = context.WithValue(ctx, ct.ContextKey(DescriptorSourceKey), source)
- dscp, err := source.FindSymbol(svc)
+ mthDesc, err := f.descriptor.getMethodDescriptor(source, clientConn,
svc, mth)
if err != nil {
- logger.Errorf("%s err {%s}", loggerHeader, "request path
invalid")
+ if _, ok := err.(*serviceNotExposedError); ok {
+ logger.Errorf("%s err {service not expose, %s}",
loggerHeader, svc)
+ errResp := http.BadRequest.WithError(err)
+ c.SendLocalReply(errResp.Status, errResp.ToJSON())
+ return filter.Stop
+ }
+ logger.Errorf("%s err {request path invalid, service: %s,
method: %s, cause: %v}", loggerHeader, svc, mth, err)
errResp := http.MethodNotAllowed.New()
c.SendLocalReply(errResp.Status, errResp.ToJSON())
return filter.Stop
}
- svcDesc, ok := dscp.(*desc.ServiceDescriptor)
- if !ok {
- logger.Errorf("%s err {service not expose, %s}", loggerHeader,
svc)
- errResp := http.BadRequest.WithError(fmt.Errorf("service not
exposed: %s", svc))
- c.SendLocalReply(errResp.Status, errResp.ToJSON())
- return filter.Stop
- }
-
- mthDesc := svcDesc.FindMethodByName(mth)
-
err = f.registerExtension(source, mthDesc)
if err != nil {
logger.Errorf("%s err {%s}", loggerHeader, "register extension
failed")
@@ -281,6 +299,7 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
logger.Errorf("%s err {gRPC client error, code: %s,
msg: %s}", loggerHeader, st.Code(), st.Message())
errResp := http.BadGateway.WithError(fmt.Errorf("gRPC
client error: %w", err))
c.SendLocalReply(errResp.Status, errResp.ToJSON())
+ f.connections.Invalidate(connectionKey, clientConn)
return filter.Stop
}
// Handle server-side gRPC errors
@@ -289,11 +308,13 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
logger.Errorf("%s err {failed to invoke grpc
service provider because timeout, err:%s}", loggerHeader, err.Error())
errResp :=
http.GatewayTimeout.WithError(fmt.Errorf("upstream timeout: %w", err))
c.SendLocalReply(errResp.Status,
errResp.ToJSON())
+ f.connections.Invalidate(connectionKey,
clientConn)
return filter.Stop
}
logger.Errorf("%s err {failed to invoke grpc service
provider, %s}", loggerHeader, err.Error())
errResp :=
http.ServiceUnavailable.WithError(fmt.Errorf("gRPC invoke error: %w", err))
c.SendLocalReply(errResp.Status, errResp.ToJSON())
+ f.connections.Invalidate(connectionKey, clientConn)
return filter.Stop
}
} else if err != nil {
@@ -301,6 +322,7 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
logger.Errorf("%s err {failed to invoke grpc service provider,
%s}", loggerHeader, err.Error())
errResp := http.ServiceUnavailable.WithError(fmt.Errorf("gRPC
invoke error: %w", err))
c.SendLocalReply(errResp.Status, errResp.ToJSON())
+ f.connections.Invalidate(connectionKey, clientConn)
return filter.Stop
}
@@ -323,19 +345,48 @@ func (f *Filter) Decode(c *http.HttpContext)
filter.FilterStatus {
Trailer: th,
Request: c.Request,
}
- p.Put(clientConn)
return filter.Continue
}
+func grpcConnectionKey(cluster, endpoint string) string {
+ return cluster + "\x00" + endpoint
+}
+
func (f *Filter) registerExtension(source DescriptorSource, mthDesc
*desc.MethodDescriptor) error {
- err := RegisterExtension(source, f.extReg, mthDesc.GetInputType(),
f.registered)
- if err != nil {
- return perrors.New("register extension failed")
+ inputDesc := mthDesc.GetInputType()
+ outputDesc := mthDesc.GetOutputType()
+ if f.extensionMu == nil || f.extReg == nil || f.registered == nil {
+ if err := RegisterExtension(source, f.extReg, inputDesc,
f.registered); err != nil {
+ return perrors.New("register extension failed")
+ }
+ if err := RegisterExtension(source, f.extReg, outputDesc,
f.registered); err != nil {
+ return perrors.New("register extension failed")
+ }
+ return nil
}
- err = RegisterExtension(source, f.extReg, mthDesc.GetOutputType(),
f.registered)
- if err != nil {
- return perrors.New("register extension failed")
+ inputName := inputDesc.GetFullyQualifiedName()
+ outputName := outputDesc.GetFullyQualifiedName()
+ f.extensionMu.RLock()
+ registered := f.registered[inputName] && f.registered[outputName]
+ f.extensionMu.RUnlock()
+ if registered {
+ return nil
+ }
+
+ f.extensionMu.Lock()
+ defer f.extensionMu.Unlock()
+ if !f.registered[inputName] {
+ if err := RegisterExtension(source, f.extReg, inputDesc,
f.registered); err != nil {
+ return perrors.New("register extension failed")
+ }
+ f.registered[inputName] = true
+ }
+ if !f.registered[outputName] {
+ if err := RegisterExtension(source, f.extReg, outputDesc,
f.registered); err != nil {
+ return perrors.New("register extension failed")
+ }
+ f.registered[outputName] = true
}
return nil
}
@@ -426,6 +477,23 @@ func (factory *FilterFactory) Apply() error {
return nil
}
+// Close releases all backend connections owned by this filter factory.
+func (factory *FilterFactory) Close() error {
+ var firstErr error
+ if factory.connections != nil {
+ if err := factory.connections.Close(); err != nil {
+ firstErr = err
+ }
+ }
+ if factory.descriptor != nil {
+ factory.descriptor.Close()
+ }
+ if factory.removeEndpointHandler != nil {
+ factory.removeEndpointHandler()
+ }
+ return firstErr
+}
+
func configCheck(cfg *Config) error {
if len(cfg.DescriptorSourceStrategy.Val()) == 0 {
return perrors.Errorf("grpc descriptor source config
`descriptor_source_strategy` is `%s`, maybe set it `%s`",
cfg.DescriptorSourceStrategy.String(), AUTO)
diff --git a/pkg/listener/http/http_listener.go
b/pkg/listener/http/http_listener.go
index 005bbaaac..5a65e631b 100644
--- a/pkg/listener/http/http_listener.go
+++ b/pkg/listener/http/http_listener.go
@@ -49,7 +49,9 @@ type (
// ListenerService the facade of a listener
HttpListenerService struct {
listener.BaseListenerService
- srv *http.Server
+ srv *http.Server
+ filterMu sync.Mutex
+ filterState *httpFilterChainState
}
// DefaultHttpListener
@@ -58,6 +60,73 @@ type (
}
)
+type httpFilterChainState struct {
+ chain *filterchain.NetworkFilterChain
+ mu sync.Mutex
+ refs int
+ retired bool
+ done chan struct{}
+ closeErr error
+ closeOnce sync.Once
+}
+
+func newHTTPFilterChainState(chain *filterchain.NetworkFilterChain)
*httpFilterChainState {
+ return &httpFilterChainState{chain: chain, done: make(chan struct{})}
+}
+
+func (state *httpFilterChainState) acquire() bool {
+ if state == nil {
+ return false
+ }
+ state.mu.Lock()
+ defer state.mu.Unlock()
+ if state.retired || state.chain == nil {
+ return false
+ }
+ state.refs++
+ return true
+}
+
+func (state *httpFilterChainState) release() {
+ if state == nil {
+ return
+ }
+ state.mu.Lock()
+ state.refs--
+ closeNow := state.retired && state.refs == 0
+ state.mu.Unlock()
+ if closeNow {
+ state.close()
+ }
+}
+
+func (state *httpFilterChainState) retire(wait bool) error {
+ if state == nil {
+ return nil
+ }
+ state.mu.Lock()
+ state.retired = true
+ closeNow := state.refs == 0
+ state.mu.Unlock()
+ if closeNow {
+ state.close()
+ }
+ if wait {
+ <-state.done
+ return state.closeErr
+ }
+ return nil
+}
+
+func (state *httpFilterChainState) close() {
+ state.closeOnce.Do(func() {
+ if state.chain != nil {
+ state.closeErr = state.chain.Close()
+ }
+ close(state.done)
+ })
+}
+
func newHttpListenerService(lc *model.Listener, bs *model.Bootstrap)
(listener.ListenerService, error) {
fc := filterchain.CreateNetworkFilterChain(lc.FilterChain)
return &HttpListenerService{
@@ -65,7 +134,8 @@ func newHttpListenerService(lc *model.Listener, bs
*model.Bootstrap) (listener.L
Config: lc,
FilterChain: fc,
},
- srv: nil,
+ srv: nil,
+ filterState: newHTTPFilterChainState(fc),
}, nil
}
@@ -83,7 +153,15 @@ func (ls *HttpListenerService) Start() error {
}
func (ls *HttpListenerService) Close() error {
- return ls.srv.Close()
+ serverErr := error(nil)
+ if ls.srv != nil {
+ serverErr = ls.srv.Close()
+ }
+ filterErr := ls.closeFilterChain()
+ if serverErr != nil {
+ return serverErr
+ }
+ return filterErr
}
func (ls *HttpListenerService) ShutDown(wg any) error {
@@ -96,13 +174,56 @@ func (ls *HttpListenerService) ShutDown(wg any) error {
cancel()
wg.(*sync.WaitGroup).Done()
}()
- return ls.srv.Shutdown(ctx)
+ serverErr := ls.srv.Shutdown(ctx)
+ filterErr := ls.closeFilterChainAfterShutdown()
+ if serverErr != nil {
+ return serverErr
+ }
+ return filterErr
+}
+
+func (ls *HttpListenerService) closeFilterChain() error {
+ ls.filterMu.Lock()
+ state := ls.filterState
+ if state == nil && ls.FilterChain != nil {
+ state = newHTTPFilterChainState(ls.FilterChain)
+ }
+ ls.FilterChain = nil
+ ls.filterState = nil
+ ls.filterMu.Unlock()
+ return state.retire(true)
+}
+
+// closeFilterChainAfterShutdown must not wait for a request that outlives the
+// HTTP server shutdown deadline while holding filterMu. If an active request
+// still owns a chain lease, defer the close until that request has released
+// it. The shutdown caller can then return the server timeout instead of
+// extending the timeout by the lifetime of the request.
+func (ls *HttpListenerService) closeFilterChainAfterShutdown() error {
+ ls.filterMu.Lock()
+ state := ls.filterState
+ if state == nil && ls.FilterChain != nil {
+ state = newHTTPFilterChainState(ls.FilterChain)
+ }
+ ls.FilterChain = nil
+ ls.filterState = nil
+ ls.filterMu.Unlock()
+ return state.retire(false)
}
func (ls *HttpListenerService) Refresh(c model.Listener) error {
- // There is no need to lock here for now, as there is at most one
NetworkFilter
fc := filterchain.CreateNetworkFilterChain(c.FilterChain)
+ ls.filterMu.Lock()
+ old := ls.filterState
+ if old == nil && ls.FilterChain != nil {
+ old = newHTTPFilterChainState(ls.FilterChain)
+ }
ls.FilterChain = fc
+ ls.filterState = newHTTPFilterChainState(fc)
+ ls.filterMu.Unlock()
+ if old != nil {
+ return old.retire(false)
+ }
return nil
}
@@ -175,7 +296,19 @@ func createDefaultHttpWorker(ls *HttpListenerService)
*DefaultHttpWorker {
// ServeHTTP http request entrance.
func (s *DefaultHttpWorker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- s.ls.FilterChain.ServeHTTP(w, r)
+ s.ls.filterMu.Lock()
+ state := s.ls.filterState
+ if state == nil && s.ls.FilterChain != nil {
+ state = newHTTPFilterChainState(s.ls.FilterChain)
+ s.ls.filterState = state
+ }
+ acquired := state.acquire()
+ s.ls.filterMu.Unlock()
+ if !acquired {
+ return
+ }
+ defer state.release()
+ state.chain.ServeHTTP(w, r)
}
func resolveInt2IntProp(currentV, defaultV int) int {
diff --git a/pkg/listener/http/http_listener_test.go
b/pkg/listener/http/http_listener_test.go
new file mode 100644
index 000000000..ab3682c3e
--- /dev/null
+++ b/pkg/listener/http/http_listener_test.go
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package http
+
+import (
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/filterchain"
+ listenerpkg "github.com/apache/dubbo-go-pixiu/pkg/listener"
+ "github.com/apache/dubbo-go-pixiu/pkg/model"
+)
+
+func TestCloseFilterChainAfterShutdownDoesNotWaitForActiveRequest(t
*testing.T) {
+ listener := &HttpListenerService{
+ BaseListenerService: listenerpkg.BaseListenerService{
+ FilterChain: &filterchain.NetworkFilterChain{},
+ },
+ }
+ listener.filterState = newHTTPFilterChainState(listener.FilterChain)
+ state := listener.filterState
+ require.True(t, state.acquire())
+
+ finished := make(chan struct{})
+ go func() {
+ _ = listener.closeFilterChainAfterShutdown()
+ close(finished)
+ }()
+
+ select {
+ case <-finished:
+ case <-time.After(100 * time.Millisecond):
+ t.Fatal("shutdown filter cleanup waited for an active request")
+ }
+
+ state.release()
+ select {
+ case <-state.done:
+ case <-time.After(time.Second):
+ t.Fatal("deferred filter cleanup did not run after the request
released its chain lease")
+ }
+}
+
+func TestRefreshSwapsChainWithoutWaitingForActiveRequest(t *testing.T) {
+ listener := &HttpListenerService{
+ BaseListenerService: listenerpkg.BaseListenerService{
+ FilterChain: &filterchain.NetworkFilterChain{},
+ },
+ }
+ listener.filterState = newHTTPFilterChainState(listener.FilterChain)
+ oldState := listener.filterState
+ require.True(t, oldState.acquire())
+
+ refreshed := make(chan error, 1)
+ go func() {
+ refreshed <- listener.Refresh(model.Listener{})
+ }()
+
+ select {
+ case err := <-refreshed:
+ require.NoError(t, err)
+ case <-time.After(100 * time.Millisecond):
+ oldState.release()
+ t.Fatal("refresh waited for an active request on the old filter
chain")
+ }
+ require.NotSame(t, oldState, listener.filterState)
+ oldState.release()
+ select {
+ case <-oldState.done:
+ case <-time.After(time.Second):
+ t.Fatal("old filter chain was not closed after the active
request released")
+ }
+}
diff --git a/pkg/server/cluster_manager.go b/pkg/server/cluster_manager.go
index c47304ef0..895bb3b20 100644
--- a/pkg/server/cluster_manager.go
+++ b/pkg/server/cluster_manager.go
@@ -44,7 +44,11 @@ type (
ClusterManager struct {
rw sync.RWMutex
- store *ClusterStore
+ store *ClusterStore
+ endpointRemovalHandlers map[uint64]func(string, string)
+ endpointStateHandlers map[uint64]func(string, string, bool,
uint64)
+ nextEndpointHandlerID uint64
+ nextEndpointEventID uint64
//cConfig []*model.ClusterConfig
}
@@ -75,6 +79,49 @@ func CreateDefaultClusterManager(bs *model.Bootstrap)
*ClusterManager {
return &ClusterManager{store: newClusterStore(bs)}
}
+// AddEndpointRemovalHandler registers a callback invoked after an endpoint is
+// removed. The returned function unregisters the callback.
+func (cm *ClusterManager) AddEndpointRemovalHandler(handler func(string,
string)) func() {
+ if handler == nil {
+ return func() {}
+ }
+ cm.rw.Lock()
+ if cm.endpointRemovalHandlers == nil {
+ cm.endpointRemovalHandlers = make(map[uint64]func(string,
string))
+ }
+ cm.nextEndpointHandlerID++
+ id := cm.nextEndpointHandlerID
+ cm.endpointRemovalHandlers[id] = handler
+ cm.rw.Unlock()
+ return func() {
+ cm.rw.Lock()
+ delete(cm.endpointRemovalHandlers, id)
+ cm.rw.Unlock()
+ }
+}
+
+// AddEndpointStateHandler registers a callback for endpoint additions and
+// removals. version increases for every cluster-store mutation, allowing
+// consumers to discard stale callbacks that run after a newer update.
+func (cm *ClusterManager) AddEndpointStateHandler(handler func(string, string,
bool, uint64)) func() {
+ if handler == nil {
+ return func() {}
+ }
+ cm.rw.Lock()
+ if cm.endpointStateHandlers == nil {
+ cm.endpointStateHandlers = make(map[uint64]func(string, string,
bool, uint64))
+ }
+ cm.nextEndpointHandlerID++
+ id := cm.nextEndpointHandlerID
+ cm.endpointStateHandlers[id] = handler
+ cm.rw.Unlock()
+ return func() {
+ cm.rw.Lock()
+ delete(cm.endpointStateHandlers, id)
+ cm.rw.Unlock()
+ }
+}
+
func newClusterStore(bs *model.Bootstrap) *ClusterStore {
store := &ClusterStore{
clustersMap: map[string]*cluster.Cluster{},
@@ -87,18 +134,28 @@ func newClusterStore(bs *model.Bootstrap) *ClusterStore {
func (cm *ClusterManager) AddCluster(c *model.ClusterConfig) {
cm.rw.Lock()
- defer cm.rw.Unlock()
-
+ old := endpointAddressSnapshot(nil)
+ version := cm.nextEndpointEventIDLocked()
cm.store.IncreaseVersion()
cm.store.AddCluster(c)
+ new := endpointAddressSnapshot([]*model.ClusterConfig{c})
+ cm.rw.Unlock()
+ cm.notifyEndpointChanges(endpointStateChanges(old, new, version))
}
func (cm *ClusterManager) UpdateCluster(new *model.ClusterConfig) {
cm.rw.Lock()
- defer cm.rw.Unlock()
-
+ oldConfig := cm.store.findClusterConfig(new.Name)
+ old := endpointAddressSnapshot([]*model.ClusterConfig{oldConfig})
+ newSnapshot := old
+ if oldConfig != nil {
+ newSnapshot =
endpointAddressSnapshot([]*model.ClusterConfig{new})
+ }
+ version := cm.nextEndpointEventIDLocked()
cm.store.IncreaseVersion()
cm.store.UpdateCluster(new)
+ cm.rw.Unlock()
+ cm.notifyEndpointChanges(endpointStateChanges(old, newSnapshot,
version))
}
// SetEndpoint registers or refreshes a single endpoint in the named
@@ -135,18 +192,168 @@ func (cm *ClusterManager) UpdateCluster(new
*model.ClusterConfig) {
// the whole cluster config via AddCluster/UpdateCluster.
func (cm *ClusterManager) SetEndpoint(clusterName string, endpoint
*model.Endpoint) {
cm.rw.Lock()
- defer cm.rw.Unlock()
-
+ old :=
endpointAddressSnapshot([]*model.ClusterConfig{cm.store.findClusterConfig(clusterName)})
+ version := cm.nextEndpointEventIDLocked()
cm.store.IncreaseVersion()
cm.store.SetEndpoint(clusterName, endpoint)
+ new :=
endpointAddressSnapshot([]*model.ClusterConfig{cm.store.findClusterConfig(clusterName)})
+ cm.rw.Unlock()
+ cm.notifyEndpointChanges(endpointStateChanges(old, new, version))
}
func (cm *ClusterManager) DeleteEndpoint(clusterName string, endpointID
string) {
cm.rw.Lock()
- defer cm.rw.Unlock()
-
+ old :=
endpointAddressSnapshot([]*model.ClusterConfig{cm.store.findClusterConfig(clusterName)})
+ version := cm.nextEndpointEventIDLocked()
cm.store.IncreaseVersion()
cm.store.DeleteEndpoint(clusterName, endpointID)
+ new :=
endpointAddressSnapshot([]*model.ClusterConfig{cm.store.findClusterConfig(clusterName)})
+ cm.rw.Unlock()
+ cm.notifyEndpointChanges(endpointStateChanges(old, new, version))
+}
+
+func (cm *ClusterManager) notifyEndpointRemoval(clusterName, endpointAddress
string) {
+ cm.notifyEndpointRemovals([]endpointRemoval{{clusterName: clusterName,
address: endpointAddress}})
+}
+
+func (cm *ClusterManager) notifyEndpointRemovals(removals []endpointRemoval) {
+ cm.rw.RLock()
+ handlers := make([]func(string, string), 0,
len(cm.endpointRemovalHandlers))
+ for _, handler := range cm.endpointRemovalHandlers {
+ handlers = append(handlers, handler)
+ }
+ cm.rw.RUnlock()
+ for _, removal := range removals {
+ if removal.address == "" {
+ continue
+ }
+ for _, handler := range handlers {
+ handler(removal.clusterName, removal.address)
+ }
+ }
+}
+
+func (cm *ClusterManager) notifyEndpointChanges(changes []endpointStateChange)
{
+ cm.rw.RLock()
+ removalHandlers := make([]func(string, string), 0,
len(cm.endpointRemovalHandlers))
+ for _, handler := range cm.endpointRemovalHandlers {
+ removalHandlers = append(removalHandlers, handler)
+ }
+ stateHandlers := make([]func(string, string, bool, uint64), 0,
len(cm.endpointStateHandlers))
+ for _, handler := range cm.endpointStateHandlers {
+ stateHandlers = append(stateHandlers, handler)
+ }
+ cm.rw.RUnlock()
+ for _, change := range changes {
+ for _, handler := range stateHandlers {
+ handler(change.clusterName, change.address,
change.present, change.version)
+ }
+ if change.present {
+ continue
+ }
+ for _, handler := range removalHandlers {
+ handler(change.clusterName, change.address)
+ }
+ }
+}
+
+type endpointRemoval struct {
+ clusterName string
+ address string
+}
+
+type endpointStateChange struct {
+ clusterName string
+ address string
+ present bool
+ version uint64
+}
+
+func (cm *ClusterManager) nextEndpointEventIDLocked() uint64 {
+ cm.nextEndpointEventID++
+ return cm.nextEndpointEventID
+}
+
+func endpointAddressSnapshot(configs []*model.ClusterConfig)
map[string]map[string]struct{} {
+ result := make(map[string]map[string]struct{})
+ for _, config := range configs {
+ if config == nil {
+ continue
+ }
+ addresses := result[config.Name]
+ if addresses == nil {
+ addresses = make(map[string]struct{})
+ result[config.Name] = addresses
+ }
+ for _, endpoint := range config.Endpoints {
+ if endpoint != nil {
+ addresses[endpoint.Address.GetAddress()] =
struct{}{}
+ }
+ }
+ }
+ return result
+}
+
+func endpointStateChanges(old, new map[string]map[string]struct{}, version
uint64) []endpointStateChange {
+ changes := make([]endpointStateChange, 0)
+ for clusterName, addresses := range old {
+ for address := range addresses {
+ if _, exists := new[clusterName][address]; !exists {
+ changes = append(changes,
endpointStateChange{clusterName: clusterName, address: address, version:
version})
+ }
+ }
+ }
+ for clusterName, addresses := range new {
+ for address := range addresses {
+ if _, exists := old[clusterName][address]; !exists {
+ changes = append(changes,
endpointStateChange{clusterName: clusterName, address: address, present: true,
version: version})
+ }
+ }
+ }
+ return changes
+}
+
+func removedEndpointAddresses(oldConfigs, newConfigs []*model.ClusterConfig)
[]endpointRemoval {
+ newAddresses := make(map[string]map[string]struct{})
+ for _, config := range newConfigs {
+ if config == nil {
+ continue
+ }
+ addresses := newAddresses[config.Name]
+ if addresses == nil {
+ addresses = make(map[string]struct{})
+ newAddresses[config.Name] = addresses
+ }
+ for _, endpoint := range config.Endpoints {
+ if endpoint != nil {
+ addresses[endpoint.Address.GetAddress()] =
struct{}{}
+ }
+ }
+ }
+
+ seen := make(map[string]struct{})
+ var removals []endpointRemoval
+ for _, config := range oldConfigs {
+ if config == nil {
+ continue
+ }
+ for _, endpoint := range config.Endpoints {
+ if endpoint == nil {
+ continue
+ }
+ address := endpoint.Address.GetAddress()
+ key := config.Name + "\x00" + address
+ if _, exists := newAddresses[config.Name][address];
exists {
+ continue
+ }
+ if _, exists := seen[key]; exists {
+ continue
+ }
+ seen[key] = struct{}{}
+ removals = append(removals,
endpointRemoval{clusterName: config.Name, address: address})
+ }
+ }
+ return removals
}
func (cm *ClusterManager) CloneStore() (*ClusterStore, error) {
@@ -177,25 +384,29 @@ func (cm *ClusterManager) NewStore(version int32)
*ClusterStore {
// CompareAndSetStore swaps the store only when versions match.
// Version mismatch must leave both stores and runtime clusters untouched.
func (cm *ClusterManager) CompareAndSetStore(store *ClusterStore) bool {
- swapped, replacedClusters := cm.compareAndSetStore(store)
+ swapped, replacedClusters, changes := cm.compareAndSetStore(store)
if !swapped {
return false
}
// Stop old runtime after publishing the swap; Stop may touch
timers/goroutines.
stopClusters(replacedClusters)
+ cm.notifyEndpointChanges(changes)
return true
}
-func (cm *ClusterManager) compareAndSetStore(store *ClusterStore) (bool,
[]*cluster.Cluster) {
+func (cm *ClusterManager) compareAndSetStore(store *ClusterStore) (bool,
[]*cluster.Cluster, []endpointStateChange) {
cm.rw.Lock()
defer cm.rw.Unlock()
if store.Version != cm.store.Version {
- return false, nil
+ return false, nil, nil
}
currentStore := cm.store
+ version := cm.nextEndpointEventIDLocked()
+ old := endpointAddressSnapshot(currentStore.Config)
+ new := endpointAddressSnapshot(store.Config)
var replacedClusters []*cluster.Cluster
if store == currentStore {
replacedClusters = store.ensureRuntimeClusters()
@@ -207,7 +418,7 @@ func (cm *ClusterManager) compareAndSetStore(store
*ClusterStore) (bool, []*clus
if store != currentStore {
replacedClusters = append(replacedClusters,
currentStore.runtimeClustersNotIn(store)...)
}
- return true, replacedClusters
+ return true, replacedClusters, endpointStateChanges(old, new, version)
}
// PickEndpoint picks an endpoint from the cluster by its name and load
balancing policy.
@@ -260,6 +471,25 @@ func (cm *ClusterManager) GetAnyEndpointByID(clusterName,
endpointID string) *mo
return runtimeCluster.EndpointSnapshot().EndpointByID(endpointID)
}
+// HasEndpointAddress reports whether the current runtime snapshot still
+// contains the given endpoint address. Connection managers use this as an
+// authoritative check after bounded lifecycle tombstones have been evicted.
+func (cm *ClusterManager) HasEndpointAddress(clusterName, address string) bool
{
+ cm.rw.RLock()
+ defer cm.rw.RUnlock()
+
+ runtimeCluster := cm.getRuntimeCluster(clusterName)
+ if runtimeCluster == nil {
+ return false
+ }
+ for _, endpoint := range
runtimeCluster.EndpointSnapshot().AllEndpoints() {
+ if endpoint != nil && endpoint.Address.GetAddress() == address {
+ return true
+ }
+ }
+ return false
+}
+
// GetHealthyEndpointByID returns the runtime endpoint by ID only when it is
// healthy in the current runtime snapshot.
func (cm *ClusterManager) GetHealthyEndpointByID(clusterName, endpointID
string) *model.Endpoint {
@@ -307,7 +537,8 @@ func (cm *ClusterManager) pickOneEndpoint(runtimeCluster
*cluster.Cluster, polic
func (cm *ClusterManager) RemoveCluster(namesToDel []string) {
cm.rw.Lock()
- defer cm.rw.Unlock()
+ old := endpointAddressSnapshot(cm.store.Config)
+ version := cm.nextEndpointEventIDLocked()
for i, c := range cm.store.Config {
if c == nil {
@@ -331,6 +562,9 @@ func (cm *ClusterManager) RemoveCluster(namesToDel
[]string) {
cm.store.Config = append(cm.store.Config[:i],
cm.store.Config[i+1:]...)
}
cm.store.IncreaseVersion()
+ new := endpointAddressSnapshot(cm.store.Config)
+ cm.rw.Unlock()
+ cm.notifyEndpointChanges(endpointStateChanges(old, new, version))
}
func (cm *ClusterManager) HasCluster(clusterName string) bool {
@@ -613,10 +847,10 @@ func (s *ClusterStore) UpdateCluster(new
*model.ClusterConfig) {
logger.Warnf("not found modified cluster %s", new.Name)
}
-func (s *ClusterStore) SetEndpoint(clusterName string, endpoint
*model.Endpoint) {
+func (s *ClusterStore) SetEndpoint(clusterName string, endpoint
*model.Endpoint) string {
endpoint = model.CloneEndpoint(endpoint)
if endpoint == nil {
- return
+ return ""
}
clusterConfig := s.findClusterConfig(clusterName)
@@ -642,9 +876,13 @@ func (s *ClusterStore) SetEndpoint(clusterName string,
endpoint *model.Endpoint)
// already correct. Returning here keeps re-registration
idempotent —
// the LLM/Nacos path can replay the same instance event without
// growing the endpoint slice.
- return
+ return ""
case setEndpointReplace:
+ oldAddress :=
clusterConfig.Endpoints[outcome.replaceIdx].Address.GetAddress()
s.replaceEndpointAt(clusterConfig, runtimeCluster,
outcome.replaceIdx, endpoint)
+ if oldAddress != endpoint.Address.GetAddress() &&
!clusterHasEndpointAddress(clusterConfig, oldAddress) {
+ return oldAddress
+ }
case setEndpointAppend:
clusterConfig.Endpoints = append(clusterConfig.Endpoints,
endpoint)
s.prepareOwnedClusterConfig(clusterConfig)
@@ -652,6 +890,16 @@ func (s *ClusterStore) SetEndpoint(clusterName string,
endpoint *model.Endpoint)
runtimeCluster.RefreshEndpoints()
runtimeCluster.AddEndpoint(endpoint)
}
+ return ""
+}
+
+func clusterHasEndpointAddress(config *model.ClusterConfig, address string)
bool {
+ for _, endpoint := range config.Endpoints {
+ if endpoint != nil && endpoint.Address.GetAddress() == address {
+ return true
+ }
+ }
+ return false
}
// replaceEndpointAt overwrites the cluster slot at idx with the incoming
diff --git a/pkg/server/cluster_manager_test.go
b/pkg/server/cluster_manager_test.go
index 264790350..08412f1d3 100644
--- a/pkg/server/cluster_manager_test.go
+++ b/pkg/server/cluster_manager_test.go
@@ -800,6 +800,23 @@ func
TestClusterManager_SetEndpointExplicitSameIDDifferentAddressOverwritesInPla
assert.Equal(t, 21101, endpoints[0].Address.Port)
}
+func TestClusterManager_SetEndpointNotifiesAddressReplacement(t *testing.T) {
+ cm := testClusterManager(testCluster("endpoint-removal",
model.LoadBalancerRoundRobin, []*model.Endpoint{
+ testEndpoint("foo", "127.0.0.1", 21120),
+ }))
+ defer stopStoreRuntimes(cm.store)
+
+ var removed []string
+ removeHandler := cm.AddEndpointRemovalHandler(func(clusterName, address
string) {
+ removed = append(removed, clusterName+"\x00"+address)
+ })
+ defer removeHandler()
+
+ cm.SetEndpoint("endpoint-removal", testEndpoint("foo", "127.0.0.2",
21121))
+
+ assert.Equal(t, []string{"endpoint-removal\x00127.0.0.1:21120"},
removed)
+}
+
// TestClusterManager_SetEndpointExplicitSameIDSameContentIsIdempotent locks
// the other side of the dedup contract: when two calls share the same
// explicit ID AND the same routing-relevant content, the second call is a
diff --git a/pkg/server/pixiu_start.go b/pkg/server/pixiu_start.go
index 38b882e57..dd4837b8f 100644
--- a/pkg/server/pixiu_start.go
+++ b/pkg/server/pixiu_start.go
@@ -145,6 +145,9 @@ func GetServer() *Server {
}
func GetClusterManager() *ClusterManager {
+ if server == nil {
+ return nil
+ }
return server.GetClusterManager()
}
diff --git a/pkg/tracing/driver.go b/pkg/tracing/driver.go
index 27002159f..36d56129c 100644
--- a/pkg/tracing/driver.go
+++ b/pkg/tracing/driver.go
@@ -33,6 +33,7 @@ import (
)
import (
+ "github.com/apache/dubbo-go-pixiu/pkg/client/dubbo"
"github.com/apache/dubbo-go-pixiu/pkg/logger"
"github.com/apache/dubbo-go-pixiu/pkg/model"
"github.com/apache/dubbo-go-pixiu/pkg/tracing/jaeger"
@@ -84,6 +85,7 @@ func InitDriver(bs *model.Bootstrap) *TraceDriver {
}
provider := newTraceProvider(exp, config)
otel.SetTracerProvider(provider)
+ dubbo.SetTracingEnabled(true)
return &TraceDriver{TracerProvider: provider}
}
diff --git a/tools/benchmark/README.md b/tools/benchmark/README.md
index 2a92a81b4..6d4706f19 100644
--- a/tools/benchmark/README.md
+++ b/tools/benchmark/README.md
@@ -1,6 +1,6 @@
# Benchmark Results
-Test environment: macOS, Apple Silicon, N=500 samples per method
+Test environment: macOS, Apple Silicon, N=500 samples per method. The three
suites were run in parallel with `go test -count=1 ./...`.
## gRPC Protocol
@@ -8,19 +8,19 @@ Test environment: macOS, Apple Silicon, N=500 samples per
method
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 100µs | 200µs | 400µs | 1.1ms | 8.2ms |
+| GetUser | 100µs | 200µs | 200µs | 400µs | 2.8ms |
| GetUsers | 100µs | 200µs | 200µs | 100µs | 400µs |
-| GetUserByName | 100µs | 200µs | 200µs | 100µs | 600µs |
-| SayHello | 100µs | 200µs | 200µs | 100µs | 400µs |
+| GetUserByName | 100µs | 200µs | 200µs | 100µs | 400µs |
+| SayHello | 100µs | 200µs | 200µs | 0s | 500µs |
### gRPC via Pixiu
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 600µs | 1.4ms | 1.8ms | 1.8ms | 12.7ms |
-| GetUsers | 500µs | 1.4ms | 1.7ms | 1.2ms | 14.6ms |
-| GetUserByName | 500µs | 1.3ms | 2ms | 2ms | 17.1ms |
-| SayHello | 600µs | 1.3ms | 1.8ms | 2ms | 32.8ms |
+| GetUser | 200µs | 600µs | 800µs | 1.1ms | 8.3ms |
+| GetUsers | 200µs | 600µs | 700µs | 400µs | 3.6ms |
+| GetUserByName | 200µs | 400µs | 500µs | 300µs | 2ms |
+| SayHello | 200µs | 500µs | 600µs | 300µs | 1.8ms |
## Triple Protocol
@@ -28,19 +28,19 @@ Test environment: macOS, Apple Silicon, N=500 samples per
method
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 200µs | 1.8ms | 2.5ms | 2.4ms | 14.5ms |
-| GetUsers | 200µs | 1.5ms | 2.0ms | 2.0ms | 12.0ms |
-| GetUserByName | 200µs | 1.2ms | 1.8ms | 1.8ms | 10.5ms |
-| SayHello | 200µs | 900µs | 1.4ms | 1.3ms | 12.5ms |
+| GetUser | 200µs | 500µs | 600µs | 300µs | 2.5ms |
+| GetUsers | 200µs | 500µs | 500µs | 200µs | 1.7ms |
+| GetUserByName | 200µs | 600µs | 800µs | 700µs | 7ms |
+| SayHello | 200µs | 500µs | 500µs | 200µs | 1.2ms |
### Triple via Pixiu
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 1ms | 2.1ms | 2.8ms | 3.1ms | 24.4ms |
-| GetUsers | 800µs | 2.0ms | 2.5ms | 2.5ms | 20.0ms |
-| GetUserByName | 700µs | 1.8ms | 2.3ms | 2.0ms | 18.0ms |
-| SayHello | 700µs | 2ms | 2.5ms | 1.4ms | 12ms |
+| GetUser | 200µs | 1ms | 1.5ms | 1.9ms | 10.7ms |
+| GetUsers | 300µs | 1.3ms | 1.4ms | 700µs | 4.1ms |
+| GetUserByName | 300µs | 900µs | 1.2ms | 1ms | 7ms |
+| SayHello | 300µs | 800µs | 900µs | 500µs | 7ms |
## Dubbo Protocol
@@ -48,31 +48,29 @@ Test environment: macOS, Apple Silicon, N=500 samples per
method
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 100µs | 200µs | 600µs | 2.8ms | 20.7ms |
-| GetGender | 0s | 100µs | 200µs | 100µs | 600µs |
-| GetUser0 | 100µs | 100µs | 200µs | 0s | 400µs |
-| GetUsers | 100µs | 200µs | 200µs | 100µs | 600µs |
-| GetUser2 | 0s | 200µs | 200µs | 100µs | 500µs |
-| GetErr | 300µs | 500µs | 500µs | 200µs | 1.4ms |
+| GetUser | 100µs | 200µs | 200µs | 100µs | 500µs |
+| GetGender | 100µs | 200µs | 200µs | 100µs | 500µs |
+| GetUser0 | 100µs | 200µs | 200µs | 100µs | 900µs |
+| GetUsers | 100µs | 200µs | 200µs | 100µs | 500µs |
### Dubbo via Pixiu
| Method | Min | Median | Mean | StdDev | Max |
|--------|-----|--------|------|--------|-----|
-| GetUser | 700µs | 2ms | 3ms | 5.9ms | 44.7ms |
-| GetGender | 300µs | 800µs | 1ms | 500µs | 3.1ms |
-| GetUser0 | 300µs | 700µs | 800µs | 500µs | 4.2ms |
-| GetUsers | 100µs | 500µs | 600µs | 400µs | 3.2ms |
+| GetUser | 200µs | 400µs | 600µs | 1.5ms | 14.2ms |
+| GetGender | 200µs | 300µs | 400µs | 500µs | 4.3ms |
+| GetUser0 | 200µs | 400µs | 400µs | 200µs | 1.6ms |
+| GetUsers | 200µs | 400µs | 400µs | 200µs | 1.7ms |
## Performance Summary
-Pixiu proxy adds approximately 0.3-1.1ms overhead compared to direct protocol
calls.
+Pixiu proxy adds approximately 0.2-0.4ms median overhead in this run. Results
vary with local load and scheduling. The summary compares the same method
(`GetUser` for Dubbo and gRPC, `SayHello` for Triple).
| Protocol | Direct (Median) | via Pixiu (Median) | Overhead |
|----------|-----------------|--------------------|-----------|
-| gRPC | ~200µs | ~1.3ms | ~1.1ms |
-| Triple | ~1.4ms | ~2ms | ~0.6ms |
-| Dubbo | ~200µs | ~700µs | ~0.5ms |
+| gRPC | ~200µs | ~600µs | ~0.4ms |
+| Triple | ~500µs | ~800µs | ~0.3ms |
+| Dubbo | ~200µs | ~400µs | ~0.2ms |
# How to Run
diff --git a/tools/benchmark/README_CN.md b/tools/benchmark/README_CN.md
index 53cfc1477..6c716d32b 100644
--- a/tools/benchmark/README_CN.md
+++ b/tools/benchmark/README_CN.md
@@ -1,70 +1,38 @@
# Benchmark 结果
-* dubbo:
+测试环境：macOS、Apple Silicon，每个方法 500 次采样。Dubbo、gRPC、Triple 三套测试并行运行。
-```
-dubbo protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- =====================================================================
- GetUser [duration] | 500 | 100µs | 200µs | 600µs | 2.8ms | 20.7ms
- ---------------------------------------------------------------------
- GetGender [duration] | 500 | 0s | 100µs | 200µs | 100µs | 600µs
- ---------------------------------------------------------------------
- GetUser0 [duration] | 500 | 100µs | 100µs | 200µs | 0s | 400µs
- ---------------------------------------------------------------------
- GetUsers [duration] | 500 | 100µs | 200µs | 200µs | 100µs | 600µs
- ---------------------------------------------------------------------
- GetUser2 [duration] | 500 | 0s | 200µs | 200µs | 100µs | 500µs
- ---------------------------------------------------------------------
- GetErr [duration] | 500 | 300µs | 500µs | 500µs | 200µs | 1.4ms
-
- pixiu to dubbo protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- ====================================================================
- GetUser [duration] | 500 | 200µs | 600µs | 1.8ms | 8.4ms | 61ms
- --------------------------------------------------------------------
- GetGender [duration] | 499 | 200µs | 500µs | 600µs | 300µs | 4.1ms
- --------------------------------------------------------------------
- GetUser0 [duration] | 497 | 200µs | 500µs | 600µs | 300µs | 2.5ms
- --------------------------------------------------------------------
- GetUsers [duration] | 495 | 200µs | 600µs | 700µs | 300µs | 2.9ms
-```
+## Dubbo
-* grpc:
+| 方法 | Min | Median | Mean | StdDev | Max |
+| --- | --- | --- | --- | --- | --- |
+| GetUser（经 Pixiu） | 200µs | **400µs** | 600µs | 1.5ms | 14.2ms |
+| GetGender（经 Pixiu） | 200µs | **300µs** | 400µs | 500µs | 4.3ms |
+| GetUser0（经 Pixiu） | 200µs | 400µs | 400µs | 200µs | 1.6ms |
+| GetUsers（经 Pixiu） | 200µs | 400µs | 400µs | 200µs | 1.7ms |
+| GetUser（直连） | 100µs | 200µs | 200µs | 100µs | 500µs |
-```
-grpc protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- ========================================================================
- GetUser [duration] | 500 | 100µs | 300µs | 400µs | 600µs | 5.2ms
- ------------------------------------------------------------------------
- GetUsers [duration] | 500 | 100µs | 300µs | 300µs | 200µs | 1.5ms
- ------------------------------------------------------------------------
- GetUserByName [duration] | 496 | 100µs | 200µs | 300µs | 100µs | 1ms
-
-pixiu to grpc protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- =========================================================================
- GetUser [duration] | 500 | 600µs | 2ms | 3.4ms | 4.3ms | 25.9ms
- -------------------------------------------------------------------------
- GetUsers [duration] | 500 | 600µs | 1.3ms | 2.1ms | 3.3ms | 25.1ms
- -------------------------------------------------------------------------
- GetUserByName [duration] | 500 | 800µs | 1.9ms | 2.7ms | 3.8ms | 33.4ms
-```
+对比 issue #820 中的 Dubbo 数据：GetUser `600µs → 400µs`（约 1.5x），GetGender `500µs →
300µs`（约 1.7x）。
-* triple:
+## gRPC
-```
-triple protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- ===================================================================
- SayHello [duration] | 500 | 100µs | 200µs | 300µs | 300µs | 2.6ms
-
-pixiu to triple protocol performance test
- Name | N | Min | Median | Mean | StdDev | Max
- ====================================================================
- SayHello [duration] | 490 | 10.1ms | 12.7ms | 14ms | 4.1ms | 43.3ms
-```
+| 方法 | Min | Median | Mean | StdDev | Max |
+| --- | --- | --- | --- | --- | --- |
+| GetUser（经 Pixiu） | 200µs | **600µs** | 800µs | 1.1ms | 8.3ms |
+| GetUsers（经 Pixiu） | 200µs | **600µs** | 700µs | 400µs | 3.6ms |
+| GetUserByName（经 Pixiu） | 200µs | 400µs | 500µs | 300µs | 2ms |
+| GetUser（直连） | 100µs | 200µs | 200µs | 400µs | 2.8ms |
+
+## Triple
+
+| 方法 | Min | Median | Mean | StdDev | Max |
+| --- | --- | --- | --- | --- | --- |
+| GetUser（经 Pixiu） | 200µs | **500µs** | 600µs | 300µs | 2.5ms |
+| GetUsers（经 Pixiu） | 200µs | **500µs** | 500µs | 200µs | 1.7ms |
+| SayHello（经 Pixiu） | 300µs | **800µs** | 900µs | 500µs | 7ms |
+| GetUser（直连） | 200µs | 500µs | 600µs | 300µs | 2.5ms |
+
+相较 issue #820 中 Triple `SayHello` 的 `12.7ms`，最新结果为 `800µs`，约降低
**15.9x**。汇总数据按同一方法对比：Dubbo/gRPC 使用 `GetUser`，Triple 使用 `SayHello`。
# 运行方法
@@ -100,4 +68,4 @@ go test -v grpc_suite/grpc_test.go
# 运行 triple 测试
go test -v triple_suite/proto_suite/proto_test.go
-```
\ No newline at end of file
+```
diff --git a/tools/benchmark/protocol/grpc/pixiu/conf/config.yaml
b/tools/benchmark/protocol/grpc/pixiu/conf/config.yaml
index 62fe14036..bb8dcd258 100644
--- a/tools/benchmark/protocol/grpc/pixiu/conf/config.yaml
+++ b/tools/benchmark/protocol/grpc/pixiu/conf/config.yaml
@@ -24,7 +24,7 @@ static_resources:
address:
socket_address:
address: "0.0.0.0"
- port: 8881
+ port: 8882
filter_chains:
filters:
- name: dgp.filter.httpconnectionmanager
diff --git a/tools/benchmark/protocol/triple/go-server/cmd/server.go
b/tools/benchmark/protocol/triple/go-server/cmd/server.go
index b50bdfd11..681d3780d 100644
--- a/tools/benchmark/protocol/triple/go-server/cmd/server.go
+++ b/tools/benchmark/protocol/triple/go-server/cmd/server.go
@@ -42,7 +42,7 @@ func main() {
// Create server using new API
srv, err := server.NewServer(
server.WithServerProtocol(
- protocol.WithPort(20000),
+ protocol.WithPort(20010),
protocol.WithTriple(),
),
)
@@ -62,7 +62,7 @@ func main() {
}
}()
- fmt.Println("triple benchmark server is now running on :20000...")
+ fmt.Println("triple benchmark server is now running on :20010...")
initSignal()
}
diff --git a/tools/benchmark/protocol/triple/pixiu/conf/config.yaml
b/tools/benchmark/protocol/triple/pixiu/conf/config.yaml
index e2ee05ccb..1f49ab696 100644
--- a/tools/benchmark/protocol/triple/pixiu/conf/config.yaml
+++ b/tools/benchmark/protocol/triple/pixiu/conf/config.yaml
@@ -24,7 +24,7 @@ static_resources:
address:
socket_address:
address: "0.0.0.0"
- port: 8881
+ port: 8883
filter_chains:
filters:
- name: dgp.filter.httpconnectionmanager
@@ -52,7 +52,7 @@ static_resources:
endpoints:
- socket_address:
address: 127.0.0.1
- port: 20000
+ port: 20010
protocol_type: "TRIPLE"
shutdown_config:
timeout: "60s"
diff --git a/tools/benchmark/test/dubbo_suite/dubbo_test.go
b/tools/benchmark/test/dubbo_suite/dubbo_test.go
index 26caebb5a..4f68bfccd 100644
--- a/tools/benchmark/test/dubbo_suite/dubbo_test.go
+++ b/tools/benchmark/test/dubbo_suite/dubbo_test.go
@@ -70,7 +70,7 @@ var _ = Describe("Dubbo protocol performance test", Ordered,
func() {
dubboServerSession = prepareDubboServer()
time.Sleep(5 * time.Second)
- pixiuSession = test.PreparePixiu("../../dist/pixiu",
test.CurPath+"/../../protocol/dubbo/pixiu/conf/config.yaml")
+ pixiuSession = test.PreparePixiu("../../dist/pixiu",
test.CurPath+"/../../protocol/dubbo/pixiu/conf/config.yaml", 18380)
time.Sleep(5 * time.Second)
})
diff --git a/tools/benchmark/test/grpc_suite/grpc_test.go
b/tools/benchmark/test/grpc_suite/grpc_test.go
index 768199c32..c82eec18f 100644
--- a/tools/benchmark/test/grpc_suite/grpc_test.go
+++ b/tools/benchmark/test/grpc_suite/grpc_test.go
@@ -65,7 +65,7 @@ var _ = Describe("gRPC protocol performance test", Ordered,
func() {
time.Sleep(1 * time.Second)
- pixiuSession = test.PreparePixiu("../../dist/pixiu",
test.CurPath+"/../../protocol/grpc/pixiu/conf/config.yaml")
+ pixiuSession = test.PreparePixiu("../../dist/pixiu",
test.CurPath+"/../../protocol/grpc/pixiu/conf/config.yaml", 18381)
time.Sleep(3 * time.Second)
@@ -133,7 +133,7 @@ var _ = Describe("gRPC protocol performance test", Ordered,
func() {
experiment := gmeasure.NewExperiment("pixiu to gRPC protocol
performance test")
AddReportEntry(experiment.Name, experiment)
- urlPrefix :=
"http://localhost:8881/api/v1/benchmark.BenchmarkService/";
+ urlPrefix :=
"http://localhost:8882/api/v1/benchmark.BenchmarkService/";
experiment.Sample(func(idx int) {
defer GinkgoRecover()
diff --git a/tools/benchmark/test/test_tool.go
b/tools/benchmark/test/test_tool.go
index 7a45806a2..51cd9e965 100644
--- a/tools/benchmark/test/test_tool.go
+++ b/tools/benchmark/test/test_tool.go
@@ -20,6 +20,7 @@ package test
import (
"io"
"os/exec"
+ "strconv"
"time"
)
@@ -38,8 +39,8 @@ var (
}
)
-func PreparePixiu(pixiu, path string) *gexec.Session {
- command := exec.Command(pixiu, "gateway", "start", "-c", path)
+func PreparePixiu(pixiu, path string, reloadPort int) *gexec.Session {
+ command := exec.Command(pixiu, "gateway", "start", "-c", path,
"--reload-port", strconv.Itoa(reloadPort))
session, err := gexec.Start(command, io.Discard, io.Discard)
//session, err := gexec.Start(command, os.Stdout, os.Stderr)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
diff --git a/tools/benchmark/test/triple_suite/proto_suite/proto_test.go
b/tools/benchmark/test/triple_suite/proto_suite/proto_test.go
index 3e748f220..1d8105d34 100644
--- a/tools/benchmark/test/triple_suite/proto_suite/proto_test.go
+++ b/tools/benchmark/test/triple_suite/proto_suite/proto_test.go
@@ -62,20 +62,20 @@ var _ = Describe("triple protocol performance test",
Ordered, func() {
test.CurPath, err = os.Getwd()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- waitForPortAvailable("20000", 10*time.Second)
- waitForPortAvailable("8881", 10*time.Second)
+ waitForPortAvailable("20010", 10*time.Second)
+ waitForPortAvailable("8883", 10*time.Second)
tripleServerSession = prepareTripleServer()
time.Sleep(8 * time.Second)
- pixiuSession = test.PreparePixiu("../../../dist/pixiu",
test.CurPath+"/../../../protocol/triple/pixiu/conf/config.yaml")
+ pixiuSession = test.PreparePixiu("../../../dist/pixiu",
test.CurPath+"/../../../protocol/triple/pixiu/conf/config.yaml", 18382)
time.Sleep(6 * time.Second)
})
It("pixiu to triple protocol performance test", func() {
defer GinkgoRecover()
- urlPrefix :=
"http://localhost:8881/dubbo.io/benchmark.BenchmarkService/%s";
+ urlPrefix :=
"http://localhost:8883/dubbo.io/benchmark.BenchmarkService/%s";
experiment := gmeasure.NewExperiment("pixiu to triple protocol
performance test")
AddReportEntry(experiment.Name, experiment)
@@ -219,7 +219,7 @@ func prepareTripleServer() *gexec.Session {
func prepareTripleClient() {
// Create client using new API
cli, err := client.NewClient(
- client.WithClientURL("tri://127.0.0.1:20000"),
+ client.WithClientURL("tri://127.0.0.1:20010"),
)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
