Merge branch 'candidate-8.2.x'

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

dali/base/dasds.cpp

@@ -8823,9 +8823,6 @@ public:
             LOG(MCdebugInfo, unknownJob, "Failed to load main store");
             throw;
         }
-        // In nas/non-local storage mode, create a published named group for 1-way files to use
-        if (isContainerized())
-            queryNamedGroupStore().ensureNasGroup(1);
         storeLoaded = true;
         manager->start();
     }

dockerfiles/platform-core-debug/Dockerfile

@@ -28,7 +28,7 @@ ENV LANG en_US.UTF-8
 ENV LANGUAGE en_US:en  
 ENV LC_ALL en_US.UTF-8     
 
-RUN apt-get install -y libcap2-bin zip libarchive13
+RUN apt-get install -y libcap2-bin zip libarchive13 libatlas3-base
 RUN setcap cap_sys_ptrace+eip /usr/bin/gdb
 
 ENV PATH="/opt/HPCCSystems/bin:${PATH}"

dockerfiles/platform-core/Dockerfile

@@ -46,6 +46,7 @@ RUN apt-get install -y \
   r-base-core \
   zip \
   libarchive13 \
+  libatlas3-base \
   curl
 
 ARG USE_CPPUNIT=1

docs/EN_US/ECLStandardLibraryReference/SLR-Mods/GetExpireDays.xml

@@ -31,11 +31,11 @@
 
   <para>The <emphasis role="bold">GetExpireDays </emphasis>function retrieves
   a logical file's expiration criteria (the <emphasis>expireDays</emphasis>
-  attribute). </para>
+  attribute). A return of -1 indicates that there is no expiration set.</para>
 
   <para>Example:</para>
 
-  <programlisting format="linespecific">A := STD.File.SetExpireDays('~samples::myscope::myfile');
+  <programlisting format="linespecific">A := STD.File.GetExpireDays('~samples::myscope::myfile');
  //returns a file's expireDays 
 </programlisting>
 

docs/EN_US/ECLStandardLibraryReference/SLR-Mods/SetExpireDays.xml

@@ -32,21 +32,21 @@
 
           <entry>Number of days before the file expires. Setting to 0
           specifies to use the system's default expire value (specified in the
-          Sasha server's <emphasis>ExpiryDefault</emphasis> attribute).
-          </entry>
+          Sasha server's <emphasis>ExpiryDefault</emphasis>
+          attribute).</entry>
         </row>
       </tbody>
     </tgroup>
   </informaltable>
 
-  <para>The <emphasis role="bold">SetExpireDays </emphasis>function sets a
+  <para>The <emphasis role="bold">SetExpireDays </emphasis>action sets a
   logical file's expiration criteria (the <emphasis>expireDays</emphasis>
   attribute). The file is deleted by the Sasha server when a file has not been
-  accessed for the number of days specified. </para>
+  accessed for the number of days specified.</para>
 
   <para>Example:</para>
 
-  <programlisting format="linespecific">A := STD.File.SetExpireDays('~samples::myscope::myfile',30);
+  <programlisting format="linespecific">STD.File.SetExpireDays('~samples::myscope::myfile',30);
  //file expires and is deleted after 30 days w/o access
 </programlisting>
 

helm/hpcc/templates/_helpers.tpl

