Logstasher instrumentation (#2559)

* Add LogStasher instrumentation

---------

Co-authored-by: fallwith <[email protected]>
Co-authored-by: Kayla Reopelle (she/her) <[email protected]>

CHANGELOG.md

@@ -1,8 +1,11 @@
 # New Relic Ruby Agent Release Notes
 
 ## <dev>
+Version <dev> introduces instrumentation for the LogStasher gem and improves instrumentation for the `redis-clustering` gem.
 
-Version <dev> improves instrumentation for the `redis-clustering` gem.
+- **Feature: Add instrumentation for LogStasher**
+
+  The agent will now record logs generated by [LogStasher](https://github.com/shadabahmed/logstasher). Versions 1.0.0 and above of the LogStasher gem are supported. [PR#2559](https://github.com/newrelic/newrelic-ruby-agent/pull/2559)
 
 - **Feature: Add instrumentation for redis-clustering**
 

lib/new_relic/agent/configuration/default_source.rb

@@ -1636,6 +1636,15 @@ def self.enforce_fallback(allowed_values: nil, fallback: nil)
           :allowed_from_server => false,
           :description => 'Controls auto-instrumentation of Ruby standard library Logger at start-up. May be one of: `auto`, `prepend`, `chain`, `disabled`.'
         },
+        :'instrumentation.logstasher' => {
+          :default => instrumentation_value_from_boolean(:'application_logging.enabled'),
+          :documentation_default => 'auto',
+          :public => true,
+          :type => String,
+          :dynamic_name => true,
+          :allowed_from_server => false,
+          :description => 'Controls auto-instrumentation of the LogStasher library at start-up. May be one of [auto|prepend|chain|disabled].'
+        },
         :'instrumentation.memcache' => {
           :default => 'auto',
           :documentation_default => 'auto',

lib/new_relic/agent/instrumentation/logstasher.rb

@@ -0,0 +1,27 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require_relative 'logstasher/instrumentation'
+require_relative 'logstasher/chain'
+require_relative 'logstasher/prepend'
+
+DependencyDetection.defer do
+  named :logstasher
+
+  depends_on do
+    defined?(LogStasher) &&
+      Gem::Version.new(LogStasher::VERSION) >= Gem::Version.new('1.0.0') &&
+      NewRelic::Agent.config[:'application_logging.enabled']
+  end
+
+  executes do
+    NewRelic::Agent.logger.info('Installing LogStasher instrumentation')
+
+    if use_prepend?
+      prepend_instrument LogStasher.singleton_class, NewRelic::Agent::Instrumentation::LogStasher::Prepend
+    else
+      chain_instrument NewRelic::Agent::Instrumentation::LogStasher::Chain
+    end
+  end
+end

lib/new_relic/agent/instrumentation/logstasher/chain.rb

@@ -0,0 +1,21 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module LogStasher::Chain
+    def self.instrument!
+      ::LogStasher.singleton_class.class_eval do
+        include NewRelic::Agent::Instrumentation::LogStasher
+
+        alias_method(:build_logstash_event_without_new_relic, :build_logstash_event)
+
+        def build_logstash_event(*args)
+          build_logstash_event_with_new_relic(*args) do
+            build_logstash_event_without_new_relic(*args)
+          end
+        end
+      end
+    end
+  end
+end

lib/new_relic/agent/instrumentation/logstasher/instrumentation.rb

@@ -0,0 +1,24 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module LogStasher
+    INSTRUMENTATION_NAME = NewRelic::Agent.base_name(name)
+
+    def self.enabled?
+      NewRelic::Agent.config[:'instrumentation.logstasher'] != 'disabled'
+    end
+
+    def build_logstash_event_with_new_relic(*args)
+      logstasher_event = yield
+      log = logstasher_event.instance_variable_get(:@data)
+
+      ::NewRelic::Agent.record_instrumentation_invocation(INSTRUMENTATION_NAME)
+      ::NewRelic::Agent.agent.log_event_aggregator.record_logstasher_event(log)
+      ::NewRelic::Agent::LocalLogDecorator.decorate(log)
+
+      logstasher_event
+    end
+  end
+end

lib/new_relic/agent/instrumentation/logstasher/prepend.rb

@@ -0,0 +1,13 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module LogStasher::Prepend
+    include NewRelic::Agent::Instrumentation::LogStasher
+
+    def build_logstash_event(*args)
+      build_logstash_event_with_new_relic(*args) { super }
+    end
+  end
+end

lib/new_relic/agent/local_log_decorator.rb

@@ -12,6 +12,12 @@ def decorate(message)
         return message unless decorating_enabled?
 
         metadata = NewRelic::Agent.linking_metadata
+
+        if message.is_a?(Hash)
+          message.merge!(metadata) unless message.frozen?
+          return
+        end
+
         formatted_metadata = " NR-LINKING|#{metadata[ENTITY_GUID_KEY]}|#{metadata[HOSTNAME_KEY]}|" \
                              "#{metadata[TRACE_ID_KEY]}|#{metadata[SPAN_ID_KEY]}|" \
                              "#{escape_entity_name(metadata[ENTITY_NAME_KEY])}|"
@@ -23,7 +29,8 @@ def decorate(message)
 
       def decorating_enabled?
         NewRelic::Agent.config[:'application_logging.enabled'] &&
-          NewRelic::Agent::Instrumentation::Logger.enabled? &&
+          (NewRelic::Agent::Instrumentation::Logger.enabled? ||
+            NewRelic::Agent::Instrumentation::LogStasher.enabled?) &&
           NewRelic::Agent.config[:'application_logging.local_decorating.enabled']
       end
 

lib/new_relic/agent/log_event_aggregator.rb

@@ -20,7 +20,8 @@ class LogEventAggregator < EventAggregator
       DROPPED_METRIC = 'Logging/Forwarding/Dropped'.freeze
       SEEN_METRIC = 'Supportability/Logging/Forwarding/Seen'.freeze
       SENT_METRIC = 'Supportability/Logging/Forwarding/Sent'.freeze
-      OVERALL_SUPPORTABILITY_FORMAT = 'Supportability/Logging/Ruby/Logger/%s'.freeze
+      LOGGER_SUPPORTABILITY_FORMAT = 'Supportability/Logging/Ruby/Logger/%s'.freeze
+      LOGSTASHER_SUPPORTABILITY_FORMAT = 'Supportability/Logging/Ruby/LogStasher/%s'.freeze
       METRICS_SUPPORTABILITY_FORMAT = 'Supportability/Logging/Metrics/Ruby/%s'.freeze
       FORWARDING_SUPPORTABILITY_FORMAT = 'Supportability/Logging/Forwarding/Ruby/%s'.freeze
       DECORATING_SUPPORTABILITY_FORMAT = 'Supportability/Logging/LocalDecorating/Ruby/%s'.freeze
@@ -58,38 +59,71 @@ def capacity
       end
 
       def record(formatted_message, severity)
-        return unless enabled?
+        return unless logger_enabled?
 
         severity = 'UNKNOWN' if severity.nil? || severity.empty?
+        increment_event_counters(severity)
+
+        return if formatted_message.nil? || formatted_message.empty?
+        return unless monitoring_conditions_met?(severity)
+
+        txn = NewRelic::Agent::Transaction.tl_current
+        priority = LogPriority.priority_for(txn)
 
-        if NewRelic::Agent.config[METRICS_ENABLED_KEY]
-          @counter_lock.synchronize do
-            @seen += 1
-            @seen_by_severity[severity] += 1
+        return txn.add_log_event(create_event(priority, formatted_message, severity)) if txn
+
+        @lock.synchronize do
+          @buffer.append(priority: priority) do
+            create_event(priority, formatted_message, severity)
           end
         end
+      rescue
+        nil
+      end
 
-        return if severity_too_low?(severity)
-        return if formatted_message.nil? || formatted_message.empty?
-        return unless NewRelic::Agent.config[FORWARDING_ENABLED_KEY]
-        return if @high_security
+      def record_logstasher_event(log)
+        return unless logstasher_enabled?
+
+        # LogStasher logs do not inherently include a message key, so most logs are recorded.
+        # But when the key exists, we should not record the log if the message value is nil or empty.
+        return if log.key?('message') && (log['message'].nil? || log['message'].empty?)
+
+        severity = determine_severity(log)
+        increment_event_counters(severity)
+
+        return unless monitoring_conditions_met?(severity)
 
         txn = NewRelic::Agent::Transaction.tl_current
         priority = LogPriority.priority_for(txn)
 
-        if txn
-          return txn.add_log_event(create_event(priority, formatted_message, severity))
-        else
-          return @lock.synchronize do
-            @buffer.append(priority: priority) do
-              create_event(priority, formatted_message, severity)
-            end
+        return txn.add_log_event(create_logstasher_event(priority, severity, log)) if txn
+
+        @lock.synchronize do
+          @buffer.append(priority: priority) do
+            create_logstasher_event(priority, severity, log)
           end
         end
       rescue
         nil
       end
 
+      def monitoring_conditions_met?(severity)
+        !severity_too_low?(severity) && NewRelic::Agent.config[FORWARDING_ENABLED_KEY] && !@high_security
+      end
+
+      def determine_severity(log)
+        log['level'] ? log['level'].to_s.upcase : 'UNKNOWN'
+      end
+
+      def increment_event_counters(severity)
+        return unless NewRelic::Agent.config[METRICS_ENABLED_KEY]
+
+        @counter_lock.synchronize do
+          @seen += 1
+          @seen_by_severity[severity] += 1
+        end
+      end
+
       def record_batch(txn, logs)
         # Ensure we have the same shared priority
         priority = LogPriority.priority_for(txn)
@@ -104,15 +138,17 @@ def record_batch(txn, logs)
         end
       end
 
-      def create_event(priority, formatted_message, severity)
-        formatted_message = truncate_message(formatted_message)
-
-        event = LinkingMetadata.append_trace_linking_metadata({
+      def add_event_metadata(formatted_message, severity)
+        metadata = {
           LEVEL_KEY => severity,
-          MESSAGE_KEY => formatted_message,
           TIMESTAMP_KEY => Process.clock_gettime(Process::CLOCK_REALTIME) * 1000
-        })
+        }
+        metadata[MESSAGE_KEY] = formatted_message unless formatted_message.nil?
+
+        LinkingMetadata.append_trace_linking_metadata(metadata)
+      end
 
+      def create_prioritized_event(priority, event)
         [
           {
             PrioritySampledBuffer::PRIORITY_KEY => priority
@@ -121,6 +157,31 @@ def create_event(priority, formatted_message, severity)
         ]
       end
 
+      def create_event(priority, formatted_message, severity)
+        formatted_message = truncate_message(formatted_message)
+        event = add_event_metadata(formatted_message, severity)
+
+        create_prioritized_event(priority, event)
+      end
+
+      def create_logstasher_event(priority, severity, log)
+        formatted_message = log['message'] ? truncate_message(log['message']) : nil
+        event = add_event_metadata(formatted_message, severity)
+        add_logstasher_event_attributes(event, log)
+
+        create_prioritized_event(priority, event)
+      end
+
+      def add_logstasher_event_attributes(event, log)
+        log_copy = log.dup
+        # Delete previously reported attributes
+        log_copy.delete('message')
+        log_copy.delete('level')
+        log_copy.delete('@timestamp')
+
+        event['attributes'] = log_copy
+      end
+
       def add_custom_attributes(custom_attributes)
         attributes.add_custom_attributes(custom_attributes)
       end
@@ -166,19 +227,23 @@ def reset!
         super
       end
 
-      def enabled?
+      def logger_enabled?
         @enabled && @instrumentation_logger_enabled
       end
 
+      def logstasher_enabled?
+        @enabled && NewRelic::Agent::Instrumentation::LogStasher.enabled?
+      end
+
       private
 
       # We record once-per-connect metrics for enabled/disabled state at the
       # point we consider the configuration stable (i.e. once we've gotten SSC)
       def register_for_done_configuring(events)
         events.subscribe(:server_source_configuration_added) do
           @high_security = NewRelic::Agent.config[:high_security]
-
-          record_configuration_metric(OVERALL_SUPPORTABILITY_FORMAT, OVERALL_ENABLED_KEY)
+          record_configuration_metric(LOGGER_SUPPORTABILITY_FORMAT, OVERALL_ENABLED_KEY)
+          record_configuration_metric(LOGSTASHER_SUPPORTABILITY_FORMAT, OVERALL_ENABLED_KEY)
           record_configuration_metric(METRICS_SUPPORTABILITY_FORMAT, METRICS_ENABLED_KEY)
           record_configuration_metric(FORWARDING_SUPPORTABILITY_FORMAT, FORWARDING_ENABLED_KEY)
           record_configuration_metric(DECORATING_SUPPORTABILITY_FORMAT, DECORATING_ENABLED_KEY)

test/multiverse/suites/logstasher/Envfile

@@ -0,0 +1,19 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+instrumentation_methods :chain, :prepend
+
+logstasher_versions = [
+  [nil, 2.7]
+]
+
+# Lock down activesupport version due to a logstasher test incompatiability with 7.1.
+def gem_list(logstasher_versions = nil)
+  <<~RB
+    gem 'logstasher'#{logstasher_versions}
+    gem 'activesupport', '< 7.1'
+  RB
+end
+
+create_gemfiles(logstasher_versions)

test/multiverse/suites/logstasher/config/newrelic.yml

@@ -0,0 +1,22 @@
+---
+development:
+  error_collector:
+    enabled: true
+  apdex_t: 0.5
+  monitor_mode: true
+  license_key: bootstrap_newrelic_admin_license_key_000
+  instrumentation:
+    logstasher: <%= $instrumentation_method %>
+  app_name: test
+  log_level: debug
+  host: 127.0.0.1
+  api_host: 127.0.0.1
+  transaction_trace:
+    record_sql: obfuscated
+    enabled: true
+    stack_trace_threshold: 0.5
+    transaction_threshold: 1.0
+  capture_params: false
+  application_logging:
+    forwarding:
+      enabled: true