用 Python 编写一个composition函数

 1"""A Crossplane composition function."""
 2
 3import grpc
 4from crossplane.function import logging, response
 5from crossplane.function.proto.v1beta1 import run_function_pb2 as fnv1beta1
 6from crossplane.function.proto.v1beta1 import run_function_pb2_grpc as grpcv1beta1
 7
 8class FunctionRunner(grpcv1beta1.FunctionRunnerService):
 9    """A FunctionRunner handles gRPC RunFunctionRequests."""
10
11    def __init__(self):
12        """Create a new FunctionRunner."""
13        self.log = logging.get_logger()
14
15    async def RunFunction(
16        self, req: fnv1beta1.RunFunctionRequest, _: grpc.aio.ServicerContext
17    ) -> fnv1beta1.RunFunctionResponse:
18        """Run the function."""
19        # Create a logger for this request.
20        log = self.log.bind(tag=req.meta.tag)
21        log.info("Running function")
22
23        # Create a response to the request. This copies the desired state and
24        # pipeline context from the request to the response.
25        rsp = response.to(req)
26
27        # Get the region and a list of bucket names from the observed composite
28        # resource (XR). Crossplane represents resources using the Struct
29        # well-known protobuf type. The Struct Python object can be accessed
30        # like a dictionary.
31        region = req.observed.composite.resource["spec"]["region"]
32        names = req.observed.composite.resource["spec"]["names"]
33
34        # Add a desired S3 bucket for each name.
35        for name in names:
36            # Crossplane represents desired composed resources using a protobuf
37            # map of messages. This works a little like a Python defaultdict.
38            # Instead of assigning to a new key in the dict-like map, you access
39            # the key and mutate its value as if it did exist.
40            #
41            # The below code works because accessing the xbuckets-{name} key
42            # automatically creates a new, empty fnv1beta1.Resource message. The
43            # Resource message has a resource field containing an empty Struct
44            # object that can be populated from a dictionary by calling update.
45            #
46            # https://protobuf.dev/reference/python/python-generated/#map-fields
47            rsp.desired.resources[f"xbuckets-{name}"].resource.update(
48                {
49                    "apiVersion": "s3.aws.upbound.io/v1beta1",
50                    "kind": "Bucket",
51                    "metadata": {
52                        "annotations": {
53                            "crossplane.io/external-name": name,
54                        },
55                    },
56                    "spec": {
57                        "forProvider": {
58                            "region": region,
59                        },
60                    },
61                }
62            )
63
64        # Log what the function did. This will only appear in the function's pod
65        # logs. A function can use response.normal() and response.warning() to
66        # emit Kubernetes events associated with the XR it's operating on.
67        log.info("Added desired buckets", region=region, count=len(names))
68
69        return rsp

 1import dataclasses
 2import unittest
 3
 4from crossplane.function import logging, resource
 5from crossplane.function.proto.v1beta1 import run_function_pb2 as fnv1beta1
 6from google.protobuf import duration_pb2 as durationpb
 7from google.protobuf import json_format
 8from google.protobuf import struct_pb2 as structpb
 9
10from function import fn
11
12class TestFunctionRunner(unittest.IsolatedAsyncioTestCase):
13    def setUp(self) -> None:
14        logging.configure(level=logging.Level.DISABLED)
15        self.maxDiff = 2000
16
17    async def test_run_function(self) -> None:
18        @dataclasses.dataclass
19        class TestCase:
20            reason: str
21            req: fnv1beta1.RunFunctionRequest
22            want: fnv1beta1.RunFunctionResponse
23
24        cases = [
25            TestCase(
26                reason="The function should compose two S3 buckets.",
27                req=fnv1beta1.RunFunctionRequest(
28                    observed=fnv1beta1.State(
29                        composite=fnv1beta1.Resource(
30                            resource=resource.dict_to_struct(
31                                {
32                                    "apiVersion": "example.crossplane.io/v1alpha1",
33                                    "kind": "XBuckets",
34                                    "metadata": {"name": "test"},
35                                    "spec": {
36                                        "region": "us-east-2",
37                                        "names": ["test-bucket-a", "test-bucket-b"],
38                                    },
39                                }
40                            )
41                        )
42                    )
43                ),
44                want=fnv1beta1.RunFunctionResponse(
45                    meta=fnv1beta1.ResponseMeta(ttl=durationpb.Duration(seconds=60)),
46                    desired=fnv1beta1.State(
47                        resources={
48                            "xbuckets-test-bucket-a": fnv1beta1.Resource(
49                                resource=resource.dict_to_struct(
50                                    {
51                                        "apiVersion": "s3.aws.upbound.io/v1beta1",
52                                        "kind": "Bucket",
53                                        "metadata": {
54                                            "annotations": {
55                                                "crossplane.io/external-name": "test-bucket-a"
56                                            },
57                                        },
58                                        "spec": {
59                                            "forProvider": {"region": "us-east-2"}
60                                        },
61                                    }
62                                )
63                            ),
64                            "xbuckets-test-bucket-b": fnv1beta1.Resource(
65                                resource=resource.dict_to_struct(
66                                    {
67                                        "apiVersion": "s3.aws.upbound.io/v1beta1",
68                                        "kind": "Bucket",
69                                        "metadata": {
70                                            "annotations": {
71                                                "crossplane.io/external-name": "test-bucket-b"
72                                            },
73                                        },
74                                        "spec": {
75                                            "forProvider": {"region": "us-east-2"}
76                                        },
77                                    }
78                                )
79                            ),
80                        },
81                    ),
82                    context=structpb.Struct(),
83                ),
84            ),
85        ]
86
87        runner = fn.FunctionRunner()
88
89        for case in cases:
90            got = await runner.RunFunction(case.req, None)
91            self.assertEqual(
92                json_format.MessageToDict(got),
93                json_format.MessageToDict(case.want),
94                "-want, +got",
95            )
96
97if __name__ == "__main__":
98    unittest.main()

 1apiVersion: example.crossplane.io/v1
 2kind: XBuckets
 3metadata:
 4  name: example-buckets
 5spec:
 6  region: us-east-2
 7  names:
 8  - crossplane-functions-example-a
 9  - crossplane-functions-example-b
10  - crossplane-functions-example-c

 1apiVersion: apiextensions.crossplane.io/v1
 2kind: Composition
 3metadata:
 4  name: create-buckets
 5spec:
 6  compositeTypeRef:
 7    apiVersion: example.crossplane.io/v1
 8    kind: XBuckets
 9  mode: Pipeline
10  pipeline:
11  - step: create-buckets
12    functionRef:
13      name: function-xbuckets

 1apiVersion: pkg.crossplane.io/v1beta1
 2kind: Function
 3metadata:
 4  name: function-xbuckets
 5  annotations:
 6    render.crossplane.io/runtime: Development
 7spec:
 8  # The CLI ignores this package when using the Development runtime.
 9  # You can set it to any value.
10  package: xpkg.upbound.io/negz/function-xbuckets:v0.1.0