Merge branch 'candidate-8.4.x'

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

ecllibrary/std/system/Store.ecl

@@ -419,7 +419,7 @@ EXPORT Store(STRING username = '',
                     TRANSFORM
                         (
                             RECORDOF(LEFT),
-                            SELF.was_found := NOT EXISTS(LEFT.exceptions.exceptions),
+                            SELF.was_found := LEFT.value != '' AND NOT EXISTS(LEFT.exceptions.exceptions),
                             SELF := LEFT
                         )
                 );

ecllibrary/teststd/system/TestStore.ecl

@@ -15,17 +15,23 @@ EXPORT TestStore := MODULE
     SHARED KEY_1 := 'some_key';
     SHARED VALUE_1 := 'fubar';
 
+    SHARED KEY_2 := 'another_key';
+    SHARED VALUE_2 := '';
+
     SHARED kvStore := Std.System.Store(USER_NAME, USER_PW, ESP);
     SHARED namedKVStore := kvStore.WithNamespace(STORE_NAMESPACE, STORE_NAME);
 
     SHARED createStoreRes := kvStore.CreateStore(STORE_NAME);
     SHARED listStoresRes := kvStore.ListStores();
     SHARED listNamespacesRes := kvStore.ListNamespaces(STORE_NAME);
+    SHARED setKeyEmptyValueRes := namedKVStore.SetKeyValue(KEY_2, VALUE_2);
+    SHARED getKeyEmptyValueRes := namedKVStore.GetKeyValue(KEY_2);
     SHARED setKeyValueRes := namedKVStore.SetKeyValue(KEY_1, VALUE_1);
     SHARED getKeyValueRes := namedKVStore.GetKeyValue(KEY_1);
     SHARED getAllKeyValuesRes := namedKVStore.GetAllKeyValues();
     SHARED getAllKeysRes := namedKVStore.GetAllKeys();
     SHARED deleteKeyValueRes := namedKVStore.DeleteKeyValue(KEY_1);
+    SHARED deleteKeyValue2Res := namedKVStore.DeleteKeyValue(KEY_2);
     SHARED deleteNamespaceRes := namedKVStore.DeleteNamespace();
 
     // Using SEQUENTIAL to reuse the above definitions in a stateful manner
@@ -40,6 +46,9 @@ EXPORT TestStore := MODULE
             // Set a new key/value
             ASSERT(setKeyValueRes.succeeded);
 
+            // Set a new key/value (empty)
+            ASSERT(setKeyEmptyValueRes.succeeded);
+
             // Ensure our namespace exists
             ASSERT(EXISTS(listNamespacesRes.namespaces(namespace = STORE_NAMESPACE)));
 
@@ -47,6 +56,10 @@ EXPORT TestStore := MODULE
             ASSERT(getKeyValueRes.was_found = TRUE);
             ASSERT(getKeyValueRes.value = VALUE_1);
 
+            // Fetch empty string value for a key
+            ASSERT(getKeyEmptyValueRes.was_found = FALSE);
+            ASSERT(getKeyEmptyValueRes.value = VALUE_2);
+
             // Fetch all key/values in a namespace
             ASSERT(getAllKeysRes.namespace = STORE_NAMESPACE);
             ASSERT(EXISTS(getAllKeysRes.keys(key = KEY_1)));
@@ -56,6 +69,9 @@ EXPORT TestStore := MODULE
             // Delete a key/value
             ASSERT(deleteKeyValueRes.succeeded);
 
+            // Delete another key/value
+            ASSERT(deleteKeyValue2Res.succeeded);
+
             // Ensure our namespace still exists
             ASSERT(EXISTS(listNamespacesRes.namespaces(namespace = STORE_NAMESPACE)));
 

esp/scm/ws_store.ecm

@@ -114,7 +114,7 @@ ESPrequest FetchRequest
 
 ESPresponse [exceptions_inline] FetchResponse
 {
-    string Value;
+    [nil_remove] string Value;
 };
 
 ESPrequest FetchAllRequest

esp/services/ws_machine/ws_machineService.cpp

@@ -908,26 +908,12 @@ void Cws_machineEx::getProcesses(IConstEnvironment* constEnv, IPropertyTree* env
             SCMStringBuffer ep;
             pMachineInfo->getNetAddress(ep);
 
-            const char* ip = ep.str();
-            if (!ip)
+            if (ep.length() == 0)
             {
                 OWARNLOG("Network address not found for machine %s", name0);
                 continue;
             }
-
-            StringBuffer netAddress;
-            StringBuffer configNetAddress(ip);
-            if (!streq(ip, "."))
-            {
-                netAddress.set(ip);
-            }
-            else
-            {
-                IpAddress ipaddr = queryHostIP();
-                ipaddr.getIpText(netAddress);
-            }
-
-            setProcessRequest(machineInfoData, uniqueProcesses, netAddress.str(), configNetAddress.str(), processType, processName, directory0);
+            setProcessRequest(machineInfoData, uniqueProcesses, ep.str(), ep.str(), processType, processName, directory0);
         }
     }
 

