Refactor grpc pkg to support general grpc service.

Signed-off-by: xuezhaojun <zxue@redhat.com>

Author: xuezhaojun

grpc-interceptor-flow.md

@@ -0,0 +1,185 @@
+# gRPC 拦截器和认证授权流程
+
+本文档描述了 gRPC 服务器中拦截器链的工作流程，特别是认证（Authentication）和授权（Authorization）的处理过程。
+
+## 系统架构概览
+
+```mermaid
+graph TB
+    Client[gRPC Client] --> Server[gRPC Server]
+    Server --> Chain[Interceptor Chain]
+    Chain --> AuthN[Authentication Interceptor]
+    Chain --> AuthZ[Authorization Interceptor]
+    Chain --> Handler[Business Handler]
+    
+    AuthN --> AuthN1[Authenticator 1<br/>Token Auth]
+    AuthN --> AuthN2[Authenticator 2<br/>mTLS Auth]
+    AuthN --> AuthN3[Authenticator N<br/>Custom Auth]
+    
+    AuthZ --> AuthZ1[Authorizer 1<br/>SAR Check]
+    AuthZ --> AuthZ2[Authorizer 2<br/>RBAC Check]
+    AuthZ --> AuthZ3[Authorizer N<br/>Custom Check]
+```
+
+## 详细请求处理流程
+
+```mermaid
+sequenceDiagram
+    participant C as gRPC Client
+    participant S as gRPC Server
+    participant IC as Interceptor Chain
+    participant AI as Auth Interceptor
+    participant A1 as TokenAuthenticator
+    participant A2 as mTLSAuthenticator
+    participant AZ as Authz Interceptor
+    participant SAR as SARAuthorizer
+    participant H as Business Handler
+    participant K8S as Kubernetes API
+
+    C->>S: gRPC Request
+    S->>IC: Process Request
+    IC->>AI: Authentication Phase
+    
+    Note over AI: OR Logic - Any authenticator success
+    AI->>A1: Try Token Authentication
+    A1->>K8S: TokenReview API Call
+    K8S-->>A1: Authentication Failed
+    A1-->>AI: Failed
+    
+    AI->>A2: Try mTLS Authentication
+    A2->>A2: Verify Client Certificate
+    A2-->>AI: Success + User Context
+    
+    Note over AI: First success, skip remaining authenticators
+    AI->>AZ: Pass with User Context
+    
+    Note over AZ: AND Logic - All authorizers must pass
+    AZ->>SAR: Check Permissions
+    SAR->>K8S: SubjectAccessReview API Call
+    K8S-->>SAR: Permission Granted
+    SAR-->>AZ: Authorized
+    
+    Note over AZ: All authorizers passed
+    AZ->>H: Execute Business Logic
+    H-->>AZ: Response
+    AZ-->>AI: Response
+    AI-->>IC: Response
+    IC-->>S: Response
+    S-->>C: gRPC Response
+```
+
+## 拦截器链配置
+
+```mermaid
+graph LR
+    subgraph "Server Configuration"
+        SO[Server Options]
+        SO --> CUI[ChainUnaryInterceptor]
+        SO --> CSI[ChainStreamInterceptor]
+    end
+    
+    subgraph "Unary Interceptor Chain"
+        CUI --> AI1[newAuthnUnaryInterceptor]
+        AI1 --> AZ1[newAuthzUnaryInterceptor]
+        AZ1 --> BH1[Business Handler]
+    end
+    
+    subgraph "Stream Interceptor Chain"
+        CSI --> AI2[newAuthnStreamInterceptor]
+        AI2 --> AZ2[newAuthzStreamInterceptor]
+        AZ2 --> BH2[Business Handler]
+    end
+    
+    subgraph "Authenticators"
+        AI1 --> TA[TokenAuthenticator]
+        AI1 --> MA[mTLSAuthenticator]
+        AI2 --> TA
+        AI2 --> MA
+    end
+    
+    subgraph "Authorizers"
+        AZ1 --> SARA[SARAuthorizer]
+        AZ2 --> SARA
+    end
+```
+
+## 认证器逻辑（OR 关系）
+
+```mermaid
+flowchart TD
+    Start([Request Arrives]) --> AuthLoop{For each Authenticator}
+    AuthLoop --> TryAuth[Try Authentication]
+    TryAuth --> AuthSuccess{Success?}
+    AuthSuccess -->|Yes| SetContext[Set User Context]
+    SetContext --> CallHandler[Call Next Handler]
+    AuthSuccess -->|No| NextAuth{More Authenticators?}
+    NextAuth -->|Yes| AuthLoop
+    NextAuth -->|No| AuthFailed[Return Auth Error]
+    CallHandler --> End([Continue to Authorization])
+    AuthFailed --> ErrorEnd([Return Error])
+    
+    style SetContext fill:#90EE90
+    style AuthFailed fill:#FFB6C1
+    style CallHandler fill:#87CEEB
+```
+
+## 授权器逻辑（AND 关系）
+
+```mermaid
+flowchart TD
+    Start([From Authentication]) --> AuthzLoop{For each Authorizer}
+    AuthzLoop --> TryAuthz[Try Authorization]
+    TryAuthz --> AuthzSuccess{Success?}
+    AuthzSuccess -->|No| AuthzFailed[Return Authz Error]
+    AuthzSuccess -->|Yes| NextAuthz{More Authorizers?}
+    NextAuthz -->|Yes| AuthzLoop
+    NextAuthz -->|No| AllPassed[All Authorizers Passed]
+    AllPassed --> CallHandler[Call Business Handler]
+    CallHandler --> End([Return Response])
+    AuthzFailed --> ErrorEnd([Return Error])
+    
+    style AllPassed fill:#90EE90
+    style AuthzFailed fill:#FFB6C1
+    style CallHandler fill:#87CEEB
+```
+
+## 性能影响分析
+
+```mermaid
+graph TB
+    subgraph "Performance Impact"
+        PI[Performance Impact]
+        PI --> FO[Framework Overhead<br/>~1μs]
+        PI --> AO[Auth Overhead<br/>1-10ms]
+        PI --> NO[Network Overhead<br/>K8s API Calls]
+    end
+    
+    subgraph "Optimization Strategies"
+        OS[Optimization]
+        OS --> Cache[Token/SAR Caching]
+        OS --> Pool[Connection Pooling]
+        OS --> Batch[Batch Processing]
+        OS --> Async[Async Prewarming]
+    end
+    
+    FO -.->|Minimal| Cache
+    AO -.->|Significant| Cache
+    NO -.->|Major| Pool
+```
+
+## 关键特性总结
+
+### 认证（Authentication）
+- **逻辑关系**: OR（任意一个成功即可）
+- **早期退出**: 第一个成功的认证器会立即返回
+- **上下文传递**: 成功的认证器会在 context 中设置用户信息
+
+### 授权（Authorization）
+- **逻辑关系**: AND（所有授权器都必须通过）
+- **严格检查**: 任意一个授权器失败都会导致请求被拒绝
+- **依赖认证**: 需要认证阶段提供的用户上下文
+
+### 性能考虑
+- **拦截器框架开销**: 微秒级，可忽略
+- **主要瓶颈**: Kubernetes API 调用（TokenReview, SubjectAccessReview）
+- **优化方案**: 缓存、连接池、批处理等策略

