Crossplane logo
Crossplane logo

Write a Composition Function in Python

This is a beta feature.

This feature was introduced in v1.11.
This feature graduated to beta status in v1.14.

For more information read the Crossplane feature lifecycle.

On this page

Composition functions (or just functions, for short) are custom programs that template Crossplane resources. Crossplane calls composition functions to determine what resources it should create when you create a composite resource (XR). Read the [concepts]https://docs.crossplane.io/latest/concepts/composition-functions/ page to learn more about composition functions.

You can write a function to template resources using a general purpose programming language. Using a general purpose programming language allows a function to use advanced logic to template resources, like loops and conditionals. This guide explains how to write a composition function in Python.

Important
It helps to be familiar with [how composition functions work]https://docs.crossplane.io/latest/concepts/composition-functions/#how-composition-functions-work before following this guide.

Understand the steps

This guide covers writing a composition function for an XBuckets composite resource (XR).

 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

An XBuckets XR has a region and an array of bucket names. The function will create an Amazon Web Services (AWS) S3 bucket for each entry in the names array.

To write a function in Python:

  1. Install the tools you need to write the function
  2. Initialize the function from a template
  3. Edit the template to add the function’s logic
  4. Test the function end-to-end
  5. Build and push the function to a package repository

This guide covers each of these steps in detail.

Install the tools you need to write the function

To write a function in Python you need:

Note
You don’t need access to a Kubernetes cluster or a Crossplane control plane to build or test a composition function.

Initialize the function from a template

Use the crossplane beta xpkg init command to initialize a new function. When you run this command it initializes your function using a GitHub repository as a template.

1crossplane beta xpkg init function-xbuckets https://github.com/crossplane/function-template-python -d function-xbuckets
2Initialized package "function-xbuckets" in directory "/home/negz/control/negz/function-xbuckets" from https://github.com/crossplane/function-template-python/tree/bfed6923ab4c8e7adeed70f41138645fc7d38111 (main)

The crossplane beta init xpkg command creates a directory named function-xbuckets. When you run the command the new directory should look like this:

1ls function-xbuckets
2Dockerfile  example/  function/  LICENSE  package/  pyproject.toml  README.md  renovate.json  tests/

Your function’s code lives in the function directory:

1ls function/
2__version__.py  fn.py  main.py

The function/fn.py file is where you add the function’s code. It’s useful to know about some other files in the template:

  • function/main.py runs the function. You don’t need to edit main.py.
  • Dockerfile builds the function runtime. You don’t need to edit Dockerfile.
  • The package directory contains metadata used to build the function package.
Tip

In v1.14 of the Crossplane CLI crossplane beta xpkg init just clones a template GitHub repository. A future CLI release will automate tasks like replacing the template name with the new function’s name. See Crossplane issue #4941 for details.

Edit package/crossplane.yaml to change the package’s name before you start adding code. Name your package function-xbuckets.

The package/input directory defines the OpenAPI schema for the a function’s input. The function in this guide doesn’t accept an input. Delete the package/input directory.

The [composition functions]https://docs.crossplane.io/latest/concepts/composition-functions/ documentation explains composition function inputs.

Tip

If you’re writing a function that uses an input, edit the input YAML file to meet your function’s requirements.

Change the input’s kind and API group. Don’t use Input and template.fn.crossplane.io. Instead use something meaningful to your function.

Edit the template to add the function’s logic

You add your function’s logic to the RunFunction method in function/fn.py. When you first open the file it contains a “hello world” function.

 1async def RunFunction(self, req: fnv1beta1.RunFunctionRequest, _: grpc.aio.ServicerContext) -> fnv1beta1.RunFunctionResponse:
 2    log = self.log.bind(tag=req.meta.tag)
 3    log.info("Running function")
 4
 5    rsp = response.to(req)
 6
 7    example = ""
 8    if "example" in req.input:
 9        example = req.input["example"]
10
11    # TODO: Add your function logic here!
12    response.normal(rsp, f"I was run with input {example}!")
13    log.info("I was run!", input=example)
14
15    return rsp

All Python composition functions have a RunFunction method. Crossplane passes everything the function needs to run in a RunFunctionRequest object.

The function tells Crossplane what resources it should compose by returning a RunFunctionResponse object.

