From 46fb4c6ab951a0cfe1e864e138ee8dd07733accc Mon Sep 17 00:00:00 2001 From: Emily Stolfo Date: Wed, 20 May 2026 12:42:06 +0200 Subject: [PATCH 1/2] Add additional fields for integration dashboards --- .../instrument/periodic_poller/otel.rb | 199 +++++++++++++++++- .../instrument/periodic_poller/otel_spec.rb | 86 +++++++- 2 files changed, 276 insertions(+), 9 deletions(-) diff --git a/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb b/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb index 41110118d1..2e5c154acf 100644 --- a/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb +++ b/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb @@ -172,6 +172,10 @@ def register_global_metrics get_metric_value(:stats, :events, :filtered) end + register_observable_counter("logstash.events.duration", "Total time spent processing events", "ms") do + get_metric_value(:stats, :events, :duration_in_millis) + end + # Global queue gauge (total across all pipelines) register_gauge("logstash.queue.events", "Total events in queues", "{event}") do get_total_queue_events @@ -331,6 +335,46 @@ def register_cgroup_metrics def register_pipeline_gauges_for(pipeline_id) attrs = create_pipeline_attributes(pipeline_id) + register_gauge( + "logstash.pipeline.workers", + "Number of worker threads for the pipeline", + "{thread}", + attrs + ) do + get_pipeline_metric_value(pipeline_id, :config, :workers) + end + + register_gauge( + "logstash.pipeline.batch.size", + "Number of events processed per batch", + "{event}", + attrs + ) do + get_pipeline_metric_value(pipeline_id, :config, :batch_size) + end + + register_gauge( + "logstash.pipeline.batch.delay", + "Delay in milliseconds before a batch is flushed", + "ms", + attrs + ) do + get_pipeline_metric_value(pipeline_id, :config, :batch_delay) + end + + register_gauge( + "logstash.pipeline.queue.type", + "Queue type: 0 = memory, 1 = persisted", + "1", + attrs + ) do + queue_type = get_pipeline_metric_value(pipeline_id, :queue, :type) + case queue_type + when "persisted" then 1 + when "memory" then 0 + end + end + register_gauge( "logstash.pipeline.queue.events", "Events in pipeline queue", @@ -417,6 +461,42 @@ def register_pipeline_counters_for(pipeline_id) ) do get_pipeline_metric_value(pipeline_id, :events, :filtered) end + + register_observable_counter( + "logstash.pipeline.events.duration", + "Time spent processing events by pipeline", + "ms", + attrs + ) do + get_pipeline_metric_value(pipeline_id, :events, :duration_in_millis) + end + + register_observable_counter( + "logstash.pipeline.events.queue_push_duration", + "Time spent pushing events to the queue", + "ms", + attrs + ) do + get_pipeline_metric_value(pipeline_id, :events, :queue_push_duration_in_millis) + end + + register_gauge( + "logstash.pipeline.batch.byte_size", + "Current batch byte size", + "By", + attrs + ) do + get_batch_current_byte_size(pipeline_id) + end + + register_gauge( + "logstash.pipeline.batch.event_count", + "Current batch event count", + "{event}", + attrs + ) do + get_batch_current_event_count(pipeline_id) + end end def register_gauge(name, description, unit, attributes = Attributes.empty, &block) @@ -478,11 +558,12 @@ def create_pipeline_attributes(pipeline_id) ) end - def create_plugin_attributes(pipeline_id, plugin_type, plugin_id) + def create_plugin_attributes(pipeline_id, plugin_type, plugin_id, plugin_name) Attributes.of( AttributeKey.stringKey("pipeline.id"), pipeline_id.to_s, AttributeKey.stringKey("plugin.type"), plugin_type.to_s, - AttributeKey.stringKey("plugin.id"), plugin_id.to_s + AttributeKey.stringKey("plugin.id"), plugin_id.to_s, + AttributeKey.stringKey("plugin.name"), plugin_name.to_s ) end @@ -495,14 +576,30 @@ def register_plugin_metrics_for(pipeline_id) plugin_key = "#{pipeline_id}:#{plugin_type}:#{plugin_id}" next if @registered_plugins.include?(plugin_key) + plugin_name = get_plugin_metric_value(pipeline_id, plugin_type, plugin_id, :name) || plugin_id logger.debug("Registering Otel metrics for plugin", :pipeline_id => pipeline_id, :plugin_type => plugin_type, - :plugin_id => plugin_id) - register_plugin_counters_for(pipeline_id, plugin_type, plugin_id) + :plugin_id => plugin_id, + :plugin_name => plugin_name) + register_plugin_counters_for(pipeline_id, plugin_type, plugin_id, plugin_name) @registered_plugins.add(plugin_key) end end + + codec_ids = get_plugin_ids(pipeline_id, :codecs) + codec_ids.each do |codec_id| + plugin_key = "#{pipeline_id}:codecs:#{codec_id}" + next if @registered_plugins.include?(plugin_key) + + codec_name = get_codec_metric_value(pipeline_id, codec_id, :name) || codec_id + logger.debug("Registering Otel metrics for codec", + :pipeline_id => pipeline_id, + :codec_id => codec_id, + :codec_name => codec_name) + register_codec_counters_for(pipeline_id, codec_id, codec_name) + @registered_plugins.add(plugin_key) + end end def get_plugin_ids(pipeline_id, plugin_type) @@ -516,8 +613,8 @@ def get_plugin_ids(pipeline_id, plugin_type) end end - def register_plugin_counters_for(pipeline_id, plugin_type, plugin_id) - attrs = create_plugin_attributes(pipeline_id, plugin_type, plugin_id) + def register_plugin_counters_for(pipeline_id, plugin_type, plugin_id, plugin_name) + attrs = create_plugin_attributes(pipeline_id, plugin_type, plugin_id, plugin_name) register_observable_counter( "logstash.plugin.events.in", @@ -545,11 +642,101 @@ def register_plugin_counters_for(pipeline_id, plugin_type, plugin_id) ) do get_plugin_metric_value(pipeline_id, plugin_type, plugin_id, :events, :duration_in_millis) end + + register_observable_counter( + "logstash.plugin.events.queue_push_duration", + "Time spent pushing events to the queue", + "ms", + attrs + ) do + get_plugin_metric_value(pipeline_id, plugin_type, plugin_id, :events, :queue_push_duration_in_millis) + end + end + + def register_codec_counters_for(pipeline_id, codec_id, codec_name) + encode_attrs = create_plugin_attributes(pipeline_id, "codec.encode", codec_id, codec_name) + decode_attrs = create_plugin_attributes(pipeline_id, "codec.decode", codec_id, codec_name) + + register_observable_counter( + "logstash.plugin.writes_in", + "Events written to plugin", + "{event}", + encode_attrs + ) do + get_codec_metric_value(pipeline_id, codec_id, :encode, :writes_in) + end + + register_observable_counter( + "logstash.plugin.events.duration", + "Time spent processing events", + "ms", + encode_attrs + ) do + get_codec_metric_value(pipeline_id, codec_id, :encode, :duration_in_millis) + end + + register_observable_counter( + "logstash.plugin.writes_in", + "Events written to plugin", + "{event}", + decode_attrs + ) do + get_codec_metric_value(pipeline_id, codec_id, :decode, :writes_in) + end + + register_observable_counter( + "logstash.plugin.writes_out", + "Events written out by plugin", + "{event}", + decode_attrs + ) do + get_codec_metric_value(pipeline_id, codec_id, :decode, :out) + end + + register_observable_counter( + "logstash.plugin.events.duration", + "Time spent processing events", + "ms", + decode_attrs + ) do + get_codec_metric_value(pipeline_id, codec_id, :decode, :duration_in_millis) + end + end + + def get_batch_current_byte_size(pipeline_id) + get_batch_current_value(pipeline_id, 1) + end + + def get_batch_current_event_count(pipeline_id) + get_batch_current_value(pipeline_id, 0) + end + + def get_batch_current_value(pipeline_id, index) + store = @snapshot.metric_store + current = store.get_shallow(:stats, :pipelines, pipeline_id.to_sym, :batch, :current) + current&.value&.[](index) + rescue LogStash::Instrument::MetricStore::MetricNotFound + nil end def get_plugin_metric_value(pipeline_id, plugin_type, plugin_id, *path) full_path = [:stats, :pipelines, pipeline_id.to_sym, :plugins, plugin_type, plugin_id.to_sym] + path get_metric_value(*full_path) end + + def get_codec_metric_value(pipeline_id, codec_id, *path) + store = @snapshot.metric_store + codecs_hash = store.get_shallow(:stats, :pipelines, pipeline_id.to_sym, :plugins, :codecs) + return nil unless codecs_hash.is_a?(Hash) + node = codecs_hash[codec_id] + return nil unless node.is_a?(Hash) + value = path.reduce(node) do |acc, key| + break nil unless acc.is_a?(Hash) + acc[key] + end + value.is_a?(Hash) ? nil : value&.value + rescue LogStash::Instrument::MetricStore::MetricNotFound + nil + end end end; end; end diff --git a/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb b/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb index bb2f012dee..9643d0cd3f 100644 --- a/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb +++ b/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb @@ -270,6 +270,9 @@ expect(otel_service).to receive(:registerObservableCounter).with( "logstash.events.filtered", anything, anything, anything, anything ) + expect(otel_service).to receive(:registerObservableCounter).with( + "logstash.events.duration", anything, anything, anything, anything + ) expect(otel_service).to receive(:registerGauge).with( "logstash.queue.events", anything, anything, anything, anything ) @@ -457,6 +460,60 @@ otel_poller.collect end end + + context "with codecs" do + let(:metric_store) do + double("metric_store").tap do |store| + allow(store).to receive(:get_shallow).and_return(nil) + allow(store).to receive(:get_shallow) + .with(:stats, :pipelines, :main, :plugins, :filters) + .and_return({}) + allow(store).to receive(:get_shallow) + .with(:stats, :pipelines, :main, :plugins, :outputs) + .and_return({}) + allow(store).to receive(:get_shallow) + .with(:stats, :pipelines, :main, :plugins, :inputs) + .and_return({}) + allow(store).to receive(:get_shallow) + .with(:stats, :pipelines, :main, :plugins, :codecs) + .and_return({ :plain_abc123 => {} }) + end + end + + let(:snapshot) do + double("snapshot", :metric_store => metric_store) + end + + before do + otel_poller.collect + allow(collector).to receive(:snapshot_metric).and_return(snapshot) + otel_poller.instance_variable_set(:@snapshot, snapshot) + end + + it "registers codec metrics with codec.encode and codec.decode plugin.type" do + expect(otel_service).to receive(:registerObservableCounter).with( + "logstash.plugin.writes_in", anything, anything, anything, anything + ).exactly(2).times + expect(otel_service).to receive(:registerObservableCounter).with( + "logstash.plugin.events.duration", anything, anything, anything, anything + ).exactly(2).times + expect(otel_service).to receive(:registerObservableCounter).with( + "logstash.plugin.writes_out", anything, anything, anything, anything + ).once + + otel_poller.collect + end + + it "only registers each codec once across multiple collects" do + otel_poller.collect + + expect(otel_service).not_to receive(:registerObservableCounter).with( + "logstash.plugin.writes_in", anything, anything, anything, anything + ) + + otel_poller.collect + end + end end describe "#stop" do @@ -682,20 +739,43 @@ def mval(*metric_path) java_import 'io.opentelemetry.api.common.AttributeKey' end - it "creates Attributes with pipeline.id, plugin.type, and plugin.id" do - attrs = otel_poller.send(:create_plugin_attributes, :main, :filters, :mutate_abc123) + it "creates Attributes with pipeline.id, plugin.type, plugin.id, and plugin.name" do + attrs = otel_poller.send(:create_plugin_attributes, :main, :filters, :mutate_abc123, "mutate") expect(attrs.get(AttributeKey.stringKey("pipeline.id"))).to eq("main") expect(attrs.get(AttributeKey.stringKey("plugin.type"))).to eq("filters") expect(attrs.get(AttributeKey.stringKey("plugin.id"))).to eq("mutate_abc123") + expect(attrs.get(AttributeKey.stringKey("plugin.name"))).to eq("mutate") end it "converts all symbol arguments to strings" do - attrs = otel_poller.send(:create_plugin_attributes, :secondary, :outputs, :elasticsearch_xyz) + attrs = otel_poller.send(:create_plugin_attributes, :secondary, :outputs, :elasticsearch_xyz, "elasticsearch") expect(attrs.get(AttributeKey.stringKey("pipeline.id"))).to eq("secondary") expect(attrs.get(AttributeKey.stringKey("plugin.type"))).to eq("outputs") expect(attrs.get(AttributeKey.stringKey("plugin.id"))).to eq("elasticsearch_xyz") + expect(attrs.get(AttributeKey.stringKey("plugin.name"))).to eq("elasticsearch") + end + end + + describe "codec attributes via #create_plugin_attributes" do + before do + java_import 'io.opentelemetry.api.common.AttributeKey' + end + + it "sets plugin.type to codec.encode for encode attributes" do + attrs = otel_poller.send(:create_plugin_attributes, :main, "codec.encode", :plain_abc123, "plain") + + expect(attrs.get(AttributeKey.stringKey("pipeline.id"))).to eq("main") + expect(attrs.get(AttributeKey.stringKey("plugin.type"))).to eq("codec.encode") + expect(attrs.get(AttributeKey.stringKey("plugin.id"))).to eq("plain_abc123") + expect(attrs.get(AttributeKey.stringKey("plugin.name"))).to eq("plain") + end + + it "sets plugin.type to codec.decode for decode attributes" do + attrs = otel_poller.send(:create_plugin_attributes, :main, "codec.decode", :plain_abc123, "plain") + + expect(attrs.get(AttributeKey.stringKey("plugin.type"))).to eq("codec.decode") end end end From eb239060dd8aeb09f4127f2a933b7c9f121f76d5 Mon Sep 17 00:00:00 2001 From: Emily Stolfo Date: Thu, 4 Jun 2026 13:03:19 +0200 Subject: [PATCH 2/2] Add CPU load metrics --- .../lib/logstash/instrument/periodic_poller/otel.rb | 12 ++++++++++++ .../logstash/instrument/periodic_poller/otel_spec.rb | 3 +++ 2 files changed, 15 insertions(+) diff --git a/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb b/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb index 2e5c154acf..94980436bc 100644 --- a/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb +++ b/logstash-core/lib/logstash/instrument/periodic_poller/otel.rb @@ -245,6 +245,18 @@ def register_jvm_metrics register_gauge("logstash.jvm.process.cpu.percent", "JVM process CPU usage", "%") do get_metric_value(:jvm, :process, :cpu, :percent) end + register_gauge("logstash.os.cpu.load_average.1m", "OS CPU 1-minute load average", "1") do + v = get_metric_value(:jvm, :process, :cpu, :load_average) + v.is_a?(Hash) ? v[:"1m"] : nil + end + register_gauge("logstash.os.cpu.load_average.5m", "OS CPU 5-minute load average", "1") do + v = get_metric_value(:jvm, :process, :cpu, :load_average) + v.is_a?(Hash) ? v[:"5m"] : nil + end + register_gauge("logstash.os.cpu.load_average.15m", "OS CPU 15-minute load average", "1") do + v = get_metric_value(:jvm, :process, :cpu, :load_average) + v.is_a?(Hash) ? v[:"15m"] : nil + end register_observable_counter("logstash.jvm.process.cpu.total", "JVM process CPU total time", "ms") do get_metric_value(:jvm, :process, :cpu, :total_in_millis) end diff --git a/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb b/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb index 9643d0cd3f..b41147c9e1 100644 --- a/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb +++ b/logstash-core/spec/logstash/instrument/periodic_poller/otel_spec.rb @@ -294,6 +294,9 @@ expect(otel_service).to receive(:registerGauge).with("logstash.jvm.process.open_file_descriptors", anything, anything, anything, anything) expect(otel_service).to receive(:registerGauge).with("logstash.jvm.process.max_file_descriptors", anything, anything, anything, anything) expect(otel_service).to receive(:registerGauge).with("logstash.jvm.process.cpu.percent", anything, anything, anything, anything) + expect(otel_service).to receive(:registerGauge).with("logstash.os.cpu.load_average.1m", anything, anything, anything, anything) + expect(otel_service).to receive(:registerGauge).with("logstash.os.cpu.load_average.5m", anything, anything, anything, anything) + expect(otel_service).to receive(:registerGauge).with("logstash.os.cpu.load_average.15m", anything, anything, anything, anything) expect(otel_service).to receive(:registerObservableCounter).with("logstash.jvm.process.cpu.total", anything, anything, anything, anything) expect(otel_service).to receive(:registerGauge).with("logstash.jvm.uptime", anything, anything, anything, anything)