esp/services/ws_store/espstorelib/CMakeLists.txt

@@ -24,7 +24,8 @@ include_directories (
     ${HPCC_SOURCE_DIR}/system/jlib
     ${HPCC_SOURCE_DIR}/dali/base                    #SDS
     ${HPCC_SOURCE_DIR}/system/mp                    #MP, included by dali base
-    ${HPCC_SOURCE_DIR}/system/security/shared/
+    ${HPCC_SOURCE_DIR}/system/security/shared
+    ${HPCC_SOURCE_DIR}/esp/smc/SMCLib
 )
 
 ADD_DEFINITIONS ( -D_USRDLL -DDALIKVSTORE_EXPORTS )

esp/services/ws_store/espstorelib/daliKVStore.cpp

@@ -410,14 +410,18 @@ bool CDALIKVStore::fetch(const char * storename, const char * ns, const char * k
     {
         xpath.appendf("/%s", key);
         if(!storetree->hasProp(xpath.str()))
-            throw makeStringExceptionV(MSGAUD_user, -1, "DALI Keystore fetch: invalid key '%s' detected!", key);
-
-        value.set(storetree->queryProp(xpath.str()));
+        {
+            throw makeStringExceptionV(ECLWATCH_INVALID_QUERY_KEY, "DALI Keystore fetch: invalid key '%s' detected!", key);
+        }
+        else
+        {
+            value.set(storetree->queryProp(xpath.str()));
+        }
 
         return value.str();
     }
     else
-        throw MakeStringException(-1, "DALI Keystore fetch: Namespace not provided!");
+        throw makeStringException(-1, "DALI Keystore fetch: Key not provided!");
 
     return true;
 }

esp/services/ws_store/espstorelib/espStoreShare.hpp

@@ -21,6 +21,7 @@
 #define _ESPSTORESHARE_HPP__
 
 #include "SecureUser.hpp"
+#include "eclwatch_errorlist.hpp"
 
 interface IEspStore : extends IInterface
 {

esp/services/ws_store/ws_storeService.cpp

@@ -312,8 +312,23 @@ bool CwsstoreEx::onFetch(IEspContext &context, IEspFetchRequest &req, IEspFetchR
             storename = m_defaultStore.get();
     }
 
-    m_storeProvider->fetch(storename, req.getNamespace(), req.getKey(), value, secuser.get(), !req.getUserSpecific());
-    resp.setValue(value.str());
+    try
+    {
+        m_storeProvider->fetch(storename, req.getNamespace(), req.getKey(), value, secuser.get(), !req.getUserSpecific());
+        resp.setValue(value.str());
+    }
+    catch(IException * e)
+    {
+        if (e->errorCode() == ECLWATCH_INVALID_QUERY_KEY)
+        {
+            StringBuffer msg;
+            LOG(MCuserInfo, "WsStore: %s", e->errorMessage(msg).str());
+            e->Release();
+            return false;
+        }
+        else
+            throw e;
+    }
 
     return true;
 }

esp/services/ws_store/ws_storeService.hpp

@@ -23,6 +23,7 @@ limitations under the License.
 
 #include "dautils.hpp"
 #include "espStoreShare.hpp"
+#include "eclwatch_errorlist.hpp"
 
 static StringBuffer g_wsstoreBuildVersion;
 

esp/smc/SMCLib/TpCommon.cpp

@@ -29,6 +29,20 @@
 #include "dautils.hpp"
 #include "dameta.hpp"
 
