Repository: kmesh-net/kmesh
Pull request #605 (2024-07-18 12:21:11)

support remove residual endpoints
**What type of PR is this?**

/kind enhancement

**What this PR does / why we need it**:
The current code cannot process the following scenarios:
1. The workload is used as the endpoint of service1 and service2.
2. Later, the workload is removed from service2.
In this case, the workload and service2 records in the endpoint are not deleted. As a result, subsequent service access requests are distributed to an incorrect endpoint.

**Which issue(s) this PR fixes**:
Fixes #

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required.
-->
```release-note

```



Patch:
diff --git a/daemon/manager/manager.go b/daemon/manager/manager.go
index 0c99e580..884697f9 100644
--- a/daemon/manager/manager.go
+++ b/daemon/manager/manager.go
@@ -23,6 +23,7 @@ import (
 	"os/signal"
 	"syscall"
 
+	"github.com/cilium/ebpf/rlimit"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 
@@ -73,8 +74,13 @@ func NewCommand() *cobra.Command {
 
 // Execute start daemon manager process
 func Execute(configs *options.BootstrapConfigs) error {
+	err := rlimit.RemoveMemlock()
+	if err != nil {
+		log.Warn("rlimit.RemoveMemlock failed")
+	}
+
 	bpfLoader := bpf.NewBpfLoader(configs.BpfConfig)
-	if err := bpfLoader.Start(configs.BpfConfig); err != nil {
+	if err = bpfLoader.Start(configs.BpfConfig); err != nil {
 		return err
 	}
 	log.Info("bpf Start successful")
diff --git a/pkg/bpf/bpf.go b/pkg/bpf/bpf.go
index 3e585156..fd064d15 100644
--- a/pkg/bpf/bpf.go
+++ b/pkg/bpf/bpf.go
@@ -30,7 +30,6 @@ import (
 	"syscall"
 
 	"github.com/cilium/ebpf"
-	"github.com/cilium/ebpf/rlimit"
 
 	"kmesh.net/kmesh/daemon/options"
 	"kmesh.net/kmesh/pkg/constants"
@@ -116,10 +115,6 @@ func StartMda() error {
 func (l *BpfLoader) Start(config *options.BpfConfig) error {
 	var err error
 
-	if err = rlimit.RemoveMemlock(); err != nil {
-		return err
-	}
-
 	if l.VersionMap == nil {
 		return fmt.Errorf("NewVersionMap failed")
 	}
diff --git a/pkg/controller/workload/bpfcache/fake_map.go b/pkg/controller/workload/bpfcache/fake_map.go
index d17a45c3..28f19690 100644
--- a/pkg/controller/workload/bpfcache/fake_map.go
+++ b/pkg/controller/workload/bpfcache/fake_map.go
@@ -21,11 +21,13 @@ import (
 	"unsafe"
 
 	"github.com/cilium/ebpf"
+	"github.com/cilium/ebpf/rlimit"
 
 	"kmesh.net/kmesh/bpf/kmesh/bpf2go"
 )
 
 func NewFakeWorkloadMap(t *testing.T) bpf2go.KmeshCgroupSockWorkloadMaps {
+	_ = rlimit.RemoveMemlock()
 	backEndMap, err := ebpf.NewMap(&ebpf.MapSpec{
 		Name:       "kmesh_backend",
 		Type:       ebpf.Hash,
diff --git a/pkg/controller/workload/cache/workload_cache.go b/pkg/controller/workload/cache/workload_cache.go
index 4763b1b3..d3180f40 100644
--- a/pkg/controller/workload/cache/workload_cache.go
+++ b/pkg/controller/workload/cache/workload_cache.go
@@ -28,9 +28,12 @@ import (
 type WorkloadCache interface {
 	GetWorkloadByUid(uid string) *workloadapi.Workload
 	GetWorkloadByAddr(networkAddress NetworkAddress) *workloadapi.Workload
-	AddWorkload(workload *workloadapi.Workload)
+	AddOrUpdateWorkload(workload *workloadapi.Workload) (deletedServices []string, newServices []string)
 	DeleteWorkload(uid string)
 	List() []*workloadapi.Workload
+	GetRelationShip(workloadId uint32, serviceId uint32) (uint32, bool)
+	UpdateRelationShip(workloadId uint32, serviceId uint32, relationId uint32)
+	DeleteRelationShip(serviceId uint32, relationId uint32)
 }
 
 type NetworkAddress struct {
@@ -38,16 +41,76 @@ type NetworkAddress struct {
 	Address netip.Addr
 }
 
+type ServiceRelationShipByWorkload struct {
+	workloadId uint32
+	serviceId  uint32
+}
+
+type ServiceRelationShipById struct {
+	serviceId  uint32
+	relationId uint32
+}
+
 type cache struct {
-	byUid  map[string]*workloadapi.Workload
-	byAddr map[NetworkAddress]*workloadapi.Workload
-	mutex  sync.RWMutex
+	byUid                  map[string]*workloadapi.Workload
+	byAddr                 map[NetworkAddress]*workloadapi.Workload
+	relationShipByWorkload map[ServiceRelationShipByWorkload]uint32
+	relationShipById       map[ServiceRelationShipById]uint32
+	mutex                  sync.RWMutex
 }
 
 func NewWorkloadCache() *cache {
 	return &cache{
-		byUid:  make(map[string]*workloadapi.Workload),
-		byAddr: make(map[NetworkAddress]*workloadapi.Workload),
+		byUid:                  make(map[string]*workloadapi.Workload),
+		byAddr:                 make(map[NetworkAddress]*workloadapi.Workload),
+		relationShipByWorkload: make(map[ServiceRelationShipByWorkload]uint32),
+		relationShipById:       make(map[ServiceRelationShipById]uint32),
+	}
+}
+
+func (w *cache) GetRelationShip(workloadId uint32, serviceId uint32) (uint32, bool) {
+	var relationKey = ServiceRelationShipByWorkload{
+		workloadId: workloadId,
+		serviceId:  serviceId,
+	}
+
+	w.mutex.RLock()
+	defer w.mutex.RUnlock()
+	relationId, exist := w.relationShipByWorkload[relationKey]
+	return relationId, exist
+}
+
+func (w *cache) UpdateRelationShip(workloadId uint32, serviceId uint32, relationId uint32) {
+	var relationKey = ServiceRelationShipByWorkload{
+		workloadId: workloadId,
+		serviceId:  serviceId,
+	}
+	var relationKeyById = ServiceRelationShipById{
+		serviceId:  serviceId,
+		relationId: relationId,
+	}
+
+	w.mutex.Lock()
+	defer w.mutex.Unlock()
+	w.relationShipByWorkload[relationKey] = relationId
+	w.relationShipById[relationKeyById] = workloadId
+}
+
+func (w *cache) DeleteRelationShip(serviceId uint32, relationId uint32) {
+	var relationKeyById = ServiceRelationShipById{
+		serviceId:  serviceId,
+		relationId: relationId,
+	}
+
+	w.mutex.Lock()
+	defer w.mutex.Unlock()
+	if workloadId, ok := w.relationShipById[relationKeyById]; ok {
+		delete(w.relationShipById, relationKeyById)
+		var relationKey = ServiceRelationShipByWorkload{
+			workloadId: workloadId,
+			serviceId:  serviceId,
+		}
+		delete(w.relationShipByWorkload, relationKey)
 	}
 }
 
@@ -70,26 +133,59 @@ func composeNetworkAddress(network string, addr netip.Addr) NetworkAddress {
 	}
 }
 
-func (w *cache) AddWorkload(workload *workloadapi.Workload) {
+func (w *cache) getUniqueServicesOnLeftWorkload(workload1, workload2 *workloadapi.Workload) []string {
+	var diff []string
+	if workload1 == nil {
+		return diff
+	}
+
+	for key := range workload1.Services {
+		if workload2 == nil {
+			diff = append(diff, key)
+			continue
+		}
+		if _, exist := workload2.Services[key]; !exist {
+			diff = append(diff, key)
+		}
+	}
+	return diff
+}
+
+func (w *cache) compareWorkloadServices(workload1, workload2 *workloadapi.Workload) ([]string, []string) {
+	dels := w.getUniqueServicesOnLeftWorkload(workload1, workload2)
+	news := w.getUniqueServicesOnLeftWorkload(workload2, workload1)
+	return dels, news
+}
+
+func (w *cache) AddOrUpdateWorkload(workload *workloadapi.Workload) ([]string, []string) {
+	var deletedServices []string
+	var newServices []string
+
 	if workload == nil {
-		return
+		return nil, nil
 	}
 	uid := workload.Uid
 
 	w.mutex.Lock()
 	defer w.mutex.Unlock()
 
-	workloadByUid, exist := w.byUid[uid]
+	oldWorkload, exist := w.byUid[uid]
 	if exist {
-		if proto.Equal(workload, workloadByUid) {
-			return
+		if proto.Equal(workload, oldWorkload) {
+			return nil, nil
 		}
-		// remove same uid but old address workload, avoid leak worklaod by address.
-		for _, ip := range workloadByUid.Addresses {
+		// remove same uid but old address workload, avoid leak workload by address.
+		for _, ip := range oldWorkload.Addresses {
 			addr, _ := netip.AddrFromSlice(ip)
-			networkAddress := composeNetworkAddress(workloadByUid.Network, addr)
+			networkAddress := composeNetworkAddress(oldWorkload.Network, addr)
 			delete(w.byAddr, networkAddress)
 		}
+
+		// compare services
+		deletedServices, newServices = w.compareWorkloadServices(oldWorkload, workload)
+	} else {
+		deletedServices = nil
+		newServices = w.getUniqueServicesOnLeftWorkload(workload, oldWorkload)
 	}
 
 	w.byUid[uid] = workload
@@ -103,6 +199,7 @@ func (w *cache) AddWorkload(workload *workloadapi.Workload) {
 			w.byAddr[networkAddress] = workload
 		}
 	}
+	return deletedServices, newServices
 }
 
 func (w *cache) DeleteWorkload(uid string) {
diff --git a/pkg/controller/workload/workload_processor.go b/pkg/controller/workload/workload_processor.go
index 1d3aa745..5b474134 100644
--- a/pkg/controller/workload/workload_processor.go
+++ b/pkg/controller/workload/workload_processor.go
@@ -156,12 +156,8 @@ func (p *Processor) removeWorkloadResource(removedResources []string) error {
 
 func (p *Processor) removeWorkloadFromBpfMap(uid string) error {
 	var (
-		err               error
-		skUpdate          = bpf.ServiceKey{}
-		svUpdate          = bpf.ServiceValue{}
-		lastEndpointKey   = bpf.EndpointKey{}
-		lastEndpointValue = bpf.EndpointValue{}
-		bkDelete          = bpf.BackendKey{}
+		err      error
+		bkDelete = bpf.BackendKey{}
 	)
 
 	backendUid := p.hashName.StrToNum(uid)
@@ -173,57 +169,20 @@ func (p *Processor) removeWorkloadFromBpfMap(uid string) error {
 
 	// 1. find all endpoint keys related to this workload
 	if eks := p.bpf.EndpointIterFindKey(backendUid); len(eks) != 0 {
-		for _, ek := range eks {
-			log.Debugf("Find EndpointKey: [%#v]", ek)
-			// 2. find the service
-			skUpdate.ServiceId = ek.ServiceId
-			if err = p.bpf.ServiceLookup(&skUpdate, &svUpdate); err == nil {
-				log.Debugf("Find ServiceValue: [%#v]", svUpdate)
-				// 3. find the last indexed endpoint of the service
-				lastEndpointKey.ServiceId = skUpdate.ServiceId
-				lastEndpointKey.BackendIndex = svUpdate.EndpointCount
-				if err = p.bpf.EndpointLookup(&lastEndpointKey, &lastEndpointValue); err == nil {
-					log.Debugf("Find EndpointValue: [%#v]", lastEndpointValue)
-					// 4. switch the index of the last with the current removed endpoint
-					if err = p.bpf.EndpointUpdate(&ek, &lastEndpointValue); err != nil {
-						log.Errorf("EndpointUpdate failed: %s", err)
-						goto failed
-					}
-					if err = p.bpf.EndpointDelete(&lastEndpointKey); err != nil {
-						log.Errorf("EndpointDelete failed: %s", err)
-						goto failed
-					}
-					svUpdate.EndpointCount = svUpdate.EndpointCount - 1
-					if err = p.bpf.ServiceUpdate(&skUpdate, &svUpdate); err != nil {
-						log.Errorf("ServiceUpdate failed: %s", err)
-						goto failed
-					}
-				} else {
-					// last indexed endpoint not exists, this should not occur
-					// we should delete the endpoint just in case leak
-					if err = p.bpf.EndpointDelete(&ek); err != nil {
-						log.Errorf("EndpointDelete failed: %s", err)
-						goto failed
-					}
-				}
-			} else { // service not exist, we should delete the endpoint
-				if err = p.bpf.EndpointDelete(&ek); err != nil {
-					log.Errorf("EndpointDelete failed: %s", err)
-					goto failed
-				}
-			}
+		err = p.deleteEndpointRecords(eks)
+		if err != nil {
+			return err
 		}
 	}
 
 	bkDelete.BackendUid = backendUid
 	if err = p.bpf.BackendDelete(&bkDelete); err != nil {
 		log.Errorf("BackendDelete failed: %s", err)
-		goto failed
+		return err
 	}
-	p.hashName.Delete(uid)
 
-failed:
-	return err
+	p.hashName.Delete(uid)
+	return nil
 }
 
 func (p *Processor) deleteFrontendData(id uint32) error {
@@ -283,6 +242,7 @@ func (p *Processor) removeServiceResourceFromBpfMap(name string) error {
 		for i = 1; i <= svDelete.EndpointCount; i++ {
 			ekDelete.ServiceId = serviceId
 			ekDelete.BackendIndex = i
+
 			if err = p.bpf.EndpointDelete(&ekDelete); err != nil {
 				log.Errorf("EndpointDelete failed: %s", err)
 				goto failed
@@ -313,6 +273,7 @@ func (p *Processor) storeEndpointWithService(sk *bpf.ServiceKey, sv *bpf.Service
 		return err
 	}
 
+	p.WorkloadCache.UpdateRelationShip(ev.BackendUid, ek.ServiceId, ek.BackendIndex)
 	return nil
 }
 
@@ -326,104 +287,92 @@ func (p *Processor) storeServiceEndpoint(workload_uid string, serviceName string
 	wls[workload_uid] = struct{}{}
 }
 
-func (p *Processor) storeBackendData(uid uint32, ip []byte, waypoint *workloadapi.GatewayAddress, portList map[string]*workloadapi.PortList, networkMode workloadapi.NetworkMode) error {
+func (p *Processor) deleteResidualServicesWithWorkload(workload *workloadapi.Workload, services []string) error {
 	var (
-		bk = bpf.BackendKey{}
-		bv = bpf.BackendValue{}
+		err       error
+		serviceId uint32
 	)
 
-	bk.BackendUid = uid
-	nets.CopyIpByteFromSlice(&bv.Ip, ip)
-	bv.ServiceCount = 0
-	for serviceName := range portList {
-		bv.Services[bv.ServiceCount] = p.hashName.StrToNum(serviceName)
-		bv.ServiceCount++
-		if bv.ServiceCount >= bpf.MaxServiceNum {
-			log.Warnf("exceed the max service count, currently, a pod can belong to a maximum of 10 services")
-			break
+	if services == nil {
+		return nil
+	}
+
+	log.Infof("deleteResidualServicesWithWorkload: %v", services)
+	eks := make([]bpf.EndpointKey, 0)
+	workloadUid := p.hashName.StrToNum(workload.GetUid())
+	serviceIds := make(map[uint32]struct{})
+	for _, serviceName := range services {
+		serviceId = p.hashName.StrToNum(serviceName)
+		if relationId, ok := p.WorkloadCache.GetRelationShip(workloadUid, serviceId); ok {
+			eks = append(eks, bpf.EndpointKey{
+				ServiceId:    serviceId,
+				BackendIndex: relationId,
+			})
 		}
+		serviceIds[serviceId] = struct{}{}
 	}
 
-	if waypoint != nil {
-		nets.CopyIpByteFromSlice(&bv.WaypointAddr, waypoint.GetAddress().Address)
-		bv.WaypointPort = nets.ConvertPortToBigEndian(waypoint.GetHboneMtlsPort())
-	}
-
-	if err := p.bpf.BackendUpdate(&bk, &bv); err != nil {
-		log.Errorf("Update backend map failed, err:%s", err)
-		return err
-	}
-
-	// we should not store frontend data of hostname network mode pods
-	// because host network pod cannot be managed by kmesh
-	// please see https://github.com/kmesh-net/kmesh/issues/631
-	if networkMode != workloadapi.NetworkMode_HOST_NETWORK {
-		if err := p.storePodFrontendData(uid, ip); err != nil {
-			log.Errorf("storePodFrontendData failed, err:%s", err)
-			return err
-		}
+	err = p.deleteEndpointRecords(eks)
+	if err != nil {
+		log.Errorf("removeResidualServices delete endpoint failed:%v", err)
 	}
-
-	return nil
+	return err
 }
 
-func (p *Processor) handleDataWithService(workload *workloadapi.Workload) error {
+func (p *Processor) addNewServicesWithWorkload(workload *workloadapi.Workload, newServices []string) error {
 	var (
 		err error
 		sk  = bpf.ServiceKey{}
 		sv  = bpf.ServiceValue{}
-
-		bk = bpf.BackendKey{}
-		bv = bpf.BackendValue{}
 	)
 
-	backend_uid := p.hashName.StrToNum(workload.GetUid())
-	for serviceName := range workload.GetServices() {
-		bk.BackendUid = backend_uid
-		// for update sense, if the backend is exist, just need update it
-		if err = p.bpf.BackendLookup(&bk, &bv); err != nil {
-			sk.ServiceId = p.hashName.StrToNum(serviceName)
-			// the service already stored in map, add endpoint
-			if err = p.bpf.ServiceLookup(&sk, &sv); err == nil {
-				if err = p.storeEndpointWithService(&sk, &sv, backend_uid); err != nil {
-					log.Errorf("storeEndpointWithService failed, err:%s", err)
-					return err
-				}
-			} else {
-				p.storeServiceEndpoint(workload.GetUid(), serviceName)
-			}
-		}
-	}
-
-	if len(workload.GetAddresses()) > 1 {
-		log.Warnf("current only support single ip of a pod")
+	if newServices == nil {
+		return nil
 	}
 
-	for _, ip := range workload.GetAddresses() { // current only support single ip, if a pod have multi ips, the last one will be stored finally
-		if err = p.storeBackendData(backend_uid, ip, workload.GetWaypoint(), workload.GetServices(), workload.GetNetworkMode()); err != nil {
-			log.Errorf("storeBackendData failed, err:%s", err)
-			return err
+	log.Infof("addNewServicesWithWorkload: %v", newServices)
+	backend_uid := p.hashName.StrToNum(workload.GetUid())
+	for _, serviceName := range newServices {
+		sk.ServiceId = p.hashName.StrToNum(serviceName)
+		// the service already stored in map, add endpoint
+		if err = p.bpf.ServiceLookup(&sk, &sv); err == nil {
+			if err = p.storeEndpointWithService(&sk, &sv, backend_uid); err != nil {
+				log.Errorf("storeEndpointWithService failed, err:%s", err)
+				return err
+			}
+		} else {
+			p.storeServiceEndpoint(workload.GetUid(), serviceName)
 		}
 	}
-
 	return nil
 }
 
-func (p *Processor) handleDataWithoutService(workload *workloadapi.Workload) error {
+func (p *Processor) updateWorkload(workload *workloadapi.Workload) error {
 	var (
-		err error
-		bk  = bpf.BackendKey{}
-		bv  = bpf.BackendValue{}
+		err         error
+		bk          = bpf.BackendKey{}
+		bv          = bpf.BackendValue{}
+		networkMode = workload.GetNetworkMode()
 	)
 
 	uid := p.hashName.StrToNum(workload.GetUid())
 	ips := workload.GetAddresses()
-	for _, ip := range ips {
-		if waypoint := workload.GetWaypoint(); waypoint != nil {
-			nets.CopyIpByteFromSlice(&bv.WaypointAddr, waypoint.GetAddress().Address)
-			bv.WaypointPort = nets.ConvertPortToBigEndian(waypoint.GetHboneMtlsPort())
+
+	if waypoint := workload.GetWaypoint(); waypoint != nil {
+		nets.CopyIpByteFromSlice(&bv.WaypointAddr, waypoint.GetAddress().Address)
+		bv.WaypointPort = nets.ConvertPortToBigEndian(waypoint.GetHboneMtlsPort())
+	}
+
+	for serviceName := range workload.GetServices() {
+		bv.Services[bv.ServiceCount] = p.hashName.StrToNum(serviceName)
+		bv.ServiceCount++
+		if bv.ServiceCount >= bpf.MaxServiceNum {
+			log.Warnf("exceed the max service count, currently, a pod can belong to a maximum of 10 services")
+			break
 		}
+	}
 
+	for _, ip := range ips {
 		bk.BackendUid = uid
 
 		nets.CopyIpByteFromSlice(&bv.Ip, ip)
@@ -433,9 +382,8 @@ func (p *Processor) handleDataWithoutService(workload *workloadapi.Workload) err
 		}
 
 		// we should not store frontend data of hostname network mode pods
-		// because host network pod cannot be managed by kmesh
 		// please see https://github.com/kmesh-net/kmesh/issues/631
-		if workload.GetNetworkMode() != workloadapi.NetworkMode_HOST_NETWORK {
+		if networkMode != workloadapi.NetworkMode_HOST_NETWORK {
 			if err = p.storePodFrontendData(uid, ip); err != nil {
 				log.Errorf("storePodFrontendData failed, err:%s", err)
 				return err
@@ -446,20 +394,28 @@ func (p *Processor) handleDataWithoutService(workload *workloadapi.Workload) err
 }
 
 func (p *Processor) handleWorkload(workload *workloadapi.Workload) error {
+	var deletedServices []string
+	var newServices []string
 	log.Debugf("handle workload: %s", workload.Uid)
-	p.WorkloadCache.AddWorkload(workload)
 
-	// if have the service name, the workload belongs to a service
-	if workload.GetServices() != nil {
-		if err := p.handleDataWithService(workload); err != nil {
-			log.Errorf("handleDataWithService %s failed: %v", workload.Uid, err)
-			return err
-		}
-	} else { // independent workload without service
-		if err := p.handleDataWithoutService(workload); err != nil {
-			log.Errorf("handleDataWithoutService %s failed: %v", workload.Uid, err)
-			return err
-		}
+	deletedServices, newServices = p.WorkloadCache.AddOrUpdateWorkload(workload)
+
+	// Delete Residual Services on the Workload
+	if err := p.deleteResidualServicesWithWorkload(workload, deletedServices); err != nil {
+		log.Errorf("deleteResidualServicesWithWorkload %s failed: %v", workload.GetUid(), err)
+		return err
+	}
+
+	// Add new services associated with the workload
+	if err := p.addNewServicesWithWorkload(workload, newServices); err != nil {
+		log.Errorf("addNewServicesWithWorkload %s failed: %v", workload.Uid, err)
+		return err
+	}
+
+	// Update workload
+	if err := p.updateWorkload(workload); err != nil {
+		log.Errorf("updateWorkload %s failed: %v", workload.Uid, err)
+		return err
 	}
 
 	return nil
@@ -534,6 +490,7 @@ func (p *Processor) storeServiceData(serviceName string, waypoint *workloadapi.G
 					log.Errorf("Update Endpoint failed, err:%s", err)
 					return err
 				}
+				p.WorkloadCache.UpdateRelationShip(ev.BackendUid, ek.ServiceId, ek.BackendIndex)
 			}
 		}
 		delete(p.endpointsByService, serviceName)
@@ -701,3 +658,90 @@ func (p *Processor) handleAuthorizationTypeResponse(rsp *service_discovery_v3.De
 
 	return nil
 }
+
+func (p *Processor) deleteEndpointRecords(endpoint_keys []bpf.EndpointKey) error {
+	var (
+		err               error
+		skUpdate          = bpf.ServiceKey{}
+		svUpdate          = bpf.ServiceValue{}
+		lastEndpointKey   = bpf.EndpointKey{}
+		lastEndpointValue = bpf.EndpointValue{}
+	)
+
+	for _, ek := range endpoint_keys {
+		log.Debugf("Find EndpointKey: [%#v]", ek)
+
+		// 2. find the service
+		skUpdate.ServiceId = ek.ServiceId
+		if err = p.bpf.ServiceLookup(&skUpdate, &svUpdate); err == nil {
+			log.Debugf("Find ServiceValue: [%#v]", svUpdate)
+			// 3. find the last indexed endpoint of the service
+			lastEndpointKey.ServiceId = skUpdate.ServiceId
+			lastEndpointKey.BackendIndex = svUpdate.EndpointCount
+			if err = p.bpf.EndpointLookup(&lastEndpointKey, &lastEndpointValue); err == nil {
+				log.Debugf("Find EndpointValue: [%#v]", lastEndpointValue)
+				// 4. switch the index of the last with the current removed endpoint
+				if err = p.updateRelationShipWithWorkloadAndService(lastEndpointValue.BackendUid, ek.ServiceId, ek.BackendIndex); err != nil {
+					log.Errorf("EndpointUpdate failed: %s", err)
+					return err
+				}
+
+				if err = p.bpf.EndpointDelete(&lastEndpointKey); err != nil {
+					log.Errorf("EndpointDelete failed: %s", err)
+					return err
+				}
+				p.WorkloadCache.DeleteRelationShip(ek.ServiceId, ek.BackendIndex)
+
+				svUpdate.EndpointCount = svUpdate.EndpointCount - 1
+				if err = p.bpf.ServiceUpdate(&skUpdate, &svUpdate); err != nil {
+					log.Errorf("ServiceUpdate failed: %s", err)
+					return err
+				}
+			} else {
+				// last indexed endpoint not exists, this should not occur
+				// we should delete the endpoint just in case leak
+				if err = p.deleteRelationShipWithWorkloadAndService(ek.ServiceId, ek.BackendIndex); err != nil {
+					log.Errorf("EndpointDelete failed: %s", err)
+					return err
+				}
+			}
+		} else { // service not exist, we should delete the endpoint
+			if err = p.deleteRelationShipWithWorkloadAndService(ek.ServiceId, ek.BackendIndex); err != nil {
+				log.Errorf("EndpointDelete failed: %s", err)
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (p *Processor) updateRelationShipWithWorkloadAndService(workloadId uint32, serviceId uint32, relationId uint32) error {
+	var ek = bpf.EndpointKey{
+		ServiceId:    serviceId,
+		BackendIndex: relationId,
+	}
+	var ev = bpf.EndpointValue{
+		BackendUid: workloadId,
+	}
+
+	if err := p.bpf.EndpointUpdate(&ek, &ev); err != nil {
+		log.Errorf("EndpointUpdate failed: %s", err)
+		return err
+	}
+	p.WorkloadCache.UpdateRelationShip(workloadId, serviceId, relationId)
+	return nil
+}
+
+func (p *Processor) deleteRelationShipWithWorkloadAndService(serviceId uint32, relationId uint32) error {
+	var ek = bpf.EndpointKey{
+		ServiceId:    serviceId,
+		BackendIndex: relationId,
+	}
+
+	if err := p.bpf.EndpointDelete(&ek); err != nil {
+		log.Errorf("EndpointDelete failed: %s", err)
+		return err
+	}
+	p.WorkloadCache.DeleteRelationShip(serviceId, relationId)
+	return nil
+}


Test patch:
diff --git a/pkg/auth/rbac_test.go b/pkg/auth/rbac_test.go
index 2980e77b..5bd72046 100644
--- a/pkg/auth/rbac_test.go
+++ b/pkg/auth/rbac_test.go
@@ -1967,7 +1967,7 @@ func TestRbac_doRbac(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			workloadCache := cache.NewWorkloadCache()
-			workloadCache.AddWorkload(tt.args.workload)
+			workloadCache.AddOrUpdateWorkload(tt.args.workload)
 			rbac := &Rbac{
 				policyStore:   tt.fields.policyStore,
 				workloadCache: workloadCache,
@@ -2145,7 +2145,7 @@ func TestRbac_Run(t *testing.T) {
 	}
 
 	workloadCache := cache.NewWorkloadCache()
-	workloadCache.AddWorkload(&workloadapi.Workload{
+	workloadCache.AddOrUpdateWorkload(&workloadapi.Workload{
 		Name: "ut-workload",
 		Uid:  "123456",
 		Addresses: [][]byte{
diff --git a/pkg/controller/telemetry/metric_test.go b/pkg/controller/telemetry/metric_test.go
index 0bb0a4f7..4085602a 100644
--- a/pkg/controller/telemetry/metric_test.go
+++ b/pkg/controller/telemetry/metric_test.go
@@ -353,7 +353,7 @@ func TestMetricGetWorkloadByAddress(t *testing.T) {
 			m := MetricController{
 				workloadCache: cache.NewWorkloadCache(),
 			}
-			m.workloadCache.AddWorkload(workload)
+			m.workloadCache.AddOrUpdateWorkload(workload)
 			if got, _ := m.getWorkloadByAddress(tt.args.address); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("Metric.getWorkloadByAddress() = %v, want %v", got, tt.want)
 			}
@@ -441,8 +441,8 @@ func TestBuildworkloadMetric(t *testing.T) {
 			m := MetricController{
 				workloadCache: cache.NewWorkloadCache(),
 			}
-			m.workloadCache.AddWorkload(dstWorkload)
-			m.workloadCache.AddWorkload(srcWorkload)
+			m.workloadCache.AddOrUpdateWorkload(dstWorkload)
+			m.workloadCache.AddOrUpdateWorkload(srcWorkload)
 			got := m.buildWorkloadMetric(tt.args.data)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("Metric.buildMetric() = %v, want %v", got, tt.want)
@@ -576,8 +576,8 @@ func TestBuildServiceMetric(t *testing.T) {
 			m := MetricController{
 				workloadCache: cache.NewWorkloadCache(),
 			}
-			m.workloadCache.AddWorkload(dstWorkload)
-			m.workloadCache.AddWorkload(srcWorkload)
+			m.workloadCache.AddOrUpdateWorkload(dstWorkload)
+			m.workloadCache.AddOrUpdateWorkload(srcWorkload)
 			got := m.buildServiceMetric(tt.args.data)
 			assert.Equal(t, tt.want, got)
 		})
diff --git a/pkg/controller/workload/cache/workload_cache_test.go b/pkg/controller/workload/cache/workload_cache_test.go
index 29e90f91..b6715ef8 100644
--- a/pkg/controller/workload/cache/workload_cache_test.go
+++ b/pkg/controller/workload/cache/workload_cache_test.go
@@ -25,7 +25,7 @@ import (
 	"kmesh.net/kmesh/api/v2/workloadapi"
 )
 
-func TestAddWorkload(t *testing.T) {
+func TestAddOrUpdateWorkload(t *testing.T) {
 	t.Run("adding a workload when none exists", func(t *testing.T) {
 		w := NewWorkloadCache()
 		workload := &workloadapi.Workload{
@@ -37,7 +37,7 @@ func TestAddWorkload(t *testing.T) {
 				[]byte("1.2.3.4"),
 			},
 		}
-		w.AddWorkload(workload)
+		w.AddOrUpdateWorkload(workload)
 		assert.Equal(t, workload, w.byUid["123456"])
 		addr1, _ := netip.AddrFromSlice([]byte("192.168.224.22"))
 		addr2, _ := netip.AddrFromSlice([]byte("1.2.3.4"))
@@ -56,7 +56,7 @@ func TestAddWorkload(t *testing.T) {
 				[]byte("1.2.3.4"),
 			},
 		}
-		w.AddWorkload(workload)
+		w.AddOrUpdateWorkload(workload)
 		assert.Equal(t, workload, w.byUid["123456"])
 		addr1, _ := netip.AddrFromSlice([]byte("192.168.224.22"))
 		addr2, _ := netip.AddrFromSlice([]byte("1.2.3.4"))
@@ -71,7 +71,7 @@ func TestAddWorkload(t *testing.T) {
 				[]byte("2.3.4.5"),
 			},
 		}
-		w.AddWorkload(newWorkload)
+		w.AddOrUpdateWorkload(newWorkload)
 		assert.Equal(t, newWorkload, w.byUid["123456"])
 		addr3, _ := netip.AddrFromSlice([]byte("192.168.10.25"))
 		addr4, _ := netip.AddrFromSlice([]byte("2.3.4.5"))
@@ -91,7 +91,7 @@ func TestAddWorkload(t *testing.T) {
 				[]byte("192.168.224.22"),
 			},
 		}
-		w.AddWorkload(workload)
+		w.AddOrUpdateWorkload(workload)
 		assert.Equal(t, workload, w.byUid["123456"])
 		addr, _ := netip.AddrFromSlice([]byte("192.168.224.22"))
 		assert.Equal(t, workload, w.byAddr[NetworkAddress{Network: workload.Network, Address: addr}])
@@ -104,7 +104,7 @@ func TestAddWorkload(t *testing.T) {
 				[]byte("2.3.4.5"),
 			},
 		}
-		w.AddWorkload(newWorkload)
+		w.AddOrUpdateWorkload(newWorkload)
 		assert.Equal(t, newWorkload, w.byUid["123456"])
 		addr1, _ := netip.AddrFromSlice([]byte("192.168.224.22"))
 		addr2, _ := netip.AddrFromSlice([]byte("2.3.4.5"))
@@ -126,7 +126,7 @@ func TestDeleteWorkload(t *testing.T) {
 				[]byte("world"),
 			},
 		}
-		w.AddWorkload(workload)
+		w.AddOrUpdateWorkload(workload)
 		assert.Equal(t, workload, w.byUid["123456"])
 		w.DeleteWorkload("123456")
 		assert.Equal(t, (*workloadapi.Workload)(nil), w.byUid["123456"])
@@ -136,3 +136,33 @@ func TestDeleteWorkload(t *testing.T) {
 		assert.Equal(t, (*workloadapi.Workload)(nil), w.byAddr[NetworkAddress{Network: "ut-net", Address: addr2}])
 	})
 }
+
+func TestWorkloadRelationShip(t *testing.T) {
+	t.Run("normal function test", func(t *testing.T) {
+		w := NewWorkloadCache()
+
+		var serviceId uint32 = 12345678
+		var relationId uint32 = 87654321
+		var workloadId uint32 = 12341234
+
+		var relationKey = ServiceRelationShipByWorkload{
+			workloadId: workloadId,
+			serviceId:  serviceId,
+		}
+		var relationKeyById = ServiceRelationShipById{
+			serviceId:  serviceId,
+			relationId: relationId,
+		}
+
+		w.UpdateRelationShip(workloadId, serviceId, relationId)
+		assert.Equal(t, workloadId, w.relationShipById[relationKeyById])
+		assert.Equal(t, relationId, w.relationShipByWorkload[relationKey])
+
+		val, ok := w.GetRelationShip(workloadId, serviceId)
+		assert.Equal(t, ok, true)
+		assert.Equal(t, val, relationId)
+
+		w.DeleteRelationShip(serviceId, relationId)
+		assert.Equal(t, (uint32)(0), w.relationShipById[relationKeyById])
+	})
+}
diff --git a/pkg/controller/workload/workload_controller_test.go b/pkg/controller/workload/workload_controller_test.go
index 883fc57b..b9aab7e3 100644
--- a/pkg/controller/workload/workload_controller_test.go
+++ b/pkg/controller/workload/workload_controller_test.go
@@ -124,7 +124,7 @@ func TestWorkloadStreamCreateAndSend(t *testing.T) {
 
 				workloadController.Processor = newProcessor(workloadMap)
 				workload := createFakeWorkload("10.10.10.1", workloadapi.NetworkMode_STANDARD)
-				workloadController.Processor.WorkloadCache.AddWorkload(workload)
+				workloadController.Processor.WorkloadCache.AddOrUpdateWorkload(workload)
 				patches2.ApplyMethodFunc(fakeClient.DeltaClient, "Send",
 					func(req *discoveryv3.DeltaDiscoveryRequest) error {
 						if req.TypeUrl == AddressType {
diff --git a/pkg/controller/workload/workload_processor_test.go b/pkg/controller/workload/workload_processor_test.go
index 046724e6..388f0b54 100644
--- a/pkg/controller/workload/workload_processor_test.go
+++ b/pkg/controller/workload/workload_processor_test.go
@@ -41,8 +41,10 @@ func Test_handleWorkload(t *testing.T) {
 
 	// 1. handle workload with service, but service not handled yet
 	// In this case, only frontend map and backend map should be updated.
-	wl := createTestWorkloadWithService()
-	_ = p.handleDataWithService(createTestWorkloadWithService())
+	wl := createTestWorkloadWithService(true)
+	err := p.handleWorkload(wl)
+	assert.NoError(t, err)
+
 	var (
 		ek bpfcache.EndpointKey
 		ev bpfcache.EndpointValue
@@ -74,13 +76,14 @@ func Test_handleWorkload(t *testing.T) {
 	// 2.3 check endpoint map now contains the workloads
 	ek.BackendIndex = 1
 	ek.ServiceId = svcID
-	err := p.bpf.EndpointLookup(&ek, &ev)
+	err = p.bpf.EndpointLookup(&ek, &ev)
 	assert.NoError(t, err)
 	assert.Equal(t, ev.BackendUid, workloadID)
 
 	// 3. add another workload with service
 	workload2 := createFakeWorkload("1.2.3.5", workloadapi.NetworkMode_STANDARD)
-	_ = p.handleDataWithService(workload2)
+	err = p.handleWorkload(workload2)
+	assert.NoError(t, err)
 
 	// 3.1 check endpoint map now contains the new workloads
 	workload2ID := checkFrontEndMap(t, workload2.Addresses[0], p)
@@ -93,13 +96,56 @@ func Test_handleWorkload(t *testing.T) {
 	// 3.2 check service map contains service
 	checkServiceMap(t, p, svcID, fakeSvc, 2)
 
-	// 4. add namespace scoped waypoint service
+	// 4 modify workload2 attribute not relationship with services
+	workload2.Waypoint = &workloadapi.GatewayAddress{
+		Destination: &workloadapi.GatewayAddress_Address{
+			Address: &workloadapi.NetworkAddress{
+				Address: netip.MustParseAddr("10.10.10.10").AsSlice(),
+			},
+		},
+		HboneMtlsPort: 15008,
+	}
+
+	err = p.handleWorkload(workload2)
+	assert.NoError(t, err)
+
+	// 4.1 check endpoint map now contains the new workloads
+	workload2ID = checkFrontEndMap(t, workload2.Addresses[0], p)
+	ek.BackendIndex = 2
+	ek.ServiceId = svcID
+	err = p.bpf.EndpointLookup(&ek, &ev)
+	assert.NoError(t, err)
+	assert.Equal(t, ev.BackendUid, workload2ID)
+
+	// 4.2 check service map contains service
+	checkServiceMap(t, p, svcID, fakeSvc, 2)
+
+	// 4.3 check backend map contains waypoint
+	checkBackendMap(t, p, workload2ID, workload2)
+
+	// 5 remove relationship with workload and service
+	workload5 := createTestWorkloadWithService(false)
+	workload5.Uid = wl.Uid
+	err = p.handleWorkload(workload5)
+	assert.NoError(t, err)
+
+	// 5.1 check service map
+	checkServiceMap(t, p, svcID, fakeSvc, 1)
+
+	// 5.2 check endpoint map
+	ek.BackendIndex = 1
+	ek.ServiceId = svcID
+	err = p.bpf.EndpointLookup(&ek, &ev)
+	assert.NoError(t, err)
+	assert.Equal(t, workload2ID, ev.BackendUid)
+
+	// 6. add namespace scoped waypoint service
 	wpSvc := createFakeService("waypoint", "10.240.10.5", "10.240.10.5")
 	_ = p.handleService(wpSvc)
 	assert.Nil(t, wpSvc.Waypoint)
-	// 4.1 check front end map contains service
+	// 6.1 check front end map contains service
 	svcID = checkFrontEndMap(t, wpSvc.Addresses[0].Address, p)
-	// 4.2 check service map contains service, but no waypoint address
+	// 6.2 check service map contains service, but no waypoint address
 	checkServiceMap(t, p, svcID, wpSvc, 0)
 
 	hashNameClean(p)
@@ -148,7 +194,7 @@ func checkServiceMap(t *testing.T, p *Processor, svcId uint32, fakeSvc *workload
 	var sv bpfcache.ServiceValue
 	err := p.bpf.ServiceLookup(&bpfcache.ServiceKey{ServiceId: svcId}, &sv)
 	assert.NoError(t, err)
-	assert.Equal(t, sv.EndpointCount, endpointCount)
+	assert.Equal(t, endpointCount, sv.EndpointCount)
 	waypointAddr := fakeSvc.GetWaypoint().GetAddress().GetAddress()
 	if waypointAddr != nil {
 		assert.Equal(t, test.EqualIp(sv.WaypointAddr, waypointAddr), true)
@@ -193,7 +239,7 @@ func checkFrontEndMap(t *testing.T, ip []byte, p *Processor) (upstreamId uint32)
 	return
 }
 
-func BenchmarkHandleDataWithService(b *testing.B) {
+func BenchmarkAddNewServicesWithWorkload(b *testing.B) {
 	t := &testing.T{}
 	config := options.BpfConfig{
 		Mode:        constants.WorkloadMode,
@@ -208,14 +254,14 @@ func BenchmarkHandleDataWithService(b *testing.B) {
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		workload := createTestWorkloadWithService()
-		err := workloadController.Processor.handleDataWithService(workload)
+		workload := createTestWorkloadWithService(true)
+		err := workloadController.Processor.handleWorkload(workload)
 		assert.NoError(t, err)
 	}
 	workloadController.Processor.hashName.Reset()
 }
 
-func createTestWorkloadWithService() *workloadapi.Workload {
+func createTestWorkloadWithService(withService bool) *workloadapi.Workload {
 	workload := workloadapi.Workload{
 		Namespace:         "ns",
 		Name:              "name",
@@ -227,7 +273,11 @@ func createTestWorkloadWithService() *workloadapi.Workload {
 		WorkloadName:      "name",
 		Status:            workloadapi.WorkloadStatus_HEALTHY,
 		ClusterId:         "cluster0",
-		Services: map[string]*workloadapi.PortList{
+		Services:          map[string]*workloadapi.PortList{},
+	}
+
+	if withService == true {
+		workload.Services = map[string]*workloadapi.PortList{
 			"default/testsvc.default.svc.cluster.local": {
 				Ports: []*workloadapi.Port{
 					{
@@ -244,7 +294,7 @@ func createTestWorkloadWithService() *workloadapi.Workload {
 					},
 				},
 			},
-		},
+		}
 	}
 	workload.Uid = "cluster0/" + rand.String(6)
 	return &workload
@@ -329,8 +379,9 @@ func Test_deleteWorkloadWithRestart(t *testing.T) {
 
 	// 1. handle workload with service, but service not handled yet
 	// In this case, only frontend map and backend map should be updated.
-	wl := createTestWorkloadWithService()
-	_ = p.handleDataWithService(createTestWorkloadWithService())
+	wl := createTestWorkloadWithService(true)
+	err := p.handleWorkload(wl)
+	assert.NoError(t, err)
 
 	workloadID := checkFrontEndMap(t, wl.Addresses[0], p)
 	checkBackendMap(t, p, workloadID, wl)
diff --git a/pkg/status/status_server_test.go b/pkg/status/status_server_test.go
index 98b24f35..c06cb2c2 100644
--- a/pkg/status/status_server_test.go
+++ b/pkg/status/status_server_test.go
@@ -265,7 +265,7 @@ func TestServer_configDumpWorkload(t *testing.T) {
 		}}
 	fakeWorkloadCache := cache.NewWorkloadCache()
 	fakeServiceCache := cache.NewServiceCache()
-	fakeWorkloadCache.AddWorkload(w1)
+	fakeWorkloadCache.AddOrUpdateWorkload(w1)
 	fakeServiceCache.AddOrUpdateService(svc)
 	// Create a new instance of the Server struct
 	server := &Server{