HPCC-29768 Protect updates with a dali lock when fetching a git repo

common/fileview2/fvtransform.cpp

@@ -295,7 +295,7 @@ void ViewTransformerRegistry::addPlugins(const char * name)
     loadedPlugins->loadFromList(name);
 
     Owned<IErrorReceiver> errorReporter = createThrowingErrorReceiver();
-    EclRepositoryManager collection;
+    EclRepositoryManager collection(nullptr);
     collection.addSharedSourceFileEclRepository(errorReporter, name, ESFallowplugins|ESFnodependencies, 0, false);
     dataServer.setown(collection.createPackage(nullptr));
 

ecl/eclcc/eclcc.cpp

@@ -207,6 +207,8 @@ struct EclCompileInstance : public CInterfaceOf<ICodegenContextCallback>
     virtual IHqlExpression *lookupDFSlayout(const char *filename, IErrorReceiver &errs, const ECLlocation &location, bool isOpt) const override;
     virtual unsigned lookupClusterSize() const override;
     virtual void getTargetPlatform(StringBuffer & result) override;
+    virtual IInterface * getGitUpdateLock(const char * key) override;
+
 
 public:
     EclCC & eclcc;
@@ -239,11 +241,11 @@ struct EclCompileInstance : public CInterfaceOf<ICodegenContextCallback>
     Linked<IErrorReceiver> errorProcessor;
 };
 
