OPA Gatekeeper กับ Internal Developer Platform

OPA Gatekeeper คืออะไร

OPA Gatekeeper เป็น Policy Engine สำหรับ Kubernetes ที่ทำหน้าที่เป็น Admission Controller ตรวจสอบทุก Request ที่เข้ามาที่ Kubernetes API Server ว่าเป็นไปตาม Policy ที่กำหนดหรือไม่ ถ้าไม่ผ่าน Request จะถูก Reject ทันที

เมื่อใช้ร่วมกับ Internal Developer Platform (IDP) Gatekeeper ทำหน้าที่เป็น Guardrails ให้ Developer สามารถ Deploy Application ได้ด้วยตัวเองอย่างปลอดภัย โดยไม่ต้องรอ Platform Team Review ทุก Manifest เพราะ Policy จะตรวจสอบอัตโนมัติ

ติดตั้ง OPA Gatekeeper

# ติดตั้ง Gatekeeper ด้วย Helm
helm repo add gatekeeper https://open-policy-agent.github.io/gatekeeper/charts
helm repo update

helm install gatekeeper gatekeeper/gatekeeper \
  --namespace gatekeeper-system \
  --create-namespace \
  --set replicas=3 \
  --set audit.replicas=1 \
  --set audit.logLevel=INFO \
  --set controllerManager.logLevel=INFO

# ตรวจสอบ
kubectl get pods -n gatekeeper-system
kubectl get crd | grep gatekeeper

# === Constraint Template: ห้ามใช้ Latest Tag ===
# k8s-required-image-tag.yaml
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredimagetag
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredImageTag
      validation:
        openAPIV3Schema:
          type: object
          properties:
            exemptImages:
              type: array
              items:
                type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredimagetag
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not valid_tag(container.image)
          msg := sprintf("Container '%v' uses invalid image '%v'. Must specify a tag (not 'latest')", [container.name, container.image])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.initContainers[_]
          not valid_tag(container.image)
          msg := sprintf("Init container '%v' uses invalid image '%v'", [container.name, container.image])
        }
        
        valid_tag(image) {
          # ตรวจสอบว่ามี tag และไม่ใช่ latest
          contains(image, ":")
          not endswith(image, ":latest")
        }
        
        valid_tag(image) {
          # ใช้ SHA digest
          contains(image, "@sha256:")
        }
        
        valid_tag(image) {
          # Exempt images
          exempt := input.parameters.exemptImages[_]
          glob.match(exempt, ["/"], image)
        }

---
# Constraint: บังคับใช้กับทุก Namespace ยกเว้น kube-system
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredImageTag
metadata:
  name: require-image-tag
spec:
  enforcementAction: deny
  match:
    kinds:
      - apiGroups: [""]
        kinds: ["Pod"]
      - apiGroups: ["apps"]
        kinds: ["Deployment", "StatefulSet", "DaemonSet"]
    excludedNamespaces:
      - kube-system
      - gatekeeper-system
  parameters:
    exemptImages:
      - "gcr.io/distroless/*"

# Apply
kubectl apply -f k8s-required-image-tag.yaml

Policy Templates สำหรับ IDP

