Keyboard shortcuts

Press ← or → to navigate between chapters

Press ? to show this help

Press Esc to hide this help

Directory 소스와 에이전트 파일의 SSH 수집

Directory 커넥터는 Data Connection 에이전트가 실행되는 호스트의 파일을 수집합니다. 원천 탐색, 배치 동기화와 증분 수집은 Sunset 단계이며 에이전트 워커(Legacy)에서만 실행됩니다. 같은 Directory 연결을 Foundry 워커로 실행하는 대응 모드는 없습니다.

파일 공유 프로토콜을 사용할 수 있으면 SFTP, SMB 또는 FTP 계열 커넥터를 우선 사용합니다. 호스트 자체에서만 파일에 접근할 수 있으면 REST API 소스와 외부 변환을 구성하여 Foundry 워커에서 SSH로 접근할 수 있습니다.

기존 Directory 소스

Data Connection의 +New Source에서 Directory를 선택하면 에이전트 워커로 구성됩니다. 필수 설정인 Root directory에는 해당 에이전트 호스트에서 모든 요청의 기준으로 사용할 디렉터리를 지정합니다.

파일 기반 동기화로 원하는 파일을 가져옵니다. 파일 형식은 유지되며 수집 중이나 수집 직후에 스키마를 적용하지 않습니다. 출력 데이터셋에 스키마를 지정하거나 하위 변환에서 내용을 해석합니다.

Foundry 워커에서 SSH 연결 준비

  1. Data Connection에서 REST API 소스를 만듭니다. 이 소스는 SSH 통신에 필요한 대상과 인증 정보를 코드에 제공하는 데 사용합니다.
  2. 도메인을 에이전트 호스트의 주소로, 포트를 SSH의 22로 설정합니다.
  3. 호스트에 SSH로 접속할 사용자의 이름과 비밀번호를 소스 비밀 정보에 저장합니다.
  4. 그 에이전트 자체를 경유하는 에이전트 프록시 외부 통신 정책을 추가합니다.
  5. 소스를 Python 변환 저장소에 가져오고 Libraries에서 paramiko를 추가합니다.

호스트 주소가 10.x.x.x, 172.16.x.x, 192.168.x.x 같은 사설 IP이면 에이전트의 host override에서 별도 도메인을 해당 IP에 매핑합니다. REST 소스와 외부 통신 정책에는 사설 IP 대신 그 도메인을 사용합니다.

SSH는 비HTTPS 연결이므로 소스의 create_socket(hostname, 22)로 연결 소켓을 만들고 Paramiko에 전달합니다. 서버 호스트 키는 관리자가 확인한 값과 비교하여 검증합니다.

공통 SSH 연결 구성

소스의 연결 소켓과 서버 호스트 키를 다음과 같이 구성합니다. 알 수 없는 서버 키를 자동 신뢰하지 않으며, 키가 없거나 일치하지 않으면 연결이 실패합니다.

소스에 hostname, username, password, known_hosts 값을 준비합니다. hostname은 소스 정책과 host override의 도메인이며, known_hosts에는 관리자가 별도로 확인한 OpenSSH known_hosts 항목을 넣습니다. 공통 연결 함수와 사용할 변환은 같은 모듈에 둡니다.

from contextlib import contextmanager
from tempfile import NamedTemporaryFile

import paramiko
from transforms.external.systems import ResolvedSource


@contextmanager
def open_agent_ssh(source: ResolvedSource):
    hostname = source.get_secret("hostname")
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    with NamedTemporaryFile(mode="w+", encoding="utf-8") as host_keys:
        host_keys.write(source.get_secret("known_hosts"))
        host_keys.flush()
        client.load_host_keys(host_keys.name)

    proxy_socket = source.create_socket(hostname, 22)
    try:
        client.connect(
            hostname,
            port=22,
            username=source.get_secret("username"),
            password=source.get_secret("password"),
            sock=proxy_socket,
            allow_agent=False,
            look_for_keys=False,
        )
        yield client
    finally:
        client.close()
        proxy_socket.close()

파일 읽기

