trace iptables

iptables_trace_test.go

/*
Copyright 2021 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package iptables

import (
	"fmt"
	"log"
	"net"
	"os"
	"runtime"
	"runtime/pprof"
	"testing"
	"time"

	v1 "k8s.io/api/core/v1"
	discovery "k8s.io/api/discovery/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/component-base/metrics/testutil"
	"k8s.io/kubernetes/pkg/proxy/metrics"

	utiliptables "k8s.io/kubernetes/pkg/util/iptables"
	iptablestest "k8s.io/kubernetes/pkg/util/iptables/testing"

	utilnet "k8s.io/utils/net"
	utilpointer "k8s.io/utils/pointer"
)

func TestNumberIptablesRules(t *testing.T) {
	numServices := 1000
	numEndpoints := 1000
	i := numServices
	j := numEndpoints
	svcType := v1.ServiceTypeLoadBalancer
	t.Run(fmt.Sprintf("%s-%d-%d", svcType, i, j), func(t *testing.T) {
		ipt := iptablestest.NewFake()
		fp := NewFakeProxier(ipt)
		metrics.RegisterMetrics()
		// t.Logf("Generate %d Services Type %s with %d endpoints each", i, svcType, j)
		svcs, eps := generateServiceEndpoints(svcType, i, j)

		// t.Logf("Updating kube-proxy")
		makeServiceMap(fp, svcs...)
		populateEndpointSlices(fp, eps...)

		// t.Logf("Syncing kube-proxy")
		start := time.Now()
		//f, err := os.Create("trace.out")
		//if err != nil {
		//	log.Fatalf("failed to create trace output file: %v", err)
		//	}
		//		defer func() {
		//			if err := f.Close(); err != nil {
		//				log.Fatalf("failed to close trace file: %v", err)
		//			}
		//		}()

		f2, err := os.Create("mem.out")
		if err != nil {
			log.Fatal("could not create memory profile: ", err)
		}
		runtime.GC() // get up-to-date statistics
		if err := pprof.WriteHeapProfile(f2); err != nil {
			log.Fatal("could not write memory profile: ", err)
		}

		//if err := trace.Start(f); err != nil {
		//		log.Fatalf("failed to start trace: %v", err)
		//	}

		fp.syncProxyRules()
		f2.Close()
		//trace.Stop()
		elapsed := time.Since(start)

		t.Logf("kube-proxy sync rules latency: %v", elapsed)

		t.Logf("Gathering kube-proxy metrics")
		nFilterRules, err := testutil.GetGaugeMetricValue(metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableFilter)))
		if err != nil {
			t.Errorf("failed to get %s value, err: %v", metrics.IptablesRulesTotal.Name, err)
		}

		nNatRules, err := testutil.GetGaugeMetricValue(metrics.IptablesRulesTotal.WithLabelValues(string(utiliptables.TableNAT)))
		if err != nil {
			t.Errorf("failed to get %s value, err: %v", metrics.IptablesRulesTotal.Name, err)
		}

		t.Logf("kube-proxy iptables rules: Filter %.0f NAT %.0f on %v", nFilterRules, nNatRules, elapsed)

	})

}

// generateServiceEndpoints generate
func generateServiceEndpoints(svcType v1.ServiceType, nServices, nEndpoints int) ([]*v1.Service, []*discovery.EndpointSlice) {
	services := make([]*v1.Service, nServices)
	endpoints := make([]*discovery.EndpointSlice, nServices)

	// base parameters
	basePort := 80
	base := utilnet.BigForIP(net.ParseIP("10.0.0.1"))
	// generate a base endpoint object
	baseEp := utilnet.BigForIP(net.ParseIP("172.16.0.1"))
	epPort := 8080

	tcpProtocol := v1.ProtocolTCP
	ep := &discovery.EndpointSlice{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "ep",
			Namespace: "namespace",
		},
		AddressType: discovery.AddressTypeIPv4,
		Endpoints:   []discovery.Endpoint{},
		Ports: []discovery.EndpointPort{{
			Name:     utilpointer.StringPtr(fmt.Sprintf("%d", epPort)),
			Port:     utilpointer.Int32(int32(epPort)),
			Protocol: &tcpProtocol,
		}},
	}
	addresses := []string{}
	for j := 0; j < nEndpoints; j++ {
		ipEp := utilnet.AddIPOffset(baseEp, j)
		addresses = append(addresses, ipEp.String())
	}
	ep.Endpoints = []discovery.Endpoint{{
		Addresses: addresses,
	}}
	// Create the Services and associate and endpoint object to each one
	for i := 0; i < nServices; i++ {
		ip := utilnet.AddIPOffset(base, i)
		svcName := fmt.Sprintf("svc%d", i)
		services[i] = &v1.Service{
			ObjectMeta: metav1.ObjectMeta{
				Name:      svcName,
				Namespace: "namespace",
			},
			Spec: v1.ServiceSpec{
				ClusterIP: ip.String(),
				Ports: []v1.ServicePort{{
					Name:       fmt.Sprintf("%d", epPort),
					Protocol:   v1.ProtocolTCP,
					Port:       int32(basePort + i),
					TargetPort: intstr.FromInt(epPort),
				}},
				Type: svcType,
			},
		}

		if svcType == v1.ServiceTypeNodePort || svcType == v1.ServiceTypeLoadBalancer {
			services[i].Spec.Ports[0].NodePort = int32(30000 + i)

		}
		if svcType == v1.ServiceTypeLoadBalancer {
			services[i].Status.LoadBalancer.Ingress = []v1.LoadBalancerIngress{{
				IP: "1.2.3.4",
			}}
			services[i].Spec.ExternalIPs = []string{"1.2.3.4"}
			services[i].Spec.LoadBalancerSourceRanges = []string{" 1.2.3.4/28"}
			services[i].Spec.HealthCheckNodePort = int32(32000 + i)
		}

		endpoints[i] = ep.DeepCopy()
		endpoints[i].Name = svcName
		endpoints[i].ObjectMeta.Labels = map[string]string{
			discovery.LabelServiceName: svcName,
		}

	}

	return services, endpoints
}