enhance: iptables support

pkg/apis/ctrlmesh/types.go

@@ -31,11 +31,13 @@ const (
 
 // Labels
 const (
-	CtrlmeshControlPrefix                 = "ctrlmesh.kusionstack.io/"
-	CtrlmeshIgnoreWebhookLabel            = "ctrlmesh.kusionstack.io/ignore-webhook"
-	CtrlmeshIgnoreValidateLabel           = "ctrlmesh.kusionstack.io/ignore-validate"
-	CtrlmeshDefaultReplicasLabel          = "ctrlmesh.kusionstack.io/default-replicas"
-	CtrlmeshEnableProxyLabel              = "ctrlmesh.kusionstack.io/enable-proxy"
+	CtrlmeshControlPrefix        = "ctrlmesh.kusionstack.io/"
+	CtrlmeshIgnoreWebhookLabel   = "ctrlmesh.kusionstack.io/ignore-webhook"
+	CtrlmeshIgnoreValidateLabel  = "ctrlmesh.kusionstack.io/ignore-validate"
+	CtrlmeshDefaultReplicasLabel = "ctrlmesh.kusionstack.io/default-replicas"
+	CtrlmeshEnableProxyLabel     = "ctrlmesh.kusionstack.io/enable-proxy"
+	CtrlmeshEnableIptableMode    = "ctrlmesh.kusionstack.io/enable-iptables"
+
 	CtrlmeshAutoShardingRootLabel         = "ctrlmesh.kusionstack.io/auto-sharding-root"
 	CtrlmeshInRollingLabel                = "ctrlmesh.kusionstack.io/rolling"
 	CtrlmeshDisableFakeKubeconfigArgLabel = "ctrlmesh.kusionstack.io/disable-fake-kubeconfig-arg"

pkg/cmd/proxy/main.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io/ioutil"
 	"net"
 	"net/http"
 	"os"
@@ -35,7 +36,6 @@ import (
 
 	"github.com/KusionStack/controller-mesh/pkg/apis/ctrlmesh/constants"
 	"github.com/KusionStack/controller-mesh/pkg/client"
-
 	proxyapiserver "github.com/KusionStack/controller-mesh/pkg/proxy/apiserver"
 	proxycache "github.com/KusionStack/controller-mesh/pkg/proxy/cache"
 	"github.com/KusionStack/controller-mesh/pkg/proxy/circuitbreaker"
@@ -56,6 +56,8 @@ var (
 	webhookCertDir     = flag.String(constants.ProxyWebhookCertDirFlag, "", "The directory where the webhook certs generated or mounted.")
 
 	proxyIptablePort = flag.Int(constants.ProxyIptablesFlag, constants.ProxyIptablesPort, "port that http-tproxy listens on")
+
+	enableIpTable = os.Getenv(constants.EnvIPTable) == "true"
 )
 
 func main() {
@@ -66,8 +68,18 @@ func main() {
 		klog.Fatalf("Environment %s=%s %s=%s not exist.",
 			constants.EnvPodNamespace, os.Getenv(constants.EnvPodNamespace), constants.EnvPodName, os.Getenv(constants.EnvPodName))
 	}
-	cfg := ctrl.GetConfigOrDie()
-	cfg.UserAgent = "ctrlmesh"
+	var cfg *rest.Config
+
+	if enableIpTable {
+		var err error
+		cfg, err = getRestConfig()
+		if err != nil {
+			klog.Fatalf("Failed to get rest config: %v", err)
+		}
+	} else {
+		cfg = ctrl.GetConfigOrDie()
+	}
+	//cfg.UserAgent = "ctrlmesh"
 	if err := client.NewRegistry(cfg); err != nil {
 		klog.Fatalf("Failed to new client registry: %v", err)
 	}
@@ -165,3 +177,41 @@ func serveHTTP(ctx context.Context, readyHandler *healthz.Handler) {
 		klog.Fatalf("Serve HTTP shutting down on :%d: %v", *metricsHealthPort, err)
 	}
 }
+
+func getRestConfig() (*rest.Config, error) {
+	const (
+		tokenFile = "/var/run/secrets/kubernetes.io/serviceaccount/token"
+		//rootCAFile = "/var/run/secrets/kubernetes.io/serviceaccount/..data/ca.crt"
+	)
+	host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
+	if len(host) == 0 || len(port) == 0 {
+		return nil, rest.ErrNotInCluster
+	}
+
+	token, err := ioutil.ReadFile(tokenFile)
+	if err != nil {
+		return nil, err
+	}
+
+	tlsClientConfig := rest.TLSClientConfig{Insecure: true}
+
+	//if _, err := certutil.NewPool(rootCAFile); err != nil {
+	//	klog.Errorf("Expected to load root CA config from %s, but got err: %v", rootCAFile, err)
+	//} else {
+	//	tlsClientConfig.CAFile = rootCAFile
+	//}
+
+	cfg := &rest.Config{
+		// TODO: switch to using cluster DNS.
+		Host:            "https://" + net.JoinHostPort(host, port),
+		TLSClientConfig: tlsClientConfig,
+		BearerToken:     string(token),
+		BearerTokenFile: tokenFile,
+
+		Burst: 3000,
+		QPS:   2000.0,
+	}
+	klog.V(3).Infof("Starting with rest config: %v", utils.DumpJSON(cfg))
+
+	return cfg, nil
+}

pkg/webhook/pod/injector.go

@@ -131,7 +131,7 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 	if *proxyImage == "" {
 		return fmt.Errorf("the images for ControllerMesh init or proxy container have not set in args")
 	}
-
+	enableIpTable := pod.Labels[ctrlmesh.CtrlmeshEnableIptableMode] == "true"
 	imagePullPolicy := v1.PullAlways
 	if *proxyImagePullPolicy == string(v1.PullIfNotPresent) {
 		imagePullPolicy = v1.PullIfNotPresent
@@ -177,6 +177,13 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 		},
 	}
 
+	if enableIpTable {
+		proxyContainer.Env = append(proxyContainer.Env, v1.EnvVar{
+			Name:  constants.EnvIPTable,
+			Value: "true",
+		})
+	}
+
 	if val, ok := pod.Annotations[ctrlmesh.CtrlmeshProxyContainerResourceAnno]; ok {
 		req := &v1.ResourceRequirements{}
 		if err := json.Unmarshal([]byte(val), req); err != nil {
@@ -213,15 +220,6 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 		proxyContainer.Env = append(proxyContainer.Env, apiserverHostPortEnvs...)
 	}
 
-	ipTableEnvs := getEnv(pod, constants.EnvIPTable)
-	enableIpTable := false
-	if len(ipTableEnvs) > 0 {
-		initContainer.Env = append(initContainer.Env, ipTableEnvs...)
-		//proxyContainer.Env = append(proxyContainer.Env, ipTableEnvs...)
-		if ipTableEnvs[0].Value == "true" {
-			enableIpTable = true
-		}
-	}
 	if !enableIpTable {
 		if err := h.applyFakeConfigMap(pod); err != nil {
 			return err
@@ -271,7 +269,7 @@ func (h *MutatingHandler) injectByShardingConfig(ctx context.Context, pod *v1.Po
 			proxyContainer.VolumeMounts = append(proxyContainer.VolumeMounts, certVolumeMounts[0])
 		}
 	}
-	if *initImage != "" {
+	if enableIpTable && *initImage != "" {
 		pod.Spec.InitContainers = append([]v1.Container{*initContainer}, pod.Spec.InitContainers...)
 	}
 	if pod.Labels == nil {