Merge pull request #1275 from weli-l/dev/auth_ip_optimize

add channel for Rbac when Kmesh restart

Author: kmesh-bot

pkg/controller/workload/workload_controller.go

@@ -19,6 +19,7 @@ package workload
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	discoveryv3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
 
@@ -66,7 +67,21 @@ func NewController(bpfWorkload *bpfwl.BpfWorkload, enableMonitoring, enablePerfM
 }
 
 func (c *Controller) Run(ctx context.Context) {
-	go c.Rbac.Run(ctx, c.bpfWorkloadObj.SockOps.KmAuthReq, c.bpfWorkloadObj.XdpAuth.KmAuthRes)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		<-c.Processor.addressDone
+		wg.Done()
+	}()
+	go func() {
+		<-c.Processor.authzDone
+		wg.Done()
+	}()
+	go func() {
+		wg.Wait()
+		c.Rbac.Run(ctx, c.bpfWorkloadObj.SockOps.KmAuthReq, c.bpfWorkloadObj.XdpAuth.KmAuthRes)
+	}()
+
 	go c.MetricController.Run(ctx, c.bpfWorkloadObj.SockConn.KmTcpProbe)
 	if c.MapMetricController != nil {
 		go c.MapMetricController.Run(ctx)

pkg/controller/workload/workload_controller_test.go

@@ -199,10 +199,12 @@ func TestWorkloadStreamCreateAndSend(t *testing.T) {
 	}
 }
 
-func TestAdsStream_AdsStreamProcess(t *testing.T) {
+func TestWorkloadStream_WorkloadstreamProcess(t *testing.T) {
 	workloadStream := Controller{
 		Processor: &Processor{
-			ack: &discoveryv3.DeltaDiscoveryRequest{},
+			ack:         &discoveryv3.DeltaDiscoveryRequest{},
+			addressDone: make(chan struct{}),
+			authzDone:   make(chan struct{}),
 		},
 	}
 
@@ -217,6 +219,12 @@ func TestAdsStream_AdsStreamProcess(t *testing.T) {
 
 	patches1 := gomonkey.NewPatches()
 	patches2 := gomonkey.NewPatches()
+	consumeSignals := func() {
+		go func() {
+			<-workloadStream.Processor.addressDone
+			<-workloadStream.Processor.authzDone
+		}()
+	}
 	tests := []struct {
 		name       string
 		beforeFunc func()
@@ -302,6 +310,7 @@ func TestAdsStream_AdsStreamProcess(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			tt.beforeFunc()
+			consumeSignals()
 			err := workloadStream.HandleWorkloadStream()
 
 			if (err != nil) != tt.wantErr {

pkg/controller/workload/workload_processor.go

@@ -66,6 +66,12 @@ type Processor struct {
 
 	once      sync.Once
 	authzOnce sync.Once
+
+	// used to notify Rbac the address/authz type response is done when Kmesh restart
+	addressDone     chan struct{}
+	authzDone       chan struct{}
+	addressRespOnce sync.Once
+	authzRespOnce   sync.Once
 }
 
 func NewProcessor(workloadMap bpf2go.KmeshCgroupSockWorkloadMaps) *Processor {
@@ -80,6 +86,8 @@ func NewProcessor(workloadMap bpf2go.KmeshCgroupSockWorkloadMaps) *Processor {
 		EndpointCache: cache.NewEndpointCache(),
 		WaypointCache: cache.NewWaypointCache(serviceCache),
 		locality:      bpf.NewLocalityCache(),
+		addressDone:   make(chan struct{}, 1),
+		authzDone:     make(chan struct{}, 1),
 	}
 }
 
@@ -119,8 +127,14 @@ func (p *Processor) processWorkloadResponse(rsp *service_discovery_v3.DeltaDisco
 	switch rsp.GetTypeUrl() {
 	case AddressType:
 		err = p.handleAddressTypeResponse(rsp)
+		p.addressRespOnce.Do(func() {
+			p.addressDone <- struct{}{}
+		})
 	case AuthorizationType:
 		err = p.handleAuthorizationTypeResponse(rsp, rbac)
+		p.authzRespOnce.Do(func() {
+			p.authzDone <- struct{}{}
+		})
 	default:
 		err = fmt.Errorf("unsupported type url %s", rsp.GetTypeUrl())
 	}