[BUG] Dynamic tasks fail if function name is too long

[kimoktm]

Describe the bug
Dynamic tasks workflow fails, raising malformed dynamic workflow error when the dynamic task function name is 52 characters or longer.

Raised error:

[User] malformed dynamic workflow, caused by: Collected Errors: 3
	Error 0: Code: NodeReferenceNotFound, Node Id: f1waozpy, Description: Referenced node [f1waozpy] not found.
	Error 1: Code: NodeReferenceNotFound, Node Id: fequ45eq, Description: Referenced node [fequ45eq] not found.
	Error 2: Code: NodeReferenceNotFound, Node Id: fzzt6uqq, Description: Referenced node [fzzt6uqq] not found.

Expected behavior
The dynamic workflow should work even if function names are longer than 52 characters

Additional context
To Reproduce
Here is a quick snippet to reproduce the issue:

@flytekit.task
def count_length(a: Dict[str, str]) -> FlyteFile:
    ctx = flytekit.current_context()
    working_dir = ctx.working_directory
    fname = os.path.join(working_dir, a["name"])
    fname = fname + ".json"
    ctx.logging.warning(f"Working dir is: {working_dir} Filename is: {fname}")
    with open(fname, "w") as fh:
        fh.write(f"{len(a['role'])}\n")

    return fname


@flytekit.workflow
def wf_demo(wf_in1: Dict[str, str]) -> FlyteFile:
    return count_length(a=wf_in1)


wf_default = flytekit.LaunchPlan.get_or_create(wf_demo)


@flytekit.dynamic
def abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz(data: str) -> List[FlyteFile]:
    ctx = flytekit.current_context()
    working_dir = ctx.working_directory
    file_path = os.path.join(working_dir, "file")

    with open(file_path, "w") as data_file:
        data_file.write(data.replace(".", "\n"))

    result_files: List[FlyteFile] = []
    with open(file_path) as f:
        reader = DictReader(f, delimiter=",")
        for row in reader:
            result_file = wf_default(wf_in1=row)
            result_files.append(result_file)
    return result_files


if __name__ == "__main__":
    data = "name,role.alice,sender.bob,receiver.eve,eavesdropper"
    out = abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz(data=data)
    print(out)

Renaming the dynamic task from abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz to abc or anything shorter than 52 characters will lead to a successful execution.

Screenshots

[welcome]

Thank you for opening your first issue here! 🛠

[kumare3]

@kimoktm thank you for the issue, we will definitely look into this soon, but as a work-around if you name the task something smaller does it work?
also cc @katrogan / @EngHabu

[kimoktm]

@kimoktm thank you for the issue, we will definitely look into this soon, but as a work-around if you name the task something smaller does it work?
also cc @katrogan / @EngHabu

Thanks for looking into it @kumare3!
Yes, renaming the task a shorter name works fine, so I'm not currently blocked

[pingsutw]

I found that this bug also occurs in "@task"

Traceback (most recent call last):
  File "/opt/venv/bin/pyflyte-execute", line 8, in <module>
    sys.exit(execute_task_cmd())
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 486, in execute_task_cmd
    _execute_task(
  File "/opt/venv/lib/python3.8/site-packages/flytekit/common/exceptions/scopes.py", line 160, in system_entry_point
    return wrapped(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 391, in _execute_task
    _task_def = resolver_obj.load_task(loader_args=resolver_args)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/core/python_auto_container.py", line 185, in load_task
    task_def = getattr(task_module, task_name)
AttributeError: module 'myapp.workflows.example' has no attribute 'say_hellooooooooooooooooooo'Traceback (most recent call last):
  File "/opt/venv/bin/pyflyte-execute", line 8, in <module>
    sys.exit(execute_task_cmd())
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/opt/venv/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 486, in execute_task_cmd
    _execute_task(
  File "/opt/venv/lib/python3.8/site-packages/flytekit/common/exceptions/scopes.py", line 160, in system_entry_point
    return wrapped(*args, **kwargs)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 391, in _execute_task
    _task_def = resolver_obj.load_task(loader_args=resolver_args)
  File "/opt/venv/lib/python3.8/site-packages/flytekit/core/python_auto_container.py", line 185, in load_task
    task_def = getattr(task_module, task_name)
AttributeError: module 'myapp.workflows.example' has no attribute 'say_hellooooooooooooooooooo'

Example:

import typing

from flytekit import task, workflow

@task
def say_hellooooooooooooooooooo(name: str) -> str:
    return f"hello world, {name}"

@workflow
def my_wf(name: str) -> str:
    return say_hellooooooooooooooooooo(name=name)

if __name__ == "__main__":
    print(f"Running my_wf(name='adam') {my_wf(name='adam')}")

[kumare3]

@eapolinario my recent fix in progress might fix this issue as well. flyteorg/flytekit#705