Compute Module 사용자 클라이언트 프로토콜

Python, Java 또는 TypeScript SDK는 작업 수신과 결과 전송을 처리합니다. 다른 언어로 작성하거나 SDK 없이 실행한다면 클라이언트가 이 역할을 구현해야 합니다. Pipeline 모드는 사용자 쿼리를 받지 않으므로 이 클라이언트가 필요하지 않습니다.

클라이언트는 시작 시 함수 명세를 게시하고, 작업을 반복 조회하여 해당 함수를 실행한 뒤 작업 ID에 결과를 전송합니다. 명세는 Functions 화면에서 수동 정의할 수도 있으므로 코드에서의 명세 게시는 선택 사항입니다.

인증과 주소

내부 서비스 요청에는 Module-Auth-Token 헤더를 사용합니다. MODULE_AUTH_TOKEN은 토큰 값이 아닌 파일 경로입니다. DEFAULT_CA_PATH는 TLS를 검증할 CA 파일이며 GET_JOB_URI, POST_RESULT_URI와 POST_SCHEMA_URI는 실제 내부 주소입니다.

: "${MODULE_AUTH_TOKEN:?MODULE_AUTH_TOKEN is required}"
: "${DEFAULT_CA_PATH:?DEFAULT_CA_PATH is required}"
: "${GET_JOB_URI:?GET_JOB_URI is required}"
module_auth_token="$(cat "$MODULE_AUTH_TOKEN")"
curl --fail-with-body \
  --header "Module-Auth-Token: $module_auth_token" \
  --cacert "$DEFAULT_CA_PATH" \
  --request GET "$GET_JOB_URI"

시작 직후 내부 서비스가 아직 준비되지 않으면 connection refused가 발생할 수 있습니다. 잠시 기다렸다가 다시 연결합니다. 인증 오류나 잘못된 주소를 연결 준비 지연으로 계속 숨기지 않습니다.

GET job

GET_JOB_URI는 long polling을 지원합니다. 204 이후 별도 지연 없이 다시 요청할 수 있습니다. 204에는 JSON 작업 본문이 없으므로 파싱하지 않습니다.

작업 응답은 computeModuleJobV1 안에 다음 값을 담습니다.

아래 ID와 토큰은 구조를 보여 주는 가상 값입니다.

{
  "computeModuleJobV1": {
    "jobId": "9a2a1e94-41d3-47d7-807f-db2f4c547b9c",
    "queryType": "multiply",
    "query": {"n": 4.0},
    "temporaryCredentialsAuthToken": "example-sidecar-token",
    "authHeader": "example-authorization-value"
  }
}

authHeader가 모든 모드에 존재하거나 호출자의 모든 권한을 전달한다고 가정하지 않습니다. 이 필드들을 사용자 로그나 함수 결과에 포함하지 않습니다.

POST result

POST_RESULT_URI 뒤에 /{jobId}를 붙입니다. Content-Type은 application/octet-stream이며 함수 결과를 직렬화한 바이트를 보냅니다. 정상 수락은 204이며 응답 본문은 없습니다.

result.json에는 해당 함수의 실제 결과 JSON을, JOB_ID에는 GET에서 받은 작업 ID를 준비합니다.

: "${POST_RESULT_URI:?POST_RESULT_URI is required}"
: "${JOB_ID:?JOB_ID must come from a received job}"
curl --fail-with-body \
  --header "Content-Type: application/octet-stream" \
  --header "Module-Auth-Token: $module_auth_token" \
  --cacert "$DEFAULT_CA_PATH" \
  --request POST --data-binary @result.json \
  "${POST_RESULT_URI%/}/$JOB_ID"

받은 작업에는 결과 보고가 필요합니다. 보고하지 않으면 진행 중 작업으로 남아 새 작업이 해당 worker에 배정되지 않을 수 있습니다. HTTP 요청을 보냈다는 사실과 204로 수락되었다는 사실을 구분합니다.

POST function schema

POST_SCHEMA_URI에 application/json으로 함수 명세의 JSON 배열을 보냅니다. 배열의 각 원소가 함수 하나이며 정상 응답은 204, 응답 본문은 없습니다.

: "${POST_SCHEMA_URI:?POST_SCHEMA_URI is required}"
curl --fail-with-body \
  --header "Content-Type: application/json" \
  --header "Module-Auth-Token: $module_auth_token" \
  --cacert "$DEFAULT_CA_PATH" \
  --request POST --data-binary @schemas.json "$POST_SCHEMA_URI"

함수 하나만 있어도 배열로 감쌉니다. functionName은 클라이언트가 실제로 처리하는 이름과 일치해야 합니다.

순차 처리 Python 클라이언트

