From 0ee711c9c8458b221604f698f748da6a654f0865 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Sat, 3 Oct 2026 21:09:39 -0600 Subject: [PATCH 1/5] fix(microservices): validate pre-identified telemetry and keep stored data out of CVT A protocol-identified packet for an unmapped target, or a stored packet with an unknown identity, is now warned about and identified by content. Stored packets that cannot be identified no longer update the UNKNOWN CVT entry. Python logs the unknown hex prefix in uppercase and writes stored as "true"/"false" like Ruby. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../microservices/interface_microservice.rb | 50 ++++++++++----- .../microservices/interface_microservice.py | 62 ++++++++++++------- .../python/openc3/topics/telemetry_topic.py | 2 +- .../test_interface_microservice.py | 60 ++++++++++++++++++ .../test/topics/test_telemetry_topic.py | 54 ++++++++++++++++ .../interface_microservice_spec.rb | 56 +++++++++++++++++ 6 files changed, 242 insertions(+), 42 deletions(-) create mode 100644 openc3/python/test/topics/test_telemetry_topic.py diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 8d5e337d8c..3b987a3463 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -758,26 +758,24 @@ 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? and !known_tlm_packet?(packet) + # Packet identified but we don't know about it or it isn't mapped to this interface + # 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 + 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, @@ -791,13 +789,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() @@ -811,6 +817,16 @@ def handle_packet(packet) end end + # @return [Boolean] Whether a pre-identified packet names a defined packet + # in one of this interface's telemetry mapped targets + def known_tlm_packet?(packet) + return false unless @interface.tlm_target_names.include?(packet.target_name) + 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 eb0fd384a3..2c955f0ae6 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -851,26 +851,23 @@ def handle_packet(self, packet): if packet.received_time is None: packet.received_time = datetime.now(timezone.utc) + if packet.identified() and not self._known_tlm_packet(packet): + # Packet identified but we don't know about it or it isn't mapped to this interface + # 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( @@ -883,22 +880,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}" ) @@ -908,6 +911,17 @@ 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 in one of this + # interface's telemetry mapped targets + def _known_tlm_packet(self, packet): + if packet.target_name not in self.interface.tlm_target_names: + return False + try: + System.telemetry.packet(packet.target_name, packet.packet_name) + return True + except Exception: + return False + def handle_connection_failed(self, connection, connect_error): self.error = connect_error self.logger.error(f"{self.interface.name}: Connection {connection} failed due to {repr(connect_error)}") 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 85255cb2bc..49c7baf696 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -24,6 +24,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 @@ -843,6 +844,65 @@ 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_falls_back_for_a_target_not_mapped_to_the_interface(self): + im, warnings, written = self.setup_handle_packet() + self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE") + self.assertIn("INST_INT: Received unknown identified telemetry: SYSTEM LIMITS_CHANGE", warnings) + self.assertEqual(len(written), 1) + self.assertEqual(written[0].target_name, "INST") + self.assertEqual(written[0].packet_name, "HEALTH_STATUS") + + 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 new file mode 100644 index 0000000000..387dc73cbc --- /dev/null +++ b/openc3/python/test/topics/test_telemetry_topic.py @@ -0,0 +1,54 @@ +# Copyright 2026 OpenC3, Inc. +# All Rights Reserved. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. +# See LICENSE.md for more details. +# +# This file may also be used under the terms of a commercial license +# if purchased from OpenC3, Inc. + +import datetime +import unittest +from unittest.mock import MagicMock, patch + +from openc3.topics.telemetry_topic import TelemetryTopic +from test.test_helper import mock_redis + + +class TestTelemetryTopic(unittest.TestCase): + def setUp(self): + mock_redis(self) + self.captured = {} + + def fake_write_topic(topic, msg_hash, db_shard=0): + self.captured["topic"] = topic + self.captured["msg_hash"] = msg_hash + + topic_patch = patch("openc3.topics.telemetry_topic.Topic.write_topic", side_effect=fake_write_topic) + topic_patch.start() + self.addCleanup(topic_patch.stop) + + shard_patch = patch("openc3.topics.telemetry_topic.Store.db_shard_for_target", return_value=0) + shard_patch.start() + self.addCleanup(shard_patch.stop) + + def _make_packet(self, stored): + packet = MagicMock() + packet.target_name = "TARGET" + packet.packet_name = "PACKET" + packet.packet_time = datetime.datetime.now() + packet.received_time = datetime.datetime.now() + packet.received_count = 1 + packet.stored = stored + packet.buffer_no_copy.return_value = b"\x01\x02" + packet.extra = None + return packet + + def test_writes_stored_as_lowercase_like_ruby(self): + TelemetryTopic.write_packet(self._make_packet(True), scope="DEFAULT") + self.assertEqual(self.captured["topic"], "DEFAULT__TELEMETRY__{TARGET}__PACKET") + self.assertEqual(self.captured["msg_hash"]["stored"], "true") + TelemetryTopic.write_packet(self._make_packet(False), scope="DEFAULT") + self.assertEqual(self.captured["msg_hash"]["stored"], "false") diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 48b5ed2760..e1f3d5a451 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -247,6 +247,62 @@ 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 "falls back to identifying a packet pre-identified with a target not mapped to the interface" do + handle(@im, health_status_buffer, target_name: 'SYSTEM', packet_name: 'LIMITS_CHANGE') + expect(@warnings).to include("INST_INT: Received unknown identified telemetry: SYSTEM LIMITS_CHANGE") + expect(@written.length).to eql 1 + expect(@written[0].target_name).to eql 'INST' + expect(@written[0].packet_name).to eql 'HEALTH_STATUS' + 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 From e9ac1095625d55b44a4904b7cbb5cb9bb879c479 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Fri, 9 Oct 2026 19:03:46 -0600 Subject: [PATCH 2/5] fix(microservices): guard against unset interface in tlm identification Resolves ty unresolved-attribute errors on self.interface, which is typed as possibly None, in the new pre-identified telemetry check. Co-Authored-By: Claude Opus 5.5 (1M context) --- openc3/python/openc3/microservices/interface_microservice.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 2c955f0ae6..4002e214cc 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -851,7 +851,7 @@ def handle_packet(self, packet): if packet.received_time is None: packet.received_time = datetime.now(timezone.utc) - if packet.identified() and not self._known_tlm_packet(packet): + if self.interface is not None and packet.identified() and not self._known_tlm_packet(packet): # Packet identified but we don't know about it or it isn't mapped to this interface # Clear packet_name and target_name and try to identify self.logger.warn( @@ -914,7 +914,7 @@ def handle_packet(self, packet): # Whether a pre-identified packet names a defined packet in one of this # interface's telemetry mapped targets def _known_tlm_packet(self, packet): - if packet.target_name not in self.interface.tlm_target_names: + if self.interface is None or packet.target_name not in self.interface.tlm_target_names: return False try: System.telemetry.packet(packet.target_name, packet.packet_name) From 4988cb41cb4d255215bb9b76699b6449d1745866 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Fri, 9 Oct 2026 19:44:59 -0600 Subject: [PATCH 3/5] fix(microservices): resolve ty errors in python interface microservice Type the microservice interface as non-optional, require scope on the handler threads (matching Ruby), and align the shutdown override with Microservice.shutdown. Also fixes router reconnect with new params writing to an undefined interface_model and logging the bound connection_string method instead of its value. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../microservices/interface_microservice.py | 30 ++++++++----------- 1 file changed, 13 insertions(+), 17 deletions(-) diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 4002e214cc..f3b7a3b401 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -23,7 +23,7 @@ from datetime import datetime, timezone from openc3.config.config_parser import ConfigParser -from openc3.interfaces.interface import WriteRejectError +from openc3.interfaces.interface import Interface, WriteRejectError from openc3.microservices.interface_decom_common import handle_inject_tlm from openc3.microservices.microservice import Microservice from openc3.models.cvt_model import CvtModel @@ -50,11 +50,11 @@ with contextlib.suppress(ModuleNotFoundError): # Should never actually be used in COSMOS Core - from openc3enterprise.models.critical_cmd_model import CriticalCmdModel + from openc3enterprise.models.critical_cmd_model import CriticalCmdModel # ty: ignore[unresolved-import] class InterfaceCmdHandlerThread: - def __init__(self, interface, tlm, logger=None, metric=None, db_shard=0, scope=None): + def __init__(self, interface, tlm, logger=None, metric=None, db_shard=0, *, scope): self.interface = interface self.tlm = tlm self.scope = scope @@ -64,9 +64,7 @@ def __init__(self, interface, tlm, logger=None, metric=None, db_shard=0, scope=N self.critical_commanding = scope_model.critical_commanding else: self.critical_commanding = "OFF" - self.logger = logger - if not self.logger: - self.logger = Logger() + self.logger = logger or Logger() self.metric = metric self.count = 0 self.directive_count = 0 @@ -438,14 +436,12 @@ def process_cmd(self, topic, msg_id, msg_hash, _redis): class RouterTlmHandlerThread: - def __init__(self, router, tlm, logger=None, metric=None, db_shard=0, scope=None): + def __init__(self, router, tlm, logger=None, metric=None, db_shard=0, *, scope): self.router = router self.tlm = tlm self.scope = scope self.db_shard = int(db_shard or 0) - self.logger = logger - if not self.logger: - self.logger = Logger + self.logger = logger or Logger() self.metric = metric self.count = 0 self.directive_count = 0 @@ -644,7 +640,6 @@ class InterfaceMicroservice(Microservice): def __init__(self, name): self.mutex = threading.Lock() - self.interface = None self.interface_or_router = None self.interface_thread_sleeper = Sleeper() self.cancel_thread = False @@ -661,6 +656,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: @@ -754,7 +750,7 @@ def attempt_connection(self, *params): else: router_model = RouterModel.get(name=self.interface.name, scope=self.scope) # config_params[0] is the filename so set the rest - interface_model["config_params"][1:] = list(params) + router_model["config_params"][1:] = list(params) RouterModel.set(router_model, scope=self.scope) self.interface.state = "ATTEMPTING" @@ -766,7 +762,7 @@ def attempt_connection(self, *params): # Need to rescue Exception so we cover LoadError except RuntimeError: 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") @@ -851,7 +847,7 @@ def handle_packet(self, packet): if packet.received_time is None: packet.received_time = datetime.now(timezone.utc) - if self.interface is not None and packet.identified() and not self._known_tlm_packet(packet): + if packet.identified() and not self._known_tlm_packet(packet): # Packet identified but we don't know about it or it isn't mapped to this interface # Clear packet_name and target_name and try to identify self.logger.warn( @@ -914,7 +910,7 @@ def handle_packet(self, packet): # Whether a pre-identified packet names a defined packet in one of this # interface's telemetry mapped targets def _known_tlm_packet(self, packet): - if self.interface is None or packet.target_name not in self.interface.tlm_target_names: + if packet.target_name not in self.interface.tlm_target_names: return False try: System.telemetry.packet(packet.target_name, packet.packet_name) @@ -1054,7 +1050,7 @@ def stop(self): if valid_interface: valid_interface.destroy() - def shutdown(self, sig=None): + def shutdown(self, state="STOPPED"): if self.shutdown_complete: return # Nothing more to do name = self.name @@ -1065,7 +1061,7 @@ def shutdown(self, sig=None): if self.interface is not None and self.interface.stream_log_pair is not None: # In python shutdown does the join and cleanup self.interface.stream_log_pair.shutdown() - super().shutdown() + super().shutdown(state) def graceful_kill(self): pass # Just to avoid warning From 247a1694d2b601496c55560917e037753aa9bad0 Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Thu, 15 Oct 2026 15:49:35 -0600 Subject: [PATCH 4/5] fix(microservices): drop pre-identified tlm for unmapped targets - Keep the header identity and drop packets whose target is not in tlm_target_names, instead of re-identifying them by content, which could publish them under a mapped packet's name - Warn once per target/packet via @unmapped_tlm_warned so replaying multi-target logs does not flood the log - Document the mapping requirement in the Preidentified Protocol docs Co-Authored-By: Claude Opus 5.5 --- .../docs/configuration/protocols.md | 2 ++ .../microservices/interface_microservice.rb | 25 +++++++++++----- .../microservices/interface_microservice.py | 30 +++++++++++-------- .../test_interface_microservice.py | 21 +++++++++---- .../interface_microservice_spec.rb | 13 ++++---- 5 files changed, 59 insertions(+), 32 deletions(-) diff --git a/docs.openc3.com/docs/configuration/protocols.md b/docs.openc3.com/docs/configuration/protocols.md index 212befe5c3..29f1fd9575 100644 --- a/docs.openc3.com/docs/configuration/protocols.md +++ b/docs.openc3.com/docs/configuration/protocols.md @@ -433,6 +433,8 @@ For a full example, please see [openc3-cosmos-ccsds-protocols](https://repos.ope 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 3b987a3463..30146a7daf 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -592,6 +592,7 @@ def initialize(name) end @queued = false + @unmapped_tlm_warned = Set.new @interface.options.each do |option_name, option_values| # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting if option_name.upcase == 'UPDATE_INTERVAL' or option_name.upcase == 'OPTIMIZE_THROUGHPUT' @@ -758,12 +759,22 @@ 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? and !known_tlm_packet?(packet) - # Packet identified but we don't know about it or it isn't mapped to this interface - # 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 + 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}" + if @unmapped_tlm_warned.add?(key) + @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 @@ -818,9 +829,7 @@ def handle_packet(packet) end # @return [Boolean] Whether a pre-identified packet names a defined packet - # in one of this interface's telemetry mapped targets def known_tlm_packet?(packet) - return false unless @interface.tlm_target_names.include?(packet.target_name) System.telemetry.packet(packet.target_name, packet.packet_name) true rescue RuntimeError diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index f3b7a3b401..6e92c6d5fd 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -677,6 +677,7 @@ def __init__(self, name): RouterStatusModel.set(self.interface.as_json(), scope=self.scope) self.queued = False + self.unmapped_tlm_warned = set() for option_name, option_values in self.interface.options.items(): # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting if option_name.upper() == "UPDATE_INTERVAL" or option_name.upper() == "OPTIMIZE_THROUGHPUT": @@ -847,14 +848,22 @@ def handle_packet(self, packet): if packet.received_time is None: packet.received_time = datetime.now(timezone.utc) - if packet.identified() and not self._known_tlm_packet(packet): - # Packet identified but we don't know about it or it isn't mapped to this interface - # 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.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: + self.unmapped_tlm_warned.add(key) + 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 @@ -907,11 +916,8 @@ 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 in one of this - # interface's telemetry mapped targets + # Whether a pre-identified packet names a defined packet def _known_tlm_packet(self, packet): - if packet.target_name not in self.interface.tlm_target_names: - return False try: System.telemetry.packet(packet.target_name, packet.packet_name) return True diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index 49c7baf696..4e5cd3efec 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -871,13 +871,22 @@ def handle(self, im, buffer, target_name=None, packet_name=None, stored=False): packet.stored = stored im.handle_packet(packet) - def test_handle_packet_falls_back_for_a_target_not_mapped_to_the_interface(self): + def test_handle_packet_drops_a_target_not_mapped_to_the_interface_and_warns_once(self): im, warnings, written = self.setup_handle_packet() - self.handle(im, self.health_status_buffer(), "SYSTEM", "LIMITS_CHANGE") - self.assertIn("INST_INT: Received unknown identified telemetry: SYSTEM LIMITS_CHANGE", warnings) - self.assertEqual(len(written), 1) - self.assertEqual(written[0].target_name, "INST") - self.assertEqual(written[0].packet_name, "HEALTH_STATUS") + 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_falls_back_for_a_stored_packet_with_an_unknown_packet(self): im, warnings, written = self.setup_handle_packet() diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index e1f3d5a451..322c5d895e 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -275,12 +275,13 @@ def handle(im, buffer, target_name: nil, packet_name: nil, stored: false) sleep 0.1 # Allow threads to exit end - it "falls back to identifying a packet pre-identified with a target not mapped to the interface" do - handle(@im, health_status_buffer, target_name: 'SYSTEM', packet_name: 'LIMITS_CHANGE') - expect(@warnings).to include("INST_INT: Received unknown identified telemetry: SYSTEM LIMITS_CHANGE") - expect(@written.length).to eql 1 - expect(@written[0].target_name).to eql 'INST' - expect(@written[0].packet_name).to eql 'HEALTH_STATUS' + 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 "falls back to identifying a stored packet pre-identified with an unknown packet" do From de4dd7b69b1661a8daf77cbeeceb2dbe22dd40d9 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Sun, 25 Oct 2026 17:08:36 -0600 Subject: [PATCH 5/5] fix(microservices): bound unmapped telemetry warning tracking Unmapped target and packet names come from the wire, so the set used to warn once per name could grow without limit. Cap it at 1000 entries and evict the oldest so every newly seen name still warns. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../openc3/microservices/interface_microservice.rb | 11 +++++++++-- .../openc3/microservices/interface_microservice.py | 11 +++++++++-- .../test/microservices/test_interface_microservice.py | 9 +++++++++ .../spec/microservices/interface_microservice_spec.rb | 8 ++++++++ 4 files changed, 35 insertions(+), 4 deletions(-) diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index c189d5b110..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,7 +602,8 @@ def initialize(name) RouterStatusModel.set(@interface.as_json(), scope: @scope) end - @unmapped_tlm_warned = Set.new + # 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| @@ -799,7 +802,11 @@ def handle_packet(packet) 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}" - if @unmapped_tlm_warned.add?(key) + 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 diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 2284bf4964..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): @@ -702,7 +704,8 @@ def __init__(self, name): else: RouterStatusModel.set(self.interface.as_json(), scope=self.scope) - self.unmapped_tlm_warned = set() + # 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(): @@ -912,7 +915,11 @@ def handle_packet(self, packet): # 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: - self.unmapped_tlm_warned.add(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. + 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): diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index bd833e2ad2..42bd1b54b8 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -1098,6 +1098,15 @@ def test_handle_packet_drops_a_target_not_mapped_to_the_interface_and_warns_once ) 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) diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 12c918f32d..53f3a69cea 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -327,6 +327,14 @@ def handle(im, buffer, target_name: nil, packet_name: nil, stored: false) 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")