From c2c42d3d95543014ee75d1767050607db35dfde2 Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Wed, 7 Oct 2026 09:49:09 -0600 Subject: [PATCH 1/6] feat(interfaces): queue redis writes by default at 0.1s interval - Interface microservices (Ruby and Python) now always queue CVT and stream writes, using DEFAULT_UPDATE_INTERVAL of 0.1 seconds - UPDATE_INTERVAL 0 (or less) disables queuing to write immediately - Document the new default and the disable option in interfaces.md Co-Authored-By: Claude Opus 5.5 --- .../docs/configuration/interfaces.md | 2 +- .../microservices/interface_microservice.rb | 13 ++++-- .../microservices/interface_microservice.py | 12 +++-- .../test_interface_microservice.py | 45 +++++++++++++++++++ .../interface_microservice_spec.rb | 26 +++++++++++ 5 files changed, 89 insertions(+), 9 deletions(-) diff --git a/docs.openc3.com/docs/configuration/interfaces.md b/docs.openc3.com/docs/configuration/interfaces.md index 5b01d29b0a..a1d11b1407 100644 --- a/docs.openc3.com/docs/configuration/interfaces.md +++ b/docs.openc3.com/docs/configuration/interfaces.md @@ -131,7 +131,7 @@ INTERFACE INTERFACE_NAME tcpip_client_interface.rb host.docker.internal 8080 808 #### UPDATE_INTERVAL -Number of seconds to wait before writing packets to Redis. By default packets are written to Redis every second (both the Current Value Table (CVT) and the stream). If you have high rate data and you want faster updates you can lower this value. Note that this will increase processor utilization! This option was previously known as `OPTIMIZE_THROUGHPUT`. +Number of seconds to wait before writing packets to Redis. Interfaces queue their Redis writes (both the Current Value Table (CVT) and the stream) and flush them in a single pipelined batch every 0.1 seconds by default. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables queuing entirely and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`. diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 8d5e337d8c..723eda07ea 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -555,6 +555,7 @@ def run class InterfaceMicroservice < Microservice UNKNOWN_BYTES_TO_PRINT = 16 + DEFAULT_UPDATE_INTERVAL = 0.1 # seconds def initialize(name) @mutex = Mutex.new @@ -591,19 +592,23 @@ def initialize(name) RouterStatusModel.set(@interface.as_json(), scope: @scope) end - @queued = false + # 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| # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting if option_name.upcase == 'UPDATE_INTERVAL' or option_name.upcase == 'OPTIMIZE_THROUGHPUT' - @queued = true update_interval = option_values[0].to_f - EphemeralStoreQueued.instance.set_update_interval(update_interval) - StoreQueued.instance.set_update_interval(update_interval) end if option_name.upcase == 'SYNC_PACKET_COUNT_DELAY_SECONDS' TargetModel.sync_packet_count_delay_seconds = option_values[0].to_f end end + # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis + @queued = update_interval > 0.0 + if @queued + EphemeralStoreQueued.instance.set_update_interval(update_interval) + StoreQueued.instance.set_update_interval(update_interval) + end @interface_thread_sleeper = Sleeper.new @cancel_thread = false diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index eb0fd384a3..120fc6f05a 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -641,6 +641,7 @@ def run(self): class InterfaceMicroservice(Microservice): UNKNOWN_BYTES_TO_PRINT = 16 DISCONNECT_WAIT_TIME = 1 + DEFAULT_UPDATE_INTERVAL = 0.1 # seconds def __init__(self, name): self.mutex = threading.Lock() @@ -680,16 +681,19 @@ def __init__(self, name): else: RouterStatusModel.set(self.interface.as_json(), scope=self.scope) - self.queued = False + # 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(): # OPTIMIZE_THROUGHPUT was changed to UPDATE_INTERVAL to better represent the setting if option_name.upper() == "UPDATE_INTERVAL" or option_name.upper() == "OPTIMIZE_THROUGHPUT": - self.queued = True update_interval = float(option_values[0]) - EphemeralStoreQueued.instance().set_update_interval(update_interval) - StoreQueued.instance().set_update_interval(update_interval) if option_name.upper() == "SYNC_PACKET_COUNT_DELAY_SECONDS": TargetModel.sync_packet_count_delay_seconds = float(option_values[0]) + # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis + self.queued = update_interval > 0.0 + if self.queued: + EphemeralStoreQueued.instance().set_update_interval(update_interval) + StoreQueued.instance().set_update_interval(update_interval) if self.interface_or_router == "INTERFACE": self.handler_thread = InterfaceCmdHandlerThread( diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index 85255cb2bc..a5d4dbe0be 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -868,3 +868,48 @@ def test_supports_optimize_throughput_option_for_backward_compatibility(self): im.shutdown() time.sleep(0.1) # Allow threads to exit + + def test_queues_writes_by_default_with_default_update_interval(self): + model = InterfaceModel( + name="INST_INT", + scope="DEFAULT", + target_names=["INST"], + cmd_target_names=["INST"], + tlm_target_names=["INST"], + config_params=["test_interface.py"], + options=[], + ) + model.update() + + # Set a different interval to verify the default gets applied + StoreQueued.instance().set_update_interval(1) + EphemeralStoreQueued.instance().set_update_interval(1) + + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + self.assertEqual(im.queued, True) + self.assertEqual(StoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) + self.assertEqual(EphemeralStoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) + self.assertEqual(InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL, 0.1) + + im.shutdown() + time.sleep(0.1) # Allow threads to exit + + def test_update_interval_zero_disables_queued_writes(self): + model = InterfaceModel( + name="INST_INT", + scope="DEFAULT", + target_names=["INST"], + cmd_target_names=["INST"], + tlm_target_names=["INST"], + config_params=["test_interface.py"], + options=[["UPDATE_INTERVAL", "0"]], + ) + model.update() + + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + self.assertEqual(im.queued, False) + + im.shutdown() + time.sleep(0.1) # Allow threads to exit diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 48b5ed2760..8b198958cf 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -176,6 +176,32 @@ class ApiTest sleep 0.1 # Allow threads to exit end + it "queues writes by default with the default update interval" do + InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: []).update + # Set a different interval to verify the default gets applied + StoreQueued.instance.set_update_interval(1) + EphemeralStoreQueued.instance.set_update_interval(1) + + im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + expect(im.instance_variable_get(:@queued)).to eql true + expect(InterfaceMicroservice::DEFAULT_UPDATE_INTERVAL).to eql 0.1 + expect(StoreQueued.instance.update_interval).to eql 0.1 + expect(EphemeralStoreQueued.instance.update_interval).to eql 0.1 + + im.shutdown + sleep 0.1 # Allow threads to exit + end + + it "disables queued writes when UPDATE_INTERVAL is 0" do + InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: [["UPDATE_INTERVAL", "0"]]).update + + im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + expect(im.instance_variable_get(:@queued)).to eql false + + im.shutdown + sleep 0.1 # Allow threads to exit + end + it "preserves existing packet counts" do # Initialize the telemetry topic with a non-zero RECEIVED_COUNT System.telemetry.packets("INST").each do |_packet_name, packet| From c33c1a8ee8e3e6c05296800a70446b923d753d33 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 22:41:28 +0000 Subject: [PATCH 2/6] fix(review): apply Claude review fixes (turn 1) - openc3/python/test/microservices/test_interface_microservice.py:890 - SonarQube python:S5906: replaced assertEqual(im.queued, True) with assertTrue(im.queued) - openc3/python/test/microservices/test_interface_microservice.py:912 - SonarQube python:S5906: replaced assertEqual(im.queued, False) with assertFalse(im.queued) AI-Review-Bot: true AI-Review-Run: 38000332622 --- .../python/test/microservices/test_interface_microservice.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index a5d4dbe0be..836dc5f715 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -887,7 +887,7 @@ def test_queues_writes_by_default_with_default_update_interval(self): im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") self.addCleanup(im.shutdown) - self.assertEqual(im.queued, True) + self.assertTrue(im.queued) self.assertEqual(StoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) self.assertEqual(EphemeralStoreQueued.instance().update_interval, InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL) self.assertEqual(InterfaceMicroservice.DEFAULT_UPDATE_INTERVAL, 0.1) @@ -909,7 +909,7 @@ def test_update_interval_zero_disables_queued_writes(self): im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") self.addCleanup(im.shutdown) - self.assertEqual(im.queued, False) + self.assertFalse(im.queued) im.shutdown() time.sleep(0.1) # Allow threads to exit From e91e1c16f8eeb2acae40deb57a78de24729b05b7 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 22:43:26 +0000 Subject: [PATCH 3/6] fix(review): apply Codex review fixes (turn 2) - openc3/lib/openc3/microservices/interface_microservice.rb:609 - Nonzero telemetry shards retained a one-second queue interval; apply the configured interval to every telemetry target shard. - openc3/python/openc3/microservices/interface_microservice.py:696 - Fixed the equivalent shard interval bug and added Python and Ruby regression coverage. AI-Review-Bot: true AI-Review-Run: 38000332622 --- .../microservices/interface_microservice.rb | 7 +++++-- .../microservices/interface_microservice.py | 9 +++++++-- .../microservices/test_interface_microservice.py | 14 ++++++++++++++ .../microservices/interface_microservice_spec.rb | 16 ++++++++++++++++ 4 files changed, 42 insertions(+), 4 deletions(-) diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 723eda07ea..3ebe28e9dd 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -606,8 +606,11 @@ def initialize(name) # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis @queued = update_interval > 0.0 if @queued - EphemeralStoreQueued.instance.set_update_interval(update_interval) - StoreQueued.instance.set_update_interval(update_interval) + db_shards = [0] + @interface.tlm_target_names.map { |target_name| Store.db_shard_for_target(target_name, scope: @scope) } + db_shards.uniq.each do |db_shard| + EphemeralStoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) + StoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) + end end @interface_thread_sleeper = Sleeper.new diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 120fc6f05a..e72e15c3dd 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -43,6 +43,7 @@ from openc3.utilities.json import JsonDecoder, JsonEncoder from openc3.utilities.logger import Logger from openc3.utilities.sleeper import Sleeper +from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.thread_manager import ThreadManager from openc3.utilities.time import from_nsec_from_epoch @@ -692,8 +693,12 @@ def __init__(self, name): # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis self.queued = update_interval > 0.0 if self.queued: - EphemeralStoreQueued.instance().set_update_interval(update_interval) - StoreQueued.instance().set_update_interval(update_interval) + db_shards = {0} + for target_name in self.interface.tlm_target_names: + db_shards.add(Store.db_shard_for_target(target_name, scope=self.scope)) + for db_shard in db_shards: + EphemeralStoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) + StoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) if self.interface_or_router == "INTERFACE": self.handler_thread = InterfaceCmdHandlerThread( diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index 836dc5f715..7eb919c6df 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -28,6 +28,7 @@ from openc3.topics.interface_topic import InterfaceTopic from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic from openc3.topics.topic import Topic +from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.time import from_nsec_from_epoch from test.test_helper import * @@ -913,3 +914,16 @@ def test_update_interval_zero_disables_queued_writes(self): im.shutdown() time.sleep(0.1) # Allow threads to exit + + def test_applies_default_update_interval_to_target_shards(self): + with ( + patch.object(Store, "db_shard_for_target", return_value=1), + patch.object(EphemeralStoreQueued, "instance") as ephemeral_instance, + patch.object(StoreQueued, "instance") as store_instance, + ): + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + for instance in (ephemeral_instance, store_instance): + instance.assert_any_call(db_shard=0) + instance.assert_any_call(db_shard=1) + instance.return_value.set_update_interval.assert_called_with(0.1) diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 8b198958cf..15aff06fb0 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -192,6 +192,22 @@ class ApiTest sleep 0.1 # Allow threads to exit end + it "applies the default update interval to target shards" do + InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: []).update + allow(Store).to receive(:db_shard_for_target).and_return(1) + [EphemeralStoreQueued, StoreQueued].each do |store_class| + [0, 1].each do |db_shard| + queued_store = double("queued store") + allow(store_class).to receive(:instance).with(db_shard: db_shard).and_return(queued_store) + expect(queued_store).to receive(:set_update_interval).with(0.1) + end + end + + im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + im.shutdown + sleep 0.1 # Allow threads to exit + end + it "disables queued writes when UPDATE_INTERVAL is 0" do InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: [["UPDATE_INTERVAL", "0"]]).update From 9ce0cf7c1d8e861ee68a0f0d19053c1b6576ddca Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Tue, 13 Oct 2026 13:33:45 -0600 Subject: [PATCH 4/6] docs(interfaces): clarify UPDATE_INTERVAL only queues raw stream writes Decom stream and CVT updates are written by the decom microservice and are not batched by this option. Co-Authored-By: Claude Opus 5.5 --- docs.openc3.com/docs/configuration/interfaces.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs.openc3.com/docs/configuration/interfaces.md b/docs.openc3.com/docs/configuration/interfaces.md index a1d11b1407..a6ec5ba52f 100644 --- a/docs.openc3.com/docs/configuration/interfaces.md +++ b/docs.openc3.com/docs/configuration/interfaces.md @@ -131,7 +131,7 @@ INTERFACE INTERFACE_NAME tcpip_client_interface.rb host.docker.internal 8080 808 #### UPDATE_INTERVAL -Number of seconds to wait before writing packets to Redis. Interfaces queue their Redis writes (both the Current Value Table (CVT) and the stream) and flush them in a single pipelined batch every 0.1 seconds by default. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables queuing entirely and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`. +Number of seconds to wait before writing packets to Redis. Interfaces queue the raw packet stream writes they make to Redis and flush them in a single pipelined batch every 0.1 seconds by default. Decommutated telemetry and the Current Value Table (CVT) are written separately by the decom microservice and are not affected by this option. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables queuing entirely and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`. From 637b19fe158249d7e04a9da991650c5ed28166ea Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Mon, 19 Oct 2026 10:25:11 -0600 Subject: [PATCH 5/6] fix(interfaces): apply update interval to lazily created telemetry shards - Pass the interface UPDATE_INTERVAL to TelemetryTopic.write_packet so a queued store created after startup (e.g. shard resolved after a failed lookup) uses it instead of the 1s default - Clarify that UPDATE_INTERVAL 0 only disables raw packet stream queuing - Move script shebangs above the license header so direct execution works Co-Authored-By: Claude Opus 5.5 --- .../docs/configuration/interfaces.md | 2 +- .../microservices/interface_microservice.rb | 3 +- openc3/lib/openc3/topics/telemetry_topic.rb | 5 +- .../microservices/interface_microservice.py | 5 +- .../python/openc3/topics/telemetry_topic.py | 5 +- .../test/topics/test_telemetry_topic.py | 54 +++++++++++++++++++ openc3/spec/topics/telemetry_topic_spec.rb | 54 +++++++++++++++++++ scripts/heritage/code_heritage.py | 2 +- scripts/license_audit/license_audit.py | 2 +- 9 files changed, 123 insertions(+), 9 deletions(-) create mode 100644 openc3/python/test/topics/test_telemetry_topic.py create mode 100644 openc3/spec/topics/telemetry_topic_spec.rb diff --git a/docs.openc3.com/docs/configuration/interfaces.md b/docs.openc3.com/docs/configuration/interfaces.md index a6ec5ba52f..c4552e8b5c 100644 --- a/docs.openc3.com/docs/configuration/interfaces.md +++ b/docs.openc3.com/docs/configuration/interfaces.md @@ -131,7 +131,7 @@ INTERFACE INTERFACE_NAME tcpip_client_interface.rb host.docker.internal 8080 808 #### UPDATE_INTERVAL -Number of seconds to wait before writing packets to Redis. Interfaces queue the raw packet stream writes they make to Redis and flush them in a single pipelined batch every 0.1 seconds by default. Decommutated telemetry and the Current Value Table (CVT) are written separately by the decom microservice and are not affected by this option. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables queuing entirely and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`. +Number of seconds to wait before writing packets to Redis. Interfaces queue the raw packet stream writes they make to Redis and flush them in a single pipelined batch every 0.1 seconds by default. Decommutated telemetry and the Current Value Table (CVT) are written separately by the decom microservice and are not affected by this option. Raising this value reduces processor utilization for high rate data at the cost of slower updates. Lowering it gives faster updates but increases processor utilization. Setting it to 0 disables raw packet stream queuing and writes every packet to Redis immediately, which is the slowest option for high rate data. This option was previously known as `OPTIMIZE_THROUGHPUT`. diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 47cf466263..b3721c8d50 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -612,6 +612,7 @@ def initialize(name) end end # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis + @update_interval = update_interval @queued = update_interval > 0.0 if @queued db_shards = [0] + @interface.tlm_target_names.map { |target_name| Store.db_shard_for_target(target_name, scope: @scope) } @@ -823,7 +824,7 @@ def handle_packet(packet) # Write to stream if @interface.tlm_target_enabled[packet.target_name] TargetModel.sync_tlm_packet_counts(packet, @interface.tlm_target_names, scope: @scope) - TelemetryTopic.write_packet(packet, queued: @queued, scope: @scope) + TelemetryTopic.write_packet(packet, queued: @queued, update_interval: @update_interval, scope: @scope) end end diff --git a/openc3/lib/openc3/topics/telemetry_topic.rb b/openc3/lib/openc3/topics/telemetry_topic.rb index 4773b49b39..1bac2e4709 100644 --- a/openc3/lib/openc3/topics/telemetry_topic.rb +++ b/openc3/lib/openc3/topics/telemetry_topic.rb @@ -20,7 +20,7 @@ module OpenC3 class TelemetryTopic < Topic - def self.write_packet(packet, queued: false, scope:) + def self.write_packet(packet, queued: false, update_interval: 1, scope:) msg_hash = { :time => packet.packet_time.to_nsec_from_epoch, :received_time => packet.received_time.to_nsec_from_epoch, @@ -34,7 +34,8 @@ def self.write_packet(packet, queued: false, scope:) topic = "#{scope}__TELEMETRY__{#{packet.target_name}}__#{packet.packet_name}" db_shard = Store.db_shard_for_target(packet.target_name, scope: scope) if queued - EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash) + # update_interval only applies if this creates the instance for a shard not configured at startup + EphemeralStoreQueued.instance(update_interval, db_shard: db_shard).write_topic(topic, msg_hash) else Topic.write_topic(topic, msg_hash, db_shard: db_shard) end diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 9082599adc..a6d8777616 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -696,6 +696,7 @@ def __init__(self, name): if option_name.upper() == "SYNC_PACKET_COUNT_DELAY_SECONDS": TargetModel.sync_packet_count_delay_seconds = float(option_values[0]) # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis + self.update_interval = update_interval self.queued = update_interval > 0.0 if self.queued: db_shards = {0} @@ -920,7 +921,9 @@ def handle_packet(self, packet): # Write to stream if self.interface.tlm_target_enabled.get(packet.target_name, False): TargetModel.sync_tlm_packet_counts(packet, self.interface.tlm_target_names, scope=self.scope) - TelemetryTopic.write_packet(packet, queued=self.queued, scope=self.scope) + TelemetryTopic.write_packet( + packet, queued=self.queued, update_interval=self.update_interval, scope=self.scope + ) def handle_connection_failed(self, connection, connect_error): self.error = connect_error diff --git a/openc3/python/openc3/topics/telemetry_topic.py b/openc3/python/openc3/topics/telemetry_topic.py index d30522fcab..5785c5d872 100644 --- a/openc3/python/openc3/topics/telemetry_topic.py +++ b/openc3/python/openc3/topics/telemetry_topic.py @@ -19,7 +19,7 @@ class TelemetryTopic(Topic): @classmethod - def write_packet(cls, packet, scope, queued=False): + def write_packet(cls, packet, scope, queued=False, update_interval=1): msg_hash = { "time": to_nsec_from_epoch(packet.packet_time), "received_time": to_nsec_from_epoch(packet.received_time), @@ -33,7 +33,8 @@ def write_packet(cls, packet, scope, queued=False): msg_hash["extra"] = json.dumps(packet.extra) db_shard = Store.db_shard_for_target(packet.target_name, scope=scope) if queued: - EphemeralStoreQueued.instance(db_shard=db_shard).write_topic( + # update_interval only applies if this creates the instance for a shard not configured at startup + EphemeralStoreQueued.instance(update_interval=update_interval, db_shard=db_shard).write_topic( f"{scope}__TELEMETRY__{{{packet.target_name}}}__{packet.packet_name}", msg_hash, ) 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..c42230a87f --- /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.packets.packet import Packet +from openc3.topics.telemetry_topic import TelemetryTopic +from test.test_helper import mock_redis + + +class TestTelemetryTopic(unittest.TestCase): + def setUp(self): + mock_redis(self) + shard_patch = patch("openc3.topics.telemetry_topic.Store.db_shard_for_target", return_value=3) + shard_patch.start() + self.addCleanup(shard_patch.stop) + + self.packet = Packet("TARGET", "PACKET") + self.packet.received_time = datetime.datetime.now(datetime.timezone.utc) + self.packet.packet_time = datetime.datetime.now(datetime.timezone.utc) + self.packet.received_count = 1 + + @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") + def test_creates_queued_store_for_target_shard_with_update_interval(self, instance): + instance.return_value = MagicMock() + TelemetryTopic.write_packet(self.packet, queued=True, update_interval=0.25, scope="DEFAULT") + instance.assert_called_once_with(update_interval=0.25, db_shard=3) + topic, msg_hash = instance.return_value.write_topic.call_args.args + self.assertEqual(topic, "DEFAULT__TELEMETRY__{TARGET}__PACKET") + self.assertEqual(msg_hash["target_name"], "TARGET") + + @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") + def test_defaults_queued_store_update_interval_to_1_second(self, instance): + instance.return_value = MagicMock() + TelemetryTopic.write_packet(self.packet, queued=True, scope="DEFAULT") + instance.assert_called_once_with(update_interval=1, db_shard=3) + + @patch("openc3.topics.telemetry_topic.Topic.write_topic") + @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") + def test_writes_directly_to_target_shard_when_not_queued(self, instance, write_topic): + TelemetryTopic.write_packet(self.packet, scope="DEFAULT") + instance.assert_not_called() + self.assertEqual(write_topic.call_args.args[0], "DEFAULT__TELEMETRY__{TARGET}__PACKET") + self.assertEqual(write_topic.call_args.kwargs["db_shard"], 3) diff --git a/openc3/spec/topics/telemetry_topic_spec.rb b/openc3/spec/topics/telemetry_topic_spec.rb new file mode 100644 index 0000000000..ca80890bd4 --- /dev/null +++ b/openc3/spec/topics/telemetry_topic_spec.rb @@ -0,0 +1,54 @@ +# encoding: ascii-8bit + +# 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. + +require 'spec_helper' +require 'openc3/topics/telemetry_topic' +require 'openc3/packets/packet' + +module OpenC3 + describe TelemetryTopic do + describe "self.write_packet" do + let(:packet) do + packet = Packet.new('TARGET', 'PACKET') + packet.received_time = Time.now + packet.packet_time = Time.now + packet.received_count = 1 + packet + end + + before(:each) do + mock_redis() + allow(Store).to receive(:db_shard_for_target).and_return(3) + end + + it "creates the queued store for the target shard with the given update interval" do + queued_store = double("queued store") + expect(EphemeralStoreQueued).to receive(:instance).with(0.25, db_shard: 3).and_return(queued_store) + expect(queued_store).to receive(:write_topic).with("DEFAULT__TELEMETRY__{TARGET}__PACKET", hash_including(target_name: 'TARGET')) + TelemetryTopic.write_packet(packet, queued: true, update_interval: 0.25, scope: 'DEFAULT') + end + + it "defaults the queued store update interval to 1 second" do + queued_store = double("queued store", write_topic: nil) + expect(EphemeralStoreQueued).to receive(:instance).with(1, db_shard: 3).and_return(queued_store) + TelemetryTopic.write_packet(packet, queued: true, scope: 'DEFAULT') + end + + it "writes directly to the target shard when not queued" do + expect(EphemeralStoreQueued).not_to receive(:instance) + expect(Topic).to receive(:write_topic).with("DEFAULT__TELEMETRY__{TARGET}__PACKET", hash_including(target_name: 'TARGET'), db_shard: 3) + TelemetryTopic.write_packet(packet, scope: 'DEFAULT') + end + end + end +end diff --git a/scripts/heritage/code_heritage.py b/scripts/heritage/code_heritage.py index f5068ee8bc..de9f651eca 100755 --- a/scripts/heritage/code_heritage.py +++ b/scripts/heritage/code_heritage.py @@ -1,3 +1,4 @@ +#!/usr/bin/env python3 # Copyright 2026 OpenC3, Inc. # All Rights Reserved. # @@ -9,7 +10,6 @@ # This file may also be used under the terms of a commercial license # if purchased from OpenC3, Inc. -#!/usr/bin/env python3 """ code_heritage.py -- How much of the current code base is still Ball Aerospace COSMOS 5.0.5? diff --git a/scripts/license_audit/license_audit.py b/scripts/license_audit/license_audit.py index a2977ba02c..e7917429ae 100755 --- a/scripts/license_audit/license_audit.py +++ b/scripts/license_audit/license_audit.py @@ -1,3 +1,4 @@ +#!/usr/bin/env python3 # Copyright 2026 OpenC3, Inc. # All Rights Reserved. # @@ -9,7 +10,6 @@ # This file may also be used under the terms of a commercial license # if purchased from OpenC3, Inc. -#!/usr/bin/env python3 """ license_audit.py -- Does anything OpenC3 COSMOS depends on carry a license we cannot ship under? From 9faf3b86f3c0d081c3a4d63afd3b12c5316ed326 Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Wed, 21 Oct 2026 12:48:35 -0600 Subject: [PATCH 6/6] fix(store): apply interface UPDATE_INTERVAL to every db shard Writes are routed to each target's db shard, so setting the interval on only shard 0 and the telemetry target shards left other queues, including ones created later, on the 1 second default. StoreQueued now remembers the interval and applies it to existing and future shard instances. Co-Authored-By: Claude Opus 5.5 --- .../microservices/interface_microservice.rb | 11 ++---- openc3/lib/openc3/topics/telemetry_topic.rb | 5 +-- openc3/lib/openc3/utilities/store_queued.rb | 18 +++++++-- .../microservices/interface_microservice.py | 15 ++----- .../python/openc3/topics/telemetry_topic.py | 5 +-- .../openc3/utilities/store_implementation.py | 3 ++ .../python/openc3/utilities/store_queued.py | 23 ++++++++++- .../python/openc3/utilities/store_queued.pyi | 7 +++- .../test_interface_microservice.py | 39 ++++++++++++------- openc3/python/test/test_helper.py | 2 + .../test/topics/test_telemetry_topic.py | 12 ++---- .../test/utilities/test_store_queued.py | 24 +++++++++++- .../interface_microservice_spec.rb | 21 +++++----- openc3/spec/spec_helper.rb | 2 + openc3/spec/topics/telemetry_topic_spec.rb | 10 +---- openc3/spec/utilities/store_queued_spec.rb | 24 ++++++++++++ 16 files changed, 151 insertions(+), 70 deletions(-) diff --git a/openc3/lib/openc3/microservices/interface_microservice.rb b/openc3/lib/openc3/microservices/interface_microservice.rb index 397b5471e2..ff83cb6a3a 100644 --- a/openc3/lib/openc3/microservices/interface_microservice.rb +++ b/openc3/lib/openc3/microservices/interface_microservice.rb @@ -612,14 +612,11 @@ def initialize(name) end end # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis - @update_interval = update_interval @queued = update_interval > 0.0 if @queued - db_shards = [0] + @interface.tlm_target_names.map { |target_name| Store.db_shard_for_target(target_name, scope: @scope) } - db_shards.uniq.each do |db_shard| - EphemeralStoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) - StoreQueued.instance(db_shard: db_shard).set_update_interval(update_interval) - end + # Apply to every db_shard since writes are routed to the shard of each target + EphemeralStoreQueued.set_update_interval_all_shards(update_interval) + StoreQueued.set_update_interval_all_shards(update_interval) end @interface_thread_sleeper = Sleeper.new @@ -832,7 +829,7 @@ def handle_packet(packet) # Write to stream if @interface.tlm_target_enabled[packet.target_name] TargetModel.sync_tlm_packet_counts(packet, @interface.tlm_target_names, scope: @scope) - TelemetryTopic.write_packet(packet, queued: @queued, update_interval: @update_interval, scope: @scope) + TelemetryTopic.write_packet(packet, queued: @queued, scope: @scope) end end diff --git a/openc3/lib/openc3/topics/telemetry_topic.rb b/openc3/lib/openc3/topics/telemetry_topic.rb index 1bac2e4709..4773b49b39 100644 --- a/openc3/lib/openc3/topics/telemetry_topic.rb +++ b/openc3/lib/openc3/topics/telemetry_topic.rb @@ -20,7 +20,7 @@ module OpenC3 class TelemetryTopic < Topic - def self.write_packet(packet, queued: false, update_interval: 1, scope:) + def self.write_packet(packet, queued: false, scope:) msg_hash = { :time => packet.packet_time.to_nsec_from_epoch, :received_time => packet.received_time.to_nsec_from_epoch, @@ -34,8 +34,7 @@ def self.write_packet(packet, queued: false, update_interval: 1, scope:) topic = "#{scope}__TELEMETRY__{#{packet.target_name}}__#{packet.packet_name}" db_shard = Store.db_shard_for_target(packet.target_name, scope: scope) if queued - # update_interval only applies if this creates the instance for a shard not configured at startup - EphemeralStoreQueued.instance(update_interval, db_shard: db_shard).write_topic(topic, msg_hash) + EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash) else Topic.write_topic(topic, msg_hash, db_shard: db_shard) end diff --git a/openc3/lib/openc3/utilities/store_queued.rb b/openc3/lib/openc3/utilities/store_queued.rb index 0943e36a82..90a3e19288 100644 --- a/openc3/lib/openc3/utilities/store_queued.rb +++ b/openc3/lib/openc3/utilities/store_queued.rb @@ -24,20 +24,32 @@ class StoreQueued # Mutex used to ensure that only one instance is created @@instance_mutex = Mutex.new + DEFAULT_UPDATE_INTERVAL = 1 # seconds + # Get the singleton instance for the given db_shard - # Sets the update interval to 1 second by default - def self.instance(update_interval = 1, db_shard: 0) # seconds + # Uses the interval from set_update_interval_all_shards if not given + def self.instance(update_interval = nil, db_shard: 0) @instances ||= [] the_instance = @instances[db_shard] return the_instance if the_instance @@instance_mutex.synchronize do @instances ||= [] - @instances[db_shard] ||= self.new(update_interval, db_shard: db_shard) + @instances[db_shard] ||= self.new(update_interval || @update_interval || DEFAULT_UPDATE_INTERVAL, db_shard: db_shard) return @instances[db_shard] end end + # Set the update interval on every db_shard instance, including ones created later + def self.set_update_interval_all_shards(interval) + return unless interval > 0.0 + + @@instance_mutex.synchronize do + @update_interval = interval + (@instances || []).compact.each { |the_instance| the_instance.set_update_interval(interval) } + end + end + # Delegate all unknown class methods to delegate to the instance (db_shard 0) def self.method_missing(message, *args, **kwargs, &) self.instance.public_send(message, *args, **kwargs, &) diff --git a/openc3/python/openc3/microservices/interface_microservice.py b/openc3/python/openc3/microservices/interface_microservice.py index 8e805881a3..b0588e03f7 100644 --- a/openc3/python/openc3/microservices/interface_microservice.py +++ b/openc3/python/openc3/microservices/interface_microservice.py @@ -44,7 +44,6 @@ from openc3.utilities.json import JsonDecoder, JsonEncoder from openc3.utilities.logger import Logger from openc3.utilities.sleeper import Sleeper -from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.thread_manager import ThreadManager from openc3.utilities.time import from_nsec_from_epoch @@ -697,15 +696,11 @@ def __init__(self, name): if option_name.upper() == "SYNC_PACKET_COUNT_DELAY_SECONDS": TargetModel.sync_packet_count_delay_seconds = float(option_values[0]) # An UPDATE_INTERVAL of 0 (or less) disables queuing and writes directly to Redis - self.update_interval = update_interval self.queued = update_interval > 0.0 if self.queued: - db_shards = {0} - for target_name in self.interface.tlm_target_names: - db_shards.add(Store.db_shard_for_target(target_name, scope=self.scope)) - for db_shard in db_shards: - EphemeralStoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) - StoreQueued.instance(db_shard=db_shard).set_update_interval(update_interval) + # Apply to every db_shard since writes are routed to the shard of each target + EphemeralStoreQueued.set_update_interval_all_shards(update_interval) + StoreQueued.set_update_interval_all_shards(update_interval) if self.interface_or_router == "INTERFACE": self.handler_thread = InterfaceCmdHandlerThread( @@ -929,9 +924,7 @@ def handle_packet(self, packet): # Write to stream if self.interface.tlm_target_enabled.get(packet.target_name, False): TargetModel.sync_tlm_packet_counts(packet, self.interface.tlm_target_names, scope=self.scope) - TelemetryTopic.write_packet( - packet, queued=self.queued, update_interval=self.update_interval, scope=self.scope - ) + TelemetryTopic.write_packet(packet, queued=self.queued, scope=self.scope) # Equivalent of the Ruby Errno lists. Use the specific ConnectionError subclasses # rather than ConnectionError itself, which also covers BrokenPipeError (EPIPE) diff --git a/openc3/python/openc3/topics/telemetry_topic.py b/openc3/python/openc3/topics/telemetry_topic.py index 5785c5d872..d30522fcab 100644 --- a/openc3/python/openc3/topics/telemetry_topic.py +++ b/openc3/python/openc3/topics/telemetry_topic.py @@ -19,7 +19,7 @@ class TelemetryTopic(Topic): @classmethod - def write_packet(cls, packet, scope, queued=False, update_interval=1): + 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), @@ -33,8 +33,7 @@ def write_packet(cls, packet, scope, queued=False, update_interval=1): msg_hash["extra"] = json.dumps(packet.extra) db_shard = Store.db_shard_for_target(packet.target_name, scope=scope) if queued: - # update_interval only applies if this creates the instance for a shard not configured at startup - EphemeralStoreQueued.instance(update_interval=update_interval, db_shard=db_shard).write_topic( + EphemeralStoreQueued.instance(db_shard=db_shard).write_topic( f"{scope}__TELEMETRY__{{{packet.target_name}}}__{packet.packet_name}", msg_hash, ) diff --git a/openc3/python/openc3/utilities/store_implementation.py b/openc3/python/openc3/utilities/store_implementation.py index 9b8e7d80e5..6cb0488625 100644 --- a/openc3/python/openc3/utilities/store_implementation.py +++ b/openc3/python/openc3/utilities/store_implementation.py @@ -65,6 +65,9 @@ class StoreMeta(type): "instance", "instance_mutex", "my_instances", + "set_update_interval_all_shards", + "all_shards_update_interval", + "DEFAULT_UPDATE_INTERVAL", "db_shard_for_target", "_db_shard_cache", "_db_shard_cache_lock", diff --git a/openc3/python/openc3/utilities/store_queued.py b/openc3/python/openc3/utilities/store_queued.py index acd24d4ecb..dbcd6ed25f 100644 --- a/openc3/python/openc3/utilities/store_queued.py +++ b/openc3/python/openc3/utilities/store_queued.py @@ -40,18 +40,36 @@ class StoreQueued(metaclass=StoreMeta): # Mutex used to ensure that only one instance is created instance_mutex = threading.Lock() + DEFAULT_UPDATE_INTERVAL = 1 # seconds + + # Interval from set_update_interval_all_shards, used for instances created later + all_shards_update_interval = None + # Get the singleton instance for a given db_shard + # Uses the interval from set_update_interval_all_shards if not given @classmethod - def instance(cls, update_interval=1, db_shard=0): + def instance(cls, update_interval=None, db_shard=0): inst = cls.my_instances.get(db_shard) if inst: return inst with cls.instance_mutex: if db_shard not in cls.my_instances: + if update_interval is None: + update_interval = cls.all_shards_update_interval or cls.DEFAULT_UPDATE_INTERVAL cls.my_instances[db_shard] = cls(update_interval, db_shard=db_shard) return cls.my_instances[db_shard] + # Set the update interval on every db_shard instance, including ones created later + @classmethod + def set_update_interval_all_shards(cls, interval): + if interval <= 0.0: + return + with cls.instance_mutex: + cls.all_shards_update_interval = interval + for inst in cls.my_instances.values(): + inst.set_update_interval(interval) + def __init__(self, update_interval, db_shard=0): self.is_shutdown = False self.update_interval = update_interval @@ -143,5 +161,8 @@ class EphemeralStoreQueued(StoreQueued): # Variable that holds the singleton instances per db_shard my_instances = {} + # Interval from set_update_interval_all_shards, used for instances created later + all_shards_update_interval = None + def store_instance(self): return EphemeralStore.instance(db_shard=self.db_shard) diff --git a/openc3/python/openc3/utilities/store_queued.pyi b/openc3/python/openc3/utilities/store_queued.pyi index ae1a7953f1..d80e7c2a8d 100644 --- a/openc3/python/openc3/utilities/store_queued.pyi +++ b/openc3/python/openc3/utilities/store_queued.pyi @@ -31,9 +31,13 @@ def kill_thread(owner, thread, timeout: float = 1.0) -> None: ... class StoreQueued(metaclass=StoreMeta): my_instances: Incomplete instance_mutex: Incomplete + DEFAULT_UPDATE_INTERVAL: int + all_shards_update_interval: Incomplete @classmethod - def instance(cls, update_interval: int = 1, db_shard: int = 0): ... + def instance(cls, update_interval=None, db_shard: int = 0): ... + @classmethod + def set_update_interval_all_shards(cls, interval) -> None: ... is_shutdown: bool update_interval: Incomplete db_shard: Incomplete @@ -60,6 +64,7 @@ class StoreQueued(metaclass=StoreMeta): class EphemeralStoreQueued(StoreQueued): my_instances: Incomplete + all_shards_update_interval: Incomplete @classmethod def store_instance(cls): ... diff --git a/openc3/python/test/microservices/test_interface_microservice.py b/openc3/python/test/microservices/test_interface_microservice.py index 7b5da3603a..d5304ddfa4 100644 --- a/openc3/python/test/microservices/test_interface_microservice.py +++ b/openc3/python/test/microservices/test_interface_microservice.py @@ -31,7 +31,6 @@ from openc3.topics.interface_topic import InterfaceTopic from openc3.topics.telemetry_decom_topic import TelemetryDecomTopic from openc3.topics.topic import Topic -from openc3.utilities.store import Store from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued from openc3.utilities.time import from_nsec_from_epoch from test.test_helper import * @@ -1125,18 +1124,32 @@ def test_update_interval_zero_disables_queued_writes(self): im.shutdown() time.sleep(0.1) # Allow threads to exit - def test_applies_default_update_interval_to_target_shards(self): - with ( - patch.object(Store, "db_shard_for_target", return_value=1), - patch.object(EphemeralStoreQueued, "instance") as ephemeral_instance, - patch.object(StoreQueued, "instance") as store_instance, - ): - im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") - self.addCleanup(im.shutdown) - for instance in (ephemeral_instance, store_instance): - instance.assert_any_call(db_shard=0) - instance.assert_any_call(db_shard=1) - instance.return_value.set_update_interval.assert_called_with(0.1) + def test_applies_update_interval_to_every_db_shard(self): + model = InterfaceModel( + name="INST_INT", + scope="DEFAULT", + target_names=["INST"], + cmd_target_names=["INST"], + tlm_target_names=["INST"], + config_params=["test_interface.py"], + options=[["UPDATE_INTERVAL", "0.25"]], + ) + model.update() + existing_store = StoreQueued.instance(db_shard=2) + existing_ephemeral = EphemeralStoreQueued.instance(db_shard=2) + + im = InterfaceMicroservice("DEFAULT__INTERFACE__INST_INT") + self.addCleanup(im.shutdown) + # Instances that already existed + self.assertEqual(existing_store.update_interval, 0.25) + self.assertEqual(existing_ephemeral.update_interval, 0.25) + # Instances first used after startup, e.g. a target on another shard + self.assertEqual(StoreQueued.instance(db_shard=1).update_interval, 0.25) + self.assertEqual(EphemeralStoreQueued.instance(db_shard=1).update_interval, 0.25) + + im.shutdown() + time.sleep(0.1) # Allow threads to exit + class TestInterfaceCmdHandlerRoutedCommands(unittest.TestCase): def setUp(self): diff --git a/openc3/python/test/test_helper.py b/openc3/python/test/test_helper.py index ca39943e28..55d7b389eb 100644 --- a/openc3/python/test/test_helper.py +++ b/openc3/python/test/test_helper.py @@ -168,6 +168,8 @@ def mock_redis(self): Store._db_shard_cache = {} EphemeralStoreQueued.my_instances = {} StoreQueued.my_instances = {} + EphemeralStoreQueued.all_shards_update_interval = None + StoreQueued.all_shards_update_interval = None return redis diff --git a/openc3/python/test/topics/test_telemetry_topic.py b/openc3/python/test/topics/test_telemetry_topic.py index c42230a87f..fd9a54f82b 100644 --- a/openc3/python/test/topics/test_telemetry_topic.py +++ b/openc3/python/test/topics/test_telemetry_topic.py @@ -31,20 +31,14 @@ def setUp(self): self.packet.received_count = 1 @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") - def test_creates_queued_store_for_target_shard_with_update_interval(self, instance): + def test_writes_to_queued_store_for_target_shard(self, instance): instance.return_value = MagicMock() - TelemetryTopic.write_packet(self.packet, queued=True, update_interval=0.25, scope="DEFAULT") - instance.assert_called_once_with(update_interval=0.25, db_shard=3) + TelemetryTopic.write_packet(self.packet, queued=True, scope="DEFAULT") + instance.assert_called_once_with(db_shard=3) topic, msg_hash = instance.return_value.write_topic.call_args.args self.assertEqual(topic, "DEFAULT__TELEMETRY__{TARGET}__PACKET") self.assertEqual(msg_hash["target_name"], "TARGET") - @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") - def test_defaults_queued_store_update_interval_to_1_second(self, instance): - instance.return_value = MagicMock() - TelemetryTopic.write_packet(self.packet, queued=True, scope="DEFAULT") - instance.assert_called_once_with(update_interval=1, db_shard=3) - @patch("openc3.topics.telemetry_topic.Topic.write_topic") @patch("openc3.topics.telemetry_topic.EphemeralStoreQueued.instance") def test_writes_directly_to_target_shard_when_not_queued(self, instance, write_topic): diff --git a/openc3/python/test/utilities/test_store_queued.py b/openc3/python/test/utilities/test_store_queued.py index cef8c232df..9facebb1c8 100644 --- a/openc3/python/test/utilities/test_store_queued.py +++ b/openc3/python/test/utilities/test_store_queued.py @@ -15,7 +15,8 @@ import unittest from unittest.mock import MagicMock -import test.test_helper # noqa: F401 +from openc3.utilities.store_queued import EphemeralStoreQueued, StoreQueued +from test.test_helper import mock_redis def load_real_store_queued(): @@ -70,5 +71,26 @@ def test_shutdown_drains_before_writing_through(self): self.assertTrue(self.store_queued.is_shutdown) +class TestStoreQueuedUpdateIntervalAllShards(unittest.TestCase): + def setUp(self): + mock_redis(self) + + def test_updates_existing_instances_and_later_ones(self): + existing = StoreQueued.instance(db_shard=0) + self.assertEqual(existing.update_interval, 1) + StoreQueued.set_update_interval_all_shards(0.25) + self.assertEqual(existing.update_interval, 0.25) + self.assertEqual(StoreQueued.instance(db_shard=1).update_interval, 0.25) + # Separate setting per class + self.assertEqual(EphemeralStoreQueued.instance(db_shard=1).update_interval, 1) + # An explicit interval still wins for a new instance + self.assertEqual(StoreQueued.instance(update_interval=0.5, db_shard=2).update_interval, 0.5) + + def test_ignores_intervals_of_zero_or_less(self): + StoreQueued.set_update_interval_all_shards(0.25) + StoreQueued.set_update_interval_all_shards(0) + self.assertEqual(StoreQueued.instance().update_interval, 0.25) + + if __name__ == "__main__": unittest.main() diff --git a/openc3/spec/microservices/interface_microservice_spec.rb b/openc3/spec/microservices/interface_microservice_spec.rb index 2745bdc219..899b704a76 100644 --- a/openc3/spec/microservices/interface_microservice_spec.rb +++ b/openc3/spec/microservices/interface_microservice_spec.rb @@ -192,18 +192,19 @@ class ApiTest sleep 0.1 # Allow threads to exit end - it "applies the default update interval to target shards" do - InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: []).update - allow(Store).to receive(:db_shard_for_target).and_return(1) - [EphemeralStoreQueued, StoreQueued].each do |store_class| - [0, 1].each do |db_shard| - queued_store = double("queued store") - allow(store_class).to receive(:instance).with(db_shard: db_shard).and_return(queued_store) - expect(queued_store).to receive(:set_update_interval).with(0.1) - end - end + it "applies UPDATE_INTERVAL to every db_shard" do + InterfaceModel.new(name: "INST_INT", scope: "DEFAULT", target_names: ["INST"], cmd_target_names: ["INST"], tlm_target_names: ["INST"], config_params: ["TestInterface"], options: [["UPDATE_INTERVAL", "0.25"]]).update + existing_store = StoreQueued.instance(db_shard: 2) + existing_ephemeral = EphemeralStoreQueued.instance(db_shard: 2) im = InterfaceMicroservice.new("DEFAULT__INTERFACE__INST_INT") + # Instances that already existed + expect(existing_store.update_interval).to eql 0.25 + expect(existing_ephemeral.update_interval).to eql 0.25 + # Instances first used after startup, e.g. a target on another shard + expect(StoreQueued.instance(db_shard: 1).update_interval).to eql 0.25 + expect(EphemeralStoreQueued.instance(db_shard: 1).update_interval).to eql 0.25 + im.shutdown sleep 0.1 # Allow threads to exit end diff --git a/openc3/spec/spec_helper.rb b/openc3/spec/spec_helper.rb index 2b99f8f985..e407c69d9c 100644 --- a/openc3/spec/spec_helper.rb +++ b/openc3/spec/spec_helper.rb @@ -247,6 +247,8 @@ def mock_redis OpenC3::EphemeralStore.instance_variable_set(:@instances, []) OpenC3::StoreQueued.instance_variable_set(:@instances, []) OpenC3::EphemeralStoreQueued.instance_variable_set(:@instances, []) + OpenC3::StoreQueued.instance_variable_set(:@update_interval, nil) + OpenC3::EphemeralStoreQueued.instance_variable_set(:@update_interval, nil) require 'openc3/models/auth_model' OpenC3::AuthModel.set($openc3_password, nil) $openc3_token = OpenC3::AuthModel.generate_session() diff --git a/openc3/spec/topics/telemetry_topic_spec.rb b/openc3/spec/topics/telemetry_topic_spec.rb index ca80890bd4..a3d202cddb 100644 --- a/openc3/spec/topics/telemetry_topic_spec.rb +++ b/openc3/spec/topics/telemetry_topic_spec.rb @@ -31,16 +31,10 @@ module OpenC3 allow(Store).to receive(:db_shard_for_target).and_return(3) end - it "creates the queued store for the target shard with the given update interval" do + it "writes to the queued store for the target shard" do queued_store = double("queued store") - expect(EphemeralStoreQueued).to receive(:instance).with(0.25, db_shard: 3).and_return(queued_store) + expect(EphemeralStoreQueued).to receive(:instance).with(db_shard: 3).and_return(queued_store) expect(queued_store).to receive(:write_topic).with("DEFAULT__TELEMETRY__{TARGET}__PACKET", hash_including(target_name: 'TARGET')) - TelemetryTopic.write_packet(packet, queued: true, update_interval: 0.25, scope: 'DEFAULT') - end - - it "defaults the queued store update interval to 1 second" do - queued_store = double("queued store", write_topic: nil) - expect(EphemeralStoreQueued).to receive(:instance).with(1, db_shard: 3).and_return(queued_store) TelemetryTopic.write_packet(packet, queued: true, scope: 'DEFAULT') end diff --git a/openc3/spec/utilities/store_queued_spec.rb b/openc3/spec/utilities/store_queued_spec.rb index 9ee0adb84b..2858e7101e 100644 --- a/openc3/spec/utilities/store_queued_spec.rb +++ b/openc3/spec/utilities/store_queued_spec.rb @@ -75,4 +75,28 @@ module OpenC3 expect(store_queued.instance_variable_get(:@shutdown)).to be true end end + + describe StoreQueued, ".set_update_interval_all_shards" do + before(:each) do + mock_redis() + end + + it "updates existing instances and is used by instances created later" do + existing = StoreQueued.instance(db_shard: 0) + expect(existing.update_interval).to eql 1 + StoreQueued.set_update_interval_all_shards(0.25) + expect(existing.update_interval).to eql 0.25 + expect(StoreQueued.instance(db_shard: 1).update_interval).to eql 0.25 + # Separate setting per class + expect(EphemeralStoreQueued.instance(db_shard: 1).update_interval).to eql 1 + # An explicit interval still wins for a new instance + expect(StoreQueued.instance(0.5, db_shard: 2).update_interval).to eql 0.5 + end + + it "ignores intervals of 0 or less" do + StoreQueued.set_update_interval_all_shards(0.25) + StoreQueued.set_update_interval_all_shards(0) + expect(StoreQueued.instance.update_interval).to eql 0.25 + end + end end