diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/PilotLoggingAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/PilotLoggingAgent.py
new file mode 100644
index 00000000000..5fda3320a24
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Agent/PilotLoggingAgent.py
@@ -0,0 +1,187 @@
+""" :mod: PilotLoggingAgent
+
+    PilotLoggingAgent sends Pilot log files to an SE.
+
+.. literalinclude:: ../ConfigTemplate.cfg
+  :start-after: ##BEGIN PilotLoggingAgent
+  :end-before: ##END
+  :dedent: 2
+  :caption: PilotLoggingAgent options
+"""
+
+# # imports
+import os
+import time
+from DIRAC import S_OK, S_ERROR, gConfig
+from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
+from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOs
+from DIRAC.Core.Base.AgentModule import AgentModule
+from DIRAC.Core.Utilities.Proxy import executeWithUserProxy
+from DIRAC.DataManagementSystem.Client.DataManager import DataManager
+from DIRAC.WorkloadManagementSystem.Client.TornadoPilotLoggingClient import TornadoPilotLoggingClient
+
+
+class PilotLoggingAgent(AgentModule):
+    """
+    .. class:: PilotLoggingAgent
+
+    The agent sends completed pilot log files to permanent storage for analysis.
+    """
+
+    def __init__(self, *args, **kwargs):
+        """c'tor"""
+        super().__init__(*args, **kwargs)
+
+    def initialize(self):
+        """
+        agent's initialisation. Use this agent's CS information to:
+        Determine what Defaults/Shifter shifter proxy to use.,
+        get the target SE name from the CS.
+        Obtain log file location from Tornado.
+
+        :param self: self reference
+        """
+        # pilot logs lifetime in days
+        self.clearPilotsDelay = self.am_getOption("ClearPilotsDelay", 30)
+        # configured VOs and setup
+        res = getVOs()
+        if not res["OK"]:
+            return res
+        self.voList = res.get("Value", [])
+
+        if isinstance(self.voList, str):
+            self.voList = [self.voList]
+
+        self.setup = gConfig.getValue("/DIRAC/Setup", None)
+
+        return S_OK()
+
+    def execute(self):
+        """
+        Execute one agent cycle. Upload log files to the SE and register them in the DFC.
+        Use a shifter proxy dynamically loaded for every VO
+
+        :param self: self reference
+        """
+        voRes = {}
+        for vo in self.voList:
+            self.opsHelper = Operations(vo=vo, setup=self.setup)
+            # is remote pilot logging enabled for the VO ?
+            pilotLogging = self.opsHelper.getValue("/Pilot/RemoteLogging", False)
+            if pilotLogging:
+                res = self.opsHelper.getOptionsDict("Shifter/DataManager")
+                if not res["OK"]:
+                    voRes[vo] = "No shifter defined - skipped"
+                    self.log.error(f"No shifter defined for VO: {vo} - skipping ...")
+                    continue
+
+                proxyUser = res["Value"].get("User")
+                proxyGroup = res["Value"].get("Group")
+                if proxyGroup is None or proxyUser is None:
+                    self.log.error(
+                        f"No proxy user or group defined for pilot: VO: {vo}, User: {proxyUser}, Group: {proxyGroup}"
+                    )
+                    voRes[vo] = "No proxy user or group defined - skipped"
+                    continue
+
+                self.log.info(f"Proxy used for pilot logging: VO: {vo}, User: {proxyUser}, Group: {proxyGroup}")
+                res = self.executeForVO(  # pylint: disable=unexpected-keyword-arg
+                    vo, proxyUserName=proxyUser, proxyUserGroup=proxyGroup
+                )
+                if not res["OK"]:
+                    voRes[vo] = res["Message"]
+        if voRes:
+            for key, value in voRes.items():
+                self.log.error(f"Error for {key} vo; message: {value}")
+            voRes.update(S_ERROR("Agent cycle for some VO finished with errors"))
+            return voRes
+        return S_OK()
+
+    @executeWithUserProxy
+    def executeForVO(self, vo):
+        """
+        Execute one agent cycle for a VO. It obtains VO-specific configuration pilot options from the CS:
+        UploadPath - the path where the VO wants to upload pilot logs. It has to start with a VO name (/vo/path).
+        UploadSE - Storage element where the logs will be kept.
+
+        :param str vo: vo enabled for remote pilot logging
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        self.log.info(f"Pilot files upload cycle started for VO: {vo}")
+        res = self.opsHelper.getOptionsDict("Pilot")
+        if not res["OK"]:
+            return S_ERROR(f"No pilot section for {vo} vo")
+        pilotOptions = res["Value"]
+        uploadSE = pilotOptions.get("UploadSE")
+        if uploadSE is None:
+            return S_ERROR("Upload SE not defined")
+        self.log.info(f"Pilot upload SE: {uploadSE}")
+
+        uploadPath = pilotOptions.get("UploadPath")
+        if uploadPath is None:
+            return S_ERROR(f"Upload path on SE {uploadSE} not defined")
+        self.log.info(f"Pilot upload path: {uploadPath}")
+
+        client = TornadoPilotLoggingClient(useCertificates=True)
+        resDict = client.getMetadata()
+
+        if not resDict["OK"]:
+            return resDict
+
+        # vo-specific source log path:
+        pilotLogPath = os.path.join(resDict["Value"]["LogPath"], vo)
+        # check for new files and upload them
+        if not os.path.exists(pilotLogPath):
+            # not a disaster, the VO is enabled, but no logfiles were ever stored.
+            return S_OK()
+        # delete old pilot log files for the vo VO
+        self.clearOldPilotLogs(pilotLogPath)
+
+        self.log.info(f"Pilot log files location = {pilotLogPath} for VO: {vo}")
+
+        # get finalised (.log) files from Tornado and upload them to the selected SE
+
+        files = [
+            f for f in os.listdir(pilotLogPath) if os.path.isfile(os.path.join(pilotLogPath, f)) and f.endswith("log")
+        ]
+
+        if not files:
+            self.log.info("No files to upload for this cycle")
+        for elem in files:
+            lfn = os.path.join(uploadPath, elem)
+            name = os.path.join(pilotLogPath, elem)
+            res = DataManager().putAndRegister(lfn=lfn, fileName=name, diracSE=uploadSE, overwrite=True)
+            if not res["OK"]:
+                self.log.error("Could not upload", f"to {uploadSE}: {res['Message']}")
+            else:
+                self.log.verbose("File uploaded: ", f"LFN = {res['Value']}")
+                try:
+                    os.remove(name)
+                except Exception as excp:
+                    self.log.exception("Cannot remove a local file after uploading", lException=excp)
+        return S_OK()
+
+    def clearOldPilotLogs(self, pilotLogPath):
+        """
+        Delete old pilot log files unconditionally. Assumes that pilotLogPath exists.
+
+        :param str pilotLogPath: log files directory
+        :return: None
+        :rtype: None
+        """
+
+        files = os.listdir(pilotLogPath)
+        seconds = int(self.clearPilotsDelay) * 86400
+        currentTime = time.time()
+
+        for file in files:
+            fullpath = os.path.join(pilotLogPath, file)
+            modifTime = os.stat(fullpath).st_mtime
+            if modifTime < currentTime - seconds:
+                self.log.debug(f" Deleting old log : {fullpath}")
+                try:
+                    os.remove(fullpath)
+                except Exception as excp:
+                    self.log.exception(f"Cannot remove an old log file after {fullpath}", lException=excp)
diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py b/src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py
index b8947afb4b1..279410393bc 100644
--- a/src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py
+++ b/src/DIRAC/WorkloadManagementSystem/Agent/SiteDirector.py
@@ -1017,10 +1017,6 @@ def _getPilotOptions(self, queue, **kwargs):
 
         pilotOptions.append("--pythonVersion=3")
 
-        # Debug
-        if self.pilotLogLevel.lower() == "debug":
-            pilotOptions.append("-ddd")
-
         # DIRAC Extensions to be used in pilots
         pilotExtensionsList = opsHelper.getValue("Pilot/Extensions", [])
         extensionsList = []
diff --git a/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PilotLoggingAgent.py b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PilotLoggingAgent.py
new file mode 100644
index 00000000000..2776a0f5042
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Agent/test/Test_Agent_PilotLoggingAgent.py
@@ -0,0 +1,190 @@
+""" Test class for PilotLoggingAgent Agent
+"""
+import os
+import time
+import tempfile
+
+import pytest
+from unittest.mock import MagicMock, patch
+
+# DIRAC Components
+import DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent as plaModule
+from DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent import PilotLoggingAgent
+from DIRAC import gLogger, gConfig, S_OK, S_ERROR
+
+gLogger.setLevel("DEBUG")
+
+# Mock Objects
+mockReply = MagicMock()
+mockReply1 = MagicMock()
+mockOperations = MagicMock()
+mockTornadoClient = MagicMock()
+mockDataManager = MagicMock()
+mockAM = MagicMock()
+mockNone = MagicMock()
+mockNone.return_value = None
+
+upDict = {
+    "OK": True,
+    "Value": {"User": "proxyUser", "Group": "proxyGroup"},
+}
+
+
+@pytest.fixture
+def plaBase(mocker):
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.AgentModule.__init__")
+    mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.JobCleaningAgent.AgentModule._AgentModule__moduleProperties",
+        side_effect=lambda x, y=None: y,
+        create=True,
+    )
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.AgentModule.am_getOption", return_value=mockAM)
+    mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.getVOs",
+        return_value={"OK": True, "Value": ["gridpp", "lz"]},
+    )
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.gConfig.getValue", return_value="GridPP")
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.Operations.getValue", side_effect=mockReply)
+    mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.Operations.getOptionsDict", side_effect=mockReply1
+    )
+    pla = PilotLoggingAgent()
+    pla.log = gLogger
+    pla._AgentModule__configDefaults = mockAM
+    return pla
+
+
+@pytest.fixture
+def pla_initialised(mocker, plaBase):
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.PilotLoggingAgent.executeForVO")
+    plaBase.initialize()
+    return plaBase
+
+
+@pytest.fixture
+def pla(mocker, plaBase):
+    mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.TornadoPilotLoggingClient",
+        side_effect=mockTornadoClient,
+    )
+    mocker.patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.Operations", side_effect=mockOperations)
+    mocker.patch(
+        "DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.DataManager",
+        side_effect=mockDataManager,
+    )
+    plaBase.initialize()
+    return plaBase
+
+
+def test_initialize(plaBase):
+    res = plaBase.initialize()
+    assert plaBase.voList == plaModule.getVOs()["Value"]
+    assert plaBase.setup is not None
+    assert res == S_OK()
+
+
+@pytest.mark.parametrize(
+    "mockReplyInput, expected, expectedExecOut, expected2",
+    [
+        ("/Pilot/RemoteLogging", [True, False], S_OK(), upDict),
+        ("/Pilot/RemoteLogging", [False, False], S_OK(), upDict),
+        ("/Pilot/RemoteLogging", [True, False], S_ERROR("Execute for VO failed"), upDict),
+    ],
+)
+def test_execute(pla_initialised, mockReplyInput, expected, expectedExecOut, expected2):
+    """Testing a thin version of execute (executeForVO is mocked)"""
+    assert pla_initialised.voList == plaModule.getVOs()["Value"]
+    mockReply.side_effect = expected
+    mockReply1.return_value = expected2
+    # remote pilot logging on (gridpp only) and off.
+    pla_initialised.executeForVO.return_value = expectedExecOut
+    res = pla_initialised.execute()
+    if not any(expected):
+        pla_initialised.executeForVO.assert_not_called()
+    else:
+        assert pla_initialised.executeForVO.called
+        pla_initialised.executeForVO.assert_called_with(
+            "gridpp",
+            proxyUserName=upDict["Value"]["User"],
+            proxyUserGroup=upDict["Value"]["Group"],
+        )
+    assert res["OK"] == expectedExecOut["OK"]
+
+
+@pytest.mark.parametrize(
+    "ppath, files, result",
+    [
+        ("pilot/log/path/", ["file1.log", "file2.log", "file3.log"], S_OK()),
+        ("pilot/log/path/", [], S_OK()),
+    ],
+)
+def test_executeForVO(pla, ppath, files, result):
+    opsHelperValues = {"OK": True, "Value": {"UploadSE": "testUploadSE", "UploadPath": "/gridpp/uploadPath"}}
+    # full local temporary path:
+    filepath = os.path.join(tempfile.TemporaryDirectory().name, ppath)
+    # this is what getMetadata returns:
+    resDict = {"OK": True, "Value": {"LogPath": filepath}}
+    mockTornadoClient.return_value.getMetadata.return_value = resDict
+    mockDataManager.return_value.putAndRegister.return_value = result
+    if files:
+        os.makedirs(os.path.join(filepath, "gridpp"), exist_ok=True)
+    for elem in files:
+        open(os.path.join(filepath, "gridpp", elem), "w")
+    mockOperations.return_value.getOptionsDict.return_value = opsHelperValues
+    pla.opsHelper = mockOperations.return_value
+    # success route
+    res = pla.executeForVO(vo="gridpp")
+    mockTornadoClient.assert_called_with(useCertificates=True)
+    assert mockTornadoClient.return_value.getMetadata.called
+    # only called with a non-empty file list:
+    if files:
+        assert mockDataManager.return_value.putAndRegister.called
+    assert res == S_OK()
+
+
+def test_executeForVOMetaFails(pla):
+    opsHelperValues = {"OK": True, "Value": {"UploadSE": "testUploadSE", "UploadPath": "/gridpp/uploadPath"}}
+    mockOperations.return_value.getOptionsDict.return_value = opsHelperValues
+    pla.opsHelper = mockOperations.return_value
+    # getMetadata call fails.
+    mockTornadoClient.return_value.getMetadata.return_value = {"OK": False, "Message": "Failed, sorry.."}
+    res = pla.executeForVO(vo="anything")
+    assert res["OK"] is False
+
+
+@pytest.mark.parametrize(
+    "opsHelperValues, expectedRes",
+    [
+        ({"OK": True, "Value": {"UploadPath": "/gridpp/uploadPath"}}, S_ERROR("Upload SE not defined")),
+        ({"OK": True, "Value": {"UploadSE": "testUploadSE"}}, S_ERROR("Upload path on SE testUploadSE not defined")),
+        ({"OK": False}, S_ERROR(f"No pilot section for gridpp vo")),
+    ],
+)
+def test_executeForVOBadConfig(pla, opsHelperValues, expectedRes):
+    """Testing an incomplete configuration"""
+    mockOperations.return_value.getOptionsDict.return_value = opsHelperValues
+    pla.opsHelper = mockOperations.return_value
+    res = pla.executeForVO(vo="gridpp")
+    assert res["OK"] is False
+    assert res["Message"] == expectedRes["Message"]
+    mockTornadoClient.return_value.getMetadata.reset_mock()
+    mockTornadoClient.return_value.getMetadata.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "filename, fileAge, ageLimit, expectedResult", [("survives.log", 10, 20, True), ("getsdeleted.log", 21, 20, False)]
+)
+def test_oldLogsCleaner(plaBase, filename, fileAge, ageLimit, expectedResult):
+    """Testing old files removal"""
+    plaBase.clearPilotsDelay = ageLimit
+    filepath = tempfile.TemporaryDirectory().name
+    os.makedirs(filepath, exist_ok=True)
+    testfile = os.path.join(filepath, filename)
+    fd = open(testfile, "w")
+    fd.close()
+    assert os.path.exists(testfile) is True
+    # cannot patch os.stat globally because os.path.exists uses it !
+    with patch("DIRAC.WorkloadManagementSystem.Agent.PilotLoggingAgent.os.stat") as mockOSStat:
+        mockOSStat.return_value.st_mtime = time.time() - fileAge * 86400  # file older that fileAge in seconds
+        plaBase.clearOldPilotLogs(filepath)
+    assert os.path.exists(testfile) is expectedResult
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/FileCacheLoggingPlugin.py b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/FileCacheLoggingPlugin.py
new file mode 100644
index 00000000000..32ca1eb65cb
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/FileCacheLoggingPlugin.py
@@ -0,0 +1,124 @@
+"""
+File cache logging plugin.
+"""
+import os
+import json
+import re
+from DIRAC import S_OK, S_ERROR, gLogger
+from DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.PilotLoggingPlugin import PilotLoggingPlugin
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class FileCacheLoggingPlugin(PilotLoggingPlugin):
+    """
+    File cache logging. Log records are appended to a file, one for each pilot.
+    It is assumed that an agent will be installed together with this plugin, which will copy
+    the files to a safe place and clear the cache.
+    """
+
+    def __init__(self):
+        """
+        Sets the pilot log files location for a WebServer.
+
+        """
+        # UUID pattern
+        self.pattern = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")
+        # pilot stamp pattern
+        self.stamppattern = re.compile(r"^[0-9a-f]{32}$")
+        self.meta = {}
+        logPath = os.path.join(os.getcwd(), "pilotlogs")
+        self.meta["LogPath"] = logPath
+        if not os.path.exists(logPath):
+            os.makedirs(logPath)
+        sLog.verbose("Pilot logging directory:", logPath)
+
+    def sendMessage(self, message, pilotUUID, vo):
+        """
+        File cache sendMessage method. Write the log message to a file line by line.
+
+        :param str message: text to log in json format
+        :param str pilotUUID: pilot id. Optimally it should be a pilot stamp if available, otherwise a generated UUID.
+        :param str vo: VO name of a pilot which sent the message.
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        if not self._verifyUUIDPattern(pilotUUID):
+            return S_ERROR("Pilot UUID is invalid")
+
+        dirname = os.path.join(self.meta["LogPath"], vo)
+        try:
+            if not os.path.exists(dirname):
+                os.mkdir(dirname)
+            with open(os.path.join(dirname, pilotUUID), "a") as pilotLog:
+                try:
+                    messageContent = json.loads(message)
+                    if isinstance(messageContent, list):
+                        for elem in messageContent:
+                            pilotLog.write(elem + "\n")
+                    else:
+                        # it could be a string, if emitted by pilot logger StringIO handler
+                        pilotLog.write(messageContent)
+                except OSError as oserr:
+                    sLog.error("Error writing to log file:", repr(oserr))
+                    return S_ERROR(repr(oserr))
+        except OSError as err:
+            sLog.exception("Error opening a pilot log file", lException=err)
+            return S_ERROR(repr(err))
+        return S_OK(f"Message logged successfully for pilot: {pilotUUID} and {vo}")
+
+    def finaliseLogs(self, payload, logfile, vo):
+        """
+        Finalise a log file. Finalised logfile can be copied to a secure location.
+
+        :param dict payload: additional info, a plugin might want to use (i.e. the system return code of a pilot script)
+        :param str logfile: log filename (pilotUUID).
+        :param str vo: VO name of a pilot which sent the message.
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        returnCode = json.loads(payload).get("retCode", 0)
+
+        if not self._verifyUUIDPattern(logfile):
+            return S_ERROR("Pilot UUID is invalid")
+
+        try:
+            filepath = self.meta["LogPath"]
+            os.rename(os.path.join(filepath, vo, logfile), os.path.join(filepath, vo, logfile + ".log"))
+            sLog.info(f"Log file {logfile} finalised for pilot: (return code: {returnCode})")
+            return S_OK()
+        except Exception as err:
+            sLog.exception("Exception when finalising log")
+            return S_ERROR(repr(err))
+
+    def getMeta(self):
+        """
+        Return any metadata related to this plugin. The "LogPath" is the minimum requirement for the dict to contain.
+
+        :return: Dirac S_OK containing the metadata or S_ERROR if the LogPath is not defined.
+        :rtype: dict
+        """
+        if "LogPath" in self.meta:
+            return S_OK(self.meta)
+        return S_ERROR("No Pilot logging directory defined")
+
+    def _verifyUUIDPattern(self, logfile):
+        """
+        Verify if the name of the log file matches the required pattern.
+
+        :param str name: file name
+        :return: re.match result
+        :rtype: re.Match object or None.
+        """
+
+        res = self.stamppattern.match(logfile)
+        if not res:
+            res = self.pattern.match(logfile)
+        if not res:
+            sLog.error(
+                "Pilot UUID does not match the UUID nor the stamp pattern. ",
+                f"UUID: {logfile}, pilot stamp pattern {self.stamppattern}, UUID pattern {self.pattern}",
+            )
+        return res
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/MQPilotLoggingPlugin.py b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/MQPilotLoggingPlugin.py
new file mode 100644
index 00000000000..6e878b642be
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/MQPilotLoggingPlugin.py
@@ -0,0 +1,57 @@
+"""
+MeessageQueue Pilot logging plugin. Just log messages.
+"""
+import re
+from DIRAC import S_OK, S_ERROR, gLogger
+from DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.PilotLoggingPlugin import PilotLoggingPlugin
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class MQPilotLoggingPlugin(PilotLoggingPlugin):
+    """
+    A template of a MQ logging plugin.
+    It gets the message and converts it to a list of Dictionaries to be shipped to a remote MQ service
+    """
+
+    def __init__(self):
+        sLog.warning("MQPilotLoggingPlugin skeleton is being used. NO-op")
+        self.rcompiled = re.compile(
+            r"(?P<date>[0-9-]+)T(?P<time>[0-9:,]+)Z (?P<loglevel>DEBUG|INFO|ERROR|NOTICE) (?:\[(?P<source>[a-zA-Z]+)\] )?(?P<message>.*)"
+        )
+
+    def sendMessage(self, message, pilotID, vo):
+        """
+        A message could of a form:
+        2022-06-10T11:02:02,823512Z DEBUG    [pilotLogger] X509_USER_PROXY=/scratch/dir_2313/user.proxy
+
+
+        :param str message: text to log
+        :param str pilotID: pilot id. Optimally it should be a pilot stamp if available, otherwise a generated UUID.
+        :param str vo: VO name of a pilot which sent the message.
+        :return: None
+        :rtype: None
+        """
+
+        res = self.rcompiled.match(message)
+        if res:
+            resDict = res.groupdict()
+            # {'date': '2022-06-10', 'loglevel': 'DEBUG',
+            # 'message': '   [pilotLogger] X509_USER_PROXY=/scratch/dir_2313/user.proxy',
+            # 'time': '11:02:02,823512', 'source': None}
+            return S_OK()
+        else:
+            return S_ERROR("No match - message could not be parsed")
+
+    def finaliseLogs(self, payload, pilotID, vo):
+        """
+        Log finaliser method. To indicate that the log is now complete.
+
+        :param dict payload: additional info, a plugin might want to use (i.e. the system return code of a pilot script)
+        :param str pilotID: unique pilot ID.
+        :param str vo: VO name of a pilot which sent the message.
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        return S_OK()
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/PilotLoggingPlugin.py b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/PilotLoggingPlugin.py
new file mode 100644
index 00000000000..9348c6a751d
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/PilotLoggingPlugin.py
@@ -0,0 +1,43 @@
+"""
+Pilot logging plugin abstract class.
+"""
+from abc import ABC, abstractmethod
+from DIRAC import S_OK, S_ERROR, gLogger
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class PilotLoggingPlugin(ABC):
+    """
+    Remote pilot logging bas abstract class. It defines abstract methods used to sent messages to the server.
+    Any pilot logger plugin should inherit from this class and implement a (sub)set of methods required by
+    :class:`TornadoPilotLoggingHandler`.
+    """
+
+    @abstractmethod
+    def sendMessage(self, message, pilotID, vo):
+        """
+        sendMessage method, carrying the unique pilot identity and a VO name.
+
+        :param str message: text to log in json format
+        :param str pilotID: pilot id. Optimally it should be a pilot stamp if available, otherwise a generated UUID.
+        :param str vo: VO name of a pilot which sent the message.
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        pass
+
+    @abstractmethod
+    def finaliseLogs(self, payload, pilotID, vo):
+        """
+        Log finaliser method. To indicate that the log is now complete.
+
+        :param dict payload: additional info, a plugin might want to use (i.e. the system return code of a pilot script)
+        :param str pilotID: unique pilot ID.
+        :param str vo: VO name of a pilot which sent the message.
+        :return: S_OK or S_ERROR
+        :rtype: dict
+        """
+
+        pass
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/__init__.py b/src/DIRAC/WorkloadManagementSystem/Client/PilotLoggingPlugins/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/TornadoPilotLoggingClient.py b/src/DIRAC/WorkloadManagementSystem/Client/TornadoPilotLoggingClient.py
new file mode 100644
index 00000000000..808d57f7a7b
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Client/TornadoPilotLoggingClient.py
@@ -0,0 +1,17 @@
+from DIRAC.Core.Base.Client import Client, createClient
+
+
+@createClient("WorkloadManagement/TornadoPilotLogging")
+class TornadoPilotLoggingClient(Client):
+    def __init__(self, url=None, **kwargs):
+        """
+        Initialise a client.
+
+        :param str url: Server URL, if None, defaults to "WorkloadManagement/TornadoPilotLogging"
+        :param dict kwargs: additional keyword arguments, currently unused.
+        """
+        super().__init__(**kwargs)
+        if not url:
+            self.serverURL = "WorkloadManagement/TornadoPilotLogging"
+        else:
+            self.serverURL = url
diff --git a/src/DIRAC/WorkloadManagementSystem/Client/test/Test_TornadoPilotLoggingClient.py b/src/DIRAC/WorkloadManagementSystem/Client/test/Test_TornadoPilotLoggingClient.py
new file mode 100644
index 00000000000..afca22d9b20
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Client/test/Test_TornadoPilotLoggingClient.py
@@ -0,0 +1,17 @@
+import unittest
+from unittest.mock import patch
+from unittest.mock import MagicMock
+
+import DIRAC.WorkloadManagementSystem.Client.TornadoPilotLoggingClient as tplc
+
+
+class TestTornadoPilotLoggingClient(unittest.TestCase):
+    def test_client(self):
+        client = tplc.TornadoPilotLoggingClient("test.server", useCertificates=True)
+        self.assertEqual(client.serverURL, "test.server")
+        client = tplc.TornadoPilotLoggingClient(useCertificates=True)
+        self.assertEqual(client.serverURL, "WorkloadManagement/TornadoPilotLogging")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg b/src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg
index 5a56a2bfb54..698646cdefd 100644
--- a/src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg
+++ b/src/DIRAC/WorkloadManagementSystem/ConfigTemplate.cfg
@@ -19,6 +19,23 @@ Services
     }
   }
   ##END
+  ##BEGIN TornadoPilotLogging
+  TornadoPilotLogging
+  {
+    Protocol = https
+    Authorization
+    {
+      Default = authenticated
+      sendMessage = "Operator"
+      sendMessage += "GenericPilot"
+      getMetadata = "Operator"
+      getMetadata += "TrustedHost"
+      finaliseLogs = "Operator"
+      finaliseLogs += "Pilot"
+      finaliseLogs += "GenericPilot"
+    }
+  }
+  ##END
   ##BEGIN JobMonitoring
   JobMonitoring
   {
@@ -184,6 +201,12 @@ Agents
     PilotAccountingEnabled = yes
   }
   ##END
+  ##BEGIN PilotLoggingAgent
+  PilotLoggingAgent
+  {
+    PollingTime = 600
+  }
+  ##END
   JobAgent
   {
     PollingTime = 20
diff --git a/src/DIRAC/WorkloadManagementSystem/Service/TornadoPilotLoggingHandler.py b/src/DIRAC/WorkloadManagementSystem/Service/TornadoPilotLoggingHandler.py
new file mode 100644
index 00000000000..cd98938096f
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Service/TornadoPilotLoggingHandler.py
@@ -0,0 +1,101 @@
+""" Tornado-based HTTPs PilotLogging service.
+
+.. literalinclude:: ../ConfigTemplate.cfg
+  :start-after: ##BEGIN TornadoPilotLogging
+  :end-before: ##END
+  :dedent: 2
+  :caption: PilotLogging options
+"""
+
+import os
+from DIRAC import gLogger, S_OK, S_ERROR
+from DIRAC.ConfigurationSystem.Client.Helpers import Registry
+from DIRAC.Core.Tornado.Server.TornadoService import TornadoService
+from DIRAC.Core.DISET.RequestHandler import getServiceOption
+from DIRAC.Core.Utilities.ObjectLoader import ObjectLoader
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class TornadoPilotLoggingHandler(TornadoService):
+    log = sLog
+
+    @classmethod
+    def initializeHandler(cls, infoDict):
+        """
+        Called once, at the first request. Create a directory where pilot logs will be stored.
+
+        :param infoDict:
+        :return: None
+        """
+
+        defaultOption, defaultClass = "LoggingPlugin", "PilotLoggingPlugin"
+        configValue = getServiceOption(infoDict, defaultOption, defaultClass)
+
+        result = ObjectLoader().loadObject(
+            f"WorkloadManagementSystem.Client.PilotLoggingPlugins.{configValue}", configValue
+        )
+        if not result["OK"]:
+            cls.log.error("Failed to load LoggingPlugin", f"{configValue}: {result['Message']}")
+            return result
+
+        componentClass = result["Value"]
+        cls.loggingPlugin = componentClass()
+        cls.log.debug("Loaded: PilotLoggingPlugin class", configValue)
+
+        cls.meta = {}
+        logPath = os.path.join(os.getcwd(), "pilotlogs")
+        cls.meta["LogPath"] = logPath
+        if not os.path.exists(logPath):
+            os.makedirs(logPath)
+        cls.log.verbose("Pilot logging directory:", logPath)
+
+    def export_sendMessage(self, message, pilotUUID):
+        # def export_sendMessage(self, message, pilotUUID):
+        """
+        The method logs messages to Tornado and forwards pilot log files, one per pilot, to a relevant plugin.
+        The pilot is identified by its UUID.
+
+        :param message: message sent by a client, a list of strings in JSON format
+        :param pilotUUID: pilot UUID
+        :return: S_OK or S_ERROR if a plugin cannot process the message.
+        :rtype: dict
+        """
+        ## Insert your method here, don't forget the return should be serializable
+        ## Returned value may be an S_OK/S_ERROR
+        ## You don't need to serialize in JSON, Tornado will do it
+
+        # determine client VO
+        vo = self.__getClientVO()
+        # the plugin returns S_OK or S_ERROR
+        # leave JSON decoding to the selected plugin:
+        result = self.loggingPlugin.sendMessage(message, pilotUUID, vo)
+        return result
+
+    def export_getMetadata(self):
+        """
+        Get PilotLoggingHandler metadata. Intended to be used by a client or an agent.
+
+        :return: S_OK containing a metadata dictionary
+        """
+        return self.loggingPlugin.getMeta()
+
+    def export_finaliseLogs(self, payload, pilotUUID):
+        """
+        Finalise a log file. Finalised logfile can be copied to a secure location, if a file cache is used.
+
+        :param payload: data passed to the plugin finaliser.
+        :type payload: dict
+        :param pilotUUID: pilot UUID
+        :return: S_OK or S_ERROR (via the plugin involved)
+        :rtype: dict
+        """
+
+        vo = self.__getClientVO()
+
+        # The plugin returns the Dirac S_OK or S_ERROR object
+        return self.loggingPlugin.finaliseLogs(payload, pilotUUID, vo)
+
+    def __getClientVO(self):
+        # get client credentials to determine the VO
+        return Registry.getVOForGroup(self.getRemoteCredentials()["group"])
diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/PilotCStoJSONSynchronizer.py b/src/DIRAC/WorkloadManagementSystem/Utilities/PilotCStoJSONSynchronizer.py
index 428ddab87c2..27da25f8a90 100644
--- a/src/DIRAC/WorkloadManagementSystem/Utilities/PilotCStoJSONSynchronizer.py
+++ b/src/DIRAC/WorkloadManagementSystem/Utilities/PilotCStoJSONSynchronizer.py
@@ -9,6 +9,7 @@
 import shutil
 import tarfile
 import datetime
+from typing import Any
 
 from git import Repo
 
@@ -16,6 +17,7 @@
 from DIRAC.ConfigurationSystem.Client.ConfigurationData import gConfigurationData
 from DIRAC.ConfigurationSystem.Client.Helpers.Operations import Operations
 from DIRAC.ConfigurationSystem.Client.Helpers.Path import cfgPath
+from DIRAC.Core.Utilities.ReturnValues import DReturnType, DOKReturnType
 
 
 class PilotCStoJSONSynchronizer:
@@ -59,8 +61,10 @@ def __init__(self):
         self.pilotRepoBranch = ops.getValue("Pilot/pilotRepoBranch", self.pilotRepoBranch)
         self.pilotVORepoBranch = ops.getValue("Pilot/pilotVORepoBranch", self.pilotVORepoBranch)
 
-    def getCSDict(self, includeMasterCS=True):
-        """Gets minimal info for running a pilot, from the CS
+    def getCSDict(self, includeMasterCS: bool = True) -> DReturnType[Any]:
+        """
+        Gets minimal info for running a pilot, from the CS. The complete Operations section is
+        dumped to a dictionary. A decision which VO to use will be delegated to a pilot.
 
         :returns: pilotDict (containing pilots run info)
         :rtype: S_OK, S_ERROR, value is pilotDict
@@ -68,41 +72,26 @@ def getCSDict(self, includeMasterCS=True):
 
         pilotDict = {
             "timestamp": datetime.datetime.utcnow().isoformat(),
-            "Setups": {},
             "CEs": {},
             "GenericPilotDNs": [],
         }
 
         self.log.info("-- Getting the content of the CS --")
 
-        # These are in fact not only setups: they may be "Defaults" sections, or VOs, in multi-VOs installations
-        setupsRes = gConfig.getSections("/Operations/")
-        if not setupsRes["OK"]:
-            self.log.error("Can't get sections from Operations", setupsRes["Message"])
-            return setupsRes
-        setupsInOperations = setupsRes["Value"]
-
-        # getting the setup(s) in this CS, and comparing with what we found in Operations
-        setupsInDIRACRes = gConfig.getSections("DIRAC/Setups")
-        if not setupsInDIRACRes["OK"]:
-            self.log.error("Can't get sections from DIRAC/Setups", setupsInDIRACRes["Message"])
-            return setupsInDIRACRes
-        setupsInDIRAC = setupsInDIRACRes["Value"]
-
-        # Handling the case of multi-VO CS
-        if not set(setupsInDIRAC).intersection(set(setupsInOperations)):
-            vos = list(setupsInOperations)
-            for vo in vos:
-                setupsFromVOs = gConfig.getSections(f"/Operations/{vo}")
-                if not setupsFromVOs["OK"]:
-                    continue
-                else:
-                    setupsInOperations = setupsFromVOs["Value"]
-
-        self.log.verbose("From Operations/[Setup]/Pilot")
-
-        for setup in setupsInOperations:
-            self._getPilotOptionsPerSetup(setup, pilotDict)
+        # Get the whole Operations section as a dict.
+        self.log.verbose("From Operations (whole section)")
+        opRes = gConfig.getOptionsDictRecursively("/Operations")
+        if not opRes["OK"]:
+            self.log.error("Can't get sections from Operations", opRes["Message"])
+            return opRes
+        pilotDict.update(opRes["Value"])
+
+        # we still need a pilotVOVersion
+        self.opsHelper = Operations(setup=self.pilotSetup)
+        self.pilotVOVersion = self.opsHelper.getValue("/Pilot/Version")
+        # if self.pilotVORepo is defined and self.pilotVOVersion is not, syncScripts is likely to fail.
+        if self.pilotVOVersion is None and self.pilotVORepo:
+            self.log.error("Pilot VO repo is set in the CS but the pilot VO version is not. Expect problems ahead")
 
         self.log.verbose("From Resources/Sites")
         sitesSection = gConfig.getSections("/Resources/Sites/")
@@ -170,85 +159,7 @@ def getCSDict(self, includeMasterCS=True):
 
         return S_OK(pilotDict)
 
-    def _getPilotOptionsPerSetup(self, setup, pilotDict):
-        """Given a setup, returns its pilot options in a dictionary"""
-
-        options = gConfig.getOptionsDict(f"/Operations/{setup}/Pilot")
-        if not options["OK"]:
-            self.log.warn("Section does not exist: skipping", f"/Operations/{setup}/Pilot ")
-            return
-
-        # We include everything that's in the Pilot section for this setup
-        if setup == self.pilotSetup:
-            self.pilotVOVersion = options["Value"]["Version"]
-        pilotDict["Setups"][setup] = options["Value"]
-        # We update separately 'GenericPilotDNs'
-        try:
-            pilotDict["GenericPilotDNs"].append(pilotDict["Setups"][setup]["GenericPilotDN"])
-        except KeyError:
-            pass
-        ceTypesCommands = gConfig.getOptionsDict(f"/Operations/{setup}/Pilot/Commands")
-        if ceTypesCommands["OK"]:
-            # It's ok if the Pilot section doesn't list any Commands too
-            pilotDict["Setups"][setup]["Commands"] = {}
-            for ceType in ceTypesCommands["Value"]:
-                # FIXME: inconsistent that we break Commands down into a proper list but other things are comma-list strings
-                pilotDict["Setups"][setup]["Commands"][ceType] = ceTypesCommands["Value"][ceType].split(", ")
-                # pilotDict['Setups'][setup]['Commands'][ceType] = ceTypesCommands['Value'][ceType]
-        if "CommandExtensions" in pilotDict["Setups"][setup]:
-            # FIXME: inconsistent that we break CommandExtensionss down into a proper
-            # list but other things are comma-list strings
-            pilotDict["Setups"][setup]["CommandExtensions"] = pilotDict["Setups"][setup]["CommandExtensions"].split(
-                ", "
-            )
-            # pilotDict['Setups'][setup]['CommandExtensions'] = pilotDict['Setups'][setup]['CommandExtensions']
-
-        # Getting the details aboout the MQ Services to be used for logging, if any
-        if "LoggingMQService" in pilotDict["Setups"][setup]:
-            loggingMQService = gConfig.getOptionsDict(
-                f"/Resources/MQServices/{pilotDict['Setups'][setup]['LoggingMQService']}"
-            )
-            if not loggingMQService["OK"]:
-                self.log.error(loggingMQService["Message"])
-                return loggingMQService
-            pilotDict["Setups"][setup]["Logging"] = {}
-            pilotDict["Setups"][setup]["Logging"]["Host"] = loggingMQService["Value"]["Host"]
-            pilotDict["Setups"][setup]["Logging"]["Port"] = loggingMQService["Value"]["Port"]
-
-            loggingMQServiceQueuesSections = gConfig.getSections(
-                f"/Resources/MQServices/{pilotDict['Setups'][setup]['LoggingMQService']}/Queues"
-            )
-            if not loggingMQServiceQueuesSections["OK"]:
-                self.log.error(loggingMQServiceQueuesSections["Message"])
-                return loggingMQServiceQueuesSections
-            pilotDict["Setups"][setup]["Logging"]["Queue"] = {}
-
-            for queue in loggingMQServiceQueuesSections["Value"]:
-                loggingMQServiceQueue = gConfig.getOptionsDict(
-                    f"/Resources/MQServices/{pilotDict['Setups'][setup]['LoggingMQService']}/Queues/{queue}"
-                )
-                if not loggingMQServiceQueue["OK"]:
-                    self.log.error(loggingMQServiceQueue["Message"])
-                    return loggingMQServiceQueue
-                pilotDict["Setups"][setup]["Logging"]["Queue"][queue] = loggingMQServiceQueue["Value"]
-
-            queuesRes = gConfig.getSections(
-                f"/Resources/MQServices/{pilotDict['Setups'][setup]['LoggingMQService']}/Queues"
-            )
-            if not queuesRes["OK"]:
-                return queuesRes
-            queues = queuesRes["Value"]
-            queuesDict = {}
-            for queue in queues:
-                queueOptionRes = gConfig.getOptionsDict(
-                    f"/Resources/MQServices/{pilotDict['Setups'][setup]['LoggingMQService']}/Queues/{queue}"
-                )
-                if not queueOptionRes["OK"]:
-                    return queueOptionRes
-                queuesDict[queue] = queueOptionRes["Value"]
-            pilotDict["Setups"][setup]["Logging"]["Queues"] = queuesDict
-
-    def syncScripts(self):
+    def syncScripts(self) -> DOKReturnType[Any]:
         """Clone the pilot scripts from the Pilot repositories (handle also extensions)"""
         tarFiles = []
 
diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/PilotWrapper.py b/src/DIRAC/WorkloadManagementSystem/Utilities/PilotWrapper.py
index a2fe5d0a372..ed5c62f96dd 100644
--- a/src/DIRAC/WorkloadManagementSystem/Utilities/PilotWrapper.py
+++ b/src/DIRAC/WorkloadManagementSystem/Utilities/PilotWrapper.py
@@ -46,18 +46,60 @@
 import time
 import tarfile
 import hashlib
+# for remote logging
+import subprocess
+import json
+import os
+import urllib
+import ssl
+import shlex
+from uuid import uuid1
+
+try:
+    # For Python 3.0 and later
+    from urllib.request import urlopen, HTTPError, URLError
+    from urllib.parse import urlencode
+except ImportError:
+    # Fall back to Python 2's urllib2
+    from urllib2 import urlopen, HTTPError, URLError
+    from urllib import urlencode
+
+try:
+  from cStringIO import StringIO
+except ImportError:
+  from io import StringIO
+
+# formatting with microsecond accuracy, (ISO-8601)
+
+class MicrosecondFormatter(logging.Formatter):
+  def formatTime(self, record, datefmt=None):
+    ct = self.converter(record.created)
+    if datefmt:
+        s = time.strftime(datefmt, ct)
+    else:
+        t = time.strftime("%%Y-%%m-%%dT%%H:%%M:%%S", ct)
+        s = "%%s,%%06dZ" %% (t, (record.created - int(record.created)) * 1e6)
+    return s
 
 # setting up the logging
-formatter = logging.Formatter(fmt='%%(asctime)s UTC %%(levelname)-8s %%(message)s', datefmt='%%Y-%%m-%%d %%H:%%M:%%S')
+# formatter = logging.Formatter(fmt='%%(asctime)s UTC %%(levelname)-8s %%(message)s', datefmt='%%Y-%%m-%%d %%H:%%M:%%S')
+formatter = MicrosecondFormatter('%%(asctime)s %%(levelname)-8s [%%(name)s] %%(message)s')
 logging.Formatter.converter = time.gmtime
 try:
   screen_handler = logging.StreamHandler(stream=sys.stdout)
 except TypeError:  # python2.6
   screen_handler = logging.StreamHandler(strm=sys.stdout)
 screen_handler.setFormatter(formatter)
+
+# add a string buffer handler
+sio = StringIO()
+buffer = logging.StreamHandler(sio)
+buffer.setFormatter(formatter)
+
 logger = logging.getLogger('pilotLogger')
 logger.setLevel(logging.DEBUG)
 logger.addHandler(screen_handler)
+logger.addHandler(buffer)
 
 # just logging the environment as first thing
 logger.debug('===========================================================')
@@ -66,6 +108,8 @@
   logger.debug(key + '=' + val)
 logger.debug('===========================================================\\n')
 
+logger.debug(sys.version)
+
 # putting ourselves in the right directory
 pilotExecDir = '%(pilotExecDir)s'
 if not pilotExecDir:
@@ -268,12 +312,29 @@ def pilotWrapperScript(
     localPilot += (
         """
 # now finally launching the pilot script (which should be called dirac-pilot.py)
