Repository: dask/dask-jobqueue
Pull request #571 (2022-08-09 08:41:59)

htcondor: add batch_name to match the name of the Dask worker
This makes it easier to identify the workers in the output of `condor_q`, `condor_history`, ... for debugging.

Before, the batch name was automatically set by HTCondor to the job id (or rather ClusterId in HTCondor language):
```
OWNER   BATCH_NAME      SUBMITTED   DONE   RUN    IDLE   HOLD  TOTAL JOB_IDS
jolange ID: 21671522   8/9  09:40      _      1      _      _      1 21671522.0
```

I think the name of the Dask worker is really useful to have here:
```
OWNER   BATCH_NAME           SUBMITTED   DONE   RUN    IDLE   HOLD  TOTAL JOB_IDS
jolange HTCondorCluster-1   8/9  10:17      _      1      _      _      1 21671526.0
jolange HTCondorCluster-0   8/9  10:17      _      1      _      _      1 21671527.0
```

So if you identify a worker (e.g. with problems) with `cluster.workers` or using the dashboard, it is much easier to find the right job in the batch system.

jolange: On a related note, it would be nice to also have the name in the worker's output (to stderr), so you can identify it in the log files:
```
2022-08-08 13:40:37,987 - distributed.nanny - INFO -         Start Nanny at: 'tcp://172.18.0.3:44283'
2022-08-08 13:40:39,274 - distributed.worker - INFO -       Start worker at:     tcp://172.18.0.3:42825
2022-08-08 13:40:39,275 - distributed.worker - INFO -          Listening to:     tcp://172.18.0.3:42825
2022-08-08 13:40:39,275 - distributed.worker - INFO -          dashboard at:           172.18.0.3:38721
2022-08-08 13:40:39,276 - distributed.worker - INFO - Waiting to connect to:     tcp://172.18.0.5:36311
2022-08-08 13:40:39,276 - distributed.worker - INFO - -------------------------------------------------
2022-08-08 13:40:39,277 - distributed.worker - INFO -               Threads:                          1
2022-08-08 13:40:39,277 - distributed.worker - INFO -                Memory:                 100.00 MiB
2022-08-08 13:40:39,278 - distributed.worker - INFO -       Local Directory: /var/lib/condor/execute/dir_85/dask-worker-space/worker-wfg9pxc7
2022-08-08 13:40:39,278 - distributed.worker - INFO - -------------------------------------------------
2022-08-08 13:40:39,285 - distributed.worker - INFO -         Registered to:     tcp://172.18.0.5:36311
2022-08-08 13:40:39,285 - distributed.worker - INFO - -------------------------------------------------
2022-08-08 13:40:39,286 - distributed.core - INFO - Starting established connection
```
I think I might propose to add it to `distributed` [here](https://github.com/dask/distributed/blob/bf537608e13e217fca7afdcab554d07be535a051/distributed/worker.py#L1371-L1383).
```
2022-08-08 13:40:37,987 - distributed.nanny - INFO -         Start Nanny at: 'tcp://172.18.0.3:44283'
2022-08-08 13:40:39,274 - distributed.worker - INFO -       Start worker at:     tcp://172.18.0.3:42825
2022-08-08 13:40:39,274 - distributed.worker - INFO -           Worker name:          HTCondorCluster-0
```

jolange: Sure, I just added this!
guillaumeeb: On of the security test is still a bit random, any way, it's safe to merge, thanks @jolange!

Patch:
diff --git a/dask_jobqueue/htcondor.py b/dask_jobqueue/htcondor.py
index 2d25c27..77ded3d 100644
--- a/dask_jobqueue/htcondor.py
+++ b/dask_jobqueue/htcondor.py
@@ -75,6 +75,7 @@ Queue
 
         self.job_header_dict = {
             "MY.DaskWorkerName": '"htcondor--$F(MY.JobId)--"',
+            "batch_name": self.name,
             "RequestCpus": "MY.DaskWorkerCores",
             "RequestMemory": "floor(MY.DaskWorkerMemory / 1048576)",
             "RequestDisk": "floor(MY.DaskWorkerDisk / 1024)",


Test patch:
diff --git a/dask_jobqueue/tests/test_htcondor.py b/dask_jobqueue/tests/test_htcondor.py
index b93b3ba..2147e21 100644
--- a/dask_jobqueue/tests/test_htcondor.py
+++ b/dask_jobqueue/tests/test_htcondor.py
@@ -38,6 +38,7 @@ def test_job_script():
         cancel_command_extra=["-forcex"],
     ) as cluster:
         job_script = cluster.job_script()
+        assert "batch_name = dummy-name" in job_script
         assert "RequestCpus = MY.DaskWorkerCores" in job_script
         assert "RequestDisk = floor(MY.DaskWorkerDisk / 1024)" in job_script
         assert "RequestMemory = floor(MY.DaskWorkerMemory / 1048576)" in job_script
@@ -78,10 +79,13 @@ def test_basic(loop):
             future = client.submit(lambda x: x + 1, 10)
             assert future.result(QUEUE_WAIT) == 11
 
+            condor_q = Job._call(["condor_q", "-batch"]).strip()
+
             workers = list(client.scheduler_info()["workers"].values())
             w = workers[0]
             assert w["memory_limit"] == 500 * 1024**2
             assert w["nthreads"] == 1
+            assert w["name"] in condor_q
 
             cluster.scale(0)