HPCC-23310 Archive tank files for ESP transaction logging

1. New tankfile tool for checking and archiving acked log files.
2. Add ILogRequestReader for CLogRequestReader so that it can be
accessed from ESP client;
3. Add GetAckedLogFiles method to the WSDecoupledLog service;
4. Add CleanAckedFiles method to the WSDecoupledLog service;
5. Encapsulate the code for checking/looping through
LogAgentGroups/LogAgents into new WSDecoupledLog methods
(such as doActions, etc) so that they can be shared by multiple
WSDecoupledLog actions, including pause, get settings,
and GetAckedLogFiles;

Revise based on review:
1. Move the doAction related methods into action classes;
2. Add more comments, etc.
3. Implement 'safe save';
4. Move the try-catch into doAction();
5. Remove the doActionForAllAgentsInGroup().
6. Re-target to 7.10.x
7. Remove the createIFile() call;
8. Change the fileNames point to reference;
9. Add override to when derived virtual functions are defined;
10. Initialise the fileNames point.

Signed-off-by: wangkx <kevin.wang@lexisnexis.com>

esp/logging/logginglib/logthread.cpp

@@ -622,6 +622,89 @@ void CLogRequestReader::threadmain()
     PROGLOG("LogRequest Reader Thread terminated.");
 }
 
+void CLogRequestReader::reportAckedLogFiles(StringArray& ackedLogFiles)
+{
+    CriticalBlock b(crit);
+    for (auto r : ackedLogFileCheckList)
+        ackedLogFiles.append(r.c_str());
+    ESPLOG(LogMax, "#### The reportAckedLogFiles() done.");
+}
+
+//This method is used to setup an acked file list which contains the acked files for all agents.
+//The first agent reports all the acked files in that agent using the reportAckedLogFiles().
+//Using this method, the list of the acked files is given to the rest agents. If any file inside
+//the list has not been acked in the rest agents, the file should be removed from the list.
+void CLogRequestReader::removeUnknownAckedLogFiles(StringArray& ackedLogFiles)
+{
+    CriticalBlock b(crit);
+    ForEachItemInRev(i, ackedLogFiles)
+    {
+        const char* node = ackedLogFiles.item(i);
+        if (ackedLogFileCheckList.find(node) == ackedLogFileCheckList.end())
+            ackedLogFiles.remove(i);
+    }
+    ESPLOG(LogMax, "#### The removeUnknownAckedLogFiles() done.");
+}
+
+void CLogRequestReader::addNewAckedFileList(const char* list, StringArray& fileNames)
+{
+    OwnedIFile newList = createIFile(list);
+    if (!newList)
+        throw makeStringExceptionV(EspLoggingErrors::UpdateLogFailed, "Failed to access %s", list);
+    if (newList->exists())
+    {
+        if (!newList->remove())
+            throw makeStringExceptionV(EspLoggingErrors::UpdateLogFailed, "Failed to remove old %s", list);
+    }
+    OwnedIFileIO newListIO = newList->open(IFOwrite);
+    if (!newListIO)
+        throw makeStringExceptionV(EspLoggingErrors::UpdateLogFailed, "Failed to open %s", list);
+
+    offset_t pos = 0;
+    ForEachItemIn(i, fileNames)
+    {
+        const char* fileName = fileNames.item(i);
+        StringBuffer line(fileName);
+        line.append("\r\n");
+
+        unsigned len = line.length();
+        newListIO->write(pos, len, line.str());
+        pos += len;
+        PROGLOG("Add AckedLogFile %s to %s", fileName, list);
+    }
+}
+//The file names in the fileNames should be removed from both ackedLogFileCheckList
+//and settings->ackedFileList.
+void CLogRequestReader::cleanAckedLogFiles(StringArray& fileNames)
+{
+    CriticalBlock b(crit);
+
+    //Find which file should not be removed from ackedLogFileCheckList.
+    StringArray fileNamesToKeep;
+    for (auto r : ackedLogFileCheckList)
+    {
+        if (!fileNames.contains(r.c_str()))
+            fileNamesToKeep.append(r.c_str());
+    }
+
+    //Create a temp file with the fileNamesToKeep for replacing the settings->ackedFileList
+    VStringBuffer tempFileName("%s.tmp", settings->ackedFileList.str());
+    addNewAckedFileList(tempFileName, fileNamesToKeep);
+
+    //Replace the settings->ackedFileList with the temp file
+    renameFile(settings->ackedFileList, tempFileName, true);
+    PROGLOG("Rename %s to %s", tempFileName.str(), settings->ackedFileList.str());
+
+    //Create new ackedLogFileCheckList based on fileNamesToKeep
+    ackedLogFileCheckList.clear();
+    ForEachItemIn(j, fileNamesToKeep)
+    {
+        const char* name = fileNamesToKeep.item(j);
+        ackedLogFileCheckList.insert(name);
+        PROGLOG("Add %s to new ackedLogFileCheckList", name);
+    }
+}
+
 void CLogRequestReader::readAcked(const char* fileName, std::set<std::string>& acked)
 {
     Owned<IFile> f = createIFile(fileName);

esp/logging/logginglib/logthread.hpp

@@ -41,7 +41,16 @@ public:
 
 class CLogThread;
 
-class CLogRequestReader : public CInterface, implements IThreaded
+interface ILogRequestReader : extends IThreaded
+{
+    virtual CLogRequestReaderSettings* getSettings() = 0;
+    virtual void setPause(bool pause) = 0;
+    virtual void reportAckedLogFiles(StringArray& ackedLogFiles) = 0;
+    virtual void removeUnknownAckedLogFiles(StringArray& ackedLogFiles) = 0;
+    virtual void cleanAckedLogFiles(StringArray& fileNames) = 0;
+};
+
+class CLogRequestReader : public CInterface, implements ILogRequestReader
 {
     Owned<CLogRequestReaderSettings> settings;
     StringArray newAckedLogFiles;
@@ -69,6 +78,7 @@ class CLogRequestReader : public CInterface, implements IThreaded
     void addPendingLogsToQueue();
     void updateAckedFileList();
     void updateAckedLogRequestList();
+    void addNewAckedFileList(const char* list, StringArray& fileNames);
 
 public:
     CLogRequestReader(CLogRequestReaderSettings* _settings, CLogThread* _logThread)
@@ -82,11 +92,11 @@ public:
     virtual void threadmain() override;
 
     void addACK(const char* GUID);
-    virtual CLogRequestReaderSettings* getSettings() { return settings; };
-    void setPause(bool pause)
-    {
-        paused = pause;
-    };
+    virtual CLogRequestReaderSettings* getSettings() override { return settings; };
+    virtual void setPause(bool pause) override { paused = pause; };
+    virtual void reportAckedLogFiles(StringArray& ackedLogFiles) override;
+    virtual void removeUnknownAckedLogFiles(StringArray& ackedLogFiles) override;
+    virtual void cleanAckedLogFiles(StringArray& fileNames) override;
 };
 
 interface IUpdateLogThread : extends IInterface
@@ -101,7 +111,7 @@ interface IUpdateLogThread : extends IInterface
     virtual bool queueLog(IEspUpdateLogRequest* logRequest) = 0;
     virtual bool queueLog(IEspUpdateLogRequestWrap* logRequest) = 0;
     virtual void sendLog() = 0;
-    virtual CLogRequestReader* getLogRequestReader() = 0;
+    virtual ILogRequestReader* getLogRequestReader() = 0;
 };
 
 class CLogThread : public Thread , implements IUpdateLogThread

