Enable config monitoring for Thor in containerized deployments

thorlcr/THOR_CONFIG_MONITORING.md

@@ -0,0 +1,59 @@
+# Thor Config Monitoring Implementation
+
+## Overview
+This implementation enables configuration monitoring for Thor (manager and workers) in containerized deployments, allowing "soft" configuration changes (e.g., logging levels) to be applied without restarting Thor.
+
+## Problem Solved
+Previously, Thor components loaded configuration without monitoring (monitor=false), meaning any configuration changes required a restart. While other components could auto-reload config, Thor couldn't because:
+- The manager sends additional settings to workers during registration
+- If workers auto-reloaded config, they would lose these manager-provided settings
+
+## Solution
+The solution reorganizes how Thor handles configuration in containerized mode:
+
+1. **Separate additional settings**: Manager-specific settings are extracted into a dedicated IPropertyTree
+2. **Send settings separately**: Only additional settings are sent to workers (not the full merged config)
+3. **Re-merge on refresh**: Both manager and workers install hooks to re-merge additional settings when config is refreshed
+4. **Enable monitoring**: Config monitoring is enabled for both manager and workers in containerized mode
+
+## Technical Details
+
+### Containerized Mode Changes
+
+#### Manager (thmastermain.cpp)
+- Enables config monitoring: `loadConfiguration(..., monitor=true)`
+- Creates `managerAdditionalSettings` tree with manager-specific settings:
+  - `@masterBuildTag`, `@channelsPerWorker`, `@name`, `@nodeGroup`
+  - `@masterTotalMem`, `@thorPath`, `@query_so_dir`, `@dllsToSlaves`, `@thorTempDirectory`
+  - `logging/@thorworkerdetail`
+  - `workerMemory/*`, `managerMemory/*`
+- Sends only `managerAdditionalSettings` to workers (not full globals)
+- Installs `ConfigModifyFunc` hook to re-merge settings on config refresh
+
+#### Worker (thslavemain.cpp)
+- Enables config monitoring: `loadConfiguration(..., monitor=true)`
+- Receives `managerAdditionalSettings` from manager
+- Merges additional settings into its own config
+- Stores additional settings for re-use
+- Installs `ConfigModifyFunc` hook to re-merge settings on config refresh
+
+### Bare-Metal Mode (Unchanged)
+- Config monitoring disabled: `loadConfiguration(..., monitor=false)`
+- Manager sends full merged globals to workers
+- Workers use master config as before
+- Complete backward compatibility
+
+## Files Modified
+- `thorlcr/master/thmastermain.cpp` (+83 lines)
+- `thorlcr/slave/thslavemain.cpp` (+49 lines)
+
+## Benefits
+1. **No restart required**: Soft config changes applied automatically
+2. **Preserved settings**: Manager-provided settings maintained across refreshes
+3. **Backward compatible**: Bare-metal mode unchanged
+4. **Clean design**: Clear separation of base config and additional settings
+
+## Related Code
+- Config update hooks: `system/jlib/jptree.cpp` (ConfigModifyFunc, ConfigUpdateFunc)
+- Config loading: `system/jlib/jptree.cpp` (loadConfiguration)
+- Config merging: `system/jlib/jptree.cpp` (mergeConfiguration)

thorlcr/master/thmastermain.cpp

@@ -131,6 +131,12 @@ class CThorEndHandler : implements IThreaded
 static CThorEndHandler *thorEndHandler = nullptr;
 static StringBuffer cloudJobName;
 
