1. NiFi 불레틴 저장 구조
프로세서 에러(불레틴)가 저장되는 위치와 보존 한계를 소스 기준으로 정리함
1.1 VolatileBulletinRepository
- 저장 위치 — 메모리에만 저장됨. 디스크 영속화 없음
- 보존 개수 — 컴포넌트당 5개, 컨트롤러 서비스는 10개까지만 유지됨
- 조회 범위 — 조회 시점 기준 5분이 지난 불레틴은 자동으로 제외됨
- 재시작 — NiFi 재시작 시 전부 소실됨
- nifi-app.log — 별도로 로그 파일에도 남지만, 남는 수준과 형식은 logback 설정에 따라 달라짐
불레틴은 "최근 몇 건을 잠깐 보여주는 알림판"이지 에러 저장소가 아니다. 에러 이력이 필요하면 flow 안에서 직접 로그를 만들어 외부로 내보내야 함
2. 에러 로그 파이프라인 설계
ExecuteSQLRecord 실패 FlowFile을 로그 레코드로 가공해 Kafka로 전송하는 구조를 설계함
2.1 전체 흐름
ExecuteSQLRecord (failure)
→ UpdateAttribute (로그 필드 조립)
├→ PublishKafka (로그 전송)
└→ ExecuteSQL (재시도 카운트 증가)
- UpdateAttribute — error.seq, error.timestamp, error.source, error.message, log.message 필드를 attribute로 조립함
- 팬아웃 — 같은 relationship을 두 프로세서에 연결해 로그 전송과 재시도 카운트 증가를 병렬 실행함
2.2 error.seq 채번 방식
- nextInt() — 소스 확인 결과 JVM 전역 static AtomicLong 기반
- 단일 JVM 내 동시성 — CAS(compare-and-swap) 방식이라 여러 스레드가 동시에 호출해도 중복 없음. 경합은 CPU 레벨 원자 연산으로 처리됨
- 클러스터 환경 — 노드마다 JVM이 따로 있으므로 노드 간 값이 겹칠 수 있음
- 보완안 — 노드 식별자를 앞에 붙여 전역 유일성 확보
${hostname()}-${nextInt()}
2.3 PublishKafka 프로퍼티 구성
| 프로퍼티 |
설정값 |
비고 |
| Kafka Connection Service |
공용 커넥션 서비스 |
브로커 접속 정보 |
| Topic Name |
에러 로그 토픽 |
|
| Record Reader / Writer |
비워둠 |
원본 content를 그대로 전송 |
| Kafka Key |
${error.seq} |
메시지 키로 채번값 사용 |
| Failure Strategy |
운영 정책에 맞게 선택 |
|
| Delivery Guarantee |
운영 정책에 맞게 선택 |
|
| Attribute Header Pattern |
error\..* |
error.* attribute를 Kafka 헤더로도 전송 |
3. PublishKafka 실패 원인 추적의 한계
PublishKafka가 실패했을 때 "왜 실패했는지"를 flow 안에서 얻을 수 있는지 소스로 확인함
3.1 PublishKafka.java 확인 결과
- 실패 attribute — 실패해도 에러 메시지 attribute가 전혀 붙지 않음
- msg.count — 붙는 건 이것뿐이며, 그마저도 부분 실패일 때만 붙음
- 실제 원인 위치 — getLogger().error(...)로만 출력되어 로그 파일과 불레틴에만 남음
3.2 SiteToSiteBulletinReportingTask 검토
- 동작 — 불레틴을 Site-to-Site로 flow에 다시 끌어오는 Reporting Task
- 한계 — bulletinMessage 필드만 전송하고 stackTrace는 아예 직렬화하지 않음
- 결론 — 이 방법으로도 근본 원인 텍스트는 얻을 수 없음
| 수단 |
원인 텍스트 확보 |
비고 |
| FlowFile attribute |
불가 |
에러 attribute 자체가 없음 |
| SiteToSiteBulletinReportingTask |
부분적 |
메시지만, stackTrace 없음 |
| nifi-app.log tail |
가능 |
원인 텍스트를 온전히 보유 |
실패 원인의 원문은 결국 nifi-app.log에만 있다. 원인까지 수집하려면 로그 파일을 직접 tail하는 경로를 따로 두는 수밖에 없음
4. CUBRID 재시도 로직 설계
상태 컬럼 기반 선점 구조에서 발생하는 레이스 컨디션과 그 해결 방식을 정리함
4.1 상태값 체계
| status |
의미 |
| 0 |
신규 |
| 1 |
선점 완료 (폴링 대상) |
| 2 |
성공 |
| 9 |
실패 (재시도 대기) |
4.2 레이스 컨디션
- 상황 — ExecuteSQLRecord가 status=1인 행을 폴링하는 동안, 실패 처리(9로 변경)는 비동기로 진행됨
- 문제 — 9로 바뀌기 전에 다음 폴링 사이클이 돌면 같은 행을 중복으로 가져감
- 해결 — 선점 시 실행마다 새로 발급한 uuid(current_flow_id)를 행에 같이 기록하고, 폴링은 그 uuid로만 매칭함. 다른 실행의 uuid와는 절대 겹치지 않음
-- 선점
UPDATE target_table
SET status = 1, job_id = '${current_flow_id}'
WHERE status = 0;
-- 폴링
SELECT * FROM target_table
WHERE status = 1 AND job_id = '${current_flow_id}';
4.3 WHERE절 식별자 혼동 교정
- 실수 — 실제 행 식별자와 선점용 uuid를 같은 조건에 섞어 쓰는 실수가 있었음
- 확정 구조 — job_id 컬럼 자체를 선점할 때마다 uuid로 덮어쓰는 방식으로 정리함
4.4 재시도 흐름 분리
- 구조 — 재시도 로직을 GenerateFlowFile로 시작하는 독립 흐름으로 분리함
- 재선점 — 9 → 1로 되돌릴 때도 uuid를 새로 발급해 동일한 원칙으로 매칭함
GenerateFlowFile
→ UpdateAttribute (current_flow_id = ${UUID()})
→ ExecuteSQL
UPDATE target_table
SET status = 1, job_id = '${current_flow_id}'
WHERE status = 9;
4.5 충돌 가능 지점 구분
| 대상 |
충돌 여부 |
근거 |
| FlowFile attribute 이름 |
없음 |
각 FlowFile이 자기만의 attributes Map을 가짐 (StandardFlowFileRecord 소스 확인) |
| DB status 컬럼 |
있음 |
여러 흐름이 공유하는 실제 상태이므로 uuid 재발급 + 매칭 원칙 필수 |
attribute는 FlowFile마다 격리되어 있어 이름이 같아도 문제없다. 위험한 건 DB처럼 흐름 바깥에서 공유되는 상태이고, 동시성 제어는 그쪽에 집중해야 함
5. URL Refang 로직
위협 인텔리전스에서 흔히 쓰는 디팽(defang) URL을 원래 형태로 복원하는 방법을 정리함
5.1 REPLACE 체이닝
- 치환 대상 — [://] → ://, [/] → /, hxxp → http
- hxxps — hxxp 치환 시 자동으로 https가 됨
REPLACE(REPLACE(REPLACE(url, '[://]', '://'), '[/]', '/'), 'hxxp', 'http')
5.2 도메인만 추출
- SUBSTRING_INDEX — 전체 URL 복원과는 목적이 다른, 도메인 추출용
- DBeaver 에러 — substring_index not registered 발생. 원인 미확정 상태이며 csql에서 직접 테스트해 볼 필요 있음
5.3 정규식으로 단순화
- 아이디어 — 대괄호만 벗기면 대부분의 디팽 패턴에 범용 대응됨
REGEXP_REPLACE(url, '[\\[\\]]', '')
re.sub(r'[\[\]]', '', url)
5.4 SQL 인젝션 관점
- 핵심 — SQL 자체가 위험한 게 아니라 외부 값을 문자열로 직접 이어붙이는 방식이 위험함
- 해법 — 파라미터 바인딩 사용. "SQL을 피한다"는 해법이 아님
6. 폐쇄망 Trino 연동
인터넷이 차단된 서버에서 Python trino 클라이언트를 설치하고 접속하기까지의 과정을 정리함
6.1 pip 오프라인 설치 문제
- 증상 — pip download, pip install 모두 의존성 해결을 위해 PyPI 접근이 필요해 실패함
- 검토안 — 외부망에서 의존성 포함 다운로드 후 반입하는 방식 (보류)
pip download trino -d ./pkgs
pip install --no-index --find-links=./pkgs trino
6.2 최종 채택 방식 — whl 직접 설치
- 원리 — .whl은 zip 포맷이므로 압축만 풀면 바로 import 가능한 구조임
- 절차 — 브라우저로 .whl 다운로드 → unzip → site-packages에 복사. pip를 전혀 쓰지 않음
unzip trino-*.whl -d ./trino_pkg
cp -r ./trino_pkg/* $(python -c "import site; print(site.getsitepackages()[0])")/
6.3 연쇄 의존성 누락 처리
- 증상 — import할 때마다 tzlocal 등 의존성이 하나씩 터짐
- 해결 — .dist-info/METADATA의 Requires-Dist를 읽고, pip show(로컬 조회, 인터넷 불필요)로 이미 설치된 패키지를 걸러내 필요한 목록만 뽑음
import re, subprocess, glob
meta = glob.glob("./trino_pkg/*.dist-info/METADATA")[0]
reqs = [re.split(r'[ ;<>=!~(\[]', l.split(":", 1)[1].strip())[0]
for l in open(meta) if l.startswith("Requires-Dist:") and "extra ==" not in l]
for r in reqs:
if subprocess.run(["pip", "show", r], capture_output=True).returncode != 0:
print("missing:", r)
6.4 접속 timeout 디버깅
| 단계 |
증상 |
원인 |
조치 |
| 1 |
curl도 timeout |
https_proxy 환경변수 때문에 Trino 요청까지 프록시로 나감 |
unset https_proxy |
| 2 |
google.com 테스트 실패 |
DNS 미해석 — 폐쇄망이라 정상, 무의미한 테스트였음 |
- |
| 3 |
여전히 timeout |
SSH 접속 서버가 곧 Trino 호스트. 자기 자신을 외부 IP로 호출하다 헤어핀(hairpin) 문제 발생 |
접속 주소를 localhost로 변경 |
| 4 |
SSL 검증 실패 |
자체서명 인증서 |
verify=False + 경고 억제 |
import urllib3
import trino
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
conn = trino.dbapi.connect(
host="localhost",
port=8443,
user="user",
http_scheme="https",
verify=False,
)
- 남은 확인 사항 — 클라이언트가 httpx 기반이라면 urllib3 경고 억제가 먹지 않을 수 있음. 현재 환경에서는 문제없이 동작함
timeout은 "네트워크가 막혔다"는 신호일 뿐 원인이 아니다. 프록시 환경변수, 자기 자신으로의 외부 IP 접속처럼 요청 경로 자체를 먼저 의심하는 게 빠름
- 결과 — Trino 접속, 데이터 조회, refang 함수 적용까지 정상 동작 확인
7. 미해결 항목
결론을 내리지 못해 후속 논의가 필요한 항목을 정리함
- job_id 원본 보존 — 선점마다 uuid로 덮어쓰면 최초 식별자가 영구 소실됨. 에러 추적이 필요하면 불변 원본 식별자 컬럼을 별도로 둬야 함
- substring_index not registered — DBeaver 환경 문제인지 CUBRID 함수 지원 문제인지 csql 테스트로 확인 필요
- httpx 기반 경고 억제 — 클라이언트 구현에 따라 억제 방식 재확인 필요
요약
| 주제 |
핵심 내용 |
| NiFi 불레틴 |
메모리 전용, 개수·시간 제한, 재시작 시 소실 |
| 에러 로그 파이프라인 |
failure → 필드 조립 → Kafka 전송 + 재시도 카운트 팬아웃 |
| error.seq |
nextInt()는 JVM 내 유일, 클러스터는 hostname 조합 필요 |
| PublishKafka 실패 원인 |
attribute·불레틴 Reporting Task 모두 원인 불충분, nifi-app.log만 온전 |
| CUBRID 재시도 |
선점 시 uuid 기록·매칭으로 중복 폴링 방지, 재시도도 동일 원칙 |
| URL Refang |
REPLACE 체이닝 → 대괄호 제거 정규식으로 단순화 |
| 폐쇄망 Trino |
whl 직접 설치, 프록시 해제, 헤어핀 회피 위해 localhost 사용 |
오늘 작업의 공통점은 "문서가 아니라 소스를 확인한 뒤에야 답이 나왔다"는 점이다. nextInt()의 범위, PublishKafka의 실패 attribute, Reporting Task의 직렬화 필드 모두 코드를 열어보고서야 한계를 정확히 알 수 있었음