esp/scm/ws_decoupledlogging.ecm

@@ -17,6 +17,15 @@
 
 EspInclude(common);
 
+ESPenum LogAgentActions : string
+{
+    Pause("Pause"),
+    Resume("Resume"),
+    GetSettings("GetSettings"),
+    GetAckedLogFileNames("GetAckedLogFileNames"),
+    CleanAckedLogFiles("CleanAckedLogFiles"),
+};
+
 ESPStruct [nil_remove] LogAgentGroup
 {
     string GroupName;
@@ -55,6 +64,13 @@ ESPStruct [nil_remove] LogAgentGroupSetting
     ESParray<ESPstruct LogAgentSetting> AgentSettings;
 };
 
+ESPStruct [nil_remove] LogAgentGroupTankFiles
+{
+    string GroupName;
+    string TankFileDir;
+    ESParray<string> TankFileNames;
+};
+
 ESPrequest [nil_remove] GetLogAgentSettingRequest
 {
     ESParray<ESPstruct LogAgentGroup> Groups;
@@ -76,10 +92,32 @@ ESPresponse [exceptions_inline, nil_remove, http_encode(0)] PauseLogResponse
     ESParray<ESPstruct LogAgentGroupStatus> Statuses;
 };
 
+ESPrequest [nil_remove] GetAckedLogFilesRequest
+{
+    ESParray<ESPstruct LogAgentGroup> Groups;
+};
+
+ESPresponse [exceptions_inline, nil_remove, http_encode(0)] GetAckedLogFilesResponse
+{
+    ESParray<ESPstruct LogAgentGroupTankFiles> AckedLogFilesInGroups;
+};
+
+ESPrequest [nil_remove] CleanAckedFilesRequest
+{
+    string GroupName;
+    ESParray<string> FileNames;
+};
+
+ESPresponse [exceptions_inline, nil_remove, http_encode(0)] CleanAckedFilesResponse
+{
+};
+
 ESPservice [auth_feature("NONE"), exceptions_inline("./smc_xslt/exceptions.xslt")] WSDecoupledLog
 {
     ESPmethod GetLogAgentSetting(GetLogAgentSettingRequest, GetLogAgentSettingResponse);
     ESPmethod PauseLog(PauseLogRequest, PauseLogResponse);
+    ESPmethod GetAckedLogFiles(GetAckedLogFilesRequest, GetAckedLogFilesResponse);
+    ESPmethod CleanAckedFiles(CleanAckedFilesRequest, CleanAckedFilesResponse);
 };
 
 SCMexportdef(WSDecoupledLog);

esp/services/ws_decoupledlogging/ws_decoupledloggingservice.cpp

