🔨 fix: Port IntOrString, add cni detector to add secgroup when using …

…cilium native routing

pkg/cni_detector/cni_detector.go

@@ -0,0 +1,94 @@
+package cni_detector
+
+import (
+	"context"
+
+	apimetav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/klog/v2"
+)
+
+// CNIType represents the CNI type detected in the cluster.
+type CNIType string
+
+const (
+	CalicoOverlay       CNIType = "Calico Overlay"
+	CiliumOverlay       CNIType = "Cilium Overlay"
+	CiliumNativeRouting CNIType = "Cilium Native Routing"
+	UnknownCNI          CNIType = "Unknown CNI"
+)
+
+// Detector detects the CNI type used in the Kubernetes cluster.
+type Detector struct {
+	kubeClient kubernetes.Interface
+}
+
+// NewDetector creates a new instance of the CNI Detector.
+func NewDetector(kubeClient kubernetes.Interface) *Detector {
+	return &Detector{kubeClient: kubeClient}
+}
+
+// DetectCNIType detects the CNI type in the cluster.
+func (d *Detector) DetectCNIType() (CNIType, error) {
+	// Check for Calico
+	if d.isCalicoOverlay() {
+		return CalicoOverlay, nil
+	}
+
+	// Check for Cilium
+	if d.isCiliumNativeRouting() {
+		return CiliumNativeRouting, nil
+	}
+
+	if d.isCiliumOverlay() {
+		return CiliumOverlay, nil
+	}
+
+	return UnknownCNI, nil
+}
+
+// Check if Calico Overlay is running
+func (d *Detector) isCalicoOverlay() bool {
+	calicoNodeDaemonSet, err := d.kubeClient.AppsV1().DaemonSets("kube-system").Get(context.TODO(), "calico-node", apimetav1.GetOptions{})
+	if err != nil {
+		klog.Errorf("Error getting Calico DaemonSet: %v", err)
+		return false
+	}
+	if calicoNodeDaemonSet == nil {
+		return false
+	}
+	return true
+}
+
+// Check if Cilium Overlay is running
+func (d *Detector) isCiliumOverlay() bool {
+	ciliumDaemonSet, err := d.kubeClient.AppsV1().DaemonSets("kube-system").Get(context.TODO(), "cilium", apimetav1.GetOptions{})
+	if err != nil {
+		klog.Errorf("Error getting Cilium DaemonSet: %v", err)
+		return false
+	}
+	if ciliumDaemonSet == nil {
+		return false
+	}
+	return true
+}
+
+// Check if Cilium Native Routing is running
+func (d *Detector) isCiliumNativeRouting() bool {
+	if d.isCiliumOverlay() {
+		// get cilium-config config map
+		ciliumConfigMap, err := d.kubeClient.CoreV1().ConfigMaps("kube-system").Get(context.TODO(), "cilium-config", apimetav1.GetOptions{})
+		if err != nil {
+			klog.Errorf("Error getting Cilium ConfigMap: %v", err)
+			return false
+		}
+		if ciliumConfigMap == nil {
+			return false
+		}
+		// check if cilium-config have routing-mode: native
+		if ciliumConfigMap.Data["routing-mode"] == "native" {
+			return true
+		}
+	}
+	return false
+}

pkg/cni_detector/cni_detector_test.go

@@ -0,0 +1,125 @@
+package cni_detector
+
+import (
+	"context"
+	"testing"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func TestDetectCNIType_CalicoOverlay(t *testing.T) {
+	// Set up a fake Kubernetes client with Calico DaemonSet
+	client := fake.NewSimpleClientset()
+	client.AppsV1().DaemonSets("kube-system").Create(context.TODO(), &appsv1.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "calico-node",
+			Namespace: "kube-system",
+		},
+	}, metav1.CreateOptions{})
+
+	detector := &Detector{kubeClient: client}
+	cniType, err := detector.DetectCNIType()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if cniType != CalicoOverlay {
+		t.Errorf("expected CNI type %s, got %s", CalicoOverlay, cniType)
+	}
+}
+
+func TestDetectCNIType_CiliumOverlay(t *testing.T) {
+	// Set up a fake Kubernetes client with Cilium DaemonSet
+	client := fake.NewSimpleClientset()
+	client.AppsV1().DaemonSets("kube-system").Create(context.TODO(), &appsv1.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cilium",
+			Namespace: "kube-system",
+		},
+	}, metav1.CreateOptions{})
+
+	detector := &Detector{kubeClient: client}
+	cniType, err := detector.DetectCNIType()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if cniType != CiliumOverlay {
+		t.Errorf("expected CNI type %s, got %s", CiliumOverlay, cniType)
+	}
+}
+
+func TestDetectCNIType_CiliumNativeRouting(t *testing.T) {
+	// Set up a fake Kubernetes client with Cilium DaemonSet and config map
+	client := fake.NewSimpleClientset()
+
+	// Create the Cilium DaemonSet
+	client.AppsV1().DaemonSets("kube-system").Create(context.TODO(), &appsv1.DaemonSet{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cilium",
+			Namespace: "kube-system",
+		},
+	}, metav1.CreateOptions{})
+
+	// Create the Cilium ConfigMap with routing-mode: native
+	client.CoreV1().ConfigMaps("kube-system").Create(context.TODO(), &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "cilium-config",
+			Namespace: "kube-system",
+		},
+		Data: map[string]string{
+			"routing-mode": "native",
+		},
+	}, metav1.CreateOptions{})
+
+	detector := &Detector{kubeClient: client}
+	cniType, err := detector.DetectCNIType()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if cniType != CiliumNativeRouting {
+		t.Errorf("expected CNI type %s, got %s", CiliumNativeRouting, cniType)
+	}
+}
+
+func TestDetectCNIType_UnknownCNI(t *testing.T) {
+	// Set up a fake Kubernetes client without any CNI configuration
+	client := fake.NewSimpleClientset()
+
+	detector := &Detector{kubeClient: client}
+	cniType, err := detector.DetectCNIType()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if cniType != UnknownCNI {
+		t.Errorf("expected CNI type %s, got %s", UnknownCNI, cniType)
+	}
+}
+
+func TestRealCluster(t *testing.T) {
+	// configPath := "/home/annd2/Downloads/annd2-clean.txt"
+	configPath := "/home/annd2/Downloads/cluster-43dd8bab-fd8.txt"
+	if configPath == "" {
+		t.Skip("Skipping test; no kubeconfig provided")
+	}
+	// init new kubernetes client
+
+	config, err := clientcmd.BuildConfigFromFlags("", configPath)
+	if err != nil {
+		t.Fatalf("failed to create Kubernetes client: %v", err)
+	}
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		t.Fatalf("failed to create Kubernetes client: %v", err)
+	}
+
+	detector := NewDetector(clientset)
+	cniType, err := detector.DetectCNIType()
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	t.Logf("Detected CNI type: %s", cniType)
+}