diff --git a/docs.openc3.com/docs/configuration/protocols.md b/docs.openc3.com/docs/configuration/protocols.md index 6053c9a2c0..5d98d9ff31 100644 --- a/docs.openc3.com/docs/configuration/protocols.md +++ b/docs.openc3.com/docs/configuration/protocols.md @@ -323,6 +323,8 @@ For a full example, please see [openc3-cosmos-gems-interface](https://repos.open The Preidentified Protocol delineates packets using the COSMOS header. This Protocol was created to allow tools to connect and receive the entire packet stream. It can be used with the [FileInterface](/docs/configuration/interfaces#file-interface) to process COSMOS 4 log files. It can also be used to chain COSMOS instances together although that should rarely be needed with the new web native implementation. +Every target named in the incoming packet headers must be mapped to the interface with [MAP_TARGET](plugins.md#map_target) or [MAP_TLM_TARGET](plugins.md#map_tlm_target). Packets for unmapped targets are dropped, and the interface logs a warning the first time it sees each unmapped packet. + | Parameter | Description | Required | Default | | ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------- | ----------------------- | | Sync Pattern | Hex string representing a byte pattern that will be searched for in the raw data. This pattern represents a packet delimiter and all data found AFTER the sync pattern will be returned. The sync pattern itself is discarded. | No | nil (no sync pattern) | diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 3964b2d064..2f51b6d555 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -559,6 +559,8 @@ def run class InterfaceMicroservice < Microservice UNKNOWN_BYTES_TO_PRINT = 16 DEFAULT_UPDATE_INTERVAL = 0.1 # seconds + # Most distinct unmapped target / packet names to remember warning about + UNMAPPED_TLM_WARNED_MAX = 1000 # Interfaces never use descriptions so don't keep them in memory def load_descriptions? @@ -600,6 +602,8 @@ def initialize(name) RouterStatusModel.set(@interface.as_json(), scope: @scope) end + # Insertion ordered so the oldest key can be evicted (see handle_packet) + @unmapped_tlm_warned = {} # Queue Redis writes by default and flush them every update_interval seconds update_interval = DEFAULT_UPDATE_INTERVAL @interface.options.each do |option_name, option_values| @@ -794,26 +798,38 @@ def handle_packet(packet) InterfaceStatusModel.set(@interface.as_json(), queued: true, scope: @scope) unless @cancel_thread packet.received_time = Time.now.sys unless packet.received_time + if packet.identified? + unless @interface.tlm_target_names.include?(packet.target_name) + # Trust the identity but drop it because the target isn't mapped to this interface + key = "#{packet.target_name} #{packet.packet_name}" + unless @unmapped_tlm_warned.key?(key) + # The names come from the wire so bound how many are remembered. + # Evict the oldest rather than stop adding so every new name still warns. + @unmapped_tlm_warned.delete(@unmapped_tlm_warned.first[0]) if @unmapped_tlm_warned.length >= UNMAPPED_TLM_WARNED_MAX + @unmapped_tlm_warned[key] = true + @logger.warn "#{@interface.name}: Dropping telemetry for unmapped target: #{key}" + end + return + end + unless known_tlm_packet?(packet) + # Packet identified but we don't know about it + # Clear packet_name and target_name and try to identify + @logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}" + packet.target_name = nil + packet.packet_name = nil + end + end + if packet.stored # Stored telemetry does not update the current value table identified_packet = System.telemetry.identify_and_define_packet(packet, @interface.tlm_target_names) else # Identify and update packet if packet.identified? - begin - # Preidentifed packet - place it into the current value table - identified_packet = System.telemetry.update!(packet.target_name, - packet.packet_name, - packet.buffer) - rescue RuntimeError - # Packet identified but we don't know about it - # Clear packet_name and target_name and try to identify - @logger.warn "#{@interface.name}: Received unknown identified telemetry: #{packet.target_name} #{packet.packet_name}" - packet.target_name = nil - packet.packet_name = nil - identified_packet = System.telemetry.identify!(packet.buffer, - @interface.tlm_target_names) - end + # Preidentifed packet - place it into the current value table + identified_packet = System.telemetry.update!(packet.target_name, + packet.packet_name, + packet.buffer) else # Packet needs to be identified identified_packet = System.telemetry.identify!(packet.buffer, @@ -827,13 +843,21 @@ def handle_packet(packet) identified_packet.extra = packet.extra packet = identified_packet else - unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer) + if packet.stored + # Stored telemetry does not update the current value table + unknown_packet = System.telemetry.packet('UNKNOWN', 'UNKNOWN').clone + unknown_packet.buffer = packet.buffer + else + unknown_packet = System.telemetry.update!('UNKNOWN', 'UNKNOWN', packet.buffer) + end unknown_packet.received_time = packet.received_time unknown_packet.stored = packet.stored unknown_packet.extra = packet.extra packet = unknown_packet - json_hash = CvtModel.build_json_from_packet(packet) - CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, queued: @queued, scope: @scope) + unless packet.stored + json_hash = CvtModel.build_json_from_packet(packet) + CvtModel.set(json_hash, target_name: packet.target_name, packet_name: packet.packet_name, queued: @queued, scope: @scope) + end num_bytes_to_print = [UNKNOWN_BYTES_TO_PRINT, packet.length].min data = packet.buffer(false)[0..(num_bytes_to_print - 1)] prefix = data.each_byte.map { | byte | sprintf("%02X", byte) }.join() @@ -847,6 +871,14 @@ def handle_packet(packet) end end + # @return [Boolean] Whether a pre-identified packet names a defined packet + def known_tlm_packet?(packet) + System.telemetry.packet(packet.target_name, packet.packet_name) + true + rescue RuntimeError + false + end + def handle_connection_failed(connection, connect_error) @error = connect_error @logger.error "#{@interface.name}: Connection #{connection} failed due to #{connect_error.formatted(false, false)}" diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index c77160b0fc..759cecda8b 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -657,6 +657,8 @@ class InterfaceMicroservice(Microservice): UNKNOWN_BYTES_TO_PRINT = 16 DISCONNECT_WAIT_TIME = 1 DEFAULT_UPDATE_INTERVAL = 0.1 # seconds + # Most distinct unmapped target / packet names to remember warning about + UNMAPPED_TLM_WARNED_MAX = 1000 # Interfaces never use descriptions so don't keep them in memory def load_descriptions(self): @@ -682,6 +684,7 @@ def __init__(self, name): self.scope = name.split("__")[0] interface_name = name.split("__")[2] + self.interface: Interface if self.interface_or_router == "INTERFACE": self.interface = InterfaceModel.get_model(name=interface_name, scope=self.scope).build() else: @@ -701,6 +704,8 @@ def __init__(self, name): else: RouterStatusModel.set(self.interface.as_json(), scope=self.scope) + # Insertion ordered so the oldest key can be evicted (see handle_packet) + self.unmapped_tlm_warned: dict[str, bool] = {} # Queue Redis writes by default and flush them every update_interval seconds update_interval = InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL for option_name, option_values in self.interface.options.items(): @@ -798,7 +803,7 @@ def attempt_connection(self, *params): # Catch Exception so any rebuild failure (bad parameters, import errors) keeps the original interface except Exception: self.logger.error( - f"Attempting connection #{self.interface.connection_string} failed due to {traceback.format_exc()}" + f"Attempting connection #{self.interface.connection_string()} failed due to {traceback.format_exc()}" ) # if SignalException === error: # self.logger.info(f"{self.interface.name}: Closing from signal") @@ -905,26 +910,35 @@ def handle_packet(self, packet): if packet.received_time is None: packet.received_time = datetime.now(timezone.utc) + if packet.identified(): + if packet.target_name not in self.interface.tlm_target_names: + # Trust the identity but drop it because the target isn't mapped to this interface + key = f"{packet.target_name} {packet.packet_name}" + if key not in self.unmapped_tlm_warned: + # The names come from the wire so bound how many are remembered. + # Evict the oldest rather than stop adding so every new name still warns. + if len(self.unmapped_tlm_warned) >= self.UNMAPPED_TLM_WARNED_MAX: + del self.unmapped_tlm_warned[next(iter(self.unmapped_tlm_warned))] + self.unmapped_tlm_warned[key] = True + self.logger.warn(f"{self.interface.name}: Dropping telemetry for unmapped target: {key}") + return + if not self._known_tlm_packet(packet): + # Packet identified but we don't know about it + # Clear packet_name and target_name and try to identify + self.logger.warn( + f"{self.interface.name}: Received unknown identified telemetry: {packet.target_name} {packet.packet_name}" + ) + packet.target_name = None + packet.packet_name = None + if packet.stored: # Stored telemetry does not update the current value table identified_packet = System.telemetry.identify_and_define_packet(packet, self.interface.tlm_target_names) else: # Identify and update packet if packet.identified(): - try: - # Preidentifed packet - place it into the current value table - identified_packet = System.telemetry.update(packet.target_name, packet.packet_name, packet.buffer) - except Exception: - # Packet identified but we don't know about it - # Clear packet_name and target_name and try to identify - self.logger.warn( - f"{self.interface.name}: Received unknown identified telemetry: {packet.target_name} {packet.packet_name}" - ) - packet.target_name = None - packet.packet_name = None - identified_packet = System.telemetry.identify_and_set_buffer( - packet.buffer, self.interface.tlm_target_names - ) + # Preidentifed packet - place it into the current value table + identified_packet = System.telemetry.update(packet.target_name, packet.packet_name, packet.buffer) else: # Packet needs to be identified identified_packet = System.telemetry.identify_and_set_buffer( @@ -937,22 +951,28 @@ def handle_packet(self, packet): identified_packet.extra = packet.extra packet = identified_packet else: - unknown_packet = System.telemetry.update("UNKNOWN", "UNKNOWN", packet.buffer) + if packet.stored: + # Stored telemetry does not update the current value table + unknown_packet = System.telemetry.packet("UNKNOWN", "UNKNOWN").clone() + unknown_packet.buffer = packet.buffer + else: + unknown_packet = System.telemetry.update("UNKNOWN", "UNKNOWN", packet.buffer) unknown_packet.received_time = packet.received_time unknown_packet.stored = packet.stored unknown_packet.extra = packet.extra packet = unknown_packet - json_hash = CvtModel.build_json_from_packet(packet) - CvtModel.set( - json_hash, - packet.target_name, - packet.packet_name, - queued=self.queued, - scope=self.scope, - ) + if not packet.stored: + json_hash = CvtModel.build_json_from_packet(packet) + CvtModel.set( + json_hash, + packet.target_name, + packet.packet_name, + queued=self.queued, + scope=self.scope, + ) num_bytes_to_print = min(InterfaceMicroservice.UNKNOWN_BYTES_TO_PRINT, len(packet.buffer)) data = packet.buffer_no_copy()[0:(num_bytes_to_print)] - prefix = "".join([format(x, "02x") for x in data]) + prefix = "".join([format(x, "02X") for x in data]) self.logger.warn( f"{self.interface.name} {packet.target_name} packet length: {len(packet.buffer)} starting with: {prefix}" ) @@ -962,6 +982,14 @@ def handle_packet(self, packet): TargetModel.sync_tlm_packet_counts(packet, self.interface.tlm_target_names, scope=self.scope) TelemetryTopic.write_packet(packet, queued=self.queued, scope=self.scope) + # Whether a pre-identified packet names a defined packet + def _known_tlm_packet(self, packet): + try: + System.telemetry.packet(packet.target_name, packet.packet_name) + return True + except Exception: + return False + # Equivalent of the Ruby Errno lists. Use the specific ConnectionError subclasses # rather than ConnectionError itself, which also covers BrokenPipeError (EPIPE) CONNECTION_FAILED_ERRORS = (ConnectionRefusedError, ConnectionResetError, TimeoutError) diff --git a/openc3/python/openc3/topics/telemetry_topic.py b/openc3/python/openc3/topics/telemetry_topic.py index d30522fcab..8361041f23 100644 --- a/openc3/python/openc3/topics/telemetry_topic.py +++ b/openc3/python/openc3/topics/telemetry_topic.py @@ -23,7 +23,7 @@ def write_packet(cls, packet, scope, queued=False): msg_hash = { "time": to_nsec_from_epoch(packet.packet_time), "received_time": to_nsec_from_epoch(packet.received_time), - "stored": str(packet.stored), + "stored": str(packet.stored).lower(), "target_name": packet.target_name, "packet_name": packet.packet_name, "received_count": packet.received_count, diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index d5304ddfa4..42bd1b54b8 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -27,6 +27,7 @@ from openc3.models.interface_status_model import InterfaceStatusModel from openc3.models.microservice_model import MicroserviceModel from openc3.models.target_model import TargetModel +from openc3.packets.packet import Packet from openc3.system.system import System from openc3.topics.interface_topic import InterfaceTopic from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic @@ -1053,6 +1054,83 @@ def test_handle_packet_does_not_write_status_after_cancel_thread_set(self): im.shutdown() time.sleep(0.1) # Allow threads to exit + def setup_handle_packet(self): + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + warnings = [] + im.logger = Mock() + im.logger.warn.side_effect = lambda msg, *args, **kwargs: warnings.append(msg) + written = [] + patcher = patch( + "openc3.microservices.interface_microservice.TelemetryTopic.write_packet", + side_effect=lambda packet, *args, **kwargs: written.append(packet), + ) + patcher.start() + self.addCleanup(patcher.stop) + return im, warnings, written + + def health_status_buffer(self): + packet = System.telemetry.packet("INST", "HEALTH_STATUS").clone() + for item in packet.id_items: + packet.write_item(item, item.id_value, "RAW") + return packet.buffer + + def handle(self, im, buffer, target_name=None, packet_name=None, stored=False): + packet = Packet(target_name, packet_name) + packet.buffer = buffer + packet.stored = stored + im.handle_packet(packet) + + def test_handle_packet_drops_a_target_not_mapped_to_the_interface_and_warns_once(self): + im, warnings, written = self.setup_handle_packet() + with ( + patch.object(System.telemetry, "identify_and_set_buffer") as mock_identify, + patch.object(System.telemetry, "identify_and_define_packet") as mock_define, + ): + for _ in range(2): + self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE") + self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE", stored=True) + mock_identify.assert_not_called() + mock_define.assert_not_called() + self.assertEqual( + [w for w in warnings if "unmapped target" in w], + ["INST_INT: Dropping telemetry for unmapped target: SYSTEM LIMITS_CHANGE"], + ) + self.assertEqual(written, []) + + def test_handle_packet_bounds_unmapped_warnings_by_evicting_the_oldest(self): + im, warnings, _ = self.setup_handle_packet() + with patch.object(InterfaceMicroservice, "UNMAPPED_TLM_WARNED_MAX", 2): + for name in ["A", "B", "C", "A"]: + self.handle(im, self.health_status_buffer(), "SYSTEM", name) + # C evicted A so A warns again, and every new name warned + self.assertEqual([w.split(" ")[-1] for w in warnings if "unmapped target" in w], ["A", "B", "C", "A"]) + self.assertEqual(len(im.unmapped_tlm_warned), 2) + + def test_handle_packet_falls_back_for_a_stored_packet_with_an_unknown_packet(self): + im, warnings, written = self.setup_handle_packet() + self.handle(im, self.health_status_buffer(), "INST", "NOPE", stored=True) + self.assertIn("INST_INT: Received unknown identified telemetry: INST NOPE", warnings) + self.assertEqual(len(written), 1) + self.assertEqual(written[0].target_name, "INST") + self.assertEqual(written[0].packet_name, "HEALTH_STATUS") + self.assertTrue(written[0].stored) + + def test_handle_packet_publishes_an_unknown_stored_packet_without_updating_the_cvt(self): + im, warnings, written = self.setup_handle_packet() + with ( + patch.object(CvtModel, "set") as mock_set, + patch.object(System.telemetry, "update", wraps=System.telemetry.update) as mock_update, + ): + self.handle(im, b"\xab\xcd", stored=True) + mock_set.assert_not_called() + mock_update.assert_not_called() + self.assertEqual(len(written), 1) + self.assertEqual(written[0].target_name, "UNKNOWN") + self.assertEqual(written[0].packet_name, "UNKNOWN") + self.assertTrue(written[0].stored) + self.assertIn("INST_INT UNKNOWN packet length: 2 starting with: ABCD", warnings) + def test_supports_optimize_throughput_option_for_backward_compatibility(self): # Update the model to use OPTIMIZE_THROUGHPUT option (legacy name) model = InterfaceModel( diff --git a/openc3/python/test/topics/test_telemetry_topic.py b/openc3/python/test/topics/test_telemetry_topic.py index fd9a54f82b..0026914bf2 100644 --- a/openc3/python/test/topics/test_telemetry_topic.py +++ b/openc3/python/test/topics/test_telemetry_topic.py @@ -30,6 +30,17 @@ def setUp(self): self.packet.packet_time = datetime.datetime.now(datetime.timezone.utc) self.packet.received_count = 1 + @patch("openc3.topics.telemetry_topic.Topic.write_topic") + def test_writes_stored_as_lowercase_like_ruby(self, write_topic): + self.packet.stored = True + TelemetryTopic.write_packet(self.packet, scope="DEFAULT") + topic, msg_hash = write_topic.call_args.args + self.assertEqual(topic, "DEFAULT__TELEMETRY__{TARGET}__PACKET") + self.assertEqual(msg_hash["stored"], "true") + self.packet.stored = False + TelemetryTopic.write_packet(self.packet, scope="DEFAULT") + self.assertEqual(write_topic.call_args.args[1]["stored"], "false") + @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") def test_writes_to_queued_store_for_target_shard(self, instance): instance.return_value = MagicMock() diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 899b704a76..53f3a69cea 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -290,6 +290,71 @@ def handle_interface_details(interface) im.shutdown sleep 0.1 # Allow threads to exit end + + context "identification" do + def health_status_buffer + packet = System.telemetry.packet('INST', 'HEALTH_STATUS').clone + packet.id_items.each { |item| packet.write_item(item, item.id_value, :RAW) } + packet.buffer + end + + def handle(im, buffer, target_name: nil, packet_name: nil, stored: false) + packet = Packet.new(target_name, packet_name) + packet.buffer = buffer + packet.stored = stored + im.send(:handle_packet, packet) + end + + before(:each) do + @im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + @warnings = [] + allow(@im.instance_variable_get(:@logger)).to receive(:warn) { |msg| @warnings << msg } + @written = [] + allow(TelemetryTopic).to receive(:write_packet) { |packet, **_| @written << packet } + end + + after(:each) do + @im.shutdown + sleep 0.1 # Allow threads to exit + end + + it "drops a packet pre-identified with a target not mapped to the interface and warns once" do + expect(System.telemetry).to_not receive(:identify!) + expect(System.telemetry).to_not receive(:identify_and_define_packet) + 2.times { handle(@im, health_status_buffer, target_name: 'SYSTEM', packet_name: 'LIMITS_CHANGE') } + handle(@im, health_status_buffer, target_name: 'SYSTEM', packet_name: 'LIMITS_CHANGE', stored: true) + expect(@warnings.grep(/unmapped target/)).to eql ["INST_INT: Dropping telemetry for unmapped target: SYSTEM LIMITS_CHANGE"] + expect(@written).to be_empty + end + + it "bounds the unmapped warnings it remembers by evicting the oldest" do + stub_const("OpenC3::InterfaceMicroservice::UNMAPPED_TLM_WARNED_MAX", 2) + %w(A B C A).each { |name| handle(@im, health_status_buffer, target_name: 'SYSTEM', packet_name: name) } + # C evicted A so A warns again, and every new name warned + expect(@warnings.grep(/unmapped target/).map { |w| w.split(' ').last }).to eql %w(A B C A) + expect(@im.instance_variable_get(:@unmapped_tlm_warned).length).to eql 2 + end + + it "falls back to identifying a stored packet pre-identified with an unknown packet" do + handle(@im, health_status_buffer, target_name: 'INST', packet_name: 'NOPE', stored: true) + expect(@warnings).to include("INST_INT: Received unknown identified telemetry: INST NOPE") + expect(@written.length).to eql 1 + expect(@written[0].target_name).to eql 'INST' + expect(@written[0].packet_name).to eql 'HEALTH_STATUS' + expect(@written[0].stored).to be true + end + + it "publishes an unidentified stored packet as UNKNOWN without updating the CVT" do + expect(CvtModel).to_not receive(:set) + expect(System.telemetry).to_not receive(:update!) + handle(@im, "\xAB\xCD", stored: true) + expect(@written.length).to eql 1 + expect(@written[0].target_name).to eql 'UNKNOWN' + expect(@written[0].packet_name).to eql 'UNKNOWN' + expect(@written[0].stored).to be true + expect(@warnings).to include("INST_INT UNKNOWN packet length: 2 starting with: ABCD") + end + end end describe "run" do