@@ -86,7 +86,7 @@ void CWSDecoupledLogEx::init(IPropertyTree* cfg, const char* process, const char
             if(!logThread)
                 throw MakeStringException(-1, "Failed to create update log thread for %s", agentName);
 
-            CLogRequestReader* logRequestReader = logThread->getLogRequestReader();
+            ILogRequestReader* logRequestReader = logThread->getLogRequestReader();
             if (!logRequestReader)
                 throw MakeStringException(-1, "CLogRequestReader not found for %s.", agentName);
 
@@ -100,40 +100,65 @@ void CWSDecoupledLogEx::init(IPropertyTree* cfg, const char* process, const char
 
 bool CWSDecoupledLogEx::onGetLogAgentSetting(IEspContext& context, IEspGetLogAgentSettingRequest& req, IEspGetLogAgentSettingResponse& resp)
 {
+    LogAgentAction action;
+    action.type = CLogAgentActions_GetSettings;
+
+    CLogAgentActionResults results;
+    WSDecoupledLogGetSettings act(action, results);
+    act.doAction(context, logGroups, req.getGroups());
+    resp.setSettings(results.queryGroupSettings());
+
+    return true;
+}
+
+bool CWSDecoupledLogEx::onPauseLog(IEspContext& context, IEspPauseLogRequest& req, IEspPauseLogResponse& resp)
+{
+    LogAgentAction action;
+    action.type = req.getPause() ? CLogAgentActions_Pause : CLogAgentActions_Resume;
+
+    CLogAgentActionResults results;
+    WSDecoupledLogPause act(action, results);
+    act.doAction(context, logGroups, req.getGroups());
+
+    resp.setStatuses(results.queryGroupStatus());
+
+    return true;
+}
+
+bool CWSDecoupledLogEx::onGetAckedLogFiles(IEspContext& context, IEspGetAckedLogFilesRequest& req, IEspGetAckedLogFilesResponse& resp)
+{
+    LogAgentAction action;
+    action.type = CLogAgentActions_GetAckedLogFileNames;
+
+    CLogAgentActionResults results;
+    WSDecoupledLogGetAckedLogFileNames act(action, results);
+    act.doAction(context, logGroups, req.getGroups());
+
+    resp.setAckedLogFilesInGroups(results.queryTankFilesInGroup());
+
+    return true;
+}
+
+bool CWSDecoupledLogEx::onCleanAckedFiles(IEspContext& context, IEspCleanAckedFilesRequest& req, IEspCleanAckedFilesResponse& resp)
+{
     try
     {
-        IArrayOf<IEspLogAgentGroupSetting> groupSettingResp;
-        IArrayOf<IConstLogAgentGroup>& groups = req.getGroups();
-        if (!groups.ordinality())
-        {
-            for (auto ml : logGroups)
-                getSettingsForLoggingAgentsInGroup(ml.second, nullptr, groupSettingResp);
-        }
-        else
-        {
-            ForEachItemIn(i, groups)
-            {
-                IConstLogAgentGroup& g = groups.item(i);
-                const char* gName = g.getGroupName();
-                if (isEmptyString(gName))
-                    throw MakeStringException(ECLWATCH_INVALID_INPUT, "Group name not specified.");
-
-                auto match = logGroups.find(gName);
-                if (match != logGroups.end())
-                {
-                    StringArray& agentNames = g.getAgentNames();
-                    getSettingsForLoggingAgentsInGroup(match->second, &agentNames, groupSettingResp);
-                }
-                else
-                {
-                    Owned<IEspLogAgentGroupSetting> groupSetting = createLogAgentGroupSetting();
-                    groupSetting->setGroupName(gName);
-                    groupSetting->setGroupStatus("NotFound");
-                    groupSettingResp.append(*groupSetting.getClear());
-                }
-            }
-        }
-        resp.setSettings(groupSettingResp);
+        const char* groupName = req.getGroupName();
+        if (isEmptyString(groupName))
+            throw makeStringException(ECLWATCH_INVALID_INPUT, "Group name not specified.");
+        auto match = logGroups.find(groupName);
+        if (match == logGroups.end())
+            throw MakeStringException(ECLWATCH_INVALID_INPUT, "Group %s not found.", groupName);
+
+        LogAgentAction action;
+        action.type = CLogAgentActions_CleanAckedLogFiles;
+        action.fileNames = &req.getFileNames();
+        if (!action.fileNames->length())
+            throw makeStringException(ECLWATCH_INVALID_INPUT, "File name not specified.");
+
+        CLogAgentActionResults results;
+        WSDecoupledLogCleanAckedLogFiles act(action, results);
+        act.doActionInGroup(match->second, nullptr);
     }
     catch(IException* e)
     {
@@ -142,157 +167,184 @@ bool CWSDecoupledLogEx::onGetLogAgentSetting(IEspContext& context, IEspGetLogAge
     return true;
 }
 
-bool CWSDecoupledLogEx::onPauseLog(IEspContext& context, IEspPauseLogRequest& req, IEspPauseLogResponse& resp)
+void WSDecoupledLogAction::doAction(IEspContext& context, std::map<std::string, Owned<WSDecoupledLogAgentGroup>>& allGroups,
+    IArrayOf<IConstLogAgentGroup>& groupsReq)
 {
     try
     {
-        IArrayOf<IEspLogAgentGroupStatus> groupStatusResp;
-        bool pause = req.getPause();
-        IArrayOf<IConstLogAgentGroup>& groups = req.getGroups();
-        if (!groups.ordinality())
+        if (groupsReq.ordinality())
         {
-            for (auto ml : logGroups)
-                pauseLoggingAgentsInGroup(ml.second, nullptr, pause, groupStatusResp);
+            checkGroupInput(allGroups, groupsReq);
+            ForEachItemIn(i, groupsReq)
+            {
+                IConstLogAgentGroup& g = groupsReq.item(i);
+                auto match = allGroups.find(g.getGroupName());
+                StringArray& agentNames = g.getAgentNames();
+                doActionInGroup(match->second, &agentNames);
+            }
         }
         else
         {
-            ForEachItemIn(i, groups)
+            for (auto ml : allGroups)
             {
-                IConstLogAgentGroup& g = groups.item(i);
-                const char* gName = g.getGroupName();
-                if (isEmptyString(gName))
-                    throw MakeStringException(ECLWATCH_INVALID_INPUT, "Group name not specified.");
-
-                auto match = logGroups.find(gName);
-                if (match != logGroups.end())
-                {
-                    StringArray& agentNames = g.getAgentNames();
-                    pauseLoggingAgentsInGroup(match->second, &agentNames, pause, groupStatusResp);
-                }
-                else
-                {
-                    Owned<IEspLogAgentGroupStatus> groupStatus = createLogAgentGroupStatus();
-                    groupStatus->setGroupName(gName);
-                    groupStatus->setGroupStatus("NotFound");
-                    groupStatusResp.append(*groupStatus.getClear());
-                }
+                doActionInGroup(ml.second, nullptr);
             }
         }
-        resp.setStatuses(groupStatusResp);
     }
     catch(IException* e)
     {
         FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
     }
-    return true;
 }
 
-void CWSDecoupledLogEx::getSettingsForLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames,
-    IArrayOf<IEspLogAgentGroupSetting>& groupSettingResp)
+void WSDecoupledLogAction::checkGroupInput(std::map<std::string, Owned<WSDecoupledLogAgentGroup>>& allGroups,
+    IArrayOf<IConstLogAgentGroup>& groupsReq)
+{
+    ForEachItemIn(i, groupsReq)
+    {
+        IConstLogAgentGroup& g = groupsReq.item(i);
+        const char* gName = g.getGroupName();
+        if (isEmptyString(gName))
+            throw makeStringException(ECLWATCH_INVALID_INPUT, "Group name not specified.");
+
+        auto match = allGroups.find(gName);
+        if (match == allGroups.end())
+            throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Group %s not found.", gName);
+
+        StringArray& agentNames = g.getAgentNames();
+        ForEachItemIn(j, agentNames)
+        {
+            const char* agentName = agentNames.item(j);
+            if (isEmptyString(agentName))
+                throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "%s: logging agent name not specified.", gName);
+
+            if (!match->second->getLoggingAgentThread(agentName))
+                throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "%s: logging agent %s not found.", gName, agentName);
+        }
+    }
+}
+
+void WSDecoupledLogAction::doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames)
 {
-    IArrayOf<IEspLogAgentSetting> agentSettingResp;
     if (!agentNames || !agentNames->ordinality())
-        getSettingsForAllLoggingAgentsInGroup(group, agentSettingResp);
+    {
+        std::map<std::string, Owned<IUpdateLogThread>>& agentThreadMap = group->getLoggingAgentThreads();
+        for (auto mt : agentThreadMap)
+        {
+            if (!doActionForAgent(mt.first.c_str(), mt.second))
+                break;
+        }
+    }
     else
     {
         ForEachItemIn(j, *agentNames)
         {
             const char* agentName = agentNames->item(j);
-            if (isEmptyString(agentName))
-                throw MakeStringException(ECLWATCH_INVALID_INPUT, "%s: logging agent name not specified.", group->getName());
-
-            getLoggingAgentSettings(agentName, group->getLoggingAgentThread(agentName), agentSettingResp);
+            if (!doActionForAgent(agentName, group->getLoggingAgentThread(agentName)))
+                break;
         }
     }
+}
 
