https://dagster.io/ logo
#ask-community
Title
# ask-community
d

Drew You

04/10/2023, 5:43 PM
Copy code
I'm trying to use an EcsRunLauncher and QueuedRunCoordinator but I keep getting this error:
Copy code
dagster._check.ParameterCheckError: Param "image" is not a str. Got None which is type <class 'NoneType'>.

  File "/usr/local/lib/python3.11/site-packages/dagster/_daemon/run_coordinator/queued_run_coordinator_daemon.py", line 333, in _dequeue_run
    instance.run_launcher.launch_run(LaunchRunContext(dagster_run=run, workspace=workspace))
  File "/usr/local/lib/python3.11/site-packages/dagster_aws/ecs/launcher.py", line 370, in launch_run
    run_task_kwargs = self._run_task_kwargs(run, image, container_context)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster_aws/ecs/launcher.py", line 570, in _run_task_kwargs
    task_definition_config = DagsterEcsTaskDefinitionConfig.from_task_definition_dict(
                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster_aws/ecs/tasks.py", line 126, in from_task_definition_dict
    return DagsterEcsTaskDefinitionConfig(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster_aws/ecs/tasks.py", line 55, in __new__
    check.str_param(image, "image"),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_check/__init__.py", line 1347, in str_param
    raise _param_type_mismatch_exception(obj, str, param_name, additional_message)
Copy code
# ECS launcher
run_launcher:
  module: dagster_aws.ecs
  class: EcsRunLauncher
  config:
    container_name: "syncdag_user_code"
    include_sidecars: true
    env_vars:
      - DAGSTER_POSTGRES_USER
      - DAGSTER_POSTGRES_PASSWORD
      - DAGSTER_POSTGRES_DB
      - PG_DB_CONN_STRING
      - SYNCRETIC_ENVIRONMENT

# Custom values via env
run_coordinator:
  module: dagster.core.run_coordinator
  class: QueuedRunCoordinator
  config:
    max_concurrent_runs: 25
    tag_concurrency_limits:
      - key: "database"
        value: "redshift"
        limit: 4
      - key: "dagster/backfill"
        limit: 10
I've tried with and without the
container_name
param, but
syncdag_user_code
is the name of the container in the ecs task definition.
d

daniel

04/10/2023, 11:39 PM
Hi drew - what image / task definition are you hoping that it use to launch the run in? Usually we recommend you run a separate Ecs task with a grpc server on it that has the DAGSTER_CURRENT_IMAGE env var set to the image to use - the daemon then uses that image to launch the run
d

Drew You

04/11/2023, 2:09 PM
Here's my current set of task definitions:
Copy code
[{
        "image": "${local.dagster_image_name}",
        "name": "${local.dagit_container_name}",
        "command": [
            "dagit","-h", "0.0.0.0", "-p","8080"
        ],
        "environment": [
          ${join(",\n", formatlist("{\"name\":\"%s\",\"value\":\"%s\"}", keys(local.dagster_variables), values(local.dagster_variables)))}
        ],
        "logConfiguration": {
          "logDriver": "awslogs",
          "options": {
            "awslogs-group": "${aws_cloudwatch_log_group.dagster.name}",
            "awslogs-region": "${var.aws_region}",
            "awslogs-stream-prefix": "dagster"
          }
        },
        "essential": true,
        "portMappings": [
            {
                "containerPort": 8080,
                "hostPort": 8080
            }
        ]
      },
      {
        "image": "${local.dagster_image_name}",
        "name": "${local.dagster_daemon_container_name}",
        "command": [
            "dagster-daemon","run"
        ],
        "environment": [
          ${join(",\n", formatlist("{\"name\":\"%s\",\"value\":\"%s\"}", keys(local.dagster_daemon_variables), values(local.dagster_daemon_variables)))}
        ],
        "logConfiguration": {
          "logDriver": "awslogs",
          "options": {
            "awslogs-group": "${aws_cloudwatch_log_group.dagster.name}",
            "awslogs-region": "${var.aws_region}",
            "awslogs-stream-prefix": "dagster"
          }
        },
        "essential": true
      },
  {
        "image": "${local.user_container_image_name}",
        "name": "${local.user_container_name}",
        "environment": [
          ${join(",\n", formatlist("{\"name\":\"%s\",\"value\":\"%s\"}", keys(local.dagster_variables), values(local.dagster_variables)))}
        ],
        "logConfiguration": {
          "logDriver": "awslogs",
          "options": {
            "awslogs-group": "${aws_cloudwatch_log_group.dagster.name}",
            "awslogs-region": "${var.aws_region}",
            "awslogs-stream-prefix": "dagster"
          }
        },
        "essential": true,
        "mountPoints": [
          {
            "sourceVolume": "${local.dagster_mounted_volume_name}",
            "containerPath": "/opt/dagster"
          }
        ],
        "portMappings": [
            {
                "containerPort": 8008,
                "hostPort": 8008
            }
        ]
      }]
but the high level is that I've got 3 containers,
user_code
,
dagit
, and
dagster-daemon
all on the same task definition
if I split
user_code
into its own task definition, how do you specify the
grpc
ip in
dagster.yml
?
d

daniel

04/11/2023, 2:19 PM
If your user_code task definition sets the DAGSTER_CURRENT_IMAGE like in the deploy_ecs example here: https://github.com/dagster-io/dagster/blob/master/examples/deploy_ecs/docker-compose.yml#L110 - i would expect it to pick up the image correctly
er sorry, your user_code container, not task definition
d

Drew You

04/11/2023, 4:00 PM
hmm, so it seems like what is happening is that dagster is running a fresh copy of that image and not copying over any of the info in the task definition:
Copy code
{
        "image": "${local.user_container_image_name}",
        "name": "${local.user_container_name}",
        "environment": [
          ${join(",\n", formatlist("{\"name\":\"%s\",\"value\":\"%s\"}", keys(local.dagster_variables), values(local.dagster_variables)))}
        ],
        "logConfiguration": {
          "logDriver": "awslogs",
          "options": {
            "awslogs-group": "${aws_cloudwatch_log_group.dagster.name}",
            "awslogs-region": "${var.aws_region}",
            "awslogs-stream-prefix": "dagster"
          }
        },
        "essential": true,
        "mountPoints": [
          {
            "sourceVolume": "${local.dagster_mounted_volume_name}",
            "containerPath": "/opt/dagster"
          }
        ],
        "portMappings": [
            {
                "containerPort": 8008,
                "hostPort": 8008
            }
        ]
      }
d

daniel

04/11/2023, 4:00 PM
What information are you hoping that it will copy over?
d

Drew You

04/11/2023, 4:00 PM
i.e. the
"mountPoints"
and the
"environment"
keys
I have an efs drive mounted to the
user_code
container that has the repo in it, and that seems to not survive the ecs runtask
d

daniel

04/11/2023, 4:12 PM
those will be copied over from a container with the same name as the "container_name" field on the run launcher
d

Drew You

04/11/2023, 4:40 PM
hmm, so I'm sure that it's not doing that. how should I go about debugging this?
d

daniel

04/11/2023, 4:42 PM
Can you pass along your latest dagster.yaml (with container_name set) and the task definition that's being used to launch the run?
oh, you know what, you're right, I gave you some bad information there. It's copying the config over from the container that was used to launch the run, not the one set in container_name
Here's the relevant code that constructs the task definition to use from the current task: https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-aws/dagster_aws/ecs/tasks.py#L178-L212 Sorry for the confusion there
d

Drew You

04/11/2023, 6:50 PM
oh, it looks like you pushed a commit 2 hours ago that allows for mountPoints?
d

daniel

04/11/2023, 6:52 PM
I did, but it should work before that if you’re using the codepath where it pulls from the calling container
The thing I added is for the path where you tell it to construct its own task definition from configuration that you provide
d

Drew You

04/11/2023, 6:53 PM
I see. how does it decide which path to follow?
d

daniel

04/11/2023, 6:55 PM
If you supply a task_definition key to the run launcher, it will create one using that config - if not, it will use the task definition that launched the run
d

Drew You

04/11/2023, 7:02 PM
so, here are the env vars I have set:
Copy code
PG_DB_CONN_STRING : "postgresql://${var.rds_username}:${var.rds_password}@${aws_db_instance.dagster[0].address}:${aws_db_instance.dagster[0].port}/${aws_db_instance.dagster[0].name}",
    DAGSTER_POSTGRES_USER: var.rds_username,
    DAGSTER_POSTGRES_PASSWORD: var.rds_password,
    DAGSTER_POSTGRES_DB: aws_db_instance.dagster[0].name,
    DAGSTER_HOME : "${var.dagster-container-home}/",
    SYNCDAG_HOME : "/opt/dagster/syncdag/repository.py" #path to a not yet mounted drive
I think it may have something to do with the
CMD
arg in the
user_code
dockerfile:
Copy code
CMD env >> /etc/environment && exec poetry run dagster api grpc -h 0.0.0.0 -p 8008 -f $SYNCDAG_HOME
except
$SYNCDAG_HOME
does get set correctly
d

daniel

04/11/2023, 7:06 PM
Just confirming understanding - This is no longer about the run launcher now / is more of a general ecs question about how env vars can be accessed in the user code container?
d

Drew You

04/11/2023, 7:08 PM
I'm not sure. The problem is that env vars aren't making it to the runlauncher. Neither the
mountPoints
nor the
DAGSTER_
vars. When I set the
DAGSTER_
vars in the
Dockerfile
, it correctly passes them to the runlauncher and I get the
mountPoint
error. It feels like somehow the runlauncher is launching the docker container without the ecs task definition
(because the env vars I posted above are defined in the task definition and not the
Dockerfile
itself)
d

daniel

04/11/2023, 7:25 PM
Can you post the latest version of the dagster.yaml that you’re using? Are the mount points and env vars that you’re hoping to include present on the container that’s launching the run? (Likely the daemon container)
d

Drew You

04/11/2023, 7:30 PM
here's my task definition taken from the cluster on the aws console:
Copy code
{
    "taskDefinitionArn": "",
    "containerDefinitions": [
        {
            "name": "dagit",
            "image": "<http://dkr.ecr.us-east-1.amazonaws.com/build_ecs/dagster:devel|dkr.ecr.us-east-1.amazonaws.com/build_ecs/dagster:devel>",
            "cpu": 0,
            "portMappings": [
                {
                    "containerPort": 8080,
                    "hostPort": 8080,
                    "protocol": "tcp"
                }
            ],
            "essential": true,
            "command": [
                "dagit",
                "-h",
                "0.0.0.0",
                "-p",
                "8080"
            ],
            "environment": [
                {
                    "name": "DAGSTER_POSTGRES_USER",
                    "value": "user"
                },
                {
                    "name": "SYNCDAG_HOME",
                    "value": "/opt/dagster/syncdag/repository.py"
                },
                {
                    "name": "SYNCRETIC_ENVIRONMENT",
                    "value": "DEVELOPMENT"
                },
                {
                    "name": "DAGSTER_POSTGRES_DB",
                    "value": "rds"
                },
                {
                    "name": "DAGSTER_HOME",
                    "value": "/opt/dagster/dagster_home//"
                },
                {
                    "name": "PG_DB_CONN_STRING",
                    "value": "user:pass@url/rds"
                },
                {
                    "name": "DAGSTER_POSTGRES_PASSWORD",
                    "value": "password"
                }
            ],
            "mountPoints": [],
            "volumesFrom": [],
            
        {
            "name": "dagster_daemon",
            "image": "<http://dkr.ecr.us-east-1.amazonaws.com/build_ecs/dagster:devel|dkr.ecr.us-east-1.amazonaws.com/build_ecs/dagster:devel>",
            "cpu": 0,
            "portMappings": [],
            "essential": true,
            "command": [
                "dagster-daemon",
                "run"
            ],
            "environment": [
                {
                    "name": "PG_DB_CONN_STRING",
                    "value": "<postgresql://user:pass@url>:port/rds"
                },
            ],
            "mountPoints": [],
            "volumesFrom": [],
           
        {
            "name": "syncdag_user_code",
            "image": "<http://dkr.ecr.us-east-1.amazonaws.com/build_ecs/syncdag_user_code:devel|dkr.ecr.us-east-1.amazonaws.com/build_ecs/syncdag_user_code:devel>",
            "cpu": 0,
            "portMappings": [
                {
                    "containerPort": 8008,
                    "hostPort": 8008,
                    "protocol": "tcp"
                }
            ],
            "essential": true,
            "environment": [
                {
                    "name": "DAGSTER_POSTGRES_USER",
                    "value": "user"
                },
                {
                    "name": "SYNCDAG_HOME",
                    "value": "/opt/dagster/syncdag/repository.py"
                },
            
                {
                    "name": "DAGSTER_POSTGRES_DB",
                    "value": "rds"
                },
                {
                    "name": "DAGSTER_HOME",
                    "value": "/opt/dagster/dagster_home//"
                },
                {
                    "name": "PG_DB_CONN_STRING",
                    "value": "user:pass@url/rds"
                },
                {
                    "name": "DAGSTER_POSTGRES_PASSWORD",
                    "value": "password"
                }
            ],
            "mountPoints": [
                {
                    "sourceVolume": "dagster",
                    "containerPath": "/opt/dagster"
                }
            ],
            "volumesFrom": [],
            
    ],
    "revision": 43,
    "volumes": [
        {
            "name": "dagster",
            "efsVolumeConfiguration": {
                "fileSystemId": "fs-ID",
                "rootDirectory": "/",
                "transitEncryption": "ENABLED",
                "authorizationConfig": {
                    "accessPointId": "fsap-ID",
                    "iam": "ENABLED"
                }
            }
        }
    ],
    "status": "ACTIVE",
    "
}
specifically, I have been assuming that it used the
user_code
container. should I be setting the mount and vars on the
dagster_daemon
container?
d

daniel

04/11/2023, 7:30 PM
It sounds like you're expecting the mountPoints from the user code container to be passed through - it's reasonable for you to expect that, but that's not currently the container that it uses
yeah, it'll use whichever container launches the run
d

Drew You

04/11/2023, 7:31 PM
ahh I see
d

daniel

04/11/2023, 7:32 PM
(it's more common from what i've seen for the three containers you're using there to each have their own task)
d

Drew You

04/11/2023, 7:33 PM
I was thinking of doing that, but then
dagster.yaml
is static and I'd have to figure out a way to set it
when they're all in the same task, you can set dagster.yaml to
0.0.0.0
err the
grpc
d

daniel

04/11/2023, 7:34 PM
I see
d

Drew You

04/12/2023, 4:00 PM
dope, finally got it working. you were very helpful. another hint for anyone else doing this in the future -- you need to do
RUN poetry config virtualenvs.create false
before
RUN poetry install
because the spawned run won't be able to interact w/ poetry (if you're using
poetry
)
I am still somewhat randomly getting:
Copy code
FileNotFoundError: [Errno 2] No such file or directory: '/opt/dagster/syncdag/repository.py'

  File "/usr/local/lib/python3.11/site-packages/dagster/_grpc/impl.py", line 120, in core_execute_run
    recon_pipeline.get_definition()
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 238, in get_definition
    return self.repository.get_definition().get_maybe_subset_job_def(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 117, in get_definition
    return repository_def_from_pointer(self.pointer, self.repository_load_data)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 789, in repository_def_from_pointer
    target = def_from_pointer(pointer)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 679, in def_from_pointer
    target = pointer.load_target()
             ^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/code_pointer.py", line 175, in load_target
    module = load_python_file(self.python_file, self.working_directory)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/dagster/_core/code_pointer.py", line 72, in load_python_file
    os.stat(python_file)
this file is on the efs drive
so, I can materialize -- and 50% of the time, it completes successfully, 50% of the time I get this error
hmm, so it appears it isn't 50% of the time. Only my assets with
StaticPartition
s will materialize and the ones w/ no partitions fail w/ that error.
d

daniel

04/12/2023, 4:26 PM
Hm I’m having a lot of trouble understanding why that would matter - I would think all runs would need to be able to load that file
d

Drew You

04/12/2023, 4:42 PM
got it. those runs launch from the
dagit
container instead of the
daemon
or
user_code
containers.
if they're triggered from the ui
d

daniel

04/12/2023, 4:43 PM
Ohhh yes sorry that would do it! If you have the run queue enabled, then every run will launch from the daemon
❤️ 1
15 Views