support dump workload bpf map

pkg/controller/workload/bpfcache/auth_policy.go

@@ -20,25 +20,30 @@
 	"github.com/cilium/ebpf"
 )
 
-type WorkloadPolicy_key struct {
+type WorkloadPolicyKey struct {
 	WorklodId uint32 // workloadIp to uint32
 }
 
-type WorkloadPolicy_value struct {
+type WorkloadPolicyValue struct {
 	PolicyIds [4]uint32 // name length is [MAX_MEMBER_NUM_PER_POLICY]
 }
 
-func (c *Cache) WorkloadPolicyUpdate(key *WorkloadPolicy_key, value *WorkloadPolicy_value) error {
+func (c *Cache) WorkloadPolicyUpdate(key *WorkloadPolicyKey, value *WorkloadPolicyValue) error {
 	log.Debugf("workload policy update: [%#v], [%#v]", *key, *value)
 	return c.bpfMap.MapOfWlPolicy.Update(key, value, ebpf.UpdateAny)
 }
 
-func (c *Cache) WorkloadPolicyDelete(key *WorkloadPolicy_key) error {
+func (c *Cache) WorkloadPolicyDelete(key *WorkloadPolicyKey) error {
 	log.Debugf("workload policy delete: [%#v]", *key)
 	return c.bpfMap.MapOfWlPolicy.Delete(key)
 }
 
-func (c *Cache) WorkloadPolicyLookup(key *WorkloadPolicy_key, value *WorkloadPolicy_value) error {
+func (c *Cache) WorkloadPolicyLookup(key *WorkloadPolicyKey, value *WorkloadPolicyValue) error {
 	log.Debugf("workload policy lookup: [%#v]", *key)
 	return c.bpfMap.MapOfWlPolicy.Lookup(key, value)
 }
+
+func (c *Cache) WorkloadPolicyLookupAll() []WorkloadPolicyValue {
+	log.Debugf("WorkloadPolicyLookupAll")
+	return LookupAll[WorkloadPolicyKey, WorkloadPolicyValue](c.bpfMap.MapOfWlPolicy)
+}

pkg/controller/workload/bpfcache/backend.go

@@ -56,15 +56,10 @@ func (c *Cache) BackendLookup(key *BackendKey, value *BackendValue) error {
 // BackendCount returns the length of backend map
 // Note only used for testing
 func (c *Cache) BackendCount() int {
-	var (
-		key   = BackendKey{}
-		value = BackendValue{}
-	)
+	return len(c.BackendLookupAll())
+}
 
-	res := 0
-	iter := c.bpfMap.KmeshBackend.Iterate()
-	for iter.Next(&key, &value) {
-		res++
-	}
-	return res
+func (c *Cache) BackendLookupAll() []BackendValue {
+	log.Debugf("BackendLookupAll")
+	return LookupAll[BackendKey, BackendValue](c.bpfMap.KmeshBackend)
 }

pkg/controller/workload/bpfcache/common.go

@@ -0,0 +1,35 @@
+/*
+ * Copyright The Kmesh Authors.
+ *
+ * Licensed 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 bpfcache
+
+import (
+	"github.com/cilium/ebpf"
+)
+
+func LookupAll[K any, V any](bpfMap *ebpf.Map) []V {
+	var (
+		key   K
+		value V
+		ret   []V
+	)
+
+	iter := bpfMap.Iterate()
+	for iter.Next(&key, &value) {
+		ret = append(ret, value)
+	}
+	return ret
+}

pkg/controller/workload/bpfcache/endpoint.go

@@ -153,15 +153,10 @@ func (c *Cache) GetAllEndpointsForService(serviceId uint32) []EndpointValue {
 // EndpointCount returns the length of endpoint map
 // Note only used for testing
 func (c *Cache) EndpointCount() int {
-	var (
-		key   = EndpointKey{}
-		value = EndpointValue{}
-	)
+	return len(c.EndpointLookupAll())
+}
 
-	res := 0
-	iter := c.bpfMap.KmeshEndpoint.Iterate()
-	for iter.Next(&key, &value) {
-		res++
-	}
-	return res
+func (c *Cache) EndpointLookupAll() []EndpointValue {
+	log.Debugf("EndpointLookupAll")
+	return LookupAll[EndpointKey, EndpointValue](c.bpfMap.KmeshEndpoint)
 }

pkg/controller/workload/bpfcache/fake_map.go

@@ -75,8 +75,8 @@ func NewFakeWorkloadMap(t *testing.T) bpf2go.KmeshCgroupSockWorkloadMaps {
 	wlPolicyMap, err := ebpf.NewMap(&ebpf.MapSpec{
 		Name:       "workload_policy",
 		Type:       ebpf.Hash,
-		KeySize:    uint32(unsafe.Sizeof(WorkloadPolicy_key{})),
-		ValueSize:  uint32(unsafe.Sizeof(WorkloadPolicy_value{})),
+		KeySize:    uint32(unsafe.Sizeof(WorkloadPolicyKey{})),
+		ValueSize:  uint32(unsafe.Sizeof(WorkloadPolicyValue{})),
 		MaxEntries: 512,
 	})
 	if err != nil {

pkg/controller/workload/bpfcache/frontend.go

@@ -68,15 +68,10 @@ func (c *Cache) FrontendIterFindKey(upstreamId uint32) []FrontendKey {
 // FrontendCount returns the length of frontend map
 // Note only used for testing
 func (c *Cache) FrontendCount() int {
-	var (
-		key   = FrontendKey{}
-		value = FrontendValue{}
-	)
+	return len(c.FrontendLookupAll())
+}
 
-	res := 0
-	iter := c.bpfMap.KmeshFrontend.Iterate()
-	for iter.Next(&key, &value) {
-		res++
-	}
-	return res
+func (c *Cache) FrontendLookupAll() []FrontendValue {
+	log.Debugf("FrontendLookupAll")
+	return LookupAll[FrontendKey, FrontendValue](c.bpfMap.KmeshFrontend)
 }

pkg/controller/workload/bpfcache/service.go

@@ -58,15 +58,10 @@ func (c *Cache) ServiceLookup(key *ServiceKey, value *ServiceValue) error {
 // ServiceCount returns the length of service map
 // Note only used for testing
 func (c *Cache) ServiceCount() int {
-	var (
-		key   = ServiceKey{}
-		value = ServiceValue{}
-	)
+	return len(c.ServiceLookupAll())
+}
 
-	res := 0
-	iter := c.bpfMap.KmeshService.Iterate()
-	for iter.Next(&key, &value) {
-		res++
-	}
-	return res
+func (c *Cache) ServiceLookupAll() []ServiceValue {
+	log.Debugf("ServiceLookupAll")
+	return LookupAll[ServiceKey, ServiceValue](c.bpfMap.KmeshService)
 }

pkg/controller/workload/workload_controller.go

@@ -45,7 +45,7 @@
 
 func NewController(bpfWorkload *bpf.BpfKmeshWorkload) *Controller {
 	c := &Controller{
-		Processor:      newProcessor(bpfWorkload.SockConn.KmeshCgroupSockWorkloadObjects.KmeshCgroupSockWorkloadMaps),
+		Processor:      NewProcessor(bpfWorkload.SockConn.KmeshCgroupSockWorkloadObjects.KmeshCgroupSockWorkloadMaps),
 		bpfWorkloadObj: bpfWorkload,
 	}
 	// do some initialization when restart

pkg/controller/workload/workload_controller_test.go

@@ -124,7 +124,7 @@ func TestWorkloadStreamCreateAndSend(t *testing.T) {
 			beforeFunc: func(t *testing.T) {
 				patches1.ApplyMethodReturn(fakeClient.Client, "DeltaAggregatedResources", fakeClient.DeltaClient, nil)
 
-				workloadController.Processor = newProcessor(workloadMap)
+				workloadController.Processor = NewProcessor(workloadMap)
 				workload := createFakeWorkload("10.10.10.1", workloadapi.NetworkMode_STANDARD)
 				workloadController.Processor.WorkloadCache.AddOrUpdateWorkload(workload)
 				patches2.ApplyMethodFunc(fakeClient.DeltaClient, "Send",
@@ -152,7 +152,7 @@ func TestWorkloadStreamCreateAndSend(t *testing.T) {
 			beforeFunc: func(t *testing.T) {
 				patches0.ApplyFuncReturn(maps_v2.AuthorizationUpdate, nil)
 				patches1.ApplyMethodReturn(fakeClient.Client, "DeltaAggregatedResources", fakeClient.DeltaClient, nil)
-				workloadController.Processor = newProcessor(workloadMap)
+				workloadController.Processor = NewProcessor(workloadMap)
 				workloadController.Rbac = auth.NewRbac(nil)
 				workloadController.Rbac.UpdatePolicy(&security.Authorization{
 					Name:      "p1",

pkg/controller/workload/workload_processor.go

@@ -62,7 +62,7 @@
 	authzOnce sync.Once
 }
 
-func newProcessor(workloadMap bpf2go.KmeshCgroupSockWorkloadMaps) *Processor {
+func NewProcessor(workloadMap bpf2go.KmeshCgroupSockWorkloadMaps) *Processor {
 	return &Processor{
 		hashName:      NewHashName(),
 		bpf:           bpf.NewCache(workloadMap),
@@ -93,6 +93,10 @@
 	}
 }
 
+func (p *Processor) GetBpfCache() *bpf.Cache {
+	return p.bpf
+}
+
 func (p *Processor) processWorkloadResponse(rsp *service_discovery_v3.DeltaDiscoveryResponse, rbac *auth.Rbac) {
 	var err error
 
@@ -172,7 +176,7 @@
 	var (
 		err       error
 		bkDelete  = bpf.BackendKey{}
-		wpkDelete = bpf.WorkloadPolicy_key{}
+		wpkDelete = bpf.WorkloadPolicyKey{}
 	)
 
 	backendUid := p.hashName.Hash(uid)
@@ -198,7 +202,7 @@
 	}
 
 	// 4. delete auth policy of workload
-	wpkValue := bpf.WorkloadPolicy_value{}
+	wpkValue := bpf.WorkloadPolicyValue{}
 	wpkDelete.WorklodId = backendUid
 	if err = p.bpf.WorkloadPolicyLookup(&wpkDelete, &wpkValue); err == nil {
 		if err = p.bpf.WorkloadPolicyDelete(&wpkDelete); err != nil {
@@ -746,8 +750,8 @@
 
 func (p *Processor) storeWorkloadPolicies(uid string, polices []string) {
 	var (
-		key   = bpf.WorkloadPolicy_key{}
-		value = bpf.WorkloadPolicy_value{}
+		key   = bpf.WorkloadPolicyKey{}
+		value = bpf.WorkloadPolicyValue{}
 	)
 	if len(polices) == 0 {
 		return

pkg/controller/workload/workload_processor_test.go

@@ -42,7 +42,7 @@ func Test_handleWorkload(t *testing.T) {
 	workloadMap := bpfcache.NewFakeWorkloadMap(t)
 	defer bpfcache.CleanupFakeWorkloadMap(workloadMap)
 
-	p := newProcessor(workloadMap)
+	p := NewProcessor(workloadMap)
 
 	var (
 		ek bpfcache.EndpointKey
@@ -175,7 +175,7 @@ func Test_handleServiceWithWaypoint(t *testing.T) {
 	// Mainly used to test whether processor can correctly handle
 	// different types of waypoint address without panic.
 	workloadMap := bpfcache.NewFakeWorkloadMap(t)
-	p := newProcessor(workloadMap)
+	p := NewProcessor(workloadMap)
 
 	// Waypoint with network address.
 	svc1 := createFakeService("svc1", "10.240.10.1", "10.240.10.200")
@@ -188,7 +188,7 @@ func Test_handleServiceWithWaypoint(t *testing.T) {
 
 func Test_hostnameNetworkMode(t *testing.T) {
 	workloadMap := bpfcache.NewFakeWorkloadMap(t)
-	p := newProcessor(workloadMap)
+	p := NewProcessor(workloadMap)
 	workload := createFakeWorkload("1.2.3.4", workloadapi.NetworkMode_STANDARD)
 	workloadWithoutService := createFakeWorkload("1.2.3.5", workloadapi.NetworkMode_STANDARD)
 	workloadWithoutService.Services = nil
@@ -485,7 +485,7 @@ func TestRestart(t *testing.T) {
 	workloadMap := bpfcache.NewFakeWorkloadMap(t)
 	defer bpfcache.CleanupFakeWorkloadMap(workloadMap)
 
-	p := newProcessor(workloadMap)
+	p := NewProcessor(workloadMap)
 
 	res := &service_discovery_v3.DeltaDiscoveryResponse{}
 
@@ -545,7 +545,7 @@ func TestRestart(t *testing.T) {
 	// Set a restart label and simulate missing data in the cache
 	bpf.SetStartType(bpf.Restart)
 	// reconstruct a new processor
-	p = newProcessor(workloadMap)
+	p = NewProcessor(workloadMap)
 	p.bpf.RestoreEndpointKeys()
 	// 2.1 simulate workload add/delete during restart
 	// simulate workload update during restart