-    Owned<IEspLogAgentGroupSetting> groupSetting = createLogAgentGroupSetting();
+void WSDecoupledLogGetSettings::doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames)
+{
+    groupSetting.setown(createLogAgentGroupSetting());
     groupSetting->setGroupName(group->getName());
-    groupSetting->setGroupStatus("Found");
     const char* tankFileDir = group->getTankFileDir();
     const char* tankFileMask = group->getTankFileMask();
     groupSetting->setTankFileDir(tankFileDir);
     if (!isEmptyString(tankFileMask))
         groupSetting->setTankFileMask(tankFileMask);
-    groupSetting->setAgentSettings(agentSettingResp);
-    groupSettingResp.append(*groupSetting.getClear());
+
+    WSDecoupledLogAction::doActionInGroup(group, agentNames);
+
+    results.appendGroupSetting(groupSetting.getClear());
 }
 
-void CWSDecoupledLogEx::pauseLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames, bool pause,
-    IArrayOf<IEspLogAgentGroupStatus>& groupStatusResp)
+bool WSDecoupledLogGetSettings::doActionForAgent(const char* agentName, IUpdateLogThread* agentThread)
 {
-    IArrayOf<IEspLogAgentStatus> agentStatusResp;
-    if (!agentNames || !agentNames->ordinality())
-        pauseAllLoggingAgentsInGroup(group, pause, agentStatusResp);
+    Owned<IEspLogAgentSetting> agentSetting = createLogAgentSetting();
+    agentSetting->setAgentName(agentName);
+
+    CLogRequestReaderSettings* settings = agentThread->getLogRequestReader()->getSettings();
+    if (!settings)
+        agentSetting->setAgentStatus("SettingsNotFound");
     else
     {
-        ForEachItemIn(j, *agentNames)
-        {
-            const char* agentName = agentNames->item(j);
-            if (isEmptyString(agentName))
-                throw MakeStringException(ECLWATCH_INVALID_INPUT, "%s: logging agent name not specified.", group->getName());
-
-            pauseLoggingAgent(agentName, group->getLoggingAgentThread(agentName), pause, agentStatusResp);
-        }
+        agentSetting->setAgentStatus("SettingsFound");
+        agentSetting->setAckedFileList(settings->ackedFileList);
+        agentSetting->setAckedLogRequestFile(settings->ackedLogRequestFile);
+        agentSetting->setWaitSeconds(settings->waitSeconds);
+        agentSetting->setPendingLogBufferSize(settings->pendingLogBufferSize);
     }
 
-    Owned<IEspLogAgentGroupStatus> groupStatus = createLogAgentGroupStatus();
-    groupStatus->setGroupName(group->getName());
-    groupStatus->setGroupStatus("Found");
-    groupStatus->setAgentStatuses(agentStatusResp);
-    groupStatusResp.append(*groupStatus.getClear());
+    IArrayOf<IConstLogAgentSetting>& agentSettings = groupSetting->getAgentSettings();
+    agentSettings.append(*agentSetting.getClear());
+    return true;
 }
 
-void CWSDecoupledLogEx::pauseAllLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, bool pause, IArrayOf<IEspLogAgentStatus>& agentStatusResp)
+void WSDecoupledLogPause::doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames)
 {
-    std::map<std::string, Owned<IUpdateLogThread>>&  agentThreadMap = group->getLoggingAgentThreads();
-    for (auto mt : agentThreadMap)
-        pauseLoggingAgent(mt.first.c_str(), mt.second, pause, agentStatusResp);
+    groupStatus.setown(createLogAgentGroupStatus());
+    groupStatus->setGroupName(group->getName());
+
+    WSDecoupledLogAction::doActionInGroup(group, agentNames);
+
+    results.appendGroupStatus(groupStatus.getClear());
 }
 
-void CWSDecoupledLogEx::pauseLoggingAgent(const char* agentName, IUpdateLogThread* agentThread, bool pause, IArrayOf<IEspLogAgentStatus>& agentStatusResp)
+bool WSDecoupledLogPause::doActionForAgent(const char* agentName, IUpdateLogThread* agentThread)
 {
-    Owned<IEspLogAgentStatus> agentStatus = createLogAgentStatus();
-    agentStatus->setAgentName(agentName);
+    agentThread->getLogRequestReader()->setPause((action.type == CLogAgentActions_Pause) ? true : false);
 
-    if (!agentThread)
-        agentStatus->setStatus("NotFound");
+    IArrayOf<IConstLogAgentStatus>& agentStatusInGroup = groupStatus->getAgentStatuses();
+    Owned<IEspLogAgentStatus> aStatus = createLogAgentStatus();
+    aStatus->setAgentName(agentName);
+    if (action.type == CLogAgentActions_Pause)
+        aStatus->setStatus("Pausing");
     else
-    {
-        agentThread->getLogRequestReader()->setPause(pause);
-        agentStatus->setStatus(pause ? "Pausing" : "Resuming");
-    }
-    agentStatusResp.append(*agentStatus.getClear());
+        aStatus->setStatus("Resuming");
+    agentStatusInGroup.append(*aStatus.getClear());
+    return true;
 }
 
-void CWSDecoupledLogEx::getSettingsForAllLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, IArrayOf<IEspLogAgentSetting>& agentSettingResp)
+void WSDecoupledLogGetAckedLogFileNames::doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames)
 {
-    std::map<std::string, Owned<IUpdateLogThread>>&  agentThreadMap = group->getLoggingAgentThreads();
-    for (auto mt : agentThreadMap)
-        getLoggingAgentSettings(mt.first.c_str(), mt.second, agentSettingResp);
+    tankFilesInGroup.setown(createLogAgentGroupTankFiles());
+    tankFilesInGroup->setGroupName(group->getName());
+    tankFilesInGroup->setTankFileDir(group->getTankFileDir());
+
+    WSDecoupledLogAction::doActionInGroup(group, agentNames);
+
+    results.appendGroupTankFiles(tankFilesInGroup.getClear());
 }
 
