Class: OpenC3::ReactionWorker

Inherits:
Object
  • Object
show all
Defined in:
lib/openc3/microservices/reaction_microservice.rb

Overview

The Reaction worker is a very simple thread pool worker. Once the manager queues a trigger to evaluate against the reactions. The worker will check the reactions to see if it needs to fire any reactions.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name:, logger:, scope:, share:, ident:) ⇒ ReactionWorker

Returns a new instance of ReactionWorker.



224
225
226
227
228
229
230
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 224

def initialize(name:, logger:, scope:, share:, ident:)
  @name = name
  @logger = logger
  @scope = scope
  @share = share
  @ident = ident
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



222
223
224
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 222

def name
  @name
end

#scope ⇒ Object (readonly)

Returns the value of attribute scope.



222
223
224
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 222

def scope
  @scope
end

#share ⇒ Object (readonly)

Returns the value of attribute share.



222
223
224
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 222

def share
  @share
end

Instance Method Details

#get_token(username) ⇒ Object 



232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 232

def get_token(username)
  if ENV['OPENC3_API_CLIENT'].nil?
    return OpenC3Authentication.new().token
  else
    # Check for offline access token
    model = nil
    model = OpenC3::OfflineAccessModel.get_model(name: username, scope: @scope) if username and username != ''
    if model and model.offline_access_token
      auth = OpenC3KeycloakAuthentication.new(ENV['OPENC3_KEYCLOAK_URL'])
      return auth.get_token_from_refresh_token(model.offline_access_token)
    else
      return nil
    end
  end
end

#process_true_trigger(data:) ⇒ Object 



271
272
273
274
275
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 271

def process_true_trigger(data:)
  @share.reaction_base.get_reactions(trigger_name: data['name']).each do |reaction|
    run_reaction(reaction: reaction)
  end
end

#reaction(data:) ⇒ Object 



248
249
250
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 248

def reaction(data:)
  return ReactionModel.from_json(data, name: data['name'], scope: data['scope'])
end

#run ⇒ Object 



252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 252

def run
  @logger.info "ReactionWorker-#{@ident} running"
  loop do
    begin
      kind, data = @share.queue_base.queue.pop
      break if kind.nil? || data.nil?
      case kind
      when 'reaction'
        run_reaction(reaction: reaction(data: data))
      when 'trigger'
        process_true_trigger(data: data)
      end
    rescue StandardError => e
      @logger.error "ReactionWorker-#{@ident} failed to evaluate kind: #{kind} data: #{data}\n#{e.formatted}"
    end
  end
  @logger.info "ReactionWorker-#{@ident} exiting"
end

#run_action(reaction:, action:) ⇒ Object 



284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 284

def run_action(reaction:, action:)
  reaction.updated_at = Time.now.to_nsec_from_epoch
  reaction_json = reaction.as_json(:allow_nan => true)
  # Let the frontend know which action is being run
  # because we can combine commands and scripts with notifications
  reaction_json['action'] = action['type']
  notification = {
    'kind' => 'run',
    'type' => 'reaction',
    'data' => JSON.generate(reaction_json),
  }
  AutonomicTopic.write_notification(notification, scope: @scope)

  case action['type']
  when 'notify'
    run_notify(reaction: reaction, action: action)
  when 'command'
    run_command(reaction: reaction, action: action)
  when 'script'
    run_script(reaction: reaction, action: action)
  end
end

#run_command(reaction:, action:) ⇒ Object 



319
320
321
322
323
324
325
326
327
328
329
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 319

def run_command(reaction:, action:)
  begin
    username = reaction.username
    token = get_token(username)
    raise "No token available for username: #{username}" unless token
    cmd_no_hazardous_check(action['value'], scope: @scope, token: token)
    @logger.info "ReactionWorker-#{@ident} #{reaction.name} command action complete, command: #{action['value']}"
  rescue StandardError => e
    @logger.error "ReactionWorker-#{@ident} #{reaction.name} command action failed, #{action}\n#{e.message}"
  end
end

#run_notify(reaction:, action:) ⇒ Object 



307
308
309
310
311
312
313
314
315
316
317
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 307

def run_notify(reaction:, action:)
  notification = NotificationModel.new(
    time: Time.now.to_nsec_from_epoch,
    severity: action['severity'],
    url: "/tools/autonomic/reactions",
    title: "#{reaction.name} run",
    body: action['value']
  )
  NotificationsTopic.write_notification(notification.as_json(:allow_nan => true), scope: @scope)
  @logger.info "ReactionWorker-#{@ident} #{reaction.name} notify action complete, body: #{action['value']}, severity: #{action['severity']}"
end

#run_reaction(reaction:) ⇒ Object 



277
278
279
280
281
282
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 277

def run_reaction(reaction:)
  reaction.actions.each do |action|
    run_action(reaction: reaction, action: action)
  end
  @share.reaction_base.sleep(name: reaction.name)
end

#run_script(reaction:, action:) ⇒ Object 



331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
 
# File 'lib/openc3/microservices/reaction_microservice.rb', line 331

def run_script(reaction:, action:)
  begin
    username = reaction.username
    token = get_token(username)
    raise "No token available for username: #{username}" unless token
    request = Net::HTTP::Post.new(
      "/script-api/scripts/#{action['value']}/run?scope=#{@scope}",
      'Content-Type' => 'application/json',
      'Authorization' => token
    )
    request.body = JSON.generate({
      'scope' => @scope,
      'environment' => action['environment'],
      'reaction' => reaction.name,
      'id' => Time.now.to_i
    })
    hostname = ENV['OPENC3_SCRIPT_HOSTNAME'] || 'openc3-cosmos-script-runner-api'
    response = Net::HTTP.new(hostname, 2902).request(request)
    raise "failed to call #{hostname}, for script: #{action['value']}, response code: #{response.code}" if response.code != '200'

    @logger.info "ReactionWorker-#{@ident} #{reaction.name} script action complete, #{action['value']} => #{response.body}"
  rescue StandardError => e
    @logger.error "ReactionWorker-#{@ident} #{reaction.name} script action failed, #{action}\n#{e.message}"
  end
end