Restart extractor on config change

Author: Toshad

ExtractorUtils.Test/unit/Unstable/RuntimeTest.cs

@@ -187,6 +187,9 @@ private ExtractorRuntimeBuilder<DummyConfig, DummyExtractor> CreateMockRuntimeBu
         public async Task TestRuntimeRestartNewConfig()
         {
             var builder = CreateMockRuntimeBuilder();
+            // Restart policy won't be the default "Always" in customer envs, but we should 
+            // still restart on config change. 
+            builder.RestartPolicy = ExtractorRestartPolicy.OnError;
 
             using var evt = new ManualResetEventSlim(false);
 
@@ -199,6 +202,7 @@ public async Task TestRuntimeRestartNewConfig()
                     extractor = ext;
                 };
             };
+            
 
             using var source = new CancellationTokenSource();
 

ExtractorUtils/Unstable/Runtime/Runtime.cs

@@ -36,6 +36,11 @@ enum ExtractorRunResult
         /// The extractor crashed.
         /// </summary>
         Error,
+        /// <summary>
+        /// The extractor was stopped with a clean shutdown.
+        /// But we need to restart it (possibly due to a revision change).
+        /// </summary>
+        RestartRequired
     }
 
     /// <summary>
@@ -138,6 +143,7 @@ public async Task Run()
                 }
                 else if (result == ExtractorRunResult.CleanShutdown)
                 {
+                    _activeLogger.LogInformation("Extractor stopped cleanly with policy {Policy}", _params.RestartPolicy);
                     // Shut down, if the extractor is configured to only restart on error.
                     if (_params.RestartPolicy != ExtractorRestartPolicy.Always)
                     {
@@ -147,6 +153,11 @@ public async Task Run()
                     // Otherwise, immediately restart.
                     backoff = 0;
                 }
+                else if (result == ExtractorRunResult.RestartRequired)
+                {
+                    _activeLogger.LogInformation("Extractor stopped cleanly with restart required, restarting with backoff");
+                    backoff = 1;
+                }
 
                 if (backoff == 0)
                 {
@@ -328,6 +339,7 @@ private async Task<ExtractorRunResult> BuildAndRunExtractor(ServiceProvider prov
         {
             using var internalTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_source.Token);
             TExtractor extractor;
+            var shouldRestart = false;
             try
             {
                 if (_params.AddMetrics)
@@ -386,6 +398,7 @@ private async Task<ExtractorRunResult> BuildAndRunExtractor(ServiceProvider prov
                 {
                     _activeLogger.LogInformation("Revision changed, reloading config");
                     internalTokenSource.Cancel();
+                    shouldRestart = true;
                     await extractorTask.ConfigureAwait(false);
                 }
 
@@ -394,6 +407,7 @@ private async Task<ExtractorRunResult> BuildAndRunExtractor(ServiceProvider prov
                 {
                     ExceptionDispatchInfo.Capture(extractorTask.Exception).Throw();
                 }
+
             }
             catch (OperationCanceledException) when (internalTokenSource.IsCancellationRequested)
             {
@@ -415,7 +429,11 @@ private async Task<ExtractorRunResult> BuildAndRunExtractor(ServiceProvider prov
 
                 return ExtractorRunResult.Error;
             }
-
+            if (shouldRestart)
+            {
+                _activeLogger.LogInformation("Extractor stopped cleanly with policy {Policy}, restart is required", _params.RestartPolicy);
+                return ExtractorRunResult.RestartRequired;
+            }
             return ExtractorRunResult.CleanShutdown;
         }