Class: Stellwerk::Metering

Inherits:
Object
  • Object
show all
Defined in:
lib/stellwerk/metering.rb

Overview

Execution metering and batch reporting

Tracks flow executions locally and periodically reports them to the Stellwerk server in batches for efficiency.

Thread-safe via Mutex.

Defined Under Namespace

Classes: Execution

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Metering

Returns a new instance of Metering.



15
16
17
18
19
20
21
 
# File 'lib/stellwerk/metering.rb', line 15

def initialize
  @mutex = Mutex.new
  @pending_executions = []
  @last_flush_at = Time.now
  @background_thread = nil
  @running = false
end

Instance Method Details

#background_flush_running? ⇒ Boolean

Check if background flush is running

Returns:

  • (Boolean) 


100
101
102
103
104
 
# File 'lib/stellwerk/metering.rb', line 100

def background_flush_running?
  @mutex.synchronize do
    @running && @background_thread&.alive?
  end
end

#flush ⇒ Hash?

Flush pending executions to the server

Returns:

  • (Hash, nil) —

    Server response or nil if nothing to flush



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/stellwerk/metering.rb', line 46

def flush
  executions_to_send = nil

  @mutex.synchronize do
    return nil if @pending_executions.empty?

    executions_to_send = @pending_executions.dup
    @pending_executions.clear
    @last_flush_at = Time.now
  end

  send_executions(executions_to_send)
rescue ApiError => e
  # Re-queue executions on failure
  @mutex.synchronize do
    @pending_executions = executions_to_send + @pending_executions
  end
  Stellwerk.logger.error("Failed to report executions: #{e.message}")
  nil
end

#pending_count ⇒ Integer

Get count of pending executions

Returns:

  • (Integer) 


70
71
72
73
74
 
# File 'lib/stellwerk/metering.rb', line 70

def pending_count
  @mutex.synchronize do
    @pending_executions.length
  end
end

#reset! ⇒ Object

Reset metering state



107
108
109
110
111
112
113
 
# File 'lib/stellwerk/metering.rb', line 107

def reset!
  stop_background_flush
  @mutex.synchronize do
    @pending_executions.clear
    @last_flush_at = Time.now
  end
end

#start_background_flush ⇒ Thread

Start background auto-flush thread

Returns:

  • (Thread) 


79
80
81
82
83
84
85
86
 
# File 'lib/stellwerk/metering.rb', line 79

def start_background_flush
  @mutex.synchronize do
    return @background_thread if @running

    @running = true
    @background_thread = Thread.new { background_flush_loop }
  end
end

#stop_background_flush ⇒ Object

Stop background auto-flush thread



89
90
91
92
93
94
95
 
# File 'lib/stellwerk/metering.rb', line 89

def stop_background_flush
  @mutex.synchronize do
    @running = false
  end
  @background_thread&.join(5)
  @background_thread = nil
end

#track(flow_id:, duration_ms:, status:) ⇒ Object

Track a flow execution

Parameters:

  • flow_id (String) —

    Identifier for the flow

  • duration_ms (Integer) —

    Execution duration in milliseconds

  • status (String) —

    Execution status ('success', 'error')



28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/stellwerk/metering.rb', line 28

def track(flow_id:, duration_ms:, status:)
  execution = Execution.new(
    flow_id: flow_id,
    timestamp: Time.now.utc.iso8601,
    duration_ms: duration_ms,
    status: status
  )

  @mutex.synchronize do
    @pending_executions << execution
  end

  auto_flush_if_needed
end