如何在Kubernetes Sidecar容器实现许可证机制，验证失败时终止应用容器？

算力阿峰

2026-6-16

问题场景

基于Kubernetes v1.31.2搭建环境，包含应用容器和许可证验证Sidecar容器，要求许可证验证失败时停止应用容器服务，且不能修改应用容器内部代码。当前Sidecar验证失败退出后，应用容器仍正常运行。

现有实现代码

许可证验证Sidecar脚本

import os
import sys
import time
import requests

LICENSE_SERVER_URL = os.getenv("LICENSE_SERVER_URL", "")  # URL for the license service
LICENSE_TOKEN = os.getenv("LICENSE_TOKEN", "")
CHECK_INTERVAL = 60  # Check every 60 seconds

def validate_license():
    """Request the license issuer service to validate the token."""
    response = requests.get(f"{LICENSE_SERVER_URL}/validate_license/{LICENSE_TOKEN}")
    if response.status_code == 200 and response.json().get("status") == "valid":
        print("License validation successful.")
        return True
    else:
        print("License validation failed.")
        return False

def main():
    if not LICENSE_TOKEN:
        print("No license token found. Exiting.")
        sys.exit(1)

    while True:
        if not validate_license():
            print("License is invalid. Terminating pod.")
            sys.exit(1)  # Exit to terminate the pod
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    main()

当前Pod配置

apiVersion: v1
kind: Pod
metadata:
  name: licensed-app-pod
  labels:
    app: licensed-app
spec:
  initContainers:
    - name: license-verifier
      image: localhost:5000/license_verifier:0.1 # Replace with your actual image for license verification
      restartPolicy: Always
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
        - name: LICENSE_SERVER_URL
          value: "http://license-server-service:5000" # URL for the license issuer service
      # Periodic check for license validity
      command: ["python", "/app/license_verifier.py"]
  containers:
    - name: application
      image: nginx:latest
      ports:
        - containerPort: 80
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
  restartPolicy: Always

解决方案与最佳实践

方法1：共享PID命名空间，让Sidecar直接终止应用进程

通过开启Pod的PID命名空间共享，允许Sidecar容器操作应用容器的主进程，达到许可证失效时终止应用的目的。

配置修改

在Pod spec中添加shareProcessNamespace: true
更新Sidecar脚本，增加终止应用进程的逻辑

修改后的Sidecar脚本：

import os
import sys
import time
import requests
import psutil

LICENSE_SERVER_URL = os.getenv("LICENSE_SERVER_URL", "")
LICENSE_TOKEN = os.getenv("LICENSE_TOKEN", "")
CHECK_INTERVAL = 60
# 替换为应用容器的主进程名称（如nginx、java等）
APP_PROCESS_NAME = "nginx"

def validate_license():
    response = requests.get(f"{LICENSE_SERVER_URL}/validate_license/{LICENSE_TOKEN}")
    if response.status_code == 200 and response.json().get("status") == "valid":
        print("License validation successful.")
        return True
    else:
        print("License validation failed.")
        return False

def terminate_app_process():
    print(f"Terminating app process: {APP_PROCESS_NAME}")
    for proc in psutil.process_iter(['name']):
        if proc.info['name'] == APP_PROCESS_NAME:
            try:
                proc.terminate()
                print(f"Killed process {proc.pid}")
            except Exception as e:
                print(f"Failed to kill process {proc.pid}: {str(e)}")
    sys.exit(1)

def main():
    if not LICENSE_TOKEN:
        print("No license token found.")
        terminate_app_process()

    while True:
        if not validate_license():
            print("License invalid, stopping app.")
            terminate_app_process()
        time.sleep(CHECK_INTERVAL)

if __name__ == "__main__":
    main()

修改后的Pod YAML：

apiVersion: v1
kind: Pod
metadata:
  name: licensed-app-pod
  labels:
    app: licensed-app
spec:
  shareProcessNamespace: true  # 开启PID命名空间共享
  containers:
    - name: license-verifier
      image: localhost:5000/license_verifier:0.1
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
        - name: LICENSE_SERVER_URL
          value: "http://license-server-service:5000"
      command: ["python", "/app/license_verifier.py"]
    - name: application
      image: nginx:latest
      ports:
        - containerPort: 80
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
  restartPolicy: Always

方法2：利用Kubernetes健康探针，让应用依赖Sidecar状态

让Sidecar暴露健康检查端点，应用容器通过livenessProbe和readinessProbe依赖该端点状态，实现许可证失效时自动停止流量或重启应用。

配置修改

给Sidecar添加HTTP健康服务，验证通过返回200，失败返回503
为应用容器配置探针，检查Sidecar的健康端点

修改后的Sidecar脚本：

import os
import sys
import time
import requests
from flask import Flask

app = Flask(__name__)
LICENSE_VALID = True

LICENSE_SERVER_URL = os.getenv("LICENSE_SERVER_URL", "")
LICENSE_TOKEN = os.getenv("LICENSE_TOKEN", "")
CHECK_INTERVAL = 60
HEALTH_PORT = 8080