@@ -140,6 +140,33 @@ Returns the largest number of workers from all the thors
 {{- end -}}
 
 {{/*
+Returns true if the given certificate issuer is enabled, otherwise false
+*/}}
+{{- define "hpcc.isIssuerEnabled" -}}
+{{- $certificates := (.root.Values.certificates | default dict) -}}
+{{- $issuers := ($certificates.issuers | default dict) -}}
+{{- $issuer := get $issuers .issuer -}}
+{{- if $issuer -}}
+  {{- (hasKey $issuer "enabled" | ternary $issuer.enabled true) }}
+{{- else -}}
+false
+{{- end -}}
+{{- end -}}
+
+{{/*
+Returns true if mtls should be enabled, otherwise false
+*/}}
+{{- define "hpcc.isMtlsEnabled" -}}
+{{- $security := .root.Values.security | default dict -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" "local")) "true" -}}
+  {{- (hasKey $security "mtls" | ternary $security.mtls true) -}}
+{{- else -}}
+false
+{{- end -}}
+{{- end -}}
+
+
+{{/*
 Generate global ConfigMap info
 Pass in root as .
 */}}
@@ -150,8 +177,7 @@ Pass in root as .
 {{- $certificates := (.Values.certificates | default dict) -}}
 {{- $issuers := ($certificates.issuers | default dict) -}}
 {{- $security := .Values.security | default dict -}}
-{{- $mtls := hasKey $security "mtls" | ternary $security.mtls true -}}
-mtls: {{ (and $mtls (and ($certificates.enabled) (hasKey $issuers "local"))) }}
+mtls: {{ (include "hpcc.isMtlsEnabled" (dict "root" $)) }}
 imageVersion: {{ .Values.global.image.version | default .Chart.Version }}
 singleNode: {{ .Values.global.singleNode | default false }}
 {{ if .Values.global.defaultEsp -}}
@@ -1149,6 +1175,7 @@ use "public" or "local"
 {{- if (.root.Values.certificates | default dict).enabled -}}
 {{- $externalCert := or (and (hasKey . "external") .external) (ne (include "hpcc.isVisibilityPublic" .) "") -}}
 {{- $issuerName := .issuer | default (ternary "public" "local" $externalCert) -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" $issuerName)) "true" -}}
 {{- $issuer := get .root.Values.certificates.issuers $issuerName -}}
 {{- if $issuer -}}
 {{- $namespace := .root.Release.Namespace -}}
@@ -1211,6 +1238,7 @@ spec:
 {{- end }}
 {{- end }}
 {{- end }}
+{{- end }}
 
 {{/*
 Experimental: Use certmanager to generate a key for roxie udp encryption.
@@ -1220,6 +1248,7 @@ Key is in pem format and the private key would need to be extracted.
 */}}
 {{- define "hpcc.addUDPCertificate" }}
 {{- if (.root.Values.certificates | default dict).enabled -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" "local")) "true" -}}
 {{- $issuer := .root.Values.certificates.issuers.local -}}
 {{- $namespace := .root.Release.Namespace -}}
 {{- $name := .name -}}
@@ -1260,6 +1289,7 @@ spec:
 {{- end }}
 {{- end }}
 {{- end }}
+{{- end }}
 
 {{/*
 Add a certficate volume mount for a component
@@ -1277,8 +1307,7 @@ use "public" or "local"
 - name: certificate-{{ .component }}-{{ $issuerName }}-{{ .name }}
   mountPath: /opt/HPCCSystems/secrets/certificates/{{ $issuerName }}
 {{- else if (.root.Values.certificates | default dict).enabled -}}
-{{- $issuer := get .root.Values.certificates.issuers $issuerName -}}
-{{- if $issuer -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" $issuerName)) "true" -}}
 - name: certificate-{{ .component }}-{{ $issuerName }}-{{ .name }}
   mountPath: /opt/HPCCSystems/secrets/certificates/{{ $issuerName }}
 {{- end }}
@@ -1302,8 +1331,7 @@ use "public" or "local"
   secret:
     secretName: {{ .certificate }}
 {{- else if (.root.Values.certificates | default dict).enabled -}}
-{{- $issuer := get .root.Values.certificates.issuers $issuerName -}}
-{{- if $issuer -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" $issuerName)) "true" -}}
 - name: certificate-{{ .component }}-{{ $issuerName }}-{{ .name }}
   secret:
     secretName: {{ .component }}-{{ $issuerName }}-{{ .name }}-tls
@@ -1316,7 +1344,7 @@ Add the certficate volume mount for a roxie udp key
 */}}
 {{- define "hpcc.addUDPCertificateVolumeMount" }}
 {{- if (.root.Values.certificates | default dict).enabled -}}
-{{- if .root.Values.certificates.issuers.local -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" "local")) "true" -}}
 - name: certificate-{{ .component }}-udp-{{ .name }}
   mountPath: /opt/HPCCSystems/secrets/certificates/udp
 {{- end -}}
@@ -1328,7 +1356,7 @@ Add a secret volume for a roxie udp key
 */}}
 {{- define "hpcc.addUDPCertificateVolume" }}
 {{- if (.root.Values.certificates | default dict).enabled -}}
-{{- if .root.Values.certificates.issuers.local -}}
+{{- if eq (include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" "local")) "true" -}}
 - name: certificate-{{ .component }}-udp-{{ .name }}
   secret:
     secretName: {{ .component }}-udp-{{ .name }}-dtls

helm/hpcc/templates/esp.yaml

@@ -37,11 +37,13 @@ data:
 {{- include "hpcc.generateLoggingConfig" . | indent 6 }}
 {{- include "hpcc.generateMetricsConfig" . | indent 6 }}
 {{- if and .root.Values.certificates .root.Values.certificates.enabled }}
+ {{- $externalCert := (ne (include "hpcc.isVisibilityPublic" (dict "root" .root "visibility" .me.service.visibility)) "") -}}
+ {{- $issuerName := ternary "public" "local" $externalCert -}}
  {{- if not (hasKey .me "tls" )}}
-      tls: true
+      tls: {{ include "hpcc.isIssuerEnabled" (dict "root" .root "issuer" $issuerName) }}
  {{- end }}
       tls_config:
- {{- if (not (eq .me.service.visibility "cluster")) }}
+ {{- if $externalCert }}
         certificate: /opt/HPCCSystems/secrets/certificates/public/tls.crt
         privatekey: /opt/HPCCSystems/secrets/certificates/public/tls.key
  {{- else }}

helm/hpcc/templates/roxie.yaml

@@ -37,9 +37,7 @@ data:
       numChannels: {{ .numChannels }}
       topologyServers: "{{ .toponame }}:{{ .topoport }}"
       resolveLocally: false
-{{- $certificates := (.root.Values.certificates | default dict) -}}
-{{- $issuers := ($certificates.issuers | default dict) -}}
-{{- $mtlsEnabled := (and ($certificates.enabled) (hasKey $issuers "local")) -}}
+{{- $mtlsEnabled := (eq (include "hpcc.isMtlsEnabled" (dict "root" .root)) "true") -}}
 {{/* By default use encryption if local certificates are enabled, but allow it to be turned off via roxie .encryptInTransit value */}}
 {{- if (hasKey .me "encryptInTransit") -}}
   {{- if and (.me.encryptInTransit) (not $mtlsEnabled) -}}

roxie/ccd/ccdserver.cpp

@@ -392,15 +392,15 @@ static const StatisticsMapping keyedJoinStatistics({ StNumServerCacheHits, StNum
                                                     StNumIndexRowsRead, StNumDiskRowsRead, StNumDiskSeeks, StNumDiskAccepted,
                                                     StNumBlobCacheHits, StNumLeafCacheHits, StNumNodeCacheHits,
                                                     StNumBlobCacheAdds, StNumLeafCacheAdds, StNumNodeCacheAdds,
-                                                    StNumDiskRejected}, joinStatistics);
+                                                    StNumDiskRejected, StSizeAgentReply}, joinStatistics);
 static const StatisticsMapping indexStatistics({StNumServerCacheHits, StNumIndexSeeks, StNumIndexScans, StNumIndexWildSeeks,
                                                 StNumIndexSkips, StNumIndexNullSkips, StNumIndexMerges, StNumIndexMergeCompares,
                                                 StNumPreFiltered, StNumPostFiltered, StNumIndexAccepted, StNumIndexRejected,
                                                 StNumBlobCacheHits, StNumLeafCacheHits, StNumNodeCacheHits,
                                                 StNumBlobCacheAdds, StNumLeafCacheAdds, StNumNodeCacheAdds,
-                                                StNumIndexRowsRead}, actStatistics);
+                                                StNumIndexRowsRead, StSizeAgentReply}, actStatistics);
 static const StatisticsMapping diskStatistics({StNumServerCacheHits, StNumDiskRowsRead, StNumDiskSeeks, StNumDiskAccepted,
-                                               StNumDiskRejected }, actStatistics);
+                                               StNumDiskRejected, StSizeAgentReply }, actStatistics);
 static const StatisticsMapping soapStatistics({ StTimeSoapcall }, actStatistics);
 static const StatisticsMapping groupStatistics({ StNumGroups, StNumGroupMax }, actStatistics);
 static const StatisticsMapping sortStatistics({ StTimeSortElapsed }, actStatistics);
@@ -4141,7 +4141,7 @@ public:
             //       activity type/activity behaviour/expected reply size .. etc).
             //       
             //       Currently (code below) based on high priority, seq=0, and none-child activity.
-            //       But this could still cause too many reply packets on the fatlane
+            //       But this could still cause too many reply packets on the fastlane
             //       (higher priority output Q), which may cause the activities on the 
             //       low priority output Q to not get service on time.
             if ((colocalArg == 0) &&     // not a child query activity??
@@ -4403,8 +4403,12 @@ public:
             ROQ->queryReceiveManager()->detachCollator(mc);
         merger.reset();
         pending.kill();
-        if (mc && ctx)
-            ctx->addAgentsReplyLen(mc->queryBytesReceived(), mc->queryDuplicates(), mc->queryResends());
+        if (mc)
+        {
+            activity.noteStatistic(StSizeAgentReply, mc->queryBytesReceived());
+            if (ctx)
+                ctx->addAgentsReplyLen(mc->queryBytesReceived(), mc->queryDuplicates(), mc->queryResends());
+        }
         mc.clear(); // Or we won't free memory for graphs that get recreated
         mu.clear(); //ditto
         deferredStart = false;
@@ -17516,6 +17520,7 @@ public:
             DBGLOG("activityid = %d  isKeyed = %d  line = %d", activityId, isKeyed, __LINE__);
         helper.onLimitExceeded();
     }
+
     virtual const void *createLimitFailRow(bool isKeyed)
     {
         UNIMPLEMENTED; // MORE - is there an ONFAIL for a limit folded into a remote?

roxie/ccd/ccdserver.hpp

@@ -187,6 +187,7 @@ interface IRoxieServerActivity : extends IActivityBase
     virtual ISectionTimer * registerTimer(unsigned activityId, const char * name) = 0;
     virtual IEngineRowAllocator * createRowAllocator(IOutputMetaData * metadata) = 0;
     virtual void gatherStatistics(IStatisticGatherer * statsBuilder) const = 0;
+    virtual void noteStatistic(StatisticKind kind, unsigned __int64 value) const = 0;
 };
 
 interface IRoxieServerActivityFactory : extends IActivityFactory

roxie/ccd/ccdsnmp.cpp

@@ -1023,7 +1023,7 @@ public:
                 CriticalBlock b(statsLock);
                 for (auto thisSlot: aggregated)
                 {
-                    if (thisSlot.inRange(from, to))
+                    if (thisSlot.timeOverlap(from, to))
                         aggregator.mergeStats(thisSlot);
                     else if (thisSlot.older(from))
                         break;

roxie/ccdcache/ccdcache.cpp

@@ -173,7 +173,7 @@ class StandaloneCacheWarmer : implements ICacheWarmer
     unsigned pagesTouched = 0;
     char *file_mmap = nullptr;
     int fd = -1;
-    struct stat file_stat;
+    struct stat file_stat = {};
     char dummy = 0;
 
     void warmRange(offset_t startOffset, offset_t endOffset)
@@ -197,9 +197,8 @@ public:
     {
         file_mmap = nullptr;
         fd = open(filename, 0);
-        if (fd != -1)
+        if (fd != -1 && fstat(fd, &file_stat)==0)
         {
-            fstat(fd, &file_stat);
             file_mmap = (char *) mmap((void *)0, file_stat.st_size, PROT_READ, MAP_SHARED, fd, 0);
             if (file_mmap == MAP_FAILED)
             {
@@ -298,6 +297,8 @@ int main(int argc, const char **argv)
             cacheFileName = argv[arg];
         arg++;
     }
+    if (!cacheFileName)
+        usage();
     StringBuffer cacheInfo;
     install_signal_handlers();
     StandaloneCacheWarmer warmer(traceLevel);

system/jlib/jmetrics.cpp

@@ -28,19 +28,11 @@ MODULE_EXIT()
 }
 
 
-struct hpccMetrics::SinkInfo
-{
-    explicit SinkInfo(MetricSink *_pSink) : pSink{_pSink} {}
-    MetricSink *pSink = nullptr;             // ptr to the sink
-    std::vector<std::string> reportMetrics;   // vector of metrics to report (empty for none)
-};
-
 MetricsReporter &hpccMetrics::queryMetricsReporter()
 {
     return *metricsReporter.query([] { return new MetricsReporter; });
 }
 
-
 MetricsReporter::~MetricsReporter()
 {
     for (auto const &sinkIt : sinks)
@@ -202,6 +194,23 @@ MetricSink *MetricsReporter::getSinkFromLib(const char *type, const char *sinkNa
     return pSink;
 }
 
+// Method for use when testing
+void MetricsReporter::addSink(MetricSink *pSink, const char *name)
+{
+    //
+    // Add the sink if it does not already exist, otherwise delete the sink because
+    // we are taking ownership.
+    auto sinkIt = sinks.find(name);
+    if (sinkIt == sinks.end())
+    {
+        sinks.insert({std::string(name), std::unique_ptr<SinkInfo>(new SinkInfo(pSink))});
+    }
+    else
+    {
+        delete pSink;
+    }
+}
+
 
 PeriodicMetricSink::PeriodicMetricSink(const char *name, const char *type, const IPropertyTree *pSettingsTree) :
     MetricSink(name, type),

system/jlib/jmetrics.hpp

@@ -120,7 +120,7 @@ class jlib_decl CounterMetric : public MetricVal
 public:
     CounterMetric(const char *name, const char *description) :
         MetricVal{name, description, MetricType::METRICS_COUNTER}  { }
-    void inc(uint64_t val)
+    void inc(uint64_t val = 1)
     {
         value.fetch_add(val);
     }
@@ -218,7 +218,12 @@ protected:
 
 extern "C" { typedef hpccMetrics::MetricSink* (*getSinkInstance)(const char *, const IPropertyTree *pSettingsTree); }
 
-struct SinkInfo;
+struct SinkInfo
+{
+    explicit SinkInfo(MetricSink *_pSink) : pSink{_pSink} {}
+    MetricSink *pSink = nullptr;             // ptr to the sink
+    std::vector<std::string> reportMetrics;   // vector of metrics to report (empty for none)
+};
 
 class jlib_decl MetricsReporter
 {
@@ -226,6 +231,7 @@ public:
     MetricsReporter() = default;
     ~MetricsReporter();
     void init(IPropertyTree *pMetricsTree);
+    void addSink(MetricSink *pSink, const char *name);  // for use by unit tests
     void addMetric(const std::shared_ptr<IMetric> &pMetric);
     void startCollecting();
     void stopCollecting();

system/jlib/jstatcodes.h

@@ -235,6 +235,7 @@ enum StatisticKind
     StTimeBlocked,
     StCycleBlockedCycles,
     StCostExecute,
+    StSizeAgentReply,
     StMax,
 
     //For any quantity there is potentially the following variants.

system/jlib/jstats.cpp

@@ -956,6 +956,7 @@ static const StatisticMeta statsMetaData[StMax] = {
     { TIMESTAT(Blocked) },
     { CYCLESTAT(Blocked) },
     { STAT(Cost, Execute, SMeasureCost) },
+    { SIZESTAT(AgentReply) },
 };
 
 //Is a 0 value likely, and useful to be reported if it does happen to be zero?

system/security/LdapSecurity/ldapconnection.cpp

@@ -3400,6 +3400,12 @@ public:
             return false;
         }
 
+        if(isEmptyString(newPassword))
+        {
+            DBGLOG("CLdapClient::updateUserPassword password must be provided");
+            return false;
+        }
+
         if (currPassword)
         {
             //User will not be authenticated if their password was expired,
@@ -3421,6 +3427,12 @@ public:
             return false;
         }
 
+        if(isEmptyString(newPassword))
+        {
+            DBGLOG("CLdapClient::updateUserPassword password must be provided");
+            return false;
+        }
+
         const char* sysuser = m_ldapconfig->getSysUser();
         if(sysuser && *sysuser && strcmp(username, sysuser) == 0)
             throw MakeStringException(-1, "You can't change password of the system user.");
@@ -5999,9 +6011,6 @@ private:
             act_ctrl_val |= 0x10000;
 #endif
 
-        // Ensure password required
-        act_ctrl_val &= ~(0x20);// UF_PASSWD_NOTREQD 0x0020
-
         StringBuffer new_act_ctrl;
         new_act_ctrl.append(act_ctrl_val);
 
@@ -6023,15 +6032,29 @@ private:
 
         // set the password.
         Owned<ISecUser> tmpuser = new CLdapSecUser(user->getName(), "");
-        const char* passwd = user->credentials().getPassword();
-        if(passwd == NULL || *passwd == '\0')
-            passwd = "password";
-
-        if (!updateUserPassword(*tmpuser, passwd, NULL))
+        if (!updateUserPassword(*tmpuser, user->credentials().getPassword(), nullptr))
         {
             DBGLOG("Error updating password for %s",username);
             throw MakeStringException(-1, "Error updating password for %s",username);
         }
+
+        //Now that the password is set, we can ensure passwords are always required
+
+        act_ctrl_val &= ~(0x20);// UF_PASSWD_NOTREQD 0x0020
+        new_act_ctrl.clear().append(act_ctrl_val);
+        LDAPMod attr = {
+            LDAP_MOD_REPLACE,
+            "userAccountControl",
+            ctrl_values
+        };
+        cattrs[0] = &attr;
+        cattrs[1] = NULL;
+        rc = ldap_modify_ext_s(ld, (char*)dn, cattrs, NULL, NULL);
+        if ( rc != LDAP_SUCCESS )
+        {
+            throw MakeStringException(-1, "error enableUser2 %s, ldap_modify_ext_s error2: %s", username, ldap_err2string( rc ));
+        }
+
     }
 
 
@@ -6044,6 +6067,12 @@ private:
             throw MakeStringException(-1, "Can't add user, username not set");
         }
 
+        const char* userPassword = user.credentials().getPassword();
+        if(isEmptyString(userPassword))
+        {
+            throw MakeStringException(-1, "Can't add user, password not set");
+        }
+
         const char* fname = user.getFirstName();
         const char* lname = user.getLastName();
         if((lname == NULL || *lname == '\0') && (fname == NULL || *fname == '\0' || m_ldapconfig->getServerType() == IPLANET))
@@ -6134,10 +6163,7 @@ private:
             actname_values
         };
 
-        const char* passwd = user.credentials().getPassword();
-        if(passwd == NULL || *passwd == '\0')
-            passwd = "password";
-        char* passwd_values[] = {(char*)passwd, NULL};
+        char* passwd_values[] = {(char*)userPassword, nullptr};//password is set later (in enableUser) if ACTIVE_DIRECTORY
         LDAPMod passwd_attr =
         {
             LDAP_MOD_ADD,

testing/unittests/CMakeLists.txt

@@ -47,6 +47,8 @@ set (    SRCS
          configmgr/ConfigMgrUnitTests.cpp
          configmgr/ConfigMgrTemplateTests.cpp
          configmgr/ConfigMgrHPCCTests.cpp
+         metrics/MetricFrameworkTests.cpp
+         metrics/PeriodicSinkTests.cpp
     )
 endif ()
 

testing/unittests/metrics/MetricFrameworkTests.cpp

@@ -0,0 +1,189 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2021 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.
+############################################################################## */
+
+#ifdef _USE_CPPUNIT
+
+#include <cppunit/TestFixture.h>
+#include "unittests.hpp"
+#include <algorithm>
+
+#include "jmetrics.hpp"
+
+using namespace hpccMetrics;
+
+class MetricFrameworkTestSink : public MetricSink
+{
+public:
+    explicit MetricFrameworkTestSink(const char *name) :
+        MetricSink(name, "test") { }
+
+    ~MetricFrameworkTestSink() override = default;
+
+    void startCollection(MetricsReporter *_pReporter) override
+    {
+        pReporter = _pReporter;
+        isCollecting = true;
+    }
+
+    void stopCollection() override
+    {
+        isCollecting = false;
+    }
+
+    std::vector<std::shared_ptr<IMetric>> getReportMetrics() const
+    {
+        return pReporter->queryMetricsForReport(name);
+    }
+
+public:
+    bool isCollecting = false;
+};
+
+
+class MetricFrameworkTests : public CppUnit::TestFixture
+{
+public:
+    MetricFrameworkTests()
+    {
+        pTestSink = new MetricFrameworkTestSink("testsink");
+        frameworkTestReporter.addSink(pTestSink, "testsink");
+    }
+
+    ~MetricFrameworkTests() = default;
+
+    CPPUNIT_TEST_SUITE(MetricFrameworkTests);
+        CPPUNIT_TEST(Test_counter_metric_increments_properly);
+        CPPUNIT_TEST(Test_gauge_metric_updates_properly);
+        CPPUNIT_TEST(Test_custom_metric);
+        CPPUNIT_TEST(Test_reporter_calls_sink_to_start_and_stop_collection);
+        CPPUNIT_TEST(Test_reporter_manages_metrics_properly);
+    CPPUNIT_TEST_SUITE_END();
+
+protected:
+
+
+    void Test_counter_metric_increments_properly()
+    {
+        std::shared_ptr<CounterMetric> pCounter = std::make_shared<CounterMetric>("test-counter", "description");
+        CPPUNIT_ASSERT_EQUAL(0, static_cast<int>(pCounter->queryValue()));
+
+        //
+        // Test default increment (by 1)
+        pCounter->inc();
+        int counterValue = pCounter->queryValue();
+        CPPUNIT_ASSERT_EQUAL(1, counterValue);
+
+        //
+        // Test increment by > 1
+        pCounter->inc(2);
+        counterValue = pCounter->queryValue();
+        CPPUNIT_ASSERT_EQUAL(3, counterValue);
+    }
+
+
+    void Test_gauge_metric_updates_properly()
+    {
+        std::shared_ptr<GaugeMetric> pGauge = std::make_shared<GaugeMetric>("test-gauge", "description");
+        int gaugeValue = pGauge->queryValue();
+        CPPUNIT_ASSERT_EQUAL(0, gaugeValue);
+
+        //
+        // Test initial setting of gauge
+        pGauge->set(25);
+        gaugeValue = pGauge->queryValue();
+        CPPUNIT_ASSERT_EQUAL(25, gaugeValue);
+
+        //
+        // Test updating gauge
+        pGauge->add(10);
+        gaugeValue = pGauge->queryValue();
+        CPPUNIT_ASSERT_EQUAL(35, gaugeValue);
+
+        pGauge->add(-5);
+        gaugeValue = pGauge->queryValue();
+        CPPUNIT_ASSERT_EQUAL(30, gaugeValue);
+    }
+
+
+    void Test_custom_metric()
+    {
+        int customCounter = 0;
+        std::shared_ptr<CustomMetric<int>> pCustomCounter = std::make_shared<CustomMetric<int>>("custom-counter", "description", METRICS_COUNTER, customCounter);
+        int customCounterValue = pCustomCounter->queryValue();
+        CPPUNIT_ASSERT_EQUAL(0, customCounterValue);
+
+        customCounter++;
+        customCounterValue = pCustomCounter->queryValue();
+        CPPUNIT_ASSERT_EQUAL(1, customCounterValue);
+    }
+
+
+    void Test_reporter_calls_sink_to_start_and_stop_collection()
+    {
+        frameworkTestReporter.startCollecting();
+        CPPUNIT_ASSERT_EQUAL(true, pTestSink->isCollecting);
+        frameworkTestReporter.stopCollecting();
+        CPPUNIT_ASSERT_EQUAL(false, pTestSink->isCollecting);
+    }
+
+
+    void Test_reporter_manages_metrics_properly()
+    {
+        int numAdded;
+        std::shared_ptr<CounterMetric> pCounter = std::make_shared<CounterMetric>("test-counter", "description");
+        std::shared_ptr<GaugeMetric> pGauge = std::make_shared<GaugeMetric>("test-gauge", "description");
+        frameworkTestReporter.addMetric(pCounter);
+        frameworkTestReporter.addMetric(pGauge);
+        numAdded = 2;
+
+        frameworkTestReporter.startCollecting();
+
+        //
+        // Make sure the initial list is correct
+        int numMetrics = frameworkTestReporter.queryMetricsForReport("testsink").size();
+        CPPUNIT_ASSERT_EQUAL(numAdded, numMetrics);
+
+        //
+        // Add a metric while reporting is enabled and make sure it is returned
+        std::shared_ptr<CounterMetric> pNewCounter = std::make_shared<CounterMetric>("test-newcounter", "description");
+        frameworkTestReporter.addMetric(pNewCounter);
+        numAdded++;
+
+        numMetrics = frameworkTestReporter.queryMetricsForReport("testsink").size();
+        CPPUNIT_ASSERT_EQUAL(numAdded, numMetrics);
+
+        //
+        // Destroy a metric and ensure it is no longer in the list of report metrics
+        pNewCounter = nullptr;
+        numAdded--;
+
+        numMetrics = frameworkTestReporter.queryMetricsForReport("testsink").size();
+        CPPUNIT_ASSERT_EQUAL(numAdded, numMetrics);
+
+        frameworkTestReporter.stopCollecting();
+    }
+
+protected:
+    MetricsReporter frameworkTestReporter;
+    MetricFrameworkTestSink *pTestSink;
+
+};
+
+CPPUNIT_TEST_SUITE_REGISTRATION( MetricFrameworkTests );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( MetricFrameworkTests, "MetricFrameworkTests" );
+
+#endif

testing/unittests/metrics/PeriodicSinkTests.cpp

@@ -0,0 +1,123 @@
+/*##############################################################################
+
+    HPCC SYSTEMS software Copyright (C) 2021 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.
+############################################################################## */
+
+#ifdef _USE_CPPUNIT
+
+#include <cppunit/TestFixture.h>
+#include "unittests.hpp"
+#include <algorithm>
+
+#include "jptree.hpp"
+#include "jmetrics.hpp"
+
+using namespace hpccMetrics;
+
+class PeriodicTestSink : public PeriodicMetricSink
+{
+public:
+    explicit PeriodicTestSink(const char *name, const IPropertyTree *pSettingsTree) :
+        PeriodicMetricSink(name, "test", pSettingsTree) { }
+
+    ~PeriodicTestSink() = default;
+
+protected:
+    virtual void prepareToStartCollecting() override
+    {
+        prepareCalled = true;
+        numCollections = 0;
+    }
+
+    virtual void collectingHasStopped() override
+    {
+        stopCollectionNotificationCalled = true;
+    }
+
+    void doCollection() override
+    {
+        numCollections++;
+    }
+
+public:
+    bool prepareCalled = false;
+    bool stopCollectionNotificationCalled = false;
+    unsigned numCollections = 0;
+};
+
+
+const char *periodicSinkSettingsTestYml = R"!!(period: 2
+)!!";
+
+static const unsigned period = 2;
+
+
+class PeriodicSinkTests : public CppUnit::TestFixture
+{
+public:
+    PeriodicSinkTests()
+    {
+        //
+        // Load the settings then set the period using the global var
+        Owned<IPropertyTree> pSettings = createPTreeFromYAMLString(periodicSinkSettingsTestYml, ipt_none, ptr_ignoreWhiteSpace, nullptr);
+        pSettings->setPropInt("@period", period);
+        pPeriodicTestSink = new PeriodicTestSink("periodic_test_sink", pSettings);
+        periodicSinkTestReporter.addSink(pPeriodicTestSink, "periodic_test_sink");
+    }
+
+    ~PeriodicSinkTests() = default;
+
+    CPPUNIT_TEST_SUITE(PeriodicSinkTests);
+        CPPUNIT_TEST(Test_sets_period_correctly);
+    CPPUNIT_TEST_SUITE_END();
+
+protected:
+
+    void Test_sets_period_correctly()
+    {
+        //
+        // To test setting the period correctly, start collection and delay a multiple of that period.
+        // Stop collection and ask the test sink how many collections were done. If the count is +/- 1
+        // from the wait period multiple used, then we are close enough
+        unsigned multiple = 3;
+        periodicSinkTestReporter.startCollecting();
+
+        //
+        // Check that the sink called to prepare for collection
+        CPPUNIT_ASSERT(pPeriodicTestSink->prepareCalled);
+
+        // wait... then stop collecting
+        sleep(multiple * period);
+        periodicSinkTestReporter.stopCollecting();
+
+        unsigned numCollections = pPeriodicTestSink->numCollections;
+
+        bool numReportsCorrect = (numCollections >= multiple-1) && (numCollections <= multiple+1);
+        CPPUNIT_ASSERT_EQUAL(true, numReportsCorrect);
+
+        //
+        // Verify collection was stopped
+        CPPUNIT_ASSERT(pPeriodicTestSink->stopCollectionNotificationCalled);
+    }
+
+protected:
+    MetricsReporter periodicSinkTestReporter;
+    PeriodicTestSink *pPeriodicTestSink = nullptr;
+};
+
+CPPUNIT_TEST_SUITE_REGISTRATION( PeriodicSinkTests );
+CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( PeriodicSinkTests, "PeriodicSinkTests" );
+
+#endif

thorlcr/activities/fetch/thfetch.cpp

@@ -20,6 +20,7 @@
 #include "thbufdef.hpp"
 #include "mptag.hpp"
 #include "dadfs.hpp"
+#include "jstats.h"
 #include "thexception.hpp"
 
 #include "../hashdistrib/thhashdistrib.ipp"
@@ -30,12 +31,13 @@ class CFetchActivityMaster : public CMasterActivity
     Owned<CSlavePartMapping> mapping;
     MemoryBuffer offsetMapMb;
     SocketEndpoint *endpoints;
+    std::vector<OwnedPtr<CThorStatsCollection>> subFileStats;
 
 protected:
     IHThorFetchArg *helper;
 
 public:
-    CFetchActivityMaster(CMasterGraphElement *info) : CMasterActivity(info)
+    CFetchActivityMaster(CMasterGraphElement *info) : CMasterActivity(info, diskReadActivityStatistics)
     {
         endpoints = NULL;
         if (!container.queryLocalOrGrouped())
@@ -73,7 +75,12 @@ public:
             }
             else if (encrypted)
                 throw MakeActivityException(this, 0, "File '%s' was published as encrypted but no encryption key provided", fetchFile->queryLogicalName());
-            mapping.setown(getFileSlaveMaps(fetchFile->queryLogicalName(), *fileDesc, container.queryJob().queryUserDescriptor(), container.queryJob().querySlaveGroup(), container.queryLocalOrGrouped(), false, NULL, fetchFile->querySuperFile()));
+            IDistributedSuperFile *super = fetchFile->querySuperFile();
+            unsigned numsubs = super?super->numSubFiles(true):0;
+            for (unsigned i=0; i<numsubs; i++)
+                subFileStats.push_back(new CThorStatsCollection(diskReadActivityStatistics));
+
+            mapping.setown(getFileSlaveMaps(fetchFile->queryLogicalName(), *fileDesc, container.queryJob().queryUserDescriptor(), container.queryJob().querySlaveGroup(), container.queryLocalOrGrouped(), false, NULL, super));
             mapping->serializeFileOffsetMap(offsetMapMb);
             addReadFile(fetchFile);
         }
@@ -93,6 +100,32 @@ public:
         if (!container.queryLocalOrGrouped())
             dst.append((int)mpTag);
     }
+    virtual void deserializeStats(unsigned node, MemoryBuffer &mb) override
+    {
+        CMasterActivity::deserializeStats(node, mb);
+        for (auto &stats: subFileStats)
+            stats->deserialize(node, mb);
+    }
+    virtual void done() override
+    {
+        if (!subFileStats.empty())
+        {
+            unsigned numSubFiles = subFileStats.size();
+            for (unsigned i=0; i<numSubFiles; i++)
+            {
+                IDistributedFile *file = queryReadFile(i);
+                if (file)
+                    file->addAttrValue("@numDiskReads", subFileStats[i]->getStatisticSum(StNumDiskReads));
+            }
+        }
+        else
+        {
+            IDistributedFile *file = queryReadFile(0);
+            if (file)
+                file->addAttrValue("@numDiskReads", statsCollection.getStatisticSum(StNumDiskReads));
+        }
+        CMasterActivity::done();
+    }
 };
 
 class CCsvFetchActivityMaster : public CFetchActivityMaster

