Sha256: 116c5207e11e883e2a51ac381eb36ad3f641edbdeacd20b0fd088872c0701a5e
Contents?: true
Size: 1.75 KB
Versions: 18
Compression:
Stored size: 1.75 KB
Contents
# frozen_string_literal: true
require_relative 'periodic_sync'
module Prefab
# This class aggregates example contexts. It dedupes based on the
# concatenation of the keys of the contexts.
#
# It shouldn't send the same context more than once per hour.
class ExampleContextsAggregator
include Prefab::PeriodicSync
LOG = Prefab::InternalLogger.new(self)
attr_reader :data, :cache
ONE_HOUR = 60 * 60
def initialize(client:, max_contexts:, sync_interval:)
@client = client
@max_contexts = max_contexts
@name = 'example_contexts_aggregator'
@data = Concurrent::Array.new
@cache = Prefab::RateLimitCache.new(ONE_HOUR)
start_periodic_sync(sync_interval)
end
def record(contexts)
key = contexts.grouped_key
return unless @data.size < @max_contexts && !@cache.fresh?(key)
@cache.set(key)
@data.push(contexts)
end
private
def on_prepare_data
@cache.prune
end
def flush(to_ship, _)
pool.post do
LOG.debug "Flushing #{to_ship.size} examples"
result = post('/api/v1/telemetry', events(to_ship))
LOG.debug "Uploaded #{to_ship.size} examples: #{result.status}"
end
end
def example_contexts(to_ship)
to_ship.map do |contexts|
PrefabProto::ExampleContext.new(
timestamp: contexts.seen_at * 1000,
contextSet: contexts.slim_proto
)
end
end
def events(to_ship)
event = PrefabProto::TelemetryEvent.new(
example_contexts: PrefabProto::ExampleContexts.new(
examples: example_contexts(to_ship)
)
)
PrefabProto::TelemetryEvents.new(
instance_hash: @client.instance_hash,
events: [event]
)
end
end
end
Version data entries
18 entries across 18 versions & 1 rubygems