Merge pull request #45622 from shyamjvs/mock-proxy

Automatic merge from submit-queue (batch tested with PRs 45571, 45657, 45638, 45663, 45622)

Use real proxier inside hollow-proxy but with mocked syscalls

Fixes https://github.com/kubernetes/kubernetes/issues/43701

This should make hollow-proxy better mimic the real kube-proxy in performance.
Maybe next we should have a more realistic implementation even for fake iptables (adding/updating/deleting rules/chains in an table, just not on the real one)? Though I'm not sure how important it is.

cc @kubernetes/sig-scalability-misc @kubernetes/sig-network-misc @wojtek-t @gmarek

cmd/kubemark/BUILD

@@ -27,7 +27,9 @@ go_library(
         "//pkg/kubelet/cm:go_default_library",
         "//pkg/kubelet/dockershim/libdocker:go_default_library",
         "//pkg/kubemark:go_default_library",
+        "//pkg/util/exec:go_default_library",
         "//pkg/util/iptables/testing:go_default_library",
+        "//pkg/util/sysctl/testing:go_default_library",
         "//pkg/version/prometheus:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/github.com/spf13/pflag:go_default_library",

cmd/kubemark/hollow-node.go

@@ -35,7 +35,9 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/cm"
 	"k8s.io/kubernetes/pkg/kubelet/dockershim/libdocker"
 	"k8s.io/kubernetes/pkg/kubemark"
+	fakeexec "k8s.io/kubernetes/pkg/util/exec"
 	fakeiptables "k8s.io/kubernetes/pkg/util/iptables/testing"
+	fakesysctl "k8s.io/kubernetes/pkg/util/sysctl/testing"
 	_ "k8s.io/kubernetes/pkg/version/prometheus" // for version metric registration
 
 	"github.com/golang/glog"
@@ -112,7 +114,6 @@ func main() {
 	if config.Morph == "kubelet" {
 		cadvisorInterface := new(cadvisortest.Fake)
 		containerManager := cm.NewStubContainerManager()
-
 		fakeDockerClient := libdocker.NewFakeDockerClient().WithTraceDisabled()
 		fakeDockerClient.EnableSleep = true
 
@@ -131,24 +132,29 @@ func main() {
 	}
 
 	if config.Morph == "proxy" {
-		eventBroadcaster := record.NewBroadcaster()
-		recorder := eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "kube-proxy", Host: config.NodeName})
-
-		iptInterface := fakeiptables.NewFake()
-
 		eventClient, err := clientgoclientset.NewForConfig(clientConfig)
 		if err != nil {
 			glog.Fatalf("Failed to create API Server client: %v", err)
 		}
+		iptInterface := fakeiptables.NewFake()
+		sysctl := fakesysctl.NewFake()
+		execer := &fakeexec.FakeExec{}
+		eventBroadcaster := record.NewBroadcaster()
+		recorder := eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "kube-proxy", Host: config.NodeName})
 
-		hollowProxy := kubemark.NewHollowProxyOrDie(
+		hollowProxy, err := kubemark.NewHollowProxyOrDie(
 			config.NodeName,
 			internalClientset,
 			eventClient,
 			iptInterface,
+			sysctl,
+			execer,
 			eventBroadcaster,
 			recorder,
 		)
+		if err != nil {
+			glog.Fatalf("Failed to create hollowProxy instance: %v", err)
+		}
 		hollowProxy.Run()
 	}
 }

pkg/kubemark/BUILD

