AlexStocks commented on code in PR #1017:
URL: https://github.com/apache/dubbo-go-pixiu/pull/1017#discussion_r3740276210
##########
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:
[P1] 不要在全局 descriptor 锁内执行网络 reflection
`methodMu` 是整个 Descriptor 共享的写锁，这里调用 `source.FindSymbol` 时仍持有该锁；REMOTE/AUTO
下它会发起网络 reflection。于是一个慢后端的首次 lookup 会阻塞所有其他连接和服务的 cache miss，直到该请求超时。我在当前 Head
用两个独立 connection 做负向探针：A 的 reflection 阻塞后，B 的立即失败 lookup 也被阻塞超过
100ms。建议把网络查询移到锁外，只在 double-check/store 时短暂加锁；若要合并并发 miss，按
connection/service/method 使用 singleflight，并补充‘慢 A 不阻塞 B’的回归测试。
##########
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:
[P1] 被注册中心删除的健康 endpoint 会永久留在连接缓存
这里把连接按 `cluster + endpoint` 永久存入 map，但现有删除只发生在再次访问同一 key 并发现 transport
不健康、Invalidate 遇到不健康连接，或关闭整个 manager。注册中心 `DeleteEndpoint` 只更新 ClusterManager；旧
endpoint 被移除后不会再被选中，而健康的旧 ClientConn 也不会触发上述任何删除路径，连带按 ClientConn 保存的
descriptor cache 一起永久保留。实例滚动或地址频繁变化时，连接、goroutine 和内存会随历史 endpoint 单调增长。建议把
endpoint 删除/cluster reconcile 接入 manager，或实现有界 idle TTL/LRU，并补充 A 建连后删除/切换到 B 时
A 被 Close、descriptor 同步清理以及多轮 churn 有界的测试。
--
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]