입력 데이터셋의 remote_file_path 열에 읽을 원격 파일 경로를 넣습니다. 아래 소스 RID, 입력 RID와 출력 RID는 저장소에 가져온 실제 리소스 값으로 바꿉니다.

출력 파일 이름은 원격 경로의 마지막 파일 이름입니다. 서로 다른 경로에 같은 파일 이름이 있으면 출력이 충돌하므로 이 예제에 전달할 경로는 파일 이름이 서로 달라야 합니다.

from transforms.api import (
    Input, Output, LightweightInput, LightweightOutput, lightweight, transform,
)
from transforms.external.systems import Source, external_systems


@lightweight
@external_systems(agent_source=Source("<source_rid>"))
@transform(
    output_dataset=Output("<output_dataset_rid>"),
    files_to_read=Input("<input_dataset_rid>"),
)
def read_agent_files(
    agent_source: ResolvedSource,
    output_dataset: LightweightOutput,
    files_to_read: LightweightInput,
):
    remote_file_paths = files_to_read.pandas()["remote_file_path"].tolist()
    with open_agent_ssh(agent_source) as client:
        sftp = client.open_sftp()
        try:
            for remote_path in remote_file_paths:
                with sftp.open(remote_path, "rb") as remote_file:
                    file_binary_data = remote_file.read()
                filename = remote_path.split("/")[-1]
                with output_dataset.filesystem().open(filename, "wb") as output_file:
                    output_file.write(file_binary_data)
        finally:
            sftp.close()

파일 내용은 메모리에 읽은 뒤 출력 데이터셋에 씁니다. 큰 파일을 처리한다면 실행 자원의 메모리 조건을 고려합니다.

수집 후 파일 삭제

파일 기반 동기화의 완료 전략은 최선 노력 방식의 삭제입니다. 삭제 대상과 결과를 별도 작업으로 관리하려면 다음 순서를 사용합니다.

  1. 상위 동기화에서 파일을 수집하고 성공적으로 수집한 파일 경로를 데이터셋으로 출력합니다.
  2. 상위 동기화가 끝난 뒤 이 경로 데이터셋을 입력으로 삭제 변환을 실행하도록 스케줄을 구성합니다.
  3. 삭제 결과 데이터셋에서 파일별 상태를 확인합니다.

이 변환은 에이전트 호스트의 파일을 영구 삭제합니다. Foundry에 성공적으로 수집한 파일인지 확인하고, 삭제를 허용한 경로만 입력에 포함해야 합니다. 출력 데이터셋 기록이 실패한다고 이미 수행한 원격 파일 삭제가 되돌아오지는 않습니다.

다음 코드는 위의 공통 연결 함수와 함께 사용합니다. pandas 의존성이 필요합니다.

import pandas as pd
from transforms.api import (
    Input, Output, LightweightInput, LightweightOutput, lightweight, transform,
)
from transforms.external.systems import Source, external_systems


@lightweight
@external_systems(agent_source=Source("<source_rid>"))
@transform(
    output_dataset=Output("<deletion_results_dataset_rid>"),
    files_to_delete=Input("<successfully_ingested_paths_dataset_rid>"),
)
def delete_agent_files(
    agent_source: ResolvedSource,
    output_dataset: LightweightOutput,
    files_to_delete: LightweightInput,
):
    remote_file_paths = files_to_delete.pandas()["remote_file_path"].tolist()
    deletion_results = []
    with open_agent_ssh(agent_source) as client:
        sftp = client.open_sftp()
        try:
            for remote_path in remote_file_paths:
                try:
                    sftp.remove(remote_path)
                    status = "deleted"
                except FileNotFoundError:
                    status = "not_found"
                except PermissionError:
                    status = "permission_denied"
                deletion_results.append({"file": remote_path, "status": status})
        finally:
            sftp.close()
    output_dataset.write_table(
        pd.DataFrame(deletion_results, columns=["file", "status"])
    )

deleted는 삭제 성공, not_found는 해당 경로에 파일이 없음, permission_denied는 삭제 권한 부족을 뜻합니다. 다른 연결 및 처리 오류는 작업 오류로 전달됩니다. 결과에 권한 부족이 남아 있으면 모든 파일의 삭제가 완료된 것으로 판단하지 않습니다.