로그를 담는 NAS 가 /LOG-NAS1 ~ /LOG-NAS4 처럼 여러 개로 늘어나고 그 개수가 환경마다 달라지면 ListFile 하나로 감당할 수 없다. ListFile 은 Input Directory 를 하나만 받기 때문이다. 마운트 수만큼 프로세서를 복제하면 마운트가 바뀔 때마다 플로우를 고쳐야 한다.
그래서 수집부를 GenerateFlowFile 로 주기 트리거만 만들고, 실제 탐색은 ExecuteScript(Groovy) 가 하도록 바꾼다. 마운트 목록은 파라미터 컨텍스트에 두고 동적 프로퍼티로 주입한다. 하류(SplitText → ExtractText → UpdateAttribute → AttributesToJSON → PutKudu) 는 그대로 쓴다.
GenerateFlowFile (cron) ─> ExecuteScript (Groovy) ─> SplitText ─> ExtractText ─> ... ─> PutKudu
import org.apache.nifi.components.state.Scope
def trigger = session.get()
if (!trigger) return
try {
// 동적 프로퍼티 nas_roots = #{nas_root_list}
def rootListStr = nas_roots.evaluateAttributeExpressions().getValue() ?: ""
def roots = rootListStr.split(",").collect { it.trim() }.findAll { it }
// 활성 파일명만 정확히 일치시킨다. 백업본(-yyyymmdd)은 자동으로 빠진다
def targetFile = "mail-send-waspod.log"
// 상태는 CLUSTER 스코프에 파일경로 -> 마지막 mtime 으로 둔다
def stateManager = context.stateManager
def stateMap = new HashMap<String, String>(stateManager.getState(Scope.CLUSTER).toMap())
roots.each { root ->
def staDir = new File(root, "sta")
if (!staDir.isDirectory()) return
staDir.listFiles()?.findAll { it.isDirectory() }?.each { lv2 ->
lv2.listFiles()?.findAll { it.isDirectory() && it.name.startsWith("cluster") }?.each { lv3 ->
lv3.listFiles()?.findAll { it.isDirectory() && it.name.contains("mail-kr") }?.each { lv4 ->
def dir = new File(lv4, "mysingle")
if (!dir.isDirectory()) return
def f = new File(dir, targetFile)
if (!f.isFile()) return
def key = f.absolutePath
def prev = stateMap[key]
if (prev == null || Long.parseLong(prev) < f.lastModified()) {
// FlowFile 생성 + stateMap[key] = f.lastModified() 기록
}
}
}
}
}
} finally {
session.remove(trigger)
}
세 가지가 핵심이다.
루트 목록은 파라미터로 받는다. 동적 프로퍼티(nas_roots) 에 #{nas_root_list} 를 넣으면 마운트가 늘거나 줄 때 파라미터 값만 바꾸면 된다. 스크립트도 플로우도 손대지 않는다.
파일명은 정확히 일치시킨다. mail-send-waspod.log 로 완전 일치를 걸면 mail-send-waspod.log.20260527 같은 로테이션 백업이 자동으로 빠진다. ListFile 의 File Filter 를 정규식으로 느슨하게 주었을 때 생기던 중복 수집 문제가 사라진다.
상태는 CLUSTER 스코프에 mtime 으로 둔다. 파일 경로를 키로, 마지막으로 처리한 lastModified 를 값으로 저장하면 append 로 갱신된 파일을 다시 잡을 수 있다. Scope.LOCAL 로 두면 프라이머리 노드가 바뀔 때 전부 다시 읽는다.
트리거 FlowFile 은 finally 에서 반드시 제거한다. 안 그러면 큐에 남아 쌓인다.
기존 ListFile 의 Path Filter 정규식에서 역산한 경로는 추정값이다. 가변 마운트로 바뀌면서 각 마운트 바로 아래 구조가 같다는 보장이 없다. 실제 파일의 절대 경로를 한 줄이라도 확인한 뒤 고정하는 편이 좋다.
/LOG-NAS*/sta/*/cluster*/*portal-kr*/mysingle/login/adminstat-v8-waspod.log
/LOG-NAS*/sta/*/cluster*/*mail-kr*/mysingle/mail-send-waspod.log
깊이가 일정하다면 재귀 탐색보다 고정 깊이로 내려가는 편이 탐색 비용이 훨씬 싸다.
UpdateAttribute 에서 regn_cd=#{regn_cd} 를 세팅해도 AttributesToJSON 의 Attributes List 에 그 이름이 없으면 JSON 에 포함되지 않고 Kudu 로도 나가지 않는다. 속성 추가는 두 군데를 함께 고쳐야 한다.
ts,pageid,epid → ts,pageid,epid,regn_cd