FROM prefecthq/prefect:latest
RUN apt update && apt install -y openssh-client

Failed to load and execute Flow's environment: HangupException('Host key verification failed.\r')

RUN mkdir ~/.ssh && ssh-keyscan -t rsa <http://github.com|github.com> >> ~/.ssh/known_hosts

job_template

job_template = '''
apiVersion: batch/v1
kind: Job
spec:
  template:
    spec:
      containers:
        - name: flow
          volumeMounts:
            - name: ssh-key
              readOnly: true
              mountPath: "/root/.ssh"
      volumes:
        - name: ssh-key
          secret:
            secretName: prefect-ssh-key
            optional: false
            defaultMode: 0600
'''

volumeMounts:
    - mountPath: /root/.ssh
      name: ssh-key
      readOnly: true
  volumes:
  - name: ssh-key
    secret:
      defaultMode: 384
      optional: false
      secretName: prefect-ssh-key

defaultMode

Failed to add the RSA host key for IP address '140.82.114.4' to the list of known hosts

/bin/sh: prefect: command not found

pip install prefect[aws]

prefect

state_handler

animal=cat

run-cat

LocalRun

LocalExecutor

sys.path.append

ModuleNotFound

MainProject/
|-coreLogicModule/
|-PrefectStack/
    flows/    -need access to coreLogic
    runflows.py

08:01:52.893 | Submitting flow run 'f37fba9f-4280-431d-9f43-889e53192f24'
08:01:52.894 | Completed submission of flow run 'f37fba9f-4280-431d-9f43-889e53192f24'
08:01:52.901 | Finished monitoring for late runs.
08:01:54.187 | Flow run 'f37fba9f-4280-431d-9f43-889e53192f24' exited with exception: KeyError('__main__')

helm install prefectprod prefecthq/prefect-server -n prod --set agent.enabled=true --set postgresql.postgresqlUsername=admin%40my-server --set postgresql.postgresqlPassword=p%40ssword123 --set postgresql.externalHostname=<http://my-server.postgres.databases.azure.net|my-server.postgres.databases.azure.net> --set postgresql.useSubChart=false

-p 12001:12001

host_config

DockerRun

0.15.7

0.14.22

Completed deployment of flow run

flow_run_id

flow_run_id

wait_for

FlowRunView

raise KeyError(
KeyError: 'Task slug RunSimulationTask-1 is not found in the current Flow. This is usually caused by a mismatch between the flow version stored in the Prefect backend and the flow that was loaded from storage.\n- Did you change the flow without re-registering it?\n- Did you register the flow without updating it in your storage location (if applicable)?'

simulation_flow.py
    SimulationFlow = Flow(
       "RunSimulationFlowDemo-3",
        storage=Local(stored_as_script=True, path=__file__),
     )
----
simulation_task.py
class RunSimulationTask(Task):
    def __init__(
        self,
        simulation_run_parameters: SimulationRunParameters,
        simulation_gateways: SimulationGateways,
        factory_scheme_gateways: FactorySchemeGateways,
        map_gtw: MAPDataGatewayBase = None,
        **kwargs,
    ):
        super(RunSimulationTask, self).__init__(name="RunSimulationTask")
        self.simulation_run_parameters = simulation_run_parameters.to_dict()
        self.simulation_gateways = simulation_gateways
        self.factory_scheme_gateways = factory_scheme_gateways
        self.map_gtw = map_gtw or MAPDataGateway()

    def run(self) -> str:
        flow_run_id = prefect.context.get("flow_run_id")
        <http://self.logger.info|self.logger.info>(f"flow_run_id: {flow_run_id}")

        simulation_controller = SimulationController(
            **self.factory_scheme_gateways.to_dict(),
            **self.simulation_gateways.to_dict(),
            map_gtw=self.map_gtw,
        )

        <http://self.logger.info|self.logger.info>(f"Run simulation task. [{self.simulation_run_parameters}]")
        simulation_controller.run(
            **self.simulation_run_parameters, flow_run_id=flow_run_id
        )

        return flow_run_id

----
simulation_manager.py

        task = RunSimulationTask(
            simulation_run_parameters=simulation_run_parameters,
            simulation_gateways=self._simulation_gateways,
            factory_scheme_gateways=self._factory_scheme_gateways,
            map_gtw=self._map_gtw,
        )
        SimulationFlow.add_task(task)

        SimulationFlow.run_config = self.flow_config.run_config
        SimulationFlow.executor = self.flow_config.executor
        SimulationFlow.state_handlers = self.flow_config.state_handlers

client.register()

client.create_flow_run()

~/.prefect/flows

mapped

"task": [
      {      
        "id": "ad65f438-062b-4f9d-ba8e-826edce68d74",
        "name": "<task_name>",
        "mapped": false,
        "task_runs": [
          {
            "name": null,
            "created": "2021-11-11T01:08:13.326713+00:00",
            "map_index": -1,
            "version": 2
          },
          {
            "name": "<task_run_0>",
            "created": "2021-11-11T01:31:55.062178+00:00",
            "map_index": 0,
            "version": 3
          },
          {
            "name": "<task_run_1>",
            "created": "2021-11-11T01:31:55.062178+00:00",
            "map_index": 1,
            "version": 3
          }
        ]
      },

map_index != -1

mapped

task_runs

Failed: "An error occurred (ThrottlingException) when calling the DeregisterTaskDefinition operation (reached max retries: 2): Rate exceeded"