Class: Kilden::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/kilden/client.rb

Overview

The Kilden server-side client: bounded in-memory queue, background worker, retries with backoff, remote feature flags. Construction is the only place that raises (spec contract 2); after that every public method is exception-safe — telemetry never takes down a request.

kilden = Kilden::Client.new(ENV["KILDEN_SECRET_KEY"])
kilden.track("user_42", "order_completed", { "revenue" => 99.9 })
kilden.close # on shutdown; an at_exit hook covers the forgetful

Constant Summary collapse

DEFAULT_HOST = 
"https://ingest.kilden.io"
MAX_EVENT_BYTES = 
200
MAX_DISTINCT_ID_BYTES = 
512
BATCH_LIMIT = 
1000
CLOSE_DEADLINE = 
10

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(write_key, host: DEFAULT_HOST, flush_at: 20, flush_interval: 10, max_queue_size: 10_000, timeout: 3, transport: nil, debug: false, enabled: true, logger: nil) ⇒ Client

Returns a new instance of Client.



30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/kilden/client.rb', line 30

def initialize(write_key, host: DEFAULT_HOST, flush_at: 20, flush_interval: 10,
               max_queue_size: 10_000, timeout: 3, transport: nil, debug: false,
               enabled: true, logger: nil)
  unless write_key.is_a?(String) && !write_key.empty?
    raise ConfigurationError,
          "a write key is required — find your project's secret key (sk_...) in the Kilden panel"
  end
  if write_key.start_with?("wk_")
    raise ConfigurationError,
          "#{write_key[0, 10]}… is a public write key. Server SDKs authenticate with the secret key (sk_...): " \
          "public keys degrade events to source=client and break verified revenue. " \
          "Never ship the secret key to a browser."
  end

  @debug = debug ? true : false
  @logger = logger || Log.new(@debug ? :debug : :warn)
  @enabled = enabled ? true : false
  @flush_interval = flush_interval
  return unless @enabled

  @queue = EventQueue.new(max_size: max_queue_size, flush_at: flush_at)
  @sender = Sender.new(
    write_key: write_key,
    host: host,
    transport: transport || Transport::NetHttp.new(timeout: timeout),
    logger: @logger
  )
  @decide = Decide.new(write_key: write_key, host: host, timeout: timeout,
                       transport: transport, logger: @logger)
  @flag_cache = FlagCache.new
  @pid = Process.pid
  @worker = nil
  @lifecycle = Mutex.new
  @closed = false
  @shutdown_deadline = nil

  at_exit { close }
end

Class Method Details

.format_time(time) ⇒ Object

Frozen wire format for timestamps (spec §4.4).



170
171
172
 
# File 'lib/kilden/client.rb', line 170

def self.format_time(time)
  time.getutc.strftime("%Y-%m-%dT%H:%M:%S.%LZ")
end

Instance Method Details

#close ⇒ Object

flush with a 10-second deadline, then worker shutdown. Idempotent; events sent after close are dropped with a warning.



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/kilden/client.rb', line 137

def close
  guard do
    next unless @enabled

    @lifecycle.synchronize do
      next if @closed

      @closed = true
      @shutdown_deadline = monotonic + CLOSE_DEADLINE
      @queue.close
    end

    worker = @worker
    if worker&.alive?
      worker.join([@shutdown_deadline - monotonic, 0].max)
      if worker.alive?
        worker.kill
        abandoned = @queue.drain.size
        if abandoned.positive?
          @sender.dropped!(abandoned)
          @logger.warn("kilden: close deadline hit; dropped #{abandoned} events")
        end
      end
    else
      @queue.drain.each_slice(BATCH_LIMIT) do |batch|
        @sender.send_batch(batch, deadline: @shutdown_deadline)
      end
    end
    nil
  end
end

#dropped_count ⇒ Object

dropped_count: events lost to a full queue or exhausted retries (contract 7/8) — observable so operators can alert on it.



24
25
26
27
28
 
# File 'lib/kilden/client.rb', line 24

def dropped_count
  return 0 unless @enabled

  @queue.dropped_count + @sender.dropped_total
end

#enabled?(flag_key, distinct_id, person_properties: nil, default: false) ⇒ Boolean

Returns:

  • (Boolean) 


108
109
110
111
112
113
114
115
 
# File 'lib/kilden/client.rb', line 108

def enabled?(flag_key, distinct_id, person_properties: nil, default: false)
  guard(default) do
    state, value = fetch_flag(flag_key, distinct_id, person_properties)
    next default unless state == :ok

    value == true || value.is_a?(String)
  end
end

#feature_flag(flag_key, distinct_id, person_properties: nil, default: false) ⇒ Object 



117
118
119
120
121
122
 
# File 'lib/kilden/client.rb', line 117

def feature_flag(flag_key, distinct_id, person_properties: nil, default: false)
  guard(default) do
    state, value = fetch_flag(flag_key, distinct_id, person_properties)
    state == :ok ? value : default
  end
end

#flush ⇒ Object

Blocking: drains everything queued at this moment, retries included.



125
126
127
128
129
130
131
132
133
 
# File 'lib/kilden/client.rb', line 125

def flush
  guard do
    next unless @enabled

    check_fork
    @queue.drain.each_slice(BATCH_LIMIT) { |batch| @sender.send_batch(batch) }
    nil
  end
end

#identify(distinct_id, traits = {}, opts = {}) ⇒ Object 



79
80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/kilden/client.rb', line 79

def identify(distinct_id, traits = {}, opts = {})
  guard do
    next unless @enabled && open_for_events?

    traits = {} if traits.nil?
    unless traits.is_a?(Hash)
      @logger.warn("kilden: identify traits must be a Hash; event dropped")
      next
    end
    payload = build_event(distinct_id, "$identify", { "$set" => traits }, opts, reserved: true)
    enqueue(payload) if payload
    nil
  end
end

#track(distinct_id, event, properties = {}, opts = {}) ⇒ Object 



69
70
71
72
73
74
75
76
77
 
# File 'lib/kilden/client.rb', line 69

def track(distinct_id, event, properties = {}, opts = {})
  guard do
    next unless @enabled && open_for_events?

    payload = build_event(distinct_id, event, properties, opts)
    enqueue(payload) if payload
    nil
  end
end