diff --git a/CHANGELOG.md b/CHANGELOG.md index a0eadf7f3..96ba7b55c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,8 @@ The `Unreleased` section name is replaced by the expected version of next releas ### Added -- `eqxProjector --source cosmos --kafka --synthesizeSequence`: Sample code for custom parsing of document changes [#84](https://github.com/jet/dotnet-templates/pull/84) +- `eqxProjector`: Added `--source cosmos --kafka --synthesizeSequence`: Sample code for custom parsing of document changes [#84](https://github.com/jet/dotnet-templates/pull/84) +- `proProjector`: Added `es` checkpoint storage mode [#81](https://github.com/jet/dotnet-templates/pull/81) ### Changed ### Removed diff --git a/propulsion-projector/Program.fs b/propulsion-projector/Program.fs index df4f37f6a..24ffde209 100644 --- a/propulsion-projector/Program.fs +++ b/propulsion-projector/Program.fs @@ -42,10 +42,10 @@ module Args = let private defaultWithEnvVar varName argName = function | None -> getEnvVarForArgumentOrThrow varName argName | Some x -> x -#if esdb +//#if esdb let private isEnvVarTrue varName = EnvVar.tryGet varName |> Option.exists (fun s -> String.Equals(s, bool.TrueString, StringComparison.OrdinalIgnoreCase)) -#endif +//#endif let private seconds (x : TimeSpan) = x.TotalSeconds open Argu #if cosmos @@ -105,7 +105,7 @@ module Args = let connector = Connector(x.Timeout, x.Retries, x.MaxRetryWaitTime, Log.Logger, mode=x.Mode) discovery, { database = x.Database; container = x.Container }, connector #endif -#if esdb +//#if esdb open Equinox.EventStore open Propulsion.EventStore type [] EsSourceParameters = @@ -130,6 +130,7 @@ module Args = | [] Password of string | [] Cosmos of ParseResults + | [] Es interface IArgParserTemplate with member a.Usage = a |> function | FromTail -> "Start the processing from the Tail" @@ -153,6 +154,7 @@ module Args = | HeartbeatTimeout _ -> "specify heartbeat timeout in seconds. Default: 1.5." | Cosmos _ -> "CosmosDB (Checkpoint) Store parameters." + | Es -> "Request storage of checkpoints in EventStore (not recommended due to feedback effects)." and EsSourceArguments(a : ParseResults) = let discovery (host, port, tcp) = match tcp, port with @@ -192,10 +194,11 @@ module Args = .Connect(appName, discovery, nodePreference) |> Async.RunSynchronously member __.CheckpointInterval = TimeSpan.FromHours 1. - member val Cosmos : CosmosArguments = + member val Checkpoints : Choice = match a.TryGetSubCommand() with - | Some (EsSourceParameters.Cosmos cosmos) -> CosmosArguments cosmos - | _ -> raise (MissingArg "Must specify `cosmos` checkpoint store when source is `es`") + | Some (EsSourceParameters.Cosmos cosmos) -> Choice1Of2 (CosmosArguments cosmos) + | Some (EsSourceParameters.Es) -> Choice2Of2 () + | _ -> raise (MissingArg "Must specify `cosmos` or `es` checkpoint store when source is `es`") and [] CosmosParameters = | [] Connection of string | [] ConnectionMode of Equinox.Cosmos.ConnectionMode @@ -230,8 +233,8 @@ module Args = seconds x.Timeout, x.Retries, seconds x.MaxRetryWaitTime) let connector = Equinox.Cosmos.Connector(x.Timeout, x.Retries, x.MaxRetryWaitTime, Log.Logger, mode=x.Mode) discovery, x.Database, x.Container, connector -#endif -//#if sss +//#endif +#if sss // TOCONSIDER: add DB connectors other than MsSql type [] SqlStreamStoreSourceParameters = | [] Tail of intervalS: float @@ -272,7 +275,7 @@ module Args = let sssConnectionString = String.Join(";", conn, creds) Log.Information("SqlStreamStore MsSql Connection {connectionString} Schema {schema} AutoCreate {autoCreate}", conn, schema, autoCreate) Equinox.SqlStreamStore.MsSql.Connector(sssConnectionString, schema, autoCreate=autoCreate).Connect() |> Async.RunSynchronously -//#endif +#endif // sss [] type Parameters = @@ -288,12 +291,12 @@ module Args = #if cosmos | [] Cosmos of ParseResults #endif -#if esdb +//#if esdb | [] Es of ParseResults -#endif -//#if sss - | [] SqlMs of ParseResults //#endif +#if sss + | [] SqlMs of ParseResults +#endif interface IArgParserTemplate with member a.Usage = match a with @@ -308,12 +311,12 @@ module Args = #if cosmos | Cosmos _ -> "specify CosmosDb input parameters" #endif -#if esdb +//#if esdb | Es _ -> "specify EventStore input parameters." -#endif -//#if sss - | SqlMs _ -> "specify SqlStreamStore input parameters." //#endif +#if sss + | SqlMs _ -> "specify SqlStreamStore input parameters." +#endif and Arguments(a : ParseResults) = member __.Verbose = a.Contains Verbose member __.ConsumerGroupName = a.GetResult ConsumerGroupName @@ -336,21 +339,29 @@ module Args = c.LagFrequency |> Option.iter (fun s -> Log.Information("Dumping lag stats at {lagS:n0}s intervals", s.TotalSeconds)) { database = c.Database; container = c.AuxContainerName }, __.ConsumerGroupName, c.FromTail, c.MaxDocuments, c.LagFrequency #endif -#if esdb +//#if esdb member val Es = EsSourceArguments(a.GetResult Es) member __.BuildEventStoreParams() = - let srcE = __.Es - let startPos, cosmos = srcE.StartPos, srcE.Cosmos + let es = __.Es Log.Information("Processing Consumer Group {groupName} from {startPos} (force: {forceRestart}) in Database {db} Container {container}", - __.ConsumerGroupName, startPos, srcE.ForceRestart, cosmos.Database, cosmos.Container) + __.ConsumerGroupName, es.StartPos, es.ForceRestart) Log.Information("Ingesting in batches of [{minBatchSize}..{batchSize}], reading up to {maxReadAhead} uncommitted batches ahead", - srcE.MinBatchSize, srcE.StartingBatchSize, __.MaxReadAhead) - srcE, cosmos, - { groupName = __.ConsumerGroupName; start = startPos; checkpointInterval = srcE.CheckpointInterval; tailInterval = srcE.TailInterval - forceRestart = srcE.ForceRestart - batchSize = srcE.StartingBatchSize; minBatchSize = srcE.MinBatchSize; gorge = srcE.Gorge; streamReaders = 0 } -#endif -//#if sss + es.MinBatchSize, es.StartingBatchSize, __.MaxReadAhead) + es, + { groupName = __.ConsumerGroupName; start = es.StartPos; checkpointInterval = es.CheckpointInterval; tailInterval = es.TailInterval + forceRestart = es.ForceRestart + batchSize = es.StartingBatchSize; minBatchSize = es.MinBatchSize; gorge = es.Gorge; streamReaders = 0 } + member __.BuildCheckpointStoreParams() = + let es = __.Es + match es.Checkpoints with + | Choice1Of2 cosmos -> + Log.Information("Checkpointing in Database {db} Container {container}", cosmos.Database, cosmos.Container) + Choice1Of2 cosmos + | Choice2Of2 () -> + Log.Information("Checkpointing in source EventStore") + Choice2Of2 () +//#endif +#if sss member val SqlStreamStore = SqlStreamStoreSourceArguments(a.GetResult SqlMs) member __.BuildSqlStreamStoreParams() = let src = __.SqlStreamStore @@ -359,7 +370,7 @@ module Args = maxBatchSize = src.MaxBatchSize tailSleepInterval = src.TailInterval } src, spec -//#endif +#endif //#if kafka member val Target = TargetInfo a and TargetInfo(a : ParseResults) = @@ -380,10 +391,11 @@ module Checkpoints = open Propulsion.EventStore + let codec = FsCodec.NewtonsoftJson.Codec.Create() + // In this implementation, we keep the checkpoints in Cosmos when consuming from EventStore module Cosmos = - let codec = FsCodec.NewtonsoftJson.Codec.Create() let access = Equinox.Cosmos.AccessStrategy.Custom (Checkpoint.Fold.isOrigin, Checkpoint.Fold.transmute) let create groupName (context, cache) = let caching = Equinox.Cosmos.CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) @@ -391,6 +403,23 @@ module Checkpoints = let resolve streamName = resolver.Resolve(streamName, Equinox.AllowStale) Checkpoint.CheckpointSeries(groupName, resolve) + // Alternately, one can keep checkpoints in EventStore, but that brings two factors into play: + // - the feedback effect of each write of a checkpoint triggering a read and so on will increase database traffic significantly + // - you'll want to limit the maximum retained events for checkpoint streams to 1 (or some other low number) + module EventStore = + + let access = Equinox.EventStore.AccessStrategy.LatestKnownEvent + let create groupName (context, cache) = + let caching = Equinox.EventStore.CachingStrategy.SlidingWindow (cache, TimeSpan.FromMinutes 20.) + let resolver = Equinox.EventStore.Resolver(context, codec, Checkpoint.Fold.fold, Checkpoint.Fold.initial, caching, access) + let resolve streamName = resolver.Resolve(streamName, Equinox.AllowStale) + Checkpoint.CheckpointSeries(groupName, resolve) + +module EventStoreContext = + + let create client = + Equinox.EventStore.Context(Equinox.EventStore.Connection(client), Equinox.EventStore.BatchingPolicy(maxBatchSize=500)) + module CosmosContext = let create appName (connector : Equinox.Cosmos.Connector) discovery (database, container) = @@ -425,16 +454,21 @@ let build (args : Args.Arguments) = aux, leaseId, startFromTail, createObserver, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency) #endif // cosmos -#if esdb - let (srcE, cosmos, spec) = args.BuildEventStoreParams() - +//#if esdb + let (srcE, spec) = args.BuildEventStoreParams() let connectEs () = srcE.Connect(Log.Logger, Log.Logger, AppName, Equinox.EventStore.NodePreference.Master) - let (discovery, database, container, connector) = cosmos.BuildConnectionDetails() - let context = CosmosContext.create AppName connector discovery (database, container) let cache = Equinox.Cache(AppName, sizeMb=10) - - let checkpoints = Checkpoints.Cosmos.create spec.groupName (context, cache) + let checkpoints = + match args.BuildCheckpointStoreParams() with + | Choice1Of2 cosmos -> + let (discovery, database, container, connector) = cosmos.BuildConnectionDetails() + let context = CosmosContext.create AppName connector discovery (database, container) + Checkpoints.Cosmos.create spec.groupName (context, cache) + | Choice2Of2 () -> + let esClient = connectEs () + let context = EventStoreContext.create esClient + Checkpoints.EventStore.create spec.groupName (context, cache) #if kafka // esdb && kafka let (broker, topic) = args.Target.BuildTargetParams() @@ -450,8 +484,8 @@ let build (args : Args.Arguments) = Propulsion.EventStore.EventStoreSource.Run( Log.Logger, sink, checkpoints, connectEs, spec, Handler.tryMapEvent filterByStreamName, args.MaxReadAhead, args.StatsInterval) -#endif // esdb -//#if sss +//#endif // esdb +#if sss let (srcSql, spec) = args.BuildSqlStreamStoreParams() let monitored = srcSql.Connect() @@ -469,7 +503,7 @@ let build (args : Args.Arguments) = let sink = Propulsion.Streams.StreamsProjector.Start(Log.Logger, maxReadAhead, maxConcurrentStreams, Handler.handle, stats, args.StatsInterval) #endif // sss && !kafka let pipeline = Propulsion.SqlStreamStore.SqlStreamStoreSource.Run(Log.Logger, monitored, checkpointer, spec, sink, args.StatsInterval) -//#endif // sss +#endif // sss sink, pipeline let run args = async {