다음 app.py는 n을 두 배로 만들거나 2로 나누는 두 함수를 등록합니다. requests가 필요하며 복제본 동시성은 1로 구성합니다. SDK를 import하지 않고 위 HTTP 프로토콜을 직접 사용합니다.

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

LOG = logging.getLogger(__name__)
SCHEMAS = [
    {
        "functionName": name,
        "inputs": [{
            "name": "n",
            "dataType": {"float": {}, "type": "float"},
            "required": True,
            "constraints": [],
        }],
        "output": {
            "single": {"dataType": {"float": {}, "type": "float"}},
            "type": "single",
        },
    }
    for name in ("multiply", "divide")
]


def calculate(query_type: str, query: dict) -> float:
    if query_type == "multiply":
        return float(query["n"]) * 2
    if query_type == "divide":
        return float(query["n"]) / 2
    raise ValueError("Unsupported query type")


def post_at_startup(session, url: str, *, json_body=None, expected=204) -> None:
    for attempt in range(5):
        try:
            with session.post(url, json=json_body, timeout=(10, 60)) as response:
                response.raise_for_status()
                if response.status_code != expected:
                    raise RuntimeError(f"Unexpected startup status: {response.status_code}")
                return
        except requests.ConnectionError:
            if attempt == 4:
                raise
            time.sleep(2 ** attempt)


def receive_job(session, url: str):
    while True:
        try:
            with session.get(url, timeout=(10, None)) as response:
                if response.status_code == 200:
                    return response.json()["computeModuleJobV1"]
                if response.status_code == 204:
                    continue
                response.raise_for_status()
                raise RuntimeError(f"Unexpected job status: {response.status_code}")
        except requests.ConnectionError:
            LOG.warning("Internal service connection unavailable")
            time.sleep(1)


def send_result(session, base_url: str, job_id: str, body: bytes) -> None:
    url = base_url.rstrip("/") + "/" + quote(job_id, safe="")
    for attempt in range(5):
        try:
            with session.post(
                url,
                data=body,
                headers={"Content-Type": "application/octet-stream"},
                timeout=(10, 60),
            ) as response:
                response.raise_for_status()
                if response.status_code != 204:
                    raise RuntimeError(f"Unexpected result status: {response.status_code}")
                return
        except requests.RequestException:
            if attempt == 4:
                raise
            time.sleep(1)


def main() -> None:
    logging.basicConfig(level=logging.INFO)
    token = Path(os.environ["MODULE_AUTH_TOKEN"]).read_text().strip()
    with requests.Session() as session:
        session.verify = os.environ["DEFAULT_CA_PATH"]
        session.headers["Module-Auth-Token"] = token
        post_at_startup(session, os.environ["POST_SCHEMA_URI"], json_body=SCHEMAS)
        restart_uri = os.environ.get("RESTART_NOTIFICATION_URI")
        if restart_uri:
            post_at_startup(session, restart_uri, expected=200)
        while True:
            job = receive_job(session, os.environ["GET_JOB_URI"])
            job_id = job["jobId"]
            try:
                result = calculate(job["queryType"], job["query"])
                body = json.dumps(result, allow_nan=False).encode("utf-8")
            except Exception as error:
                body = json.dumps({
                    "exception": f"{type(error).__name__}: {error}"
                }).encode("utf-8")
            send_result(session, os.environ["POST_RESULT_URI"], job_id, body)


if __name__ == "__main__":
    main()

초기 연결 거부는 최대 다섯 번 시도하며 재시도 사이에 1, 2, 4, 8초 기다립니다. 결과 전송도 최대 다섯 번 시도합니다. 끝내 수락되지 않은 결과를 버리고 다음 작업으로 넘어가지 않고 프로세스에 실패를 전파합니다.

계산 및 결과 직렬화 예외는 Python SDK의 exception 필드 형식으로 같은 작업에 보고합니다. 알 수 없는 함수 이름이나 잘못된 입력을 정상 숫자 결과로 바꾸지 않습니다. 일반 결과와 오류 전송 모두 TLS 및 HTTP 상태를 확인합니다.

런타임이 RESTART_NOTIFICATION_URI를 제공하면 시작 시 POST하고 200을 확인합니다. 이 알림은 이전 컨테이너에 남은 진행 중 작업을 정리합니다. 해당 알림을 지원하지 않는 이전 런타임은 재시작 뒤 작업이 오래 진행 중으로 남을 수 있으므로 런타임의 복구 조건을 확인해야 합니다.

이 예제는 한 번에 한 작업을 처리합니다. 여러 작업을 병렬로 처리하는 클라이언트는 시작 시 MAX_CONCURRENT_TASKS를 읽어 worker 수를 맞추고 작업별 결과 및 실패를 각각 보고해야 합니다.