-void CWSDecoupledLogEx::getLoggingAgentSettings(const char* agentName, IUpdateLogThread* agentThread, IArrayOf<IEspLogAgentSetting>& agentSettingResp)
+bool WSDecoupledLogGetAckedLogFileNames::doActionForAgent(const char* agentName, IUpdateLogThread* agentThread)
 {
-    Owned<IEspLogAgentSetting> agentSetting = createLogAgentSetting();
-    agentSetting->setAgentName(agentName);
-
-    if (!agentThread)
-        agentSetting->setAgentStatus("NotFound");
+    StringArray& ackedFiles = tankFilesInGroup->getTankFileNames();
+    //The ackedFiles stores the tank files which have been acked for all logging agents
+    //in an agent group. At the beginning, it is empty. The reportAckedLogFiles() will
+    //be called for the 1st logging agent. The ackedFiles is filled with the acked tank
+    //files in the 1st logging agent. If the ackedFiles is still empty, this method
+    //returns false and the outside loop for other logging agents in the group will be
+    //stopped. If the ackedFiles is not empty, the outside loop calls this method for
+    //the rest of logging agents in the group. For those logging agents, the
+    //removeUnknownAckedLogFiles() will be called because ackedFiles.length() != 0.
+    //In the removeUnknownAckedLogFiles(), if any file inside the ackedFiles has not
+    //been acked in that agent, the file should be removed from the ackedFiles. After
+    //the removeUnknownAckedLogFiles() call, if the ackedFiles is empty, the outside
+    //loop for the rest of logging agents in the group will be stopped.
+    if (!ackedFiles.length())
+        agentThread->getLogRequestReader()->reportAckedLogFiles(ackedFiles);
     else
-    {
-        CLogRequestReaderSettings* settings = agentThread->getLogRequestReader()->getSettings();
-        if (!settings)
-            agentSetting->setAgentStatus("SettingsNotFound");
-        else
-        {
-            agentSetting->setAgentStatus("Found");
-            agentSetting->setAckedFileList(settings->ackedFileList);
-            agentSetting->setAckedLogRequestFile(settings->ackedLogRequestFile);
-            agentSetting->setWaitSeconds(settings->waitSeconds);
-            agentSetting->setPendingLogBufferSize(settings->pendingLogBufferSize);
-        }
-    }
-    agentSettingResp.append(*agentSetting.getClear());
+        agentThread->getLogRequestReader()->removeUnknownAckedLogFiles(ackedFiles);
+    return !ackedFiles.empty();
+}
+
+bool WSDecoupledLogCleanAckedLogFiles::doActionForAgent(const char* agentName, IUpdateLogThread* agentThread)
+{
+    agentThread->getLogRequestReader()->cleanAckedLogFiles(*action.fileNames);
+    return true;
 }
 
 IUpdateLogThread* WSDecoupledLogAgentGroup::getLoggingAgentThread(const char* name)

esp/services/ws_decoupledlogging/ws_decoupledloggingservice.hpp

@@ -23,6 +23,29 @@
 #include "environment.hpp"
 #include "logthread.hpp"
 
+struct LogAgentAction
+{
+    CLogAgentActions type;
+    StringArray* fileNames = nullptr;
+};
+
+class CLogAgentActionResults : public CSimpleInterfaceOf<IInterface>
+{
+    IArrayOf<IEspLogAgentGroupStatus> groupStatus;
+    IArrayOf<IEspLogAgentGroupSetting> groupSettings;
+    IArrayOf<IEspLogAgentGroupTankFiles> tankFilesInGroups;
+
+public:
+    CLogAgentActionResults() {};
+
+    IArrayOf<IEspLogAgentGroupStatus>& queryGroupStatus() { return groupStatus; }
+    void appendGroupStatus(IEspLogAgentGroupStatus* status) { groupStatus.append(*status); }
+    IArrayOf<IEspLogAgentGroupSetting>& queryGroupSettings() { return groupSettings; }
+    void appendGroupSetting(IEspLogAgentGroupSetting* setting) { groupSettings.append(*setting); }
+    IArrayOf<IEspLogAgentGroupTankFiles>& queryTankFilesInGroup() { return tankFilesInGroups; }
+    void appendGroupTankFiles(IEspLogAgentGroupTankFiles* groupTankFiles) { tankFilesInGroups.append(*groupTankFiles); }
+};
+
 class CWSDecoupledLogSoapBindingEx : public CWSDecoupledLogSoapBinding
 {
 public:
@@ -52,6 +75,69 @@ public:
     IUpdateLogThread* getLoggingAgentThread(const char* name);
 };
 
