Tsukikage7 commented on code in PR #1017:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1017#discussion_r3810959455
##########
pkg/listener/http/http_listener.go:
##########
@@ -83,7 +83,18 @@ 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 := error(nil)
+ if ls.FilterChain != nil {
+ filterErr = ls.FilterChain.Close()
Review Comment:
Fixed in d54b2ad1. ServeHTTP now acquires a per-chain lease and releases the
listener lock before running filters. Refresh swaps to the new chain
immediately, then drains and closes the old chain after active requests release
their leases. Added a regression test covering refresh with an active request.
##########
pkg/filter/http/grpcproxy/connection_manager.go:
##########
@@ -0,0 +1,200 @@
+/*
+ * 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"
+ "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
+
+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
+}
+
+func newGRPCConnectionManager() *grpcConnectionManager {
+ return &grpcConnectionManager{
+ dial: dialGRPCConnection,
+ dialTimeout: defaultGRPCDialTimeout,
+ }
+}
+
+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 conn, ok := m.loadHealthy(key); ok {
+ return conn, nil
+ }
+
+ result := m.creates.DoChan(key, func() (any, error) {
+ if conn, ok := m.loadHealthy(key); ok {
+ 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
+ if !closed {
+ m.connections.Store(key, conn)
Review Comment:
Fixed in d54b2ad1. The bounded tombstone LRU is no longer the only source of
truth: grpcproxy now checks the current ClusterManager endpoint snapshot before
reusing or publishing a connection, including after a tombstone has been
evicted. Added a regression test for delete + 1025 unique endpoint churn +
delayed Get; it rejects without dialing.
##########
pkg/filter/http/grpcproxy/connection_manager.go:
##########
@@ -0,0 +1,423 @@
+/*
+ * 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
+}
+
+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) 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 conn, ok := m.loadHealthy(key); ok {
+ 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) {
+ 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 {
+ 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 {
+ _ = 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 && state !=
connectivity.TransientFailure
Review Comment:
Fixed in d54b2ad1. TransientFailure is now treated as recoverable and is no
longer evicted by loadHealthy or Invalidate; only Shutdown is considered
unusable here. Added a regression test that keeps the same ClientConn cached
while it is in TransientFailure.
##########
pkg/client/dubbo/dubbo.go:
##########
@@ -311,15 +364,14 @@ 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 {
+ // fast path: no attachments, no tracing -> reuse the context as-is
+ if !tracingEnabled.Load() && len(outbound) == 0 &&
ctx.Value(constant.AttachmentKey) == nil {
Review Comment:
Fixed in d54b2ad1. The fast path now checks for an external SpanContext
instead of relying only on Pixiu tracingEnabled, so an external remote span
with empty business attachments still reaches the global propagator. Added a
regression test that verifies traceparent propagation.
##########
pkg/filter/http/grpcproxy/descriptor.go:
##########
@@ -117,12 +128,79 @@ 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) removeConnection(cc *grpc.ClientConn) {
+ if cc == nil {
+ return
+ }
+ dr.methodMu.Lock()
+ delete(dr.methodDescs, cc)
+ dr.methodMu.Unlock()
+}
+
+func (dr *Descriptor) Close() {
+ dr.methodMu.Lock()
+ dr.methodDescs = nil
+ dr.methodMu.Unlock()
+}
+
+func (dr *Descriptor) getMethodDescriptor(source DescriptorSource, cc
*grpc.ClientConn, service, method string) (*desc.MethodDescriptor, error) {
+ key := service + "\x00" + method
+ dr.methodMu.RLock()
+ if methods := dr.methodDescs[cc]; methods != nil {
+ if descriptor, ok := methods[key]; ok {
+ dr.methodMu.RUnlock()
+ return descriptor, nil
+ }
+ }
+ dr.methodMu.RUnlock()
+
+ dr.methodMu.Lock()
+ defer dr.methodMu.Unlock()
+ if methods := dr.methodDescs[cc]; methods != nil {
+ if descriptor, ok := methods[key]; ok {
+ return descriptor, nil
+ }
+ }
+
+ dscp, err := source.FindSymbol(service)
Review Comment:
Fixed in d54b2ad1. I removed the cross-request singleflight around
descriptor cache misses because DescriptorSource is bound to the request
context. Each lookup now keeps its own deadline and cancellation, while
successful method descriptors remain cached per connection. Added a regression
test showing that a slow first source does not block the next request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
