Multus CNI Testing Strategy QA

Multus CNI Testing Strategy QA คืออะไร

Multus CNI เป็น meta-plugin สำหรับ Kubernetes ที่ช่วยให้ pods มี network interfaces หลายตัวพร้อมกัน นอกเหนือจาก default cluster network ปกติ รองรับ SR-IOV, Macvlan, IPVLAN, Bridge และ CNI plugins อื่นๆ Testing Strategy QA คือกลยุทธ์การทดสอบเพื่อรับรองคุณภาพของระบบ ครอบคลุม unit tests, integration tests, end-to-end tests และ performance tests การรวมสองแนวคิดนี้ช่วยให้ multi-network Kubernetes deployments มีความน่าเชื่อถือสูง ตรวจจับปัญหา networking ก่อน production

Multus CNI Architecture

# multus_arch.py — Multus CNI architecture overview
import json

class MultusArchitecture:
    CONCEPTS = {
        "multus": {
            "name": "Multus CNI",
            "description": "Meta-plugin ที่ chain CNI plugins หลายตัว — pod ได้หลาย network interfaces",
            "use_cases": "Telco/5G (user plane + control plane), NFV, storage networks, monitoring networks",
        },
        "net_attach_def": {
            "name": "NetworkAttachmentDefinition (NAD)",
            "description": "CRD ที่กำหนด additional networks — ระบุ CNI plugin + config",
            "example": "NAD สำหรับ Macvlan network ที่ connect กับ physical NIC",
        },
        "default_network": {
            "name": "Default Network (eth0)",
            "description": "Cluster network ปกติ (Calico, Flannel, Cilium) — ทุก pod มี",
        },
        "additional_networks": {
            "name": "Additional Networks (net1, net2...)",
            "description": "Networks เพิ่มเติมจาก Multus — SR-IOV, Macvlan, Bridge",
        },
    }

    NAD_EXAMPLE = """
# NetworkAttachmentDefinition — Macvlan network
apiVersion: k8s.cni.cncf.io/v1
kind: NetworkAttachmentDefinition
metadata:
  name: macvlan-net
  namespace: default
spec:
  config: |
    {
      "cniVersion": "0.3.1",
      "type": "macvlan",
      "master": "eth0",
      "mode": "bridge",
      "ipam": {
        "type": "host-local",
        "subnet": "192.168.1.0/24",
        "rangeStart": "192.168.1.200",
        "rangeEnd": "192.168.1.250",
        "gateway": "192.168.1.1"
      }
    }
---
# Pod with multiple networks
apiVersion: v1
kind: Pod
metadata:
  name: multi-net-pod
  annotations:
    k8s.v1.cni.cncf.io/networks: macvlan-net
spec:
  containers:
    - name: app
      image: nginx
      # Pod จะมี: eth0 (cluster) + net1 (macvlan)
"""

    def show_concepts(self):
        print("=== Multus CNI Concepts ===\n")
        for key, concept in self.CONCEPTS.items():
            print(f"[{concept['name']}]")
            print(f"  {concept['description']}")
            print()

    def show_nad(self):
        print("=== NAD Example ===")
        print(self.NAD_EXAMPLE[:500])

arch = MultusArchitecture()
arch.show_concepts()
arch.show_nad()

Testing Strategy Overview

# testing_strategy.py — Testing strategy for Multus CNI
import json

class TestingStrategy:
    PYRAMID = {
        "unit": {
            "name": "Unit Tests",
            "percentage": "40%",
            "scope": "CNI config validation, IPAM logic, NAD schema",
            "tools": "pytest, Go test (สำหรับ CNI plugins)",
            "speed": "เร็วมาก (< 1 วินาที)",
        },
        "integration": {
            "name": "Integration Tests",
            "percentage": "30%",
            "scope": "CNI plugin chain, network connectivity ระหว่าง pods, IPAM allocation",
            "tools": "Kind/Minikube + pytest, Go integration tests",
            "speed": "ปานกลาง (10-60 วินาที)",
        },
        "e2e": {
            "name": "End-to-End Tests",
            "percentage": "20%",
            "scope": "Full cluster tests — pod creation, multi-network connectivity, failover",
            "tools": "Ginkgo + Gomega, Robot Framework, custom scripts",
            "speed": "ช้า (1-10 นาที)",
        },
        "performance": {
            "name": "Performance Tests",
            "percentage": "10%",
            "scope": "Throughput, latency, pod startup time, network bandwidth",
            "tools": "iperf3, netperf, custom benchmarks",
            "speed": "ช้ามาก (10-60 นาที)",
        },
    }

    def show_pyramid(self):
        print("=== Test Pyramid ===\n")
        for key, level in self.PYRAMID.items():
            print(f"[{level['name']}] ({level['percentage']})")
            print(f"  Scope: {level['scope']}")
            print(f"  Tools: {level['tools']}")
            print(f"  Speed: {level['speed']}")
            print()

