Merge pull request #7754 from fstagni/90_mountCVMFS

test: mount cvmfs in pilot container

.github/workflows/integration.yml

@@ -49,6 +49,7 @@ jobs:
     - uses: actions/setup-python@v5
       with:
         python-version: '3.11'
+    - uses: cvmfs-contrib/github-action-cvmfs@v4
     - name: Installing dependencies
       run: |
         python -m pip install \

dirac.cfg

@@ -582,6 +582,11 @@ Systems
         index_name=a_different_name
       }
     }
+    JobWrapper
+    {
+      # Minimum output buffer requested for running jobs
+      MinOutputDataBufferGB = 5
+    }
   }
 }
 Resources

integration_tests.py

@@ -779,6 +779,12 @@ def _make_env(flags):
     env["MYSQL_VER"] = flags.pop("MYSQL_VER", DEFAULT_MYSQL_VER)
     env["ES_VER"] = flags.pop("ES_VER", DEFAULT_ES_VER)
     env["IAM_VER"] = flags.pop("IAM_VER", DEFAULT_IAM_VER)
+    if Path("/cvmfs").is_dir():
+        env["CVMFS_DIR"] = "/cvmfs"
+    else:
+        # create a directory in tmp
+        with tempfile.TemporaryDirectory() as tmpdir:
+            env["CVMFS_DIR"] = tmpdir
     return env
 
 

src/DIRAC/WorkloadManagementSystem/Client/DownloadInputData.py

@@ -5,7 +5,8 @@
 import random
 import tempfile
 
-from DIRAC import S_ERROR, S_OK, gLogger
+from DIRAC import S_ERROR, S_OK, gConfig, gLogger
+from DIRAC.ConfigurationSystem.Client.PathFinder import getSystemSection
 from DIRAC.Core.Utilities.Os import getDiskSpace
 from DIRAC.Core.Utilities.ReturnValues import returnSingleResult
 from DIRAC.DataManagementSystem.Utilities.DMSHelpers import DMSHelpers
