Add chart flink cdc session cluster chart
This commit is contained in:
wbsong111
@@ -0,0 +1,105 @@
|
|||||||
|
# flink-cdc-session 버전 갱신 가이드
|
||||||
|
|
||||||
|
## 1. git 작업 환경 구성
|
||||||
|
|
||||||
|
- DIP 카탈로그 git 다운로드
|
||||||
|
```
|
||||||
|
$ git clone https://github.com/paasup/dip-catalog.git
|
||||||
|
```
|
||||||
|
|
||||||
|
- 작업 브랜치로 체크아웃
|
||||||
|
```
|
||||||
|
$ git checkout -b update-flink-cdc-session/0.1.0
|
||||||
|
```
|
||||||
|
|
||||||
|
## 2. helm 차트 개발 및 업데이트
|
||||||
|
|
||||||
|
- 현재 flink-cdc-session 차트는 직접 작성하여 업데이트 중입니다.
|
||||||
|
- 기존 오픈소스 차트를 기반으로 하지 않고, 프로젝트 요구사항에 맞게 커스텀 개발하고 있습니다.
|
||||||
|
|
||||||
|
### 2.1 차트 구조
|
||||||
|
```
|
||||||
|
flink-cdc-session/0.1.0/
|
||||||
|
├── Chart.yaml # 차트 메타데이터
|
||||||
|
├── values.yaml # 기본 설정값
|
||||||
|
├── custom-values.yaml # 커스텀 설정값
|
||||||
|
├── BUILD-README.md # 빌드 가이드 (본 파일)
|
||||||
|
├── CUSTOM-README.md # 배포 가이드
|
||||||
|
├── README.md # 차트 상세 문서
|
||||||
|
└── templates/
|
||||||
|
├── _helpers.tpl # 헬퍼 템플릿
|
||||||
|
├── configmap.yaml # flink-cdc.yaml + pipeline.yaml
|
||||||
|
├── flinkdeployment.yaml # Flink Session Cluster
|
||||||
|
├── ingress.yaml # Flink Web UI Ingress
|
||||||
|
├── job.yaml # CDC 파이프라인 제출 Job (조건부)
|
||||||
|
└── secrets.yaml # truststore / gcs-key Secret (조건부)
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2.2 개발 중인 주요 기능
|
||||||
|
- PostgreSQL → Apache Iceberg CDC 파이프라인 (Flink Session Mode)
|
||||||
|
- Flink CDC 3.5.0 기반
|
||||||
|
- S3/MinIO 및 GCS 체크포인트 저장소 지원
|
||||||
|
- 사설 CA 인증서(JKS truststore) 지원
|
||||||
|
- Iceberg REST 카탈로그(Lakekeeper) + OAuth2 연동
|
||||||
|
- Ingress 지원 (Flink Web UI)
|
||||||
|
|
||||||
|
## 3. 이미지 정보
|
||||||
|
|
||||||
|
- 기본 이미지: `paasup/flink-cdc-pipeline:3.5.0-flink1.20-r1`
|
||||||
|
- Flink CDC 3.5.0 + Flink 1.20 기반 커스텀 이미지
|
||||||
|
- PostgreSQL CDC 커넥터 및 Iceberg 싱크 포함
|
||||||
|
|
||||||
|
## 4. git push 및 tag 추가
|
||||||
|
|
||||||
|
- 갱신작업 진행후 commit
|
||||||
|
```sh
|
||||||
|
$ git add .
|
||||||
|
$ git commit -m "update flink-cdc-session/0.1.0"
|
||||||
|
```
|
||||||
|
|
||||||
|
- main 브랜치에 체크아웃 후 merge
|
||||||
|
```sh
|
||||||
|
$ git checkout main
|
||||||
|
$ git merge update-flink-cdc-session/0.1.0
|
||||||
|
```
|
||||||
|
|
||||||
|
- git에 push 후 작업 브랜치 삭제
|
||||||
|
```sh
|
||||||
|
$ git push -u origin main
|
||||||
|
$ git branch -d update-flink-cdc-session/0.1.0
|
||||||
|
```
|
||||||
|
|
||||||
|
- git tag 추가 후 push
|
||||||
|
```sh
|
||||||
|
$ git tag flink-cdc-session/0.1.0
|
||||||
|
$ git push origin flink-cdc-session/0.1.0
|
||||||
|
```
|
||||||
|
|
||||||
|
## 5. 차트 버전 정보
|
||||||
|
|
||||||
|
- flink-cdc-session/0.1.0
|
||||||
|
- Apache Flink CDC 3.5.0 + Flink 1.20 기반
|
||||||
|
- PostgreSQL WAL → Apache Iceberg REST 카탈로그 CDC 파이프라인
|
||||||
|
- S3/MinIO 및 GCS 체크포인트 저장소 지원
|
||||||
|
- 사설 CA 인증서(JKS truststore) 지원
|
||||||
|
- Iceberg REST 카탈로그(Lakekeeper) + Keycloak OAuth2 연동
|
||||||
|
- 서비스 배포를 위하여 custom-values.yaml에 정의하였다.
|
||||||
|
- 차트의 빌드 방법과 배포 방법을 BUILD-README.md, CUSTOM-README.md 문서에 작성하였다.
|
||||||
|
|
||||||
|
## 6. 개발 참고사항
|
||||||
|
|
||||||
|
### 6.1 주요 의존성
|
||||||
|
- Flink Kubernetes Operator 설치 필요
|
||||||
|
- Kubernetes 1.25+ 권장
|
||||||
|
- Helm 3.x 필요
|
||||||
|
- PostgreSQL logical replication 활성화 필요 (`wal_level=logical`)
|
||||||
|
|
||||||
|
### 6.2 보안 설정
|
||||||
|
- JKS 기반 truststore로 사설 CA 인증서 관리
|
||||||
|
- Kubernetes Secret을 통한 MinIO/GCS 자격증명 관리
|
||||||
|
- OAuth2 client_id:client_secret을 통한 Lakekeeper 인증
|
||||||
|
|
||||||
|
## 참고) 관련 문서
|
||||||
|
- [Flink CDC 공식 문서](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/)
|
||||||
|
- [Flink Kubernetes Operator 문서](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/)
|
||||||
|
- [Apache Iceberg 문서](https://iceberg.apache.org/docs/latest/)
|
||||||
@@ -0,0 +1,165 @@
|
|||||||
|
# Flink CDC Session 배포
|
||||||
|
|
||||||
|
## 1. 배포 방법
|
||||||
|
|
||||||
|
```sh
|
||||||
|
helm upgrade --install flink-cdc-session ./ \
|
||||||
|
--namespace flink-test \
|
||||||
|
--create-namespace \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml
|
||||||
|
```
|
||||||
|
|
||||||
|
### Session Cluster 먼저 배포 후 Job 활성화 (권장)
|
||||||
|
|
||||||
|
```sh
|
||||||
|
# 1단계: Session Cluster만 배포 (custom-values.yaml에 job.enabled: false 설정)
|
||||||
|
helm upgrade --install flink-cdc-session ./ \
|
||||||
|
--namespace flink-test \
|
||||||
|
--create-namespace \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml
|
||||||
|
|
||||||
|
# 2단계: Cluster RUNNING 확인
|
||||||
|
kubectl get flinkdeployment -n flink-test -w
|
||||||
|
|
||||||
|
# 3단계: Job 포함해서 upgrade
|
||||||
|
helm upgrade flink-cdc-session ./ \
|
||||||
|
--namespace flink-test \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml \
|
||||||
|
--set job.enabled=true
|
||||||
|
```
|
||||||
|
|
||||||
|
## 2. custom-values.yaml 설명
|
||||||
|
|
||||||
|
### 1) 체크포인트 저장소
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `checkpointStorage.storageType` | 저장소 유형 (`s3` \| `gcs`) | `s3` |
|
||||||
|
| `checkpointStorage.checkpointDir` | 체크포인트 저장 경로 | `s3://flink/checkpoints` |
|
||||||
|
| `checkpointStorage.savepointDir` | 세이브포인트 저장 경로 | `s3://flink/savepoints` |
|
||||||
|
| `checkpointStorage.s3.endpoint` | S3/MinIO 엔드포인트 | `http://minio.minio.svc.cluster.local:9000` |
|
||||||
|
| `checkpointStorage.s3.pathStyleAccess` | Path-style 접근 여부 (MinIO는 `"true"`) | `"true"` |
|
||||||
|
| `checkpointStorage.s3.existingSecret` | 자격증명 Secret 이름 | `minio-credentials` |
|
||||||
|
|
||||||
|
### 2) Ingress 설정
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `ingress.enabled` | Ingress 활성화 여부 | `true` |
|
||||||
|
| `ingress.host` | Flink Web UI 호스트명 | `flink-cdc-session.example.org` |
|
||||||
|
| `ingress.className` | IngressClass 이름 | `kong` |
|
||||||
|
| `ingress.annotations` | Ingress 어노테이션 | cert-manager + Kong HTTPS 설정 |
|
||||||
|
|
||||||
|
### 3) Truststore (사설 CA 환경)
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `truststore.enabled` | 외부 JKS 사용 여부 | `true` |
|
||||||
|
| `truststore.secretName` | truststore Secret 이름 | `flink-truststore` |
|
||||||
|
|
||||||
|
### 4) Job 설정
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `job.enabled` | CDC Job 활성화 여부 | `false` |
|
||||||
|
|
||||||
|
> **주의**: Session Cluster가 RUNNING 상태가 된 후 `job.enabled: true`로 upgrade하세요.
|
||||||
|
|
||||||
|
### 5) Flink 클러스터 리소스 (values.yaml에서 조정)
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `jobManager.memory` | JobManager 메모리 | `1024m` |
|
||||||
|
| `jobManager.cpu` | JobManager CPU | `0.5` |
|
||||||
|
| `taskManager.replicas` | TaskManager 인스턴스 수 | `3` |
|
||||||
|
| `taskManager.memory` | TaskManager 메모리 | `2048m` |
|
||||||
|
| `taskManager.cpu` | TaskManager CPU | `1` |
|
||||||
|
| `taskManager.taskSlots` | TaskManager 슬롯 수 | `2` |
|
||||||
|
|
||||||
|
### 6) PostgreSQL 소스 (values.yaml에서 조정)
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `postgres.hostname` | PostgreSQL 호스트 | `postgres-postgresql.flink-test.svc.cluster.local` |
|
||||||
|
| `postgres.port` | PostgreSQL 포트 | `5432` |
|
||||||
|
| `postgres.username` | CDC 전용 사용자 | `flink_cdc` |
|
||||||
|
| `postgres.password` | 사용자 패스워드 | `password` |
|
||||||
|
| `postgres.slotName` | Replication slot 이름 | `flink_cdc_slot` |
|
||||||
|
| `postgres.tables` | CDC 대상 테이블 목록 | `testdb.public.orders,testdb.public.users` |
|
||||||
|
|
||||||
|
### 7) Iceberg 싱크 (values.yaml에서 조정)
|
||||||
|
|
||||||
|
| Name | 설명 | 기본값 |
|
||||||
|
|------|------|--------|
|
||||||
|
| `sink.catalog.uri` | Lakekeeper REST URI | `https://lakekeeper.example.org/catalog` |
|
||||||
|
| `sink.catalog.warehouse` | 카탈로그 warehouse 이름 | `minio` |
|
||||||
|
| `sink.catalog.s3Endpoint` | Iceberg 데이터 저장소 엔드포인트 | `https://minio.example.org` |
|
||||||
|
| `sink.catalog.oauth2Uri` | Keycloak OAuth2 토큰 URI | `https://keycloak.example.org/…` |
|
||||||
|
| `sink.catalog.credential` | `client_id:client_secret` | - |
|
||||||
|
| `sink.catalog.scope` | OAuth2 scope | `lakekeeper` |
|
||||||
|
|
||||||
|
## 3. 사전 준비: Secret 생성
|
||||||
|
|
||||||
|
### 3.1 MinIO 자격증명 Secret
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl create secret generic minio-credentials \
|
||||||
|
--from-literal=access-key-id=<ACCESS_KEY> \
|
||||||
|
--from-literal=secret-access-key=<SECRET_KEY> \
|
||||||
|
-n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
### 3.2 Truststore Secret (`truststore.enabled: true` 인 경우)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# JVM 기본 cacerts를 베이스로 truststore 생성
|
||||||
|
IMAGE="paasup/flink-cdc-pipeline:3.5.0-flink1.20-r1"
|
||||||
|
docker run --rm --entrypoint="" "${IMAGE}" \
|
||||||
|
cat /opt/java/openjdk/lib/security/cacerts > truststore.jks
|
||||||
|
|
||||||
|
# 사설 CA 추가 (cert-manager 사용 시)
|
||||||
|
kubectl get secret root-ca-secret -n cert-manager \
|
||||||
|
-o jsonpath='{.data.ca\.crt}' | base64 -d > /tmp/internal-ca.pem
|
||||||
|
|
||||||
|
keytool -import -noprompt \
|
||||||
|
-keystore truststore.jks \
|
||||||
|
-storepass changeit \
|
||||||
|
-alias internal-root-ca \
|
||||||
|
-file /tmp/internal-ca.pem
|
||||||
|
|
||||||
|
# Kubernetes Secret 생성
|
||||||
|
kubectl create secret generic flink-truststore \
|
||||||
|
--from-file=truststore.jks=./truststore.jks \
|
||||||
|
-n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
> 공인 CA 환경(Let's Encrypt 등)에서는 `truststore.enabled: false`로 설정하면 truststore Secret 불필요합니다.
|
||||||
|
|
||||||
|
## 4. 배포 후 확인
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Session Cluster 상태 확인
|
||||||
|
kubectl get flinkdeployment -n flink-test
|
||||||
|
|
||||||
|
# Pod 상태 확인
|
||||||
|
kubectl get pods -n flink-test
|
||||||
|
|
||||||
|
# Job 로그 확인
|
||||||
|
kubectl logs -n flink-test job/submit-cdc
|
||||||
|
|
||||||
|
# Flink Web UI (포트포워딩)
|
||||||
|
kubectl port-forward svc/flink-cdc-session-rest 8081:8081 -n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
## 5. 삭제
|
||||||
|
|
||||||
|
```bash
|
||||||
|
helm uninstall flink-cdc-session -n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
## 6. 참고
|
||||||
|
|
||||||
|
자세한 파라미터 설명, truststore 생성 방법, 렌더링 확인 방법은 [README.md](README.md)를 참조하세요.
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
apiVersion: v2
|
||||||
|
appVersion: 3.5.0-flink1.20
|
||||||
|
description: Flink CDC Session Mode cluster with PostgreSQL-to-Iceberg pipeline
|
||||||
|
name: flink-cdc-session
|
||||||
|
type: application
|
||||||
|
version: 0.1.0
|
||||||
@@ -0,0 +1,472 @@
|
|||||||
|
# flink-cdc-session Helm Chart
|
||||||
|
|
||||||
|
PostgreSQL → Apache Iceberg CDC 파이프라인을 Flink Session Mode로 배포하는 Helm chart입니다.
|
||||||
|
|
||||||
|
## 아키텍처
|
||||||
|
|
||||||
|
```
|
||||||
|
PostgreSQL (WAL)
|
||||||
|
│ pgoutput
|
||||||
|
▼
|
||||||
|
Flink Session Cluster (FlinkDeployment)
|
||||||
|
│ Flink CDC 3.5.0
|
||||||
|
▼
|
||||||
|
Iceberg REST Catalog (Lakekeeper)
|
||||||
|
│ S3FileIO
|
||||||
|
▼
|
||||||
|
Object Storage (MinIO / GCS / S3)
|
||||||
|
```
|
||||||
|
|
||||||
|
## 사전 요구사항
|
||||||
|
|
||||||
|
| 항목 | 버전 |
|
||||||
|
|------|------|
|
||||||
|
| Helm | 3.x |
|
||||||
|
| Flink Kubernetes Operator | 1.x |
|
||||||
|
| Kubernetes | 1.25+ |
|
||||||
|
|
||||||
|
## 파일 구조
|
||||||
|
|
||||||
|
```
|
||||||
|
flink-cdc-session/
|
||||||
|
├── Chart.yaml
|
||||||
|
├── values.yaml
|
||||||
|
├── custom-values.yaml # 환경별 오버라이드 (배포 시 사용)
|
||||||
|
└── templates/
|
||||||
|
├── _helpers.tpl
|
||||||
|
├── configmap.yaml # flink-cdc.yaml + pipeline.yaml
|
||||||
|
├── flinkdeployment.yaml # Flink Session Cluster
|
||||||
|
├── job.yaml # CDC 파이프라인 제출 Job (조건부)
|
||||||
|
└── secrets.yaml # truststore / gcs-key Secret (조건부)
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 외부 JKS(truststore) 사용 여부 결정
|
||||||
|
|
||||||
|
이 chart는 `truststore.enabled` 플래그로 외부 JKS 사용 여부를 제어합니다.
|
||||||
|
**배포 전에 반드시 아래 기준에 따라 값을 결정하세요.**
|
||||||
|
|
||||||
|
### 판단 기준
|
||||||
|
|
||||||
|
Flink는 다음 엔드포인트와 HTTPS로 통신합니다.
|
||||||
|
|
||||||
|
| 엔드포인트 | 용도 |
|
||||||
|
|-----------|------|
|
||||||
|
| Lakekeeper (`sink.catalog.uri`) | Iceberg REST 카탈로그 |
|
||||||
|
| Keycloak (`sink.catalog.oauth2Uri`) | OAuth2 토큰 발급 |
|
||||||
|
| MinIO (`sink.catalog.s3Endpoint`) | Iceberg 데이터 파일 읽기/쓰기 |
|
||||||
|
|
||||||
|
**`truststore.enabled: true` (외부 JKS 사용)**
|
||||||
|
→ 위 엔드포인트 중 하나라도 **사설 CA**가 서명한 인증서를 사용하는 경우
|
||||||
|
|
||||||
|
**`truststore.enabled: false` (외부 JKS 불필요)**
|
||||||
|
→ 모든 엔드포인트가 **공인 CA**(Let's Encrypt, DigiCert 등)로 서명된 인증서를 사용하는 경우
|
||||||
|
→ JVM 기본 cacerts만으로 TLS 검증이 가능하며, `flink-truststore` Secret도 불필요
|
||||||
|
|
||||||
|
> **왜 외부 JKS가 필요한가**
|
||||||
|
> Flink JVM 옵션 `-Djavax.net.ssl.trustStore`는 JVM 기본 cacerts를 **완전히 교체**합니다.
|
||||||
|
> 따라서 사설 CA 인증서뿐만 아니라 공인 CA 체인도 포함된 truststore를 직접 만들어야 합니다.
|
||||||
|
> 반대로 이 옵션을 설정하지 않으면 JVM 기본 cacerts가 그대로 사용됩니다.
|
||||||
|
|
||||||
|
### 엔드포인트 인증서 확인 방법
|
||||||
|
|
||||||
|
로컬 또는 클러스터 내부에서 아래 명령으로 인증서 발급자를 확인합니다.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 인증서 발급자(Issuer) 확인
|
||||||
|
openssl s_client -connect lakekeeper.example.org:443 -showcerts </dev/null 2>/dev/null \
|
||||||
|
| openssl x509 -noout -issuer -subject
|
||||||
|
|
||||||
|
openssl s_client -connect keycloak.example.org:443 -showcerts </dev/null 2>/dev/null \
|
||||||
|
| openssl x509 -noout -issuer -subject
|
||||||
|
```
|
||||||
|
|
||||||
|
출력 예시:
|
||||||
|
|
||||||
|
```
|
||||||
|
# 사설 CA → truststore.enabled: true 필요
|
||||||
|
issuer=O=paasup, CN=paasup-root-ca
|
||||||
|
subject=CN=lakekeeper.example.org
|
||||||
|
|
||||||
|
# 공인 CA → truststore.enabled: false 가능
|
||||||
|
issuer=C=US, O=Let's Encrypt, CN=R11
|
||||||
|
subject=CN=lakekeeper.example.org
|
||||||
|
```
|
||||||
|
|
||||||
|
클러스터 내부에서 확인하는 경우:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl run ssl-check --rm -it --image=alpine/openssl --restart=Never -- \
|
||||||
|
s_client -connect lakekeeper.example.org:443 -showcerts </dev/null 2>/dev/null \
|
||||||
|
| grep -E "issuer|subject"
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## Secret 준비
|
||||||
|
|
||||||
|
### 1. truststore.jks 생성 (`truststore.enabled: true` 인 경우만 필요)
|
||||||
|
|
||||||
|
#### Step 1 — JVM 기본 cacerts 추출
|
||||||
|
|
||||||
|
Docker 이미지 내부의 JVM cacerts를 truststore 베이스로 사용합니다.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
IMAGE="wbsong111/flink-cdc-pipeline:3.5.0-flink1.20-r1"
|
||||||
|
|
||||||
|
docker run --rm --entrypoint="" "${IMAGE}" \
|
||||||
|
cat /opt/java/openjdk/lib/security/cacerts > truststore.jks
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Step 2 — 사설 CA 추가
|
||||||
|
|
||||||
|
**cert-manager ClusterIssuer에서 추출하는 경우:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl get secret root-ca-secret -n cert-manager \
|
||||||
|
-o jsonpath='{.data.ca\.crt}' | base64 -d > /tmp/internal-ca.pem
|
||||||
|
|
||||||
|
keytool -import -noprompt \
|
||||||
|
-keystore truststore.jks \
|
||||||
|
-storepass changeit \
|
||||||
|
-alias internal-root-ca \
|
||||||
|
-file /tmp/internal-ca.pem
|
||||||
|
|
||||||
|
rm -f /tmp/internal-ca.pem
|
||||||
|
```
|
||||||
|
|
||||||
|
**PEM 파일에서 직접 추가하는 경우:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
keytool -import -noprompt \
|
||||||
|
-keystore truststore.jks \
|
||||||
|
-storepass changeit \
|
||||||
|
-alias my-ca \
|
||||||
|
-file ./certs/my-ca.crt
|
||||||
|
```
|
||||||
|
|
||||||
|
추가된 인증서 목록 확인:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
keytool -list -keystore truststore.jks -storepass changeit | tail -5
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Step 3 — truststore 동작 검증
|
||||||
|
|
||||||
|
배포 전에 truststore로 실제 엔드포인트 TLS 연결이 성공하는지 확인합니다.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 사설 CA PEM으로 curl 검증
|
||||||
|
curl -v --cacert /tmp/internal-ca.pem https://lakekeeper.example.org/catalog 2>&1 \
|
||||||
|
| grep -E "SSL|TLS|certificate"
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Step 4 — Kubernetes Secret 생성
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl create secret generic flink-truststore \
|
||||||
|
--from-file=truststore.jks=./truststore.jks \
|
||||||
|
-n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
기존 Secret 업데이트:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl create secret generic flink-truststore \
|
||||||
|
--from-file=truststore.jks=./truststore.jks \
|
||||||
|
-n flink-test \
|
||||||
|
--dry-run=client -o yaml | kubectl apply -f -
|
||||||
|
```
|
||||||
|
|
||||||
|
> `create-truststore.sh` 스크립트로 Step 1~4를 한번에 실행할 수 있습니다:
|
||||||
|
> ```bash
|
||||||
|
> bash ../../create-truststore.sh
|
||||||
|
> ```
|
||||||
|
|
||||||
|
#### 배포 후 마운트 확인
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# truststore.jks가 Pod에 정상 마운트되었는지 확인
|
||||||
|
kubectl exec -n flink-test deploy/flink-cdc-session -- ls -la /opt/flink/certs/
|
||||||
|
|
||||||
|
# JVM이 truststore를 인식하는지 확인 (JobManager 로그에서 SSL 관련 출력)
|
||||||
|
kubectl logs -n flink-test -l component=jobmanager | grep -i "truststore\|ssl\|tls" | head -20
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
### 2. 체크포인트 저장소 자격증명
|
||||||
|
|
||||||
|
`checkpointStorage.storageType` 값에 따라 필요한 Secret이 다릅니다.
|
||||||
|
|
||||||
|
#### MinIO / S3 — existingSecret 방식 (권장)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl create secret generic minio-credentials \
|
||||||
|
--from-literal=access-key-id=<ACCESS_KEY> \
|
||||||
|
--from-literal=secret-access-key=<SECRET_KEY> \
|
||||||
|
-n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
`custom-values.yaml`에서 참조:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
checkpointStorage:
|
||||||
|
storageType: s3
|
||||||
|
checkpointDir: s3://flink/checkpoints
|
||||||
|
savepointDir: s3://flink/savepoints
|
||||||
|
s3:
|
||||||
|
endpoint: "http://minio.minio.svc.cluster.local:9000" # 클러스터 내부 MinIO
|
||||||
|
pathStyleAccess: "true"
|
||||||
|
existingSecret: minio-credentials
|
||||||
|
```
|
||||||
|
|
||||||
|
클러스터 내부 MinIO를 사용할 경우 plain HTTP 엔드포인트(`http://`)를 사용하면 TLS 없이 통신할 수 있습니다.
|
||||||
|
|
||||||
|
#### GCS — 서비스 계정 키 방식
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl create secret generic gcs-key \
|
||||||
|
--from-file=key.json=./gcs_key/key.json \
|
||||||
|
-n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
`custom-values.yaml`에서 참조:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
checkpointStorage:
|
||||||
|
storageType: gcs
|
||||||
|
checkpointDir: gs://my-bucket/flink/checkpoints
|
||||||
|
savepointDir: gs://my-bucket/flink/savepoints
|
||||||
|
gcs:
|
||||||
|
credentialSecret: gcs-key
|
||||||
|
credentialPath: /opt/flink/gcs-key/key.json
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 배포
|
||||||
|
|
||||||
|
### custom-values.yaml 사용 (권장)
|
||||||
|
|
||||||
|
환경별 설정은 `values.yaml`을 수정하지 않고 `custom-values.yaml`에 오버라이드합니다.
|
||||||
|
|
||||||
|
```bash
|
||||||
|
helm install flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml
|
||||||
|
```
|
||||||
|
|
||||||
|
### Session Cluster만 먼저 배포 (Job 제외)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 1단계: Session Cluster만 배포 (custom-values.yaml에 job.enabled: false 설정)
|
||||||
|
helm install flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml
|
||||||
|
|
||||||
|
# 2단계: Cluster 기동 확인
|
||||||
|
kubectl get flinkdeployment -n flink-test -w
|
||||||
|
|
||||||
|
# 3단계: Job 포함해서 업그레이드
|
||||||
|
helm upgrade flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml \
|
||||||
|
--set job.enabled=true
|
||||||
|
```
|
||||||
|
|
||||||
|
### truststore 불필요한 경우 (공인 CA 환경)
|
||||||
|
|
||||||
|
```bash
|
||||||
|
helm install flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
-f values.yaml \
|
||||||
|
-f custom-values.yaml \
|
||||||
|
--set truststore.enabled=false
|
||||||
|
```
|
||||||
|
|
||||||
|
`truststore.enabled=false` 시 다음이 모두 제거됩니다:
|
||||||
|
- `env.java.opts.jobmanager/taskmanager`의 SSL JVM 옵션
|
||||||
|
- Pod의 truststore volumeMount 및 volume
|
||||||
|
|
||||||
|
### Secret도 chart에서 생성하는 경우
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# truststore Secret chart 생성
|
||||||
|
helm install flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
--set truststore.create=true \
|
||||||
|
--set truststore.base64jks="$(base64 < truststore.jks)"
|
||||||
|
|
||||||
|
# GCS Secret chart 생성
|
||||||
|
helm install flink-cdc-session ./flink/helm/flink-cdc-session \
|
||||||
|
-n flink-test \
|
||||||
|
--set checkpointStorage.storageType=gcs \
|
||||||
|
--set checkpointStorage.gcs.create=true \
|
||||||
|
--set "checkpointStorage.gcs.base64json=$(base64 < gcs_key/key.json)"
|
||||||
|
```
|
||||||
|
|
||||||
|
### 삭제
|
||||||
|
|
||||||
|
```bash
|
||||||
|
helm uninstall flink-cdc-session -n flink-test
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## values.yaml 파라미터
|
||||||
|
|
||||||
|
### 이미지
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `image.repository` | `wbsong111/flink-cdc-pipeline` | 컨테이너 이미지 저장소 |
|
||||||
|
| `image.tag` | `3.5.0-flink1.20-r1` | 이미지 태그 |
|
||||||
|
| `image.pullPolicy` | `Always` | 이미지 풀 정책 |
|
||||||
|
|
||||||
|
### Flink 클러스터
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `flinkVersion` | `v1_20` | Flink 버전 |
|
||||||
|
| `serviceAccount` | `flink` | Pod 서비스 어카운트 |
|
||||||
|
| `jobManager.replicas` | `1` | JobManager 인스턴스 수 |
|
||||||
|
| `jobManager.memory` | `1024m` | JobManager 메모리 |
|
||||||
|
| `jobManager.cpu` | `0.5` | JobManager CPU |
|
||||||
|
| `taskManager.replicas` | `3` | TaskManager 인스턴스 수 |
|
||||||
|
| `taskManager.memory` | `2048m` | TaskManager 메모리 |
|
||||||
|
| `taskManager.cpu` | `1` | TaskManager CPU |
|
||||||
|
| `taskManager.taskSlots` | `2` | TaskManager 슬롯 수 |
|
||||||
|
|
||||||
|
### CheckpointStorage (저장 위치 + 자격증명)
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `checkpointStorage.storageType` | `s3` | 저장소 유형 (`gcs` \| `s3`) |
|
||||||
|
| `checkpointStorage.checkpointDir` | `s3://my-bucket/…` | 체크포인트 저장 경로 |
|
||||||
|
| `checkpointStorage.savepointDir` | `s3://my-bucket/…` | 세이브포인트 저장 경로 |
|
||||||
|
| `checkpointStorage.s3.endpoint` | `""` | S3/MinIO 엔드포인트. 비워두면 AWS S3 |
|
||||||
|
| `checkpointStorage.s3.pathStyleAccess` | `"false"` | Path-style 접근 여부. MinIO는 `"true"` |
|
||||||
|
| `checkpointStorage.s3.existingSecret` | `""` | 자격증명 Secret 이름. 설정 시 아래 두 항목보다 우선 |
|
||||||
|
| `checkpointStorage.s3.accessKeyIdKey` | `access-key-id` | Secret 내 Access Key ID 필드명 |
|
||||||
|
| `checkpointStorage.s3.secretAccessKeyKey` | `secret-access-key` | Secret 내 Secret Access Key 필드명 |
|
||||||
|
| `checkpointStorage.s3.accessKeyId` | `""` | Access Key ID 직접 입력 (테스트용) |
|
||||||
|
| `checkpointStorage.s3.secretAccessKey` | `""` | Secret Access Key 직접 입력 (테스트용) |
|
||||||
|
| `checkpointStorage.gcs.credentialSecret` | `gcs-key` | GCS 키 Secret 이름 |
|
||||||
|
| `checkpointStorage.gcs.credentialPath` | `/opt/flink/gcs-key/key.json` | 컨테이너 내 키 파일 경로 |
|
||||||
|
| `checkpointStorage.gcs.create` | `false` | `true`이면 chart가 Secret 생성 |
|
||||||
|
| `checkpointStorage.gcs.base64json` | `""` | `create: true` 시 필수, base64 인코딩된 key.json |
|
||||||
|
|
||||||
|
### 체크포인트 동작 설정
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `checkpoint.interval` | `60s` | 체크포인트 주기 |
|
||||||
|
| `checkpoint.mode` | `EXACTLY_ONCE` | 체크포인트 모드 |
|
||||||
|
|
||||||
|
### Truststore (외부 JKS)
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `truststore.enabled` | `true` | 외부 JKS 사용 여부. 사설 CA 환경이면 `true` |
|
||||||
|
| `truststore.secretName` | `flink-truststore` | truststore Secret 이름 |
|
||||||
|
| `truststore.password` | `changeit` | JKS 패스워드 |
|
||||||
|
| `truststore.create` | `false` | `true`이면 chart가 Secret 생성 |
|
||||||
|
| `truststore.base64jks` | `""` | `create: true` 시 필수, base64 인코딩된 .jks |
|
||||||
|
|
||||||
|
### PostgreSQL 소스
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `postgres.hostname` | `postgres-postgresql.flink-test.svc…` | PostgreSQL 호스트 |
|
||||||
|
| `postgres.port` | `5432` | PostgreSQL 포트 |
|
||||||
|
| `postgres.username` | `flink_cdc` | CDC 전용 사용자 |
|
||||||
|
| `postgres.password` | `password` | 사용자 패스워드 |
|
||||||
|
| `postgres.slotName` | `flink_cdc_slot` | Replication slot 이름 |
|
||||||
|
| `postgres.decodingPlugin` | `pgoutput` | WAL 디코딩 플러그인 |
|
||||||
|
| `postgres.tables` | `testdb.public.orders,…` | CDC 대상 테이블 목록 |
|
||||||
|
|
||||||
|
### 파이프라인
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `flinkCdc.parallelism` | `1` | Flink CDC 글로벌 병렬도 |
|
||||||
|
| `flinkCdc.schemaChangeBehavior` | `EVOLVE` | 스키마 변경 처리 방식 |
|
||||||
|
| `pipeline.name` | `pg-to-iceberg-test` | 파이프라인 이름 |
|
||||||
|
| `pipeline.parallelism` | `2` | 파이프라인 병렬도 |
|
||||||
|
| `pipeline.checkpointInterval` | `60s` | 파이프라인 체크포인트 주기 |
|
||||||
|
|
||||||
|
### Iceberg 싱크
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `sink.catalog.uri` | `https://lakekeeper.example.org/catalog` | Lakekeeper REST URI |
|
||||||
|
| `sink.catalog.warehouse` | `minio` | 카탈로그 warehouse 이름 |
|
||||||
|
| `sink.catalog.s3Endpoint` | `https://minio.example.org` | S3 호환 오브젝트 스토리지 엔드포인트 |
|
||||||
|
| `sink.catalog.s3PathStyleAccess` | `"true"` | Path-style S3 접근 여부 |
|
||||||
|
| `sink.catalog.oauth2Uri` | `https://keycloak.example.org/…` | Keycloak OAuth2 토큰 URI |
|
||||||
|
| `sink.catalog.credential` | `lakekeeper-admin:…` | `client_id:client_secret` |
|
||||||
|
| `sink.catalog.scope` | `lakekeeper` | OAuth2 scope |
|
||||||
|
|
||||||
|
### 라우팅
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
route:
|
||||||
|
- sourceTable: public.orders # PostgreSQL schema.table
|
||||||
|
sinkTable: testdb_public.orders # Iceberg namespace.table
|
||||||
|
- sourceTable: public.users
|
||||||
|
sinkTable: testdb_public.users
|
||||||
|
```
|
||||||
|
|
||||||
|
### Job
|
||||||
|
|
||||||
|
| 파라미터 | 기본값 | 설명 |
|
||||||
|
|----------|--------|------|
|
||||||
|
| `job.enabled` | `true` | `false`이면 Job 리소스 미생성 |
|
||||||
|
| `job.name` | `submit-cdc` | Job 리소스 이름 |
|
||||||
|
| `job.restAddress` | `flink-cdc-session-rest` | Session Cluster REST 서비스명 |
|
||||||
|
| `job.restPort` | `8081` | REST 포트 |
|
||||||
|
| `job.pipelineFile` | `./conf/pipeline.yaml` | 파이프라인 설정 파일 경로 |
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
## 렌더링 확인
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 전체 렌더링 확인
|
||||||
|
helm template flink-cdc-session . -n flink-test \
|
||||||
|
-f values.yaml -f custom-values.yaml
|
||||||
|
|
||||||
|
# lint 검사
|
||||||
|
helm lint .
|
||||||
|
|
||||||
|
# S3/MinIO 렌더링 확인
|
||||||
|
helm template flink-cdc-session . -n flink-test \
|
||||||
|
-f values.yaml -f custom-values.yaml \
|
||||||
|
| grep -E "checkpointDir|savepointDir|AWS_ACCESS|s3.endpoint"
|
||||||
|
|
||||||
|
# truststore 비활성화 시 SSL 옵션이 제거되는지 확인
|
||||||
|
helm template flink-cdc-session . -n flink-test \
|
||||||
|
-f values.yaml -f custom-values.yaml \
|
||||||
|
--set truststore.enabled=false | grep -A3 "java.opts"
|
||||||
|
```
|
||||||
|
|
||||||
|
## 배포 후 확인
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Session Cluster 상태 확인
|
||||||
|
kubectl get flinkdeployment -n flink-test
|
||||||
|
|
||||||
|
# Job 로그 확인
|
||||||
|
kubectl logs -n flink-test job/submit-cdc
|
||||||
|
|
||||||
|
# Job 삭제 후 재제출
|
||||||
|
kubectl delete job submit-cdc -n flink-test
|
||||||
|
helm upgrade flink-cdc-session . -n flink-test \
|
||||||
|
-f values.yaml -f custom-values.yaml \
|
||||||
|
--set job.enabled=true
|
||||||
|
```
|
||||||
@@ -0,0 +1,31 @@
|
|||||||
|
# custom-values.yaml
|
||||||
|
# 실제 배포에 사용하는 환경별 오버라이드 값
|
||||||
|
# 사용법: helm install ... -f values.yaml -f custom-values.yaml
|
||||||
|
|
||||||
|
checkpointStorage:
|
||||||
|
storageType: s3
|
||||||
|
checkpointDir: s3://flink/checkpoints
|
||||||
|
savepointDir: s3://flink/savepoints
|
||||||
|
s3:
|
||||||
|
endpoint: "http://minio.minio.svc.cluster.local:9000"
|
||||||
|
pathStyleAccess: "true"
|
||||||
|
existingSecret: minio-credentials
|
||||||
|
|
||||||
|
ingress:
|
||||||
|
enabled: true
|
||||||
|
host: "flink-cdc-session.example.org"
|
||||||
|
className: kong
|
||||||
|
annotations:
|
||||||
|
cert-manager.io/cluster-issuer: root-ca-issuer
|
||||||
|
cert-manager.io/duration: 8760h
|
||||||
|
cert-manager.io/renew-before: 720h
|
||||||
|
konghq.com/protocols: https
|
||||||
|
konghq.com/https-redirect-status-code: "301"
|
||||||
|
|
||||||
|
truststore:
|
||||||
|
enabled: true
|
||||||
|
secretName: flink-truststore
|
||||||
|
|
||||||
|
# Session Cluster만 먼저 배포. RUNNING 확인 후 job.enabled: true로 upgrade.
|
||||||
|
job:
|
||||||
|
enabled: false
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
{{/*
|
||||||
|
Chart label
|
||||||
|
*/}}
|
||||||
|
{{- define "flink-cdc-session.chart" -}}
|
||||||
|
{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
|
||||||
|
{{- end }}
|
||||||
|
|
||||||
|
{{/*
|
||||||
|
Common labels
|
||||||
|
*/}}
|
||||||
|
{{- define "flink-cdc-session.labels" -}}
|
||||||
|
helm.sh/chart: {{ include "flink-cdc-session.chart" . }}
|
||||||
|
app.kubernetes.io/name: {{ .Chart.Name }}
|
||||||
|
app.kubernetes.io/instance: {{ .Release.Name }}
|
||||||
|
{{- if .Chart.AppVersion }}
|
||||||
|
app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
|
||||||
|
{{- end }}
|
||||||
|
app.kubernetes.io/managed-by: {{ .Release.Service }}
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,45 @@
|
|||||||
|
apiVersion: v1
|
||||||
|
kind: ConfigMap
|
||||||
|
metadata:
|
||||||
|
name: cdc-pipeline-config
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
data:
|
||||||
|
flink-cdc.yaml: |
|
||||||
|
parallelism: {{ .Values.flinkCdc.parallelism }}
|
||||||
|
schema.change.behavior: {{ .Values.flinkCdc.schemaChangeBehavior }}
|
||||||
|
|
||||||
|
pipeline.yaml: |
|
||||||
|
source:
|
||||||
|
type: postgres
|
||||||
|
hostname: {{ .Values.postgres.hostname }}
|
||||||
|
port: {{ .Values.postgres.port }}
|
||||||
|
username: {{ .Values.postgres.username }}
|
||||||
|
password: {{ .Values.postgres.password }}
|
||||||
|
decoding.plugin.name: {{ .Values.postgres.decodingPlugin }}
|
||||||
|
slot.name: {{ .Values.postgres.slotName }}
|
||||||
|
tables: {{ .Values.postgres.tables }}
|
||||||
|
|
||||||
|
pipeline:
|
||||||
|
name: {{ .Values.pipeline.name }}
|
||||||
|
parallelism: {{ .Values.pipeline.parallelism }}
|
||||||
|
execution.checkpointing.interval: {{ .Values.pipeline.checkpointInterval }}
|
||||||
|
|
||||||
|
sink:
|
||||||
|
type: iceberg
|
||||||
|
catalog.properties.type: rest
|
||||||
|
catalog.properties.uri: {{ .Values.sink.catalog.uri | quote }}
|
||||||
|
catalog.properties.warehouse: {{ .Values.sink.catalog.warehouse }}
|
||||||
|
catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO
|
||||||
|
catalog.properties.s3.endpoint: {{ .Values.sink.catalog.s3Endpoint | quote }}
|
||||||
|
catalog.properties.s3.path-style-access: {{ .Values.sink.catalog.s3PathStyleAccess | quote }}
|
||||||
|
catalog.properties.oauth2-server-uri: {{ .Values.sink.catalog.oauth2Uri | quote }}
|
||||||
|
catalog.properties.credential: {{ .Values.sink.catalog.credential }}
|
||||||
|
catalog.properties.scope: {{ .Values.sink.catalog.scope }}
|
||||||
|
|
||||||
|
route:
|
||||||
|
{{- range .Values.route }}
|
||||||
|
- source-table: {{ .sourceTable }}
|
||||||
|
sink-table: {{ .sinkTable }}
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,103 @@
|
|||||||
|
apiVersion: flink.apache.org/v1beta1
|
||||||
|
kind: FlinkDeployment
|
||||||
|
metadata:
|
||||||
|
name: flink-cdc-session
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
spec:
|
||||||
|
image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
|
||||||
|
imagePullPolicy: {{ .Values.image.pullPolicy }}
|
||||||
|
flinkVersion: {{ .Values.flinkVersion }}
|
||||||
|
flinkConfiguration:
|
||||||
|
classloader.resolve-order: parent-first
|
||||||
|
taskmanager.numberOfTaskSlots: {{ .Values.taskManager.taskSlots | quote }}
|
||||||
|
execution.checkpointing.interval: {{ .Values.checkpoint.interval }}
|
||||||
|
execution.checkpointing.mode: {{ .Values.checkpoint.mode }}
|
||||||
|
execution.checkpointing.dir: {{ .Values.checkpointStorage.checkpointDir }}
|
||||||
|
execution.checkpointing.savepoint-dir: {{ .Values.checkpointStorage.savepointDir }}
|
||||||
|
classloader.parent-first-patterns.additional: org.apache.iceberg.
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "s3" }}
|
||||||
|
{{- if .Values.checkpointStorage.s3.endpoint }}
|
||||||
|
s3.endpoint: {{ .Values.checkpointStorage.s3.endpoint }}
|
||||||
|
s3.path-style-access: {{ .Values.checkpointStorage.s3.pathStyleAccess | quote }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
|
{{- if .Values.truststore.enabled }}
|
||||||
|
env.java.opts.jobmanager: >-
|
||||||
|
-Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks
|
||||||
|
-Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }}
|
||||||
|
-Djavax.net.ssl.trustStoreType=JKS
|
||||||
|
env.java.opts.taskmanager: >-
|
||||||
|
-Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks
|
||||||
|
-Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }}
|
||||||
|
-Djavax.net.ssl.trustStoreType=JKS
|
||||||
|
{{- end }}
|
||||||
|
serviceAccount: {{ .Values.serviceAccount }}
|
||||||
|
jobManager:
|
||||||
|
replicas: {{ .Values.jobManager.replicas }}
|
||||||
|
resource:
|
||||||
|
memory: {{ .Values.jobManager.memory }}
|
||||||
|
cpu: {{ .Values.jobManager.cpu }}
|
||||||
|
taskManager:
|
||||||
|
replicas: {{ .Values.taskManager.replicas }}
|
||||||
|
resource:
|
||||||
|
memory: {{ .Values.taskManager.memory }}
|
||||||
|
cpu: {{ .Values.taskManager.cpu }}
|
||||||
|
podTemplate:
|
||||||
|
spec:
|
||||||
|
containers:
|
||||||
|
- name: flink-main-container
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
env:
|
||||||
|
- name: GOOGLE_APPLICATION_CREDENTIALS
|
||||||
|
value: {{ .Values.checkpointStorage.gcs.credentialPath }}
|
||||||
|
{{- else if eq .Values.checkpointStorage.storageType "s3" }}
|
||||||
|
{{- if or .Values.checkpointStorage.s3.existingSecret .Values.checkpointStorage.s3.accessKeyId }}
|
||||||
|
env:
|
||||||
|
{{- if .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
- name: AWS_ACCESS_KEY_ID
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: {{ .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
key: {{ .Values.checkpointStorage.s3.accessKeyIdKey }}
|
||||||
|
- name: AWS_SECRET_ACCESS_KEY
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: {{ .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
key: {{ .Values.checkpointStorage.s3.secretAccessKeyKey }}
|
||||||
|
{{- else }}
|
||||||
|
- name: AWS_ACCESS_KEY_ID
|
||||||
|
value: {{ .Values.checkpointStorage.s3.accessKeyId }}
|
||||||
|
- name: AWS_SECRET_ACCESS_KEY
|
||||||
|
value: {{ .Values.checkpointStorage.s3.secretAccessKey }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
|
volumeMounts:
|
||||||
|
- name: pipeline-config
|
||||||
|
mountPath: /opt/flink/flink-cdc-3.5.0/conf
|
||||||
|
{{- if .Values.truststore.enabled }}
|
||||||
|
- name: truststore
|
||||||
|
mountPath: /opt/flink/certs
|
||||||
|
readOnly: true
|
||||||
|
{{- end }}
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
- name: gcs-key
|
||||||
|
mountPath: /opt/flink/gcs-key
|
||||||
|
readOnly: true
|
||||||
|
{{- end }}
|
||||||
|
volumes:
|
||||||
|
- name: pipeline-config
|
||||||
|
configMap:
|
||||||
|
name: cdc-pipeline-config
|
||||||
|
{{- if .Values.truststore.enabled }}
|
||||||
|
- name: truststore
|
||||||
|
secret:
|
||||||
|
secretName: {{ .Values.truststore.secretName }}
|
||||||
|
{{- end }}
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
- name: gcs-key
|
||||||
|
secret:
|
||||||
|
secretName: {{ .Values.checkpointStorage.gcs.credentialSecret }}
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,32 @@
|
|||||||
|
{{- if .Values.ingress.enabled }}
|
||||||
|
apiVersion: networking.k8s.io/v1
|
||||||
|
kind: Ingress
|
||||||
|
metadata:
|
||||||
|
name: {{ .Release.Name }}
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
{{- if .Values.ingress.annotations }}
|
||||||
|
annotations:
|
||||||
|
{{- toYaml .Values.ingress.annotations | nindent 4 }}
|
||||||
|
{{- end }}
|
||||||
|
spec:
|
||||||
|
{{- if .Values.ingress.className }}
|
||||||
|
ingressClassName: {{ .Values.ingress.className }}
|
||||||
|
{{- end }}
|
||||||
|
rules:
|
||||||
|
- host: {{ .Values.ingress.host }}
|
||||||
|
http:
|
||||||
|
paths:
|
||||||
|
- path: /
|
||||||
|
pathType: ImplementationSpecific
|
||||||
|
backend:
|
||||||
|
service:
|
||||||
|
name: {{ .Release.Name }}-rest
|
||||||
|
port:
|
||||||
|
number: 8081
|
||||||
|
tls:
|
||||||
|
- hosts:
|
||||||
|
- {{ .Values.ingress.host }}
|
||||||
|
secretName: {{ .Release.Name }}-tls
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,80 @@
|
|||||||
|
{{- if .Values.job.enabled }}
|
||||||
|
apiVersion: batch/v1
|
||||||
|
kind: Job
|
||||||
|
metadata:
|
||||||
|
name: {{ .Values.job.name }}
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
spec:
|
||||||
|
backoffLimit: 0
|
||||||
|
template:
|
||||||
|
spec:
|
||||||
|
serviceAccountName: {{ .Values.serviceAccount }}
|
||||||
|
containers:
|
||||||
|
- name: submit
|
||||||
|
image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
env:
|
||||||
|
- name: GOOGLE_APPLICATION_CREDENTIALS
|
||||||
|
value: {{ .Values.checkpointStorage.gcs.credentialPath }}
|
||||||
|
{{- else if eq .Values.checkpointStorage.storageType "s3" }}
|
||||||
|
{{- if or .Values.checkpointStorage.s3.existingSecret .Values.checkpointStorage.s3.accessKeyId }}
|
||||||
|
env:
|
||||||
|
{{- if .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
- name: AWS_ACCESS_KEY_ID
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: {{ .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
key: {{ .Values.checkpointStorage.s3.accessKeyIdKey }}
|
||||||
|
- name: AWS_SECRET_ACCESS_KEY
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: {{ .Values.checkpointStorage.s3.existingSecret }}
|
||||||
|
key: {{ .Values.checkpointStorage.s3.secretAccessKeyKey }}
|
||||||
|
{{- else }}
|
||||||
|
- name: AWS_ACCESS_KEY_ID
|
||||||
|
value: {{ .Values.checkpointStorage.s3.accessKeyId }}
|
||||||
|
- name: AWS_SECRET_ACCESS_KEY
|
||||||
|
value: {{ .Values.checkpointStorage.s3.secretAccessKey }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
|
{{- end }}
|
||||||
|
command: ["/bin/sh", "-c"]
|
||||||
|
args:
|
||||||
|
- |
|
||||||
|
cd /opt/flink/flink-cdc-3.5.0
|
||||||
|
./bin/flink-cdc.sh \
|
||||||
|
{{ .Values.job.pipelineFile }} \
|
||||||
|
-Dexecution.checkpointing.interval={{ .Values.pipeline.checkpointInterval }} \
|
||||||
|
-Drest.address={{ .Values.job.restAddress }} \
|
||||||
|
-Drest.port={{ .Values.job.restPort }}
|
||||||
|
volumeMounts:
|
||||||
|
- name: pipeline-config
|
||||||
|
mountPath: /opt/flink/flink-cdc-3.5.0/conf
|
||||||
|
{{- if .Values.truststore.enabled }}
|
||||||
|
- name: truststore
|
||||||
|
mountPath: /opt/flink/certs
|
||||||
|
readOnly: true
|
||||||
|
{{- end }}
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
- name: gcs-key
|
||||||
|
mountPath: /opt/flink/gcs-key
|
||||||
|
readOnly: true
|
||||||
|
{{- end }}
|
||||||
|
volumes:
|
||||||
|
- name: pipeline-config
|
||||||
|
configMap:
|
||||||
|
name: cdc-pipeline-config
|
||||||
|
{{- if .Values.truststore.enabled }}
|
||||||
|
- name: truststore
|
||||||
|
secret:
|
||||||
|
secretName: {{ .Values.truststore.secretName }}
|
||||||
|
{{- end }}
|
||||||
|
{{- if eq .Values.checkpointStorage.storageType "gcs" }}
|
||||||
|
- name: gcs-key
|
||||||
|
secret:
|
||||||
|
secretName: {{ .Values.checkpointStorage.gcs.credentialSecret }}
|
||||||
|
{{- end }}
|
||||||
|
restartPolicy: Never
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,25 @@
|
|||||||
|
{{- if .Values.truststore.create }}
|
||||||
|
apiVersion: v1
|
||||||
|
kind: Secret
|
||||||
|
metadata:
|
||||||
|
name: {{ .Values.truststore.secretName }}
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
type: Opaque
|
||||||
|
data:
|
||||||
|
truststore.jks: {{ required "truststore.base64jks is required when truststore.create is true" .Values.truststore.base64jks }}
|
||||||
|
{{- end }}
|
||||||
|
{{- if and (eq .Values.checkpointStorage.storageType "gcs") .Values.checkpointStorage.gcs.create }}
|
||||||
|
---
|
||||||
|
apiVersion: v1
|
||||||
|
kind: Secret
|
||||||
|
metadata:
|
||||||
|
name: {{ .Values.checkpointStorage.gcs.credentialSecret }}
|
||||||
|
namespace: {{ .Release.Namespace }}
|
||||||
|
labels:
|
||||||
|
{{- include "flink-cdc-session.labels" . | nindent 4 }}
|
||||||
|
type: Opaque
|
||||||
|
data:
|
||||||
|
key.json: {{ required "checkpointStorage.gcs.base64json is required when checkpointStorage.gcs.create is true" .Values.checkpointStorage.gcs.base64json }}
|
||||||
|
{{- end }}
|
||||||
@@ -0,0 +1,123 @@
|
|||||||
|
image:
|
||||||
|
repository: paasup/flink-cdc-pipeline
|
||||||
|
tag: 3.5.0-flink1.20-r1
|
||||||
|
pullPolicy: Always
|
||||||
|
|
||||||
|
flinkVersion: v1_20
|
||||||
|
serviceAccount: flink
|
||||||
|
|
||||||
|
ingress:
|
||||||
|
enabled: false
|
||||||
|
host: "flink-cdc-session.example.org"
|
||||||
|
className: ""
|
||||||
|
annotations: {}
|
||||||
|
|
||||||
|
jobManager:
|
||||||
|
replicas: 1
|
||||||
|
memory: 1024m
|
||||||
|
cpu: 0.5
|
||||||
|
|
||||||
|
taskManager:
|
||||||
|
replicas: 3
|
||||||
|
memory: 2048m
|
||||||
|
cpu: 1
|
||||||
|
taskSlots: 2
|
||||||
|
|
||||||
|
# ──────────────────────────────────────────────────────────────
|
||||||
|
# CheckpointStorage: 체크포인트/세이브포인트 저장 위치 + 자격증명 (WHERE)
|
||||||
|
# storageType에 해당하는 하위 블록(gcs 또는 s3)만 설정
|
||||||
|
# ──────────────────────────────────────────────────────────────
|
||||||
|
checkpointStorage:
|
||||||
|
storageType: s3 # gcs | s3
|
||||||
|
checkpointDir: s3://my-bucket/flink/checkpoints
|
||||||
|
savepointDir: s3://my-bucket/flink/savepoints
|
||||||
|
|
||||||
|
# S3 (storageType: s3 일 때 설정)
|
||||||
|
# endpoint 비어있음 → AWS S3 / MinIO 등: https://minio.example.org
|
||||||
|
# accessKeyId/secretAccessKey 모두 비어있음 → IAM Role / IRSA 사용
|
||||||
|
# existingSecret 지정 시 → 기존 Secret에서 자격증명 참조 (accessKeyId 값보다 우선)
|
||||||
|
s3:
|
||||||
|
endpoint: ""
|
||||||
|
pathStyleAccess: "false"
|
||||||
|
accessKeyId: ""
|
||||||
|
secretAccessKey: ""
|
||||||
|
existingSecret: ""
|
||||||
|
accessKeyIdKey: access-key-id
|
||||||
|
secretAccessKeyKey: secret-access-key
|
||||||
|
# GCS (storageType: gcs 일 때 설정)
|
||||||
|
gcs:
|
||||||
|
credentialSecret: gcs-key
|
||||||
|
credentialPath: /opt/flink/gcs-key/key.json
|
||||||
|
create: false
|
||||||
|
# create: true 시 필요: base64 인코딩된 key.json 파일 내용
|
||||||
|
# base64json: ""
|
||||||
|
|
||||||
|
# 체크포인트 동작 설정 (HOW)
|
||||||
|
checkpoint:
|
||||||
|
interval: 60s
|
||||||
|
mode: EXACTLY_ONCE
|
||||||
|
|
||||||
|
# Secret 관리
|
||||||
|
# create: false → 기존 Secret 참조 (kubectl create secret으로 별도 생성 필요)
|
||||||
|
# create: true → 차트가 Secret 리소스를 직접 생성 (base64 값 필수)
|
||||||
|
truststore:
|
||||||
|
# enabled: true → 외부 JKS를 사용. 싱크/소스 엔드포인트가 사설 CA 인증서를 사용할 때 필요.
|
||||||
|
# enabled: false → 외부 JKS 미사용. 모든 엔드포인트가 공인 CA 인증서를 사용할 때 설정.
|
||||||
|
# JVM 기본 cacerts로 TLS 검증을 수행하며, truststore Secret도 불필요.
|
||||||
|
enabled: true
|
||||||
|
secretName: flink-truststore
|
||||||
|
password: changeit
|
||||||
|
create: false
|
||||||
|
# create: true 시 필요: base64 인코딩된 truststore.jks 파일 내용
|
||||||
|
# base64jks: ""
|
||||||
|
|
||||||
|
# Flink CDC 글로벌 설정 (flink-cdc.yaml)
|
||||||
|
flinkCdc:
|
||||||
|
parallelism: 1
|
||||||
|
schemaChangeBehavior: EVOLVE
|
||||||
|
|
||||||
|
# PostgreSQL 소스
|
||||||
|
postgres:
|
||||||
|
hostname: postgres-postgresql.flink-test.svc.cluster.local
|
||||||
|
port: 5432
|
||||||
|
username: flink_cdc
|
||||||
|
password: password
|
||||||
|
slotName: flink_cdc_slot
|
||||||
|
decodingPlugin: pgoutput
|
||||||
|
tables: testdb.public.orders,testdb.public.users
|
||||||
|
|
||||||
|
# 파이프라인 설정
|
||||||
|
pipeline:
|
||||||
|
name: pg-to-iceberg-test
|
||||||
|
parallelism: 2
|
||||||
|
checkpointInterval: 60s
|
||||||
|
|
||||||
|
# Iceberg REST 카탈로그 싱크
|
||||||
|
sink:
|
||||||
|
catalog:
|
||||||
|
uri: https://lakekeeper.example.org/catalog
|
||||||
|
warehouse: minio
|
||||||
|
s3Endpoint: https://minio.example.org
|
||||||
|
s3PathStyleAccess: "true"
|
||||||
|
oauth2Uri: https://keycloak.example.org/realms/paasup/protocol/openid-connect/token
|
||||||
|
credential: lakekeeper-admin:wMyE9WNn50DH8yycLXwLTmMmu66JU7GV
|
||||||
|
scope: lakekeeper
|
||||||
|
|
||||||
|
# 소스→싱크 테이블 라우팅
|
||||||
|
# source-table: PostgreSQL CDC emit 형식 (schema.table)
|
||||||
|
# sink-table: Iceberg namespace.table 형식
|
||||||
|
route:
|
||||||
|
- sourceTable: public.orders
|
||||||
|
sinkTable: testdb_public.orders
|
||||||
|
- sourceTable: public.users
|
||||||
|
sinkTable: testdb_public.users
|
||||||
|
|
||||||
|
# CDC 파이프라인 Job 제출
|
||||||
|
# enabled: false → Session Cluster만 배포 (Job 생략)
|
||||||
|
# enabled: true → Session Cluster + Job 함께 배포
|
||||||
|
job:
|
||||||
|
enabled: true
|
||||||
|
name: submit-cdc
|
||||||
|
restAddress: flink-cdc-session-rest
|
||||||
|
restPort: 8081
|
||||||
|
pipelineFile: ./conf/pipeline.yaml
|
||||||
Reference in New Issue
Block a user