argo workflow in starlark

common.bzl

metav1 = proto.package("k8s.io.apimachinery.pkg.apis.meta.v1")
wfv1 = proto.package("github.com.argoproj.argo.pkg.apis.workflow.v1alpha1")

def workflow(name = None):
    """ Common workflow metadata to be used in many workflows """
    if not name:
        name = "workflow"

    return wfv1.Workflow(
        metadata = metav1.ObjectMeta(
            namespace = "argo",
            generateName = "{}-".format(name),
            labels = {
                "workflows.argoproj.io/controller-instanceid": "argo",
            },
        ),
    )

nested-diamond.bzl

load("//common.bzl", "workflow")

corev1 = proto.package("k8s.io.api.core.v1")
metav1 = proto.package("k8s.io.apimachinery.pkg.apis.meta.v1")
wfv1 = proto.package("github.com.argoproj.argo.pkg.apis.workflow.v1alpha1")

def main(ctx):
    w = workflow(name = "nested-dag")

    w.spec = wfv1.WorkflowSpec(
        entrypoint = "diamond",
        ttlSecondsAfterFinished = 90,
        templates = [
            wfv1.Template(
                name = "echo",
                inputs = wfv1.Inputs(
                    parameters = [
                        wfv1.Parameter(name = "message"),
                    ],
                ),
                container = corev1.Container(
                    name = "alpine",
                    image = "alpine:2.7",
                    command = [
                        "echo",
                        "{{inputs.parameters.message}}",
                    ],
                ),
            ),
            wfv1.Template(
                name = "diamond",
                dag = wfv1.DAGTemplate(
                    tasks = [
                        wfv1.DAGTask(
                            name = "A",
                            template = "nested-diamond",
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "A"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "B",
                            template = "nested-diamond",
                            dependencies = ["A"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "B"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "C",
                            template = "nested-diamond",
                            dependencies = ["A"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "C"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "D",
                            template = "nested-diamond",
                            dependencies = ["B", "C"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "D"),
                                ],
                            ),
                        ),
                    ],
                ),
            ),
            wfv1.Template(
                name = "nested-diamond",
                inputs = wfv1.Inputs(
                    parameters = [
                        wfv1.Parameter(name = "message"),
                    ],
                ),
                dag = wfv1.DAGTemplate(
                    tasks = [
                        wfv1.DAGTask(
                            name = "A",
                            template = "echo",
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "{{inputs.parameters.message}}A"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "B",
                            template = "echo",
                            dependencies = ["A"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "{{inputs.parameters.message}}B"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "C",
                            template = "echo",
                            dependencies = ["A"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "{{inputs.parameters.message}}C"),
                                ],
                            ),
                        ),
                        wfv1.DAGTask(
                            name = "D",
                            template = "echo",
                            dependencies = ["B", "C"],
                            arguments = wfv1.Arguments(
                                parameters = [
                                    wfv1.Parameter(name = "message", value = "{{inputs.parameters.message}}D"),
                                ],
                            ),
                        ),
                    ],
                ),
            ),
        ],
    )

    return [w]