+// Additional settings that the manager adds to the configuration before sending to workers.
+// These need to be re-merged when the configuration is refreshed (containerized mode only).
+static Owned<IPropertyTree> managerAdditionalSettings;
+static CriticalSection managerAdditionalSettingsCrit;
+static CConfigUpdateHook managerConfigHook;
+
 MODULE_INIT(INIT_PRIORITY_STANDARD)
 {
     /* NB: CThorEndHandler starts the thread now, although strictly it is not needed until later.
@@ -397,11 +403,48 @@ class CRegistryServer : public CSimpleInterface
         //Check that nothing has caused the global configuration to be refreshed - otherwise inconsistent values may be used by the slave
         assertex(globals == getComponentConfigSP());
 
+        if (isContainerized())
+        {
+            // Create additional settings tree to send to workers
+            // These settings will be re-merged when config is refreshed
+            // Only include settings that are actually SET by the manager (not those from config)
+            CriticalBlock b(managerAdditionalSettingsCrit);
+            managerAdditionalSettings.setown(createPTree("ThorManagerAdditionalSettings"));
+
+            // Properties that the manager dynamically sets and workers need
+            managerAdditionalSettings->setProp("@masterBuildTag", globals->queryProp("@masterBuildTag"));
+            managerAdditionalSettings->setPropInt("@masterTotalMem", globals->getPropInt("@masterTotalMem"));
+            managerAdditionalSettings->setProp("@thorPath", globals->queryProp("@thorPath"));
+
+            if (globals->hasProp("@query_so_dir"))
+                managerAdditionalSettings->setProp("@query_so_dir", globals->queryProp("@query_so_dir"));
+            if (globals->hasProp("@dllsToSlaves"))
+                managerAdditionalSettings->setPropBool("@dllsToSlaves", globals->getPropBool("@dllsToSlaves"));
+            if (globals->hasProp("@thorTempDirectory"))
+                managerAdditionalSettings->setProp("@thorTempDirectory", globals->queryProp("@thorTempDirectory"));
+
+            // Copy worker memory settings that manager computed
+            IPropertyTree *workerMemory = globals->queryPropTree("workerMemory");
+            if (workerMemory)
+                managerAdditionalSettings->setPropTree("workerMemory", createPTreeFromIPT(workerMemory));
+        }
+
         PROGLOG("Workers connected, initializing..");
         msg.clear();
         msg.append(THOR_VERSION_MAJOR).append(THOR_VERSION_MINOR);
         processGroup->serialize(msg);
-        globals->serialize(msg);
+        if (isContainerized())
+        {
+            // In containerized mode, workers already have the base config loaded.
+            // Only send the additional manager settings that need to be merged.
+            CriticalBlock b(managerAdditionalSettingsCrit);
+            managerAdditionalSettings->serialize(msg);
+        }
+        else
+        {
+            // In bare-metal mode, send the full merged globals as before
+            globals->serialize(msg);
+        }
         getGlobalConfigSP()->serialize(msg);
         msg.append(managerWorkerMpTag);
         msg.append(kjServiceMpTag);
@@ -645,10 +688,24 @@ int main( int argc, const char *argv[]  )
     InitModuleObjects();
     NoQuickEditSection xxx;
     {
-        bool monitorConfig = false; // Do not allow updates to the config file, otherwise the slave may not be in sync.
-        //MORE: What about updates to storage planes - they will not be passed through to the slaves
+        bool monitorConfig = isContainerized(); // Enable monitoring in containerized mode only
         globals.setown(loadConfiguration(thorDefaultConfigYaml, argv, "thor", "THOR", "thor.xml", nullptr, nullptr, monitorConfig));
     }
+
+    if (isContainerized())
+    {
+        // Install config update hook to re-merge manager additional settings when config is refreshed
+        managerConfigHook.installModifierOnce([](IPropertyTree *newComponentConfiguration, IPropertyTree *newGlobalConfiguration)
+        {
+            // Re-merge additional manager settings into refreshed config (before it becomes active)
+            CriticalBlock b(managerAdditionalSettingsCrit);
+            if (managerAdditionalSettings)
+            {
+                mergeConfiguration(*newComponentConfiguration, *managerAdditionalSettings);
+            }
+        }, true); // true = thread safe (we're in main thread during init)
+    }
+
     updateTraceFlags(loadTraceFlags(globals, thorTraceOptions, queryTraceFlags()), true);
 #ifdef _DEBUG
     unsigned holdWorker = globals->getPropInt("@holdSlave", NotFound);

thorlcr/slave/thslavemain.cpp

@@ -79,6 +79,11 @@ static const unsigned defaultForceNumStrands = 0;
 static const char **cmdArgs;
 static ILogMsgHandler *logHandler = nullptr;
 
+// Additional settings from manager that need to be re-merged on config refresh (containerized mode only)
+static Owned<IPropertyTree> workerStoredManagerSettings;
+static CriticalSection workerManagerSettingsCrit;
+static CConfigUpdateHook workerConfigHook;
+
 static void replyError(unsigned errorCode, const char *errorMsg)
 {
     SocketEndpoint myEp = queryMyNode()->endpoint();
@@ -139,7 +144,17 @@ static bool RegisterSelf(SocketEndpoint &masterEp)
         msg.read(vmajor);
         msg.read(vminor);
         Owned<IGroup> processGroup = deserializeIGroup(msg);
-        Owned<IPropertyTree> masterComponentConfig = createPTree(msg);
+        Owned<IPropertyTree> masterComponentConfig;
+        if (isContainerized())
+        {
+            // In containerized mode, receive only the additional manager settings
+            masterComponentConfig.setown(createPTree(msg));
+        }
+        else
+        {
+            // In bare-metal mode, receive the full merged component config from manager
+            masterComponentConfig.setown(createPTree(msg));
+        }
         Owned<IPropertyTree> masterGlobalConfig = createPTree(msg);
         mySlaveNum = (unsigned)processGroup->rank(queryMyNode());
         assertex(NotFound != mySlaveNum);
@@ -152,10 +167,38 @@ static bool RegisterSelf(SocketEndpoint &masterEp)
             assertex(mySlaveNum == configSlaveNum);
 
         Owned<IPropertyTree> mergedComponentConfig = createPTreeFromIPT(globals);
-        mergeConfiguration(*mergedComponentConfig, *masterComponentConfig);
-        if (masterComponentConfig->hasProp("logging/@thorworkerdetail"))
+        if (isContainerized())
+        {
+            // In containerized mode, merge the additional manager settings into our existing config
+            mergeConfiguration(*mergedComponentConfig, *masterComponentConfig);
+
+            // Store additional settings for re-merging on config refresh
+            {
+                CriticalBlock b(workerManagerSettingsCrit);
+                workerStoredManagerSettings.setown(createPTreeFromIPT(masterComponentConfig));
+            }
+
+            // Install config update hook to re-merge manager settings when config is refreshed
+            workerConfigHook.installModifierOnce([](IPropertyTree *newComponentConfiguration, IPropertyTree *newGlobalConfiguration)
+            {
+                // Re-merge additional manager settings into refreshed config (before it becomes active)
+                CriticalBlock b(workerManagerSettingsCrit);
+                if (workerStoredManagerSettings)
+                {
+                    mergeConfiguration(*newComponentConfiguration, *workerStoredManagerSettings);
+                }
+            }, true); // true = thread safe (we're in RegisterSelf, single-threaded at this point)
+        }
+        else
         {
-            unsigned workerDetailLevel = masterComponentConfig->getPropInt("logging/@thorworkerdetail");
+            // In bare-metal mode, merge the full master config as before
+            mergeConfiguration(*mergedComponentConfig, *masterComponentConfig);
+        }
+
+        // Handle logging detail level override if present
+        if (mergedComponentConfig->hasProp("logging/@thorworkerdetail"))
+        {
+            unsigned workerDetailLevel = mergedComponentConfig->getPropInt("logging/@thorworkerdetail");
             mergedComponentConfig->setPropInt("logging/@detail", workerDetailLevel);
             ILogMsgFilter *existingLogFilter = queryLogMsgManager()->queryMonitorFilter(logHandler);
             dbgassertex(existingLogFilter);
@@ -414,18 +457,23 @@ int main( int argc, const char *argv[]  )
             return 1;
         }
         cmdArgs = argv+1;
-#ifdef _CONTAINERIZED
-        globals.setown(loadConfiguration(thorDefaultConfigYaml, argv, "thor", "THOR", nullptr, nullptr, nullptr, false));
-        // pickup the default logging level from the thor default config yaml
-        if (globals->hasProp("logging/@thorworkerdetail"))
+        if (isContainerized())
         {
-            unsigned workerDetailLevel = globals->getPropInt("logging/@thorworkerdetail");
-            globals->setPropInt("logging/@detail", workerDetailLevel);
-            // NB: may be overridden by Thor config settings during RegisterSelf
+            // In containerized mode, enable config monitoring for workers
+            globals.setown(loadConfiguration(thorDefaultConfigYaml, argv, "thor", "THOR", nullptr, nullptr, nullptr, true));
+            // pickup the default logging level from the thor default config yaml
+            if (globals->hasProp("logging/@thorworkerdetail"))
+            {
+                unsigned workerDetailLevel = globals->getPropInt("logging/@thorworkerdetail");
+                globals->setPropInt("logging/@detail", workerDetailLevel);
+                // NB: may be overridden by Thor config settings during RegisterSelf
+            }
+        }
+        else
+        {
+            // In bare-metal mode, no monitoring
+            globals.setown(loadConfiguration(globals, nullptr, argv, "thor", "THOR", nullptr, nullptr, nullptr, false));
         }
-#else
-        globals.setown(loadConfiguration(globals, nullptr, argv, "thor", "THOR", nullptr, nullptr, nullptr, false));
-#endif
 
         // NB: the thor configuration is serialized from the manager and only available after RegisterSelf
         // Until that point, only properties on the command line are available.