Compute Module 실행 모드와 권한

항목	Functions 모드	Pipeline 모드
작업 입력	사용자가 플랫폼 함수를 호출	빌드 시스템이 관리하는 입력 자원
주된 용도	Workshop, OSDK 또는 다른 함수에서 사용자 로직 실행	스트리밍 및 실시간 미디어 등 데이터 처리
Foundry 접근	No platform permissions 또는 Application permissions	선언한 입력/출력에 범위가 제한된 작업 토큰
계보	함수가 수행하는 작업에 따라 결정	빌드 시스템을 통한 입력/출력 계보
직접 Query	함수 쿼리 사용	비대화형. 사용자가 직접 쿼리할 수 없음

Functions 권한 모델

No platform permissions는 플랫폼 API나 OSDK에 접근할 권한을 제공하지 않습니다. Foundry 데이터를 쓰지 않는 계산에 사용할 수 있으며 외부 소스의 egress는 별도로 구성할 수 있습니다.

Application permissions는 연결한 서비스 사용자의 권한을 사용합니다. 모듈을 호출한 사용자가 달라져도 서비스 사용자 권한은 같습니다. 앱 전체에 필요한 데이터 범위를 설정하며, 앱의 함수를 호출할 수 있는 사용자에게 어떤 결과를 공개하는지도 함께 제한해야 합니다. 일부 플랫폼 인스턴스에서는 이 모드를 사용할 수 없습니다.

서비스 사용자 만들기

Configure의 Execution mode에서 Functions module과 Application's permissions를 선택합니다.

Functions module의 Application permissions 선택

구성을 저장합니다. 연결한 타사 애플리케이션을 자동으로 만들고 client ID와 서비스 사용자 이름을 표시합니다.

구성 저장 뒤 표시되는 앱 client ID와 서비스 사용자

필요하면 Control Panel > Third party applications에서 서비스 사용자의 접근 보안 표지 등을 제한합니다. 이 변경에는 조직의 Manage OAuth 2.0 clients 권한이 필요합니다.

공개 Foundry URL을 호출하려면 그 호스트를 허용하는 소스를 추가합니다. 네트워크 정책만으로 데이터 권한이 부여되지는 않습니다.

애플리케이션 토큰 사용

SDK의 oauth 보조 함수는 예약 환경 변수 CLIENT_ID와 CLIENT_SECRET을 읽습니다. FOUNDRY_HOST에는 프로토콜을 제외한 실제 호스트를 설정합니다.

import os
from compute_modules.auth import oauth

access_token = oauth(
    os.environ["FOUNDRY_HOST"],
    ["api:datasets-read", "api:datasets-write"],
)
if not access_token:
    raise RuntimeError("Unable to obtain an application token")

장시간 실행하며 재사용하려면 Python SDK의 RefreshingOauthToken 또는 Java wrapper를 사용합니다.

SDK 없이 client credentials로 토큰을 교환하고 데이터셋 이름을 읽는 예는 다음과 같습니다. DATASET_RID에 실제 접근할 RID를 설정합니다. 이 조회에 필요한 범위는 api:datasets-read이며 파일을 쓰는 앱이라면 api:datasets-write도 요청합니다.

import os
from urllib.parse import quote
import requests