-class EclCC
+class EclCC final : implements CUnsharedInterfaceOf<ICodegenContextCallback>
 {
 public:
     EclCC(int _argc, const char **_argv)
-        : programName(_argv[0])
+        : programName(_argv[0]), repositoryManager(this)
     {
         argc = _argc;
         argv = _argv;
@@ -284,12 +286,14 @@ class EclCC
 
     // interface ICodegenContextCallback
 
-    void pushCluster(const char *clusterName);
-    void popCluster();
-    bool allowAccess(const char * category, bool isSigned);
-    IHqlExpression *lookupDFSlayout(const char *filename, IErrorReceiver &errs, const ECLlocation &location, bool isOpt) const;
-    unsigned lookupClusterSize() const;
-    void getTargetPlatform(StringBuffer & result);
+    virtual void noteCluster(const char *clusterName) override;
+    virtual void pushCluster(const char *clusterName) override;
+    virtual void popCluster() override;
+    virtual bool allowAccess(const char * category, bool isSigned) override;
+    virtual IHqlExpression *lookupDFSlayout(const char *filename, IErrorReceiver &errs, const ECLlocation &location, bool isOpt) const override;
+    virtual unsigned lookupClusterSize() const override;
+    virtual void getTargetPlatform(StringBuffer & result) override;
+    virtual IInterface * getGitUpdateLock(const char * key) override;
 
 protected:
     void appendNeverSimplifyList(const char *attribsList);
@@ -387,6 +391,7 @@ class EclCC
     StringAttr optMetaLocation;
     StringBuffer neverSimplifyRegEx;
     StringAttr optDefaultGitPrefix;
+    StringAttr optGitLock; // A key used to lock access to git updates
     StringAttr optGitUser;
     StringAttr optGitPasswordPath;
 
@@ -1719,7 +1724,7 @@ void EclCC::processXmlFile(EclCompileInstance & instance, const char *archiveXML
     if (optCheckEclVersion)
         instance.checkEclVersionCompatible();
 
-    EclRepositoryManager localRepositoryManager;
+    EclRepositoryManager localRepositoryManager(&instance);
     processDefinitions(localRepositoryManager);
     localRepositoryManager.inherit(repositoryManager); // Definitions, plugins, std library etc.
     Owned<IFileContents> contents;
@@ -1839,7 +1844,7 @@ void EclCC::processFile(EclCompileInstance & instance)
                 attributePackage = optDefaultRepo.str();
         }
 
-        EclRepositoryManager localRepositoryManager;
+        EclRepositoryManager localRepositoryManager(&instance);
         processDefinitions(localRepositoryManager);
         localRepositoryManager.inherit(repositoryManager); // don't include -I
         if (!optNoBundles)
@@ -2066,7 +2071,7 @@ void EclCC::processReference(EclCompileInstance & instance, const char * queryAt
     if (optArchive || optGenerateDepend || optSaveQueryArchive)
         instance.archive.setown(createAttributeArchive());
 
-    EclRepositoryManager localRepositoryManager;
+    EclRepositoryManager localRepositoryManager(&instance);
     processDefinitions(localRepositoryManager);
     localRepositoryManager.inherit(repositoryManager);
     if (!optNoBundles)
@@ -2381,6 +2386,7 @@ bool EclCompileInstance::reportErrorSummary()
 
 void EclCompileInstance::noteCluster(const char *clusterName)
 {
+    eclcc.noteCluster(clusterName);
 }
 
 void EclCompileInstance::pushCluster(const char *clusterName)
@@ -2398,6 +2404,12 @@ unsigned EclCompileInstance::lookupClusterSize() const
     return eclcc.lookupClusterSize();
 }
 
+IInterface * EclCompileInstance::getGitUpdateLock(const char * key)
+{
+    return eclcc.getGitUpdateLock(key);
+}
+
+
 bool EclCompileInstance::allowAccess(const char * category, bool isSigned)
 {
     return eclcc.allowAccess(category, isSigned);
@@ -2443,6 +2455,10 @@ void EclCC::appendNeverSimplifyList(const char *attribsList)
     }
 }
 
+void EclCC::noteCluster(const char *clusterName)
+{
+}
+
 void EclCC::pushCluster(const char *clusterName)
 {
     clusters.append(clusterName);
@@ -2460,6 +2476,9 @@ bool EclCC::checkDaliConnected() const
 {
     if (!daliConnected)
     {
+        if (isEmptyString(optDFS) || disconnectReported)
+            return false;
+
         try
         {
             Owned<IGroup> serverGroup = createIGroup(optDFS.str(), DALI_SERVER_PORT);
@@ -2489,7 +2508,7 @@ unsigned EclCC::lookupClusterSize() const
 {
     CriticalBlock b(dfsCrit);  // Overkill at present but maybe one day codegen will start threading? If it does the stack is also iffy!
 #ifndef _CONTAINERIZED
-    if (!optDFS || disconnectReported || !checkDaliConnected())
+    if (!checkDaliConnected())
         return 0;
 #endif
     if (prevClusterSize != -1)
@@ -2515,10 +2534,56 @@ unsigned EclCC::lookupClusterSize() const
     return prevClusterSize;
 }
 
+IInterface * EclCC::getGitUpdateLock(const char * path)
+{
+    if (optGitLock.isEmpty())
+        return nullptr;
+
+    VStringBuffer lockPath("/GitUpdateLocks/%s/hash%llx", optGitLock.str(), rtlHash64VStr(path, HASH64_INIT));
+
+    CriticalBlock b(dfsCrit);
+    if (!checkDaliConnected())
+        return nullptr;
+
+    DBGLOG("Get git update lock for '%s':'%s'", optGitLock.str(), path);
+    const unsigned lockTimeout = 30 * 60 * 1000; // 30 minutes - fetches from git can take a long time
+    const unsigned connectTimeout = 3 * 1000;
+    unsigned traceTimeout = connectTimeout * 2;
+    CCycleTimer elapsed;
+    for (;;)
+    {
+        try
+        {
+            unsigned remaining = elapsed.remainingMs(lockTimeout);
+            if (remaining == 0)
+                break;
+
+            Owned<IInterface> connection = querySDS().connect(lockPath, myProcessSession(), RTM_LOCK_WRITE|RTM_CREATE_QUERY, connectTimeout);
+            if (connection)
+                return connection.getClear();
+        }
+        catch (IException * e)
+        {
+            unsigned errcode = e->errorCode();
+            e->Release();
+            if (errcode != SDSExcpt_LockTimeout)
+                break;
+        }
+        if (elapsed.elapsedMs() >= traceTimeout)
+        {
+            DBGLOG("Blocked waiting for a git update lock on '%s' for %u seconds", path, elapsed.elapsedMs() / 1000);
+            traceTimeout *= 2;
+        }
+    }
+    DBGLOG("Failed to get git update lock for '%s'", path);
+    return nullptr;
+}
+
+
 IHqlExpression *EclCC::lookupDFSlayout(const char *filename, IErrorReceiver &errs, const ECLlocation &location, bool isOpt) const
 {
     CriticalBlock b(dfsCrit);  // Overkill at present but maybe one day codegen will start threading?
-    if (!optDFS || disconnectReported)
+    if (isEmptyString(optDFS) || disconnectReported)
     {
         // Dali lookup disabled, yet translation requested. Should we report if OPT set?
         if (!(optArchive || optGenerateDepend || optSyntax || optGenerateMeta || optEvaluateResult || disconnectReported))
@@ -2842,6 +2907,9 @@ int EclCC::parseCommandLineOptions(int argc, const char* argv[])
         {
             optScope.set(tempArg);
         }
+        else if (iter.matchOption(optGitLock, "--gitlock"))
+        {
+        }
         else if (iter.matchOption(optGitUser, "--gituser"))
         {
         }

ecl/eclcc/eclcc.hpp

@@ -96,6 +96,7 @@ const char * const helpText[] = {
     "?!  --fastsyntax  Delay expanding functions when parsing.  May speed up processing for some queries",
     "?   --fetchrepos  Automatically download missing repositories associated with dependencies",
     "!   --gituser=x   Which user should be used for accessing git repositories (for servers)",
+    "!   --gitlock=key The dali key (e.g. plane name) that should be used to protect updates to git repositories",
     "    -help, --help Display this message",
     "    -help -v      Display verbose help message",
     "!   --ignoresimplified Do not use simplified expressions when syntax checking",

ecl/eclccserver/eclccserver.cpp

@@ -235,6 +235,45 @@ static bool getHomeFolder(StringBuffer & homepath)
     return true;
 }
 
+static bool guardGitUpdates = false;
+static StringBuffer gitLockKey;
+static void configGitLock()
+{
+    Owned<IPropertyTree> config = getComponentConfig();
+    if (config->getPropBool("@enableEclccDali", true))
+    {
+        if (config->getPropBool("@guardGitUpdates", true))
+        {
+            if (isContainerized())
+            {
+                //Containerized: each git plane needs to be protected independently
+                gitLockKey.append(config->queryProp("@gitPlane"));
+            }
+            else
+            {
+                //Bare metal - git repos are fetched locally, so protect per host-ip
+                const char * hostname = GetCachedHostName();
+                if (hostname)
+                {
+                    gitLockKey.append("host");
+
+                    for (const byte * cur = (const byte *)hostname; *cur; cur++)
+                    {
+                        //remove '.' and other unsupported characters from the key name
+                        if (isalnum(*cur))
+                            gitLockKey.append(*cur);
+                        else
+                            gitLockKey.append("_");
+                    }
+                }
+            }
+
+            if (!gitLockKey.isEmpty())
+                guardGitUpdates = true;
+        }
+    }
+}
+
 class EclccCompileThread : implements IPooledThread, implements IErrorReporter, public CInterface
 {
     StringAttr wuid;
@@ -644,6 +683,9 @@ class EclccCompileThread : implements IPooledThread, implements IErrorReporter,
         if (!repoRootPath.isEmpty())
             eclccCmd.appendf(" \"--repocachepath=%s\"", repoRootPath.str());
 
+        if (guardGitUpdates)
+            eclccCmd.appendf(" \"--gitlock=%s\"", gitLockKey.str());
+
         if (config->queryProp("@defaultRepo"))
             eclccCmd.appendf(" --defaultrepo=%s", config->queryProp("@defaultRepo"));
         if (config->queryProp("@defaultRepoVersion"))
@@ -835,7 +877,13 @@ class EclccCompileThread : implements IPooledThread, implements IErrorReporter,
             if (GetCurrentDirectory(sizeof(dir), dir))
                 repoRootPath.append(dir);
         }
-        if (repoRootPath.length())
+
+        if (guardGitUpdates)
+        {
+            addPathSepChar(repoRootPath).append("repos");
+            recursiveCreateDirectory(repoRootPath.str());
+        }
+        else if (repoRootPath.length())
         {
             addPathSepChar(repoRootPath).append("repos_").append(idxStr);
             recursiveCreateDirectory(repoRootPath.str());
@@ -1421,6 +1469,7 @@ int main(int argc, const char *argv[])
     {
         initClientProcess(serverGroup, DCR_EclCCServer);
         openLogFile();
+        configGitLock();
         const char *wuid = globals->queryProp("@workunit");
         if (wuid)
         {

ecl/hql/hql.hpp

@@ -226,6 +226,9 @@ interface ICodegenContextCallback : public IInterface
      * Which platform was this query originally targeted to?
      */
     virtual void getTargetPlatform(StringBuffer & result) = 0;
+    /*
+    */
+    virtual IInterface * getGitUpdateLock(const char * key) = 0;
 };
 
 

ecl/hql/hqlplugininfo.cpp

@@ -29,7 +29,7 @@ namespace repositoryCommon {
 IEclPackage * loadPlugins(const char * pluginPath)
 {
     MultiErrorReceiver errs;
-    EclRepositoryManager collection;
+    EclRepositoryManager collection(nullptr);
     collection.addQuerySourceFileEclRepository(&errs, pluginPath, ESFallowplugins|ESFnodependencies, (unsigned) -1);//Preload implicits/dlls
     if (errs.errCount())
     {

ecl/hql/hqlrepository.cpp

@@ -793,6 +793,7 @@ IEclSourceCollection * EclRepositoryManager::resolveGitCollection(const char * r
         throw makeStringExceptionV(99, "Unsupported repository link format '%s'", defaultUrl);
 
     bool alreadyExists = false;
+    Owned<IInterface> gitUpdateLock(getGitUpdateLock(repoPath));
     if (checkDirExists(repoPath))
     {
         if (options.cleanRepos)
@@ -855,6 +856,10 @@ IEclSourceCollection * EclRepositoryManager::resolveGitCollection(const char * r
             ok = true;
         }
     }
+    //All following operations are read-only and should not be affected if the git repo is updated behind the scenes
+    //this could become a read/write lock if that proved to be an issue.
+    gitUpdateLock.clear();
+
     gitDownloadCycles += gitDownloadTimer.elapsedCycles();
     if (error)
     {

ecl/hql/hqlrepository.hpp

@@ -38,7 +38,9 @@ class EclRepositoryMapping : public CInterface
 class HQL_API EclRepositoryManager
 {
 public:
-    EclRepositoryManager() = default;
+    EclRepositoryManager(ICodegenContextCallback * _callback) : callback(_callback)
+    {
+    }
     EclRepositoryManager(const EclRepositoryManager & other) = delete;
 
     void addNestedRepository(IIdAtom * scopeId, IEclSourceCollection * source, bool includeInArchive);
@@ -86,10 +88,17 @@ class HQL_API EclRepositoryManager
 
     unsigned runGitCommand(StringBuffer * output, const char *args, const char * cwd, bool needCredentials);
     IEclPackage * queryRepository(IIdAtom * name, const char * defaultUrl, IEclSourceCollection * overrideSource, bool includeDefinitions);
+    IInterface * getGitUpdateLock(const char * path)
+    {
+        if (!callback)
+            return nullptr;
+        return callback->getGitUpdateLock(path);
+    }
 
 private:
     mutable IErrorReceiver * errorReceiver = nullptr; // mutable to allow const methods to set it, it logically doesn't change the object
     using DependencyInfo = std::pair<std::string, Shared<IEclPackage>>;
+    ICodegenContextCallback * callback;
     CIArrayOf<EclRepositoryMapping> repos;
     std::vector<DependencyInfo> dependencies;
     IArrayOf<IEclRepository> sharedSources;     // plugins, std library, bundles

ecl/hqlcpp/hqlecl.cpp

@@ -150,6 +150,8 @@ class NullContextCallback : implements ICodegenContextCallback, public CInterfac
     virtual bool allowAccess(const char * category, bool isSigned) override { return true; }
     virtual IHqlExpression *lookupDFSlayout(const char *filename, IErrorReceiver &errs, const ECLlocation &location, bool isOpt) const override { return nullptr; }
     virtual unsigned lookupClusterSize() const override { return 0; }
+    virtual IInterface * getGitUpdateLock(const char * key) override { return nullptr; }
+
     virtual void getTargetPlatform(StringBuffer & result) override
     {
         workunit->getDebugValue("targetClusterType", StringBufferAdaptor(result));

helm/hpcc/templates/eclccserver.yaml

@@ -27,6 +27,7 @@ Pass in dict with root and me
 */}}
 {{- define "hpcc.eclccServerConfigMap" -}}
 {{- $compileJobName := printf "compile-job-_HPCC_JOBNAME_" }}
+{{- $gitPlane := .me.gitPlane | default (include "hpcc.getDefaultGitPlane" .root) }}
 apiVersion: v1
 metadata:
   name: {{ .me.name }}-configmap
@@ -37,6 +38,9 @@ data:
 {{ toYaml (omit .me "logging" "tracing") | indent 6 }}
 {{- include "hpcc.generateLoggingConfig" . | indent 6 }}
 {{- include "hpcc.generateTracingConfig" . | indent 6 }}
+{{- if $gitPlane }}
+      gitPlane: {{ $gitPlane }}
+{{- end }}
       queues:
 {{ include "hpcc.generateConfigMapQueues" .root | indent 6 }}
 {{ include "hpcc.generateVaultConfig" . | indent 6 }}

helm/hpcc/values.schema.json

@@ -1421,6 +1421,10 @@
           "type": "string",
           "description": "The default repo version used if not supplied for the defaultRepo"
         },
+        "guardGitUpdates": {
+          "type": "boolean",
+          "description": "If enabled all updates of the git repositories are protected by holding a lock in dali"
+        },
         "terminationGracePeriodSeconds": {
           "$ref": "#/definitions/terminationGracePeriodSeconds"
         },