+class WSDecoupledLogAction : public CSimpleInterfaceOf<IInterface>
+{
+    void checkGroupInput(std::map<std::string, Owned<WSDecoupledLogAgentGroup>>& allGroups, IArrayOf<IConstLogAgentGroup>& groupsReq);
+
+protected:
+    LogAgentAction& action;
+    CLogAgentActionResults& results;
+
+public:
+    WSDecoupledLogAction(LogAgentAction& _action, CLogAgentActionResults& _results)
+        : action(_action), results(_results) {}
+
+    void doAction(IEspContext& context, std::map<std::string, Owned<WSDecoupledLogAgentGroup>>& _allGroups,
+        IArrayOf<IConstLogAgentGroup>& _groups);
+    virtual bool doActionForAgent(const char* agentName, IUpdateLogThread* agentThread) = 0;
+    virtual void doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames);
+};
+
+class WSDecoupledLogGetSettings : public WSDecoupledLogAction
+{
+    Owned<IEspLogAgentGroupSetting> groupSetting;
+
+public:
+    WSDecoupledLogGetSettings(LogAgentAction& _action, CLogAgentActionResults& _results)
+        : WSDecoupledLogAction(_action, _results) {}
+
+    virtual bool doActionForAgent(const char* agentName, IUpdateLogThread* agentThread);
+    virtual void doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames);
+};
+
+class WSDecoupledLogPause : public WSDecoupledLogAction
+{
+    Owned<IEspLogAgentGroupStatus> groupStatus;
+
+public:
+    WSDecoupledLogPause(LogAgentAction& _action, CLogAgentActionResults& _results)
+        : WSDecoupledLogAction(_action, _results) {}
+
+    virtual bool doActionForAgent(const char* agentName, IUpdateLogThread* agentThread);
+    virtual void doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames);
+};
+
+class WSDecoupledLogGetAckedLogFileNames : public WSDecoupledLogAction
+{
+    Owned<IEspLogAgentGroupTankFiles> tankFilesInGroup;
+
+public:
+    WSDecoupledLogGetAckedLogFileNames(LogAgentAction& _action, CLogAgentActionResults& _results)
+        : WSDecoupledLogAction(_action, _results) {}
+
+    virtual bool doActionForAgent(const char* agentName, IUpdateLogThread* agentThread);
+    virtual void doActionInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames);
+};
+
+class WSDecoupledLogCleanAckedLogFiles : public WSDecoupledLogAction
+{
+public:
+    WSDecoupledLogCleanAckedLogFiles(LogAgentAction& _action, CLogAgentActionResults& _results)
+        : WSDecoupledLogAction(_action, _results) {}
+
+    virtual bool doActionForAgent(const char* agentName, IUpdateLogThread* agentThread);
+};
+
 class CWSDecoupledLogEx : public CWSDecoupledLog
 {
     StringAttr espProcess;
@@ -59,14 +145,6 @@ class CWSDecoupledLogEx : public CWSDecoupledLog
     std::map<std::string, Owned<WSDecoupledLogAgentGroup>> logGroups;
 
     IEspLogAgent* loadLoggingAgent(const char* name, const char* dll, const char* service, IPropertyTree* cfg);
-    void pauseLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames, bool pause,
-        IArrayOf<IEspLogAgentGroupStatus>& groupStatusResp);
-    void pauseAllLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, bool pause, IArrayOf<IEspLogAgentStatus>& agentStatusResp);
-    void getSettingsForLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, StringArray* agentNames,
-        IArrayOf<IEspLogAgentGroupSetting>& groupSettingResp);
-    void getSettingsForAllLoggingAgentsInGroup(WSDecoupledLogAgentGroup* group, IArrayOf<IEspLogAgentSetting>& agentSettingResp);
-    void pauseLoggingAgent(const char* agentName, IUpdateLogThread* agentThread, bool pause, IArrayOf<IEspLogAgentStatus>& agentStatusResp);
-    void getLoggingAgentSettings(const char* agentName, IUpdateLogThread* agentThread, IArrayOf<IEspLogAgentSetting>& agentSettingResp);
 
 public:
     IMPLEMENT_IINTERFACE;
@@ -79,6 +157,8 @@ public:
     virtual void init(IPropertyTree* cfg, const char* process, const char* service);
     virtual bool onGetLogAgentSetting(IEspContext& context, IEspGetLogAgentSettingRequest& req, IEspGetLogAgentSettingResponse& resp);
     virtual bool onPauseLog(IEspContext& context, IEspPauseLogRequest& req, IEspPauseLogResponse& resp);
+    virtual bool onGetAckedLogFiles(IEspContext& context, IEspGetAckedLogFilesRequest& req, IEspGetAckedLogFilesResponse& resp);
+    virtual bool onCleanAckedFiles(IEspContext& context, IEspCleanAckedFilesRequest& req, IEspCleanAckedFilesResponse& resp);
 };
 
 #endif //_ESPWIZ_ws_decoupledlogging_HPP__

esp/tools/CMakeLists.txt

@@ -13,4 +13,5 @@
 #    See the License for the specific language governing permissions and
 #    limitations under the License.
 ################################################################################
-add_subdirectory (soapplus)
+add_subdirectory (soapplus)
+add_subdirectory (tankfile)

esp/tools/tankfile/CMakeLists.txt

@@ -0,0 +1,70 @@
+################################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+################################################################################
+
+# Component: tankfile 
+
+#####################################################
+# Description:
+# ------------
+#    Cmake Input File for tankfile
+#####################################################
+
+
+project( tankfile )
+
+include(${HPCC_SOURCE_DIR}/esp/scm/espscm.cmake)
+
+set (    SRCS
+         archive.cpp
+         main.cpp
+         ${ESPSCM_GENERATED_DIR}/common_esp.cpp
+         ${ESPSCM_GENERATED_DIR}/ws_decoupledlogging_esp.cpp 
+    )
+
+include_directories (
+         ${CMAKE_BINARY_DIR}
+         ${CMAKE_BINARY_DIR}/oss
+         ${HPCC_SOURCE_DIR}/system/include
+         ${HPCC_SOURCE_DIR}/system/jlib
+         ${HPCC_SOURCE_DIR}/system/security/shared
+         ${HPCC_SOURCE_DIR}/system/security/securesocket
+         ${HPCC_SOURCE_DIR}/system/xmllib
+         ${HPCC_SOURCE_DIR}/esp/platform
+         ${HPCC_SOURCE_DIR}/esp/bindings
+         ${HPCC_SOURCE_DIR}/esp/bindings/SOAP/xpp
+         ${HPCC_SOURCE_DIR}/esp/clients
+    )
+
+ADD_DEFINITIONS( -D_CONSOLE )
+
+HPCC_ADD_EXECUTABLE ( tankfile ${SRCS} )
+add_dependencies ( tankfile espscm )
+install ( TARGETS tankfile RUNTIME DESTINATION ${EXEC_DIR} COMPONENT Runtime)
+
+target_link_libraries ( tankfile
+         jlib
+         esphttp
+    )
+IF (USE_OPENSSL)
+    target_link_libraries ( tankfile
+         securesocket
+    )
+ENDIF()
+
+if ( PLATFORM )
+    install ( PROGRAMS tankfile.install DESTINATION etc/init.d/install COMPONENT Runtime )
+    install ( PROGRAMS tankfile.uninstall DESTINATION etc/init.d/uninstall COMPONENT Runtime )
+endif()

esp/tools/tankfile/README.txt

@@ -0,0 +1,25 @@
+/*##############################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+############################################################################## */
+
+Usage:
+    tankfile action=[archive] [<options>]
+
+options:
+    server=<[http|https]://host:port> : log server url
+    user=<username>
+    password=<password>
+    group=<log-agent-group-name>
+    dir=<directory-name> : directory to archive the tank files

esp/tools/tankfile/archive.cpp

