daemon add the orphan ip warning

default       0s          Warning   ResourceInvalid    node/cn-hangzhou.172.16.1.196   orphan ip found on ecs metadata, ip: 172.16.64.4, restart terway to resync data

Signed-off-by: l1b0k <[email protected]>

Author: l1b0k

pkg/eni/local.go

@@ -13,6 +13,7 @@ import (
 
 	"golang.org/x/time/rate"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/cache"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
@@ -27,6 +28,8 @@ import (
 	"github.com/AliyunContainerService/terway/pkg/metric"
 )
 
+const defaultSyncPeriod = 1 * time.Minute
+
 var _ NetworkInterface = &Local{}
 var _ Usage = &Local{}
 var _ ReportStatus = &Trunk{}
@@ -177,7 +180,7 @@ func (l *Local) Run(ctx context.Context, podResources []daemon.PodResources, wg
 
 	go l.notify(ctx)
 
-	go wait.JitterUntil(l.sync, 1*time.Minute, 1.0, true, ctx.Done())
+	go wait.JitterUntil(l.sync, defaultSyncPeriod, 1.0, true, ctx.Done())
 
 	return nil
 }
@@ -370,6 +373,7 @@ func (l *Local) sync() {
 
 	syncIPLocked(l.ipv4, ipv4)
 	syncIPLocked(l.ipv6, ipv6)
+	report()
 
 	l.cond.Broadcast()
 }
@@ -1037,8 +1041,40 @@ func syncIPLocked(lo Set, remote []netip.Addr) {
 			}
 		}
 	}
+	orphanIP(lo, s)
+}
+
+func orphanIP(lo Set, remote sets.Set[netip.Addr]) {
+	for key := range remote {
+		if _, ok := lo[key]; !ok {
+
+			prev, ok := invalidIPCache.Get(key)
+			if !ok {
+				invalidIPCache.Add(key, 1, 5*defaultSyncPeriod)
+			} else {
+				invalidIPCache.Add(key, prev.(int)+1, 5*defaultSyncPeriod)
+			}
+		} else {
+			invalidIPCache.Remove(key)
+		}
+	}
+}
+
+func report() {
+	for _, key := range invalidIPCache.Keys() {
+		count, ok := invalidIPCache.Get(key)
+		if !ok {
+			continue
+		}
+		if count.(int) > 1 {
+			_ = tracing.RecordNodeEvent(corev1.EventTypeWarning, string(types.ErrResourceInvalid), fmt.Sprintf("orphan ip found on ecs metadata, ip: %s", key))
+			logf.Log.Info("orphan ip found on ecs metadata", "ip", key)
+		}
+	}
 }
 
+var invalidIPCache = cache.NewLRUExpireCache(100)
+
 func parseResourceID(id string) (string, string, error) {
 	parts := strings.SplitN(id, ".", 2)
 	if len(parts) < 2 {

pkg/eni/local_test.go

@@ -10,6 +10,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/time/rate"
+	"k8s.io/apimachinery/pkg/util/cache"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/AliyunContainerService/terway/pkg/factory"
 	"github.com/AliyunContainerService/terway/types"
@@ -309,3 +311,30 @@ func Test_parseResourceID(t *testing.T) {
 		})
 	}
 }
+
+func Test_orphanIP(t *testing.T) {
+	invalidIPCache = cache.NewLRUExpireCache(100)
+
+	lo1 := map[netip.Addr]*IP{
+		netip.MustParseAddr("127.0.0.1"): {
+			ip: netip.MustParseAddr("127.0.0.1"),
+		},
+	}
+
+	remote1 := sets.Set[netip.Addr]{
+		netip.MustParseAddr("127.0.0.1"): {},
+		netip.MustParseAddr("127.0.0.2"): {},
+	}
+
+	orphanIP(lo1, remote1)
+
+	v, _ := invalidIPCache.Get(netip.MustParseAddr("127.0.0.1"))
+	assert.Equal(t, nil, v)
+
+	v, _ = invalidIPCache.Get(netip.MustParseAddr("127.0.0.2"))
+	assert.Equal(t, 1, v)
+
+	orphanIP(lo1, remote1)
+	v, _ = invalidIPCache.Get(netip.MustParseAddr("127.0.0.2"))
+	assert.Equal(t, 2, v)
+}

pkg/eni/types.go

@@ -94,7 +94,7 @@ func (ip *IP) Allocatable() bool {
 	return ip.Valid() && !ip.InUse()
 }
 
-type Set map[any]*IP
+type Set map[netip.Addr]*IP
 
 func (s Set) Idles() []*IP {
 	var result []*IP