Default Lag Reporting Frequency of Cosmos consumers to 1m (#95)

CHANGELOG.md

@@ -10,6 +10,9 @@ The `Unreleased` section name is replaced by the expected version of next releas
 
 ### Added
 ### Changed
+
+- Added defaulting of 1 min for lag reporting frequency to all Cosmos consumers [#95](https:/jet/dotnet-templates/pull/95)
+
 ### Removed
 ### Fixed
 
@@ -548,4 +551,4 @@ The `Unreleased` section name is replaced by the expected version of next releas
 [2.1.2]: https:/jet/dotnet-templates/compare/2.0.0...2.1.2
 [2.0.0]: https:/jet/dotnet-templates/compare/1.2.0...2.0.0
 [1.2.0]: https:/jet/dotnet-templates/compare/1.1.1...1.2.0
-[1.1.1]: https:/jet/dotnet-templates/compare/1061b32ff1d86633e4adb0ce591992aea9c48c1e...1.1.1
+[1.1.1]: https:/jet/dotnet-templates/compare/1061b32ff1d86633e4adb0ce591992aea9c48c1e...1.1.1

equinox-shipping/Watchdog/Program.fs

@@ -60,9 +60,10 @@ module Args =
  Log.Information("Monitoring Group {processorName} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, leases.Database.Id, leases.Id, Option.toNullable srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let storeClient, monitored = srcC.ConnectStoreAndMonitored()
  (storeClient, monitored, leases, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)
+
  and [<NoEquality; NoComparison>] CosmosSourceParameters =
  | [<AltCommandLine "-Z"; Unique>] FromTail
  | [<AltCommandLine "-md"; Unique>] MaxDocuments of int
@@ -80,7 +81,7 @@ module Args =
  member a.Usage = a |> function
  | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited"
- | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: 1"
  | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`."
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
@@ -103,7 +104,7 @@ module Args =
 
  member val FromTail = a.Contains CosmosSourceParameters.FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency : TimeSpan = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val LeaseContainerId = a.TryGetResult CosmosSourceParameters.LeaseContainer
  member private _.ConnectLeases containerId = connector.CreateUninitialized(database, containerId)
  member x.ConnectLeases() = match x.LeaseContainerId with
@@ -152,7 +153,7 @@ let build (args : Args.Arguments) =
  let pipeline =
  let log = Log.ForContext<CosmosStoreSource>()
  use observer = CosmosStoreSource.CreateObserver(log, watchdogSink.StartIngester, Seq.collect Handler.transformOrFilter)
- CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
  watchdogSink, pipeline
 
 let run args = async {

propulsion-archiver/Program.fs

@@ -77,7 +77,7 @@ module Args =
  Log.Information("Monitoring Group {processorName} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, leases.Database.Id, leases.Id, Option.toNullable srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let monitored = srcC.MonitoredContainer()
  (monitored, leases, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)
  and [<NoEquality; NoComparison>] CosmosSourceParameters =
@@ -99,7 +99,7 @@ module Args =
  member a.Usage = a |> function
  | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited"
- | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: 1"
  | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`."
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
@@ -124,7 +124,7 @@ module Args =
 
  member val FromTail = a.Contains CosmosSourceParameters.FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency : TimeSpan = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val LeaseContainer = a.TryGetResult CosmosSourceParameters.LeaseContainer
  member private _.ConnectLeases containerId = connector.CreateUninitialized(database, containerId)
  member x.ConnectLeases() = match x.LeaseContainer with
@@ -191,7 +191,7 @@ let build (args : Args.Arguments, log, storeLog : ILogger) =
  let pipeline =
  use observer = CosmosStoreSource.CreateObserver(log, archiverSink.StartIngester, Seq.collect Handler.selectArchivable)
  let monitored, leases, processorName, startFromTail, maxDocuments, lagFrequency = args.MonitoringParams()
- CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
  archiverSink, pipeline
 
 // A typical app will likely have health checks etc, implying the wireup would be via `UseMetrics()` and thus not use this ugly code directly

propulsion-projector/Program.fs

@@ -61,7 +61,7 @@ module Args =
  | CfpVerbose -> "request Verbose Logging from ChangeFeedProcessor. Default: off"
  | FromTail _ -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum document count to supply for the Change Feed query. Default: use response size limit"
- | LagFreqM _ -> "specify frequency (minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "specify frequency (minutes) to dump lag stats. Default: 1"
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
  | Connection _ -> "specify a connection string for a Cosmos account. (optional if environment variable EQUINOX_COSMOS_CONNECTION specified)"
@@ -85,7 +85,7 @@ module Args =
  member val CfpVerbose = a.Contains CfpVerbose
  member val FromTail = a.Contains FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val LeaseContainerId = a.TryGetResult LeaseContainer
  member private x.ConnectLeases containerId = connector.CreateUninitialized(database, containerId)
  member x.ConnectLeases() = match x.LeaseContainerId with
@@ -313,7 +313,7 @@ module Args =
  Log.Information("Monitoring Group {processorName} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, leases.Database.Id, leases.Id, Option.toNullable srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let monitored = srcC.MonitoredContainer()
  (monitored, leases, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)
 #endif
@@ -394,7 +394,7 @@ let build (args : Args.Arguments) =
  let pipeline =
  use observer = Propulsion.CosmosStore.CosmosStoreSource.CreateObserver(Log.Logger, sink.StartIngester, Handler.mapToStreamItems)
  let monitored, leases, processorName, startFromTail, maxDocuments, lagFrequency = args.MonitoringParams()
- Propulsion.CosmosStore.CosmosStoreSource.Run(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ Propulsion.CosmosStore.CosmosStoreSource.Run(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
 #endif // cosmos
 #if esdb
  let srcE, context, spec = args.BuildEventStoreParams()

propulsion-pruner/Program.fs

@@ -70,7 +70,7 @@ module Args =
  Log.Information("Monitoring Group {leaseId} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, leases.Database.Id, leases.Id, Option.toNullable srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let monitored = srcC.MonitoredContainer()
  (monitored, leases, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)
  and [<NoEquality; NoComparison>] CosmosSourceParameters =
@@ -92,7 +92,7 @@ module Args =
  member a.Usage = a |> function
  | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited"
- | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: 1"
  | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`."
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
@@ -117,7 +117,7 @@ module Args =
 
  member val FromTail = a.Contains CosmosSourceParameters.FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency : TimeSpan = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val LeaseContainerId = a.TryGetResult CosmosSourceParameters.LeaseContainer
  member x.ConnectLeases containerId = connector.CreateUninitialized(x.DatabaseId, containerId)
 
@@ -188,7 +188,7 @@ let build (args : Args.Arguments, log : ILogger, storeLog : ILogger) =
  let pipeline =
  use observer = CosmosStoreSource.CreateObserver(log.ForContext<CosmosStoreSource>(), deletingEventsSink.StartIngester, Seq.collect Handler.selectPrunable)
  let monitored, leases, processorName, startFromTail, maxDocuments, lagFrequency = args.MonitoringParams()
- CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ CosmosStoreSource.Run(log, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
  deletingEventsSink, pipeline
 
 // A typical app will likely have health checks etc, implying the wireup would be via `UseMetrics()` and thus not use this ugly code directly

propulsion-reactor/Program.fs

@@ -155,7 +155,7 @@ module Args =
  Log.Information("Monitoring Group {processorName} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, srcC.DatabaseId, srcC.ContainerId, Option.toNullable srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let storeClient, monitored = srcC.ConnectStoreAndMonitored()
  let context = CosmosStoreContext.create storeClient
 #if changeFeedOnly
@@ -184,7 +184,7 @@ module Args =
  | Broker _ -> "specify Kafka Broker, in host:port format. (optional if environment variable PROPULSION_KAFKA_BROKER specified)"
  | Topic _ -> "specify Kafka Topic Id. (optional if environment variable PROPULSION_KAFKA_TOPIC specified)"
  | MaxInflightMb _ -> "maximum MiB of data to read ahead. Default: 10."
- | LagFreqM _ -> "specify frequency (minutes) to dump lag stats. Default: off."
+ | LagFreqM _ -> "specify frequency (minutes) to dump lag stats. Default: 1."
 #if (kafka && blank)
  | Kafka _ -> "Kafka Source parameters."
 #else
@@ -231,7 +231,7 @@ module Args =
  member a.Usage = a |> function
  | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited"
- | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: 1"
  | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`."
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
@@ -259,7 +259,7 @@ module Args =
 
  member val FromTail = a.Contains CosmosSourceParameters.FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency : TimeSpan = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val private LeaseContainerId = a.TryGetResult CosmosSourceParameters.LeaseContainer
  member private x.ConnectLeases containerId = connector.CreateUninitialized(x.DatabaseId, containerId)
  member x.ConnectLeases() = match x.LeaseContainerId with
@@ -630,7 +630,7 @@ let build (args : Args.Arguments) =
 #endif
  let pipeline =
  use observer = Propulsion.CosmosStore.CosmosStoreSource.CreateObserver(Log.Logger, sink.StartIngester, mapToStreamItems)
- Propulsion.CosmosStore.CosmosStoreSource.Run(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ Propulsion.CosmosStore.CosmosStoreSource.Run(Log.Logger, monitored, leases, processorName, observer, startFromTail, ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
  sink, pipeline
 #endif // !kafkaEventSpans
 

propulsion-sync/Program.fs

@@ -131,7 +131,7 @@ module Args =
  Log.Information("Monitoring Group {leaseId} in Database {db} Container {container} with maximum document count limited to {maxDocuments}",
  x.ConsumerGroupName, leases.Database.Id, leases.Id, srcC.MaxDocuments)
  if srcC.FromTail then Log.Warning("(If new projector group) Skipping projection of all existing events.")
- srcC.LagFrequency |> Option.iter<TimeSpan> (fun i -> Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", i.TotalSeconds))
+ Log.Information("ChangeFeed Lag stats interval {lagS:n0}s", let f = srcC.LagFrequency in f.TotalSeconds)
  let monitored = srcC.MonitoredContainer()
  Choice1Of2 (monitored, leases, x.ConsumerGroupName, srcC.FromTail, srcC.MaxDocuments, srcC.LagFrequency)
  | Choice2Of2 srcE ->
@@ -165,7 +165,7 @@ module Args =
  member a.Usage = a |> function
  | FromTail -> "(iff the Consumer Name is fresh) - force skip to present Position. Default: Never skip an event."
  | MaxDocuments _ -> "maximum item count to request from feed. Default: unlimited"
- | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: off"
+ | LagFreqM _ -> "frequency (in minutes) to dump lag stats. Default: 1"
  | LeaseContainer _ -> "specify Container Name for Leases container. Default: `sourceContainer` + `-aux`."
 
  | ConnectionMode _ -> "override the connection mode. Default: Direct."
@@ -191,7 +191,7 @@ module Args =
 
  member val FromTail = a.Contains CosmosSourceParameters.FromTail
  member val MaxDocuments = a.TryGetResult MaxDocuments
- member val LagFrequency = a.TryGetResult LagFreqM |> Option.map TimeSpan.FromMinutes
+ member val LagFrequency : TimeSpan = a.GetResult(LagFreqM, 1.) |> TimeSpan.FromMinutes
  member val LeaseContainerId = a.TryGetResult CosmosSourceParameters.LeaseContainer
  member private _.ConnectLeases containerId = connector.CreateUninitialized(database, containerId)
  member x.ConnectLeases() = match x.LeaseContainerId with
@@ -537,7 +537,7 @@ let build (args : Args.Arguments, log, storeLog : ILogger) =
  let runPipeline =
  Propulsion.CosmosStore.CosmosStoreSource.Run(
  Log.Logger, monitored, leases, processorName, observer, startFromTail,
- ?maxDocuments=maxDocuments, ?lagReportFreq=lagFrequency)
+ ?maxDocuments=maxDocuments, lagReportFreq=lagFrequency)
  sink, runPipeline
  | Choice2Of2 (srcE, checkpointsContext, spec) ->
  match maybeDstCosmos with