Class: FFWD::Plugin::GoogleCloud::Hook

Inherits:
FlushingOutputHook
  • Object
show all
Includes:
Logging
Defined in:
lib/ffwd/plugin/google_cloud/hook.rb

Constant Summary collapse

HEADER_BASE = 
{
  "Content-Type" => "application/json"
}

Instance Method Summary collapse

Constructor Details

#initialize(c) ⇒ Hook

Returns a new instance of Hook.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 32

def initialize c
  @api_url = c[:api_url]
  @metadata_url = c[:metadata_url]
  @project_id = c[:project_id]
  @project = c[:project]
  @client_id = c[:client_id]
  @scope = c[:scope]
  @debug = c[:debug]

  @api_pipeline = nil
  @metadata_pipeline = nil
  @token = nil
  @expires_at = nil
  # list of blocks waiting for a token.
  @pending = []

  @write_api = "/cloudmonitoring/v2beta2/projects/#{@project_id}/timeseries:write"
  @md_api = "/cloudmonitoring/v2beta2/projects/#{@project_id}/metricDescriptors"
  @acquire = "/0.1/meta-data/service-accounts/default/acquire"
  @expire_threshold = 10

  # cache of seen metric descriptors
  @md_cache = {}
end

Instance Method Details

#active? ⇒ Boolean

Returns:

  • (Boolean) 


103
104
105
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 103

def active?
  not @api_pipeline.nil? and not @metadata_pipeline.nil?
end

#api_request ⇒ Object

avoid pipelining requests by creating a new handle for each



177
178
179
180
181
182
183
184
185
186
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 177

def api_request
  if @debug
    return DebugRequest.new(@api_url, :responses => {
      [:get, @md_api] => {"metrics" => []},
      [:post, @write_api] => {}
    })
  end

  EM::HttpRequest.new(@api_url)
end

#bind_proxy(http, p) ⇒ Object 



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 236

def bind_proxy http, p
  http.errback do
    p.err "#{http.error}"
  end

  http.callback do
    if http.response_header.status == 200
      p.call
    else
      p.err "#{http.response_header.status}: #{http.response}"
    end
  end

  p
end

#close ⇒ Object 



194
195
196
197
198
199
200
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 194

def close
  @metadata_pipeline.close if @metadata_pipeline
  @api_pipeline.close if @api_pipeline

  @metadata_pipeline = nil
  @api_pipeline = nil
end

#connect ⇒ Object 



188
189
190
191
192
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 188

def connect
  @metadata_pipeline = metadata_request
  @api_pipeline = api_request
  populate_metadata_cache
end

#metadata_request ⇒ Object 



167
168
169
170
171
172
173
174
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 167

def metadata_request
  if @debug
    return DebugRequest.new(@metadata_url, :response => {
      :accessToken => "FAKE", :expiresAt => (Time.now.to_i + 3600)})
  end

  EM::HttpRequest.new(@metadata_url)
end

#new_proxy(http) ⇒ Object

Setup a new proxy object for http request. This makes sure that the callback/errback/error api as is expected by flushing_output to be consistent.



232
233
234
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 232

def new_proxy http
  bind_proxy http, CallbackProxy.new
end

#reporter_meta ⇒ Object 



252
253
254
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 252

def reporter_meta
  {:component => :google_cloud}
end

#send(metrics) ⇒ Object 



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 202

def send metrics
  dropped, timeseries = Utils.make_timeseries(metrics)

  common_labels = Utils.make_common_labels(metrics)

  if dropped > 0
    log.warning "Dropped #{dropped} points (duplicate entries)"
  end

  request = {:commonLabels => common_labels, :timeseries => timeseries}
  metrics = JSON.dump(request)

  verify_descriptors(request) do
    with_token do |token|
      head = Hash[HEADER_BASE]
      head['Authorization'] = "Bearer #{token}"

      if log.debug?
        log.debug "Sending: #{metrics}"
      end

      new_proxy api_request.post(
        :path => @write_api, :head => head, :body => metrics)
    end
  end
end

#verify_descriptors(metrics, &block) ⇒ Object 



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
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
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 107

def verify_descriptors(metrics, &block)
  missing = []

  metrics[:timeseries].each do |ts|
    desc = ts[:timeseriesDesc]
    name = desc[:metric]
    missing << desc unless @md_cache[name]
  end

  if missing.empty?
    return block.call
  end

  all = EM::All.new

  # Example: {:metric=>"custom.cloudmonitoring.googleapis.com/lol", :labels=>{}}
  missing.each do |m|
    name = m[:metric]

    descriptor = {
     :name => m[:metric], :description => "",
     :labels => Utils.extract_labels(m[:labels]),
     :typeDescriptor => {:metricType => "gauge", :valueType => "double"}
    }

    descriptor = JSON.dump(descriptor)

    log.info "Creating descriptor for: #{name}"

    all << with_token{|token|
      head = Hash[HEADER_BASE]
      head['Authorization'] = "Bearer #{token}"
      p = new_proxy(api_request.post(
        :path => @md_api, :head => head, :body => descriptor))

      p.callback do
        log.info "Created (caching): #{name}"
        @md_cache[name] = true
      end

      p.errback do |error|
        log.error "Failed to create descriptor (#{error}): #{descriptor}"
      end

      p
    }
  end

  # TODO: call block _after_ descriptors have been verified.
  all.callback do
    block.call
  end

  all.errback do |errors|
    log.error "Failed to create descriptors: #{errors}"
  end

  SingleProxy.new all
end

#with_token(&block) ⇒ Object 



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
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/ffwd/plugin/google_cloud/hook.rb', line 57

def with_token &block
  # join a pending call
  unless @pending.empty?
    proxy = CallbackProxy.new
    @pending << [block, proxy]
    return proxy
  end

  # cached, valid token
  if @token and Time.now + @expire_threshold < @expires_at
    return block.call(@token)
  end

  current_p = CallbackProxy.new
  @pending << [block, current_p]

  log.debug "Requesting token"

  http = @metadata_pipeline.get(
    :path => @acquire,
    :query => {:client_id => @client_id, :scope => @scope})

  token_p = new_proxy(http)

  token_p.errback do
    @pending.each do |b, block_p|
      block_p.err "Token request failed: #{token_p.error}"
    end.clear
  end

  token_p.callback do
    result = JSON.load(http.response)

    @token = result['accessToken']
    @expires_at = Time.at(result['expiresAt'])

    log.debug "Got token: #{@token} (expires_at: #{@expires_at}}"

    @pending.each do |b, block_p|
      b.call(@token).into block_p
    end.clear
  end

  return current_p
end