@@ -227,9 +228,9 @@ def __checkDiskSpace(self, totalSize):
         """
         diskSpace = getDiskSpace(self.__getDownloadDir(False))  # MB
         availableBytes = diskSpace * 1024 * 1024  # bytes
-        # below can be a configuration option sent via the job wrapper in the future
-        # Moved from 3 to 5 GB (PhC 130822) for standard output file
-        bufferGBs = 5.0
+        bufferGBs = gConfig.getValue(
+            os.path.join(getSystemSection("WorkloadManagement/JobWrapper"), "JobWrapper", "MinOutputDataBufferGB"), 5.0
+        )
         data = bufferGBs * 1024 * 1024 * 1024  # bufferGBs in bytes
         if (data + totalSize) < availableBytes:
             msg = f"Enough disk space available ({availableBytes} bytes)"

src/DIRAC/WorkloadManagementSystem/Client/test/Test_Client_DownloadInputData.py

@@ -113,16 +113,20 @@ def test_DLIDownloadFromBestSE_Fail(dli, mockSE, osPathExists):
     assert not res["OK"]
 
 
-def test_DLI_execute(dli, mockSE):
+def test_DLI_execute(mocker, dli, mockSE):
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     dli._downloadFromSE = MagicMock(return_value=S_OK({"path": "/local/path/1.txt"}))
     res = dli.execute(dataToResolve=["/a/lfn/1.txt"])
     assert res["OK"]
     assert not res["Value"]["Failed"]
     assert "/a/lfn/1.txt" in res["Value"]["Successful"], res
 
 
-def test_DLI_execute_getFileMetadata_Fails(dli, mockSE):
+def test_DLI_execute_getFileMetadata_Fails(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {"Successful": {}, "Failed": {"/a/lfn/1.txt": "Error Getting MetaData"}}
@@ -137,8 +141,10 @@ def test_DLI_execute_getFileMetadata_Fails(dli, mockSE):
     assert res["Value"]["Successful"]["/a/lfn/1.txt"]["path"] == "/local/path/1.txt", res
 
 
-def test_DLI_execute_getFileMetadata_Lost(dli, mockSE):
+def test_DLI_execute_getFileMetadata_Lost(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {
@@ -161,8 +167,10 @@ def test_DLI_execute_getFileMetadata_Lost(dli, mockSE):
     assert res["Value"]["Successful"]["/a/lfn/1.txt"]["path"] == "/local/path/1.txt", res
 
 
-def test_DLI_execute_getFileMetadata_Unavailable(dli, mockSE):
+def test_DLI_execute_getFileMetadata_Unavailable(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {
@@ -185,8 +193,10 @@ def test_DLI_execute_getFileMetadata_Unavailable(dli, mockSE):
     assert res["Value"]["Successful"]["/a/lfn/1.txt"]["path"] == "/local/path/1.txt", res
 
 
-def test_DLI_execute_getFileMetadata_Cached(dli, mockSE):
+def test_DLI_execute_getFileMetadata_Cached(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {
@@ -209,8 +219,10 @@ def test_DLI_execute_getFileMetadata_Cached(dli, mockSE):
     assert res["Value"]["Successful"]["/a/lfn/1.txt"]["path"] == "/local/path/1.txt", res
 
 
-def test_DLI_execute_FirstDownFailed(dli, mockSE):
+def test_DLI_execute_FirstDownFailed(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {"Successful": {"/a/lfn/1.txt": {"Cached": 1, "Accessible": 0}}, "Failed": {}}
@@ -224,8 +236,10 @@ def test_DLI_execute_FirstDownFailed(dli, mockSE):
     assert res["Value"]["Successful"]["/a/lfn/1.txt"]["path"] == "/local/path/1.txt", res
 
 
-def test_DLI_execute_AllDownFailed(dli, mockSE):
+def test_DLI_execute_AllDownFailed(mocker, dli, mockSE):
     """When getFileMetadata fails for the first SE, we should fall back to the second."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     mockObjectSE = mockSE.return_value
     mockObjectSE.getFileMetadata.return_value = S_OK(
         {"Successful": {"/a/lfn/1.txt": {"Cached": 1, "Accessible": 0}}, "Failed": {}}
@@ -239,8 +253,10 @@ def test_DLI_execute_AllDownFailed(dli, mockSE):
     assert res["Value"]["Failed"][0] == "/a/lfn/1.txt", res
 
 
-def test_DLI_execute_NoLocal(dli, mockSE):
+def test_DLI_execute_NoLocal(mocker, dli, mockSE):
     """Data only at the remote SE."""
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.getSystemSection", return_value="pippo")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Client.DownloadInputData.gConfig.getValue", return_value=2)
     dli = DownloadInputData(
         {
             "InputData": [],

src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg

@@ -378,5 +378,6 @@ JobWrapper
   # Retry the upload of the output file if only one output SE is defined
   RetryUpload = False
   TapeSE = ['-tape', '-RDST', '-RAW']
+  MinOutputDataBufferGB = 5
 }
 ##END

tests/CI/docker-compose.yml

@@ -121,6 +121,8 @@ services:
     user: "${DIRAC_UID}:${DIRAC_GID}"
     depends_on:
       - dirac-server
+    volumes:
+      - ${CVMFS_DIR}:/cvmfs:ro
     ulimits:
       nofile: 8192
     pull_policy: always

tests/Jenkins/dirac-cfg-update-server.py

@@ -557,6 +557,8 @@
 # to avoid having to wait while testing rescheduling
 csAPI.setOption("Systems/WorkloadManagement/Production/Executors/Optimizers/JobScheduling/RescheduleDelays", "0")
 
+csAPI.createSection("Systems/WorkloadManagement/Production/JobWrapper/")
+csAPI.setOption("Systems/WorkloadManagement/Production/JobWrapper/MinOutputDataBufferGB", 1)
 
 # Final action: commit in CS
 res = csAPI.commit()