External Secrets 구성 예제 (YAML/Python)
# SecretStore 정의 - AWS Secrets Manager 연동
apiVersion: external-secrets.io/v1beta1
kind: SecretStore
metadata:
name: aws-secrets-store
namespace: production
spec:
provider:
aws:
service: SecretsManager
region: ap-northeast-2
auth:
jwt:
serviceAccountRef:
name: external-secrets-sa
---
# ExternalSecret 정의 - 외부 시크릿을 Kubernetes Secret으로 동기화
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
name: database-credentials
namespace: production
spec:
refreshInterval: 1h # 동기화 주기
secretStoreRef:
name: aws-secrets-store
kind: SecretStore
target:
name: db-secret # 생성될 Kubernetes Secret 이름
creationPolicy: Owner
template:
type: Opaque
data:
# 템플릿을 사용한 값 변환
connection-string: |
postgresql://{{ .username }}:{{ .password }}@{{ .host }}:5432/{{ .database }}
data:
- secretKey: username
remoteRef:
key: prod/database/credentials
property: username
- secretKey: password
remoteRef:
key: prod/database/credentials
property: password
- secretKey: host
remoteRef:
key: prod/database/credentials
property: host
- secretKey: database
remoteRef:
key: prod/database/credentials
property: database
Python으로 External Secrets 상태 모니터링
from kubernetes import client, config
from kubernetes.client.rest import ApiException
import json
# Kubernetes 설정 로드
config.load_incluster_config() # 클러스터 내부에서 실행 시
# config.load_kube_config() # 로컬에서 실행 시
# Custom Objects API 클라이언트
api = client.CustomObjectsApi()
def get_external_secret_status(namespace: str, name: str) -> dict:
"""ExternalSecret 리소스의 동기화 상태 확인"""
try:
external_secret = api.get_namespaced_custom_object(
group="external-secrets.io",
version="v1beta1",
namespace=namespace,
plural="externalsecrets",
name=name
)
status = external_secret.get("status", {})
conditions = status.get("conditions", [])
# SecretSynced 조건 확인
for condition in conditions:
if condition["type"] == "Ready":
return {
"name": name,
"namespace": namespace,
"synced": condition["status"] == "True",
"reason": condition.get("reason", "Unknown"),
"message": condition.get("message", ""),
"last_sync": status.get("refreshTime")
}
return {"name": name, "synced": False, "reason": "NoCondition"}
except ApiException as e:
return {"name": name, "error": str(e)}
def list_all_external_secrets() -> list:
"""클러스터 내 모든 ExternalSecret 상태 조회"""
results = []
external_secrets = api.list_cluster_custom_object(
group="external-secrets.io",
version="v1beta1",
plural="externalsecrets"
)
for es in external_secrets.get("items", []):
metadata = es["metadata"]
status = get_external_secret_status(
namespace=metadata["namespace"],
name=metadata["name"]
)
results.append(status)
return results
def check_secret_sync_health() -> dict:
"""전체 ExternalSecret 동기화 상태 헬스체크"""
all_secrets = list_all_external_secrets()
total = len(all_secrets)
synced = sum(1 for s in all_secrets if s.get("synced", False))
failed = [s for s in all_secrets if not s.get("synced", False)]
return {
"total": total,
"synced": synced,
"failed_count": len(failed),
"failed_secrets": failed,
"health": "healthy" if len(failed) == 0 else "degraded"
}
# 사용 예시
if __name__ == "__main__":
# 특정 ExternalSecret 상태 확인
status = get_external_secret_status("production", "database-credentials")
print(f"동기화 상태: {status}")
# 전체 헬스체크
health = check_secret_sync_health()
print(f"전체 상태: {health['synced']}/{health['total']} 동기화됨")
if health["failed_secrets"]:
print("동기화 실패한 시크릿:")
for secret in health["failed_secrets"]:
print(f" - {secret['namespace']}/{secret['name']}: {secret.get('reason')}")