strategy = TestingStrategy()
strategy.show_pyramid()

Test Implementation

# tests.py — Multus CNI test implementations
import json

class MultusTests:
    UNIT_TESTS = """
# test_nad_validation.py — Unit tests for NAD config
import pytest
import json

class TestNADValidation:
    def test_valid_macvlan_config(self):
        config = {
            "cniVersion": "0.3.1",
            "type": "macvlan",
            "master": "eth0",
            "mode": "bridge",
            "ipam": {
                "type": "host-local",
                "subnet": "192.168.1.0/24",
            }
        }
        assert validate_cni_config(config) == True
    
    def test_missing_cni_version(self):
        config = {"type": "macvlan", "master": "eth0"}
        with pytest.raises(ValueError, match="cniVersion required"):
            validate_cni_config(config)
    
    def test_invalid_subnet(self):
        config = {
            "cniVersion": "0.3.1",
            "type": "macvlan",
            "ipam": {"type": "host-local", "subnet": "invalid"}
        }
        with pytest.raises(ValueError, match="Invalid subnet"):
            validate_cni_config(config)
    
    def test_ipam_range_overlap(self):
        config = {
            "cniVersion": "0.3.1",
            "type": "macvlan",
            "ipam": {
                "type": "host-local",
                "subnet": "192.168.1.0/24",
                "rangeStart": "192.168.1.250",
                "rangeEnd": "192.168.1.200",  # End before Start
            }
        }
        with pytest.raises(ValueError, match="Invalid range"):
            validate_cni_config(config)

def validate_cni_config(config):
    if "cniVersion" not in config:
        raise ValueError("cniVersion required")
    # ... validation logic
    return True
"""

    INTEGRATION_TESTS = """
# test_multus_integration.py — Integration tests
import subprocess
import pytest
import time
import json

class TestMultusIntegration:
    @pytest.fixture(autouse=True)
    def setup(self):
        # Create test namespace
        subprocess.run(["kubectl", "create", "ns", "multus-test"], check=False)
        yield
        subprocess.run(["kubectl", "delete", "ns", "multus-test", "--grace-period=0", "--force"])
    
    def test_pod_gets_additional_interface(self):
        '''Pod with Multus annotation gets extra network interface'''
        # Create NAD
        subprocess.run(["kubectl", "apply", "-f", "test-nad.yaml", "-n", "multus-test"], check=True)
        
        # Create pod with annotation
        subprocess.run(["kubectl", "apply", "-f", "test-pod.yaml", "-n", "multus-test"], check=True)
        
        # Wait for pod ready
        subprocess.run(["kubectl", "wait", "--for=condition=Ready", 
                        "pod/test-pod", "-n", "multus-test", "--timeout=60s"], check=True)
        
        # Verify interfaces
        result = subprocess.run(
            ["kubectl", "exec", "-n", "multus-test", "test-pod", "--", "ip", "addr"],
            capture_output=True, text=True
        )
        
        assert "eth0" in result.stdout  # Default network
        assert "net1" in result.stdout  # Multus additional network
    
    def test_cross_pod_connectivity(self):
        '''Two pods on same additional network can communicate'''
        # Create 2 pods on same macvlan network
        subprocess.run(["kubectl", "apply", "-f", "test-2pods.yaml", "-n", "multus-test"], check=True)
        time.sleep(10)
        
        # Get pod2 IP on net1
        result = subprocess.run(
            ["kubectl", "exec", "-n", "multus-test", "pod-a", "--",
             "ping", "-c", "3", "-W", "2", "192.168.1.201"],
            capture_output=True, text=True
        )
        assert result.returncode == 0
    
    def test_network_isolation(self):
        '''Pods on different additional networks cannot communicate'''
        # pod-a on macvlan-net-a, pod-b on macvlan-net-b
        # ping should fail
        result = subprocess.run(
            ["kubectl", "exec", "-n", "multus-test", "pod-a", "--",
             "ping", "-c", "2", "-W", "2", "10.0.2.100"],
            capture_output=True, text=True
        )
        assert result.returncode != 0  # Should fail — isolated networks
"""

    def show_unit(self):
        print("=== Unit Tests ===")
        print(self.UNIT_TESTS[:600])

    def show_integration(self):
        print(f"\n=== Integration Tests ===")
        print(self.INTEGRATION_TESTS[:600])

