Merge pull request #4154 from DataDog/dedup-logs

Telemetry: Deduplicate log entries

Author: marcotc

lib/datadog/core/telemetry/event.rb

@@ -29,6 +29,22 @@ def type
           def payload
             {}
           end
+
+          # Override equality to allow for deduplication
+          # The basic implementation is to check if the other object is an instance of the same class.
+          # This works for events that have no attributes.
+          # For events with attributes, you should override this method to compare the attributes.
+          def ==(other)
+            other.is_a?(self.class)
+          end
+
+          # @see #==
+          alias eql? ==
+
+          # @see #==
+          def hash
+            self.class.hash
+          end
         end
 
         # Telemetry class for the 'app-started' event
@@ -263,6 +279,8 @@ def patch_error(integration)
 
         # Telemetry class for the 'app-client-configuration-change' event
         class AppClientConfigurationChange < Base
+          attr_reader :changes, :origin
+
           def type
             'app-client-configuration-change'
           end
@@ -301,6 +319,16 @@ def configuration
 
             res
           end
+
+          def ==(other)
+            other.is_a?(AppClientConfigurationChange) && other.changes == @changes && other.origin == @origin
+          end
+
+          alias eql? ==
+
+          def hash
+            [self.class, @changes, @origin].hash
+          end
         end
 
         # Telemetry class for the 'app-heartbeat' event
@@ -319,6 +347,8 @@ def type
 
         # Telemetry class for the 'generate-metrics' event
         class GenerateMetrics < Base
+          attr_reader :namespace, :metric_series
+
           def type
             'generate-metrics'
           end
@@ -335,24 +365,54 @@ def payload
               series: @metric_series.map(&:to_h)
             }
           end
+
+          def ==(other)
+            other.is_a?(GenerateMetrics) && other.namespace == @namespace && other.metric_series == @metric_series
+          end
+
+          alias eql? ==
+
+          def hash
+            [self.class, @namespace, @metric_series].hash
+          end
         end
 
-        # Telemetry class for the 'logs' event
+        # Telemetry class for the 'logs' event.
+        # Logs with the same content are deduplicated at flush time.
         class Log < Base
           LEVELS = {
             error: 'ERROR',
             warn: 'WARN',
           }.freeze
 
+          LEVELS_STRING = LEVELS.values.freeze
+
+          attr_reader :message, :level, :stack_trace, :count
+
           def type
             'logs'
           end
 
-          def initialize(message:, level:, stack_trace: nil)
+          # @param message [String] the log message
+          # @param level [Symbol, String] the log level. Either :error, :warn, 'ERROR', or 'WARN'.
+          # @param stack_trace [String, nil] the stack trace
+          # @param count [Integer] the number of times the log was emitted. Used for deduplication.
+          def initialize(message:, level:, stack_trace: nil, count: 1)
             super()
             @message = message
             @stack_trace = stack_trace
-            @level = LEVELS.fetch(level) { |k| raise ArgumentError, "Invalid log level :#{k}" }
+
+            if level.is_a?(String) && LEVELS_STRING.include?(level)
+              # String level is used during object copy for deduplication
+              @level = level
+            elsif level.is_a?(Symbol)
+              # Symbol level is used by the regular log emitter user
+              @level = LEVELS.fetch(level) { |k| raise ArgumentError, "Invalid log level :#{k}" }
+            else
+              raise ArgumentError, "Invalid log level #{level}"
+            end
+
+            @count = count
           end
 
           def payload
@@ -362,10 +422,24 @@ def payload
                   message: @message,
                   level: @level,
                   stack_trace: @stack_trace,
+                  count: @count,
                 }.compact
               ]
             }
           end
+
+          # override equality to allow for deduplication
+          def ==(other)
+            other.is_a?(Log) &&
+              other.message == @message &&
+              other.level == @level && other.stack_trace == @stack_trace && other.count == @count
+          end
+
+          alias eql? ==
+
+          def hash
+            [self.class, @message, @level, @stack_trace, @count].hash
+          end
         end
 
         # Telemetry class for the 'distributions' event
@@ -395,6 +469,16 @@ def payload
               }
             end
           end
+
+          def ==(other)
+            other.is_a?(MessageBatch) && other.events == @events
+          end
+
+          alias eql? ==
+
+          def hash
+            [self.class, @events].hash
+          end
         end
       end
     end

lib/datadog/core/telemetry/metric.rb

@@ -41,6 +41,18 @@ def to_h
             }
           end
 
