apiVersion: batch/v1
kind: Job
metadata:
  name: prefect-dask-job
  labels:
    app: prefect-dask-job
spec:
  template:
    metadata:
      labels:
        app: prefect-dask-job
    spec:
      containers:
        - name: flow
          image: prefecthq/prefect:latest
          imagePullPolicy: IfNotPresent
          command: ["/bin/sh", "-c"]
          args:
            [
              'python -c "import prefect; prefect.environments.DaskKubernetesEnvironment().run_flow()"',
            ]
          env:
            - name: PREFECT__CLOUD__GRAPHQL
              value: PREFECT__CLOUD__GRAPHQL
            - name: PREFECT__CLOUD__AUTH_TOKEN
              value: PREFECT__CLOUD__AUTH_TOKEN
            - name: PREFECT__CONTEXT__FLOW_RUN_ID
              value: PREFECT__CONTEXT__FLOW_RUN_ID
            - name: PREFECT__CONTEXT__NAMESPACE
              value: PREFECT__CONTEXT__NAMESPACE
            - name: PREFECT__CONTEXT__IMAGE
              value: PREFECT__CONTEXT__IMAGE
            - name: PREFECT__CLOUD__USE_LOCAL_SECRETS
              value: "false"
            - name: PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS
              value: "prefect.engine.cloud.CloudFlowRunner"
            - name: PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS
              value: "prefect.engine.cloud.CloudTaskRunner"
            - name: PREFECT__ENGINE__EXECUTOR__DEFAULT_CLASS
              value: "prefect.engine.executors.DaskExecutor"
            - name: PREFECT__LOGGING__LOG_TO_CLOUD
              value: "true"
            - name: PREFECT__LOGGING__LEVEL
              value: "DEBUG"
            - name: PREFECT__DEBUG
              value: "true"
            - name: DASK_DISTRIBUTED__SCHEDULER__WORK_STEALING
              value: "False"
            - name: DASK_DISTRIBUTED__SCHEDULER__BLOCKED_HANDLERS
              value: "['feed', 'run_function']"
            - name: PREFECT__LOGGING__EXTRA_LOGGERS
              value: PREFECT__LOGGING__EXTRA_LOGGERS
          resources:
            requests:
              cpu: "100m"
            limits:
              cpu: "100m"
      restartPolicy: Never