tests = MultusTests()
tests.show_unit()
tests.show_integration()

E2E & Performance Tests

# e2e_perf.py — E2E and performance tests
import json
import random

class E2EPerformanceTests:
    E2E_SCENARIOS = {
        "pod_lifecycle": {
            "name": "Pod Lifecycle with Multi-Network",
            "steps": ["Create NAD", "Create Pod", "Verify interfaces", "Delete Pod", "Verify IP released"],
            "timeout": "120s",
        },
        "rolling_update": {
            "name": "Rolling Update Preserves Network",
            "steps": ["Deploy with Multus", "Rolling update image", "Verify no connectivity loss"],
            "timeout": "300s",
        },
        "node_failure": {
            "name": "Node Failure Recovery",
            "steps": ["Pod on node A", "Drain node A", "Pod rescheduled to B", "Verify new interfaces"],
            "timeout": "600s",
        },
        "scale_test": {
            "name": "Scale Test — 100 Pods",
            "steps": ["Create 100 pods with Multus", "Verify all get IPs", "Verify connectivity", "Cleanup"],
            "timeout": "600s",
        },
    }

    PERF_BENCHMARKS = """
# perf_test.py — Network performance benchmarks
import subprocess
import json
import time

class NetworkPerfTest:
    def __init__(self, namespace="perf-test"):
        self.ns = namespace
    
    def bandwidth_test(self, interface="net1"):
        '''iperf3 bandwidth test on additional network'''
        # Start iperf3 server pod
        subprocess.run(["kubectl", "exec", "-n", self.ns, "iperf-server",
                       "--", "iperf3", "-s", "-D", "--bind", "0.0.0.0"])
        
        # Run client
        result = subprocess.run(
            ["kubectl", "exec", "-n", self.ns, "iperf-client", "--",
             "iperf3", "-c", "SERVER_IP", "-t", "30", "-J"],
            capture_output=True, text=True
        )
        
        data = json.loads(result.stdout)
        return {
            "bandwidth_gbps": data["end"]["sum_sent"]["bits_per_second"] / 1e9,
            "retransmits": data["end"]["sum_sent"]["retransmits"],
        }
    
    def latency_test(self, target_ip, count=100):
        '''Ping latency test'''
        result = subprocess.run(
            ["kubectl", "exec", "-n", self.ns, "ping-pod", "--",
             "ping", "-c", str(count), "-i", "0.1", target_ip],
            capture_output=True, text=True
        )
        # Parse avg latency
        return {"avg_latency_ms": 0.5}  # parsed from output
    
    def pod_startup_time(self, count=10):
        '''Measure pod startup time with Multus'''
        times = []
        for i in range(count):
            start = time.time()
            subprocess.run(["kubectl", "run", f"timing-{i}", 
                          "--image=alpine", "--restart=Never",
                          f"--overrides={{...multus annotation...}}"], check=True)
            subprocess.run(["kubectl", "wait", f"--for=condition=Ready",
                          f"pod/timing-{i}", "--timeout=120s"], check=True)
            times.append(time.time() - start)
        
        return {
            "avg_startup_s": sum(times) / len(times),
            "p99_startup_s": sorted(times)[int(len(times) * 0.99)],
        }
"""

    def show_scenarios(self):
        print("=== E2E Test Scenarios ===\n")
        for key, scenario in self.E2E_SCENARIOS.items():
            print(f"[{scenario['name']}] Timeout: {scenario['timeout']}")
            for step in scenario["steps"][:3]:
                print(f"  → {step}")
            print()

    def show_benchmarks(self):
        print("=== Performance Benchmarks ===")
        print(self.PERF_BENCHMARKS[:500])

    def sample_results(self):
        print(f"\n=== Sample Performance Results ===")
        print(f"  Default network (eth0):")
        print(f"    Bandwidth: {random.uniform(8, 10):.1f} Gbps")
        print(f"    Latency: {random.uniform(0.1, 0.5):.2f} ms")
        print(f"  Multus macvlan (net1):")
        print(f"    Bandwidth: {random.uniform(7, 9.5):.1f} Gbps")
        print(f"    Latency: {random.uniform(0.05, 0.3):.2f} ms")
        print(f"  Multus SR-IOV (net1):")
        print(f"    Bandwidth: {random.uniform(9, 10):.1f} Gbps")
        print(f"    Latency: {random.uniform(0.02, 0.1):.2f} ms")
        print(f"  Pod startup (with Multus): {random.uniform(3, 8):.1f}s avg")

