Merge pull request #687 from l1b0k/release-1.8

backport fix for 1.8

Author: BSWANG

Diff for: 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 {

Diff for: 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)
+}

Diff for: 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

Diff for: pkg/factory/aliyun/aliyun.go

@@ -191,6 +191,19 @@ func (a *Aliyun) CreateNetworkInterface(ipv4, ipv6 int, eniType string) (*daemon
 		return r, nil, nil, err
 	}
 
+	// wait mac
+	err = wait.PollUntilContextTimeout(ctx, metadataPollInterval, metadataWaitTimeout, true, func(ctx context.Context) (bool, error) {
+		macs, err := metadata.GetENIsMAC()
+		if err != nil {
+			klog.Errorf("metadata: error get mac: %v", err)
+			return false, nil
+		}
+		return sets.NewString(macs...).Has(r.MAC), nil
+	})
+	if err != nil {
+		return r, nil, nil, err
+	}
+
 	prefix, err := metadata.GetVSwitchCIDR(eni.MacAddress)
 	if err != nil {
 		return r, nil, nil, err

Diff for: plugin/terway/cni.go

@@ -2,13 +2,16 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"runtime"
 	"time"
 
 	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/credentials/insecure"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/util/retry"
 
 	"github.com/AliyunContainerService/terway/pkg/link"
 	"github.com/AliyunContainerService/terway/plugin/datapath"
@@ -298,7 +301,17 @@ func parseSetupConf(args *skel.CmdArgs, alloc *rpc.NetConf, conf *types.CNIConf,
 	if alloc.GetENIInfo() != nil {
 		mac := alloc.GetENIInfo().GetMAC()
 		if mac != "" {
-			deviceID, err = link.GetDeviceNumber(mac)
+			err = retry.OnError(wait.Backoff{
+				Steps:    10,
+				Duration: 1 * time.Second,
+				Factor:   1.0,
+				Jitter:   0,
+			}, func(err error) bool {
+				return errors.Is(err, link.ErrNotFound)
+			}, func() error {
+				deviceID, err = link.GetDeviceNumber(mac)
+				return err
+			})
 			if err != nil {
 				return nil, err
 			}