def validate_license():
    global LICENSE_VALID
    try:
        response = requests.get(f"{LICENSE_SERVER_URL}/validate_license/{LICENSE_TOKEN}")
        LICENSE_VALID = (response.status_code == 200 and response.json().get("status") == "valid")
        print(f"License validation: {'successful' if LICENSE_VALID else 'failed'}")
    except Exception as e:
        print(f"License check error: {str(e)}")
        LICENSE_VALID = False

@app.route("/health")
def health_check():
    return {"status": "healthy"} if LICENSE_VALID else {"status": "unhealthy"}, 200 if LICENSE_VALID else 503

def main():
    if not LICENSE_TOKEN:
        print("No license token found.")
        sys.exit(1)

    # 启动后台线程定期检查许可证
    import threading
    def check_loop():
        while True:
            validate_license()
            time.sleep(CHECK_INTERVAL)
    checker_thread = threading.Thread(target=check_loop)
    checker_thread.daemon = True
    checker_thread.start()

    # 启动健康服务
    app.run(host="0.0.0.0", port=HEALTH_PORT)

if __name__ == "__main__":
    main()

修改后的Pod YAML：

apiVersion: v1
kind: Pod
metadata:
  name: licensed-app-pod
  labels:
    app: licensed-app
spec:
  containers:
    - name: license-verifier
      image: localhost:5000/license_verifier:0.1
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
        - name: LICENSE_SERVER_URL
          value: "http://license-server-service:5000"
      ports:
        - containerPort: 8080  # 暴露健康检查端口
      command: ["python", "/app/license_verifier.py"]
    - name: application
      image: nginx:latest
      ports:
        - containerPort: 80
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
      livenessProbe:
        httpGet:
          host: localhost
          path: /health
          port: 8080
        initialDelaySeconds: 10
        periodSeconds: 10
      readinessProbe:
        httpGet:
          host: localhost
          path: /health
          port: 8080
        initialDelaySeconds: 5
        periodSeconds: 5
  restartPolicy: Always

方法3：InitContainer+Sidecar组合，实现启动验证+定期检查

通过InitContainer确保Pod启动前许可证有效，Sidecar定期检查，失效时退出Pod，触发重启后InitContainer再次验证，阻止应用再次启动。

修改后的Pod YAML：

apiVersion: v1
kind: Pod
metadata:
  name: licensed-app-pod
  labels:
    app: licensed-app
spec:
  initContainers:
    - name: license-init-verifier
      image: localhost:5000/license_verifier:0.1
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
        - name: LICENSE_SERVER_URL
          value: "http://license-server-service:5000"
      # 仅做一次初始验证，失败则阻止Pod启动
      command: ["python", "-c", "
import os
import requests
LICENSE_SERVER_URL = os.getenv('LICENSE_SERVER_URL', '')
LICENSE_TOKEN = os.getenv('LICENSE_TOKEN', '')
if not LICENSE_TOKEN:
    print('No license token found.')
    exit(1)
response = requests.get(f'{LICENSE_SERVER_URL}/validate_license/{LICENSE_TOKEN}')
if not (response.status_code == 200 and response.json().get('status') == 'valid'):
    print('Initial license validation failed.')
    exit(1)
print('Initial license validation successful.')
"]
  containers:
    - name: license-sidecar
      image: localhost:5000/license_verifier:0.1
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
        - name: LICENSE_SERVER_URL
          value: "http://license-server-service:5000"
      command: ["python", "/app/license_verifier.py"]
    - name: application
      image: nginx:latest
      ports:
        - containerPort: 80
      env:
        - name: LICENSE_TOKEN
          valueFrom:
            secretKeyRef:
              name: license-secret
              key: token
  restartPolicy: Always

方案对比与推荐

方法2（健康探针）：最符合Kubernetes设计理念，无侵入性，既能自动停止流量（readinessProbe）又能触发重启（livenessProbe），推荐优先使用。
方法1（共享PID）：适合需要立即终止应用的场景，但需要Sidecar安装psutil依赖，且共享PID存在一定安全风险。
方法3（InitContainer+Sidecar）：适合严格控制应用启动的场景，确保应用仅在许可证有效时运行，但失效后Pod会反复重启直到许可证恢复。

内容的提问来源于stack exchange，提问作者San-Zero

如何在Kubernetes Sidecar容器实现许可证机制，验证失败时终止应用容器？

现有实现代码 #

许可证验证Sidecar脚本 #

当前Pod配置 #

解决方案与最佳实践 #

方法1：共享PID命名空间，让Sidecar直接终止应用进程 #

配置修改

方法2：利用Kubernetes健康探针，让应用依赖Sidecar状态 #

配置修改

方法3：InitContainer+Sidecar组合，实现启动验证+定期检查 #

方案对比与推荐 #

现有实现代码

许可证验证Sidecar脚本

当前Pod配置

解决方案与最佳实践

方法1：共享PID命名空间，让Sidecar直接终止应用进程

方法2：利用Kubernetes健康探针，让应用依赖Sidecar状态

方法3：InitContainer+Sidecar组合，实现启动验证+定期检查

方案对比与推荐