pkg/cloudevents/server/grpc/authz/kube/sar.go

@@ -16,8 +16,8 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/lease"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authz"
+	"open-cluster-management.io/sdk-go/pkg/grpc/authn"
 )
 
 type SARAuthorizer struct {

pkg/cloudevents/server/grpc/authz/kube/sar_test.go

@@ -17,7 +17,7 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/lease"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/clients/work/payload"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
+	"open-cluster-management.io/sdk-go/pkg/grpc/authn"
 )
 
 func TestSARAuthorize(t *testing.T) {

pkg/cloudevents/server/grpc/server.go pkg/cloudevents/server/grpc/broker.gopkg/cloudevents/server/grpc/server.go renamed to pkg/cloudevents/server/grpc/broker.go

@@ -4,12 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net"
 	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
-	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/binding"
@@ -44,23 +42,24 @@ var _ server.AgentEventServer = &GRPCBroker{}
 // It broadcasts resource spec to agents and listens for resource status updates from them.
 type GRPCBroker struct {
 	pbv1.UnimplementedCloudEventServiceServer
-	grpcServer  *grpc.Server
 	services    map[types.CloudEventsDataType]server.Service
 	subscribers map[string]*subscriber // registered subscribers
 	mu          sync.RWMutex
 }
 
 // NewGRPCBroker creates a new gRPC broker with the given gRPC server.
-func NewGRPCBroker(srv *grpc.Server) server.AgentEventServer {
+func NewGRPCBroker() *GRPCBroker {
 	broker := &GRPCBroker{
-		grpcServer:  srv,
 		subscribers: make(map[string]*subscriber),
 		services:    make(map[types.CloudEventsDataType]server.Service),
 	}
-	pbv1.RegisterCloudEventServiceServer(broker.grpcServer, broker)
 	return broker
 }
 
+func (bkr *GRPCBroker) Register(grpcServer *grpc.Server) {
+	pbv1.RegisterCloudEventServiceServer(grpcServer, bkr)
+}
+
 func (bkr *GRPCBroker) RegisterService(t types.CloudEventsDataType, service server.Service) {
 	bkr.services[t] = service
 	service.RegisterHandler(bkr)
@@ -78,26 +77,6 @@ func (bkr *GRPCBroker) Subscribers() sets.Set[string] {
 	return subscribers
 }
 
-// Start starts the gRPC broker at the given address
-func (bkr *GRPCBroker) Start(ctx context.Context, addr string) {
-	logger := klog.FromContext(ctx)
-	logger.Info("Starting gRPC broker at addr", "addr", addr)
-	lis, err := net.Listen("tcp", addr)
-	if err != nil {
-		utilruntime.Must(fmt.Errorf("failed to listen: %v", err))
-	}
-	go func() {
-		if err := bkr.grpcServer.Serve(lis); err != nil {
-			utilruntime.Must(fmt.Errorf("failed to serve gRPC broker: %v", err))
-		}
-	}()
-
-	// wait until context is canceled
-	<-ctx.Done()
-	klog.Infof("Shutting down gRPC broker")
-	bkr.grpcServer.GracefulStop()
-}
-
 // Publish in stub implementation for agent publish resource status.
 func (bkr *GRPCBroker) Publish(ctx context.Context, pubReq *pbv1.PublishRequest) (*emptypb.Empty, error) {
 	logger := klog.FromContext(ctx)

…g/cloudevents/server/grpc/server_test.go …g/cloudevents/server/grpc/broker_test.gopkg/cloudevents/server/grpc/server_test.go renamed to pkg/cloudevents/server/grpc/broker_test.go pkg/cloudevents/server/grpc/server_test.go renamed to pkg/cloudevents/server/grpc/broker_test.go

@@ -3,6 +3,7 @@ package grpc
 import (
 	"context"
 	"errors"
+	"net"
 	"testing"
 
 	cloudevents "github.com/cloudevents/sdk-go/v2"
@@ -59,14 +60,24 @@ func (s *testService) create(evt *cloudevents.Event) error {
 func TestServer(t *testing.T) {
 	grpcServerOptions := []grpc.ServerOption{}
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := NewGRPCBroker(grpcServer)
+	grpcEventServer := NewGRPCBroker()
+	grpcEventServer.Register(grpcServer)
 
 	svc := &testService{evts: make(map[string]*cloudevents.Event)}
 	grpcEventServer.RegisterService(dataType, svc)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go grpcEventServer.Start(ctx, ":8888")
+
+	lis, err := net.Listen("tcp", ":8888")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+	go func() {
+		if err := grpcServer.Serve(lis); err != nil {
+			t.Errorf("failed to serve: %v", err)
+		}
+	}()
 
 	grpcClientOptions := grpccli.NewGRPCOptions()
 	grpcClientOptions.Dialer = &grpccli.GRPCDialer{URL: "localhost:8888"}

pkg/cloudevents/server/grpc/options/server.go

@@ -18,8 +18,8 @@ import (
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/generic/types"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server"
 	grpcserver "open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
 	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authz"
+	"open-cluster-management.io/sdk-go/pkg/grpc/authn"
 )
 
 // PreStartHook is an interface to start hook before grpc server is started.
@@ -116,7 +116,8 @@ func (s *Server) Run(ctx context.Context) error {
 			newAuthzStreamInterceptor(s.authorizers...)))
 
 	grpcServer := grpc.NewServer(grpcServerOptions...)
-	grpcEventServer := grpcserver.NewGRPCBroker(grpcServer)
+	grpcEventServer := grpcserver.NewGRPCBroker()
+	grpcEventServer.Register(grpcServer)
 
 	for t, service := range s.services {
 		grpcEventServer.RegisterService(t, service)
@@ -127,7 +128,7 @@ func (s *Server) Run(ctx context.Context) error {
 		hook.Run(ctx)
 	}
 
-	go grpcEventServer.Start(ctx, ":"+s.options.ServerBindPort)
+	// go grpcEventServer.Start(ctx, ":"+s.options.ServerBindPort)
 	<-ctx.Done()
 	return nil
 }

pkg/cloudevents/server/grpc/options/server_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 
 	certutil "k8s.io/client-go/util/cert"
-	"open-cluster-management.io/sdk-go/pkg/cloudevents/server/grpc/authn"
+	"open-cluster-management.io/sdk-go/pkg/grpc/authn"
 )
 
 type testHook struct{}

pkg/cloudevents/server/interface.go

@@ -17,9 +17,6 @@ type AgentEventServer interface {
 	// RegisterService registers a backend service with a certain data type.
 	RegisterService(t types.CloudEventsDataType, service Service)
 
-	// Start initiates the EventServer to listen to agents.
-	Start(ctx context.Context, addr string)
-
 	// Subscribers returns all current subscribers who subscribe to this server.
 	Subscribers() sets.Set[string]
 }