# === Policy: ต้องมี Resource Limits ===
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredresources
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredResources
      validation:
        openAPIV3Schema:
          type: object
          properties:
            maxCpuLimit:
              type: string
            maxMemoryLimit:
              type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredresources
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.limits
          msg := sprintf("Container '%v' must have resource limits", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.limits.cpu
          msg := sprintf("Container '%v' must have CPU limit", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.limits.memory
          msg := sprintf("Container '%v' must have memory limit", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.resources.requests
          msg := sprintf("Container '%v' must have resource requests", [container.name])
        }

---
# === Policy: ต้องมี Required Labels ===
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8srequiredlabels
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredLabels
      validation:
        openAPIV3Schema:
          type: object
          properties:
            labels:
              type: array
              items:
                type: object
                properties:
                  key:
                    type: string
                  allowedRegex:
                    type: string
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredlabels
        
        violation[{"msg": msg}] {
          provided := {label | input.review.object.metadata.labels[label]}
          required := {label | label := input.parameters.labels[_].key}
          missing := required - provided
          count(missing) > 0
          msg := sprintf("Missing required labels: %v", [missing])
        }
        
        violation[{"msg": msg}] {
          label := input.parameters.labels[_]
          value := input.review.object.metadata.labels[label.key]
          label.allowedRegex != ""
          not re_match(label.allowedRegex, value)
          msg := sprintf("Label '%v' value '%v' does not match regex '%v'", [label.key, value, label.allowedRegex])
        }

---
apiVersion: constraints.gatekeeper.sh/v1beta1
kind: K8sRequiredLabels
metadata:
  name: require-team-labels
spec:
  enforcementAction: deny
  match:
    kinds:
      - apiGroups: ["apps"]
        kinds: ["Deployment"]
    excludedNamespaces: ["kube-system"]
  parameters:
    labels:
      - key: "app.kubernetes.io/name"
      - key: "app.kubernetes.io/team"
        allowedRegex: "^(platform|backend|frontend|data|ml)$"
      - key: "app.kubernetes.io/environment"
        allowedRegex: "^(dev|staging|production)$"

---
# === Policy: ห้ามใช้ Root Container ===
apiVersion: templates.gatekeeper.sh/v1
kind: ConstraintTemplate
metadata:
  name: k8snoroot
spec:
  crd:
    spec:
      names:
        kind: K8sNoRoot
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8snoroot
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          container.securityContext.runAsUser == 0
          msg := sprintf("Container '%v' must not run as root", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input.review.object.spec.containers[_]
          not container.securityContext.runAsNonRoot
          not container.securityContext.runAsUser
          msg := sprintf("Container '%v' must set securityContext.runAsNonRoot=true", [container.name])
        }

Python Script ทดสอบ Policy

# test_policies.py — ทดสอบ OPA Policies ด้วย Python
import subprocess
import json
import yaml
import sys

class PolicyTester:
    """ทดสอบ OPA Gatekeeper Policies"""

    def __init__(self):
        self.results = []

    def test_manifest(self, manifest_path, expected_result="pass"):
        """ทดสอบ Manifest กับ Gatekeeper"""
        # Dry-run ด้วย kubectl
        cmd = [
            "kubectl", "apply", "-f", manifest_path,
            "--dry-run=server", "-o", "json"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)

        passed = result.returncode == 0
        test_passed = (passed and expected_result == "pass") or \
                      (not passed and expected_result == "fail")

        self.results.append({
            "manifest": manifest_path,
            "expected": expected_result,
            "actual": "pass" if passed else "fail",
            "test_passed": test_passed,
            "error": result.stderr if not passed else "",
        })

        status = "PASS" if test_passed else "FAIL"
        print(f"  [{status}] {manifest_path} (expected: {expected_result})")
        if not test_passed:
            print(f"    Error: {result.stderr[:200]}")

    def run_suite(self, test_cases):
        """รัน Test Suite"""
        print("=== Policy Test Suite ===\n")
        for tc in test_cases:
            self.test_manifest(tc["manifest"], tc["expected"])

        passed = sum(1 for r in self.results if r["test_passed"])
        total = len(self.results)
        print(f"\nResults: {passed}/{total} passed")

        if passed < total:
            print("\nFailed tests:")
            for r in self.results:
                if not r["test_passed"]:
                    print(f"  - {r['manifest']}: expected {r['expected']}, "
                          f"got {r['actual']}")
            sys.exit(1)

# Test Cases
tests = [
    # ควร PASS: มี Tag, Resource Limits, Labels
    {"manifest": "tests/valid-deployment.yaml", "expected": "pass"},
    # ควร FAIL: ใช้ Latest Tag
    {"manifest": "tests/latest-tag.yaml", "expected": "fail"},
    # ควร FAIL: ไม่มี Resource Limits
    {"manifest": "tests/no-limits.yaml", "expected": "fail"},
    # ควร FAIL: ไม่มี Required Labels
    {"manifest": "tests/no-labels.yaml", "expected": "fail"},
    # ควร FAIL: Run as Root
    {"manifest": "tests/root-container.yaml", "expected": "fail"},
]

tester = PolicyTester()
tester.run_suite(tests)

IDP Integration Workflow

Developer สร้าง PR: แก้ไข Kubernetes Manifests ใน Git Repository
CI Pipeline ทดสอบ: รัน conftest (OPA ใน CI) ตรวจสอบ Policy ก่อน Merge
Merge และ Deploy: ArgoCD/Flux Sync Manifests ไป Cluster
Gatekeeper ตรวจสอบ: Admission Controller ตรวจสอบ Policy อีกครั้งตอน Apply
Audit Mode: Policy ใหม่เริ่มด้วย warn ก่อน แล้วเปลี่ยนเป็น deny เมื่อพร้อม
Dashboard: Backstage หรือ Port แสดง Policy Violations ให้ Developer เห็น

Best Practices

เริ่มด้วย Audit Mode: ตั้ง enforcementAction: warn ก่อน ดูว่า Policy กระทบอะไรบ้าง แล้วค่อยเปลี่ยนเป็น deny
ใช้ Constraint Library: ใช้ gatekeeper-library ที่มี Template สำเร็จรูปหลายสิบตัว ไม่ต้องเขียนเอง
ทดสอบใน CI: ใช้ conftest หรือ gator CLI ทดสอบ Policy ใน CI Pipeline ก่อน Deploy
Exclude System Namespaces: ยกเว้น kube-system, gatekeeper-system จาก Policy เพื่อไม่ให้กระทบ System Components
Version Control: เก็บ Constraint Templates และ Constraints ใน Git ใช้ GitOps Deploy
ข้อความ Error ที่ดี: เขียน Error Message ที่ชัดเจน บอก Developer ว่าต้องแก้ไขอย่างไร

OPA Gatekeeper คืออะไร

OPA Gatekeeper เป็น Kubernetes Admission Controller ที่ใช้ OPA บังคับ Policy ตรวจสอบทุก Resource ที่สร้างหรือแก้ไขว่าเป็นไปตาม Policy เช่น ห้ามใช้ Latest Tag ต้องมี Resource Limits ต้องมี Labels ถ้าไม่ผ่าน Request ถูก Reject

OPA Gatekeeper กับ Internal Developer Platform —

OPA Gatekeeper คืออะไร

ติดตั้ง OPA Gatekeeper

Policy Templates สำหรับ IDP

Python Script ทดสอบ Policy

IDP Integration Workflow

Best Practices

OPA Gatekeeper คืออะไร

บทความที่เกี่ยวข้อง