def get_dataset_name() -> str:
    host = os.environ["FOUNDRY_HOST"]
    rid = quote(os.environ["DATASET_RID"], safe="")
    with requests.Session() as session:
        session.verify = os.environ["DEFAULT_CA_PATH"]
        token_response = session.post(
            f"https://{host}/multipass/api/oauth2/token",
            data={
                "grant_type": "client_credentials",
                "client_id": os.environ["CLIENT_ID"],
                "client_secret": os.environ["CLIENT_SECRET"],
                "scope": "api:datasets-read",
            },
            timeout=30,
        )
        token_response.raise_for_status()
        token = token_response.json()["access_token"]
        dataset_response = session.get(
            f"https://{host}/api/v1/datasets/{rid}",
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        dataset_response.raise_for_status()
        return dataset_response.json()["name"]

인증 실패, 권한 부족과 HTTP 오류는 호출 실패로 전파합니다. TypeScript SDK가 제공하는 내부 서비스 검색 경로는 공개 호스트의 ingress 소스와 다르지만 API 권한은 여전히 필요합니다.

Pipeline 입력과 출력

Pipeline 모드의 입력/출력은 빌드 시스템을 거칩니다. 보안 표지와 데이터 출처 통제가 필요하거나 Data Lineage에서 계보를 추적할 때 사용합니다. 기본으로는 아무 Foundry 자원에도 접근할 수 없습니다.

Pipeline 모드의 구성에서 input 또는 output을 추가합니다.

Pipeline 모드에서 입력과 출력 자원을 선언하는 화면

같은 프로젝트의 자원을 선택하고 고유 별칭을 지정합니다. 지원 자원은 데이터셋, 스트리밍 데이터셋과 미디어 집합입니다.

입력/출력 자원의 유형과 고유 식별자를 지정하는 대화상자

빌드 시스템은 입력과 출력 권한으로 범위를 제한한 bearer token을 파일로 마운트하고 BUILD2_TOKEN에 그 파일 경로를 설정합니다. 자원 정보는 RESOURCE_ALIAS_MAP의 경로에 JSON으로 마운트합니다.

가상 별칭 inputDataset과 outputDataset을 설정한 경우 다음처럼 읽습니다.

import json
import os
from pathlib import Path

bearer_token = Path(os.environ["BUILD2_TOKEN"]).read_text().strip()
with open(os.environ["RESOURCE_ALIAS_MAP"], encoding="utf-8") as file:
    resources = json.load(file)

input_info = resources["inputDataset"]
output_info = resources["outputDataset"]
input_rid = input_info["rid"]
input_branch = input_info.get("branch") or "master"
output_rid = output_info["rid"]
output_branch = output_info.get("branch") or "master"

각 항목은 rid와 선택 branch를 갖습니다. branch가 없거나 비어 있으면 예제에서는 master를 사용합니다. 입출력 접근에는 이 빌드 토큰을 사용하며 별도의 호출자 토큰으로 선언 범위를 우회하지 않습니다.

스트림의 최근 레코드

구성한 스트리밍 입력에 대해 stream-proxy의 최근 레코드를 조회할 수 있습니다. 아래 함수의 alias에는 실제 입력 별칭을 넘깁니다.

import json
import os
from pathlib import Path
from urllib.parse import quote
import requests


def latest_records(alias: str):
    with open(os.environ["RESOURCE_ALIAS_MAP"], encoding="utf-8") as file:
        resource = json.load(file)[alias]
    token = Path(os.environ["BUILD2_TOKEN"]).read_text().strip()
    rid = quote(resource["rid"], safe="")
    branch = quote(resource.get("branch") or "master", safe="")
    host = os.environ["FOUNDRY_HOST"]
    with requests.get(
        f"https://{host}/stream-proxy/api/streams/{rid}/branches/{branch}/records",
        headers={"Authorization": f"Bearer {token}"},
        verify=os.environ["DEFAULT_CA_PATH"],
        timeout=60,
    ) as response:
        response.raise_for_status()
        return response.json()

이 호출은 최근 레코드 조회이며 처리한 위치를 저장하는 소비자 커서가 아닙니다. 같은 범위를 반복 조회할 수 있으므로 반복 호출만으로 중복 없는 처리나 정확히 한 번 처리가 성립하지 않습니다.

데이터셋 파일

아래는 입력 별칭이 가리키는 브랜치에서 파일 목록 한 페이지와 지정한 파일의 바이트를 읽는 예입니다. requests와 앞서 설명한 환경 변수들이 필요합니다.

import json
import os
from pathlib import Path
from urllib.parse import quote
import requests


def dataset_paths(alias: str) -> tuple[str, str, dict[str, str]]:
    with open(os.environ["RESOURCE_ALIAS_MAP"], encoding="utf-8") as file:
        resource = json.load(file)[alias]
    rid = quote(resource["rid"], safe="")
    branch = quote(resource.get("branch") or "master", safe="")
    token = Path(os.environ["BUILD2_TOKEN"]).read_text().strip()
    return rid, branch, {"Authorization": f"Bearer {token}"}


def list_file_page(alias: str):
    rid, branch, headers = dataset_paths(alias)
    host = os.environ["FOUNDRY_HOST"]
    with requests.get(
        f"https://{host}/foundry-catalog/api/catalog/datasets/{rid}/views2/{branch}/files",
        params={
            "pageSize": 100,
            "includeOpenExclusiveTransaction": "false",
            "excludeHiddenFiles": "false",
        },
        headers=headers,
        verify=os.environ["DEFAULT_CA_PATH"],
        timeout=60,
    ) as response:
        response.raise_for_status()
        return response.json()


def read_file(alias: str, file_name: str) -> bytes:
    rid, branch, headers = dataset_paths(alias)
    host = os.environ["FOUNDRY_HOST"]
    path = quote(file_name, safe="/")
    with requests.get(
        f"https://{host}/foundry-data-proxy/api/dataproxy/datasets/{rid}/views/{branch}/{path}",
        headers=headers,
        verify=os.environ["DEFAULT_CA_PATH"],
        timeout=60,
    ) as response:
        response.raise_for_status()
        return response.content

list_file_page는 pageSize 100인 한 페이지를 반환합니다. 데이터셋 전체 파일 목록을 모두 읽었다는 의미는 아닙니다. CSV는 read_file("inputDataset", "example.csv").decode("utf-8")처럼 텍스트로 읽을 수 있으며 파일 본문을 무조건 JSON으로 해석하지 않습니다.

공개 Foundry 호스트를 호출하는 위 경로에는 해당 호스트의 소스 네트워크 정책이 필요합니다.

계보, 트랜잭션과 지속 실행

빌드 시스템은 출력 데이터셋의 트랜잭션을 열고 파이프라인이 성공적으로 끝나면 커밋합니다. 입력과 출력이 빌드 시스템을 통과하므로 Data Lineage에서 처리 경로와 출력의 출처를 추적할 수 있습니다.

Pipeline 모드에는 사용자 쿼리가 없으므로 쿼리 부하 기반 자동 확장이 0까지 줄일 수 있습니다. 계속 실행해야 한다면 Minimum replicas를 1 이상으로 설정합니다. 이 모드에는 작업 쿼리를 poll하는 Compute Module client를 구현할 필요가 없습니다.

Keyboard shortcuts

팔란티어 플랫폼 한국어 해설