-cmd = "$py dirac-pilot.py %s"
-logger.info('Executing: %%s' %% cmd)
-sys.stdout.flush()
-ret = os.system(cmd)
+# get the setup name an -z, if present to get remote logging in place
+opt = "%s"
+# try to get a pilot stamp from the environment:
+UUID =  os.environ.get('DIRAC_PILOT_STAMP')
+if UUID is None:
+    UUID = str(uuid1())
+opt = opt + " --pilotUUID " + UUID
+
+args = opt.split()
+
+# let's see if we have remote logging enabled (-z), if not run the pilot script with os.system, as before
+
+logger.info("dirac-pilot.py  will be called: with %%s " %% args)
+
+# call dirac-pilot.py and pass log records accumulated so far as a standard input. A decision to use the buffer
+# in a remote logger is made later in dirac-pilot.py script
 
+proc = subprocess.Popen(shlex.split("$py dirac-pilot.py " + opt), bufsize = 1, stdin = subprocess.PIPE,
+                        stdout=sys.stdout, stderr=sys.stderr, universal_newlines = True)
+proc.communicate(buffer.stream.getvalue())
+ret = proc.returncode
 # and cleaning up
+buffer.stream.close()
 shutil.rmtree(pilotWorkingDirectory)
 
 # did it fail?
diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_PilotWrapper.py b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_PilotWrapper.py
index 0b7f33f3c3d..ee925a649e4 100644
--- a/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_PilotWrapper.py
+++ b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_PilotWrapper.py
@@ -12,7 +12,7 @@ def test_scriptEmpty():
     """test script creation"""
     res = pilotWrapperScript()
 
-    assert 'cmd = "$py dirac-pilot.py "' in res
+    assert "$py dirac-pilot.py " in res
     assert 'os.environ["someName"]="someValue"' not in res
 
 
diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_TornadoPilotLoggingHandler.py b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_TornadoPilotLoggingHandler.py
new file mode 100644
index 00000000000..4b1ae29a197
--- /dev/null
+++ b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_TornadoPilotLoggingHandler.py
@@ -0,0 +1,154 @@
+import unittest
+from unittest.mock import patch
+import os
+import json
+import tempfile
+import DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler
+from DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin import FileCacheLoggingPlugin
+from DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler import TornadoPilotLoggingHandler
+
+
+class TornadoPilotLoggingHandlerTestCase(unittest.TestCase):
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os, "makedirs")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os, "getcwd")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os.path, "exists")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler, "getServiceOption")
+    def test_initializeHandlerBasic(self, mockOpt, mockExists, mockGetcwd, mockMakedirs):
+        """
+        Test the initialiser with a PilotLoggingPlugin. os.* calls used are mocked.
+
+        :param mockExists:
+        :type mockExists:
+        :param mockGetcwd:
+        :type mockGetcwd:
+        :param mockMakedirs:
+        :type mockMakedirs:
+        :return:
+        :rtype:
+        """
+        mockExists.return_value = False  # will create a file
+        mockGetcwd.return_value = "/tornado/document/root"
+        mockOpt.return_value = "FileCacheLoggingPlugin"
+        TornadoPilotLoggingHandler.initializeHandler(
+            {"csPaths": "noexistent"}
+        )  # should return FileCacheLoggingPlugin plugin
+        mockMakedirs.assert_called_with(os.path.join("/tornado/document/root", "pilotlogs"))
+        self.assertEqual(TornadoPilotLoggingHandler.loggingPlugin.__class__.__name__, "FileCacheLoggingPlugin")
+
+        mockMakedirs.reset_mock()
+        mockExists.return_value = True
+        mockGetcwd.return_value = "/tornado/document/root"
+        TornadoPilotLoggingHandler.initializeHandler(
+            {"csPaths": "noexistent"}
+        )  # should return the default, basic plugin
+        mockMakedirs.assert_not_called()
+        self.assertEqual(TornadoPilotLoggingHandler.loggingPlugin.__class__.__name__, "FileCacheLoggingPlugin")
+
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler, "getServiceOption")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os, "makedirs")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os, "getcwd")
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os.path, "exists")
+    def test_initializeHandlerFileCache(self, mockExists, mockGetcwd, mockMakedirs, mockgetServiceOption):
+        """
+        Test loading the (real) file cache plugin.
+
+        :return:
+        :rtype:
+        """
+
+        mockExists.return_value = False  # will create a file
+        mockGetcwd.return_value = "/tornado/document/root"
+        mockgetServiceOption.return_value = "FileCacheLoggingPlugin"
+        TornadoPilotLoggingHandler.initializeHandler(
+            {"csPaths": "noexistent"}
+        )  # should return the default, basic plugin
+        mockMakedirs.assert_called_with(os.path.join("/tornado/document/root", "pilotlogs"))
+        self.assertEqual(TornadoPilotLoggingHandler.loggingPlugin.__class__.__name__, "FileCacheLoggingPlugin")
+        self.assertEqual(2, mockMakedirs.call_count)  # twice: once in the handler, once in the plugin
+
+    @patch.object(DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin.os, "makedirs")
+    @patch.object(DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin.os, "getcwd")
+    @patch.object(DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin.os.path, "exists")
+    def test_FileCachePlugin(self, mockExists, mockGetcwd, mockMakedirs):
+        """
+        Test fileCachePlugin
+        """
+        mockExists.return_value = False  # will create a file
+        mockGetcwd.return_value = "/tornado/document/root"
+        plugin = FileCacheLoggingPlugin()
+        mockGetcwd.assert_called_once()
+        mockExists.assert_called_once_with(os.path.join(mockGetcwd.return_value, "pilotlogs"))
+        mockMakedirs.assert_called_once_with(os.path.join(mockGetcwd.return_value, "pilotlogs"))
+
+        # sendMessage()
+        messsageText = (
+            "2022-02-23 13:48:35.123456 UTC DEBUG [PilotParams] JSON file loaded: pilot.json\n"
+            + "2022-02-23 13:48:36.123456 UTC DEBUG [PilotParams] JSON file analysed: pilot.json"
+        )
+        messageJSON = json.dumps(messsageText)
+        vo = "anyVO"
+        pilotUUID = "78f39a90-2073-11ec-98d7-b496913c0cf4"
+
+        # use a temporary dir, not the one above. Plugin will create the file to write into.
+        with tempfile.TemporaryDirectory(suffix="pilottests") as d:
+            plugin.meta["LogPath"] = d
+            res = plugin.sendMessage(messageJSON, pilotUUID, vo)
+            self.assertTrue(res["OK"])
+            with open(os.path.join(d, vo, pilotUUID)) as pilotLog:
+                content = pilotLog.read()
+                self.assertEqual(content, messsageText)
+
+        # failures ?
+        with tempfile.TemporaryDirectory(suffix="pilottests") as d:
+            plugin.meta["LogPath"] = d
+            os.chmod(d, 0o0000)
+            res = plugin.sendMessage(messageJSON, pilotUUID, vo)
+            self.assertFalse(res["OK"])
+
+        pilotUUID = "whatever"
+        res = plugin.sendMessage(messageJSON, pilotUUID, vo)
+        self.assertFalse(res["OK"])
+
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os.path, "exists")
+    @patch.object(DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin.os, "getcwd")
+    def test_getMeta(self, mockGetcwd, mockExists):
+        mockExists.return_value = True  # will not create a file
+        mockGetcwd.return_value = "/tornado/document/root"  # so we have a path defined
+        plugin = FileCacheLoggingPlugin()
+        res = plugin.getMeta()
+        self.assertTrue(res["OK"])
+        plugin.meta = {}
+        res = plugin.getMeta()
+        self.assertFalse(res["OK"])
+
+    @patch.object(DIRAC.WorkloadManagementSystem.Service.TornadoPilotLoggingHandler.os.path, "exists")
+    @patch.object(DIRAC.WorkloadManagementSystem.Client.PilotLoggingPlugins.FileCacheLoggingPlugin.os, "getcwd")
+    def test_finaliseLogs(self, mockGetcwd, mockExists):
+        mockExists.return_value = True  # will not create a file
+        mockGetcwd.return_value = "/tornado/document/root"  # so we have a path defined (will overwrite it below)
+        plugin = FileCacheLoggingPlugin()
+        vo = "anyVO"
+
+        with tempfile.TemporaryDirectory(suffix="pilottests") as d:
+            plugin.meta["LogPath"] = d
+            payload = '{"retCode": 0}'
+            logfile = "78f39a90-2073-11ec-98d7-b496913c0cf4"  # == pilotUUID
+            os.mkdir(os.path.join(d, vo))  # vo specific directory, normally created by sending the first message
+            # will fail here...
+            res = plugin.finaliseLogs(payload, logfile, vo)
+            self.assertFalse(res["OK"])
+            # create a file ..
+            with open(os.path.join(d, vo, logfile), "w") as f:
+                f.write("Create a dummy logfile")
+            res = plugin.finaliseLogs(payload, logfile, vo)
+            self.assertTrue(res["OK"])
+
+            logfile = "invalid!"
+            res = plugin.finaliseLogs(payload, logfile, vo)
+            self.assertFalse(res["OK"])
+
+
+if __name__ == "__main__":
+    suite = unittest.defaultTestLoader.loadTestsFromTestCase(TornadoPilotLoggingHandlerTestCase)
+    runner = unittest.TextTestRunner(verbosity=2)
+    runner.run(suite)