Merge branch 'add_logo_and_diagram' of https://github.com/koralium/fl…

…owtide into add_logo_and_diagram

docs/docs/deployment/index.md

@@ -0,0 +1,13 @@
+---
+sidebar_position: 4
+---
+
+import DocCardList from '@theme/DocCardList';
+
+
+
+# Deployment
+
+This section covers information that can be useful when deploying flowtide.
+
+<DocCardList />

docs/docs/deployment/pauseresume.md

@@ -0,0 +1,73 @@
+---
+sidebar_position: 1
+---
+
+# Pause And Resume
+
+It is possible to pause and resume a flowtide stream, this is useful when there might be maintainance work on a source system, or as a panic button to stop
+the execution.
+
+A pause does not stop the stream, but stops the traversal of all events inside of the stream. This means that resuming a paused stream is quick and does
+not need to download any state from persistent storage.
+
+## Pause and stop using IConfiguration
+
+One of the easier ways to add pause and resume is to utilize `IConfiguration` in .NET. Flowtide uses an `IOptionsMonitor<FlowtidePauseOptions>` to check
+for changes on the configuration and pauses and resumes the stream based on the value.
+
+The class looks as follows:
+
+```
+public class FlowtidePauseOptions
+{
+    public bool IsPaused { get; set; }
+}
+```
+
+So to utilize this when using `FlowtideDotNet.AspNetCore` or `FlowtideDotNet.DependencyInjection` you add the following:
+
+```
+var builder = WebApplication.CreateBuilder(args);
+
+...
+
+// Map flowtide pause options to enable pausing and resuming from configuration
+builder.Services.AddOptions<FlowtidePauseOptions>()
+    .Bind(builder.Configuration.GetSection("your_section"));
+
+...
+
+// Add the stream as normal
+builder.Services.AddFlowtideStream("stream_name")
+...
+```
+
+This is best fitted with an `IConfiguration` provider that supports loading changes dynamically such as Hashicorp Vault or Azure Key Vault.
+Pausing and resuming using `IConfiguration`provider is dependent on the update frequency of the provider, so to utilize this fully this interval should be
+kept quite low. 
+
+## Pause and stop using API endpoint
+
+When using `FlowtideDotNet.AspNetCore` you can also map API endpoints to allow for pause and resume.
+
+Example:
+
+```
+var builder = WebApplication.CreateBuilder(args);
+
+...
+
+// Add the stream as normal
+builder.Services.AddFlowtideStream("stream_name")
+
+...
+
+var app = builder.Build();
+
+...
+
+// Map pause and resume endpoints
+app.UseFlowtidePauseResumeEndpoints("base_path");
+```
+
+Two endpoints are registered under the base path, `/pause` and `/resume`.

docs/docs/internal/storage/objectstate.md

@@ -0,0 +1,32 @@
+---
+sidebar_position: 2
+---
+
+# Object State
+
+The object state is the simplest type of state for an operator, it saves a C# object using JSON serialization to store it in persistent storage.
+
+An object state can be fetched from the `IStateManagerClient` using `GetOrCreateObjectStateAsync<T>(string name)`.
+
+Example:
+
+```csharp
+protected override async Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
+{
+    _state = await stateManagerClient.GetOrCreateObjectStateAsync<MyState>("my_state");
+}
+
+protected override async Task OnCheckpoint()
+{
+    // Commit any changes made to the state
+    await _state.Commit();
+}
+
+public void OtherMethod()
+{
+    _state.Value.Test = "hello";
+}
+```
+
+The object state saves an internal copy of the value that was last commited and checkpointed, if the value has not changed nothing will be written to persistent storage.
+So a user of the object state does not have to handle conditional calls to `Commit` to reduce the number of writes to persistent storage.

docs/docs/statepersistence.md

@@ -196,24 +196,18 @@ A data page is fetched using the following logic:
 ## Compression
 
 It is possible to compress pages in the state.
-This is done by providing two functions to state serialize options, a compress function and a decompress function.
+The option that exist today is to compress pages with Zstd. Most storage backends add zstd compression by default to save on network throughput and storage size.
 