@@ -29,11 +29,15 @@ go_library(
         "//pkg/kubelet/container/testing:go_default_library",
         "//pkg/kubelet/dockershim/libdocker:go_default_library",
         "//pkg/kubelet/types:go_default_library",
+        "//pkg/proxy/iptables:go_default_library",
         "//pkg/util:go_default_library",
+        "//pkg/util/exec:go_default_library",
         "//pkg/util/io:go_default_library",
         "//pkg/util/iptables:go_default_library",
         "//pkg/util/mount:go_default_library",
+        "//pkg/util/node:go_default_library",
         "//pkg/util/oom:go_default_library",
+        "//pkg/util/sysctl:go_default_library",
         "//pkg/volume/empty_dir:go_default_library",
         "//pkg/volume/secret:go_default_library",
         "//test/utils:go_default_library",

pkg/kubemark/hollow_proxy.go

@@ -17,17 +17,23 @@ limitations under the License.
 package kubemark
 
 import (
+	"fmt"
+	"net"
 	"time"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
 	clientv1 "k8s.io/client-go/pkg/api/v1"
 	"k8s.io/client-go/tools/record"
 	proxyapp "k8s.io/kubernetes/cmd/kube-proxy/app"
-	"k8s.io/kubernetes/pkg/api"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
+	"k8s.io/kubernetes/pkg/proxy/iptables"
 	"k8s.io/kubernetes/pkg/util"
+	utilexec "k8s.io/kubernetes/pkg/util/exec"
 	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
+	nodeutil "k8s.io/kubernetes/pkg/util/node"
+	utilsysctl "k8s.io/kubernetes/pkg/util/sysctl"
 
 	"github.com/golang/glog"
 )
@@ -36,32 +42,35 @@ type HollowProxy struct {
 	ProxyServer *proxyapp.ProxyServer
 }
 
-type FakeProxyHandler struct{}
-
-func (*FakeProxyHandler) OnServiceAdd(service *api.Service)                        {}
-func (*FakeProxyHandler) OnServiceUpdate(oldService, service *api.Service)         {}
-func (*FakeProxyHandler) OnServiceDelete(service *api.Service)                     {}
-func (*FakeProxyHandler) OnServiceSynced()                                         {}
-func (*FakeProxyHandler) OnEndpointsAdd(endpoints *api.Endpoints)                  {}
-func (*FakeProxyHandler) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {}
-func (*FakeProxyHandler) OnEndpointsDelete(endpoints *api.Endpoints)               {}
-func (*FakeProxyHandler) OnEndpointsSynced()                                       {}
-
-type FakeProxier struct{}
-
-func (*FakeProxier) Sync() {}
-func (*FakeProxier) SyncLoop() {
-	select {}
-}
-
 func NewHollowProxyOrDie(
 	nodeName string,
 	client clientset.Interface,
 	eventClient v1core.EventsGetter,
 	iptInterface utiliptables.Interface,
+	sysctl utilsysctl.Interface,
+	execer utilexec.Interface,
 	broadcaster record.EventBroadcaster,
 	recorder record.EventRecorder,
-) *HollowProxy {
+) (*HollowProxy, error) {
+	// Create a proxier with fake iptables underneath it.
+	proxier, err := iptables.NewProxier(
+		iptInterface,
+		sysctl,
+		execer,
+		30*time.Second,
+		5*time.Second,
+		false,
+		0,
+		"10.0.0.0/8",
+		nodeName,
+		getNodeIP(client, nodeName),
+		recorder,
+		nil,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create proxier: %v", err)
+	}
+
 	// Create and start Hollow Proxy
 	nodeRef := &clientv1.ObjectReference{
 		Kind:      "Node",
@@ -75,7 +84,7 @@ func NewHollowProxyOrDie(
 			Client:                client,
 			EventClient:           eventClient,
 			IptInterface:          iptInterface,
-			Proxier:               &FakeProxier{},
+			Proxier:               proxier,
 			Broadcaster:           broadcaster,
 			Recorder:              recorder,
 			ProxyMode:             "fake",
@@ -83,10 +92,10 @@ func NewHollowProxyOrDie(
 			OOMScoreAdj:           util.Int32Ptr(0),
 			ResourceContainer:     "",
 			ConfigSyncPeriod:      30 * time.Second,
-			ServiceEventHandler:   &FakeProxyHandler{},
+			ServiceEventHandler:   proxier,
-			EndpointsEventHandler: &FakeProxyHandler{},
+			EndpointsEventHandler: proxier,
 		},
-	}
+	}, nil
 }
 
 func (hp *HollowProxy) Run() {
@@ -94,3 +103,18 @@ func (hp *HollowProxy) Run() {
 		glog.Fatalf("Error while running proxy: %v\n", err)
 	}
 }
+
+func getNodeIP(client clientset.Interface, hostname string) net.IP {
+	var nodeIP net.IP
+	node, err := client.Core().Nodes().Get(hostname, metav1.GetOptions{})
+	if err != nil {
+		glog.Warningf("Failed to retrieve node info: %v", err)
+		return nil
+	}
+	nodeIP, err = nodeutil.InternalGetNodeHostIP(node)
+	if err != nil {
+		glog.Warningf("Failed to retrieve node IP: %v", err)
+		return nil
+	}
+	return nodeIP
+}

pkg/proxy/iptables/proxier.go

@@ -356,7 +356,7 @@ func NewProxier(ipt utiliptables.Interface,
 ) (*Proxier, error) {
 	// check valid user input
 	if minSyncPeriod > syncPeriod {
-		return nil, fmt.Errorf("min-sync (%v) must be < sync(%v)", minSyncPeriod, syncPeriod)
+		return nil, fmt.Errorf("min-sync (%v) must be <= sync(%v)", minSyncPeriod, syncPeriod)
 	}
 
 	// Set the route_localnet sysctl we need for