mashhurs · Sep 19, 2022 · Sep 9, 2022 · Sep 9, 2022 · Sep 9, 2022 · Sep 13, 2022
diff --git a/docs/static/monitoring/monitoring-apis.asciidoc b/docs/static/monitoring/monitoring-apis.asciidoc
@@ -315,6 +315,9 @@ Gets process stats, including stats about file descriptors, memory consumption,
 <<event-stats,`events`>>::
 Gets event-related statistics for the Logstash instance (regardless of how many
 pipelines were created and destroyed).
+<<flow-stats,`flow`>>::
+Gets flow-related statistics for the Logstash instance (regardless of how many
+pipelines were created and destroyed).
 <<pipeline-stats,`pipelines`>>::
 Gets runtime stats about each Logstash pipeline.
 <<reload-stats,`reloads`>>::
@@ -454,6 +457,85 @@ Example response:
   }
 --------------------------------------------------
 
+[discrete]
+[[flow-stats]]
+==== Flow stats
+
+The following request returns a JSON document containing flow-rates
+for the Logstash instance:
+
+[source,js]
+--------------------------------------------------
+curl -XGET 'localhost:9600/_node/stats/flow?pretty'
+--------------------------------------------------
+
+Example response:
+
+[source,js]
+--------------------------------------------------
+{
+  "flow" : {
+    "input_throughput" : {
+      "current": 189.720,
+      "lifetime": 201.841
+    },
+    "filter_throughput" : {
+      "current": 187.810,
+      "lifetime": 201.799
+    },
+    "output_throughput" : {
+      "current": 191.087,
+      "lifetime": 201.761
+    },
+    "queue_backpressure" : {
+      "current": 0.277,
+      "lifetime": 0.031
+    },
+    "worker_concurrency" : {
+      "current": 1.973,
+      "lifetime": 1.721
+    }
+  }
+}
+--------------------------------------------------
+
+Flow rates provide visibility into how a Logstash instance or an individual pipeline is _currently_ performing relative to _itself_ over time.
+This allows us to attach _meaning_ to the cumulative-value metrics that are also presented by this API, and to determine whether an instance or pipeline is behaving better or worse than it has in the past.
+
+[%autowidth.stretch]
+|===
+|Flow Rate | Definition
+
+| `input_throughput` |
+This metric is expressed in events-per-second, and is the rate of events being pushed into the pipeline(s) queue(s) relative to wall-clock time (`events.in` / second).
+It includes events that are blocked by the queue and have not yet been accepted.
+
+| `filter_throughput` |
+This metric is expressed in events-per-second, and is the rate of events flowing through the filter phase of the pipeline(s) relative to wall-clock time (`events.filtered` / second).
+
+| `output_throughput` |
+This metric is expressed in events-per-second, and is the rate of events flowing through the output phase of the pipeline(s) relative to wall-clock time (`events.out` / second).
+
+| `worker_concurrency` |
+This is a unitless metric representing the cumulative time spent by all workers relative to wall-clock time (`duration_in_millis` / millisecond).
+
+A _pipeline_ is considered "saturated" when its `worker_concurrency` flow metric approaches its available `pipeline.workers`, because it indicates that all of its available workers are being kept busy.
+Tuning a saturated pipeline to have more workers can often work to increase that pipeline's throughput and decrease back-pressure to its queue, unless the pipeline is experiencing back-pressure from its outputs.
+
+A _process_ is also considered "saturated" when its top-level `worker_concurrency` flow metric approaches the _cumulative_ `pipeline.workers` across _all_ pipelines, and similarly can be addressed by tuning the <<pipeline-stats,individual pipelines>> that are saturated.
+
+| `queue_backpressure` |
+This is a unitless metric representing the cumulative time spent by all inputs blocked pushing events into their pipeline's queue, relative to wall-clock time (`queue_push_duration_in_millis` / millisecond).
+It is typically most useful when looking at the stats for an <<pipeline-stats,individual pipeline>>.
+
+While a "zero" value indicates no back-pressure to the queue, the magnitude of this metric is highly dependent on the _shape_ of the pipelines and their inputs.
+It cannot be used to compare one pipeline to another or even one process to _itself_ if the quantity or shape of its pipelines changes.
+A pipeline with only one single-threaded input may contribute up to 1.00, a pipeline whose inputs have hundreds of inbound connections may contribute much higher numbers to this combined value.
+
+Additionally, some amount of back-pressure is both _normal_ and _expected_ for pipelines that are _pulling_ data, as this back-pressure allows them to slow down and pull data at a rate its downstream pipeline can tolerate.
+
+|===
+
 [discrete]
 [[pipeline-stats]]
 ==== Pipeline stats
