Module: OpenC3::DecomCommon

Extended by:
DecomCommon
Included in:
DecomCommon
Defined in:
lib/openc3/microservices/decom_common.rb

Overview

Shared decom pipeline used by DecomMicroservice (live telemetry) and ReingestJob (historical raw log replay). The reingest path passes check_limits: false so historical data does not re-fire limits events.

Instance Method Summary collapse

Instance Method Details

#decom_and_publish(packet, scope:, target_names:, logger:, name:, check_limits: true, stored_limits_mode: 'PROCESS', metric: nil, error_callback: nil) ⇒ Integer

Decommutate a Packet and publish it on the TelemetryDecomTopic. This is the step that lands data in the CVT and in the Python TsdbMicroservice → QuestDB.

Parameters:

  • packet (Packet) —

    A fully buffered Packet. Caller sets received_time, received_count, stored, extra, buffer.

  • scope (String) —

    Scope name.

  • target_names (Array<String>) —

    Used when a subpacket must be re-identified.

  • logger (Logger) —

    Destination for warnings.

  • name (String) —

    Identifier used in subpacket warning messages (microservice name, or "REINGEST:<job_id>").

  • check_limits (Boolean) (defaults to: true) —

    When false, skips the Packet#check_limits call. Reingest passes false so historical data does not re-fire limits events.

  • stored_limits_mode (String) (defaults to: 'PROCESS') —

    Controls limits handling for stored packets. 'PROCESS' (default) processes limits normally. 'LOG' still evaluates limits (callback handles suppressing reactions). 'DISABLE' skips limits evaluation entirely for stored packets and omits limits states from the decommutated output.

  • metric (Metric, nil) (defaults to: nil) —

    Optional; when set, records decom_duration_seconds.

  • error_callback (Proc, nil) (defaults to: nil) —

    Called as error_callback.call(exception) when Packet#process or Packet#check_limits raises. The microservice uses this to bump its decom_error_total metric.

Returns:

  • (Integer) —

    Number of (sub)packets published.



48
49
50
51
52
53
54
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
82
83
84
85
86
87
88
 
# File 'lib/openc3/microservices/decom_common.rb', line 48

def decom_and_publish(packet, scope:, target_names:, logger:, name:,
                      check_limits: true, stored_limits_mode: 'PROCESS',
                      metric: nil, error_callback: nil)
  start = Process.clock_gettime(Process::CLOCK_MONOTONIC) if metric
  published = 0

  packet_and_subpackets = packet.subpacketize
  packet_and_subpackets.each do |packet_or_subpacket|
    if packet_or_subpacket.subpacket
      packet_or_subpacket = handle_subpacket(packet, packet_or_subpacket,
                                             target_names: target_names,
                                             scope: scope,
                                             logger: logger,
                                             name: name)
    end

    begin
      packet_or_subpacket.process
    rescue Exception => e
      error_callback&.call(e)
      logger.error e.message
    end

    disable_stored_limits = packet_or_subpacket.stored && stored_limits_mode == 'DISABLE'
    if check_limits && !disable_stored_limits
      packet_or_subpacket.check_limits(System.limits_set)
    end

    TelemetryDecomTopic.write_packet(packet_or_subpacket,
                                     include_limits_states: !disable_stored_limits,
                                     scope: scope)
    published += 1
  end

  if metric
    diff = Process.clock_gettime(Process::CLOCK_MONOTONIC) - start
    metric.set(name: 'decom_duration_seconds', value: diff, type: 'gauge', unit: 'seconds')
  end

  published
end

#handle_subpacket(packet, subpacket, target_names:, scope:, logger:, name:) ⇒ Object

Identify a subpacket and (except for stored telemetry) update the CVT. Extracted from DecomMicroservice so reingest can handle subpackets too.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
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
 
# File 'lib/openc3/microservices/decom_common.rb', line 92

def handle_subpacket(packet, subpacket, target_names:, scope:, logger:, name:)
  subpacket.received_time = packet.received_time
  subpacket.stored = packet.stored
  subpacket.extra = packet.extra

  if subpacket.stored
    identified_subpacket = System.telemetry.identify_and_define_packet(subpacket, target_names, subpackets: true)
  else
    if subpacket.identified?
      begin
        identified_subpacket = System.telemetry.update!(subpacket.target_name,
                                                        subpacket.packet_name,
                                                        subpacket.buffer)
      rescue RuntimeError
        logger.warn "#{name}: Received unknown identified subpacket: #{subpacket.target_name} #{subpacket.packet_name}"
        subpacket.target_name = nil
        subpacket.packet_name = nil
        identified_subpacket = System.telemetry.identify!(subpacket.buffer,
                                                          target_names, subpackets: true)
      end
    else
      identified_subpacket = System.telemetry.identify!(subpacket.buffer,
                                                        target_names, subpackets: true)
    end
  end

  if identified_subpacket
    identified_subpacket.received_time = subpacket.received_time
    identified_subpacket.stored = subpacket.stored
    identified_subpacket.extra = subpacket.extra
    subpacket = identified_subpacket
  else
    unknown_subpacket = System.telemetry.update!('UNKNOWN', 'UNKNOWN', subpacket.buffer)
    unknown_subpacket.received_time = subpacket.received_time
    unknown_subpacket.stored = subpacket.stored
    unknown_subpacket.extra = subpacket.extra
    subpacket = unknown_subpacket
    num_bytes_to_print = [InterfaceMicroservice::UNKNOWN_BYTES_TO_PRINT, subpacket.length].min
    data = subpacket.buffer(false)[0..(num_bytes_to_print - 1)]
    prefix = data.each_byte.map { |byte| sprintf("%02X", byte) }.join()
    logger.warn "#{name} #{subpacket.target_name} packet length: #{subpacket.length} starting with: #{prefix}"
  end

  TargetModel.sync_tlm_packet_counts(subpacket, target_names, scope: scope)
  subpacket
end