Base extractor (#522)

First take on a new base extractor class. Some more work goes into this in the next step, adding startup reporting and integrating things a bit more tightly, but this is a nice first draft that contains some of the core logic.

The base extractor has a few roles, currently:

 - It contains the config object, in the future it will also handle changes to config revision and restart.
 - It acts as an error reporter, for raw errors.
 - It manages "monitored tasks", which are essentially just C# Tasks. The idea is that if something fails in a background process, this needs to be properly reported to the extractor so that it can take appropriate action. Background tasks have a name, can be cancelled independently, and can be expected to end or not, so if a task ends without an error, the extractor can take appropriate action.
 - It also contains shutdown logic, which is highly customizable, since extractor implementations may want to shut components down in a specific order. In general, different parts of the process are supposed to be independently cancelled, so that graceful shutdown is possible.

As with the last iteration of the base extractor, it is intended to be used as a base class for user extractors.

* Comments from review

---------

Co-authored-by: Ozan Göktan <72358629+ozangoktan@users.noreply.github.com>

Author: einarmo

ExtractorUtils.Test/unit/Unstable/BaseExtractorTest.cs

@@ -0,0 +1,173 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Cognite.Extractor.Common;
+using Cognite.Extractor.Testing;
+using Cognite.Extractor.Utils;
+using Cognite.ExtractorUtils.Unstable;
+using Cognite.ExtractorUtils.Unstable.Tasks;
+using ExtractorUtils.Test.unit.Unstable;
+using Microsoft.Extensions.DependencyInjection;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace ExtractorUtils.Test.Unit.Unstable
+{
+    class DummyConfig { }
+
+    class DummyExtractor : Cognite.ExtractorUtils.Unstable.BaseExtractor<DummyConfig>
+    {
+        public Action<ExtractorTaskScheduler> InitAction { get; set; }
+
+        public DummyExtractor(
+            DummyConfig config,
+            IServiceProvider provider,
+            ExtractorTaskScheduler taskScheduler,
+            IIntegrationSink sink,
+            CogniteDestination destination = null) : base(config, provider,
+            taskScheduler, sink, destination)
+        {
+        }
+
+        public void AddMonitoredTaskPub(Func<CancellationToken, Task> task, ExtractorTaskResult staticResult, string name)
+        {
+            AddMonitoredTask(task, staticResult, name);
+        }
+
+        public async Task CancelMonitoredTaskAndWaitPub(string name)
+        {
+            await CancelMonitoredTaskAndWait(name);
+        }
+
+        protected override Task InitTasks()
+        {
+            InitAction?.Invoke(TaskScheduler);
+            return Task.CompletedTask;
+        }
+    }
+
+    public class BaseExtractorTest
+    {
+        private readonly ITestOutputHelper _output;
+        public BaseExtractorTest(ITestOutputHelper output)
+        {
+            _output = output;
+        }
+
+        private (DummyExtractor, DummySink) CreateExtractor()
+        {
+            var sink = new DummySink();
+            var services = new ServiceCollection();
+            services.AddSingleton(new DummyConfig());
+            services.AddSingleton<IIntegrationSink>(sink);
+            services.AddTestLogging(_output);
+            services.AddTransient<ExtractorTaskScheduler>();
+            services.AddTransient<DummyExtractor>();
+            var provider = services.BuildServiceProvider();
+            return (provider.GetRequiredService<DummyExtractor>(), sink);
+        }
+
+        [Fact]
+        public async Task TestBaseExtractor()
+        {
+            var (ext, sink) = CreateExtractor();
+            var taskCompletedEvent = new ManualResetEvent(false);
+            // Run the extractor and verify that scheduled tasks are run.
+            ext.InitAction = (sched) =>
+            {
+                sched.AddScheduledTask(new RunQuickTask("task1", async (task, token) =>
+                {
+                    await Task.Delay(100, token);
+                    taskCompletedEvent.Set();
+                    return new TaskUpdatePayload();
+                }), true);
+            };
+            var runTask = ext.Start(CancellationToken.None);
+            Assert.True(await CommonUtils.WaitAsync(taskCompletedEvent, TimeSpan.FromSeconds(2), CancellationToken.None));
+            await ext.DisposeAsync();
+
+            Assert.Single(sink.TaskStart);
+            Assert.Single(sink.TaskEnd);
+            Assert.Empty(sink.Errors);
+        }
+
+        [Fact]
+        public async Task TestBaseExtractorInnerError()
+        {
+            var (ext, sink) = CreateExtractor();
+            var taskCompletedEvent = new ManualResetEvent(false);
+            // Run the extractor and verify that scheduled tasks are run.
+            ext.InitAction = (sched) =>
+            {
+                sched.AddScheduledTask(new RunQuickTask("task1", async (task, token) =>
+                {
+                    await Task.Delay(100, token);
+                    taskCompletedEvent.Set();
+                    throw new Exception("Inner error");
+                }), true);
+            };
+            var runTask = ext.Start(CancellationToken.None);
+            Assert.True(await CommonUtils.WaitAsync(taskCompletedEvent, TimeSpan.FromSeconds(2), CancellationToken.None));
+            await ext.DisposeAsync();
+
+            Assert.Single(sink.TaskStart);
+            Assert.Single(sink.TaskEnd);
+            Assert.Equal(2, sink.Errors.Count);
+            Assert.Equal("Inner error", sink.Errors[0].Description);
+        }
+
+        [Fact]
+        public async Task TestBaseExtractorMonitoredError()
+        {
+            var (ext, sink) = CreateExtractor();
+            var runTask = ext.Start(CancellationToken.None);
+            ext.AddMonitoredTaskPub(async t =>
+            {
+                await Task.Delay(100, t);
+                throw new Exception("Monitored error");
+            }, ExtractorTaskResult.Unexpected, "task1");
+            var delayTask = Task.Delay(2000);
+            Assert.NotEqual(delayTask, await Task.WhenAny(runTask, delayTask));
+            Assert.Equal(2, sink.Errors.Count);
+            Assert.Equal("Internal task task1 failed, restarting extractor: Monitored error", sink.Errors[0].Description);
+        }
+
+        [Fact]
+        public async Task TestBaseExtractorUnexpectedExit()
+        {
+            var (ext, sink) = CreateExtractor();
+            var runTask = ext.Start(CancellationToken.None);
+            ext.AddMonitoredTaskPub(async t =>
+            {
+                await Task.Delay(100, t);
+            }, ExtractorTaskResult.Unexpected, "task1");
+            var delayTask = Task.Delay(2000);
+            Assert.NotEqual(delayTask, await Task.WhenAny(runTask, delayTask));
+            Assert.Equal(2, sink.Errors.Count);
+            Assert.Equal("Internal task task1 completed, but was not expected to stop, restarting extractor.", sink.Errors[0].Description);
+        }
+
+        [Fact]
+        public async Task TestCancelMonitoredTask()
+        {
+            var (ext, sink) = CreateExtractor();
+            var runTask = ext.Start(CancellationToken.None);
+            ext.AddMonitoredTaskPub(async t =>
+            {
+                while (!t.IsCancellationRequested)
+                {
+                    await Task.Delay(100, t);
+                }
+            }, ExtractorTaskResult.Unexpected, "task1");
+            var delayTask = Task.Delay(2000);
+            Assert.NotEqual(delayTask, await Task.WhenAny(ext.CancelMonitoredTaskAndWaitPub("task1"), delayTask));
+            Assert.NotEqual(delayTask, await Task.WhenAny(ext.Shutdown(), delayTask));
+            Assert.NotEqual(delayTask, await Task.WhenAny(runTask, delayTask));
+
+            // Dispose should work, even if we're already shut-down.
+            await ext.DisposeAsync();
+        }
+    }
+}

ExtractorUtils.Test/unit/Unstable/CheckInWorkerTests.cs

@@ -88,7 +88,7 @@ public async Task TestCheckInWorker()
             using var p = provider;
             using var source = new CancellationTokenSource();
             // Check that this doesn't crash, and properly cancels out at the end.
-            var runTask = checkIn.Run(source.Token, Timeout.InfiniteTimeSpan);
+            var runTask = checkIn.RunPeriodicCheckin(source.Token, Timeout.InfiniteTimeSpan);
             // First, we should very quickly report a checkin on the start of the run task...
             await TestUtils.WaitForCondition(() => _checkInCount == 1, 5);
 
@@ -136,7 +136,7 @@ public async Task TestCheckInWorkerBatch()
             using var source = new CancellationTokenSource();
 
             // Check that this doesn't crash, and properly cancels out at the end.
-            var runTask = checkIn.Run(source.Token, Timeout.InfiniteTimeSpan);
+            var runTask = checkIn.RunPeriodicCheckin(source.Token, Timeout.InfiniteTimeSpan);
             // First, we should very quickly report a checkin on the start of the run task...
             await TestUtils.WaitForCondition(() => _checkInCount == 1, 5);
 

ExtractorUtils.Test/unit/Unstable/TaskSchedulerTest.cs

@@ -5,6 +5,7 @@
 using Cognite.Extractor.Common;
 using Cognite.ExtractorUtils.Unstable.Tasks;
 using CogniteSdk.Alpha;
+using Microsoft.Extensions.Logging.Abstractions;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -16,6 +17,11 @@ class DummySink : BaseErrorReporter, IIntegrationSink
         public List<(string, DateTime)> TaskStart { get; } = new();
         public List<(string, DateTime)> TaskEnd { get; } = new();
 
+        public Task Flush(CancellationToken token)
+        {
+            return Task.CompletedTask;
+        }
+
         public override ExtractorError NewError(ErrorLevel level, string description, string details = null, DateTime? now = null)
         {
             return new ExtractorError(level, description, this, details, null, now);
@@ -35,6 +41,11 @@ public void ReportTaskStart(string taskName, TaskUpdatePayload update = null, Da
         {
             TaskStart.Add((taskName, timestamp ?? DateTime.UtcNow));
         }
+
+        public async Task RunPeriodicCheckin(CancellationToken token, TimeSpan? interval = null)
+        {
+            while (!token.IsCancellationRequested) await Task.Delay(100000, token);
+        }
     }
 
     class RunQuickTask : BaseSchedulableTask
@@ -78,7 +89,7 @@ public TaskSchedulerTest(ITestOutputHelper output)
         public async Task TestScheduler()
         {
             var sink = new DummySink();
-            using var sched = new ExtractorTaskScheduler(sink);
+            using var sched = new ExtractorTaskScheduler(sink, new NullLogger<ExtractorTaskScheduler>());
             using var source = new CancellationTokenSource();
 
             var running = sched.Run(source.Token);
@@ -157,7 +168,7 @@ public async Task TestScheduler()
         public async Task TestSchedulerErrors()
         {
             var sink = new DummySink();
-            using var sched = new ExtractorTaskScheduler(sink);
+            using var sched = new ExtractorTaskScheduler(sink, new NullLogger<ExtractorTaskScheduler>());
             using var source = new CancellationTokenSource();
 
             var running = sched.Run(source.Token);
@@ -197,7 +208,7 @@ public async Task TestSchedulerErrors()
         public async Task TestSchedulerFatalError()
         {
             var sink = new DummySink();
-            using var sched = new ExtractorTaskScheduler(sink);
+            using var sched = new ExtractorTaskScheduler(sink, new NullLogger<ExtractorTaskScheduler>());
             using var source = new CancellationTokenSource();
 
             var running = sched.Run(source.Token);