26Y27a5
작성한 Python 모듈들을 하나로 결합하여 단일 실행 파이프라인으로 묶고, 이를 Air-Gapped 환경용 컨테이너 이미지로 빌드한 뒤 Kubernetes CronJob으로 배포·검증하는 전체 과정을 정리해 드립니다.
1. 전체 Python 파이프라인 코드 구조화
각 스텝을 하나의 메인 스크립트(main.py)로 연결합니다. 실행 상태(마지막 동기화 시간)는 Persistent Volume(PVC)이나 로컬 파일에 기록하여 다음 실행 시 활용합니다.
/app
├── main.py # 전체 흐름제어 (CQL API -> Clean MD -> Git Sync -> Dual Indexing)
├── confluence_client.py # Confluence REST API / CQL 증분 조회
├── converter.py # HTML -> Clean MD & Front-Matter 주입
├── git_sync.py # Git Commit & Push
├── indexer.py # Header Chunking & OpenSearch / Qdrant Bulk Upsert
├── requirements.txt # 의존성 라이브러리 목록
└── Dockerfile
main.py (전체 프로세스 조합)
import os
import json
import logging
from datetime import datetime, timezone
from confluence_client import get_updated_pages
from converter import convert_confluence_html_to_md
from git_sync import sync_md_to_git
from indexer import chunk_and_dual_index
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
STATE_FILE = "/workspace/ops-knowledge-repo/.sync_state.json"
def get_last_sync_time():
if os.path.exists(STATE_FILE):
with open(STATE_FILE, "r") as f:
return json.load(f).get("last_sync_time")
# 최초 실행 시 기본값 (예: 1일 전)
return "2026-07-25T00:00:00Z"
def save_last_sync_time(sync_time_str):
with open(STATE_FILE, "w") as f:
json.dump({"last_sync_time": sync_time_str}, f)
def run_pipeline():
last_sync_time = get_last_sync_time()
current_execution_time = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
logging.info(f"==> Sync Pipeline Started. Last Sync Time: {last_sync_time}")
# 1. Confluence CQL 증분 페이지 조회
updated_pages = get_updated_pages(last_sync_time)
logging.info(f"Found {len(updated_pages)} updated pages.")
if not updated_pages:
logging.info("No documents updated. Exiting pipeline.")
return
for page in updated_pages:
page_id = page["id"]
title = page["title"]
space = page["space"]["key"]
raw_html = page["body"]["storage"]["value"]
logging.info(f"Processing Page ID: {page_id} | Title: {title}")
# 2. HTML Cleaning & Markdown 변환
md_content = convert_confluence_html_to_md(page, raw_html)
# 3. Git 저장 및 Sync (Commit Hash 반환)
commit_hash = sync_md_to_git(page_id, title, space, md_content)
# 4. Chunking & Dual-Indexing (OpenSearch + Qdrant)
chunk_and_dual_index(page_id, md_content, commit_hash)
# 파이프라인 성공 시 동기화 시간 업데이트
save_last_sync_time(current_execution_time)
logging.info("==> Sync Pipeline Completed Successfully.")
if __name__ == "__main__":
run_pipeline()
2. Docker Container 이미지 생성 (Dockerfile)
망분리 환경의 사내 Container Registry(Harbor, Nexus 등)에 배포하기 위한 Dockerfile과 requirements.txt 예시입니다.
requirements.txt
beautifulsoup4==4.12.3
html2text==2024.2.26
GitPython==3.1.43
langchain-text-splitters==0.2.2
opensearch-py==2.6.0
qdrant-client==1.9.1
requests==2.32.3
pyyaml==6.0.1
Dockerfile
FROM python:3.11-slim
# Git 바이너리 설치 (GitPython 동작용)
RUN apt-get update && apt-get install -y --no-install-recommends \
git \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
# 파이썬 의존성 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 애플리케이션 소스 복사
COPY . .
# 실행 유저 권한 비루트(Non-root) 설정 (K8s Security Best Practice)
RUN useradd -m appuser && chown -R appuser:appuser /app
USER appuser
ENTRYPOINT ["python", "main.py"]
이미지 빌드 및 Harbor/Nexus Push
# 1. 빌드
docker build -t harbor.internal/aiops/confluence-sync:v1.0.0 .
# 2. 사내 레지스트리 로그인 & Push
docker login harbor.internal
docker push harbor.internal/aiops/confluence-sync:v1.0.0
3. Kubernetes CronJob manifest 작성
Git SSH Key 또는 Personal Access Token을 K8s Secret으로 관리하고, Git 저장소 동기화를 위한 Persistent Volume(PVC)을 마운트하여 배포합니다.
apiVersion: batch/v1
kind: CronJob
metadata:
name: confluence-aiops-sync
namespace: aiops
spec:
schedule: "*/10 * * * *" # 10분 주기 실행
concurrencyPolicy: Forbid # 이전 작업 미완료 시 중복 실행 방지 (Lock)
successfulJobsHistoryLimit: 3
failedJobsHistoryLimit: 5
jobTemplate:
spec:
template:
spec:
containers:
- name: sync-worker
image: harbor.internal/aiops/confluence-sync:v1.0.0
imagePullPolicy: IfNotPresent
env:
- name: CONFLUENCE_URL
value: "http://confluence.internal"
- name: CONFLUENCE_TOKEN
valueFrom:
secretKeyRef:
name: aiops-secrets
key: confluence-token
- name: OPENSEARCH_URL
value: "http://opensearch.internal:9200"
- name: QDRANT_URL
value: "http://qdrant.internal:6333"
volumeMounts:
- name: git-volume
mountPath: /workspace/ops-knowledge-repo
volumes:
- name: git-volume
persistentVolumeClaim:
claimName: ops-knowledge-git-pvc
restartPolicy: OnFailure
4. 파이프라인 검증 가이드 (동작 확인 항목)
CronJob 실행 후, 파이프라인이 정상 동작했는지 4개 레이어에서 검증을 수행합니다.
① CronJob 실행 상태 및 Pod 로그 확인
# CronJob 및 생성된 Job 확인
kubectl -n aiops get cronjob,jobs
# 최신 실행 Pod의 로그 실시간 확인
kubectl -n aiops logs -f job/confluence-aiops-sync-<POD_SUFFIX>
# 정상 출력 확인 키워드:
# - "Found N updated pages."
# - "Processing Page ID: CONF-10429"
# - "==> Sync Pipeline Completed Successfully."
② Git Repository (Bitbucket / GitLab) 확인
- Bitbucket/GitLab의
ops-knowledge/{Space}/디렉토리에 대상 Confluence 페이지가.md파일로 푸시되었는지 확인. .md파일 최상단에 YAML Front-Matter(id,title,version,tags등)가 정상 주입되었는지 확인.- Commit 히스토리에
docs(sync): update Confluence doc ...메시지가 남았는지 확인.
③ OpenSearch (Sparse Index) 검증
OpenSearch REST API를 호출하여 최근 수집된 Document의 BM25 키워드 인덱싱 결과를 확인합니다.
# 특정 doc_id 기준 쿼리
curl -X GET "http://opensearch.internal:9200/ops-knowledge-sparse/_search" \
-H 'Content-Type: application/json' \
-d '{
"query": {
"term": { "doc_id.keyword": "CONF-10429" }
}
}'
- 확인 포인트:
hits.total.value가 1 이상이며, Header 기반으로 잘린chunk_id(CONF-10429#0,CONF-10429#1)들과commit_hash가 저장되어 있는지 확인.
④ Qdrant (Dense Vector Index) 검증
Qdrant API를 통해 백터와 메타데이터 Payload가 제대로 Upsert 되었는지 확인합니다.
# Qdrant Payload 필터 쿼리
curl -X POST "http://qdrant.internal:6333/collections/ops-knowledge-dense/points/scroll" \
-H 'Content-Type: application/json' \
-d '{
"filter": {
"must": [
{ "key": "doc_id", "match": { "value": "CONF-10429" } }
]
},
"limit": 5,
"with_vector": false,
"with_payload": true
}'
- 확인 포인트:
points배열 내에payload.content와commit_hash,chunk_id가 빠짐없이 저장되었는지 확인.
engineer