From 4c903c42c56bcf32780ceb26369da570611c6942 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 16:27:56 +0300 Subject: [PATCH 01/13] feat(onboarding): paced staggered onboarding engine + lifecycle interface --- .../shared/onboarding/EntityLifecycle.java | 33 +++++ .../onboarding/StaggeredOnboardingEngine.java | 126 ++++++++++++++++++ .../StaggeredOnboardingEngineTest.java | 116 ++++++++++++++++ 3 files changed, 275 insertions(+) create mode 100644 src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java create mode 100644 src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java create mode 100644 src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java diff --git a/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java b/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java new file mode 100644 index 0000000..ddbf625 --- /dev/null +++ b/src/main/java/org/thingsboard/tools/service/shared/onboarding/EntityLifecycle.java @@ -0,0 +1,33 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.tools.service.shared.onboarding; + +/** + * One entity's onboarding step for the staggered mode: bring a single gateway/device fully online + * (connect -> optionally announce sub-devices -> subscribe -> start its own telemetry cadence). + * Implementations are mode-specific; the engine only paces the calls. + */ +public interface EntityLifecycle { + + /** Number of entities to onboard; the engine drives indices [0, entityCount()). */ + int entityCount(); + + /** + * Onboard entity {@code idx} synchronously. Must throw on failure — the engine counts it as + * failed, releases its slot, and continues (never blocks the ramp). + */ + void onboard(int idx) throws Exception; +} diff --git a/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java new file mode 100644 index 0000000..e566bb7 --- /dev/null +++ b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java @@ -0,0 +1,126 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.tools.service.shared.onboarding; + +import lombok.extern.slf4j.Slf4j; +import org.thingsboard.tools.service.gateway.EphemeralSchedule; + +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Mode-agnostic paced onboarding: schedules each entity's first onboard over a jittered window and + * runs at most {@code maxConcurrentOnboards} onboards at once, signalling ramp-complete when every + * entity has reached a terminal state (onboarded or failed). Reuses the ephemeral engine's + * tryAcquire+reschedule pacing, but drives a persistent (synchronous) onboard instead of a churn cycle. + */ +@Slf4j +public class StaggeredOnboardingEngine { + + public interface RampCompleteCallback { + void onRampComplete(int onboarded, int failed); + } + + private final EntityLifecycle lifecycle; + private final int maxConcurrentOnboards; + private final long firstJitterMillis; + private final Random rng; + + private final ScheduledExecutorService timer; + private final ExecutorService workers; + private final Semaphore permits; + + private final AtomicInteger onboarded = new AtomicInteger(); + private final AtomicInteger failed = new AtomicInteger(); + private final AtomicInteger terminal = new AtomicInteger(); + private volatile boolean running; + private volatile RampCompleteCallback onComplete; + + public StaggeredOnboardingEngine(EntityLifecycle lifecycle, int maxConcurrentOnboards, + int firstJitterSec, int schedulerThreads, long seed) { + this.lifecycle = lifecycle; + this.maxConcurrentOnboards = Math.max(1, maxConcurrentOnboards); + this.firstJitterMillis = Math.max(0, firstJitterSec) * 1000L; + this.rng = new Random(EphemeralSchedule.scheduleSeed(seed, 0)); + this.timer = Executors.newScheduledThreadPool(Math.max(1, schedulerThreads)); + this.workers = Executors.newFixedThreadPool(this.maxConcurrentOnboards); + this.permits = new Semaphore(this.maxConcurrentOnboards); + } + + public void start(RampCompleteCallback cb) { + this.onComplete = cb; + this.running = true; + int count = lifecycle.entityCount(); + log.info("Staggered onboarding starting: {} entities, maxConcurrent={}, firstJitter={}ms", + count, maxConcurrentOnboards, firstJitterMillis); + if (count <= 0) { + fireComplete(); + return; + } + for (int i = 0; i < count; i++) { + final int idx = i; + long offset = EphemeralSchedule.firstOffsetMillis(rng, firstJitterMillis); + timer.schedule(() -> onboardOne(idx), offset, TimeUnit.MILLISECONDS); + } + } + + private void onboardOne(int idx) { + if (!running) { + return; + } + if (!permits.tryAcquire()) { + // no free slot: reschedule on the timer (never block a timer thread) + timer.schedule(() -> onboardOne(idx), 1 + rng.nextInt(50), TimeUnit.MILLISECONDS); + return; + } + workers.submit(() -> { + try { + lifecycle.onboard(idx); + onboarded.incrementAndGet(); + } catch (Exception e) { + failed.incrementAndGet(); + log.warn("Onboard failed for entity {}: {}", idx, e.toString()); + } finally { + permits.release(); + if (terminal.incrementAndGet() == lifecycle.entityCount()) { + fireComplete(); + } + } + }); + } + + private void fireComplete() { + log.info("Ramp complete: {} onboarded, {} failed", onboarded.get(), failed.get()); + RampCompleteCallback cb = this.onComplete; + if (cb != null) { + cb.onRampComplete(onboarded.get(), failed.get()); + } + } + + public void stop() { + running = false; + timer.shutdownNow(); + workers.shutdownNow(); + } + + public int onboardedCount() { return onboarded.get(); } + public int failedCount() { return failed.get(); } +} diff --git a/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java new file mode 100644 index 0000000..c1e74a6 --- /dev/null +++ b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java @@ -0,0 +1,116 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.tools.service.shared.onboarding; + +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; + +class StaggeredOnboardingEngineTest { + + /** Stub lifecycle: records each onboard, tracks peak concurrency, can fail chosen indices. */ + static final class StubLifecycle implements EntityLifecycle { + final int count; + final int failEvery; // 0 = never fail + final AtomicInteger inFlight = new AtomicInteger(); + final AtomicInteger peakInFlight = new AtomicInteger(); + final AtomicInteger onboardCalls = new AtomicInteger(); + final CountDownLatch allTerminal; + + StubLifecycle(int count, int failEvery) { + this.count = count; + this.failEvery = failEvery; + this.allTerminal = new CountDownLatch(count); + } + + @Override public int entityCount() { return count; } + + @Override public void onboard(int idx) throws Exception { + int now = inFlight.incrementAndGet(); + peakInFlight.accumulateAndGet(now, Math::max); + onboardCalls.incrementAndGet(); + try { + Thread.sleep(5); // simulate work so concurrency is observable + if (failEvery > 0 && idx % failEvery == 0) { + throw new RuntimeException("stub onboard failure for " + idx); + } + } finally { + inFlight.decrementAndGet(); + } + } + } + + @Test + void onboardsEveryEntityExactlyOnceAndNeverExceedsConcurrencyCap() throws Exception { + StubLifecycle stub = new StubLifecycle(200, 0); + AtomicInteger rampOnboarded = new AtomicInteger(-1); + AtomicInteger rampFailed = new AtomicInteger(-1); + CountDownLatch complete = new CountDownLatch(1); + + StaggeredOnboardingEngine engine = + new StaggeredOnboardingEngine(stub, 10, 0, 2, 42L); + engine.start((onboarded, failed) -> { + rampOnboarded.set(onboarded); + rampFailed.set(failed); + complete.countDown(); + }); + + assertThat(complete.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(stub.onboardCalls.get()).isEqualTo(200); + assertThat(stub.peakInFlight.get()).isLessThanOrEqualTo(10); + assertThat(rampOnboarded.get()).isEqualTo(200); + assertThat(rampFailed.get()).isEqualTo(0); + engine.stop(); + } + + @Test + void countsFailuresAsTerminalSoRampStillCompletes() throws Exception { + StubLifecycle stub = new StubLifecycle(50, 10); // idx 0,10,20,30,40 fail -> 5 failures + CountDownLatch complete = new CountDownLatch(1); + AtomicInteger rampOnboarded = new AtomicInteger(); + AtomicInteger rampFailed = new AtomicInteger(); + + StaggeredOnboardingEngine engine = + new StaggeredOnboardingEngine(stub, 8, 0, 2, 1L); + engine.start((onboarded, failed) -> { + rampOnboarded.set(onboarded); + rampFailed.set(failed); + complete.countDown(); + }); + + assertThat(complete.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(rampOnboarded.get()).isEqualTo(45); + assertThat(rampFailed.get()).isEqualTo(5); + assertThat(engine.onboardedCount()).isEqualTo(45); + assertThat(engine.failedCount()).isEqualTo(5); + engine.stop(); + } + + @Test + void firesRampCompleteImmediatelyForZeroEntities() throws Exception { + StubLifecycle stub = new StubLifecycle(0, 0); + CountDownLatch complete = new CountDownLatch(1); + StaggeredOnboardingEngine engine = + new StaggeredOnboardingEngine(stub, 4, 0, 1, 0L); + engine.start((onboarded, failed) -> complete.countDown()); + assertThat(complete.await(2, TimeUnit.SECONDS)).isTrue(); + engine.stop(); + } +} From e410cf6d02d3f8e06363b991a91a141f6f250604 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 16:37:04 +0300 Subject: [PATCH 02/13] feat(onboarding): add ONBOARD_MODE / ONBOARD_MAX_CONCURRENT / ONBOARD_FIRST_JITTER_SEC config --- .../thingsboard/tools/service/shared/AbstractAPITest.java | 7 +++++++ src/main/resources/tb-ce-performance-tests.yml | 6 ++++++ 2 files changed, 13 insertions(+) diff --git a/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java b/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java index 078494f..7d53781 100644 --- a/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/shared/AbstractAPITest.java @@ -119,6 +119,13 @@ protected synchronized StatsReporter statsReporter() { @Value("${gateway.overwriteActivityTime:false}") protected boolean gatewayOverwriteActivityTime; + @Value("${onboard.mode:PHASED}") + protected String onboardMode; + @Value("${onboard.maxConcurrent:200}") + protected int onboardMaxConcurrent; + @Value("${onboard.firstJitterSec:60}") + protected int onboardFirstJitterSec; + @Autowired @Qualifier("randomTelemetryGenerator") protected MessageGenerator tsMsgGenerator; diff --git a/src/main/resources/tb-ce-performance-tests.yml b/src/main/resources/tb-ce-performance-tests.yml index 8118b85..b7393a3 100644 --- a/src/main/resources/tb-ce-performance-tests.yml +++ b/src/main/resources/tb-ce-performance-tests.yml @@ -200,6 +200,12 @@ gateway: queue: "${GATEWAY_RPC_SENDER_QUEUE:RpcCalls}" # rule-engine queue name in the URL timeoutMs: "${GATEWAY_RPC_SENDER_TIMEOUT_MS:10000}" # rule-engine call timeout; MUST equal the rule chain's hardcoded TIMEOUT_MS +# Onboarding strategy for the persistent gateway/device modes. +onboard: + mode: "${ONBOARD_MODE:PHASED}" # PHASED (default, today's warmup) | STAGGERED + maxConcurrent: "${ONBOARD_MAX_CONCURRENT:200}" # STAGGERED: max entities onboarding at once + firstJitterSec: "${ONBOARD_FIRST_JITTER_SEC:60}" # STAGGERED: each entity's first onboard spread over [0,this)s + customer: startIdx: "${CUSTOMER_START_IDX:0}" endIdx: "${CUSTOMER_END_IDX:0}" From 285ff49b9ac242179032321e63f20da5d3174f99 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 16:50:03 +0300 Subject: [PATCH 03/13] feat(onboarding): STAGGERED lifecycle for persistent gateway mode (RPC starts at ramp-complete) --- .../service/gateway/MqttGatewayAPITest.java | 293 +++++++++++++++++- 1 file changed, 283 insertions(+), 10 deletions(-) diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java index 5170104..9fdd4ed 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java @@ -17,6 +17,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; import io.netty.buffer.Unpooled; import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.util.concurrent.Future; @@ -27,14 +28,18 @@ import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.IdBased; import org.thingsboard.mqtt.MqttClient; +import org.thingsboard.mqtt.MqttConnectResult; import org.thingsboard.tools.service.gateway.rpc.GatewayRpcReceiver; import org.thingsboard.tools.service.gateway.rpc.RpcBurstSender; import org.thingsboard.tools.service.gateway.rpc.RpcLatencyStats; import org.thingsboard.tools.service.gateway.rpc.RpcMessageProcessor; import org.thingsboard.tools.service.gateway.rpc.RpcResponseTemplate; +import org.thingsboard.tools.service.msg.NodeMsg; import org.thingsboard.tools.service.mqtt.DeviceClient; import org.thingsboard.tools.service.shared.BaseMqttAPITest; import org.thingsboard.tools.service.shared.StatsBlock; +import org.thingsboard.tools.service.shared.onboarding.EntityLifecycle; +import org.thingsboard.tools.service.shared.onboarding.StaggeredOnboardingEngine; import jakarta.annotation.PostConstruct; import java.nio.charset.StandardCharsets; @@ -45,6 +50,9 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -124,6 +132,22 @@ public class MqttGatewayAPITest extends BaseMqttAPITest implements GatewayAPITes private RpcBurstSender rpcBurstSender; + // STAGGERED onboarding (onboard.mode=STAGGERED): PHASED (default) never touches any of these. + private StaggeredOnboardingEngine onboardingEngine; + // Precomputed gateway-index -> name / sub-device-names model, built once by prepareStaggeredModel(). + // Unlike PHASED's mapDevicesToGatewayClientConnections (which derives the mapping from mqttClients' + // connect order, established only after ALL gateways connect), this is index-based so onboard(idx) + // can look up its own assignment before any connection exists. + private List staggeredGatewayNames; + private Map> staggeredGatewayDeviceNames; + // Per-gateway telemetry timers started by the STAGGERED path (one per onboarded gateway); cancelled + // when the test duration elapses. + private final List> gatewayTelemetryTimers = Collections.synchronizedList(new ArrayList<>()); + + private boolean staggered() { + return "STAGGERED".equalsIgnoreCase(onboardMode); + } + @Override protected boolean isInboundHandlingEnabled() { return rpcEnabled; @@ -169,6 +193,13 @@ public void createGateways() throws Exception { @Override public void connectGateways() throws InterruptedException { + if (staggered()) { + // STAGGERED: no bulk connect here. Build the gateway/device name model only; the engine + // (driven from runApiTests()) connects + announces + subscribes each gateway on its own + // paced schedule. + prepareStaggeredModel(); + return; + } AtomicInteger totalConnectedCount = new AtomicInteger(); List pack = null; List gatewayNames; @@ -219,6 +250,36 @@ private void mapDevicesToGatewayClientConnections() { } } + /** + * STAGGERED-only: same gateway-name resolution and deviceIdx % gatewayCount assignment as + * {@link #mapDevicesToGatewayClientConnections()}, but keyed by gateway INDEX (0-based, local to + * this instance's [gatewayStartIdx, gatewayEndIdx)) instead of by connected {@link MqttClient} — + * so the assignment exists before any gateway has connected, and {@code onboard(idx)} can look up + * its own sub-device names deterministically regardless of connect order/timing. + */ + private void prepareStaggeredModel() { + List gatewayNames; + if (!gateways.isEmpty()) { + gatewayNames = gateways.stream().map(Device::getName).collect(Collectors.toList()); + } else { + gatewayNames = new ArrayList<>(); + for (int i = gatewayStartIdx; i < gatewayEndIdx; i++) { + gatewayNames.add(getToken(true, i)); + } + } + this.staggeredGatewayNames = gatewayNames; + int gatewayCount = gatewayNames.size(); + Map> byGatewayIdx = new ConcurrentHashMap<>(); + for (int i = deviceStartIdx; i < deviceEndIdx; i++) { + int deviceIdx = i - deviceStartIdx; + int gatewayIdx = deviceIdx % gatewayCount; + byGatewayIdx.computeIfAbsent(gatewayIdx, k -> Collections.synchronizedList(new ArrayList<>())) + .add(getToken(false, i)); + } + this.staggeredGatewayDeviceNames = byGatewayIdx; + log.info("STAGGERED model prepared: {} gateways, {} devices", gatewayCount, deviceEndIdx - deviceStartIdx); + } + /** Re-announce a reconnected gateway's sub-devices so the server re-routes their RPC through it. * Reuses the gateway connect topic and payload used at warm-up. */ private void reannounceDevices(MqttClient gatewayClient) { @@ -246,15 +307,82 @@ protected Future warmUpPublish(DeviceClient deviceClient) { return super.warmUpPublish(deviceClient); // RPC off: legacy QoS-0 warm-up } + @Override + public void warmUpDevices() throws InterruptedException { + if (staggered()) { + return; // STAGGERED: announcement happens inside EntityLifecycle.onboard(), driven from runApiTests() + } + super.warmUpDevices(); + } + @Override public void runApiTests() throws InterruptedException { - if (rpcSenderEnabled) { - startRpcBurstSender(); + if (!staggered()) { + if (rpcSenderEnabled) { + startRpcBurstSender(); + } + try { + super.runApiTests(deviceClients.size()); + } finally { + // Stop firing new bursts BEFORE draining so the tail can settle without fresh inbound. + if (rpcBurstSender != null) { + rpcBurstSender.stop(); + } + if (rpcEnabled && rpcReceiver != null) { + long quietMs = rpcDrainQuietSec * 1000L; + long maxMs = GatewayRpcReceiver.resolveDrainMaxMs( + rpcDrainMaxSecConfig, rpcSenderEnabled, rpcSenderTimeoutMs, rpcResponseDelayMs, rpcDrainQuietSec); + log.info("Gateway RPC drain: waiting for in-flight RPCs to settle (quietSec={}, maxSec={})...", + rpcDrainQuietSec, maxMs / 1000); + GatewayRpcReceiver.DrainResult result = rpcReceiver.drain(quietMs, maxMs, rpcRespond); + rpcReceiver.finalizeLostReplies(); // replies still buffered (client never reconnected) are lost + rpcReceiver.logPending(); // name the distinct still-unanswered RPCs for DB EXPIRED correlation + String drainLine = String.format("Gateway RPC drain complete [drained %.1fs, quiesced=%b]", + result.elapsedMs / 1000.0, result.quiesced); + if (result.quiesced) { + log.info(drainLine); + } else { + log.warn(drainLine); + } + log.info(rpcReceiver.inTotalSummary()); // RPC In [total]: publish=… (new …, redelivered …) + log.info(rpcReceiver.outTotalSummary()); // RPC Out [total]: publish=…, pubAck=…, failed=…, recovered=…, lost=… + + } + } + return; + } + runStaggeredApiTests(); + } + + /** + * STAGGERED: ramp gateways in through the engine (each onboard connects + announces + subscribes + + * schedules its own telemetry timer — see {@link #connectAnnounceSubscribeAndSchedule(int)}), start + * the RPC sender only once the ramp completes (PHASED starts it up-front, before any connection + * exists), then hold for the test duration. Reuses the same RPC drain/summary block as the PHASED + * {@code finally} above. + */ + private void runStaggeredApiTests() throws InterruptedException { + statsReporter().start(); + if (rpcEnabled) { + initRpcReceiver(); } + onboardingEngine = new StaggeredOnboardingEngine( + gatewayLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed); + onboardingEngine.start((onboarded, failed) -> { + log.info("STAGGERED gateway ramp complete: {} onboarded, {} failed — starting RPC sender", onboarded, failed); + if (rpcSenderEnabled) { + startRpcBurstSender(); // existing method, unchanged: full device list + } + }); try { - super.runApiTests(deviceClients.size()); + Thread.sleep(testDurationInSec * 1000L); } finally { - // Stop firing new bursts BEFORE draining so the tail can settle without fresh inbound. + for (ScheduledFuture timer : gatewayTelemetryTimers) { + timer.cancel(false); + } + if (onboardingEngine != null) { + onboardingEngine.stop(); + } if (rpcBurstSender != null) { rpcBurstSender.stop(); } @@ -265,8 +393,8 @@ public void runApiTests() throws InterruptedException { log.info("Gateway RPC drain: waiting for in-flight RPCs to settle (quietSec={}, maxSec={})...", rpcDrainQuietSec, maxMs / 1000); GatewayRpcReceiver.DrainResult result = rpcReceiver.drain(quietMs, maxMs, rpcRespond); - rpcReceiver.finalizeLostReplies(); // replies still buffered (client never reconnected) are lost - rpcReceiver.logPending(); // name the distinct still-unanswered RPCs for DB EXPIRED correlation + rpcReceiver.finalizeLostReplies(); + rpcReceiver.logPending(); String drainLine = String.format("Gateway RPC drain complete [drained %.1fs, quiesced=%b]", result.elapsedMs / 1000.0, result.quiesced); if (result.quiesced) { @@ -274,11 +402,141 @@ public void runApiTests() throws InterruptedException { } else { log.warn(drainLine); } - log.info(rpcReceiver.inTotalSummary()); // RPC In [total]: publish=… (new …, redelivered …) - log.info(rpcReceiver.outTotalSummary()); // RPC Out [total]: publish=…, pubAck=…, failed=…, recovered=…, lost=… + log.info(rpcReceiver.inTotalSummary()); + log.info(rpcReceiver.outTotalSummary()); + } + } + } + + private EntityLifecycle gatewayLifecycle() { + return new EntityLifecycle() { + @Override + public int entityCount() { + return gatewayEndIdx - gatewayStartIdx; + } + @Override + public void onboard(int idx) throws Exception { + int gwIdx = gatewayStartIdx + idx; + // 1) connect this gateway's client (persistent, autoReconnect via createClient) + // 2) subscribe RPC for this client via rpcReceiver (single-client attach) + wire reconnect recovery + // 3) announce its sub-devices through deviceAnnouncer.announce(...) (via the inherited warmUpPublish) + // 4) schedule this gateway's batch-telemetry timer + connectAnnounceSubscribeAndSchedule(gwIdx); } + }; + } + + /** + * One STAGGERED gateway's full onboarding step, composed entirely from existing pieces: the same + * connect sequence {@link #initClientBlocking} performs for PHASED's bulk connect, the same + * subscribe+reconnect wiring {@link #attachClientsToRpc} performs for PHASED's bulk attach, and the + * same per-device announce path ({@link #warmUpPublish}) PHASED's warm-up uses. Synchronous per the + * {@link EntityLifecycle#onboard} contract: throws on any failure so the engine counts this gateway + * as failed rather than onboarded. + */ + private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception { + int localIdx = gwIdx - gatewayStartIdx; + String gatewayName = staggeredGatewayNames.get(localIdx); + List deviceNames = staggeredGatewayDeviceNames.getOrDefault(localIdx, Collections.emptyList()); + + // 1) connect (persistent; createClient() applies autoReconnect() same as every other gateway client) + MqttClient client = initClientBlocking(gatewayName); + mqttClients.add(client); + clientNames.put(client, gatewayName); + + // Register this gateway's device group before any reconnect-triggered re-announce can occur. + gatewayDeviceNames.put(client, Collections.synchronizedList(new ArrayList<>(deviceNames))); + for (String deviceName : deviceNames) { + DeviceClient dc = new DeviceClient(); + dc.setMqttClient(client); + dc.setDeviceName(deviceName); + dc.setGatewayName(gatewayName); + deviceClients.add(dc); + } + + // 2) subscribe RPC for this client alone + wire its reconnect recovery (mirrors attachRpcReceiver's + // per-client wiring, done here per-gateway instead of once in bulk over mqttClients). + if (rpcEnabled) { + attachClientsToRpc(Collections.singletonList(client), 0); + } + + // 3) announce sub-devices through the same reliable (RPC on) / legacy QoS-0 (RPC off) path warm-up + // uses. No timeout here: an announce under retry can legitimately take longer than CONNECT_TIMEOUT; + // GatewayDeviceAnnouncer always eventually settles the future (acked or unconfirmed-after-retries). + for (String deviceName : deviceNames) { + DeviceClient dc = new DeviceClient(); + dc.setMqttClient(client); + dc.setDeviceName(deviceName); + warmUpPublish(dc).get(); } + + // 4) schedule this gateway's own batch-telemetry timer, starting immediately (its onboarding time + // is already spread by the engine's ramp jitter; an extra small per-gateway startup offset avoids + // every gateway's tick landing on the exact same millisecond). + scheduleGatewayTelemetry(gwIdx, client, gatewayName, deviceNames); + } + + /** Blocking connect for one gateway token, identical in behavior to the private {@code initClient} used + * by PHASED's {@code connectDevices} — reconstructed here (rather than reused) only because that method + * is {@code private} in {@link org.thingsboard.tools.service.shared.BaseMqttAPITest}. */ + private MqttClient initClientBlocking(String token) throws Exception { + MqttClient client = createClient(token); + Future connectFuture = connectAsync(client); + MqttConnectResult result; + try { + result = connectFuture.get(CONNECT_TIMEOUT, TimeUnit.SECONDS); + } catch (TimeoutException ex) { + connectFuture.cancel(true); + client.disconnect(); + throw new RuntimeException(String.format("STAGGERED: timed out connecting gateway [%s]", token), ex); + } + if (!result.isSuccess()) { + connectFuture.cancel(true); + client.disconnect(); + throw new RuntimeException(String.format("STAGGERED: failed to connect gateway [%s]. Result code: %s", token, result.getReturnCode())); + } + return client; + } + + /** Schedules this gateway's own periodic batch-telemetry publish (one MQTT publish carrying all of its + * sub-devices' next messages, same construction as {@link MqttGatewayBatchAPITest#nextPublishTask}), + * independent of every other gateway's timer. No-op when publishing is disabled (MESSAGES_PER_SECOND=0) + * or this gateway has no sub-devices. */ + private void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatewayName, List deviceNames) { + if (testMessagesPerSecond <= 0 || deviceNames.isEmpty()) { + return; + } + DeviceClient logClient = new DeviceClient(); + logClient.setMqttClient(client); + logClient.setGatewayName(gatewayName); + logClient.setDeviceName("batch[" + deviceNames.size() + " devices]"); + AtomicInteger tick = new AtomicInteger(); + long initialDelayMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + gwIdx), 1000L); + ScheduledFuture timer = restClientService.getScheduler().scheduleAtFixedRate(() -> { + try { + ObjectNode batch = mapper.createObjectNode(); + for (String deviceName : deviceNames) { + NodeMsg nodeMsg = getNextNodeMessage(deviceName, false); + batch.setAll(nodeMsg.getNode()); + } + byte[] data = mapper.writeValueAsBytes(batch); + int iteration = tick.incrementAndGet(); + client.publish(getTestTopic(), Unpooled.wrappedBuffer(data), MqttQoS.AT_MOST_ONCE) + .addListener(f -> { + if (f.isSuccess()) { + totalSuccessPublishedCount.incrementAndGet(); + logSuccessTestMessage(iteration, logClient); + } else { + totalFailedPublishedCount.incrementAndGet(); + logFailureTestMessage(iteration, logClient, f); + } + }); + } catch (Exception e) { + log.warn("STAGGERED telemetry publish failed for gateway [{}]", gatewayName, e); + } + }, initialDelayMs, 1000L, TimeUnit.MILLISECONDS); + gatewayTelemetryTimers.add(timer); } private void startRpcBurstSender() { @@ -332,6 +590,14 @@ protected void logFailureTestMessage(int iteration, DeviceClient client, Future< } protected void attachRpcReceiver() throws InterruptedException { + initRpcReceiver(); + attachClientsToRpc(mqttClients, warmUpPackSize); + } + + /** Builds {@code rpcReceiver}/{@code deviceAnnouncer} and registers their stats blocks. Split out of + * {@link #attachRpcReceiver()} (which still does exactly this + the bulk attach below, unchanged) + * so STAGGERED can construct these once, up front, before any gateway has connected. */ + private void initRpcReceiver() { ObjectMapper mapper = new ObjectMapper(); RpcResponseTemplate template = rpcRespond ? RpcResponseTemplate.load(rpcResponseTemplate) : null; RpcMessageProcessor processor = new RpcMessageProcessor(mapper, rpcSendTsPath, rpcRespond, template); @@ -346,11 +612,18 @@ protected void attachRpcReceiver() throws InterruptedException { statsReporter().register(StatsBlock.RPC_SUBSCRIPTION, rpcReceiver::subscriptionSummary); statsReporter().register(StatsBlock.RPC_IN, rpcReceiver::inSummary); statsReporter().register(StatsBlock.RPC_OUT, rpcReceiver::outSummary); - rpcReceiver.attach(mqttClients, warmUpPackSize); + } + + /** Subscribes the given clients to the RPC topic and wires each one's reconnect recovery. Split out + * of {@link #attachRpcReceiver()} (unchanged for PHASED: called once with {@code mqttClients} + + * {@code warmUpPackSize}) so STAGGERED can call it per-gateway with a singleton list + packSize=0 + * (pacing is meaningless for a single client). */ + private void attachClientsToRpc(List clients, int packSize) throws InterruptedException { + rpcReceiver.attach(clients, packSize); // On reconnect, a gateway loses its RPC subscription (cleanSession) and its server-side // sub-device routing; restore both so RPC delivery resumes instead of silently dropping. // Also flush any replies buffered while the channel was down so they land within the RPC expiry. - for (MqttClient client : mqttClients) { + for (MqttClient client : clients) { setReconnectAction(client, () -> { rpcReceiver.resubscribe(client); reannounceDevices(client); From 57b5d0a61db9224dbb19c07b39c033d151016332 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 16:54:45 +0300 Subject: [PATCH 04/13] fix(onboarding): derive STAGGERED per-gateway telemetry cadence from MPS, not 1s --- .../tools/service/gateway/MqttGatewayAPITest.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java index 9fdd4ed..603fb0e 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java @@ -502,7 +502,12 @@ private MqttClient initClientBlocking(String token) throws Exception { /** Schedules this gateway's own periodic batch-telemetry publish (one MQTT publish carrying all of its * sub-devices' next messages, same construction as {@link MqttGatewayBatchAPITest#nextPublishTask}), * independent of every other gateway's timer. No-op when publishing is disabled (MESSAGES_PER_SECOND=0) - * or this gateway has no sub-devices. */ + * or this gateway has no sub-devices. + *

Period is MPS-derived so STAGGERED's steady-state aggregate matches PHASED's: today's metronome + * does {@code testMessagesPerSecond} gateway-batch publishes/sec by sweeping the whole fleet, i.e. + * each gateway publishes once every {@code entityCount / testMessagesPerSecond} seconds — so each + * independent per-gateway timer here fires on that same period, jittered so the first fires aren't + * synchronized across gateways. */ private void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatewayName, List deviceNames) { if (testMessagesPerSecond <= 0 || deviceNames.isEmpty()) { return; @@ -512,7 +517,9 @@ private void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatew logClient.setGatewayName(gatewayName); logClient.setDeviceName("batch[" + deviceNames.size() + " devices]"); AtomicInteger tick = new AtomicInteger(); - long initialDelayMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + gwIdx), 1000L); + int entityCount = gatewayEndIdx - gatewayStartIdx; + long periodMs = Math.max(1L, (entityCount * 1000L) / testMessagesPerSecond); + long initialJitterMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + gwIdx), periodMs); ScheduledFuture timer = restClientService.getScheduler().scheduleAtFixedRate(() -> { try { ObjectNode batch = mapper.createObjectNode(); @@ -535,7 +542,7 @@ private void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatew } catch (Exception e) { log.warn("STAGGERED telemetry publish failed for gateway [{}]", gatewayName, e); } - }, initialDelayMs, 1000L, TimeUnit.MILLISECONDS); + }, initialJitterMs, periodMs, TimeUnit.MILLISECONDS); gatewayTelemetryTimers.add(timer); } From e9b8b4a3a6a9c9351a40b5850d621d478ec4c85b Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 17:13:35 +0300 Subject: [PATCH 05/13] fix(onboarding): commit STAGGERED gateway state only on full onboarding success; announce before subscribe; fail fast on unsupported config; add unit tests --- .../service/gateway/MqttGatewayAPITest.java | 116 ++++++++---- .../gateway/MqttGatewayAPITestTest.java | 175 ++++++++++++++++++ 2 files changed, 259 insertions(+), 32 deletions(-) create mode 100644 src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java index 603fb0e..1a7e4a8 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java @@ -68,6 +68,12 @@ public class MqttGatewayAPITest extends BaseMqttAPITest implements GatewayAPITes @Value("${gateway.count}") int gatewayCount; + // STAGGERED-only support check (see checkStaggeredSupported()): mirrors the same property that + // selects which bean loads (this class when gateway.batch!=true, MqttGatewayBatchAPITest when it's + // true), read directly here rather than inferred from the bean type. + @Value("${gateway.batch:false}") + boolean gatewayBatchEnabled; + @Value("${gateway.rpc.enabled:false}") boolean rpcEnabled; @Value("${gateway.rpc.topic:v1/gateway/rpc}") @@ -138,11 +144,13 @@ public class MqttGatewayAPITest extends BaseMqttAPITest implements GatewayAPITes // Unlike PHASED's mapDevicesToGatewayClientConnections (which derives the mapping from mqttClients' // connect order, established only after ALL gateways connect), this is index-based so onboard(idx) // can look up its own assignment before any connection exists. - private List staggeredGatewayNames; - private Map> staggeredGatewayDeviceNames; + // Package-private (not private): MqttGatewayAPITestTest reads/exercises these directly, following + // the existing broker-free unit-test idiom (test class extends the SUT and touches its own members). + List staggeredGatewayNames; + Map> staggeredGatewayDeviceNames; // Per-gateway telemetry timers started by the STAGGERED path (one per onboarded gateway); cancelled // when the test duration elapses. - private final List> gatewayTelemetryTimers = Collections.synchronizedList(new ArrayList<>()); + final List> gatewayTelemetryTimers = Collections.synchronizedList(new ArrayList<>()); private boolean staggered() { return "STAGGERED".equalsIgnoreCase(onboardMode); @@ -197,6 +205,7 @@ public void connectGateways() throws InterruptedException { // STAGGERED: no bulk connect here. Build the gateway/device name model only; the engine // (driven from runApiTests()) connects + announces + subscribes each gateway on its own // paced schedule. + checkStaggeredSupported(); prepareStaggeredModel(); return; } @@ -234,7 +243,9 @@ public void connectGateways() throws InterruptedException { } } - private void mapDevicesToGatewayClientConnections() { + // Package-private (not private): exercised directly by MqttGatewayAPITestTest for parity with + // prepareStaggeredModel(). + void mapDevicesToGatewayClientConnections() { int gatewayCount = mqttClients.size(); for (int i = deviceStartIdx; i < deviceEndIdx; i++) { int deviceIdx = i - deviceStartIdx; @@ -257,7 +268,8 @@ private void mapDevicesToGatewayClientConnections() { * so the assignment exists before any gateway has connected, and {@code onboard(idx)} can look up * its own sub-device names deterministically regardless of connect order/timing. */ - private void prepareStaggeredModel() { + // Package-private (not private): exercised directly by MqttGatewayAPITestTest. + void prepareStaggeredModel() { List gatewayNames; if (!gateways.isEmpty()) { gatewayNames = gateways.stream().map(Device::getName).collect(Collectors.toList()); @@ -280,6 +292,29 @@ private void prepareStaggeredModel() { log.info("STAGGERED model prepared: {} gateways, {} devices", gatewayCount, deviceEndIdx - deviceStartIdx); } + /** + * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it. + * {@link #scheduleGatewayTelemetry} always publishes a whole-gateway batch (needs {@code + * gateway.batch=true}) and never injects an alarm (needs {@code test.alarms.aps <= 0}). Deliberately + * narrow: STAGGERED currently supports exactly the combination case-a runs. + */ + void checkStaggeredSupported() { + if (!gatewayBatchEnabled) { + String msg = "onboard.mode=STAGGERED currently supports gateway.batch=true only (with no alarms); " + + "gateway.batch=false is not supported yet. Set GATEWAY_BATCH=true, or use onboard.mode=PHASED."; + log.error(msg); + throw new IllegalStateException(msg); + } + if (alarmsPerSecond > 0) { + String msg = String.format( + "onboard.mode=STAGGERED currently supports gateway.batch=true with NO alarms; " + + "test.alarms.aps=%d (> 0) is not supported yet. Set ALARMS_PER_SECOND=0, or use onboard.mode=PHASED.", + alarmsPerSecond); + log.error(msg); + throw new IllegalStateException(msg); + } + } + /** Re-announce a reconnected gateway's sub-devices so the server re-routes their RPC through it. * Reuses the gateway connect topic and payload used at warm-up. */ private void reannounceDevices(MqttClient gatewayClient) { @@ -419,8 +454,8 @@ public int entityCount() { public void onboard(int idx) throws Exception { int gwIdx = gatewayStartIdx + idx; // 1) connect this gateway's client (persistent, autoReconnect via createClient) - // 2) subscribe RPC for this client via rpcReceiver (single-client attach) + wire reconnect recovery - // 3) announce its sub-devices through deviceAnnouncer.announce(...) (via the inherited warmUpPublish) + // 2) announce its sub-devices through deviceAnnouncer.announce(...) (via the inherited warmUpPublish) + // 3) subscribe RPC for this client via rpcReceiver (single-client attach) + wire reconnect recovery // 4) schedule this gateway's batch-telemetry timer connectAnnounceSubscribeAndSchedule(gwIdx); } @@ -430,10 +465,17 @@ public void onboard(int idx) throws Exception { /** * One STAGGERED gateway's full onboarding step, composed entirely from existing pieces: the same * connect sequence {@link #initClientBlocking} performs for PHASED's bulk connect, the same - * subscribe+reconnect wiring {@link #attachClientsToRpc} performs for PHASED's bulk attach, and the - * same per-device announce path ({@link #warmUpPublish}) PHASED's warm-up uses. Synchronous per the - * {@link EntityLifecycle#onboard} contract: throws on any failure so the engine counts this gateway - * as failed rather than onboarded. + * per-device announce path ({@link #warmUpPublish}) PHASED's warm-up uses, and the same + * subscribe+reconnect wiring {@link #attachClientsToRpc} performs for PHASED's bulk attach — in that + * order (connect -> announce -> subscribe), per the {@link EntityLifecycle#onboard} contract. + * Synchronous: throws on any failure so the engine counts this gateway as failed rather than + * onboarded. + *

