Module: MetricSystem::Server

Extended by:
Server
Includes:
EM::P::LineProtocol
Included in:
Server
Defined in:
lib/metric_system/server.rb

Overview

require ‘eventmachine/timer’

Defined Under Namespace

Modules: Buffer Classes: Event

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.flush ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/metric_system/server.rb', line 55

def self.flush
  return unless Buffer.length > 0

  if @busy
    STDERR.puts "    Waiting for writer: backlog is #{Buffer.length} entries"
    return
  end

  @busy = true

  events = Buffer.take

  operation = proc {
    flush_events events
  }
  callback = proc {
    @busy = nil

    if shutting_down?
      flush_events(Buffer.take)
      EM.stop
    end
  }
  EventMachine.defer operation, callback
rescue
  STDERR.puts "#{$!}, from\n\t" + $!.backtrace.join("\t")
end

.flush_events(events) ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/metric_system/server.rb', line 83

def self.flush_events(events)
  return if events.empty?

  starts_at = Time.now
  MetricSystem.transaction do
    events.each do |event|
      MetricSystem.add_event event.table, event.name, event.value, event.time
    end
  end
  STDERR.puts "    Writing #{events.count} events: %.3f secs" % (Time.now - starts_at)

  starts_at = Time.now
  MetricSystem.aggregate
  STDERR.puts "    Merging #{events.count} events: %.3f secs" % (Time.now - starts_at)
end

.run(db, socket_path, options = {}) ⇒ Object

Note that this will block current thread.



100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/metric_system/server.rb', line 100

def self.run(db, socket_path, options = {})
  @options = options || {}
  MetricSystem.target = db

  STDERR.puts "Starting server at socket: #{socket_path}"

  EventMachine.run {
    EventMachine::PeriodicTimer.new(1) do
      MetricSystem::Server.flush
    end

    EventMachine.start_server socket_path, MetricSystem::Server
  }
end

.shutdown ⇒ Object 



119
120
121
122
123
124
125
 
# File 'lib/metric_system/server.rb', line 119

def self.shutdown
  return if shutting_down?
  return unless @options[:quit_server]

  @shutting_down = true
  MetricSystem::Server.flush
end

.shutting_down? ⇒ Boolean

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/metric_system/server.rb', line 115

def self.shutting_down?
  @shutting_down
end

Instance Method Details

#receive_line(line) ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/metric_system/server.rb', line 38

def receive_line(line)
  if line == "SHUTDOWN:SERVER"
    MetricSystem::Server.shutdown
    return
  end

  return if MetricSystem::Server.shutting_down?

  return unless event = Event.parse(line)

  Buffer.push event

  MetricSystem::Server.flush if Buffer.length % 1000 == 0
rescue
  STDERR.puts "#{$!}, from\n\t" + $!.backtrace.join("\t")
end