Edit the RunFunction method to replace it with this code.

 1async def RunFunction(self, req: fnv1beta1.RunFunctionRequest, _: grpc.aio.ServicerContext) -> fnv1beta1.RunFunctionResponse:
 2    log = self.log.bind(tag=req.meta.tag)
 3    log.info("Running function")
 4
 5    rsp = response.to(req)
 6
 7    region = req.observed.composite.resource["spec"]["region"]
 8    names = req.observed.composite.resource["spec"]["names"]
 9
10    for name in names:
11        rsp.desired.resources[f"xbuckets-{name}"].resource.update(
12            {
13                "apiVersion": "s3.aws.upbound.io/v1beta1",
14                "kind": "Bucket",
15                "metadata": {
16                    "annotations": {
17                        "crossplane.io/external-name": name,
18                    },
19                },
20                "spec": {
21                    "forProvider": {
22                        "region": region,
23                    },
24                },
25            }
26        )
27
28    log.info("Added desired buckets", region=region, count=len(names))
29
30    return rsp

Expand the below block to view the full fn.py, including imports and commentary explaining the function’s logic.

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

This code:

  1. Gets the observed composite resource from the RunFunctionRequest.
  2. Gets the region and bucket names from the observed composite resource.
  3. Adds one desired S3 bucket for each bucket name.
  4. Returns the desired S3 buckets in a RunFunctionResponse.

Crossplane provides a software development kit (SDK) for writing composition functions in Python. This function uses utilities from the SDK.

Tip
Read the Python Function SDK documentation.
Important

The Python SDK automatically generates the RunFunctionRequest and RunFunctionResponse Python objects from a Protocol Buffers schema. You can see the schema in the Buf Schema Registry.

The fields of the generated Python objects behave similarly to builtin Python types like dictionaries and lists. Be aware that there are some differences.

Notably, you access the map of observed and desired resources like a dictionary but you can’t add a new desired resource by assigning to a map key. Instead, access and mutate the map key as if it already exists.

Instead of adding a new resource like this:

1resource = {"apiVersion": "example.org/v1", "kind": "Composed", ...}
2rsp.desired.resources["new-resource"] = fnv1beta1.Resource(resource=resource)

Pretend it already exists and mutate it, like this:

1resource = {"apiVersion": "example.org/v1", "kind": "Composed", ...}
2rsp.desired.resources["new-resource"].resource.update(resource)

Refer to the Protocol Buffers Python Generated Code Guide for further details.

Test the function end-to-end

Test your function by adding unit tests, and by using the crossplane beta render command.

When you initialize a function from the template it adds some unit tests to tests/test_fn.py. These tests use the unittest module from the Python standard library.

To add test cases, update the cases list in test_run_function. Expand the below block to view the full tests/test_fn.py file for the function.

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

Run the unit tests using hatch run:

1hatch run test:unit
2.
3----------------------------------------------------------------------
4Ran 1 test in 0.003s
5
6OK
Tip
Hatch is a Python build tool. It builds Python artifacts like wheels. It also manages virtual environments, similar to virtualenv or venv. The hatch run command creates a virtual environment and runs a command in that environment.

You can preview the output of a Composition that uses this function using the Crossplane CLI. You don’t need a Crossplane control plane to do this.

Create a directory under function-xbuckets named example and create Composite Resource, Composition and Function YAML files.

Expand the following block to see example files.

You can recreate the output below using by running crossplane beta render with these files.

The xr.yaml file contains the composite resource to render:

 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

The composition.yaml file contains the Composition to use to render the composite resource:

 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

The functions.yaml file contains the Functions the Composition references in its pipeline steps:

 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

The Function in functions.yaml uses the Development runtime. This tells crossplane beta render that your function is running locally. It connects to your locally running function instead of using Docker to pull and run the function.

1apiVersion: pkg.crossplane.io/v1beta1
2kind: Function
3metadata:
4  name: function-xbuckets
5  annotations:
6    render.crossplane.io/runtime: Development

Use hatch run development to run your function locally.

1hatch run development
Warning
hatch run development runs the function without encryption or authentication. Only use it during testing and development.

In a separate terminal, run crossplane beta render.

1crossplane beta render xr.yaml composition.yaml functions.yaml

This command calls your function. In the terminal where your function is running you should now see log output:

1hatch run development
22024-01-11T22:12:58.153572Z [info     ] Running function               filename=fn.py lineno=22 tag=
32024-01-11T22:12:58.153792Z [info     ] Added desired buckets          count=3 filename=fn.py lineno=68 region=us-east-2 tag=

