Skip to content
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 10 additions & 11 deletions lib/statsd/instrument/aggregator.rb
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ def increment(name, value = 1, tags: [], no_prefix: false)
return
end

tags = tags_sorted(tags)
tags = tags_normalized(tags)
key = packet_key(name, tags, no_prefix, COUNT)

@aggregation_state_mutex.synchronize do
Expand Down Expand Up @@ -213,7 +213,7 @@ def aggregate_timing(name, value, tags: [], no_prefix: false, type: DISTRIBUTION
return
end

tags = tags_sorted(tags)
tags = tags_normalized(tags)
key = packet_key(name, tags, no_prefix, type, sample_rate: sample_rate)

aggregation_state = nil
Expand All @@ -236,7 +236,7 @@ def gauge(name, value, tags: [], no_prefix: false)
return
end

tags = tags_sorted(tags)
tags = tags_normalized(tags)
key = packet_key(name, tags, no_prefix, GAUGE)

@aggregation_state_mutex.synchronize do
Expand Down Expand Up @@ -304,18 +304,17 @@ def do_flush(aggregation_state)
end
end

def tags_sorted(tags)
return "" if tags.nil? || tags.empty?
def tags_normalized(tags)
return {} if tags.nil? || tags.empty?

if tags.is_a?(Hash)
tags = tags.sort_by { |k, _v| k.to_s }.map! { |k, v| "#{k}:#{v}" }
else
tags.sort!
unless tags.is_a?(Hash)
tags = tags.to_h { |tag| tag.include?(":") ? tag.split(":") : [tag, ""] }.transform_keys(&:to_sym)
end
datagram_builder(no_prefix: false).normalize_tags(tags)

tags
end

def packet_key(name, tags = "".b, no_prefix = false, type = COUNT, sample_rate: CONST_SAMPLE_RATE)
def packet_key(name, tags = {}, no_prefix = false, type = COUNT, sample_rate: CONST_SAMPLE_RATE)
AggregationKey.new(
DatagramBuilder.normalize_string(name),
tags,
Expand Down