-Example using ZLib compression:
+To set compression, it is set under add storage:
 
 ```csharp
-builder.WithStateOptions(() => new StateManagerOptions()
-{
+builder.AddStorage(b => {
     ...
-    SerializeOptions = new StateSerializeOptions()
-    {
-        CompressFunc = (stream) =>
-        {
-            return new System.IO.Compression.ZLibStream(stream, CompressionMode.Compress);
-        },
-        DecompressFunc = (stream) =>
-        {
-            return new System.IO.Compression.ZLibStream(stream, CompressionMode.Decompress);
-        }
-    }
-})
+
+    // Use zstd page compression
+    b.ZstdPageCompression();
+
+    // Use no compression even if the storage medium added compression
+    b.NoCompression();
+});
 ```

samples/MonitoringAzureMonitor/DummyReadOperator.cs

@@ -33,7 +33,7 @@ public override IStreamIngressVertex CreateSource(ReadRelation readRelation, IFu
         }
     }
 
-    public class DummyReadOperator : ReadBaseOperator<object>
+    public class DummyReadOperator : ReadBaseOperator
     {
         public DummyReadOperator(DataflowBlockOptions options) : base(options)
         {
@@ -56,14 +56,14 @@ protected override Task<IReadOnlySet<string>> GetWatermarkNames()
             return Task.FromResult<IReadOnlySet<string>>(new HashSet<string>() { "dummy" });
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override async Task SendInitial(IngressOutput<StreamEventBatch> output)

samples/MonitoringAzureMonitor/DummyWriteOperator.cs

@@ -33,7 +33,7 @@ public override IStreamEgressVertex CreateSink(WriteRelation writeRelation, IFun
         }
     }
 
-    public class DummyWriteOperator : WriteBaseOperator<object>
+    public class DummyWriteOperator : WriteBaseOperator
     {
         public DummyWriteOperator(ExecutionDataflowBlockOptions executionDataflowBlockOptions) : base(executionDataflowBlockOptions)
         {
@@ -51,14 +51,14 @@ public override Task DeleteAsync()
             return Task.CompletedTask;
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override Task OnRecieve(StreamEventBatch msg, long time)

samples/MonitoringPrometheus/DummyReadOperator.cs

@@ -36,7 +36,7 @@ public override IStreamIngressVertex CreateSource(ReadRelation readRelation, IFu
         }
     }
 
-    public class DummyReadOperator : ReadBaseOperator<object>
+    public class DummyReadOperator : ReadBaseOperator
     {
         public DummyReadOperator(DataflowBlockOptions options) : base(options)
         {
@@ -59,14 +59,14 @@ protected override Task<IReadOnlySet<string>> GetWatermarkNames()
             return Task.FromResult<IReadOnlySet<string>>(new HashSet<string>() { "dummy" });
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override async Task SendInitial(IngressOutput<StreamEventBatch> output)

samples/MonitoringPrometheus/DummyWriteOperator.cs

@@ -33,7 +33,7 @@ public override IStreamEgressVertex CreateSink(WriteRelation writeRelation, IFun
         }
     }
 
-    public class DummyWriteOperator : WriteBaseOperator<object>
+    public class DummyWriteOperator : WriteBaseOperator
     {
         public DummyWriteOperator(ExecutionDataflowBlockOptions executionDataflowBlockOptions) : base(executionDataflowBlockOptions)
         {
@@ -51,14 +51,14 @@ public override Task DeleteAsync()
             return Task.CompletedTask;
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override Task OnRecieve(StreamEventBatch msg, long time)

samples/SqlSampleWithUI/DummyReadOperator.cs

@@ -36,7 +36,7 @@ public override IStreamIngressVertex CreateSource(ReadRelation readRelation, IFu
         }
     }
 
-    public class DummyReadOperator : ReadBaseOperator<object>
+    public class DummyReadOperator : ReadBaseOperator
     {
         public DummyReadOperator(DataflowBlockOptions options) : base(options)
         {
@@ -59,12 +59,12 @@ protected override Task<IReadOnlySet<string>> GetWatermarkNames()
             return Task.FromResult<IReadOnlySet<string>>(new HashSet<string>() { "dummy" });
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
             return Task.FromResult(new object());
         }

samples/SqlSampleWithUI/DummyWriteOperator.cs

@@ -33,7 +33,7 @@ public override IStreamEgressVertex CreateSink(WriteRelation writeRelation, IFun
         }
     }
 
-    public class DummyWriteOperator : WriteBaseOperator<object>
+    public class DummyWriteOperator : WriteBaseOperator
     {
         public DummyWriteOperator(ExecutionDataflowBlockOptions executionDataflowBlockOptions) : base(executionDataflowBlockOptions)
         {
@@ -51,14 +51,14 @@ public override Task DeleteAsync()
             return Task.CompletedTask;
         }
 
-        protected override Task InitializeOrRestore(long restoreTime, object? state, IStateManagerClient stateManagerClient)
+        protected override Task InitializeOrRestore(long restoreTime, IStateManagerClient stateManagerClient)
         {
             return Task.CompletedTask;
         }
 
-        protected override Task<object> OnCheckpoint(long checkpointTime)
+        protected override Task OnCheckpoint(long checkpointTime)
         {
-            return Task.FromResult(new object());
+            return Task.CompletedTask;
         }
 
         protected override Task OnRecieve(StreamEventBatch msg, long time)

samples/SqlSampleWithUI/Program.cs

@@ -22,9 +22,14 @@
 using FlowtideDotNet.Core.Sources.Generic;
 using OpenTelemetry.Metrics;
 using FlowtideDotNet.Core.Sinks;
+using FlowtideDotNet.Base;
 
 var builder = WebApplication.CreateBuilder(args);
 
+// Map flowtide pause options to enable pausing and resuming from configuration
+builder.Services.AddOptions<FlowtidePauseOptions>()
+    .Bind(builder.Configuration.GetSection("flowtide"));
+
 var sqlText = @"
 CREATE TABLE testtable (
   val any
@@ -69,5 +74,4 @@ LEFT JOIN other o
 app.UseHealthChecks("/health");
 app.UseFlowtideUI("/");
 
-
 app.Run();

samples/SqlSampleWithUI/appsettings.json

@@ -5,5 +5,8 @@
       "Microsoft.AspNetCore": "Warning"
     }
   },
-  "AllowedHosts": "*"
+  "AllowedHosts": "*",
+  "flowtide": {
+    "isPaused": false
+  }
 }

src/FlowtideDotNet.AspNetCore/Extensions/UIApplicationBuilderExtensions.cs

@@ -13,7 +13,9 @@
 using FlowtideDotNet.AspNetCore.Internal.TimeSeries.Middleware;
 using FlowtideDotNet.AspNetCore.TimeSeries;
 using Microsoft.AspNetCore.Builder;
+using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.DependencyInjection;
+using System.Text.Json;
 
 namespace FlowtideDotNet.AspNetCore.Extensions
 {
@@ -48,5 +50,34 @@ public static IApplicationBuilder UseFlowtideUI(this IApplicationBuilder app, Fl
             return app.UseMiddleware<UiMiddleware>(new UiMiddlewareState(new DiagnosticsEndpoint(dataflowStream), new ReactEndpoint(path), path));
         }
 
+        public static IApplicationBuilder UseFlowtidePauseResumeEndpoints(this IApplicationBuilder app, string basePath = "/ui/stream")
+        {
+            if (basePath == "/")
+            {
+                basePath = string.Empty;
+            }
+            app.Map($"{basePath}/pause", appBuilder =>
+            {
+                appBuilder.Use((HttpContext context,Func<Task> next) =>
+                {
+                    context.RequestServices.GetRequiredService<FlowtideDotNet.Base.Engine.DataflowStream>().Pause();
+                    context.Response.StatusCode = 200;
+                    context.Response.ContentType = "application/json";
+                    return context.Response.BodyWriter.WriteAsync(JsonSerializer.SerializeToUtf8Bytes(new { message = "Stream paused" })).AsTask();
+                });
+            });
+
+            app.Map($"{basePath}/resume", appBuilder =>
+            {
+                appBuilder.Use((HttpContext context, Func<Task> next) =>
+                {
+                    context.RequestServices.GetRequiredService<FlowtideDotNet.Base.Engine.DataflowStream>().Resume();
+                    context.Response.StatusCode = 200;
+                    context.Response.ContentType = "application/json";
+                    return context.Response.BodyWriter.WriteAsync(JsonSerializer.SerializeToUtf8Bytes(new { message = "Stream resumed" })).AsTask();
+                });
+            });
+            return app;
+        }
     }
 }