The crossplane beta render command prints the desired resources the function returns.

 1---
 2apiVersion: example.crossplane.io/v1
 3kind: XBuckets
 4metadata:
 5  name: example-buckets
 6---
 7apiVersion: s3.aws.upbound.io/v1beta1
 8kind: Bucket
 9metadata:
10  annotations:
11    crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-b
12    crossplane.io/external-name: crossplane-functions-example-b
13  generateName: example-buckets-
14  labels:
15    crossplane.io/composite: example-buckets
16  ownerReferences:
17    # Omitted for brevity
18spec:
19  forProvider:
20    region: us-east-2
21---
22apiVersion: s3.aws.upbound.io/v1beta1
23kind: Bucket
24metadata:
25  annotations:
26    crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-c
27    crossplane.io/external-name: crossplane-functions-example-c
28  generateName: example-buckets-
29  labels:
30    crossplane.io/composite: example-buckets
31  ownerReferences:
32    # Omitted for brevity
33spec:
34  forProvider:
35    region: us-east-2
36---
37apiVersion: s3.aws.upbound.io/v1beta1
38kind: Bucket
39metadata:
40  annotations:
41    crossplane.io/composition-resource-name: xbuckets-crossplane-functions-example-a
42    crossplane.io/external-name: crossplane-functions-example-a
43  generateName: example-buckets-
44  labels:
45    crossplane.io/composite: example-buckets
46  ownerReferences:
47    # Omitted for brevity
48spec:
49  forProvider:
50    region: us-east-2
Tip
Read the composition functions documentation to learn more about testing composition functions.

Build and push the function to a package registry

You build a function in two stages. First you build the function’s runtime. This is the Open Container Initiative (OCI) image Crossplane uses to run your function. You then embed that runtime in a package, and push it to a package registry. The Crossplane CLI uses xpkg.upbound.io as its default package registry.

A function supports a single platform, like linux/amd64, by default. You can support multiple platforms by building a runtime and package for each platform, then pushing all the packages to a single tag in the registry.

Pushing your function to a registry allows you to use your function in a Crossplane control plane. See the [composition functions documentation]https://docs.crossplane.io/latest/concepts/composition-functions/. to learn how to use a function in a control plane.

Use Docker to build a runtime for each platform.

1docker build . --quiet --platform=linux/amd64 --tag runtime-amd64
2sha256:fdf40374cc6f0b46191499fbc1dbbb05ddb76aca854f69f2912e580cfe624b4b

1docker build . --quiet --platform=linux/arm64 --tag runtime-arm64
2sha256:cb015ceabf46d2a55ccaeebb11db5659a2fb5e93de36713364efcf6d699069af
Tip
You can use whatever tag you want. There’s no need to push the runtime images to a registry. The tag is only used to tell crossplane xpkg build what runtime to embed.
Important
Docker uses emulation to create images for different platforms. If building an image for a different platform fails, make sure you have installed binfmt. See the Docker documentation for instructions.

Use the Crossplane CLI to build a package for each platform. Each package embeds a runtime image.

The --package-root flag specifies the package directory, which contains crossplane.yaml. This includes metadata about the package.

The --embed-runtime-image flag specifies the runtime image tag built using Docker.

The --package-file flag specifies specifies where to write the package file to disk. Crossplane package files use the extension .xpkg.

1crossplane xpkg build \
2    --package-root=package \
3    --embed-runtime-image=runtime-amd64 \
4    --package-file=function-amd64.xpkg

1crossplane xpkg build \
2    --package-root=package \
3    --embed-runtime-image=runtime-arm64 \
4    --package-file=function-arm64.xpkg
Tip
Crossplane packages are special OCI images. Read more about packages in the packages documentation.

Push both package files to a registry. Pushing both files to one tag in the registry creates a multi-platform package that runs on both linux/arm64 and linux/amd64 hosts.

1crossplane xpkg push \
2  --package-files=function-amd64.xpkg,function-arm64.xpkg \
3  negz/function-xbuckets:v0.1.0
Tip

If you push the function to a GitHub repository the template automatically sets up continuous integration (CI) using GitHub Actions. The CI workflow will lint, test, and build your function. You can see how the template configures CI by reading .github/workflows/ci.yaml.

The CI workflow can automatically push packages to xpkg.upbound.io. For this to work you must create a repository at https://marketplace.upbound.io. Give the CI workflow access to push to the Marketplace by creating an API token and adding it to your repository. Save your API token access ID as a secret named XPKG_ACCESS_ID and your API token as a secret named XPKG_TOKEN.