thorlcr/activities/fetch/thfetchslave.cpp

@@ -159,7 +159,7 @@ public:
                 e->base = part.queryProperties().getPropInt64("@offset");
                 e->top = e->base + part.queryProperties().getPropInt64("@size");
                 e->index = f;
-                e->file = queryThor().queryFileCache().lookupIFileIO(owner, logicalFilename, part); // NB: freed by FPosTableEntryIFileIO dtor
+                e->file = queryThor().queryFileCache().lookupIFileIO(owner, logicalFilename, part, nullptr, diskReadActivityStatistics); // NB: freed by FPosTableEntryIFileIO dtor
             }
         }
     }
@@ -262,6 +262,32 @@ public:
         }
         return NULL;
     }
+    virtual void getFileStats(CRuntimeStatisticCollection & stats) override
+    {
+        for (unsigned f=0; f<files; f++)
+        {
+            IFileIO *file = fPosMultiPartTable[f].file;
+            mergeStats(stats, file);
+        }
+    }
+    virtual void getSubFileStats(std::vector<OwnedPtr<CRuntimeStatisticCollection>> & subFileStats) override
+    {
+        if (subFileStats.size()>0)
+        {
+            ISuperFileDescriptor *super = parts.item(0).queryOwner().querySuperFileDescriptor();
+            dbgassertex(super);
+            for (unsigned f=0; f<files; f++)
+            {
+                IPartDescriptor &part = parts.item(f);
+                unsigned subfile, lnum;
+                if(super->mapSubPart(part.queryPartIndex(), subfile, lnum))
+                {
+                    IFileIO *file = fPosMultiPartTable[f].file;
+                    mergeStats(*subFileStats[subfile], file);
+                }
+            }
+        }
+    }
 };
 
 
