Support remote config from extraction pipelines [DEGR-2723]

Cognite.Simulator.Tests/UtilsTests/ConnectorBaseTest.cs

@@ -1,4 +1,5 @@
 using Cognite.Extractor.Common;
+using Cognite.Extractor.Logging;
 using Cognite.Extractor.Utils;
 using Cognite.Simulator.Extensions;
 using Cognite.Simulator.Utils;
@@ -23,6 +24,9 @@
         {
             var services = new ServiceCollection();
             services.AddCogniteTestClient();
+            services.AddLogger();
+            services.AddSingleton<RemoteConfigManager<BaseConfig>>(provider => null!);
+            services.AddSingleton<BaseConfig>();
             services.AddTransient<TestConnector>();
             services.AddSingleton<ExtractionPipeline>();
             var simConfig = new SimulatorConfig
@@ -90,7 +94,7 @@
                 Assert.True(long.Parse(lastHeartbeat) > long.Parse(heartbeat));
 
                 var pipelines = await cdf.ExtPipes.RetrieveAsync(
                     new List<string> { cdfConfig.ExtractionPipeline.PipelineId },
                     true,
                     source.Token).ConfigureAwait(false);
                 Assert.Contains(pipelines, p => p.ExternalId == cdfConfig.ExtractionPipeline.PipelineId);
@@ -110,20 +114,24 @@
         }
     }
 
+
     /// <summary>
     /// Implements a simple mock connector that only reports
     /// status back to CDF (Heartbeat)
     /// </summary>
-    internal class TestConnector : ConnectorBase
+    internal class TestConnector : ConnectorBase<BaseConfig>
     {
         private readonly ExtractionPipeline _pipeline;
+        private readonly RemoteConfigManager<BaseConfig> _remoteConfigManager;
         private readonly SimulatorConfig _config;
-
+
+
         public TestConnector(
             CogniteDestination cdf,
             ExtractionPipeline pipeline,
             SimulatorConfig config,
-            ILogger<ConnectorBase> logger) : 
+            ILogger<TestConnector> logger,
+            RemoteConfigManager<BaseConfig> remoteConfigManager) :
             base(
                 cdf,
                 new ConnectorConfig
@@ -135,10 +143,12 @@
                 {
                     config
                 },
-                logger)
+                logger,
+                remoteConfigManager)
         {
             _pipeline = pipeline;
             _config = config;
+            _remoteConfigManager = remoteConfigManager;
         }
 
         public override string GetConnectorVersion()

Cognite.Simulator.Utils/ConnectorBase.cs

@@ -8,6 +8,7 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Cognite.Extractor.Configuration;
 
 namespace Cognite.Simulator.Utils
 {
@@ -16,7 +17,7 @@ namespace Cognite.Simulator.Utils
     /// The connector information is saved as a CDF sequence, where the rows
     /// are key/value pairs (see <seealso cref="SimulatorIntegrationSequenceRows"/>)
     /// </summary>
-    public abstract class ConnectorBase
+    public abstract class ConnectorBase<T> where T : BaseConfig
     {
         /// <summary>
         /// CDF client wrapper
@@ -30,11 +31,14 @@ public abstract class ConnectorBase
         private ConnectorConfig Config { get; }
 
         private readonly Dictionary<string, string> _simulatorSequenceIds;
-        private readonly ILogger<ConnectorBase> _logger;
+        private readonly ILogger<ConnectorBase<T>> _logger;
         private readonly ConnectorConfig _config;
 
         private long LastLicenseCheckTimestamp { get; set; }
         private string LastLicenseCheckResult { get; set; }
+        private const int FIFTEEN_MIN = 9000;
+
+        private readonly RemoteConfigManager<T> _remoteconfigmanager;
 
         /// <summary>
         /// Initialize the connector with the given parameters
@@ -43,18 +47,21 @@ public abstract class ConnectorBase
         /// <param name="config">Connector configuration</param>
         /// <param name="simulators">List of simulator configurations</param>
         /// <param name="logger">Logger</param>
+        /// <param name="remoteConfigManager"></param>
         public ConnectorBase(
             CogniteDestination cdf,
             ConnectorConfig config,
             IList<SimulatorConfig> simulators,
-            ILogger<ConnectorBase> logger)
+            ILogger<ConnectorBase<T>> logger,
+            RemoteConfigManager<T> remoteConfigManager)
         {
             Cdf = cdf;
             Simulators = simulators;
             Config = config;
             _simulatorSequenceIds = new Dictionary<string, string>();
             _logger = logger;
             _config = config;
+            _remoteconfigmanager = remoteConfigManager;
         }
 
         /// <summary>
@@ -284,6 +291,33 @@ await UpdateIntegrationRows(false, token)
                     .ConfigureAwait(false);
             }
         }
+
+        /// <summary>
+        /// Task that runs in a loop, checking for new config in extraction pipelines
+        /// </summary>
+        public async Task CheckForNewConfig(CancellationToken token)
+        {
+            while (!token.IsCancellationRequested)
+            {
+                await Task
+                    .Delay(FIFTEEN_MIN, token) // Run every 15 minutes
+                    .ConfigureAwait(false);
+                _logger.LogDebug("Checking remote config");
+                if (_remoteconfigmanager == null) return;
+                var newConfig = await _remoteconfigmanager.FetchLatest(token).ConfigureAwait(false);
+                if (newConfig != null)
+                {
+                    throw new NewConfigDetected();
+                }
+            }
+        }
+    }
+
+    /// <summary>
+    /// Exception used to restart connector
+    /// </summary>
+    public class NewConfigDetected : Exception
+    {
     }
 
     /// <summary>

Cognite.Simulator.Utils/ExtractionPipeline.cs

@@ -8,6 +8,8 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Cognite.Extractor.Configuration;
+using System.IO;
 
 namespace Cognite.Simulator.Utils
 {
@@ -237,5 +239,47 @@
             services.AddSingleton(config.PipelineNotification);
             services.AddScoped<ExtractionPipeline>();
         }
+        /// <summary>
+        /// Uses a minimum config that needs "cognite" and type: "specified",
+        /// then uses that to either fetch remote Config Extraction Pipeline, or just adds
+        /// the local one based on if "type" is remote or local.
+        /// </summary>
+        /// <typeparam name="T">The complete config object to be parsed</typeparam>
+        /// <param name="services">Service collection</param>
+        /// <param name="path">Path to config file</param>
+        /// <param name="types">Types to use for deserialization</param>
+        /// <param name="appId">App ID for measuring network data</param>
+        /// <param name="token">Cancellation token</param>
+        /// <param name="version">Config version</param>
+        /// <param name="acceptedConfigVersions">Accepted config versions</param>
+        public static async Task<T> AddConfiguration<T>(
+            this IServiceCollection services,
+            string path,
+            Type[] types,
+            string appId,
+            CancellationToken token,
+            int version = 1,
+            int[] acceptedConfigVersions = null) where T : BaseConfig
+        {
+            var localConfig = services.AddConfig<T>(path, version);
+            var remoteConfig = new RemoteConfig
+            {
+                Type = localConfig.Type,
+                Cognite = localConfig.Cognite
+            };
+
+            return await services.AddRemoteConfig<T>(
+                logger: null,
+                path: path,
+                types: types,
+                appId: appId,
+                userAgent: null,
+                setDestination: true,
+                bufferConfigFile: false,
+                remoteConfig: remoteConfig,
+                token: token,
+                acceptedConfigVersions: acceptedConfigVersions
+            );
+        }
     }
 }