added ReportingManager

Author: AntyaDev

examples/CSharpDev/CustomReporting/CustomReportingExample.cs

@@ -1,6 +1,5 @@
 using System;
 using CSharpDev.ClientFactory;
-using CSharpDev.CustomReporting;
 using CSharpDev.DataFeed;
 using CSharpDev.Features;
 using CSharpDev.HelloWorld;

examples/CSharpDev/Program.cs

@@ -21,7 +21,6 @@
         <None Include="HelloWorld\config.json">
           <CopyToOutputDirectory>Always</CopyToOutputDirectory>
         </None>
-        <Compile Include="CustomReporting\CustomReporting.fs" />
         <Compile Include="HttpTests\SimpleHttpTest.fs" />
         <None Include="HttpTests\infra-config.json">
           <CopyToOutputDirectory>Always</CopyToOutputDirectory>

examples/FSharpDev/CustomReporting/CustomReporting.fs

@@ -15,10 +15,10 @@ open NBomber.Contracts.Stats
 type Response = {
     StatusCode: Nullable<int>
     IsError: bool
-    Message: string
+    mutable Message: string
     SizeBytes: int
     LatencyMs: float
-    Payload: obj
+    mutable Payload: obj
 }
 
 type ScenarioInfo = {

examples/FSharpDev/FSharpDev.fsproj

@@ -262,13 +262,6 @@ type NBomberRunner =
     static member WithReportingInterval(context: NBomberContext, interval: TimeSpan) =
         context |> FSharp.NBomberRunner.withReportingInterval interval
 
-    /// Sets reporting sinks.
-    /// Reporting sink is used to save real-time metrics to correspond database.
-    [<Extension>]
-    static member WithReportingSinks(context: NBomberContext, [<ParamArray>]reportingSinks: IReportingSink[]) =
-        let sinks = reportingSinks |> Seq.toList
-        context |> FSharp.NBomberRunner.withReportingSinks sinks
-
     /// Sets worker plugins.
     /// Worker plugin is a plugin that starts at the test start and works as a background worker.
     [<Extension>]

src/NBomber.Contracts/Contracts.fs

@@ -268,12 +268,6 @@ module NBomberRunner =
         let report = { context.Reporting with ReportingInterval = interval }
         { context with Reporting = report }
 
-    /// Sets reporting sinks.
-    /// Reporting sink is used to save real-time metrics to correspond database
-    let withReportingSinks (reportingSinks: IReportingSink list) (context: NBomberContext) =
-        let report = { context.Reporting with Sinks = reportingSinks }
-        { context with Reporting = report }
-
     /// Sets worker plugins.
     /// Worker plugin is a plugin that starts at the test start and works as a background worker.
     let withWorkerPlugins (plugins: IWorkerPlugin list) (context: NBomberContext) =

src/NBomber/Api/CSharp.fs

@@ -52,6 +52,7 @@ type NBomberContext = {
 
 namespace NBomber.Contracts.Internal
 
+open System
 open CommandLine
 open NBomber.Contracts
 
@@ -67,3 +68,9 @@ type StepResponse = {
     EndTimeMs: float
     LatencyMs: float
 }
+
+type ScenarioRawStats = {
+    ScenarioName: string
+    Data: StepResponse list
+    Timestamp: TimeSpan
+}

src/NBomber/Api/FSharp.fs

@@ -7,6 +7,7 @@ open FSharp.Control.Reactive
 
 open NBomber
 open NBomber.Contracts
+open NBomber.Contracts.Internal
 open NBomber.Contracts.Stats
 open NBomber.Domain
 open NBomber.Domain.DomainTypes
@@ -80,7 +81,7 @@ let emptyExec (dep: ActorDep) (actorPool: ScenarioActor list) (scheduledActorCou
 
 type ScenarioScheduler(dep: ActorDep, scenarioClusterCount: int) =
 
-    let log = dep.Logger.ForContext<ScenarioScheduler>()
+    let _log = dep.Logger.ForContext<ScenarioScheduler>()
     let mutable _warmUp = false
     let mutable _scenario = dep.Scenario
     let mutable _currentSimulation = dep.Scenario.LoadTimeLine.Head.LoadSimulation
@@ -110,8 +111,15 @@ type ScenarioScheduler(dep: ActorDep, scenarioClusterCount: int) =
             getOneTimeActorCount()
         )
 
-    let getRealtimeStats (executionTime) =
-        let executedDuration = _scenario |> Scenario.getDuration |> correctExecutedDuration executionTime
+    let addRawStats (rawStats: ScenarioRawStats) =
+        dep.ScenarioStatsActor.Publish(AddResponses rawStats.Data)
+
+    let getRawStats (timestamp) =
+        let executedDuration = _scenario |> Scenario.getDuration |> correctExecutedDuration timestamp
+        dep.ScenarioStatsActor.GetRawStats executedDuration
+
+    let getRealtimeStats (timestamp) =
+        let executedDuration = _scenario |> Scenario.getDuration |> correctExecutedDuration timestamp
         let simulationStats = getCurrentSimulationStats()
         dep.ScenarioStatsActor.GetRealtimeStats(simulationStats, executedDuration)
 
@@ -198,13 +206,16 @@ type ScenarioScheduler(dep: ActorDep, scenarioClusterCount: int) =
 
     member _.EventStream = _eventStream :> IObservable<_>
     member _.Scenario = dep.Scenario
-    member _.PublishStatsToCoordinator() = dep.ScenarioStatsActor.Publish(ActorMessage.PublishStatsToCoordinator)
-    member _.GetRealtimeStats(executionTime) = getRealtimeStats executionTime
+
+    member _.AddRawStats(rawStats) = addRawStats rawStats
+    member _.GetRawStats(timestamp) = getRawStats timestamp
+    member _.GetRealtimeStats(timestamp) = getRealtimeStats timestamp
     member _.GetFinalStats() = getFinalStats()
+    member _.GetStatusMessages() = ()
 
     interface IDisposable with
         member _.Dispose() =
             stop()
             _eventStream.Dispose()
-            log.Verbose $"{nameof ScenarioScheduler} disposed"
+            _log.Verbose $"{nameof ScenarioScheduler} disposed"
 

src/NBomber/Contracts.fs

@@ -13,42 +13,48 @@ open NBomber.Domain.DomainTypes
 open NBomber.Domain.Stats.Statistics
 
 type ActorMessage =
-    | AddResponse      of StepResponse
-    | AddResponses     of StepResponse[]
-    | PublishStatsToCoordinator
-    | GetRealtimeStats of reply:TaskCompletionSource<ScenarioStats> * LoadSimulationStats * duration:TimeSpan
-    | GetFinalStats    of reply:TaskCompletionSource<ScenarioStats> * LoadSimulationStats * duration:TimeSpan
+    | AddResponse       of StepResponse
+    | AddResponses      of StepResponse list
+    | GetRawStats       of reply:TaskCompletionSource<ScenarioRawStats> * timestamp:TimeSpan
+    | GetRealtimeStats  of reply:TaskCompletionSource<ScenarioStats> * LoadSimulationStats * duration:TimeSpan
+    | GetFinalStats     of reply:TaskCompletionSource<ScenarioStats> * LoadSimulationStats * duration:TimeSpan
 
 type IScenarioStatsActor =
     abstract Publish: ActorMessage -> unit
+    abstract GetRawStats: timestamp:TimeSpan -> Task<ScenarioRawStats>
     abstract GetRealtimeStats: LoadSimulationStats * duration:TimeSpan -> Task<ScenarioStats>
     abstract GetFinalStats: LoadSimulationStats * duration:TimeSpan -> Task<ScenarioStats>
 
-type ScenarioStatsActor(logger: ILogger, scenario: Scenario, reportingInterval: TimeSpan) =
+type ScenarioStatsActor(logger: ILogger, scenario: Scenario, reportingInterval: TimeSpan, keepRawStats: bool) =
 
     let _allStepsData = Array.init scenario.Steps.Length (fun _ -> StepStatsRawData.createEmpty())
     let mutable _intervalStepsData = Array.init scenario.Steps.Length (fun _ -> StepStatsRawData.createEmpty())
+    let mutable _intervalRawStats = List.empty
 
-    let addResponse (allData: StepStatsRawData[]) (intervalData: StepStatsRawData[]) (resp: StepResponse) =
-        let allStData = allData.[resp.StepIndex]
-        let intervalStData = intervalData.[resp.StepIndex]
-        allData.[resp.StepIndex] <- StepStatsRawData.addResponse allStData resp
-        intervalData.[resp.StepIndex] <- StepStatsRawData.addResponse intervalStData resp
+    let addResponse (resp: StepResponse) =
+        let allStData = _allStepsData.[resp.StepIndex]
+        let intervalStData = _intervalStepsData.[resp.StepIndex]
+        _allStepsData.[resp.StepIndex] <- StepStatsRawData.addResponse allStData resp
+        _intervalStepsData.[resp.StepIndex] <- StepStatsRawData.addResponse intervalStData resp
+
+        if keepRawStats then
+            resp.ClientResponse.Payload <- null // to prevent sending in cluster mode
+            resp.ClientResponse.Message <- null
+            _intervalRawStats <- resp :: _intervalRawStats
 
     let createScenarioStats (stepsData, simulationStats, operation, duration, interval) =
         ScenarioStats.create scenario stepsData simulationStats operation duration interval
 
     let _actor = ActionBlock(fun msg ->
         try
             match msg with
-            | AddResponse response ->
-                addResponse _allStepsData _intervalStepsData response
-
-            | AddResponses responses ->
-                responses |> Array.iter(addResponse _allStepsData _intervalStepsData)
+            | AddResponse response   -> addResponse response
+            | AddResponses responses -> responses |> List.iter addResponse
 
-            | PublishStatsToCoordinator ->
-                failwith "invalid operation" // it's only needed for cluster
+            | GetRawStats (reply, timestamp) ->
+                let stats = { ScenarioName = scenario.ScenarioName; Data = _intervalRawStats; Timestamp = timestamp }
+                reply.TrySetResult(stats) |> ignore
+                _intervalRawStats <- List.empty
 
             | GetRealtimeStats (reply, simulationStats, duration) ->
                 let scnStats = createScenarioStats(_intervalStepsData, simulationStats, OperationType.Bombing, duration, reportingInterval)
@@ -60,23 +66,31 @@ type ScenarioStatsActor(logger: ILogger, scenario: Scenario, reportingInterval:
                 let scnStats = createScenarioStats(_allStepsData, simulationStats, OperationType.Complete, duration, duration)
                 reply.TrySetResult(scnStats) |> ignore
         with
-        | ex -> logger.Error(ex, "GlobalScenarioStatsActor failed")
+        | ex -> logger.Error $"{nameof ScenarioStatsActor} failed: {ex.ToString()}"
     )
 
     interface IScenarioStatsActor with
 
         [<MethodImpl(MethodImplOptions.AggressiveInlining)>]
         member _.Publish(msg) = _actor.Post(msg) |> ignore
 
+        member _.GetRawStats(timestamp) =
+            let reply = TaskCompletionSource<ScenarioRawStats>()
+            GetRawStats(reply, timestamp) |> _actor.Post |> ignore
+            reply.Task
+
         member _.GetRealtimeStats(simulationStats, duration) =
-            let tcs = TaskCompletionSource<ScenarioStats>()
-            GetRealtimeStats(tcs, simulationStats, duration) |> _actor.Post |> ignore
-            tcs.Task
+            let reply = TaskCompletionSource<ScenarioStats>()
+            GetRealtimeStats(reply, simulationStats, duration) |> _actor.Post |> ignore
+            reply.Task
 
         member _.GetFinalStats(simulationStats, duration) =
-            let tcs = TaskCompletionSource<ScenarioStats>()
-            GetFinalStats(tcs, simulationStats, duration) |> _actor.Post |> ignore
-            tcs.Task
+            let reply = TaskCompletionSource<ScenarioStats>()
+            GetFinalStats(reply, simulationStats, duration) |> _actor.Post |> ignore
+            reply.Task
 
 let create (logger: ILogger) (scenario: Scenario) (reportingInterval: TimeSpan) =
-    ScenarioStatsActor(logger, scenario, reportingInterval) :> IScenarioStatsActor
+    ScenarioStatsActor(logger, scenario, reportingInterval, keepRawStats = false) :> IScenarioStatsActor
+
+let createWithRawStats (logger: ILogger) (scenario: Scenario) (reportingInterval: TimeSpan) =
+    ScenarioStatsActor(logger, scenario, reportingInterval, keepRawStats = true) :> IScenarioStatsActor