Kubernetes Ingress与网络配置

apiVersion: networking.k8s.io/v1kind: Ingressmetadata:  name: ai-demo-ingress  annotations:    nginx.ingress.kubernetes.io/rewrite-target: /$2spec:  rules:  - host: ai.example.com    http:      paths:      - path: /api(/|$)(.*)        pathType: ImplementationSpecific        backend:          service:            name: ai-api-service            port:              number: 8080      - path: /ui(/|$)(.*)        pathType: ImplementationSpecific        backend:          service:            name: ai-ui-service            port:              number: 80

apiVersion: networking.k8s.io/v1kind: Ingressmetadata:  name: ai-demo-ingressspec:  tls:  - hosts:    - ai.example.com    secretName: ai-tls-secret  rules:  - host: ai.example.com    http:      paths:      - path: /        pathType: Prefix        backend:          service:            name: ai-api-service            port:              number: 8080

apiVersion: networking.k8s.io/v1kind: NetworkPolicymetadata:  name: db-allow-only-apispec:  podSelector:    matchLabels:      app: mysql  policyTypes:  - Ingress  ingress:  - from:    - podSelector:        matchLabels:          app: ai-api    ports:    - protocol: TCP      port: 3306

import reimport timefrom dataclasses import dataclassfrom typing import List, Optional # 模拟 k8s Service 后端@dataclassclass BackendService:    name: str    port: int # 模拟一条 Ingress path 规则@dataclassclass PathRule:    path: str    path_type: str  # Prefix 或 Exact    backend: BackendService # 模拟一条 Ingress host 规则@dataclassclass HostRule:    host: str    paths: List[PathRule] class IngressSimulator:    def __init__(self) -> None:        self.rules: List[HostRule] = []        self.default_backend: Optional[BackendService] = None     def add_rule(self, rule: HostRule) -> None:        self.rules.append(rule)     def match(self, host: str, path: str) -> Optional[BackendService]:        best_score = -1        result: Optional[BackendService] = None        for rule in self.rules:            if rule.host != host:                continue  # host 不匹配则跳过            for pr in rule.paths:                if pr.path_type == "Exact" and pr.path == path:                    score = 3  # 精确匹配得分最高                elif pr.path_type == "Prefix" and path.startswith(pr.path):                    score = 2  # 前缀匹配次之                else:                    continue                if score > best_score:                    best_score = score                    result = pr.backend        return result     def route(self, host: str, path: str) -> str:        backend = self.match(host, path)        if backend is None:            return "404 Not Found; no rule matched"        return f"200 OK; -> {backend.name}:{backend.port}" def main() -> None:    sim = IngressSimulator()    sim.add_rule(HostRule(        host="ai.example.com",        paths=[            PathRule("/api", "Prefix", BackendService("ai-api-service", 8080)),            PathRule("/ui", "Prefix", BackendService("ai-ui-service", 80)),        ],    ))    sim.add_rule(HostRule(        host="test.example.com",        paths=[            PathRule("/api", "Exact", BackendService("ai-test-service", 9090)),        ],    ))     requests = [        ("ai.example.com", "/api/v1/predict"),        ("ai.example.com", "/ui/index.html"),        ("bad.example.com", "/api/v1/predict"),        ("test.example.com", "/api"),        ("test.example.com", "/api/v2"),    ]    for host, path in requests:        start = time.perf_counter()        resp = sim.route(host, path)        cost_ms = (time.perf_counter() - start) * 1000        print(f"REQ  {host}{path}")        print(f"RESP {resp}  ({cost_ms:.2f} ms)\n") if __name__ == "__main__":    main()