Class: OpenC3::TextLogMicroservice

Inherits:
Microservice show all
Defined in:
lib/openc3/microservices/text_log_microservice.rb

Instance Attribute Summary

Attributes inherited from Microservice

#count, #custom, #error, #logger, #microservice_status_thread, #name, #scope, #secrets, #state

Instance Method Summary collapse

Methods inherited from Microservice

#as_json, run

Constructor Details

#initialize(name) ⇒ TextLogMicroservice

Returns a new instance of TextLogMicroservice.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/openc3/microservices/text_log_microservice.rb', line 29

def initialize(name)
  super(name)
  @config['options'].each do |option|
    case option[0].upcase
    when 'CYCLE_TIME' # Maximum time between log files
      @cycle_time = option[1].to_i
    when 'CYCLE_SIZE' # Maximum size of a log file
      @cycle_size = option[1].to_i
    else
      @logger.error("Unknown option passed to microservice #{@name}: #{option}")
    end
  end

  # These settings limit the log file to 10 minutes or 50MB of data, whichever comes first
  @cycle_time = 600 unless @cycle_time # 10 minutes
  @cycle_size = 50_000_000 unless @cycle_size # ~50 MB

  @error_count = 0
  @metric.set(name: 'text_log_total', value: @count, type: 'counter')
  @metric.set(name: 'text_error_total', value: @error_count, type: 'counter')
end

Instance Method Details

#log_data(topic, msg_id, msg_hash, redis) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/openc3/microservices/text_log_microservice.rb', line 77

def log_data(topic, msg_id, msg_hash, redis)
  msgid_seconds_from_epoch = msg_id.split('-')[0].to_i / 1000.0
  delta = Time.now.to_f - msgid_seconds_from_epoch
  @metric.set(name: 'text_log_topic_delta_seconds', value: delta, type: 'gauge', unit: 'seconds', help: 'Delta time between data written to stream and text log start')
  @tlws[topic].write(msg_hash["time"].to_i, msg_hash.as_json(allow_nan: true).to_json(allow_nan: true), topic, msg_id)
  @count += 1
rescue => err
  @error = err
  @logger.error("#{@name} error: #{err.formatted}")
  @error_count += 1
  @metric.set(name: 'text_log_error_total', value: @error_count, type: 'counter')
end

#run ⇒ Object 



51
52
53
54
55
56
57
58
59
60
61
62
63
64
 
# File 'lib/openc3/microservices/text_log_microservice.rb', line 51

def run
  setup_tlws()
  while true
    break if @cancel_thread

    Topic.read_topics(@topics) do |topic, msg_id, msg_hash, redis|
      break if @cancel_thread

      log_data(topic, msg_id, msg_hash, redis)
      @count += 1
      @metric.set(name: 'text_log_total', value: @count, type: 'counter')
    end
  end
end

#setup_tlws ⇒ Object 



66
67
68
69
70
71
72
73
74
75
 
# File 'lib/openc3/microservices/text_log_microservice.rb', line 66

def setup_tlws
  @tlws = {}
  @topics.each do |topic|
    topic_split = topic.gsub(/{|}/, '').split("__") # Remove the redis hashtag curly braces
    scope = topic_split[0]
    log_name = topic_split[1]
    remote_log_directory = "#{scope}/text_logs/#{log_name}"
    @tlws[topic] = TextLogWriter.new(remote_log_directory, true, @cycle_time, @cycle_size, nil, nil, false)
  end
end

#shutdown ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/openc3/microservices/text_log_microservice.rb', line 90

def shutdown
  # Make sure all the existing logs are properly closed down
  threads = []
  @tlws.each do |topic, tlw|
    threads.concat(tlw.shutdown)
  end
  # Wait for all the logging threads to move files to buckets
  threads.flatten.compact.each do |thread|
    thread.join
  end
  super()
end