Python으로 Composition 함수 작성하기
Python으로 Composition 함수 작성하기 (Write a Composition Function in Python)
Composition 함수(줄여서 함수)는 크로스플레인 리소스를 템플릿화하는 사용자 정의 프로그램입니다. 크로스플레인은 복합 리소스(XR)를 생성할 때 생성할 리소스를 결정하기 위해 Composition 함수를 호출합니다. Composition 함수에 대한 자세한 내용은 concepts 페이지를 읽어 보세요.
출처: 문서
본문
일반 목적 프로그래밍 언어를 사용해 리소스를 템플릿화하는 함수를 작성할 수 있습니다. 일반 목적 프로그래밍 언어를 사용하면 함수가 루프(loop)나 조건문(conditional) 같은 고급 로직으로 리소스를 템플릿화할 수 있습니다. 이 가이드는 Python으로 Composition 함수를 작성하는 방법을 설명합니다.
중요 이 가이드를 따르기 전에 Composition 함수가 어떻게 동작하는지 익숙해지는 것이 좋습니다.
단계 이해하기 (Understand the steps)
이 가이드는 XBuckets 복합 리소스(XR)를 위한 Composition 함수 작성 방법을 다룹니다.
apiVersion: example.crossplane.io/v1
kind: XBuckets
metadata:
name: example-buckets
spec:
region: us-east-2
names:
- crossplane-functions-example-a
- crossplane-functions-example-b
- crossplane-functions-example-c
XBuckets XR은 리전(region)과 버킷 이름 배열을 가집니다. 함수는 names 배열의 각 항목에 대해 Amazon Web Services(AWS) S3 버킷을 생성합니다.
Python으로 함수를 작성하려면:
- 함수 작성에 필요한 도구 설치
- 템플릿에서 함수 초기화
- 함수의 로직을 추가하도록 템플릿 편집
- 함수를 엔드투엔드로 테스트
- 함수를 빌드하고 패키지 저장소에 푸시
이 가이드는 각 단계를 자세히 다룹니다.
함수 작성에 필요한 도구 설치 (Install the tools)
Python으로 함수를 작성하려면 다음이 필요합니다.
- Python v3.11.
- Hatch — Python 빌드 도구. 이 가이드는 v1.7을 사용합니다.
- Docker Engine. 이 가이드는 Engine v24를 사용합니다.
- Crossplane CLI v1.14 이상. 이 가이드는 Crossplane CLI v1.14를 사용합니다.
참고 Composition 함수를 빌드하거나 테스트하기 위해 쿠버네티스 클러스터나 크로스플레인 컨트롤 플레인에 접근할 필요는 없습니다.
템플릿에서 함수 초기화 (Initialize the function from a template)
crossplane xpkg init 명령을 사용하여 새 함수를 초기화합니다. 이 명령을 실행하면 GitHub 저장소를 템플릿으로 사용하여 함수를 초기화합니다.
crossplane xpkg init function-xbuckets https://github.com/crossplane/function-template-python -d function-xbuckets
Initialized package "function-xbuckets" in directory "/home/negz/control/negz/function-xbuckets" from https://github.com/crossplane/function-template-python/tree/bfed6923ab4c8e7adeed70f41138645fc7d38111 (main)
crossplane xpkg init 명령은 function-xbuckets라는 디렉터리를 생성합니다. 명령을 실행하면 새 디렉터리는 다음과 같아야 합니다.
ls function-xbuckets
Dockerfile example/ function/ LICENSE package/ pyproject.toml README.md renovate.json tests/
함수 코드는 function 디렉터리에 있습니다.
ls function/
__version__.py fn.py main.py
function/fn.py 파일은 함수 코드를 추가하는 곳입니다. 템플릿의 다른 파일들에 대해 알아두면 유용합니다.
function/main.py는 함수를 실행합니다. main.py를 편집할 필요는 없습니다.Dockerfile은 함수 런타임을 빌드합니다. Dockerfile을 편집할 필요는 없습니다.package디렉터리는 함수 패키지를 빌드하는 데 사용되는 메타데이터를 포함합니다.
팁 Crossplane CLI v1.14에서
crossplane xpkg init은 템플릿 GitHub 저장소를 복제할 뿐입니다. 향후 CLI 릴리스는 템플릿 이름을 새 함수 이름으로 바꾸는 것 같은 작업을 자동화할 것입니다. 자세한 내용은 Crossplane issue #4941을 참고하세요.
코드를 추가하기 전에 package/crossplane.yaml을 편집하여 패키지의 이름을 바꿉니다. 패키지 이름을 function-xbuckets로 지정합니다.
package/input 디렉터리는 함수 입력의 OpenAPI 스키마를 정의합니다. 이 가이드의 함수는 입력을 받지 않습니다. package/input 디렉터리를 삭제하세요.
Composition 함수 문서는 Composition 함수 입력을 설명합니다.
팁 입력을 사용하는 함수를 작성한다면 함수 요구 사항에 맞게 입력 YAML 파일을 편집하세요. 입력의 kind와 API group을 변경하세요.
Input과template.fn.crossplane.io를 사용하지 말고 함수에 의미 있는 값을 사용하세요.
함수 로직을 추가하도록 템플릿 편집 (Edit the template)
function/fn.py의 RunFunction 메서드에 함수 로직을 추가합니다. 파일을 처음 열면 "hello world" 함수가 들어 있습니다.
async def RunFunction(self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext) -> fnv1.RunFunctionResponse:
log = self.log.bind(tag=req.meta.tag)
log.info("Running function")
rsp = response.to(req)
example = ""
if "example" in req.input:
example = req.input["example"]
# TODO: Add your function logic here!
response.normal(rsp, f"I was run with input {example}!")
log.info("I was run!", input=example)
return rsp
모든 Python Composition 함수는 RunFunction 메서드를 가집니다. 크로스플레인은 함수를 실행하는 데 필요한 모든 것을 RunFunctionRequest 객체에 전달합니다.
함수는 RunFunctionResponse 객체를 반환하여 크로스플레인에 구성해야 할 리소스를 알려줍니다.
RunFunction 메서드를 편집하여 다음 코드로 바꿉니다.
async def RunFunction(self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext) -> fnv1.RunFunctionResponse:
log = self.log.bind(tag=req.meta.tag)
log.info("Running function")
rsp = response.to(req)
region = req.observed.composite.resource["spec"]["region"]
names = req.observed.composite.resource["spec"]["names"]
for name in names:
rsp.desired.resources[f"xbuckets-{name}"].resource.update(
{
"apiVersion": "s3.aws.m.upbound.io/v1beta1",
"kind": "Bucket",
"metadata": {
"annotations": {
"crossplane.io/external-name": name,
},
},
"spec": {
"forProvider": {
"region": region,
},
},
}
)
log.info("Added desired buckets", region=region, count=len(names))
return rsp
아래 블록을 확장하여 imports와 함수 로직을 설명하는 주석을 포함한 전체 fn.py를 확인하세요.
전체 fn.py 파일:
"""A Crossplane composition function."""
import grpc
from crossplane.function import logging, response
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from crossplane.function.proto.v1 import run_function_pb2_grpc as grpcv1
class FunctionRunner(grpcv1.FunctionRunnerService):
"""A FunctionRunner handles gRPC RunFunctionRequests."""
def __init__(self):
"""Create a new FunctionRunner."""
self.log = logging.get_logger()
async def RunFunction(
self, req: fnv1.RunFunctionRequest, _: grpc.aio.ServicerContext
) -> fnv1.RunFunctionResponse:
"""Run the function."""
# Create a logger for this request.
log = self.log.bind(tag=req.meta.tag)
log.info("Running function")
# Create a response to the request. This copies the desired state and
# pipeline context from the request to the response.
rsp = response.to(req)
# Get the region and a list of bucket names from the observed composite
# resource (XR). Crossplane represents resources using the Struct
# well-known protobuf type. The Struct Python object can be accessed
# like a dictionary.
region = req.observed.composite.resource["spec"]["region"]
names = req.observed.composite.resource["spec"]["names"]
# Add a desired S3 bucket for each name.
for name in names:
# Crossplane represents desired composed resources using a protobuf
# map of messages. This works a little like a Python defaultdict.
# Instead of assigning to a new key in the dict-like map, you access
# the key and mutate its value as if it did exist.
#
# The below code works because accessing the xbuckets-{name} key
# automatically creates a new, empty fnv1.Resource message. The
# Resource message has a resource field containing an empty Struct
# object that can be populated from a dictionary by calling update.
#
# https://protobuf.dev/reference/python/python-generated/#map-fields
rsp.desired.resources[f"xbuckets-{name}"].resource.update(
{
"apiVersion": "s3.aws.m.upbound.io/v1beta1",
"kind": "Bucket",
"metadata": {
"annotations": {
"crossplane.io/external-name": name,
},
},
"spec": {
"forProvider": {
"region": region,
},
},
}
)
# Log what the function did. This will only appear in the function's pod
# logs. A function can use response.normal() and response.warning() to
# emit Kubernetes events associated with the XR it's operating on.
log.info("Added desired buckets", region=region, count=len(names))
return rsp
이 코드는:
- RunFunctionRequest에서 관찰된 복합 리소스(observed composite resource)를 가져옵니다.
- 관찰된 복합 리소스에서 리전과 버킷 이름을 가져옵니다.
- 각 버킷 이름에 대해 하나의 원하는 S3 버킷을 추가합니다.
- RunFunctionResponse에서 원하는 S3 버킷을 반환합니다.
크로스플레인은 Python으로 Composition 함수를 작성하기 위한 소프트웨어 개발 키트(SDK)를 제공합니다. 이 함수는 SDK의 유틸리티를 사용합니다.
팁 Python Function SDK 문서를 읽어 보세요.
중요 Python SDK는 Protocol Buffers 스키마에서
RunFunctionRequest와RunFunctionResponsePython 객체를 자동으로 생성합니다. 스키마는 Buf Schema Registry에서 확인할 수 있습니다.
생성된 Python 객체의 필드는 dict와 list 같은 내장 Python 유형과 비슷하게 동작합니다. 몇 가지 차이점이 있음을 인지하세요.
특히, 관찰된 리소스와 원하는 리소스의 맵에는 딕셔너리처럼 접근하지만, 맵 키에 할당하여 새 원하는 리소스를 추가할 수는 없습니다. 대신 맵 키에 접근해서 이미 존재하는 것처럼 변경합니다.
이렇게 새 리소스를 추가하는 대신:
resource = {"apiVersion": "example.org/v1", "kind": "Composed", ...}
rsp.desired.resources["new-resource"] = fnv1.Resource(resource=resource)
이미 존재한다고 가정하고 이렇게 변경합니다:
resource = {"apiVersion": "example.org/v1", "kind": "Composed", ...}
rsp.desired.resources["new-resource"].resource.update(resource)
자세한 내용은 Protocol Buffers Python Generated Code Guide를 참고하세요.
함수를 엔드투엔드로 테스트 (Test the function end-to-end)
단위 테스트(unit tests)를 추가하고 crossplane composition render 명령을 사용하여 함수를 테스트합니다.
템플릿에서 함수를 초기화하면 tests/test_fn.py에 일부 단위 테스트가 추가됩니다. 이 테스트들은 Python 표준 라이브러리의 unittest 모듈을 사용합니다.
테스트 케이스를 추가하려면 test_run_function의 cases 목록을 업데이트합니다. 아래 블록을 확장하여 이 함수의 전체 tests/test_fn.py 파일을 확인하세요.
전체 test_fn.py 파일:
import dataclasses
import unittest
from crossplane.function import logging, resource
from crossplane.function.proto.v1 import run_function_pb2 as fnv1
from google.protobuf import duration_pb2 as durationpb
from google.protobuf import json_format
from google.protobuf import struct_pb2 as structpb
from function import fn
class TestFunctionRunner(unittest.IsolatedAsyncioTestCase):
def setUp(self) -> None:
logging.configure(level=logging.Level.DISABLED)
self.maxDiff = 2000
async def test_run_function(self) -> None:
@dataclasses.dataclass
class TestCase:
reason: str
req: fnv1.RunFunctionRequest
want: fnv1.RunFunctionResponse
cases = [
TestCase(
reason="The function should compose two S3 buckets.",
req=fnv1.RunFunctionRequest(
observed=fnv1.State(
composite=fnv1.Resource(
resource=resource.dict_to_struct(
{
"apiVersion": "example.crossplane.io/v1alpha1",
"kind": "XBuckets",
"metadata": {"name": "test"},
"spec": {
"region": "us-east-2",
"names": ["test-bucket-a", "test-bucket-b"],
},
}
)
)
)
),
want=fnv1.RunFunctionResponse(
meta=fnv1.ResponseMeta(ttl=durationpb.Duration(seconds=60)),
desired=fnv1.State(
resources={
"xbuckets-test-bucket-a": fnv1.Resource(
resource=resource.dict_to_struct(
{
"apiVersion": "s3.aws.m.upbound.io/v1beta1",
"kind": "Bucket",
"metadata": {
"annotations": {
"crossplane.io/external-name": "test-bucket-a"
},
},
"spec": {
"forProvider": {"region": "us-east-2"}
},
}
)
),
"xbuckets-test-bucket-b": fnv1.Resource(
resource=resource.dict_to_struct(
{
"apiVersion": "s3.aws.m.upbound.io/v1beta1",
"kind": "Bucket",
"metadata": {
"annotations": {
"crossplane.io/external-name": "test-bucket-b"
},
},
"spec": {
"forProvider": {"region": "us-east-2"}
},
}
)
),
},
),
context=structpb.Struct(),
),
),
]
runner = fn.FunctionRunner()
for case in cases:
got = await runner.RunFunction(case.req, None)
self.assertEqual(
json_format.MessageToDict(got),
json_format.MessageToDict(case.want),
"-want, +got",
)
if __name__ == "__main__":
unittest.main()
hatch run을 사용하여 단위 테스트를 실행합니다.
hatch run test:unit
.
----------------------------------------------------------------------
Ran 1 test in 0.003s
OK
팁 Hatch는 Python 빌드 도구입니다. wheel 같은 Python 아티팩트를 빌드합니다. 또한
virtualenv나venv와 유사하게 가상 환경을 관리합니다.hatch run명령은 가상 환경을 만들고 그 환경에서 명령을 실행합니다.
Crossplane CLI를 사용하여 이 함수를 사용하는 Composition의 출력을 미리 볼 수 있습니다. 이를 위해 크로스플레인 컨트롤 플레인이 필요하지 않습니다.
function-xbuckets 아래에 example이라는 디렉터리를 만들고 Composite Resource, Composition, Function YAML 파일을 생성합니다.
다음 블록을 확장하여 예시 파일을 확인하세요.
xr.yaml, composition.yaml, function.yaml 파일 — 다음 파일들로 crossplane composition render를 실행하여 아래 출력을 재현할 수 있습니다.
xr.yaml 파일은 렌더링할 복합 리소스를 포함합니다.
apiVersion: example.crossplane.io/v1
kind: XBuckets
metadata:
name: example-buckets
spec:
region: us-east-2
names:
- crossplane-functions-example-a
- crossplane-functions-example-b
- crossplane-functions-example-c
composition.yaml 파일은 복합 리소스를 렌더링하는 데 사용할 Composition을 포함합니다.
apiVersion: apiextensions.crossplane.io/v1
kind: Composition
metadata:
name: create-buckets
spec:
compositeTypeRef:
apiVersion: example.crossplane.io/v1
kind: XBuckets
mode: Pipeline
pipeline:
- step: create-buckets
functionRef:
name: function-xbuckets
functions.yaml 파일은 Composition이 파이프라인 단계에서 참조하는 Functions를 포함합니다.
apiVersion: pkg.crossplane.io/v1
kind: Function
metadata:
name: function-xbuckets
annotations:
render.crossplane.io/runtime: Development
spec:
# The CLI ignores this package when using the Development runtime.
# You can set it to any value.
package: xpkg.crossplane.io/negz/function-xbuckets:v0.1.0
functions.yaml의 Function은 Development 런타임을 사용합니다. 이것은 crossplane composition render에게 함수가 로컬에서 실행 중임을 알려줍니다. Docker를 사용해 함수를 가져와 실행하는 대신 로컬에서 실행 중인 함수에 연결합니다.
apiVersion: pkg.crossplane.io/v1
kind: Function
metadata:
name: function-xbuckets
annotations:
render.crossplane.io/runtime: Development
hatch run development를 사용하여 함수를 로컬에서 실행합니다.
hatch run development
경고
hatch run development는 함수를 암호화나 인증 없이 실행합니다. 테스트와 개발 중에만 사용하세요.
별도의 터미널에서 crossplane composition render를 실행합니다.
crossplane composition render xr.yaml composition.yaml functions.yaml
이 명령은 함수를 호출합니다. 함수가 실행 중인 터미널에서 로그 출력을 볼 수 있어야 합니다.
hatch run development
2024-01-11T22:12:58.153572Z [info ] Running function filename=fn.py lineno=22 tag=
2024-01-11T22:12:58.153792Z [info ] Added desired buckets count=3 filename=fn.py lineno=68 region=us-east-2 tag=
crossplane composition render 명령은 함수가 반환하는 원하는 리소스를 출력합니다.
---
apiVersion: example.crossplane.io/v1
kind: XBuckets
metadata:
name: example-buckets
---
apiVersion: s3.aws.m.upbound.io/v1beta1
kind: Bucket
metadata:
annotations:
crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-b
crossplane.io/external-name: crossplane-functions-example-b
generateName: example-buckets-
labels:
crossplane.io/composite: example-buckets
ownerReferences:
# Omitted for brevity
spec:
forProvider:
region: us-east-2
---
apiVersion: s3.aws.m.upbound.io/v1beta1
kind: Bucket
metadata:
annotations:
crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-c
crossplane.io/external-name: crossplane-functions-example-c
generateName: example-buckets-
labels:
crossplane.io/composite: example-buckets
ownerReferences:
# Omitted for brevity
spec:
forProvider:
region: us-east-2
---
apiVersion: s3.aws.m.upbound.io/v1beta1
kind: Bucket
metadata:
annotations:
crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-a
crossplane.io/external-name: crossplane-functions-example-a
generateName: example-buckets-
labels:
crossplane.io/composite: example-buckets
ownerReferences:
# Omitted for brevity
spec:
forProvider:
region: us-east-2
팁 Composition 함수 테스트에 대해 더 자세히 알아보려면 Composition 함수 문서를 읽어 보세요.
함수를 빌드하고 패키지 레지스트리에 푸시 (Build and push)
함수는 두 단계로 빌드합니다. 먼저 함수의 런타임을 빌드합니다. 이것은 크로스플레인이 함수를 실행하는 데 사용하는 Open Container Initiative(OCI) 이미지입니다. 그런 다음 그 런타임을 패키지에 포함시키고 패키지 레지스트리에 푸시합니다. Crossplane CLI는 기본 패키지 레지스트리로 xpkg.crossplane.io를 사용합니다.
함수는 기본적으로 linux/amd64 같은 단일 플랫폼을 지원합니다. 각 플랫폼에 대해 런타임과 패키지를 빌드한 다음 모든 패키지를 레지스트리의 단일 태그에 푸시하여 여러 플랫폼을 지원할 수 있습니다.
함수를 레지스트리에 푸시하면 크로스플레인 컨트롤 플레인에서 함수를 사용할 수 있습니다. 컨트롤 플레인에서 함수를 사용하는 방법은 Composition 함수 문서를 참고하세요.
각 플랫폼에 대한 런타임을 빌드하려면 Docker를 사용합니다.
docker build . --quiet --platform=linux/amd64 --tag runtime-amd64
sha256:fdf40374cc6f0b46191499fbc1dbbb05ddb76aca854f69f2912e580cfe624b4b
docker build . --quiet --platform=linux/arm64 --tag runtime-arm64
sha256:cb015ceabf46d2a55ccaeebb11db5659a2fb5e93de36713364efcf6d699069af
팁 원하는 태그를 사용할 수 있습니다. 런타임 이미지를 레지스트리에 푸시할 필요는 없습니다. 태그는
crossplane xpkg build에게 포함할 런타임을 알려주는 데만 사용됩니다.
중요 Docker는 에뮬레이션을 사용하여 다른 플랫폼용 이미지를 만듭니다. 다른 플랫폼용 이미지 빌드가 실패하면
binfmt를 설치했는지 확인하세요. 지침은 Docker 문서를 참고하세요.
각 플랫폼에 대한 패키지를 빌드하려면 Crossplane CLI를 사용합니다. 각 패키지는 런타임 이미지를 포함합니다.
--package-root 플래그는 crossplane.yaml을 포함하는 package 디렉터리를 지정합니다. 이것은 패키지에 대한 메타데이터를 포함합니다.
--embed-runtime-image 플래그는 Docker로 빌드한 런타임 이미지 태그를 지정합니다.
--package-file 플래그는 패키지 파일을 디스크에 쓸 위치를 지정합니다. 크로스플레인 패키지 파일은 .xpkg 확장자를 사용합니다.
crossplane xpkg build \
--package-root=package \
--embed-runtime-image=runtime-amd64 \
--package-file=function-amd64.xpkg
crossplane xpkg build \
--package-root=package \
--embed-runtime-image=runtime-arm64 \
--package-file=function-arm64.xpkg
팁 크로스플레인 패키지는 특별한 OCI 이미지입니다. 패키지에 대한 자세한 내용은 packages 문서에서 읽어 보세요.
두 패키지 파일을 레지스트리에 푸시합니다. 두 파일을 레지스트리의 한 태그에 푸시하면 linux/arm64와 linux/amd64 호스트에서 모두 실행되는 다중 플랫폼(multi-platform) 패키지가 생성됩니다.
crossplane xpkg push \
--package-files=function-amd64.xpkg,function-arm64.xpkg \
negz/function-xbuckets:v0.1.0
팁 함수를 GitHub 저장소에 푸시하면 템플릿이 GitHub Actions를 사용하여 지속적 통합(CI)을 자동으로 설정합니다. CI 워크플로우는 함수를 lint, test, build합니다. 템플릿이 CI를 구성하는 방법은
.github/workflows/ci.yaml을 읽어 확인할 수 있습니다.