@@ -0,0 +1,216 @@
+/*##############################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+##############################################################################
+ */
+
+#pragma warning(disable:4786)
+
+#include "archive.hpp"
+
+CArchiveTankFileHelper::CArchiveTankFileHelper(IProperties* input)
+{
+    logServer.set(input->queryProp("server"));
+    if (logServer.isEmpty())
+        throw makeStringException(-1, "Please specify url.");
+    groupName.set(input->queryProp("group"));
+    archiveToDir.set(input->queryProp("dir"));
+    if (archiveToDir.isEmpty())
+        throw makeStringException(-1, "Please specify archive dir.");
+    client.setown(createLogServerClient(input));
+}
+
+void CArchiveTankFileHelper::archive()
+{
+    printf("Starting archive() ...\n");
+    Owned<IClientGetAckedLogFilesResponse> resp = getAckedLogFiles();
+    IArrayOf<IConstLogAgentGroupTankFiles>& ackedLogFilesInGroups = resp->getAckedLogFilesInGroups();
+
+    //Check any invalid response and print out the files to be archived just in case
+    if (!validateAckedLogFiles(ackedLogFilesInGroups))
+    {
+        printf("Found invalid response. No file will be archived.\n");
+        return;
+    }
+
+    ForEachItemIn(i, ackedLogFilesInGroups)
+    {
+        IConstLogAgentGroupTankFiles& ackedLogFilesInGroup = ackedLogFilesInGroups.item(i);
+        StringArray& ackedLogFiles = ackedLogFilesInGroup.getTankFileNames();
+        if (!ackedLogFiles.length())
+            continue;
+
+        const char* groupName = ackedLogFilesInGroup.getGroupName();
+        const char* tankFileDir = ackedLogFilesInGroup.getTankFileDir();
+        printf("%s: archiving files from %s to %s.\n", groupName, tankFileDir, archiveToDir.str());
+        archiveAckedLogFilesForLogAgentGroup(groupName, tankFileDir, ackedLogFiles);
+        printf("%s: files archived.\n", groupName);
+        cleanAckedLogFilesForLogAgentGroup(groupName, ackedLogFiles);
+        printf("%s: files cleaned.\n", groupName);
+    }
+    printf("The archive() done\n");
+}
+
+IClientGetAckedLogFilesResponse* CArchiveTankFileHelper::getAckedLogFiles()
+{
+    Owned<IClientGetAckedLogFilesRequest> req = client->createGetAckedLogFilesRequest();
+    if (!groupName.isEmpty())
+    {
+        IArrayOf<IEspLogAgentGroup> groups;
+        Owned<IEspLogAgentGroup> group = createLogAgentGroup();
+        group->setGroupName(groupName.get());
+        groups.append(*group.getClear());
+        req->setGroups(groups);
+    }
+    Owned<IClientGetAckedLogFilesResponse> resp = client->GetAckedLogFiles(req);
+    if (!resp)
+        throw makeStringException(-1, "Failed in GetAckedLogFiles.");
+
+    const IMultiException* excep = &resp->getExceptions();
+    if (excep != nullptr && excep->ordinality() > 0)
+    {
+        StringBuffer msg;
+        printf("%s\n", excep->errorMessage(msg).str());
+        throw makeStringException(-1, "Cannot archiveTankFiles.");
+    }
+    return resp.getClear();
+}
+
+bool CArchiveTankFileHelper::validateAckedLogFiles(IArrayOf<IConstLogAgentGroupTankFiles>& ackedLogFilesInGroups)
+{
+    bool invalidResponse = false;
+    ForEachItemIn(i, ackedLogFilesInGroups)
+    {
+        IConstLogAgentGroupTankFiles& ackedLogFilesInGroup = ackedLogFilesInGroups.item(i);
+        const char* groupName = ackedLogFilesInGroup.getGroupName();
+        if (isEmptyString(groupName))
+        {
+            printf("Empty group name in GetAckedLogFilesResponse.\n");
+            invalidResponse = true;
+            continue;
+        }
+        const char* tankFileDir = ackedLogFilesInGroup.getTankFileDir();
+        if (isEmptyString(tankFileDir))
+        {
+            printf("Empty TankFile directory for group %s.\n", groupName);
+            invalidResponse = true;
+            continue;
+        }
+
+        printf("Group %s, TankFile directory %s:\n", groupName, tankFileDir);
+        StringArray& ackedLogFiles = ackedLogFilesInGroup.getTankFileNames();
+        if (!ackedLogFiles.length())
+        {
+            printf("No TankFile to be archived for %s.\n", groupName);
+        }
+        else
+        {
+            ForEachItemIn(i, ackedLogFiles)
+                printf("TankFile to be archived: %s.\n", ackedLogFiles.item(i));
+        }
+    }
+    return !invalidResponse;
+}
+
+void CArchiveTankFileHelper::archiveAckedLogFilesForLogAgentGroup(const char* groupName, const char* archiveFromDir, StringArray& ackedLogFiles)
+{
+    Owned<IFile> archiveToFolder = createIFile(archiveToDir);
+    if (!archiveToFolder->exists())
+        archiveToFolder->createDirectory();
+
+    StringBuffer timeStr;
+    ForEachItemIn(i, ackedLogFiles)
+    {
+        const char* ackedFileName = ackedLogFiles.item(i);
+        printf("Archiving %s.\n", ackedFileName);
+
+        StringBuffer ackedFile(archiveFromDir), archivedFile(archiveToDir);
+        addPathSepChar(ackedFile);
+        addPathSepChar(archivedFile);
+        ackedFile.append(ackedFileName);
+        archivedFile.append(ackedFileName);
+
+        Owned<IFile> srcfile = createIFile(ackedFile);
+        Owned<IFile> dstFile = createIFile(archivedFile);
+        if (dstFile->exists())
+        {
+            StringBuffer newName;
+            if (timeStr.isEmpty())
+            {
+                CDateTime now;
+                now.setNow();
+                now.getString(timeStr);
+            }
+            newName.append(archivedFile).append(".").append(timeStr);
+            dstFile->move(newName);
+            printf("Old %s is moved to %s.\n", archivedFile.str(), newName.str());
+        }
+        srcfile->move(archivedFile);
+        printf("%s is archived to %s.\n", ackedFile.str(), archivedFile.str());
+    }
+}
+
+void CArchiveTankFileHelper::cleanAckedLogFilesForLogAgentGroup(const char* groupName, StringArray& ackedLogFiles)
+{
+    Owned<IClientCleanAckedFilesRequest> req = client->createCleanAckedFilesRequest();
+    req->setGroupName(groupName);
+    req->setFileNames(ackedLogFiles);
+
+    Owned<IClientCleanAckedFilesResponse> resp = client->CleanAckedFiles(req);
+    if (!resp)
+        throw makeStringException(-1, "Failed in CleanAckedFiles.");
+
+    const IMultiException* excep = &resp->getExceptions();
+    if (excep != nullptr && excep->ordinality() > 0)
+    {
+        StringBuffer msg;
+        printf("%s\n", excep->errorMessage(msg).str());
+    }
+}
+
+IClientWSDecoupledLog* createLogServerClient(IProperties* input)
+{
+    Owned<IClientWSDecoupledLog> client = createWSDecoupledLogClient();
+
+    const char* server = input->queryProp("server");
+    if (isEmptyString(server))
+        throw MakeStringException(0, "Server url not defined");
+
+    StringBuffer url(server);
+    addPathSepChar(url);
+    url.append("WSDecoupledLog");
+    client->addServiceUrl(url.str());
+    const char* user = input->queryProp("user");
+    const char* password = input->queryProp("password");
+    if (!isEmptyString(user))
+        client->setUsernameToken(user, password, nullptr);
+
+    return client.getClear();
+}
+
+void archiveTankFiles(IProperties* input)
+{
+    try
+    {
+        printf("Starting archiveTankFiles\n");
+        Owned<CArchiveTankFileHelper> helper = new CArchiveTankFileHelper(input);
+        helper->archive();
+        printf("Finished archiveTankFiles\n");
+    }
+    catch (IException *e)
+    {
+        EXCLOG(e);
+        e->Release();
+    }
+}

