カスタム Compute Module クライアントを作成する（高度な内容）

なぜクライアントが必要なのですか？

サポートされている言語（Python、Java、または TypeScript）で Compute Module を記述し、利用可能な SDK を使用する場合は、クライアントロジックを手動で作成する必要はありません。

ただし、SDK を使用しない場合や、現在サポートされていない言語で Compute Module を作成する必要がある場合は、以下で説明する手順に従って Compute Module クライアントをご自身で作成する必要があります。

カスタムクライアントの責務

Compute Module クライアントは、Compute Module 内のロジックの実行を管理し、次の3つの主要な責務を担います。

Compute Module の関数スキーマを送信する（任意）

クライアントのメイン実行サイクルを開始する前に、Compute Module の関数スキーマを公開することを推奨します。これにより、Compute Module のスキーマが Foundry のほかの部分に公開されます。または、Compute Module の関数タブでこの関数スキーマを手動で定義することもできます。

詳細については、関数スキーマの自動推論に関するドキュメントを参照してください。

新しいジョブをポーリングする

クライアントは、処理が必要な新しいジョブがないか、内部の Compute Module サービスを継続的にポーリングします。ジョブが存在する場合、クライアントはそのジョブに対応する関数を見つけ、関連付けられたペイロードを渡してその関数を呼び出します。

Compute Module の結果（出力）を送信する

関数が完了してクライアントに結果を返すと、クライアントはその出力を Compute Module サービスに報告する責務を担います。

以下は、Compute Module クライアントの実行ライフサイクルを簡単に図示したものです。

API リファレンス

ジョブの GET

変数

MODULE_AUTH_TOKEN string

• 詳細については、環境変数に関するドキュメントを参照してください。

DEFAULT_CA_PATH string

• 詳細については、環境変数に関するドキュメントを参照してください。

GET_JOB_URI string

• 詳細については、環境変数に関するドキュメントを参照してください。

想定されるステータスコード

• 200: 処理対象の新しいジョブが存在します。
• 204: 処理対象の新しいジョブは存在しません。

curl --header "Module-Auth-Token: $MODULE_AUTH_TOKEN" \
    --cacert $DEFAULT_CA_PATH \
    --request GET \
    $GET_JOB_URI

レスポンスパラメーター

jobId string

• 処理対象のジョブの一意の識別子です。

queryType string

• 呼び出す関数の名前です。

query JSON object

• 関数に渡すペイロードです。

temporaryCredentialsAuthToken string

• Foundry データサイドカーで使用する一時トークンです。

authHeader string

• Foundry 内のほかのサービスを呼び出すために使用できる Foundry 認可トークンです。
• 特定のモードでのみ利用可能です。

Copied!
1
2
3
4
5
6
7
8
9
10
11
{
    "computeModuleJobV1": {
        "jobId": "9a2a1e94-41d3-47d7-807f-db2f4c547b9c",
        "queryType": "multiply",
        "query": {
            "n": 4.0
        },
        "temporaryCredentialsAuthToken": "token-data",
        "authHeader": "auth-header"
    }
}

結果の POST

変数

result_data octet-stream

• Compute Module の関数から返された結果です。

jobId string

• 対応するジョブの GET リクエストから返された jobId です。

MODULE_AUTH_TOKEN string

• 詳細については、環境変数に関するドキュメントを参照してください。

DEFAULT_CA_PATH string

• 詳細については、環境変数に関するドキュメントを参照してください。

POST_RESULT_URI string

• 詳細については、環境変数に関するドキュメントを参照してください。

想定されるステータスコード

• 204: リクエストが受け付けられました。

レスポンスパラメーター

なし

curl --header "Content-Type: application/octet-stream" \
    --header "Module-Auth-Token: $MODULE_AUTH_TOKEN" \
    --cacert $DEFAULT_CA_PATH \
    --request POST \
    --data $result_data \
    $POST_RESULT_URI/$jobId

関数スキーマの POST

変数

schema_data JSON array

• JSON 形式の Compute Module の関数スキーマです。

MODULE_AUTH_TOKEN string

• 詳細については、環境変数に関するドキュメントを参照してください。

DEFAULT_CA_PATH string

• 詳細については、環境変数に関するドキュメントを参照してください。