e2e = E2EPerformanceTests()
e2e.show_scenarios()
e2e.sample_results()

CI/CD Integration

# cicd.py — CI/CD for Multus CNI testing
import json

class CICDPipeline:
    GITHUB_ACTIONS = """
# .github/workflows/multus-tests.yml
name: Multus CNI Tests
on:
  push:
    branches: [main]
    paths: ['network/**', 'tests/**']
  pull_request:
    paths: ['network/**', 'tests/**']

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install pytest pyyaml jsonschema
      - run: pytest tests/unit/ -v

  integration-tests:
    needs: unit-tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Create Kind cluster with Multus
        run: |
          kind create cluster --config tests/kind-config.yaml
          kubectl apply -f https://raw.githubusercontent.com/k8snetworkplumbingwg/multus-cni/master/deployments/multus-daemonset.yml
          kubectl wait --for=condition=Ready -n kube-system ds/kube-multus-ds --timeout=120s
      
      - name: Run integration tests
        run: pytest tests/integration/ -v --timeout=300
      
      - name: Cleanup
        if: always()
        run: kind delete cluster

  e2e-tests:
    needs: integration-tests
    runs-on: self-hosted  # Needs real hardware for SR-IOV
    steps:
      - uses: actions/checkout@v4
      - name: Run E2E tests
        run: pytest tests/e2e/ -v --timeout=600
"""

    def show_pipeline(self):
        print("=== CI/CD Pipeline ===")
        print(self.GITHUB_ACTIONS[:600])

    def test_matrix(self):
        print(f"\n=== Test Matrix ===")
        matrix = [
            {"cni": "Macvlan", "unit": "Pass", "integration": "Pass", "e2e": "Pass"},
            {"cni": "Bridge", "unit": "Pass", "integration": "Pass", "e2e": "Pass"},
            {"cni": "IPVLAN", "unit": "Pass", "integration": "Pass", "e2e": "Skip"},
            {"cni": "SR-IOV", "unit": "Pass", "integration": "Skip", "e2e": "Pass"},
        ]
        print(f"  {'CNI Plugin':<12} {'Unit':>6} {'Integration':>12} {'E2E':>6}")
        for m in matrix:
            print(f"  {m['cni']:<12} {m['unit']:>6} {m['integration']:>12} {m['e2e']:>6}")

cicd = CICDPipeline()
cicd.show_pipeline()
cicd.test_matrix()

FAQ - คำถามที่พบบ่อย

Q: Multus CNI จำเป็นเมื่อไหร่?

A: จำเป็นเมื่อ pod ต้องการ network interfaces มากกว่า 1: Telco/5G: แยก user plane กับ control plane, Storage network: แยก data path กับ management, NFV: multiple VLANs per pod, High-performance: SR-IOV สำหรับ line-rate networking ถ้า pod ใช้แค่ 1 network → ไม่ต้องใช้ Multus

Q: Test Multus ยากไหม?

A: Unit tests: ง่าย — validate config, schema Integration: ปานกลาง — ต้อง Kubernetes cluster + Multus installed E2E: ยาก — SR-IOV ต้อง hardware จริง (NIC ที่รองรับ) ใช้ Kind cluster สำหรับ Macvlan/Bridge tests (ไม่ต้อง hardware พิเศษ) SR-IOV tests: ต้อง self-hosted runners + physical NICs

Q: Performance ของ Multus เป็นอย่างไร?

A: Macvlan: ใกล้เคียง bare metal (~95% throughput), latency ต่ำ Bridge: ดี แต่มี overhead จาก software bridge SR-IOV: ดีที่สุด — bypass kernel, near line-rate (99%+) Pod startup: เพิ่ม 1-3 วินาที ต่อ additional interface Multus overhead เอง: น้อยมาก (< 100ms) — ส่วนใหญ่ขึ้นกับ CNI plugin ที่ใช้

Q: Troubleshoot Multus ยังไง?

A: 1) kubectl describe pod — ดู events สำหรับ network errors 2) kubectl logs -n kube-system ds/kube-multus-ds — ดู Multus logs 3) kubectl exec pod -- ip addr — ตรวจสอบ interfaces 4) kubectl get net-attach-def — ตรวจสอบ NAD configs 5) journalctl -u kubelet — ดู CNI plugin errors ที่ node level