esp/tools/tankfile/archive.hpp

@@ -0,0 +1,48 @@
+/*##############################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+##############################################################################
+ */
+
+#ifndef _ARCHIVE_HPP__
+#define _ARCHIVE_HPP__
+
+#include "jstring.hpp"
+#include "jptree.hpp"
+#include "jtime.hpp"
+#include "jfile.hpp"
+#include "jutil.hpp"
+#include "ws_decoupledlogging.hpp"
+
+class CArchiveTankFileHelper : public CSimpleInterface
+{
+    Owned<IClientWSDecoupledLog> client;
+    StringAttr logServer, groupName;
+    StringBuffer archiveToDir;
+
+    IClientGetAckedLogFilesResponse* getAckedLogFiles();
+    bool validateAckedLogFiles(IArrayOf<IConstLogAgentGroupTankFiles>& ackedLogFilesInGroups);
+    void archiveAckedLogFilesForLogAgentGroup(const char* groupName, const char* tankFileDir, StringArray& ackedLogFiles);
+    void cleanAckedLogFilesForLogAgentGroup(const char* groupName, StringArray& ackedLogFiles);
+
+public:
+    CArchiveTankFileHelper(IProperties* input);
+
+    void archive();
+};
+
+IClientWSDecoupledLog* createLogServerClient(IProperties* input);
+void archiveTankFiles(IProperties* input);
+
+#endif

esp/tools/tankfile/main.cpp

@@ -0,0 +1,112 @@
+/*##############################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+##############################################################################
+ */
+
+#include "jliball.hpp"
+#include "archive.hpp"
+
+const char* version = "1.0";
+
+void usage()
+{
+    printf("Tankfile version %s.\n\nUsage:\n", version);
+    printf("  tankfile action=[archive] [<options>]\n\n");
+    printf("options:\n");
+    printf("    server=<[http|https]://host:port> : log server url\n");
+    printf("    user=<username>\n");
+    printf("    password=<password>\n");
+    printf("    group=<log-agent-group-name>\n");
+    printf("    dir=<directory-name> : directory to archive the tank files\n");
+}
+
+bool getInput(int argc, const char* argv[], IProperties* input)
+{
+    for (unsigned i = 1; i < argc; i++)
+    {
+        const char* arg = argv[i];
+        if (strchr(arg, '='))
+            input->loadProp(arg);
+        else if (strieq(argv[i], "-v") || strieq(argv[i], "-version"))
+        {
+            printf("Tankfile version %s\n", version);
+            return false;
+        }
+        else if (strieq(arg, "-h") || strieq(arg, "-?"))
+        {
+            usage();
+            return false;
+        }
+        else
+        {
+            printf("Error: unknown command parameter: %s\n", argv[i]);
+            usage();
+            return false;
+        }
+    }
+    return true;
+}
+
+bool processRequest(IProperties* input)
+{
+    const char* action = input->queryProp("action");
+    if (isEmptyString(action))
+    {
+        printf("Error: 'action' not specified\n");
+        return false;
+    }
+
+    printf("Tankfile version %s: %s", version, action);
+    if (strieq(action, "archive"))
+    {
+        archiveTankFiles(input);
+    }
+    else
+    {
+        printf("Error: unknown 'action': %s\n", action);
+        return false;
+    }
+
+    return true;
+}
+
+int main(int argc, const char** argv)
+{
+    InitModuleObjects();
+
+    Owned<IProperties> input = createProperties(true);
+    if (!getInput(argc, argv, input))
+    {
+        releaseAtoms();
+        return 0;
+    }
+
+    try
+    {
+        processRequest(input);
+    }
+    catch(IException *excpt)
+    {
+        StringBuffer errMsg;
+        printf("Exception: %d:%s\n", excpt->errorCode(), excpt->errorMessage(errMsg).str());
+    }
+    catch(...)
+    {
+        printf("Unknown exception\n");
+    }
+    releaseAtoms();
+
+    return 0;
+}

esp/tools/tankfile/sourcedoc.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+################################################################################
+#    HPCC SYSTEMS software Copyright (C) 2020 HPCC Systems®.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License");
+#    you may not use this file except in compliance with the License.
+#    You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS,
+#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#    See the License for the specific language governing permissions and
+#    limitations under the License.
+################################################################################
+-->
+
+<!DOCTYPE section PUBLIC "-//OASIS//DTD DocBook XML V4.3//EN" "http://www.oasis-open.org/docbook/xml/4.3/docbookx.dtd">
+<section>
+    <title>esp/tools/tankfile</title>
+
+    <para>
+        The esp/tools/tankfile directory contains the sources for the esp/tools/tankfile library.
+    </para>
+</section>

esp/tools/tankfile/tankfile.install

@@ -0,0 +1 @@
+installFile "$binPath/tankfile" "/usr/bin/tankfile" 1 || exit 1

esp/tools/tankfile/tankfile.uninstall

@@ -0,0 +1 @@
+removeSymlink "/usr/bin/tankfile"