[BUG]: DFP DUO integrated training pipeline fails with dask error in 24.10

[dagardner-nv]

Version

24.10

Which installation method(s) does this occur on?

Source

Describe the bug.

Creating dask cluster...
Creating dask cluster... Done. Dashboard: http://192.168.4.51:8787/status
S3 objects to DF complete. Rows: 86, Cache: miss, Duration: 791.4469242095947 ms, Rate: 108.66174012349194 rows/s
Stopping dask cluster...
Stopping dask cluster... Done.
Failed to download logs. Error: 
Traceback (most recent call last):
  File "/home/dagardner/work/m2/python/morpheus/morpheus/controllers/file_to_df_controller.py", line 166, in _get_or_create_dataframe_from_batch
    dfs = self._downloader.download(download_buckets, download_method_func)
  File "/home/dagardner/work/m2/python/morpheus/morpheus/utils/downloader.py", line 160, in download
    with self.get_dask_client() as dist:
  File "/home/dagardner/work/m2/python/morpheus/morpheus/utils/downloader.py", line 125, in get_dask_client
    return dask.distributed.Client(self.get_dask_cluster())
  File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/distributed/client.py", line 914, in __init__
    raise RuntimeError(
RuntimeError: Trying to connect to an already closed or closing Cluster LocalCluster(8ada266f, 'inproc://192.168.4.51/156737/1', workers=0, threads=0, memory=0 B).

Minimum reproducible example

Terminal 1:

cd examples/digital_fingerprinting/production
docker compose build
docker compose up mlflow

Terminal 2:

cd examples/digital_fingerprinting/production/morpheus
python dfp_integrated_training_batch_pipeline.py --tracking_uri="http://localhost:5000" \
        --log_level DEBUG \
        --silence_monitors \
        --source duo \
        --start_time "2022-08-01" \
        --duration "60d" \
        --train_users generic \
        --input_file "./control_messages/duo_payload_training.json"

Relevant log output

Click here to see error details

Creating dask cluster...

Creating dask cluster... Done. Dashboard: http://192.168.4.51:8787/status

S3 objects to DF complete. Rows: 86, Cache: miss, Duration: 791.4469242095947 ms, Rate: 108.66174012349194 rows/s

Stopping dask cluster...

Stopping dask cluster... Done.

Failed to download logs. Error:

Traceback (most recent call last):

File "/home/dagardner/work/m2/python/morpheus/morpheus/controllers/file_to_df_controller.py", line 166, in _get_or_create_dataframe_from_batch

dfs = self._downloader.download(download_buckets, download_method_func)

File "/home/dagardner/work/m2/python/morpheus/morpheus/utils/downloader.py", line 160, in download

with self.get_dask_client() as dist:

File "/home/dagardner/work/m2/python/morpheus/morpheus/utils/downloader.py", line 125, in get_dask_client

return dask.distributed.Client(self.get_dask_cluster())

File "/home/dagardner/work/conda/envs/m2/lib/python3.10/site-packages/distributed/client.py", line 914, in init

raise RuntimeError(

RuntimeError: Trying to connect to an already closed or closing Cluster LocalCluster(8ada266f, 'inproc://192.168.4.51/156737/1', workers=0, threads=0, memory=0 B).

Error while converting S3 buckets to DF.

Full env printout

Click here to see environment details

[Paste the results of print_env.sh here, it will be hidden by default]

Other/Misc.

No response

Code of Conduct

• I agree to follow Morpheus' Code of Conduct
• I have searched the open bugs and have found no duplicates for this bug report