@@ -281,6 +307,7 @@ class CFetchSlaveBase : public CSlaveActivity, implements IFetchHandler
     MemoryBuffer offsetMapBytes;
     Owned<IExpander> eexp;
     Owned<IEngineRowAllocator> keyRowAllocator;
+    std::vector<OwnedPtr<CRuntimeStatisticCollection>> subFileStats;
 
 protected:
     Owned<IThorRowInterfaces> fetchDiskRowIf;
@@ -299,7 +326,7 @@ protected:
 public:
     IMPLEMENT_IINTERFACE_USING(CSlaveActivity);
 
-    CFetchSlaveBase(CGraphElementBase *_container) : CSlaveActivity(_container)
+    CFetchSlaveBase(CGraphElementBase *_container) : CSlaveActivity(_container, diskReadActivityStatistics)
     {
         fetchBaseHelper = (IHThorFetchBaseArg *)queryHelper();
         reInit = 0 != (fetchBaseHelper->getFetchFlags() & (FFvarfilename|FFdynamicfilename));
@@ -359,6 +386,10 @@ public:
                     prefetchers.append(prefetcher.getClear());
                 }
             }
+            ISuperFileDescriptor *super = parts.item(0).queryOwner().querySuperFileDescriptor();
+            if (super)
+                for (unsigned i=0; i<files; i++)
+                    subFileStats.push_back(new CRuntimeStatisticCollection(diskReadActivityStatistics));
         }
 
         unsigned encryptedKeyLen;