+          def ==(other)
+            other.is_a?(self.class) &&
+              name == other.name &&
+              values == other.values && tags == other.tags && common == other.common && type == other.type
+          end
+
+          alias eql? ==
+
+          def hash
+            [self.class, name, values, tags, common, type].hash
+          end
+
           private
 
           def tags_to_array(tags)
@@ -71,6 +83,16 @@ def to_h
             res[:interval] = interval
             res
           end
+
+          def ==(other)
+            super && interval == other.interval
+          end
+
+          alias eql? ==
+
+          def hash
+            [super, interval].hash
+          end
         end
 
         # Count metric adds up all the submitted values in a time interval. This would be suitable for a

lib/datadog/core/telemetry/worker.rb

@@ -97,6 +97,8 @@ def flush_events(events)
           return if events.empty?
           return if !enabled? || !sent_started_event?
 
+          events = deduplicate_logs(events)
+
           Datadog.logger.debug { "Sending #{events&.count} telemetry events" }
           send_event(Event::MessageBatch.new(events))
         end
@@ -167,6 +169,37 @@ def disable_on_not_found!(response)
           Datadog.logger.debug('Agent does not support telemetry; disabling future telemetry events.')
           disable!
         end
+
+        # Deduplicate logs by counting the number of repeated occurrences of the same log
+        # entry and replacing them with a single entry with the calculated `count` value.
+        # Non-log events are unchanged.
+        def deduplicate_logs(events)
+          return events if events.empty?
+
+          all_logs = []
+          other_events = events.reject do |event|
+            if event.is_a?(Event::Log)
+              all_logs << event
+              true
+            else
+              false
+            end
+          end
+
+          return events if all_logs.empty?
+
+          uniq_logs = all_logs.group_by(&:itself).map do |_, logs|
+            log = logs.first
+            if logs.size > 1
+              # New log event with a count of repeated occurrences
+              Event::Log.new(message: log.message, level: log.level, stack_trace: log.stack_trace, count: logs.size)
+            else
+              log
+            end
+          end
+
+          other_events + uniq_logs
+        end
       end
     end
   end

sig/datadog/core/telemetry/component.rbs

@@ -20,7 +20,7 @@ module Datadog
 
         def disable!: () -> void
 
-        def client_configuration_change!: (Enumerable[[String, Numeric | bool | String]] changes) -> void
+        def client_configuration_change!: (Array[[String, Numeric | bool | String]] changes) -> void
 
         def emit_closing!: () -> void
 

sig/datadog/core/telemetry/event.rbs

@@ -46,10 +46,10 @@ module Datadog
         end
 
         class AppClientConfigurationChange < Base
-          @changes: Enumerable[[String, Numeric | bool | String | int]]
-          @origin: String
+          attr_reader changes: Array[[String, Numeric | bool | String | int]]
+          attr_reader origin: String
 
-          def initialize: (Enumerable[[String, Numeric | bool | String]] changes, String origin) -> void
+          def initialize: (Array[[String, Numeric | bool | String]] changes, String origin) -> void
 
           def configuration: () -> Array[Hash[Symbol, untyped]]
         end
@@ -61,22 +61,25 @@ module Datadog
         end
 
         class GenerateMetrics < Base
-          @namespace: String
-          @metric_series: Enumerable[Datadog::Core::Telemetry::Metric::Base]
+          attr_reader namespace: String
+          attr_reader metric_series: Array[Metric::Base]
 
-          def initialize: (String namespace, Enumerable[Datadog::Core::Telemetry::Metric::Base] metric_series) -> void
+          def initialize: (String namespace, Array[Metric::Base] metric_series) -> void
         end
 
         class Log < Base
           LEVELS: Hash[Symbol, String]
 
-          @message: String
-          @level: "ERROR" | "DEBUG" | "WARN"
-          @stack_trace: String?
+          LEVELS_STRING: Array[String]
 
-          def initialize: (message: String, level: Symbol, ?stack_trace: String?) -> void
+          attr_reader count: Integer
+          attr_reader message: String
+          attr_reader level: String
+          attr_reader stack_trace: String?
 
-          def payload: () -> { logs: [Hash[Symbol, String]] }
+          def initialize: (message: String, level: (Symbol|String), ?stack_trace: String?, ?count: Integer) -> void
+
+          def payload: () -> { logs: [Hash[Symbol, (String|Integer)]] }
         end
 
         class Distributions < GenerateMetrics

sig/datadog/core/telemetry/worker.rbs

@@ -35,6 +35,8 @@ module Datadog
 
         private
 
+        def deduplicate_logs: (Array[Event::Base] events) -> Array[Event::Base]
+
         def heartbeat!: () -> void
 
         def started!: () -> void