+void CTpWrapper::appendTpMachine(double clientVersion, IConstEnvironment* constEnv, IConstInstanceInfo& instanceInfo, IArrayOf<IConstTpMachine>& machines)
+{
+    SCMStringBuffer name, networkAddress, description, directory;
+    Owned<IEspTpMachine> machine = createTpMachine();
+    Owned<IConstMachineInfo> machineInfo = instanceInfo.getMachine();
+    machine->setName(machineInfo->getName(name).str());
+    machine->setOS(machineInfo->getOS());
+    machine->setNetaddress(machineInfo->getNetAddress(networkAddress).str());
+    machine->setDirectory(instanceInfo.getDirectory(directory).str());
+    machine->setPort(instanceInfo.getPort());
+    machine->setType(eqSparkThorProcess); //for now, the appendTpMachine is only used for SparkThor.
+    machines.append(*machine.getLink());
+}
+
 extern TPWRAPPER_API ISashaCommand* archiveOrRestoreWorkunits(StringArray& wuids, IProperties* params, bool archive, bool dfu)
 {
     StringBuffer sashaAddress;

esp/smc/SMCLib/TpContainer.cpp

@@ -87,13 +87,9 @@ void CTpWrapper::getTpEspServers(IArrayOf<IConstTpEspServer>& list)
 static IEspTpMachine * createHostTpMachine(const char * hostname, const char *path)
 {
     Owned<IEspTpMachine> machine = createTpMachine();
-    IpAddress ipAddr;
-    ipAddr.ipset(hostname);
-    StringBuffer localHost;
-    ipAddr.getIpText(localHost);
-    machine->setName(localHost.str());
-    machine->setNetaddress(localHost.str());
-    machine->setConfigNetaddress(hostname);
+    machine->setName(hostname);
+    machine->setNetaddress(hostname);
+    machine->setConfigNetaddress(hostname); //May be used by legacy ECLWatch. Leave it for now.
     machine->setDirectory(path);
     machine->setOS(getPathSepChar(path) == '/' ? MachineOsLinux : MachineOsW2K);
     return machine.getClear();
@@ -397,33 +393,6 @@ void CTpWrapper::getTpSparkThors(double clientVersion, const char* name, IArrayO
     UNIMPLEMENTED_X("CONTAINERIZED(CTpWrapper::getTpSparkThors)");
 }
 
-void CTpWrapper::appendTpMachine(double clientVersion, IConstEnvironment* constEnv, IConstInstanceInfo& instanceInfo, IArrayOf<IConstTpMachine>& machines)
-{
-    SCMStringBuffer name, networkAddress, description, directory;
-    Owned<IConstMachineInfo> machineInfo = instanceInfo.getMachine();
-    machineInfo->getName(name);
-    machineInfo->getNetAddress(networkAddress);
-    instanceInfo.getDirectory(directory);
-
-    Owned<IEspTpMachine> machine = createTpMachine();
-    machine->setName(name.str());
-
-    if (networkAddress.length() > 0)
-    {
-        IpAddress ipAddr;
-        ipAddr.ipset(networkAddress.str());
-
-        StringBuffer networkAddressStr;
-        ipAddr.getIpText(networkAddressStr);
-        machine->setNetaddress(networkAddressStr);
-    }
-    machine->setPort(instanceInfo.getPort());
-    machine->setOS(machineInfo->getOS());
-    machine->setDirectory(directory.str());
-    machine->setType(eqSparkThorProcess);
-    machines.append(*machine.getLink());
-}
-
 IEspTpMachine* CTpWrapper::createTpMachineEx(const char* name, const char* type, IConstMachineInfo* machineInfo)
 {
     if (!machineInfo)
@@ -467,7 +436,6 @@ IEspTpMachine* CTpWrapper::createTpMachineEx(const char* name, const char* type,
     return machine.getClear();
 }
 
-
 void CTpWrapper::setAttPath(StringBuffer& Path,const char* PathToAppend,const char* AttName,const char* AttValue,StringBuffer& returnStr)
 {
     Path.append("/");

esp/smc/SMCLib/TpWrapper.cpp

@@ -1635,11 +1635,8 @@ void CTpWrapper::appendTpDropZone(double clientVersion, IConstEnvironment* const
             machine->setName(name.str());
         if (!server.isEmpty())
         {
-            IpAddress ipAddr;
-            ipAddr.ipset(server.str());
-            ipAddr.getIpText(networkAddress);
-            machine->setNetaddress(networkAddress.str());
-            machine->setConfigNetaddress(server.str());
+            machine->setNetaddress(server);
+            machine->setConfigNetaddress(server); //May be used by legacy ECLWatch. Leave it for now.
         }
         if (directory.length() > 0)
         {
@@ -1713,33 +1710,6 @@ void CTpWrapper::appendTpSparkThor(double clientVersion, IConstEnvironment* cons
     list.append(*sparkThor.getLink());
 }
 
-void CTpWrapper::appendTpMachine(double clientVersion, IConstEnvironment* constEnv, IConstInstanceInfo& instanceInfo, IArrayOf<IConstTpMachine>& machines)
-{
-    SCMStringBuffer name, networkAddress, description, directory;
-    Owned<IConstMachineInfo> machineInfo = instanceInfo.getMachine();
-    machineInfo->getName(name);
-    machineInfo->getNetAddress(networkAddress);
-    instanceInfo.getDirectory(directory);
-
-    Owned<IEspTpMachine> machine = createTpMachine();
-    machine->setName(name.str());
-
-    if (networkAddress.length() > 0)
-    {
-        IpAddress ipAddr;
-        ipAddr.ipset(networkAddress.str());
-
-        StringBuffer networkAddressStr;
-        ipAddr.getIpText(networkAddressStr);
-        machine->setNetaddress(networkAddressStr);
-    }
-    machine->setPort(instanceInfo.getPort());
-    machine->setOS(machineInfo->getOS());
-    machine->setDirectory(directory.str());
-    machine->setType(eqSparkThorProcess);
-    machines.append(*machine.getLink());
-}
-
 IEspTpMachine* CTpWrapper::createTpMachineEx(const char* name, const char* type, IConstMachineInfo* machineInfo)
 {
     if (!machineInfo)
@@ -1794,30 +1764,9 @@ void CTpWrapper::setMachineInfo(const char* name,const char* type,IEspTpMachine&
             SCMStringBuffer ep;
 
             pMachineInfo->getNetAddress(ep);
-
-            const char* ip = ep.str();
-            if (!ip || stricmp(ip, "."))
-            {
-                machine.setNetaddress(ep.str());
-                machine.setConfigNetaddress(ep.str());
-            }
-            else
-            {
-                StringBuffer ipStr;
-                IpAddress ipaddr = queryHostIP();
-                ipaddr.getIpText(ipStr);
-                if (ipStr.length() > 0)
-                {
-#ifdef MACHINE_IP
-                    machine.setNetaddress(MACHINE_IP);
-#else
-                    machine.setNetaddress(ipStr.str());
-#endif
-                    machine.setConfigNetaddress(".");
-                }
-            }
+            machine.setNetaddress(ep.str());
+            machine.setConfigNetaddress(ep.str());
             machine.setOS(pMachineInfo->getOS());
-                
             
             switch(pMachineInfo->getState())
             {

esp/smc/SMCLib/eclwatch_errorlist.hpp

@@ -128,6 +128,7 @@
 #define ECLWATCH_INVALID_ECLRECDEF          ECLWATCH_ERROR_START+107
 #define ECLWATCH_MISSING_FILETYPE           ECLWATCH_ERROR_START+108
 
+#define ECLWATCH_INVALID_QUERY_KEY          ECLWATCH_ERROR_START+109
 
 #endif //_ECLWATCH_ERRORLIST_HPP__
 

esp/src/src-react/components/Files.tsx

@@ -4,10 +4,12 @@ import * as domClass from "dojo/dom-class";
 import * as put from "put-selector/put";
 import * as WsDfu from "src/WsDfu";
 import * as ESPLogicalFile from "src/ESPLogicalFile";
+import { formatCost } from "src/Session";
 import * as Utility from "src/Utility";
 import nlsHPCC from "src/nlsHPCC";
 import { useConfirm } from "../hooks/confirm";
 import { useGrid } from "../hooks/grid";
+import { useBuildInfo } from "../hooks/platform";
 import { HolyGrail } from "../layouts/HolyGrail";
 import { pushParams } from "../util/history";
 import { AddToSuperfile } from "./forms/AddToSuperfile";
@@ -74,6 +76,7 @@ export const Files: React.FunctionComponent<FilesProps> = ({
     const [showDesprayFile, setShowDesprayFile] = React.useState(false);
     const [mine, setMine] = React.useState(false);
     const [uiState, setUIState] = React.useState({ ...defaultUIState });
+    const [, { currencyCode }] = useBuildInfo();
 
     //  Grid ---
     const [Grid, selection, refreshTable, copyButtons] = useGrid({
@@ -160,7 +163,13 @@ export const Files: React.FunctionComponent<FilesProps> = ({
                     node.innerText = Utility.valueCleanUp(value);
                 },
             },
-            Modified: { label: nlsHPCC.ModifiedUTCGMT, width: 162 }
+            Modified: { label: nlsHPCC.ModifiedUTCGMT, width: 162 },
+            Cost: {
+                label: nlsHPCC.Cost, width: 100,
+                formatter: function (cost, row) {
+                    return `${formatCost(cost ?? 0)} (${currencyCode || "$"})`;
+                }
+            }
         }
     });
 

esp/src/src-react/components/Workunits.tsx

@@ -2,12 +2,14 @@ import * as React from "react";
 import { CommandBar, ContextualMenuItemType, ICommandBarItemProps } from "@fluentui/react";
 import { scopedLogger } from "@hpcc-js/util";
 import * as domClass from "dojo/dom-class";
-import * as WsWorkunits from "src/WsWorkunits";
 import * as ESPWorkunit from "src/ESPWorkunit";
+import * as WsWorkunits from "src/WsWorkunits";
+import { formatCost } from "src/Session";
 import * as Utility from "src/Utility";
 import nlsHPCC from "src/nlsHPCC";
 import { useConfirm } from "../hooks/confirm";
 import { useGrid } from "../hooks/grid";
+import { useBuildInfo } from "../hooks/platform";
 import { HolyGrail } from "../layouts/HolyGrail";
 import { pushParams } from "../util/history";
 import { Fields } from "./forms/Fields";
@@ -78,6 +80,7 @@ export const Workunits: React.FunctionComponent<WorkunitsProps> = ({
     const [showFilter, setShowFilter] = React.useState(false);
     const [mine, setMine] = React.useState(false);
     const [uiState, setUIState] = React.useState({ ...defaultUIState });
+    const [, { currencyCode }] = useBuildInfo();
 
     //  Grid ---
     const [Grid, selection, refreshTable, copyButtons] = useGrid({
@@ -105,22 +108,34 @@ export const Workunits: React.FunctionComponent<WorkunitsProps> = ({
             },
             Wuid: {
                 label: nlsHPCC.WUID, width: 180,
-                formatter: function (Wuid) {
+                formatter: function (Wuid, row) {
                     const wu = ESPWorkunit.Get(Wuid);
-                    return `${wu.getStateImageHTML()}&nbsp;<a href='#/workunits/${Wuid}' class='dgrid-row-url''>${Wuid}</a>`;
+                    return `${wu.getStateImageHTML()}&nbsp;<a href='#/workunits/${Wuid}'>${Wuid}</a>`;
                 }
             },
             Owner: { label: nlsHPCC.Owner, width: 90 },
-            Jobname: { label: nlsHPCC.JobName, width: 500 },
-            Cluster: { label: nlsHPCC.Cluster, width: 90 },
-            RoxieCluster: { label: nlsHPCC.RoxieCluster, width: 99 },
-            State: { label: nlsHPCC.State, width: 90 },
+            Jobname: { label: nlsHPCC.JobName, width: 350 },
+            Cluster: { label: nlsHPCC.Cluster, width: 60 },
+            RoxieCluster: { label: nlsHPCC.RoxieCluster, width: 90 },
+            State: { label: nlsHPCC.State, width: 60 },
             TotalClusterTime: {
-                label: nlsHPCC.TotalClusterTime, width: 117,
+                label: nlsHPCC.TotalClusterTime, width: 115,
                 renderCell: function (object, value, node) {
                     domClass.add(node, "justify-right");
                     node.innerText = value;
                 }
+            },
+            ExecuteCost: {
+                label: nlsHPCC.ExecuteCost, width: 100,
+                formatter: function (cost, row) {
+                    return `${formatCost(cost ?? 0)} (${currencyCode || "$"})`;
+                }
+            },
+            FileAccessCost: {
+                label: nlsHPCC.FileAccessCost, width: 100,
+                formatter: function (cost, row) {
+                    return `${formatCost(cost ?? 0)} (${currencyCode || "$"})`;
+                }
             }
         }
     });

helm/managed/logging/elastic/Chart.yaml

@@ -5,10 +5,10 @@ type: application
 
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 1.0.2
+version: 1.2.0
 
 # Elastic Stack version
-appVersion: 7.12.0
+appVersion: 7.16.1
 
 # Dependencies can be automatically updated via HELM dependancy update command:
 # > 'helm dependency update' command
@@ -16,12 +16,12 @@ appVersion: 7.12.0
 # > helm install myelastic ./ —-dependency-update
 dependencies:
 - name: filebeat
-  version: 7.12.0
+  version: 7.16.1
   repository: https://helm.elastic.co
 - name: elasticsearch
-  version: 7.12.0
+  version: 7.16.1
   repository: https://helm.elastic.co
 - name: kibana # Optional managed logging processor front-end
-  version: 7.12.0
+  version: 7.16.1
   repository: https://helm.elastic.co
   condition: kibana.enabled

helm/managed/logging/elastic/README.md

@@ -1,7 +1,28 @@
 ## This folder contains lightweight Elastic Stack deployment chart and HPCC Systems preferred values
 
+<table>
+  <thead>
+    <tr>
+      <td align="left">
+        :zap: <b>Note:</b> Elastic Stack components have been reported to be affected by the high-severity vulnerability (CVE-2021-44228) impacting multiple versions of the Apache Log4j 2 utility
+      </td>
+    </tr>
+  </thead>
+
+  <tbody>
+    <tr>
+      <td>
+        <ul>
+          <li>Users of elastic4hpcclogs are strongly encouraged to update to chart version 1.2.0 which references Elastic Stack 7.16.1</li>
+          <li>Learn more about Elastic's response to the vulnerability: https://discuss.elastic.co/t/apache-log4j2-remote-code-execution-rce-vulnerability-cve-2021-44228-esa-2021-31/291476</li>
+        </ul>
+      </td>
+    </tr>
+  </tbody>
+</table>
+
 This chart describes a local, minimal Elastic Stack instance for HPCC Systems component log processing.
-Once successfully deployed, HPCC component logs produced within the same namespace should be automatically index
+Once successfully deployed, HPCC component logs produced within the same namespace should be automatically indexed
 on the Elastic Search end-point. Users can query those logs by issuing Elastic Search RESTful API queries, or via
 the Kibana UI (after creating a simple index pattern).
 

roxie/udplib/udpsha.cpp

@@ -678,6 +678,7 @@ fake read socket that
 
 #ifdef SOCKET_SIMULATION
 bool isUdpTestMode = false;
+bool udpTestUseUdpSockets = true;
 
 CSimulatedQueueWriteSocket* CSimulatedQueueWriteSocket::udp_connect(const SocketEndpoint &ep)
 {
@@ -800,12 +801,17 @@ unsigned getMappedSocketPort(const SocketEndpoint & ep)
 
 CSimulatedUdpReadSocket::CSimulatedUdpReadSocket(const SocketEndpoint &_me)
 {
-    unsigned port = getMappedSocketPort(_me);
+    port = getMappedSocketPort(_me);
     if (connected[port-basePort].exchange(true))
         throw makeStringException(0, "Two ip/ports mapped to the same port - improve the hash (or change maxPorts)!");
     realSocket.setown(ISocket::udp_create(port));
 }
 
+CSimulatedUdpReadSocket::~CSimulatedUdpReadSocket()
+{
+    connected[port-basePort].exchange(false);
+}
+
 size32_t CSimulatedUdpReadSocket::get_receive_buffer_size() { return realSocket->get_receive_buffer_size(); }
 void CSimulatedUdpReadSocket::set_receive_buffer_size(size32_t sz) { realSocket->set_receive_buffer_size(sz); }
 void CSimulatedUdpReadSocket::read(void* buf, size32_t min_size, size32_t max_size, size32_t &size_read, unsigned timeoutsecs)

roxie/udplib/udpsha.hpp

@@ -264,7 +264,6 @@ inline bool checkTraceLevel(unsigned category, unsigned level)
     return (udpTraceLevel >= level);
 }
 #define SOCKET_SIMULATION
-#define SOCKET_SIMULATION_UDP
 
 #ifdef SOCKET_SIMULATION
 #ifdef _DEBUG
@@ -278,6 +277,7 @@ extern unsigned flowPacketsSent[flowType::max_flow_cmd];
 #endif
 
 extern UDPLIB_API bool isUdpTestMode;
+extern UDPLIB_API bool udpTestUseUdpSockets;
 
 class CSocketSimulator : public CInterfaceOf<ISocket>
 {
@@ -412,11 +412,14 @@ class CSimulatedUdpSocket : public CSocketSimulator
 protected:
     Owned<ISocket> realSocket;
 };
+
 class CSimulatedUdpReadSocket : public CSimulatedUdpSocket
 {
     CSimulatedUdpReadSocket(const SocketEndpoint &_me);
+    ~CSimulatedUdpReadSocket();
 
 public:
+    unsigned port;
     static CSimulatedUdpReadSocket* udp_create(const SocketEndpoint &_me);
 
     virtual size32_t get_receive_buffer_size() override;
@@ -438,13 +441,6 @@ public:
     virtual void close() override;
 };
 
-#ifdef SOCKET_SIMULATION_UDP
-using CSimulatedWriteSocket = CSimulatedUdpWriteSocket;
-using CSimulatedReadSocket = CSimulatedUdpReadSocket;
-#else
-using CSimulatedWriteSocket = CSimulatedQueueWriteSocket;
-using CSimulatedReadSocket = CSimulatedQueueReadSocket;
-#endif
 
 #endif
 

roxie/udplib/udpsim.cpp

@@ -30,7 +30,10 @@ using roxiemem::IDataBufferManager;
 
 Owned<IDataBufferManager> dbm;
 
-unsigned numThreads = 20;
+static unsigned numThreads = 20;
+static unsigned packetsPerThread = 0;
+static bool restartSender = false;
+static bool restartReceiver = false;
 
 static constexpr const char * defaultYaml = R"!!(
 version: "1.0"
@@ -44,6 +47,9 @@ udpsim:
   help: false
   numThreads: 20
   outputconfig: false
+  packetsPerThread: 10000
+  restartReceiver: false
+  restartSender: false
   udpTraceLevel: 1
   udpTraceTimeouts: true
   udpResendLostPackets: true
@@ -51,6 +57,7 @@ udpsim:
   udpRequestToSendAckTimeout: 1000
   udpMaxPendingPermits: 1
   udpTraceFlow: false
+  useQueue: false
 )!!";
 
 bool isNumeric(const char *str)
@@ -134,6 +141,9 @@ void initOptions(int argc, const char **argv)
     udpDropFlowPackets[flowType::request_to_send_more] = options->getPropInt("@dropRequestToSendMorePackets", 0);  // drop 1 in N
     udpDropFlowPackets[flowType::send_completed] = options->getPropInt("@dropSendCompletedPackets", 0);  // drop 1 in N
 #endif
+    restartSender = options->getPropBool("@restartSender");
+    restartReceiver = options->getPropBool("@restartReceiver");
+
     numThreads = options->getPropInt("@numThreads", 0);
     udpTraceLevel = options->getPropInt("@udpTraceLevel", 1);
     udpTraceTimeouts = options->getPropBool("@udpTraceTimeouts", true);
@@ -142,12 +152,22 @@ void initOptions(int argc, const char **argv)
     udpRequestToSendAckTimeout = options->getPropInt("@udpRequestToSendAckTimeout", 1000);
     udpMaxPendingPermits = options->getPropInt("@udpMaxPendingPermits", 1);
     udpTraceFlow = options->getPropBool("@udpTraceFlow", false);
+    packetsPerThread = options->getPropInt("@packetsPerThread");
+    udpTestUseUdpSockets = !options->getPropBool("@useQueue");
 
     isUdpTestMode = true;
     roxiemem::setTotalMemoryLimit(false, false, false, 20*1024*1024, 0, NULL, NULL);
     dbm.setown(roxiemem::createDataBufferManager(roxiemem::DATA_ALIGNMENT_SIZE));
 }
 
+// How many times the simulated sender [i] should start
+unsigned numStarts(unsigned i)
+{
+    if (i==1 && restartSender)
+        return 2;
+    return 1;
+}
+
 void simulateTraffic()
 {
     constexpr unsigned numReceiveSlots = 100;
@@ -159,31 +179,52 @@ void simulateTraffic()
         Owned<IReceiveManager> rm = createReceiveManager(CCD_SERVER_FLOW_PORT, CCD_DATA_PORT, CCD_CLIENT_FLOW_PORT, numReceiveSlots, maxSlotsPerClient, false);
         unsigned begin = msTick();
         printf("Start test\n");
-        asyncFor(numThreads, numThreads, [maxSendQueueSize](unsigned i)
+        asyncFor(numThreads+1, numThreads+1, [maxSendQueueSize, numReceiveSlots, maxSlotsPerClient, &rm](unsigned i)
         {
-            unsigned header = 0;
-            IpAddress pretendIP(VStringBuffer("8.8.8.%d", i));
-            // Note - this is assuming we send flow on the data port (that option defaults true in roxie too)
-            Owned<ISendManager> sm = createSendManager(CCD_DATA_PORT, CCD_DATA_PORT, CCD_CLIENT_FLOW_PORT, maxSendQueueSize, 3, pretendIP, nullptr, false);
-            Owned<IMessagePacker> mp = sm->createMessagePacker(0, 0, &header, sizeof(header), myNode, 0);
-            for (unsigned j = 0; j < 10000; j++)
+            if (!i)
             {
-                void *buf = mp->getBuffer(500, false);
-                memset(buf, i, 500);
-                mp->putBuffer(buf, 500, false);
+                if (restartReceiver)
+                {
+                    Sleep(100);
+                    rm.clear();
+                    rm.setown(createReceiveManager(CCD_SERVER_FLOW_PORT, CCD_DATA_PORT, CCD_CLIENT_FLOW_PORT, numReceiveSlots, maxSlotsPerClient, false));
+                }
+            }
+            else
+            {
+                unsigned header = 0;
+                unsigned myStarts = numStarts(i);
+                for (unsigned startNo = 0; startNo < myStarts; startNo++)
+                {
+                    IpAddress pretendIP(VStringBuffer("8.8.8.%d", i));
+                    // Note - this is assuming we send flow on the data port (that option defaults true in roxie too)
+                    Owned<ISendManager> sm = createSendManager(CCD_DATA_PORT, CCD_DATA_PORT, CCD_CLIENT_FLOW_PORT, maxSendQueueSize, 3, pretendIP, nullptr, false);
+                    unsigned numPackets = packetsPerThread / myStarts;
+                    for (unsigned j = 0; j < packetsPerThread; j++)
+                    {
+                        Owned<IMessagePacker> mp = sm->createMessagePacker(0, 0, &header, sizeof(header), myNode, 0);
+                        void *buf = mp->getBuffer(500, false);
+                        memset(buf, i, 500);
+                        mp->putBuffer(buf, 500, false);
+                        mp->flush();
+                    }
+
+                    // Wait until all the packets have been sent and acknowledged, for last start only
+                    // For prior starts, we are trying to simulate a sender stopping abruptly (e.g. from a restart) so we don't want to close it down cleanly.
+                    if (startNo == myStarts-1)
+                        while (!sm->allDone())
+                            Sleep(50);
+                    DBGLOG("UdpSim sender thread %d sent %d packets", i, numPackets);
+                }
+                DBGLOG("UdpSim sender thread %d completed", i);
             }
-            mp->flush();
-
-            //wait until all the packets have been sent and acknowledged
-            while(!sm->allDone())
-                Sleep(50);
         });
-        printf("End test %u\n", msTick() - begin);
+        printf("UdpSim test took %ums\n", msTick() - begin);
     }
     catch (IException * e)
     {
-        StringBuffer msg;
-        printf("Exception: %s\n", e->errorMessage(msg).str());
+        EXCLOG(e);
+        e->Release();
     }
 }
 

roxie/udplib/udptrr.cpp

@@ -132,7 +132,10 @@ class CReceiveManager : implements IReceiveManager, public CInterface
             SocketEndpoint ep(port, dest);
 #ifdef SOCKET_SIMULATION
             if (isUdpTestMode)
-                flowSocket = CSimulatedWriteSocket::udp_connect(ep);
+                if (udpTestUseUdpSockets)
+                    flowSocket = CSimulatedUdpWriteSocket::udp_connect(ep);
+                else
+                    flowSocket = CSimulatedQueueWriteSocket::udp_connect(ep);
             else
 #endif
                 flowSocket = ISocket::udp_connect(ep);
@@ -426,7 +429,10 @@ class CReceiveManager : implements IReceiveManager, public CInterface
                 throw MakeStringException(ROXIE_UDP_ERROR, "System Socket max read buffer is less than %i", udpFlowSocketsSize);
 #ifdef SOCKET_SIMULATION
             if (isUdpTestMode)
-                flow_socket.setown(CSimulatedReadSocket::udp_create(SocketEndpoint(flow_port, myNode.getIpAddress())));
+                if (udpTestUseUdpSockets)
+                    flow_socket.setown(CSimulatedUdpReadSocket::udp_create(SocketEndpoint(flow_port, myNode.getIpAddress())));
+                else
+                    flow_socket.setown(CSimulatedQueueReadSocket::udp_create(SocketEndpoint(flow_port, myNode.getIpAddress())));
             else
 #endif
                 flow_socket.setown(ISocket::udp_create(flow_port));
@@ -619,8 +625,16 @@ class CReceiveManager : implements IReceiveManager, public CInterface
 #ifdef SOCKET_SIMULATION
             if (isUdpTestMode)
             {
-                receive_socket = CSimulatedReadSocket::udp_create(SocketEndpoint(parent.data_port, myNode.getIpAddress()));
-                selfFlowSocket = CSimulatedWriteSocket::udp_connect(SocketEndpoint(parent.receive_flow_port, myNode.getIpAddress()));
+                if (udpTestUseUdpSockets)
+                {
+                    receive_socket = CSimulatedUdpReadSocket::udp_create(SocketEndpoint(parent.data_port, myNode.getIpAddress()));
+                    selfFlowSocket = CSimulatedUdpWriteSocket::udp_connect(SocketEndpoint(parent.receive_flow_port, myNode.getIpAddress()));
+                }
+                else
+                {
+                    receive_socket = CSimulatedQueueReadSocket::udp_create(SocketEndpoint(parent.data_port, myNode.getIpAddress()));
+                    selfFlowSocket = CSimulatedQueueWriteSocket::udp_connect(SocketEndpoint(parent.receive_flow_port, myNode.getIpAddress()));
+                }
            }
             else
 #endif

roxie/udplib/udptrs.cpp

@@ -583,8 +583,16 @@ public:
 #ifdef SOCKET_SIMULATION
                 if (isUdpTestMode)
                 {
-                    send_flow_socket = CSimulatedWriteSocket::udp_connect(sendFlowEp);
-                    data_socket = CSimulatedWriteSocket::udp_connect(dataEp);
+                    if (udpTestUseUdpSockets)
+                    {
+                        send_flow_socket = CSimulatedUdpWriteSocket::udp_connect(sendFlowEp);
+                        data_socket = CSimulatedUdpWriteSocket::udp_connect(dataEp);
+                    }
+                    else
+                    {
+                        send_flow_socket = CSimulatedQueueWriteSocket::udp_connect(sendFlowEp);
+                        data_socket = CSimulatedQueueWriteSocket::udp_connect(dataEp);
+                    }
                 }
                 else
 #endif
@@ -666,8 +674,11 @@ class CSendManager : implements ISendManager, public CInterface
 
         ~StartedThread()
         {
-            running = false;
-            join();
+            if (running)
+            {
+                running = false;
+                join();
+            }
         }
 
         virtual void start()
@@ -779,7 +790,12 @@ class CSendManager : implements ISendManager, public CInterface
                 throw MakeStringException(ROXIE_UDP_ERROR, "System Socket max read buffer is less than %i", udpFlowSocketsSize);
 #ifdef SOCKET_SIMULATION
             if (isUdpTestMode)
-                flow_socket.setown(CSimulatedReadSocket::udp_create(SocketEndpoint(receive_port, parent.myIP)));
+            {
+                if (udpTestUseUdpSockets)
+                    flow_socket.setown(CSimulatedUdpReadSocket::udp_create(SocketEndpoint(receive_port, parent.myIP)));
+                else
+                    flow_socket.setown(CSimulatedQueueReadSocket::udp_create(SocketEndpoint(receive_port, parent.myIP)));
+            }
             else
 #endif
                 flow_socket.setown(ISocket::udp_create(receive_port));

system/security/plugins/jwtSecurity/CMakeLists.txt

@@ -43,6 +43,7 @@ include_directories("${CMAKE_CURRENT_SOURCE_DIR}/jwt-cpp/include"
                     "${HPCC_SOURCE_DIR}/dali/base"
                     "${HPCC_SOURCE_DIR}/dali/server"
                     "${HPCC_SOURCE_DIR}/esp/services/ws_store/espstorelib"
+                    "${HPCC_SOURCE_DIR}/esp/smc/SMCLib"
                     "${HPCC_SOURCE_DIR}/system/include"
                     "${HPCC_SOURCE_DIR}/system/mp"
                     "${HPCC_SOURCE_DIR}/system/security/shared"