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