Nothing is registered into the shared {@code mqttClients}/{@code deviceClients}/ + * {@code gatewayDeviceNames}/{@code clientNames} collections — which {@link #startRpcBurstSender()}, + * the telemetry scheduler, and reconnect recovery all read from wholesale — until the ENTIRE sequence + * below has succeeded. A gateway that fails partway (e.g. its subscribe throws after announce + * succeeded) is closed and left out of every shared collection entirely, so a partial onboarding can + * never leak un-announced/un-subscribed devices into the RPC-outcome measurement. */ private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception { int localIdx = gwIdx - gatewayStartIdx; @@ -442,34 +484,43 @@ private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception { // 1) connect (persistent; createClient() applies autoReconnect() same as every other gateway client) MqttClient client = initClientBlocking(gatewayName); + try { + // 2) announce sub-devices through the same reliable (RPC on) / legacy QoS-0 (RPC off) path + // warm-up uses. No timeout here: an announce under retry can legitimately take longer than + // CONNECT_TIMEOUT; GatewayDeviceAnnouncer always eventually settles the future (acked or + // unconfirmed-after-retries). + for (String deviceName : deviceNames) { + DeviceClient dc = new DeviceClient(); + dc.setMqttClient(client); + dc.setDeviceName(deviceName); + warmUpPublish(dc).get(); + } + + // 3) subscribe RPC for this client alone + wire its reconnect recovery (mirrors + // attachRpcReceiver's per-client wiring, done here per-gateway instead of once in bulk over + // mqttClients). + if (rpcEnabled) { + attachClientsToRpc(Collections.singletonList(client), 0); + } + } catch (Exception e) { + client.disconnect(); + throw e; + } + + // Onboarding succeeded end-to-end: only now commit this gateway's client + devices into the + // shared collections and start its telemetry timer. mqttClients.add(client); clientNames.put(client, gatewayName); - - // Register this gateway's device group before any reconnect-triggered re-announce can occur. gatewayDeviceNames.put(client, Collections.synchronizedList(new ArrayList<>(deviceNames))); + List newDeviceClients = new ArrayList<>(deviceNames.size()); for (String deviceName : deviceNames) { DeviceClient dc = new DeviceClient(); dc.setMqttClient(client); dc.setDeviceName(deviceName); dc.setGatewayName(gatewayName); - deviceClients.add(dc); - } - - // 2) subscribe RPC for this client alone + wire its reconnect recovery (mirrors attachRpcReceiver's - // per-client wiring, done here per-gateway instead of once in bulk over mqttClients). - if (rpcEnabled) { - attachClientsToRpc(Collections.singletonList(client), 0); - } - - // 3) announce sub-devices through the same reliable (RPC on) / legacy QoS-0 (RPC off) path warm-up - // uses. No timeout here: an announce under retry can legitimately take longer than CONNECT_TIMEOUT; - // GatewayDeviceAnnouncer always eventually settles the future (acked or unconfirmed-after-retries). - for (String deviceName : deviceNames) { - DeviceClient dc = new DeviceClient(); - dc.setMqttClient(client); - dc.setDeviceName(deviceName); - warmUpPublish(dc).get(); + newDeviceClients.add(dc); } + deviceClients.addAll(newDeviceClients); // 4) schedule this gateway's own batch-telemetry timer, starting immediately (its onboarding time // is already spread by the engine's ramp jitter; an extra small per-gateway startup offset avoids @@ -507,8 +558,9 @@ private MqttClient initClientBlocking(String token) throws Exception { * does {@code testMessagesPerSecond} gateway-batch publishes/sec by sweeping the whole fleet, i.e. * each gateway publishes once every {@code entityCount / testMessagesPerSecond} seconds — so each * independent per-gateway timer here fires on that same period, jittered so the first fires aren't - * synchronized across gateways. */ - private void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatewayName, List deviceNames) { + * synchronized across gateways. + *

Package-private (not private): exercised directly by MqttGatewayAPITestTest. */ + void scheduleGatewayTelemetry(int gwIdx, MqttClient client, String gatewayName, List deviceNames) { if (testMessagesPerSecond <= 0 || deviceNames.isEmpty()) { return; } diff --git a/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java new file mode 100644 index 0000000..db0beeb --- /dev/null +++ b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java @@ -0,0 +1,175 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.tools.service.gateway; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.thingsboard.mqtt.MqttClient; +import org.thingsboard.tools.service.mqtt.DeviceClient; +import org.thingsboard.tools.service.shared.RestClientService; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +// Same broker-free idiom as MqttGatewayBatchAPITestTest: this test class IS the SUT (extends +// MqttGatewayAPITest) so it can reach the package-private/protected inherited state +// (gatewayStartIdx/gatewayEndIdx/deviceStartIdx/deviceEndIdx/mqttClients/deviceClients/seed/...) without +// a Spring context or a real MQTT broker. init()/@PostConstruct is never invoked, so every field a test +// relies on is set explicitly. +class MqttGatewayAPITestTest extends MqttGatewayAPITest { + + ScheduledExecutorService scheduler; + + @BeforeEach + void setUp() { + scheduler = mock(ScheduledExecutorService.class); + RestClientService rcs = mock(RestClientService.class); + when(rcs.getScheduler()).thenReturn(scheduler); + restClientService = rcs; + } + + // --- prepareStaggeredModel() vs mapDevicesToGatewayClientConnections(): same device assignment --- + + @Test + void prepareStaggeredModelAssignsDevicesIdenticallyToPhasedMapping() { + gatewayStartIdx = 0; + gatewayEndIdx = 3; + deviceStartIdx = 0; + deviceEndIdx = 7; // not a multiple of 3: exercises the uneven-remainder case too + + // PHASED reference: 3 already-connected gateway clients, grouped by mqttClients position. + MqttClient gw0 = mock(MqttClient.class); + MqttClient gw1 = mock(MqttClient.class); + MqttClient gw2 = mock(MqttClient.class); + mqttClients.add(gw0); + mqttClients.add(gw1); + mqttClients.add(gw2); + clientNames.put(gw0, "GW00000000"); + clientNames.put(gw1, "GW00000001"); + clientNames.put(gw2, "GW00000002"); + mapDevicesToGatewayClientConnections(); + + Map> phasedGroups = new HashMap<>(); + for (DeviceClient dc : deviceClients) { + phasedGroups.computeIfAbsent(dc.getMqttClient(), k -> new ArrayList<>()).add(dc.getDeviceName()); + } + + // STAGGERED model: index-based, built without any connected client. + prepareStaggeredModel(); + + List byPosition = List.of(gw0, gw1, gw2); + for (int gwIdx = 0; gwIdx < 3; gwIdx++) { + List staggered = staggeredGatewayDeviceNames.getOrDefault(gwIdx, List.of()); + List phased = phasedGroups.getOrDefault(byPosition.get(gwIdx), List.of()); + assertThat(staggered).containsExactlyElementsOf(phased); + } + assertThat(staggeredGatewayNames).containsExactly("GW00000000", "GW00000001", "GW00000002"); + } + + // --- scheduleGatewayTelemetry(): MPS-derived period + jitter, and the no-timer guard --- + + @Test + void telemetryPeriodIsDerivedFromEntityCountAndMessagesPerSecond() { + gatewayStartIdx = 0; + gatewayEndIdx = 10; // entityCount = 10 + testMessagesPerSecond = 5; + seed = 0; + + // Raw type (not ScheduledFuture): a wildcard-typed mock hits a generic-capture mismatch on + // thenReturn (javac can't unify two independent "capture of ?" instantiations). + @SuppressWarnings({"unchecked", "rawtypes"}) + ScheduledFuture fakeFuture = mock(ScheduledFuture.class); + when(scheduler.scheduleAtFixedRate(any(), anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS))) + .thenReturn(fakeFuture); + + MqttClient client = mock(MqttClient.class); + scheduleGatewayTelemetry(3, client, "GW00000003", List.of("DW00000000")); + + ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class); + ArgumentCaptor periodCaptor = ArgumentCaptor.forClass(Long.class); + verify(scheduler).scheduleAtFixedRate(any(), delayCaptor.capture(), periodCaptor.capture(), eq(TimeUnit.MILLISECONDS)); + + long expectedPeriodMs = (10 * 1000L) / 5; // entityCount * 1000 / MPS = 2000ms + assertThat(periodCaptor.getValue()).isEqualTo(expectedPeriodMs); + assertThat(delayCaptor.getValue()).isBetween(0L, expectedPeriodMs - 1); + assertThat(gatewayTelemetryTimers).containsExactly(fakeFuture); + } + + @Test + void noTelemetryTimerScheduledWhenMessagesPerSecondIsZeroOrLess() { + gatewayStartIdx = 0; + gatewayEndIdx = 10; + testMessagesPerSecond = 0; // no-publish mode: mirrors AbstractAPITest.runApiTests' no-publish branch + + MqttClient client = mock(MqttClient.class); + scheduleGatewayTelemetry(0, client, "GW00000000", List.of("DW00000000")); + + verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any()); + assertThat(gatewayTelemetryTimers).isEmpty(); + } + + @Test + void noTelemetryTimerScheduledWhenGatewayHasNoSubDevices() { + gatewayStartIdx = 0; + gatewayEndIdx = 10; + testMessagesPerSecond = 5; + + MqttClient client = mock(MqttClient.class); + scheduleGatewayTelemetry(0, client, "GW00000000", List.of()); + + verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any()); + assertThat(gatewayTelemetryTimers).isEmpty(); + } + + // --- checkStaggeredSupported(): fail fast on the unsupported configs instead of silently diverging --- + + @Test + void checkStaggeredSupportedThrowsWhenGatewayBatchDisabled() { + gatewayBatchEnabled = false; + alarmsPerSecond = 0; + assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class); + } + + @Test + void checkStaggeredSupportedThrowsWhenAlarmsEnabled() { + gatewayBatchEnabled = true; + alarmsPerSecond = 1; + assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class); + } + + @Test + void checkStaggeredSupportedPassesForTheSupportedCombination() { + gatewayBatchEnabled = true; + alarmsPerSecond = 0; + checkStaggeredSupported(); // must not throw + } +} From 1e09aafe98e196710d1971bb8c16f732a7f9d230 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 17:24:33 +0300 Subject: [PATCH 06/13] feat(onboarding): STAGGERED lifecycle for persistent device mode --- .../service/device/MqttDeviceAPITest.java | 225 +++++++++++++++++- .../service/device/MqttDeviceAPITestTest.java | 121 ++++++++++ 2 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java diff --git a/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java b/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java index f192b1f..666e35b 100644 --- a/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java @@ -15,15 +15,22 @@ */ package org.thingsboard.tools.service.device; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.mqtt.MqttQoS; import io.netty.util.concurrent.Future; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Service; import org.thingsboard.mqtt.MqttClient; +import org.thingsboard.mqtt.MqttConnectResult; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.IdBased; +import org.thingsboard.tools.service.gateway.EphemeralSchedule; +import org.thingsboard.tools.service.msg.Msg; import org.thingsboard.tools.service.mqtt.DeviceClient; import org.thingsboard.tools.service.shared.BaseMqttAPITest; +import org.thingsboard.tools.service.shared.onboarding.EntityLifecycle; +import org.thingsboard.tools.service.shared.onboarding.StaggeredOnboardingEngine; import java.nio.charset.StandardCharsets; import java.util.ArrayList; @@ -32,6 +39,9 @@ import java.util.List; import java.util.Random; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -43,6 +53,23 @@ public class MqttDeviceAPITest extends BaseMqttAPITest implements DeviceAPITest static String dataAsStr = "{\"t1\":73}"; static byte[] data = dataAsStr.getBytes(StandardCharsets.UTF_8); + // STAGGERED onboarding (onboard.mode=STAGGERED): PHASED (default) never touches any of these. + private StaggeredOnboardingEngine onboardingEngine; + // Precomputed device-index -> name model, built once by prepareStaggeredModel(). Unlike PHASED's + // mapDevicesToDeviceClientConnections (which derives each device's name from its already-connected + // MqttClient's username), this is index-based so onboard(idx) can look up its own name before any + // connection exists. + // Package-private (not private): MqttDeviceAPITestTest reads/exercises these directly, following the + // existing broker-free unit-test idiom (test class extends the SUT and touches its own members). + List staggeredDeviceNames; + // Per-device telemetry timers started by the STAGGERED path (one per onboarded device); cancelled + // when the test duration elapses. + final List> deviceTelemetryTimers = Collections.synchronizedList(new ArrayList<>()); + + private boolean staggered() { + return "STAGGERED".equalsIgnoreCase(onboardMode); + } + @Override public void createDevices() throws Exception { createDevices(true); @@ -53,9 +80,176 @@ public void removeDevices() throws Exception { removeEntities(devices.stream().map(IdBased::getId).collect(Collectors.toList()), "devices"); } + @Override + public void warmUpDevices() throws InterruptedException { + if (staggered()) { + // STAGGERED: no separate warm-up phase; each device's own periodic telemetry timer (started + // by EntityLifecycle.onboard(), driven from runApiTests()) sends its own first message. + return; + } + super.warmUpDevices(); + } + @Override public void runApiTests() throws InterruptedException { - super.runApiTests(mqttClients.size()); + if (!staggered()) { + super.runApiTests(mqttClients.size()); + return; + } + runStaggeredApiTests(); + } + + /** + * STAGGERED: ramp devices in through the engine (each onboard connects and schedules its own + * telemetry timer — see {@link #connectAndScheduleDevice(int)}), then hold for the test duration. + * Direct-device mode has no RPC subscribe step (case-a is gateway-based) — STAGGERED here is + * telemetry-only, matching what {@link MqttDeviceAPITest} already supports in PHASED. + */ + private void runStaggeredApiTests() throws InterruptedException { + statsReporter().start(); + onboardingEngine = new StaggeredOnboardingEngine( + deviceLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed); + onboardingEngine.start((onboarded, failed) -> + log.info("STAGGERED device ramp complete: {} onboarded, {} failed", onboarded, failed)); + try { + Thread.sleep(testDurationInSec * 1000L); + } finally { + for (ScheduledFuture timer : deviceTelemetryTimers) { + timer.cancel(false); + } + if (onboardingEngine != null) { + onboardingEngine.stop(); + } + } + } + + private EntityLifecycle deviceLifecycle() { + return new EntityLifecycle() { + @Override + public int entityCount() { + return deviceEndIdx - deviceStartIdx; + } + + @Override + public void onboard(int idx) throws Exception { + int devIdx = deviceStartIdx + idx; + // 1) connect this device's client (persistent, autoReconnect via createClient) + // 2) schedule this device's telemetry timer + connectAndScheduleDevice(devIdx); + } + }; + } + + /** + * One STAGGERED device's full onboarding step, composed entirely from existing pieces: the same + * connect sequence {@link #initClientBlocking} performs for PHASED's bulk connect. Synchronous: + * throws on any failure so the engine counts this device as failed rather than onboarded. + *

Nothing is registered into the shared {@code mqttClients}/{@code deviceClients} collections — + * which the telemetry scheduler reads from — until the connect above has succeeded; a device that + * fails to connect is closed (by {@link #initClientBlocking}) and left out of every shared collection + * entirely. + */ + private void connectAndScheduleDevice(int devIdx) throws Exception { + int localIdx = devIdx - deviceStartIdx; + String deviceName = staggeredDeviceNames.get(localIdx); + + // 1) connect (persistent; createClient() applies autoReconnect() same as every other device client) + MqttClient client = initClientBlocking(deviceName); + + // Onboarding succeeded: only now commit this device's client into the shared collections and + // start its telemetry timer. + mqttClients.add(client); + clientNames.put(client, deviceName); + DeviceClient dc = new DeviceClient(); + dc.setMqttClient(client); + dc.setDeviceName(deviceName); + deviceClients.add(dc); + + // 2) schedule this device's own telemetry timer, starting immediately (its onboarding time is + // already spread by the engine's ramp jitter; an extra small per-device startup offset avoids + // every device's tick landing on the exact same millisecond). + scheduleDeviceTelemetry(devIdx, client, deviceName); + } + + /** Blocking connect for one device token, identical in behavior to the private {@code initClient} used + * by PHASED's {@code connectDevices} — reconstructed here (rather than reused) only because that method + * is {@code private} in {@link org.thingsboard.tools.service.shared.BaseMqttAPITest}. */ + private MqttClient initClientBlocking(String token) throws Exception { + MqttClient client = createClient(token); + Future connectFuture = connectAsync(client); + MqttConnectResult result; + try { + result = connectFuture.get(CONNECT_TIMEOUT, TimeUnit.SECONDS); + } catch (TimeoutException ex) { + connectFuture.cancel(true); + client.disconnect(); + throw new RuntimeException(String.format("STAGGERED: timed out connecting device [%s]", token), ex); + } + if (!result.isSuccess()) { + connectFuture.cancel(true); + client.disconnect(); + throw new RuntimeException(String.format("STAGGERED: failed to connect device [%s]. Result code: %s", token, result.getReturnCode())); + } + return client; + } + + /** Schedules this device's own periodic telemetry publish (one MQTT publish per device, same message + * construction {@link org.thingsboard.tools.service.shared.BaseMqttAPITest#nextPublishTask} uses for + * PHASED's per-device publish), independent of every other device's timer. No-op when publishing is + * disabled (MESSAGES_PER_SECOND=0). + *

Period is MPS-derived so STAGGERED's steady-state aggregate matches PHASED's: today's metronome + * does {@code testMessagesPerSecond} single-device publishes/sec by sweeping the whole fleet, i.e. + * each device publishes once every {@code entityCount / testMessagesPerSecond} seconds — so each + * independent per-device timer here fires on that same period, jittered so the first fires aren't + * synchronized across devices. + *

Package-private (not private): exercised directly by MqttDeviceAPITestTest. */ + void scheduleDeviceTelemetry(int devIdx, MqttClient client, String deviceName) { + if (testMessagesPerSecond <= 0) { + return; + } + DeviceClient logClient = new DeviceClient(); + logClient.setMqttClient(client); + logClient.setDeviceName(deviceName); + AtomicInteger tick = new AtomicInteger(); + int entityCount = deviceEndIdx - deviceStartIdx; + long periodMs = Math.max(1L, (entityCount * 1000L) / testMessagesPerSecond); + long initialJitterMs = EphemeralSchedule.firstOffsetMillis(new Random(seed + devIdx), periodMs); + ScheduledFuture timer = restClientService.getScheduler().scheduleAtFixedRate(() -> { + try { + Msg message = getNextMessage(deviceName, false); + int iteration = tick.incrementAndGet(); + client.publish(getTestTopic(), Unpooled.wrappedBuffer(message.getData()), MqttQoS.AT_MOST_ONCE) + .addListener(f -> { + if (f.isSuccess()) { + totalSuccessPublishedCount.incrementAndGet(); + logSuccessTestMessage(iteration, logClient); + } else { + totalFailedPublishedCount.incrementAndGet(); + logFailureTestMessage(iteration, logClient, f); + } + }); + } catch (Exception e) { + log.warn("STAGGERED telemetry publish failed for device [{}]", deviceName, e); + } + }, initialJitterMs, periodMs, TimeUnit.MILLISECONDS); + deviceTelemetryTimers.add(timer); + } + + /** + * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it. + * {@link #scheduleDeviceTelemetry} always publishes one plain per-device message and never injects an + * alarm (needs {@code test.alarms.aps <= 0}). Deliberately narrow: STAGGERED currently supports + * exactly the no-alarms combination case-a runs. + */ + void checkStaggeredSupported() { + if (alarmsPerSecond > 0) { + String msg = String.format( + "onboard.mode=STAGGERED currently supports NO alarms for direct-device mode; " + + "test.alarms.aps=%d (> 0) is not supported yet. Set ALARMS_PER_SECOND=0, or use onboard.mode=PHASED.", + alarmsPerSecond); + log.error(msg); + throw new IllegalStateException(msg); + } } @Override @@ -90,6 +284,13 @@ protected void logFailureTestMessage(int iteration, DeviceClient client, Future< @Override public void connectDevices() throws InterruptedException { + if (staggered()) { + // STAGGERED: no bulk connect here. Build the device name model only; the engine (driven from + // runApiTests()) connects + schedules each device's telemetry timer on its own paced schedule. + checkStaggeredSupported(); + prepareStaggeredModel(); + return; + } AtomicInteger totalConnectedCount = new AtomicInteger(); List pack = null; List devicesNames; @@ -124,6 +325,28 @@ public void generationX509() { } + /** + * STAGGERED-only: same device-name resolution as this method's PHASED body above, but index-based + * (0-based, local to this instance's [deviceStartIdx, deviceEndIdx)) instead of derived from each + * already-connected {@link MqttClient}'s username — so the assignment exists before any device has + * connected, and {@code onboard(idx)} can look up its own name deterministically regardless of + * connect order/timing. + */ + // Package-private (not private): exercised directly by MqttDeviceAPITestTest. + void prepareStaggeredModel() { + List devicesNames; + if (!devices.isEmpty()) { + devicesNames = devices.stream().map(Device::getName).collect(Collectors.toList()); + } else { + devicesNames = new ArrayList<>(); + for (int i = deviceStartIdx; i < deviceEndIdx; i++) { + devicesNames.add(getToken(false, i)); + } + } + this.staggeredDeviceNames = devicesNames; + log.info("STAGGERED model prepared: {} devices", devicesNames.size()); + } + private void mapDevicesToDeviceClientConnections() { for (MqttClient mqttClient : mqttClients) { DeviceClient client = new DeviceClient(); diff --git a/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java b/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java new file mode 100644 index 0000000..18c2178 --- /dev/null +++ b/src/test/java/org/thingsboard/tools/service/device/MqttDeviceAPITestTest.java @@ -0,0 +1,121 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.tools.service.device; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.thingsboard.mqtt.MqttClient; +import org.thingsboard.tools.service.shared.RestClientService; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +// Same broker-free idiom as MqttGatewayAPITestTest: this test class IS the SUT (extends +// MqttDeviceAPITest) so it can reach the package-private/protected inherited state +// (deviceStartIdx/deviceEndIdx/mqttClients/deviceClients/seed/...) without a Spring context or a real +// MQTT broker. init()/@PostConstruct is never invoked, so every field a test relies on is set explicitly. +class MqttDeviceAPITestTest extends MqttDeviceAPITest { + + ScheduledExecutorService scheduler; + + @BeforeEach + void setUp() { + scheduler = mock(ScheduledExecutorService.class); + RestClientService rcs = mock(RestClientService.class); + when(rcs.getScheduler()).thenReturn(scheduler); + restClientService = rcs; + } + + // --- scheduleDeviceTelemetry(): MPS-derived period + jitter, and the no-timer guard --- + + @Test + void telemetryPeriodIsDerivedFromEntityCountAndMessagesPerSecond() { + deviceStartIdx = 0; + deviceEndIdx = 10; // entityCount = 10 + testMessagesPerSecond = 5; + seed = 0; + + // Raw type (not ScheduledFuture): a wildcard-typed mock hits a generic-capture mismatch on + // thenReturn (javac can't unify two independent "capture of ?" instantiations). + @SuppressWarnings({"unchecked", "rawtypes"}) + ScheduledFuture fakeFuture = mock(ScheduledFuture.class); + when(scheduler.scheduleAtFixedRate(any(), anyLong(), anyLong(), eq(TimeUnit.MILLISECONDS))) + .thenReturn(fakeFuture); + + MqttClient client = mock(MqttClient.class); + scheduleDeviceTelemetry(3, client, "DW00000003"); + + ArgumentCaptor delayCaptor = ArgumentCaptor.forClass(Long.class); + ArgumentCaptor periodCaptor = ArgumentCaptor.forClass(Long.class); + verify(scheduler).scheduleAtFixedRate(any(), delayCaptor.capture(), periodCaptor.capture(), eq(TimeUnit.MILLISECONDS)); + + long expectedPeriodMs = (10 * 1000L) / 5; // entityCount * 1000 / MPS = 2000ms + assertThat(periodCaptor.getValue()).isEqualTo(expectedPeriodMs); + assertThat(delayCaptor.getValue()).isBetween(0L, expectedPeriodMs - 1); + assertThat(deviceTelemetryTimers).containsExactly(fakeFuture); + } + + @Test + void noTelemetryTimerScheduledWhenMessagesPerSecondIsZeroOrLess() { + deviceStartIdx = 0; + deviceEndIdx = 10; + testMessagesPerSecond = 0; // no-publish mode: mirrors AbstractAPITest.runApiTests' no-publish branch + + MqttClient client = mock(MqttClient.class); + scheduleDeviceTelemetry(0, client, "DW00000000"); + + verify(scheduler, never()).scheduleAtFixedRate(any(), anyLong(), anyLong(), any()); + assertThat(deviceTelemetryTimers).isEmpty(); + } + + // --- checkStaggeredSupported(): fail fast on the unsupported configs instead of silently diverging --- + + @Test + void checkStaggeredSupportedThrowsWhenAlarmsEnabled() { + alarmsPerSecond = 1; + assertThatThrownBy(this::checkStaggeredSupported).isInstanceOf(IllegalStateException.class); + } + + @Test + void checkStaggeredSupportedPassesWhenAlarmsDisabled() { + alarmsPerSecond = 0; + checkStaggeredSupported(); // must not throw + } + + // --- prepareStaggeredModel(): same device-name resolution as PHASED's connectDevices() body --- + + @Test + void prepareStaggeredModelResolvesNamesFromIndexRangeWhenNoDevicesLoaded() { + deviceStartIdx = 100; + deviceEndIdx = 103; + + prepareStaggeredModel(); + + assertThat(staggeredDeviceNames).hasSize(3); + } +} From ad030cc6bc07c6c247e1684b3466fe229a868978 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 17:42:57 +0300 Subject: [PATCH 07/13] docs(onboarding): STAGGERED smoke-test runbook + README usage section --- README.md | 28 ++++++ STAGGERED-ONBOARDING-RUNBOOK.md | 164 ++++++++++++++++++++++++++++++++ 2 files changed, 192 insertions(+) create mode 100644 STAGGERED-ONBOARDING-RUNBOOK.md diff --git a/README.md b/README.md index f9950ee..2b8b638 100644 --- a/README.md +++ b/README.md @@ -129,3 +129,31 @@ docker run -it --rm --network host --name tb-perf-test \ --env TEST_PAYLOAD_TYPE=SMART_METER \ thingsboard/tb-ce-performance-test:latest ``` + +## Staggered onboarding mode + +The persistent gateway (`TEST_API=gateway`) and direct-device (`TEST_API=device`) modes support two +onboarding strategies, controlled by `ONBOARD_MODE`: + +- **`PHASED`** (default) — today's behavior: connect the whole fleet in packs, warm it up, then run the + fixed-rate telemetry metronome. No config change is needed to keep this behavior. +- **`STAGGERED`** — pace the fleet in one entity (gateway or device) at a time, each at a random offset + within a jitter window, capped at a configurable number onboarding concurrently. Each entity starts + publishing its own telemetry as soon as *it* onboards (not after the whole fleet finishes), on a cadence + derived from `MESSAGES_PER_SECOND` so the steady-state aggregate throughput matches `PHASED`. In gateway + mode, if the in-tool RPC burst sender is enabled it starts only once every entity has reached a terminal + state (onboarded or failed) — i.e. after the whole fleet has ramped in, not before any connection exists + as `PHASED` does. + +| Variable | Default | Description | +|---|---|---| +| `ONBOARD_MODE` | `PHASED` | `PHASED` or `STAGGERED` (see above) | +| `ONBOARD_MAX_CONCURRENT` | `200` | `STAGGERED` only: max entities onboarding at once | +| `ONBOARD_FIRST_JITTER_SEC` | `60` | `STAGGERED` only: each entity's first onboard attempt is scheduled at a random offset in `[0, this)` seconds | + +`STAGGERED` currently supports only the combination this feature was built for: gateway mode requires +`GATEWAY_BATCH=true`, and both gateway and device mode require `ALARMS_PER_SECOND=0`. An unsupported +combination fails fast at startup with a clear error instead of silently behaving like `PHASED`. + +See [`STAGGERED-ONBOARDING-RUNBOOK.md`](./STAGGERED-ONBOARDING-RUNBOOK.md) for a smoke-test procedure +(exact env + the expected log sequence) and known observability gaps in this mode. diff --git a/STAGGERED-ONBOARDING-RUNBOOK.md b/STAGGERED-ONBOARDING-RUNBOOK.md new file mode 100644 index 0000000..c679b28 --- /dev/null +++ b/STAGGERED-ONBOARDING-RUNBOOK.md @@ -0,0 +1,164 @@ +# STAGGERED onboarding — smoke-test runbook + +This is a manual verification procedure for `ONBOARD_MODE=STAGGERED` against a real MQTT broker / +ThingsBoard instance. It has **not been executed** by this task — no live broker was available in the +environment that produced this document. Run it yourself against a local/dev ThingsBoard instance before +relying on STAGGERED at scale. All log lines quoted below are copied verbatim from the current source +(`StaggeredOnboardingEngine`, `MqttGatewayAPITest`, `MqttDeviceAPITest`, `RpcBurstSender`) — grep the code +if a line doesn't show up; it may mean the guard in front of it (see "Known gaps" below) suppressed it. + +## 1. Environment (small, fast, observable) + +Gateway mode, RPC on, so the "RPC sender starts only after ramp-complete" behavior is exercised: + +```bash +TEST_API=gateway +REST_URL=http://127.0.0.1:8080 +MQTT_HOST=127.0.0.1 +REST_USERNAME=tenant@thingsboard.org +REST_PASSWORD=tenant + +# small fleet: 6 gateways x 2 sub-devices +GATEWAY_START_IDX=0 +GATEWAY_END_IDX=6 +DEVICE_START_IDX=0 +DEVICE_END_IDX=12 +GATEWAY_CREATE_ON_START=true +GATEWAY_DELETE_ON_COMPLETE=true + +# STAGGERED currently requires these two (checkStaggeredSupported fails fast otherwise) +GATEWAY_BATCH=true +ALARMS_PER_SECOND=0 + +ONBOARD_MODE=STAGGERED +ONBOARD_MAX_CONCURRENT=2 # low cap relative to 6 gateways so pacing is visible +ONBOARD_FIRST_JITTER_SEC=20 # short but long enough to see the ramp spread out + +MESSAGES_PER_SECOND=6 +DURATION_IN_SECONDS=90 + +GATEWAY_RPC_ENABLED=true +GATEWAY_RPC_SENDER_ENABLED=true +GATEWAY_RPC_SENDER_INTERVAL_SEC=60 + +STATS_LOG_ENABLED=true +STATS_LOG_INTERVAL_SEC=10 +``` + +For a **device**-mode smoke instead, set `TEST_API=device`, drop the gateway/RPC keys, and use +`DEVICE_START_IDX`/`DEVICE_END_IDX` for the fleet size — direct-device STAGGERED has no RPC step. + +## 2. Expected log sequence + +All lines below are `INFO` unless noted; `%` placeholders are the actual `{}` slots from the code. + +1. **Model built (gateway mode only; runs inside `connectGateways()`, before the engine starts):** + ``` + STAGGERED model prepared: 6 gateways, 12 devices + ``` + (`MqttGatewayAPITest.prepareStaggeredModel()`). Device mode logs the device-only equivalent: + ``` + STAGGERED model prepared: 12 devices + ``` + +2. **Ramp starts** (`StaggeredOnboardingEngine.start()`): + ``` + Staggered onboarding starting: 6 entities, maxConcurrent=2, firstJitter=20000ms + ``` + Confirm `maxConcurrent` and `firstJitter` match the env above. + +3. **Paced onboarding, peak concurrency ≤ cap.** There is no per-success log line at INFO (only + per-*failure* is logged — see below), so pacing is verified structurally + by timing rather than by + counting a log line: + - The engine bounds concurrent onboards with a `Semaphore(ONBOARD_MAX_CONCURRENT)` — this is enforced + in code (`StaggeredOnboardingEngine.onboardOne`), not just logged, so "≤ cap" holds by construction + as long as `Ramp complete` (step 5) doesn't fire suspiciously fast. + - With `STATS_LOG_ENABLED=true`, watch for periodic `THROUGHPUT`/telemetry `DEBUG` lines (enable + `logging.level.org.thingsboard.tools=DEBUG` to see them) — each gateway's own + `[N] Message was successfully published to device: ... and gateway: ...` line should start appearing + at different wall-clock times as each gateway finishes its own onboard, not all at once. With + `ONBOARD_MAX_CONCURRENT=2` and 6 gateways, expect the *last* gateway's first telemetry line noticeably + later than the first gateway's — never all 6 appearing within the same second. + - Sanity bound: `Ramp complete` (step 5) should land at least `ONBOARD_FIRST_JITTER_SEC` after step 2 + (the jitter alone spreads first-attempt times over that window), and later still if any gateway had + to queue for a permit. + - A connect/announce/subscribe failure for one entity logs (`WARN`, from the engine, not the caller): + ``` + Onboard failed for entity 3: java.lang.RuntimeException: ... + ``` + None expected in a clean smoke run against a healthy broker. + +4. **Per-gateway telemetry throughout.** Each gateway's own batch-telemetry timer starts immediately after + *that* gateway's onboarding succeeds (not after the whole ramp) — `MqttGatewayAPITest. + scheduleGatewayTelemetry()`. At `DEBUG`: + ``` + [1] Message was successfully published to device: batch[2 devices] and gateway: GW00000000 + ``` + one such line per gateway per publish tick, ticks starting at different times per gateway (jittered) + and continuing at a steady period for the rest of the run. A publish failure (should not happen against + a healthy broker) logs at `ERROR`: + ``` + [1] Error while publishing message to device: batch[...] and gateway: GW00000000 ... + ``` + Device mode: same idea, `scheduleDeviceTelemetry()`, message text + `Message was successfully published to device: ` (device mode has no `and gateway:` suffix). + +5. **Ramp complete, RPC sender starts only now.** Two lines fire back-to-back from the same callback + (`MqttGatewayAPITest.runStaggeredApiTests`'s `onComplete`), immediately preceded by the engine's own + completion line: + ``` + Ramp complete: 6 onboarded, 0 failed + STAGGERED gateway ramp complete: 6 onboarded, 0 failed — starting RPC sender + ``` + Note: the second line's "— starting RPC sender" text is unconditional — it prints even if + `GATEWAY_RPC_SENDER_ENABLED=false` — the sender only actually starts inside the `if (rpcSenderEnabled)` + guard right after. With the env above (`GATEWAY_RPC_SENDER_ENABLED=true`) it does start, confirmed by + `RpcBurstSender` itself: + ``` + RPC burst sender: 12 devices in ... chunks of 500, every 60s, first burst in ...ms (url ...) + ``` + **This device count (12) must equal the full device range**, not just the devices belonging to + gateways that onboarded early — confirming the sender was built from the complete post-ramp device + list, not a partial one. Device mode has no RPC step, so step 5 for device mode is just the single + `STAGGERED device ramp complete: 12 onboarded, 0 failed` line (no RPC sender to start). + +6. **Shutdown** (after `DURATION_IN_SECONDS`): gateway/device telemetry timers cancel, the engine stops, + the RPC burst sender stops, and (gateway+RPC only) the usual drain block runs: + ``` + Gateway RPC drain: waiting for in-flight RPCs to settle (quietSec=5, maxSec=...)... + Gateway RPC drain complete [drained ...s, quiesced=true] + RPC In [total]: publish=... (new ..., redelivered ...) + RPC Out [total]: publish=..., pubAck=..., failed=..., recovered=..., lost=... + ``` + +## 3. Pass/fail checklist + +- [ ] `STAGGERED model prepared: ...` appears once, with the expected entity counts. +- [ ] `Staggered onboarding starting: N entities, maxConcurrent=2, firstJitter=20000ms` — cap and jitter match config. +- [ ] Zero (or explained) `Onboard failed for entity ...` lines. +- [ ] Per-gateway/device telemetry `DEBUG` lines appear at staggered times, not bunched at one instant. +- [ ] `Ramp complete: N onboarded, 0 failed` fires only after step 2's timestamp + roughly the jitter/cap-bounded ramp time — not immediately. +- [ ] `STAGGERED gateway ramp complete: ...` fires immediately after, and `RPC burst sender: devices ...` (if `GATEWAY_RPC_SENDER_ENABLED=true`) shows the **complete** device range, proving the sender started from the full post-ramp list and only after ramp-complete (there is no earlier `RPC burst sender: ...` line anywhere above it in the log). +- [ ] Drain + `RPC In [total]` / `RPC Out [total]` lines appear at shutdown (RPC runs only). + +## 4. Known gaps to account for when reading the log (not smoke-test failures) + +- **No periodic `Connections [window Ns]: live=.../...` line during STAGGERED.** `registerConnectionStats()` + is only called on the PHASED connect path; STAGGERED's `connectGateways()`/`connectDevices()` return + early before reaching it (by design — see Task 3/4 notes: the fixed-fleet connections gauge doesn't fit + a paced ramp). Don't wait for this line; it will not appear. +- **Gateway + `GATEWAY_RPC_ENABLED=true`: the periodic `RPC Subscription` / `RPC In` / `RPC Out` / + `Gateway device announce` interval lines do not appear during a STAGGERED run.** `runStaggeredApiTests()` + calls `statsReporter().start()` *before* `initRpcReceiver()` registers those four blocks + (`MqttGatewayAPITest.java`, `statsReporter().start()` a few lines above the `if (rpcEnabled) + { initRpcReceiver(); }` that follows it). `StatsReporter.start()` checks `sources.isEmpty()` and, finding + it empty at that point, logs `Stats logging: no active sources for this run` and returns without ever + scheduling the periodic task — and nothing calls `start()` again afterward, so the four blocks registered + moments later are silently never reported for the rest of the run. **This looks like a real ordering + defect** (unlike the connections-gauge gap above, these blocks clearly are meant to report — the + `register()` calls exist), reported separately rather than fixed here per this task's scope (verification + only, no production-code changes). It does **not** lose data: the one-shot drain-time + `RPC In [total]` / `RPC Out [total]` lines (step 5/6 above) are logged directly, not through + `StatsReporter`, so end-of-run totals are still correct — only the per-`STATS_LOG_INTERVAL_SEC` snapshots + during the run are missing. If you need those during the run, that's a sign the ordering bug needs + fixing, not that your run is broken. From ea124c2b4b6bf778e3c1ee57ebf673a4fbf872e1 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 17:46:41 +0300 Subject: [PATCH 08/13] fix(onboarding): register STAGGERED RPC/announce stats sources before starting the reporter --- .../tools/service/gateway/MqttGatewayAPITest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java index 1a7e4a8..f343b05 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java @@ -397,10 +397,15 @@ public void runApiTests() throws InterruptedException { * {@code finally} above. */ private void runStaggeredApiTests() throws InterruptedException { - statsReporter().start(); + // Register stats sources BEFORE starting the reporter — StatsReporter.start() snapshots + // sources.isEmpty() once at call time; if it's empty then, it logs "no active sources" and never + // schedules, so a source registered afterward would never print periodically (matches PHASED's + // order: connectGateways()'s attachRpcReceiver()/initRpcReceiver() always runs, via connectGateways(), + // before runApiTests() reaches statsReporter().start()). if (rpcEnabled) { initRpcReceiver(); } + statsReporter().start(); onboardingEngine = new StaggeredOnboardingEngine( gatewayLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed); onboardingEngine.start((onboarded, failed) -> { From 030d79dd314ac9b0e0ad72db1b24cef31b7611aa Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 18:05:39 +0300 Subject: [PATCH 09/13] fix(onboarding): STAGGERED throughput stats, ramp-timeout warning, engagement-neutral comments, regression tests --- README.md | 3 +- STAGGERED-ONBOARDING-RUNBOOK.md | 48 +++++++++++------ .../service/device/MqttDeviceAPITest.java | 29 +++++++++-- .../service/gateway/MqttGatewayAPITest.java | 49 ++++++++++++++---- .../gateway/MqttGatewayAPITestTest.java | 51 +++++++++++++++++++ .../StaggeredOnboardingEngineTest.java | 26 ++++++++++ 6 files changed, 176 insertions(+), 30 deletions(-) diff --git a/README.md b/README.md index 2b8b638..44ecde8 100644 --- a/README.md +++ b/README.md @@ -156,4 +156,5 @@ onboarding strategies, controlled by `ONBOARD_MODE`: combination fails fast at startup with a clear error instead of silently behaving like `PHASED`. See [`STAGGERED-ONBOARDING-RUNBOOK.md`](./STAGGERED-ONBOARDING-RUNBOOK.md) for a smoke-test procedure -(exact env + the expected log sequence) and known observability gaps in this mode. +(exact env + the expected log sequence) and the one remaining observability limitation in this mode (no +live connections gauge during the ramp). diff --git a/STAGGERED-ONBOARDING-RUNBOOK.md b/STAGGERED-ONBOARDING-RUNBOOK.md index c679b28..5626775 100644 --- a/STAGGERED-ONBOARDING-RUNBOOK.md +++ b/STAGGERED-ONBOARDING-RUNBOOK.md @@ -130,6 +130,16 @@ All lines below are `INFO` unless noted; `%` placeholders are the actual `{}` sl RPC In [total]: publish=... (new ..., redelivered ...) RPC Out [total]: publish=..., pubAck=..., failed=..., recovered=..., lost=... ``` + If `DURATION_IN_SECONDS` is too short for the ramp to finish (e.g. `ONBOARD_FIRST_JITTER_SEC` + + cap-bounded ramp time exceeds it), shutdown instead starts with a `WARN` naming the cause — not every + gateway/device onboarded, and (gateway mode) the RPC sender never started: + ``` + STAGGERED: test.duration (90s) elapsed before the onboarding ramp completed — not every gateway may + have onboarded, and the RPC sender (if enabled) never started. Consider raising DURATION_IN_SECONDS or + lowering ONBOARD_MAX_CONCURRENT/ONBOARD_FIRST_JITTER_SEC. + ``` + Not expected in this runbook's env (90s duration comfortably exceeds the 20s jitter + ramp time for 6 + gateways at cap 2) — if you see it here, raise `DURATION_IN_SECONDS`. ## 3. Pass/fail checklist @@ -139,6 +149,9 @@ All lines below are `INFO` unless noted; `%` placeholders are the actual `{}` sl - [ ] Per-gateway/device telemetry `DEBUG` lines appear at staggered times, not bunched at one instant. - [ ] `Ramp complete: N onboarded, 0 failed` fires only after step 2's timestamp + roughly the jitter/cap-bounded ramp time — not immediately. - [ ] `STAGGERED gateway ramp complete: ...` fires immediately after, and `RPC burst sender: devices ...` (if `GATEWAY_RPC_SENDER_ENABLED=true`) shows the **complete** device range, proving the sender started from the full post-ramp list and only after ramp-complete (there is no earlier `RPC burst sender: ...` line anywhere above it in the log). +- [ ] No `STAGGERED: test.duration (...) elapsed before the onboarding ramp completed` `WARN` line (it should only appear if `DURATION_IN_SECONDS` is too short for the ramp — not expected with this env's settings). +- [ ] Periodic `Throughput [window Ns]: publishOk=..., publishFail=..., ~N msg/s ...` lines appear at each `STATS_LOG_INTERVAL_SEC` tick (both gateway and device mode). +- [ ] Gateway + `GATEWAY_RPC_ENABLED=true`: periodic `RPC Subscription`/`RPC In`/`RPC Out`/`Gateway device announce` lines also appear at each `STATS_LOG_INTERVAL_SEC` tick (not just at shutdown). - [ ] Drain + `RPC In [total]` / `RPC Out [total]` lines appear at shutdown (RPC runs only). ## 4. Known gaps to account for when reading the log (not smoke-test failures) @@ -146,19 +159,22 @@ All lines below are `INFO` unless noted; `%` placeholders are the actual `{}` sl - **No periodic `Connections [window Ns]: live=.../...` line during STAGGERED.** `registerConnectionStats()` is only called on the PHASED connect path; STAGGERED's `connectGateways()`/`connectDevices()` return early before reaching it (by design — see Task 3/4 notes: the fixed-fleet connections gauge doesn't fit - a paced ramp). Don't wait for this line; it will not appear. -- **Gateway + `GATEWAY_RPC_ENABLED=true`: the periodic `RPC Subscription` / `RPC In` / `RPC Out` / - `Gateway device announce` interval lines do not appear during a STAGGERED run.** `runStaggeredApiTests()` - calls `statsReporter().start()` *before* `initRpcReceiver()` registers those four blocks - (`MqttGatewayAPITest.java`, `statsReporter().start()` a few lines above the `if (rpcEnabled) - { initRpcReceiver(); }` that follows it). `StatsReporter.start()` checks `sources.isEmpty()` and, finding - it empty at that point, logs `Stats logging: no active sources for this run` and returns without ever - scheduling the periodic task — and nothing calls `start()` again afterward, so the four blocks registered - moments later are silently never reported for the rest of the run. **This looks like a real ordering - defect** (unlike the connections-gauge gap above, these blocks clearly are meant to report — the - `register()` calls exist), reported separately rather than fixed here per this task's scope (verification - only, no production-code changes). It does **not** lose data: the one-shot drain-time - `RPC In [total]` / `RPC Out [total]` lines (step 5/6 above) are logged directly, not through - `StatsReporter`, so end-of-run totals are still correct — only the per-`STATS_LOG_INTERVAL_SEC` snapshots - during the run are missing. If you need those during the run, that's a sign the ordering bug needs - fixing, not that your run is broken. + a paced ramp). Don't wait for this line; it will not appear. **This is the only remaining known gap** — + it is a deliberate scope boundary, not a defect, and there is no plan to close it (a live-ramp connection + gauge would need its own design, not a reuse of the fixed-fleet one). + +Two items that used to be listed here have been fixed and no longer apply: + +- ~~Gateway `RPC Subscription`/`RPC In`/`RPC Out`/`Gateway device announce` periodic lines don't + appear~~ — **fixed** (commit `ea124c2`). `runStaggeredApiTests()` now calls `initRpcReceiver()` (which + registers those four blocks) *before* `statsReporter().start()`, matching PHASED's order. The periodic + lines print normally now; re-run the smoke test and confirm you see them at each `STATS_LOG_INTERVAL_SEC` + tick when `GATEWAY_RPC_ENABLED=true`. +- ~~No periodic `Throughput [window Ns]: ...` line~~ — **fixed**. Both STAGGERED paths now register + `StatsBlock.THROUGHPUT` the same way `AbstractAPITest.runApiTests(int)` does for PHASED (guarded by + `MESSAGES_PER_SECOND > 0`), before `statsReporter().start()`. This was actually the more serious of the + two gaps for direct-device mode: device STAGGERED registers no other stats block, so before this fix + `statsReporter().start()` found an empty source map, logged `Stats logging: no active sources for this + run`, and the reporter stayed inert for the entire run — no periodic output of any kind. Confirm the + `Throughput [window Ns]: publishOk=..., publishFail=..., ~N msg/s ...` line now appears periodically in + both gateway and device STAGGERED runs with `MESSAGES_PER_SECOND > 0`. diff --git a/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java b/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java index 666e35b..cbc75ae 100644 --- a/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/device/MqttDeviceAPITest.java @@ -29,6 +29,8 @@ import org.thingsboard.tools.service.msg.Msg; import org.thingsboard.tools.service.mqtt.DeviceClient; import org.thingsboard.tools.service.shared.BaseMqttAPITest; +import org.thingsboard.tools.service.shared.StatsBlock; +import org.thingsboard.tools.service.shared.ThroughputStats; import org.thingsboard.tools.service.shared.onboarding.EntityLifecycle; import org.thingsboard.tools.service.shared.onboarding.StaggeredOnboardingEngine; @@ -42,6 +44,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -102,18 +105,36 @@ public void runApiTests() throws InterruptedException { /** * STAGGERED: ramp devices in through the engine (each onboard connects and schedules its own * telemetry timer — see {@link #connectAndScheduleDevice(int)}), then hold for the test duration. - * Direct-device mode has no RPC subscribe step (case-a is gateway-based) — STAGGERED here is + * Direct-device mode has no RPC subscribe step (RPC is a gateway-mode concept) — STAGGERED here is * telemetry-only, matching what {@link MqttDeviceAPITest} already supports in PHASED. */ private void runStaggeredApiTests() throws InterruptedException { + // Register the THROUGHPUT stats source BEFORE starting the reporter (same order + same block + // AbstractAPITest.runApiTests(int) uses for PHASED) — StatsReporter.start() snapshots + // sources.isEmpty() once at call time and never reschedules if it was empty then, so without this + // the reporter would log "no active sources" and stay inert for the whole run (direct-device + // STAGGERED registers nothing else). + if (testMessagesPerSecond > 0) { + statsReporter().register(StatsBlock.THROUGHPUT, + new ThroughputStats(totalSuccessPublishedCount, totalFailedPublishedCount)::summaryAndReset); + } statsReporter().start(); + AtomicBoolean rampCompleted = new AtomicBoolean(false); onboardingEngine = new StaggeredOnboardingEngine( deviceLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed); - onboardingEngine.start((onboarded, failed) -> - log.info("STAGGERED device ramp complete: {} onboarded, {} failed", onboarded, failed)); + onboardingEngine.start((onboarded, failed) -> { + rampCompleted.set(true); + log.info("STAGGERED device ramp complete: {} onboarded, {} failed", onboarded, failed); + }); try { Thread.sleep(testDurationInSec * 1000L); } finally { + if (!rampCompleted.get()) { + log.warn("STAGGERED: test.duration ({}s) elapsed before the onboarding ramp completed — " + + "not every device may have onboarded or started publishing telemetry. " + + "Consider raising DURATION_IN_SECONDS or lowering ONBOARD_MAX_CONCURRENT/ONBOARD_FIRST_JITTER_SEC.", + testDurationInSec); + } for (ScheduledFuture timer : deviceTelemetryTimers) { timer.cancel(false); } @@ -239,7 +260,7 @@ void scheduleDeviceTelemetry(int devIdx, MqttClient client, String deviceName) { * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it. * {@link #scheduleDeviceTelemetry} always publishes one plain per-device message and never injects an * alarm (needs {@code test.alarms.aps <= 0}). Deliberately narrow: STAGGERED currently supports - * exactly the no-alarms combination case-a runs. + * exactly the no-alarms scenario this mode targets. */ void checkStaggeredSupported() { if (alarmsPerSecond > 0) { diff --git a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java index f343b05..5e890b2 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITest.java @@ -38,6 +38,7 @@ import org.thingsboard.tools.service.mqtt.DeviceClient; import org.thingsboard.tools.service.shared.BaseMqttAPITest; import org.thingsboard.tools.service.shared.StatsBlock; +import org.thingsboard.tools.service.shared.ThroughputStats; import org.thingsboard.tools.service.shared.onboarding.EntityLifecycle; import org.thingsboard.tools.service.shared.onboarding.StaggeredOnboardingEngine; @@ -53,6 +54,7 @@ import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; @@ -172,7 +174,8 @@ protected boolean isInboundHandlingEnabled() { // Gateway client -> its sub-device names, for re-announcing sub-devices on reconnect. // ConcurrentHashMap (not HashMap): populated on the main thread before start, then read from reconnect // callbacks on netty event-loop threads — safe-publish without relying on incidental happens-before. - private final Map> gatewayDeviceNames = new ConcurrentHashMap<>(); + // Package-private (not private): MqttGatewayAPITestTest asserts on this directly (commit-on-success-only). + final Map> gatewayDeviceNames = new ConcurrentHashMap<>(); @PostConstruct @@ -296,7 +299,8 @@ void prepareStaggeredModel() { * Fails fast on an unsupported STAGGERED configuration, instead of silently diverging from it. * {@link #scheduleGatewayTelemetry} always publishes a whole-gateway batch (needs {@code * gateway.batch=true}) and never injects an alarm (needs {@code test.alarms.aps <= 0}). Deliberately - * narrow: STAGGERED currently supports exactly the combination case-a runs. + * narrow: STAGGERED currently supports exactly the persistent gateway-batch, no-alarms scenario this + * mode targets. */ void checkStaggeredSupported() { if (!gatewayBatchEnabled) { @@ -405,10 +409,19 @@ private void runStaggeredApiTests() throws InterruptedException { if (rpcEnabled) { initRpcReceiver(); } + // Same THROUGHPUT registration AbstractAPITest.runApiTests(int) does for PHASED — otherwise + // STAGGERED's per-gateway telemetry timers (which feed the same totalSuccess/totalFailed counters) + // have no periodic reporter block at all. + if (testMessagesPerSecond > 0) { + statsReporter().register(StatsBlock.THROUGHPUT, + new ThroughputStats(totalSuccessPublishedCount, totalFailedPublishedCount)::summaryAndReset); + } statsReporter().start(); + AtomicBoolean rampCompleted = new AtomicBoolean(false); onboardingEngine = new StaggeredOnboardingEngine( gatewayLifecycle(), onboardMaxConcurrent, onboardFirstJitterSec, /*schedulerThreads*/ 2, seed); onboardingEngine.start((onboarded, failed) -> { + rampCompleted.set(true); log.info("STAGGERED gateway ramp complete: {} onboarded, {} failed — starting RPC sender", onboarded, failed); if (rpcSenderEnabled) { startRpcBurstSender(); // existing method, unchanged: full device list @@ -417,6 +430,12 @@ private void runStaggeredApiTests() throws InterruptedException { try { Thread.sleep(testDurationInSec * 1000L); } finally { + if (!rampCompleted.get()) { + log.warn("STAGGERED: test.duration ({}s) elapsed before the onboarding ramp completed — " + + "not every gateway may have onboarded, and the RPC sender (if enabled) never started. " + + "Consider raising DURATION_IN_SECONDS or lowering ONBOARD_MAX_CONCURRENT/ONBOARD_FIRST_JITTER_SEC.", + testDurationInSec); + } for (ScheduledFuture timer : gatewayTelemetryTimers) { timer.cancel(false); } @@ -474,13 +493,8 @@ public void onboard(int idx) throws Exception { * subscribe+reconnect wiring {@link #attachClientsToRpc} performs for PHASED's bulk attach — in that * order (connect -> announce -> subscribe), per the {@link EntityLifecycle#onboard} contract. * Synchronous: throws on any failure so the engine counts this gateway as failed rather than - * onboarded. - *

Nothing is registered into the shared {@code mqttClients}/{@code deviceClients}/ - * {@code gatewayDeviceNames}/{@code clientNames} collections — which {@link #startRpcBurstSender()}, - * the telemetry scheduler, and reconnect recovery all read from wholesale — until the ENTIRE sequence - * below has succeeded. A gateway that fails partway (e.g. its subscribe throws after announce - * succeeded) is closed and left out of every shared collection entirely, so a partial onboarding can - * never leak un-announced/un-subscribed devices into the RPC-outcome measurement. + * onboarded. The connect step is isolated here; everything after it (the part with a commit-on-success + * invariant to preserve) lives in {@link #onboardConnectedGateway}. */ private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception { int localIdx = gwIdx - gatewayStartIdx; @@ -489,6 +503,23 @@ private void connectAnnounceSubscribeAndSchedule(int gwIdx) throws Exception { // 1) connect (persistent; createClient() applies autoReconnect() same as every other gateway client) MqttClient client = initClientBlocking(gatewayName); + onboardConnectedGateway(gwIdx, client, gatewayName, deviceNames); + } + + /** + * Everything after "the client is already connected": announce -> subscribe -> commit-on-success -> + * schedule telemetry. Split out of {@link #connectAnnounceSubscribeAndSchedule} purely so the + * commit-on-success invariant is unit-testable with a mocked, already-"connected" {@link MqttClient} + * (no real broker needed) — the production call site above still invokes this immediately after a + * real connect, so behavior is unchanged. + *

Nothing is registered into the shared {@code mqttClients}/{@code deviceClients}/ + * {@code gatewayDeviceNames}/{@code clientNames} collections — which {@link #startRpcBurstSender()}, + * the telemetry scheduler, and reconnect recovery all read from wholesale — until the ENTIRE sequence + * below has succeeded. A gateway that fails partway (e.g. its subscribe throws after announce + * succeeded) is closed and left out of every shared collection entirely, so a partial onboarding can + * never leak un-announced/un-subscribed devices into the RPC-outcome measurement. + */ + void onboardConnectedGateway(int gwIdx, MqttClient client, String gatewayName, List deviceNames) throws Exception { try { // 2) announce sub-devices through the same reliable (RPC on) / legacy QoS-0 (RPC off) path // warm-up uses. No timeout here: an announce under retry can legitimately take longer than diff --git a/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java index db0beeb..83fc9b5 100644 --- a/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java +++ b/src/test/java/org/thingsboard/tools/service/gateway/MqttGatewayAPITestTest.java @@ -15,6 +15,8 @@ */ package org.thingsboard.tools.service.gateway; +import io.netty.util.concurrent.ImmediateEventExecutor; +import io.netty.util.concurrent.Promise; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; @@ -172,4 +174,53 @@ void checkStaggeredSupportedPassesForTheSupportedCombination() { alarmsPerSecond = 0; checkStaggeredSupported(); // must not throw } + + // --- onboardConnectedGateway(): commit-on-success-only (no partial-onboarding leak) --- + + @Test + void midOnboardFailureLeavesNoEntryInSharedCollections() { + gatewayStartIdx = 100; + gatewayEndIdx = 101; // single gateway + + MqttClient client = mock(MqttClient.class); + // Simulate a mid-onboard failure: the connect step already "succeeded" (this test starts past + // it, with an already-"connected" mock client — see onboardConnectedGateway's javadoc), but the + // announce step's publish never confirms. + Promise failedAnnounce = ImmediateEventExecutor.INSTANCE.newPromise(); + failedAnnounce.setFailure(new RuntimeException("simulated announce failure")); + when(client.publish(any(), any(), any())).thenReturn(failedAnnounce); + + assertThatThrownBy(() -> + onboardConnectedGateway(100, client, "GW00000100", List.of("DW00000000"))) + .isInstanceOf(Exception.class); + + // The defect this guards: a gateway that fails partway must leave NO trace in any of the shared + // collections startRpcBurstSender()/the telemetry scheduler/reconnect recovery read from wholesale + // — otherwise a partially-onboarded gateway's un-announced devices would contaminate the RPC + // target list. + assertThat(mqttClients).isEmpty(); + assertThat(deviceClients).isEmpty(); + assertThat(gatewayDeviceNames).isEmpty(); + assertThat(gatewayTelemetryTimers).isEmpty(); + verify(client).disconnect(); + } + + @Test + void fullyOnboardedGatewayCommitsAllThreeCollections() throws Exception { + gatewayStartIdx = 200; + gatewayEndIdx = 201; + testMessagesPerSecond = 0; // keep this test focused on the commit, not the telemetry timer + + MqttClient client = mock(MqttClient.class); + Promise ackedAnnounce = ImmediateEventExecutor.INSTANCE.newPromise(); + ackedAnnounce.setSuccess(null); + when(client.publish(any(), any(), any())).thenReturn(ackedAnnounce); + + onboardConnectedGateway(200, client, "GW00000200", List.of("DW00000000", "DW00000001")); + + assertThat(mqttClients).containsExactly(client); + assertThat(deviceClients).hasSize(2); + assertThat(gatewayDeviceNames.get(client)).containsExactly("DW00000000", "DW00000001"); + verify(client, never()).disconnect(); + } } diff --git a/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java index c1e74a6..a628372 100644 --- a/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java +++ b/src/test/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngineTest.java @@ -113,4 +113,30 @@ void firesRampCompleteImmediatelyForZeroEntities() throws Exception { assertThat(complete.await(2, TimeUnit.SECONDS)).isTrue(); engine.stop(); } + + /** + * Guards the ramp-complete-fires-exactly-once contract with a counter, not a {@code + * CountDownLatch(1)}: a latch's {@code countDown()} is a silent no-op once it's already at zero, so a + * test that only awaits the latch would pass even if the callback fired twice. An {@link + * AtomicInteger}, checked after giving a hypothetical duplicate a grace window to land, actually + * catches a double-fire. + */ + @Test + void rampCompleteFiresExactlyOnce() throws Exception { + StubLifecycle stub = new StubLifecycle(50, 0); + AtomicInteger completeCalls = new AtomicInteger(); + CountDownLatch firstComplete = new CountDownLatch(1); + + StaggeredOnboardingEngine engine = + new StaggeredOnboardingEngine(stub, 8, 0, 2, 7L); + engine.start((onboarded, failed) -> { + completeCalls.incrementAndGet(); + firstComplete.countDown(); + }); + + assertThat(firstComplete.await(10, TimeUnit.SECONDS)).isTrue(); + Thread.sleep(200); // grace window: let any hypothetical duplicate fire land before asserting + assertThat(completeCalls.get()).isEqualTo(1); + engine.stop(); + } } From b4c50088f5d4ab853e0b46775739156a439af78e Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 18:22:16 +0300 Subject: [PATCH 10/13] Bump version to 4.0.1-GW-22 --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index 40d51e5..fb9d24d 100644 --- a/pom.xml +++ b/pom.xml @@ -21,7 +21,7 @@ org.thingsboard performance-tests - 4.0.1-GW-21 + 4.0.1-GW-22 jar ThingsBoard Performance Tests From 0d622a2d69ae2e3266d28972229df9220cb8a674 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 18:46:44 +0300 Subject: [PATCH 11/13] fix(onboarding): log RPC stats legend + subscribe summary once, not per gateway under STAGGERED --- .../service/gateway/rpc/GatewayRpcReceiver.java | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java b/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java index 6419444..757eaca 100644 --- a/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java +++ b/src/main/java/org/thingsboard/tools/service/gateway/rpc/GatewayRpcReceiver.java @@ -71,6 +71,10 @@ public class GatewayRpcReceiver { // Clients whose current v1/gateway/rpc subscription is not (yet) SUBACK-confirmed — a live gauge, // so a slow-but-real SUBACK is never a false positive (it self-clears whenever the SUBACK lands). private final java.util.Set unconfirmedSubscriptions = ConcurrentHashMap.newKeySet(); + // The RPC stats legend is one-time ceremony. PHASED calls attach() once (bulk), but STAGGERED calls it + // per gateway (singleton list), so without this guard the multi-line legend would repeat ~once per + // gateway and flood the log during onboarding. Log it exactly once, on the first attach(). + private final AtomicBoolean legendLogged = new AtomicBoolean(false); private static final long DRAIN_POLL_MS = 500L; @@ -98,7 +102,9 @@ public GatewayRpcReceiver(String topic, MqttQoS qos, RpcMessageProcessor process } public void attach(List clients, int packSize) throws InterruptedException { - log.info("Gateway RPC stats key:\n{}", RpcLatencyStats.legend()); + if (legendLogged.compareAndSet(false, true)) { + log.info("Gateway RPC stats key:\n{}", RpcLatencyStats.legend()); + } int n = 0; for (MqttClient client : clients) { subscribe(client); @@ -109,7 +115,12 @@ public void attach(List clients, int packSize) throws InterruptedExc Thread.sleep(100 + ThreadLocalRandom.current().nextInt(100)); } } - log.info("Subscribed {} gateways to RPC topic {}", clients.size(), topic); + // Only log the batch summary for a true bulk attach (PHASED). STAGGERED attaches one gateway at a + // time, so logging here would repeat once per gateway; the STAGGERED ramp-complete line reports the + // total instead. + if (clients.size() > 1) { + log.info("Subscribed {} gateways to RPC topic {}", clients.size(), topic); + } } /** Re-issue the RPC-topic subscription for one client after it reconnects (netty-mqtt clears all From 982b1ba7bdde4d4127e441a9a43197130f7b49e9 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 18:58:30 +0300 Subject: [PATCH 12/13] feat(onboarding): periodic STAGGERED onboarding-progress line every 10s until ramp-complete --- .../onboarding/StaggeredOnboardingEngine.java | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java index e566bb7..b2dbe41 100644 --- a/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java +++ b/src/main/java/org/thingsboard/tools/service/shared/onboarding/StaggeredOnboardingEngine.java @@ -22,6 +22,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -53,6 +54,11 @@ public interface RampCompleteCallback { private final AtomicInteger terminal = new AtomicInteger(); private volatile boolean running; private volatile RampCompleteCallback onComplete; + // Periodic onboarding-progress line (replaces per-entity subscribe/announce log spam under STAGGERED); + // cancelled at ramp-complete so the final "Ramp complete: ..." line closes it out. + private volatile ScheduledFuture progressTask; + + private static final long PROGRESS_LOG_INTERVAL_SEC = 10L; public StaggeredOnboardingEngine(EntityLifecycle lifecycle, int maxConcurrentOnboards, int firstJitterSec, int schedulerThreads, long seed) { @@ -80,6 +86,15 @@ public void start(RampCompleteCallback cb) { long offset = EphemeralSchedule.firstOffsetMillis(rng, firstJitterMillis); timer.schedule(() -> onboardOne(idx), offset, TimeUnit.MILLISECONDS); } + this.progressTask = timer.scheduleAtFixedRate(this::logProgress, + PROGRESS_LOG_INTERVAL_SEC, PROGRESS_LOG_INTERVAL_SEC, TimeUnit.SECONDS); + } + + private void logProgress() { + log.info("STAGGERED onboarding progress: {} / {} onboarded, {} in-flight, {} failed", + String.format(java.util.Locale.US, "%,d", onboarded.get()), + String.format(java.util.Locale.US, "%,d", lifecycle.entityCount()), + inFlightCount(), failed.get()); } private void onboardOne(int idx) { @@ -108,6 +123,10 @@ private void onboardOne(int idx) { } private void fireComplete() { + ScheduledFuture pt = this.progressTask; + if (pt != null) { + pt.cancel(false); + } log.info("Ramp complete: {} onboarded, {} failed", onboarded.get(), failed.get()); RampCompleteCallback cb = this.onComplete; if (cb != null) { @@ -123,4 +142,5 @@ public void stop() { public int onboardedCount() { return onboarded.get(); } public int failedCount() { return failed.get(); } + public int inFlightCount() { return maxConcurrentOnboards - permits.availablePermits(); } } From 5b1143f826fbf5b806beb0bf0e0ebff330c64d07 Mon Sep 17 00:00:00 2001 From: dshvaika Date: Thu, 13 Aug 2026 19:04:31 +0300 Subject: [PATCH 13/13] Bump version to 4.0.1-GW-23 --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index fb9d24d..334bf30 100644 --- a/pom.xml +++ b/pom.xml @@ -21,7 +21,7 @@ org.thingsboard performance-tests - 4.0.1-GW-22 + 4.0.1-GW-23 jar ThingsBoard Performance Tests