@@ -541,6 +572,17 @@ public:
             return fpos;
         }
     }
+    virtual void serializeStats(MemoryBuffer &mb) override
+    {
+        if (fetchStream)
+        {
+            fetchStream->getFileStats(stats);
+            fetchStream->getSubFileStats(subFileStats);
+        }
+        PARENT::serializeStats(mb);
+        for (auto &stats: subFileStats)
+            stats->serialize(mb);
+    }
     virtual void onLimitExceeded() = 0;
 };
 

thorlcr/activities/fetch/thfetchslave.ipp

@@ -37,6 +37,8 @@ interface IFetchStream : extends IInterface
     virtual IFileIO *getPartIO(unsigned part) = 0;
     virtual StringBuffer &getPartName(unsigned part, StringBuffer &out) = 0;
     virtual void abort() = 0;
+    virtual void getFileStats(CRuntimeStatisticCollection & stats) = 0;
+    virtual void getSubFileStats(std::vector<OwnedPtr<CRuntimeStatisticCollection>> & subFileStats) = 0;
 };
 
 IFetchStream *createFetchStream(CSlaveActivity &owner, IThorRowInterfaces *keyRowIf, IThorRowInterfaces *fetchRowIf, bool &abortSoon, const char *logicalFilename, CPartDescriptorArray &parts, unsigned offsetCount, size32_t offsetMapSz, const void *offsetMap, IFetchHandler *iFetchHandler, mptag_t tag, IExpander *eexp=NULL);

thorlcr/mfilemanager/thmfilemanager.cpp

@@ -226,13 +226,6 @@ public:
     CFileManager()
     {
         replicateOutputs = globals->getPropBool("@replicateOutputs");
-
-        /* In nas/non-local storage mode, create a published named group for files to use
-         * that matches the width of the cluster.
-         * Also create a 1-way named group, that is used in special cases, e.g. BUILDINDEX,FEW
-         */
-        if (isContainerized())
-            queryNamedGroupStore().ensureNasGroup(queryClusterWidth());
     }
     StringBuffer &mangleLFN(CJobBase &job, const char *lfn, StringBuffer &out)
     {