POST_SCHEMA_URI string

• 詳細については、環境変数に関するドキュメントを参照してください。

想定されるステータスコード

204: リクエストが受け付けられました。

レスポンスパラメーター

なし

curl --header "Content-Type: application/json" \
    --header "Module-Auth-Token: $MODULE_AUTH_TOKEN" \
    --cacert $DEFAULT_CA_PATH \
    --request POST \
    --data $schema_data \
    $POST_SCHEMA_URI

Python の例

app.py

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import json
import logging as log
import os
import socket
import time

import requests

log.basicConfig(level=log.INFO)

certPath = os.environ["DEFAULT_CA_PATH"]

with open(os.environ["MODULE_AUTH_TOKEN"], "r") as f:
    moduleAuthToken = f.read()

ip = socket.gethostbyname(socket.gethostname())

getJobUri = f"https://{ip}:8945/interactive-module/api/internal-query/job"
postResultUri = f"https://{ip}:8945/interactive-module/api/internal-query/results"
postSchemaUri = f"https://{ip}:8945/interactive-module/api/internal-query/schemas"

SCHEMAS = [
    {
        "functionName": "multiply",
        "inputs": [
            {
                "name": "n",
                "dataType": {"float": {}, "type": "float"},
                "required": True,
                "constraints": [],
            },
        ],
        "output": {
            "single": {
                "dataType": {
                    "float": {},
                    "type": "float",
                }
            },
            "type": "single",
        },
    },
    {
        "functionName": "divide",
        "inputs": [
            {
                "name": "n",
                "dataType": {"float": {}, "type": "float"},
                "required": True,
                "constraints": [],
            },
        ],
        "output": {
            "single": {
                "dataType": {
                    "float": {},
                    "type": "float",
                }
            },
            "type": "single",
        },
    },
]


# Compute Module サービスからジョブを取得します。ジョブが存在するのは、
# ステータスコードが200の場合のみです。ステータスコード204が返された場合は、再試行します。
# このエンドポイントではロングポーリングが有効で、待機せずに呼び出せます。
def getJobBlocking():
    while True:
        response = requests.get(getJobUri, headers={"Module-Auth-Token": moduleAuthToken}, verify=certPath)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 204:
            log.info("No job found, trying again!")


# 種類に基づいてクエリを処理します
def get_result(query_type, query):
    if query_type == "multiply":
        return float(query["n"]) * 2
    elif query_type == "divide":
        return float(query["n"]) / 2
    else:
        log.info(f"Unknown query type: {query_type}")


# ジョブの結果を Compute Module サービスに送信します。受信したすべてのジョブについて結果を送信する必要があります。
# そうしないと、新しいジョブがこのワーカーにルーティングされない場合があります。
def postResult(jobId, result):
    response = requests.post(
        f"{postResultUri}/{jobId}",
        data=json.dumps(result).encode("utf-8"),
        headers={"Module-Auth-Token": moduleAuthToken, "Content-Type": "application/octet-stream"},
        verify=certPath,
    )
    if response.status_code != 204:
        log.info(f"Failed to post result: {response.status_code}")


# この Compute Module の関数のスキーマを Compute Module サービスに送信します。
# 呼び出しは1回のみ必要なため、メインループに入る前に呼び出します。
def postSchema():
    num_tries = 0
    success = False
    while not success and num_tries < 5:
        try:
            response = requests.post(
                postSchemaUri,
                json=SCHEMAS,
                headers={"Module-Auth-Token": moduleAuthToken, "Content-Type": "application/json"},
                verify=certPath,
            )
            success = True
            log.info(f"POST schema status: {response.status_code}")
            log.info(f"POST schema text: {response.text} reason: {response.reason}")
        except Exception as e:
            log.error(f"Exception occurred posting schema: {e}")
            time.sleep(2**num_tries)
            num_tries += 1


postSchema()

# 無期限に試行します
while True:
    try:
        job = getJobBlocking()
        v1 = job["computeModuleJobV1"]
        job_id = v1["jobId"]
        query_type = v1["queryType"]
        query = v1["query"]
        result = get_result(query_type, query)
        postResult(job_id, result)
    except Exception as e:
        log.info(f"Something failed {str(e)}")
        time.sleep(1)