Add KuberenetesWorker.submit for ad-hoc submission to a Kubernetes work pool

[desertaxle]

This PR adds an experimental .submit method to the KubernetesWorker class to allow a flow to be submitted to a Kubernetes cluster. The updates to the work pool API haven't been made yet, so the steps for bundle upload and execution are loaded via an environment variable.

Here's an example:

import asyncio
import json
import os

from prefect_kubernetes import KubernetesWorker

from prefect import flow

UPLOAD_STEP = {
    "prefect_aws.experimental.bundles.upload": {
        "requires": "prefect-aws==0.5.5",
        "bucket": "storage-blocks-test-bucket",
        "aws_credentials_block_name": "my-creds",
    }
}

EXECUTE_STEP = {
    "prefect_aws.experimental.bundles.execute": {
        "requires": "prefect-aws==0.5.5",
        "bucket": "storage-blocks-test-bucket",
        "aws_credentials_block_name": "my-creds",
    }
}


@flow(log_prints=True, result_storage="s3-bucket/results-bucket")
def my_flow():
    print("Returning greeting...")
    return "Hello, world!"


async def main():
    async with KubernetesWorker(work_pool_name="olympic") as worker:
        future = await worker.submit(flow=my_flow)

    print(future.result()) # prints "Hello, world!"


if __name__ == "__main__":
    os.environ["PREFECT__BUNDLE_UPLOAD_STEP"] = json.dumps(UPLOAD_STEP)
    os.environ["PREFECT__BUNDLE_EXECUTE_STEP"] = json.dumps(EXECUTE_STEP)
    asyncio.run(main())

Related to #17194

[codspeed-hq]

CodSpeed Performance Report

Merging #17218 will not alter performance

_{Comparing k8s-worker-submit (d01c250) with main (fdc2b8a)}

Summary

✅ 2 untouched benchmarks