feat: remove iptables dependency

cloud/pkg/cloudstream/cloudstream.go

@@ -42,7 +42,7 @@ func newCloudStream(enable bool, tunnelPort int) *cloudStream {
 
 func Register(controller *v1alpha1.CloudStream, commonConfig *v1alpha1.CommonConfig) {
 	config.InitConfigure(controller)
-	core.Register(newCloudStream(controller.Enable, commonConfig.TunnelPort))
+	core.Register(newCloudStream(controller.Enable, int(controller.StreamPort)))
 }
 
 func (s *cloudStream) Name() string {

cloud/pkg/cloudstream/tunnelserver.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/emicklei/go-restful"
 	"github.com/gorilla/websocket"
+	"github.com/kubeedge/kubeedge/cloud/pkg/common/messagelayer"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	certutil "k8s.io/client-go/util/cert"
@@ -203,6 +204,7 @@ func (s *TunnelServer) updateNodeKubeletEndpoint(nodeName string) error {
 			return false, nil
 		}
 
+		getNode = messagelayer.HijackInternalIP(getNode)
 		getNode.Status.DaemonEndpoints.KubeletEndpoint.Port = int32(s.tunnelPort)
 		_, err = client.GetKubeClient().CoreV1().Nodes().UpdateStatus(context.Background(), getNode, metav1.UpdateOptions{})
 		if err != nil {

cloud/pkg/common/messagelayer/util.go

@@ -21,11 +21,11 @@ import (
 	"fmt"
 	"strings"
 
-	"k8s.io/klog/v2"
-
 	"github.com/kubeedge/beehive/pkg/core/model"
 	"github.com/kubeedge/kubeedge/common/constants"
 	pkgutil "github.com/kubeedge/kubeedge/pkg/util"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2"
 )
 
 const (
@@ -147,3 +147,61 @@ func GetResourceTypeForDevice(resource string) (string, error) {
 
 	return "", fmt.Errorf("unknown resource, found: %s", resource)
 }
+
+func HijackInternalIP(node *v1.Node) *v1.Node {
+	// TODO POD IP or SVC IP
+	podIP, err := pkgutil.GetLocalIP(pkgutil.GetHostname())
+	if err != nil {
+		klog.Errorf("Failed to get Local IP address: %v", err)
+		return node
+	}
+	klog.Infoln("pod ip", podIP)
+
+	internalIP := ""
+	for i, address := range node.Status.Addresses {
+		if address.Type == v1.NodeInternalIP {
+			internalIP = node.Status.Addresses[i].Address
+			node.Status.Addresses[i].Address = podIP
+			break
+		}
+	}
+	if node.Status.DaemonEndpoints.KubeletEndpoint.Port == 0 {
+		node.Status.DaemonEndpoints.KubeletEndpoint.Port = 10003
+	}
+	if node.GetAnnotations() != nil {
+		node.Annotations["kubeedge.io/internal-ip"] = internalIP
+	} else {
+		node.Annotations = make(map[string]string)
+		node.Annotations["kubeedge.io/internal-ip"] = internalIP
+	}
+	return node
+}
+
+func RegainInternalIP(node *v1.Node) *v1.Node {
+	if node.GetAnnotations() != nil {
+		if val, ok := node.Annotations["kubeedge.io/internal-ip"]; ok {
+			SetInternalIP(node, val)
+		}
+	}
+	return node
+}
+
+func SetInternalIP(node *v1.Node, target string) {
+	for i, address := range node.Status.Addresses {
+		if address.Type == v1.NodeInternalIP {
+			node.Status.Addresses[i].Address = target
+			break
+		}
+	}
+}
+
+func GetInternalIP(node *v1.Node) string {
+	internalIP := ""
+	for i, address := range node.Status.Addresses {
+		if address.Type == v1.NodeInternalIP {
+			internalIP = node.Status.Addresses[i].Address
+			break
+		}
+	}
+	return internalIP
+}

cloud/pkg/edgecontroller/controller/upstream.go

@@ -26,6 +26,7 @@ we grab some functions from `kubelet/status/status_manager.go and do some modifi
 package controller
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	stderrors "errors"
@@ -34,19 +35,6 @@ import (
 	"strings"
 	"time"
 
-	authenticationv1 "k8s.io/api/authentication/v1"
-	coordinationv1 "k8s.io/api/coordination/v1"
-	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
-	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/runtime"
-	apimachineryType "k8s.io/apimachinery/pkg/types"
-	k8sinformer "k8s.io/client-go/informers"
-	"k8s.io/client-go/kubernetes"
-	coordinationlisters "k8s.io/client-go/listers/coordination/v1"
-	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/klog/v2"
-
 	beehiveContext "github.com/kubeedge/beehive/pkg/core/context"
 	"github.com/kubeedge/beehive/pkg/core/model"
 	"github.com/kubeedge/kubeedge/cloud/pkg/common/client"
@@ -62,6 +50,19 @@ import (
 	rulesv1 "github.com/kubeedge/kubeedge/pkg/apis/rules/v1"
 	crdClientset "github.com/kubeedge/kubeedge/pkg/client/clientset/versioned"
 	"github.com/kubeedge/kubeedge/pkg/metaserver/util"
+	pkgutil "github.com/kubeedge/kubeedge/pkg/util"
+	authenticationv1 "k8s.io/api/authentication/v1"
+	coordinationv1 "k8s.io/api/coordination/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	apimachineryType "k8s.io/apimachinery/pkg/types"
+	k8sinformer "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	coordinationlisters "k8s.io/client-go/listers/coordination/v1"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"k8s.io/klog/v2"
 )
 
 // SortedContainerStatuses define A type to help sort container statuses based on container names.
@@ -578,6 +579,7 @@ func (uc *UpstreamController) updateNodeStatus() {
 				}
 
 				getNode.Status = nodeStatusRequest.Status
+				getNode = messagelayer.HijackInternalIP(getNode)
 
 				node, err := uc.kubeClient.CoreV1().Nodes().UpdateStatus(context.Background(), getNode, metaV1.UpdateOptions{})
 				if err != nil {
@@ -632,7 +634,9 @@ func kubeClientGet(uc *UpstreamController, namespace string, name string, queryT
 	case common.ResourceTypeVolumeAttachment:
 		obj, err = uc.kubeClient.StorageV1().VolumeAttachments().Get(context.Background(), name, metaV1.GetOptions{})
 	case model.ResourceTypeNode:
-		obj, err = uc.nodeLister.Get(name)
+		var node *v1.Node
+		node, err = uc.nodeLister.Get(name)
+		obj = messagelayer.RegainInternalIP(node)
 	case model.ResourceTypeServiceAccountToken:
 		obj, err = uc.getServiceAccountToken(namespace, name, msg)
 	case model.ResourceTypeLease:
@@ -846,6 +850,7 @@ func (uc *UpstreamController) registerNode() {
 			if err != nil {
 				klog.Errorf("create node %s error: %v , register node failed", name, err)
 			}
+			resp = messagelayer.RegainInternalIP(resp)
 
 			resMsg := model.NewMessage(msg.GetID()).
 				FillBody(&ObjectResp{Object: resp, Err: err}).
@@ -867,7 +872,7 @@ func (uc *UpstreamController) patchNode() {
 			klog.Warning("stop patchNode")
 			return
 		case msg := <-uc.patchNodeChan:
-			klog.V(5).Infof("message: %s, operation is: %s, and resource is %s", msg.GetID(), msg.GetOperation(), msg.GetResource())
+			klog.Infof("message: %s, operation is: %s, and resource is %s", msg.GetID(), msg.GetOperation(), msg.GetResource())
 
 			namespace, err := messagelayer.GetNamespace(msg)
 			if err != nil {
@@ -886,10 +891,29 @@ func (uc *UpstreamController) patchNode() {
 				continue
 			}
 
+			podIP, err := pkgutil.GetLocalIP(pkgutil.GetHostname())
+			if err != nil {
+				klog.Errorf("Failed to get Local IP address: %v", err)
+				continue
+			}
+
+			nodeToPatch := &v1.Node{}
+			err = json.Unmarshal(patchBytes, nodeToPatch)
+			if err != nil {
+				klog.Warningf("message: %s process failure, unmarshal msg data failed with error: %v", msg.GetID(), err)
+				continue
+			}
+			originIP := messagelayer.GetInternalIP(nodeToPatch)
+
+			if originIP != "" {
+				patchBytes = bytes.ReplaceAll(patchBytes, []byte(originIP), []byte(podIP))
+			}
+
 			node, err := uc.kubeClient.CoreV1().Nodes().Patch(context.TODO(), name, apimachineryType.StrategicMergePatchType, patchBytes, metaV1.PatchOptions{}, "status")
 			if err != nil {
 				klog.Errorf("message: %s process failure, patch node failed with error: %v, namespace: %s, name: %s", msg.GetID(), err, namespace, name)
 			}
+			messagelayer.SetInternalIP(node, originIP)
 
 			resMsg := model.NewMessage(msg.GetID()).
 				SetResourceVersion(node.ResourceVersion).