@@ -462,6 +544,7 @@ The following request returns a JSON document containing pipeline stats,
 including:
 
 * the number of events that were input, filtered, or output by each pipeline
+* the current and lifetime <<flow-stats,_flow_ rates>> for each pipeline
 * stats for each configured filter or output stage
 * info about config reload successes and failures
 (when <<reloading-config,config reload>> is enabled)
@@ -487,6 +570,28 @@ Example response:
         "out" : 216485,
         "queue_push_duration_in_millis" : 342466
       },
+      "flow" : {
+        "input_throughput" : {
+          "current": 189.720,
+          "lifetime": 201.841
+        },
+        "filter_throughput" : {
+          "current": 187.810,
+          "lifetime": 201.799
+        },
+        "output_throughput" : {
+          "current": 191.087,
+          "lifetime": 201.761
+        },
+        "queue_backpressure" : {
+          "current": 0.277,
+          "lifetime": 0.031
+        },
+        "worker_concurrency" : {
+          "current": 1.973,
+          "lifetime": 1.721
+        }
+      },
       "plugins" : {
         "inputs" : [ {
           "id" : "35131f351e2dc5ed13ee04265a8a5a1f95292165-1",
@@ -546,6 +651,28 @@ Example response:
         "out" : 87247,
         "queue_push_duration_in_millis" : 1532
       },
+      "flow" : {
+        "input_throughput" : {
+          "current": 189.720,
+          "lifetime": 201.841
+        },
+        "filter_throughput" : {
+          "current": 187.810,
+          "lifetime": 201.799
+        },
+        "output_throughput" : {
+          "current": 191.087,
+          "lifetime": 201.761
+        },
+        "queue_backpressure" : {
+          "current": 0.871,
+          "lifetime": 0.031
+        },
+        "worker_concurrency" : {
+          "current": 4.71,
+          "lifetime": 1.201
+        }
+      },
       "plugins" : {
         "inputs" : [ {
           "id" : "d7ea8941c0fc48ac58f89c84a9da482107472b82-1",
@@ -601,6 +728,28 @@ Example response:
         "out" : 216485,
         "queue_push_duration_in_millis" : 342466
       },
+      "flow" : {
+        "input_throughput" : {
+          "current": 189.720,
+          "lifetime": 201.841
+        },
+        "filter_throughput" : {
+          "current": 187.810,
+          "lifetime": 201.799
+        },
+        "output_throughput" : {
+          "current": 191.087,
+          "lifetime": 201.761
+        },
+        "queue_backpressure" : {
+          "current": 0.277,
+          "lifetime": 0.031
+        },
+        "worker_concurrency" : {
+          "current": 1.973,
+          "lifetime": 1.721
+        }
+      },
       "plugins" : {
         "inputs" : [ {
           "id" : "35131f351e2dc5ed13ee04265a8a5a1f95292165-1",

diff --git a/logstash-core/lib/logstash/agent.rb b/logstash-core/lib/logstash/agent.rb
@@ -92,6 +92,7 @@ def initialize(settings = LogStash::SETTINGS, source_loader = nil)
     @pipeline_reload_metric = metric.namespace([:stats, :pipelines])
     @instance_reload_metric = metric.namespace([:stats, :reloads])
     initialize_agent_metrics
+    initialize_flow_metrics
 
     initialize_geoip_database_metrics(metric)
 
@@ -534,6 +535,51 @@ def initialize_agent_metrics
     @instance_reload_metric.increment(:failures, 0)
   end
 
+  def initialize_flow_metrics
+    if collect_metrics? && metric.collector
+
+      java_import org.logstash.instrument.metrics.UptimeMetric
+      java_import org.logstash.instrument.metrics.UptimeMetric::ScaleUnits
+
+      uptime_metric = UptimeMetric.new
+      uptime_precise_millis = uptime_metric.with_units_precise(ScaleUnits::MILLISECONDS)
+      uptime_precise_seconds = uptime_metric.with_units_precise(ScaleUnits::SECONDS)
+
+      events_namespace = metric.namespace([:stats,:events])
+      flow_metrics = []
+      flow_metrics << create_flow_metric("input_throughput", get_counter(events_namespace, :in), uptime_precise_seconds)
+      flow_metrics << create_flow_metric("filter_throughput", get_counter(events_namespace, :out), uptime_precise_seconds)
+      flow_metrics << create_flow_metric("output_throughput", get_counter(events_namespace, :filtered), uptime_precise_seconds)
+      flow_metrics << create_flow_metric("queue_backpressure", get_counter(events_namespace, :queue_push_duration_in_millis), uptime_precise_millis)
+      flow_metrics << create_flow_metric("worker_concurrency", get_counter(events_namespace, :duration_in_millis), uptime_precise_millis)
+
+      registered, unregistered = flow_metrics.partition do |flow_metric|
+        @metric.collector.register?([:stats,:flow], flow_metric.name.to_sym, flow_metric)
+      end
+
+      unregistered.each do |unregistered_flow_metric|
+        logger.warn("Failed to register global flow metric #{unregistered_flow_metric.name}.")
+      end
+
+      @flow_metrics = registered.freeze
+    end
+  end
+
+  def get_counter(namespace, key)
+    org.logstash.instrument.metrics.counter.LongCounter.fromRubyBase(namespace, key)
+  end
+  private :get_counter
+
+  def create_flow_metric(name, numerator_metric, denominator_metric)
+    org.logstash.instrument.metrics.FlowMetric.new(name, numerator_metric, denominator_metric)
+  end
+  private :create_flow_metric
+
+  def capture_flow_metrics
+    @flow_metrics&.each(&:capture)
+  end
+  public :capture_flow_metrics
+
   def initialize_pipeline_metrics(action)
     @pipeline_reload_metric.namespace([action.pipeline_id, :reloads]).tap do |n|
       n.increment(:successes, 0)

diff --git a/logstash-core/lib/logstash/api/commands/stats.rb b/logstash-core/lib/logstash/api/commands/stats.rb
@@ -81,6 +81,16 @@ def events
           {}
         end
 
+        def flow
+          extract_metrics(
+            [:stats,:flow],
+            :input_throughput, :filter_throughput, :output_throughput, :queue_backpressure, :worker_concurrency
+          )
+        rescue LogStash::Instrument::MetricStore::MetricNotFound
+          # if the stats/events metrics have not yet been populated, return an empty map
+          {}
+        end
+
         def pipeline(pipeline_id = nil, opts={})
           extended_stats = LogStash::Config::PipelinesInfo.format_pipelines_info(
             service.agent,
@@ -165,6 +175,7 @@ def plugin_stats(stats, plugin_type)
           def report(stats, extended_stats=nil, opts={})
             ret = {
               :events => stats[:events],
+              :flow => stats[:flow],
               :plugins => {
                 :inputs => plugin_stats(stats, :inputs),
                 :codecs => plugin_stats(stats, :codecs),

diff --git a/logstash-core/lib/logstash/api/modules/node_stats.rb b/logstash-core/lib/logstash/api/modules/node_stats.rb
@@ -35,6 +35,7 @@ class NodeStats < ::LogStash::Api::Modules::Base
             :jvm => jvm_payload,
             :process => process_payload,
             :events => events_payload,
+            :flow => flow_payload,
             :pipelines => pipeline_payload,
             :reloads => reloads_payload,
             :os => os_payload,
@@ -61,6 +62,10 @@ def events_payload
           @stats.events
         end
 
+        def flow_payload
+          @stats.flow
+        end
+
         def jvm_payload
           @stats.jvm
         end

diff --git a/logstash-core/lib/logstash/instrument/collector.rb b/logstash-core/lib/logstash/instrument/collector.rb
@@ -67,6 +67,28 @@ def get(namespaces_path, key, type)
       end
     end
 
+    ##
+    # Ensures that a metric on the provided `namespaces_path` with the provided `key`
+    # is registered, using the provided `metric_instance` IFF it is not already present.
+    #
+    # @param namespaces_path [Array<Symbol>]
+    # @param key [Symbol]
+    # @param metric_instance [Metric]
+    #
+    # @return [Boolean] true IFF the provided `metric_instance` was registered
+    def register?(namespaces_path, key, metric_instance)
+      registered = false
+
+      # Relies on MetricStore#fetch_or_store yielding the block
+      # EXACTLY ONCE to the winner in a race-condition.
+      @metric_store.fetch_or_store(namespaces_path, key) do
+        registered = true
+        metric_instance
+      end
+
+      registered
+    end
+
     # Snapshot the current Metric Store and return it immediately,
     # This is useful if you want to get access to the current metric store without
     # waiting for a periodic call.