fix(telemetry): release a removed driver's resilience tracker state so the publisher stops re-emitting it (#609)
Review R1 M1: StopChild evicted the hub cache, but DriverResilienceStatusPublisherService re-emits the tracker every tick. IDriverCapabilityInvokerFactory.Release (default no-op) now invalidates the pipeline builder + status tracker; the host calls it before Evict. Test runs a publisher tick after removal, with a neighbour instance as positive control. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -255,6 +255,11 @@ driver node**:
|
|||||||
reconnect loses whatever transitions happened while disconnected, same as a DPS resubscribe would.
|
reconnect loses whatever transitions happened while disconnected, same as a DPS resubscribe would.
|
||||||
The cache is evicted per instance (`ITelemetryLocalHub.Evict`) when `DriverHostActor.StopChild` removes a
|
The cache is evicted per instance (`ITelemetryLocalHub.Evict`) when `DriverHostActor.StopChild` removes a
|
||||||
driver from the plan (#609), so a decommissioned instance's last state is not replayed to new subscribers.
|
driver from the plan (#609), so a decommissioned instance's last state is not replayed to new subscribers.
|
||||||
|
`StopChild` first releases the instance's resilience state (`IDriverCapabilityInvokerFactory.Release` →
|
||||||
|
pipeline-builder + `DriverResilienceStatusTracker.Invalidate`): `DriverResilienceStatusPublisherService`
|
||||||
|
re-emits the whole tracker every 5 s, so evicting only the hub was undone within one tick (review R1 M1).
|
||||||
|
Residual: a Health emit already in flight from the stopping child can land after the eviction and
|
||||||
|
re-cache that one snapshot — a narrow race, not a recurring replay.
|
||||||
- **Connection indicator** — the `/alerts` and `/script-log` live/disconnected pill is driven by
|
- **Connection indicator** — the `/alerts` and `/script-log` live/disconnected pill is driven by
|
||||||
**aggregate** stream health across all dialers: connected when **at least one** node stream is
|
**aggregate** stream health across all dialers: connected when **at least one** node stream is
|
||||||
up, disconnected when all are down. This matches today's DPS `SubscribeAck`/`PostStop` pill
|
up, disconnected when all are down. This matches today's DPS `SubscribeAck`/`PostStop` pill
|
||||||
|
|||||||
@@ -28,6 +28,16 @@ public interface IDriverCapabilityInvokerFactory
|
|||||||
/// </param>
|
/// </param>
|
||||||
/// <returns>A per-instance <see cref="IDriverCapabilityInvoker"/>.</returns>
|
/// <returns>A per-instance <see cref="IDriverCapabilityInvoker"/>.</returns>
|
||||||
IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null);
|
IDriverCapabilityInvoker Create(string driverInstanceId, string driverType, string? resilienceConfigJson = null);
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Release every piece of shared resilience state held for <paramref name="driverInstanceId"/> — called
|
||||||
|
/// by the host when the instance leaves the deployed plan (#609). A concrete factory drops the
|
||||||
|
/// instance's cached pipelines and its status-tracker snapshots, so the periodic resilience publisher
|
||||||
|
/// stops re-emitting (and re-priming new telemetry subscribers with) a decommissioned driver's last
|
||||||
|
/// breaker/failure counters. Default: no-op (a factory that holds no state).
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="driverInstanceId">The driver instance that was removed.</param>
|
||||||
|
void Release(string driverInstanceId) { }
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
|
|||||||
@@ -90,4 +90,16 @@ public sealed class DriverCapabilityInvokerFactory : IDriverCapabilityInvokerFac
|
|||||||
statusTracker: _statusTracker,
|
statusTracker: _statusTracker,
|
||||||
optionsGeneration: generation);
|
optionsGeneration: generation);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <inheritdoc />
|
||||||
|
/// <remarks>
|
||||||
|
/// #609: without this the tracker kept a removed instance's snapshots forever (only <see cref="Create"/>
|
||||||
|
/// invalidated, and a removed instance is never re-created), and the resilience publisher re-emitted
|
||||||
|
/// them every tick — undoing the host's telemetry-hub eviction within one interval.
|
||||||
|
/// </remarks>
|
||||||
|
public void Release(string driverInstanceId)
|
||||||
|
{
|
||||||
|
_builder.Invalidate(driverInstanceId);
|
||||||
|
_statusTracker.Invalidate(driverInstanceId);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+71
@@ -227,6 +227,77 @@ public sealed class DriverHostActorUnreadableArtifactTests : RuntimeActorTestBas
|
|||||||
replayed.ShouldContain("other-node-driver");
|
replayed.ShouldContain("other-node-driver");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// #609 (review R1 M1) — the Resilience half of the eviction must SURVIVE the next resilience-publisher
|
||||||
|
/// tick. <c>DriverResilienceStatusPublisherService</c> re-emits <c>tracker.Snapshot()</c> into the hub
|
||||||
|
/// every interval, so evicting only the hub cache was undone within one tick: the host must also
|
||||||
|
/// release the instance's tracker state. The tick below is the publisher's own body
|
||||||
|
/// (<c>BuildMessages</c> → <c>hub.Emit</c>) minus the DPS publish, run against the REAL factory +
|
||||||
|
/// tracker the Host wires. The neighbour instance is the positive control: it is still emitted by
|
||||||
|
/// the same tick, so an absent drv-1 is not an empty tick.
|
||||||
|
/// </summary>
|
||||||
|
[Fact]
|
||||||
|
public void Removing_a_driver_instance_releases_its_resilience_state_so_the_next_publisher_tick_omits_it()
|
||||||
|
{
|
||||||
|
var db = NewInMemoryDbFactory();
|
||||||
|
var factory = new SubscribingDriverFactory("Modbus");
|
||||||
|
var hub = new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub();
|
||||||
|
var tracker = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker();
|
||||||
|
var invokerFactory = new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverCapabilityInvokerFactory(
|
||||||
|
new ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResiliencePipelineBuilder(statusTracker: tracker),
|
||||||
|
tracker,
|
||||||
|
_ => ZB.MOM.WW.OtOpcUa.Core.Abstractions.DriverTier.A);
|
||||||
|
var coordinator = CreateTestProbe();
|
||||||
|
var actor = Sys.ActorOf(DriverHostActor.Props(
|
||||||
|
db, TestNode, coordinator.Ref,
|
||||||
|
driverFactory: factory,
|
||||||
|
localRoles: new HashSet<string> { "driver" },
|
||||||
|
invokerFactory: invokerFactory,
|
||||||
|
telemetryHub: hub));
|
||||||
|
|
||||||
|
actor.Tell(new DispatchDeployment(SeedV3Deployment(db, RevA), RevA, CorrelationId.NewId()));
|
||||||
|
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||||
|
|
||||||
|
// Resilience counters for the running driver and for an instance this host does not own.
|
||||||
|
tracker.RecordBreakerOpen("drv-1", "dev1", DateTime.UtcNow);
|
||||||
|
tracker.RecordFailure("neighbour-drv", "dev9", DateTime.UtcNow);
|
||||||
|
PublisherTick(tracker, hub);
|
||||||
|
ReplayedResilienceIds(hub).ShouldContain("drv-1"); // control: drv-1 is primed before the removal
|
||||||
|
|
||||||
|
actor.Tell(new DispatchDeployment(SeedDriverlessDeployment(db, RevB), RevB, CorrelationId.NewId()));
|
||||||
|
coordinator.ExpectMsg<ApplyAck>(Timeout).Outcome.ShouldBe(ApplyAckOutcome.Applied);
|
||||||
|
AwaitAssert(
|
||||||
|
() => AskDiagnostics(actor).Drivers.Select(d => d.Name).ShouldNotContain("drv-1"),
|
||||||
|
duration: Timeout);
|
||||||
|
|
||||||
|
// The publisher's next tick, after the removal.
|
||||||
|
PublisherTick(tracker, hub);
|
||||||
|
|
||||||
|
var replayed = ReplayedResilienceIds(hub);
|
||||||
|
replayed.ShouldNotContain("drv-1");
|
||||||
|
replayed.ShouldContain("neighbour-drv");
|
||||||
|
|
||||||
|
static void PublisherTick(
|
||||||
|
ZB.MOM.WW.OtOpcUa.Core.Resilience.DriverResilienceStatusTracker t,
|
||||||
|
ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h)
|
||||||
|
{
|
||||||
|
var now = DateTime.UtcNow;
|
||||||
|
foreach (var e in t.Snapshot())
|
||||||
|
h.Emit(new ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience(new(
|
||||||
|
e.DriverInstanceId, e.HostName, e.Snapshot.IsBreakerOpen, e.Snapshot.ConsecutiveFailures,
|
||||||
|
e.Snapshot.CurrentInFlight, e.Snapshot.LastBreakerOpenUtc, e.Snapshot.LastSampledUtc, now)));
|
||||||
|
}
|
||||||
|
|
||||||
|
static List<string> ReplayedResilienceIds(ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryLocalHub h)
|
||||||
|
{
|
||||||
|
using var sub = h.Subscribe(64);
|
||||||
|
var ids = new List<string>();
|
||||||
|
while (sub.Reader.TryRead(out var i))
|
||||||
|
if (i is ZB.MOM.WW.OtOpcUa.Runtime.Telemetry.TelemetryItem.Resilience r) ids.Add(r.E.DriverInstanceId);
|
||||||
|
return ids;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Spawns the host with the subscribing factory, dispatches <paramref name="deploymentId"/> and
|
/// <summary>Spawns the host with the subscribing factory, dispatches <paramref name="deploymentId"/> and
|
||||||
/// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place.</summary>
|
/// waits for the Applied ACK so the child + the initial SubscribeBulk pass are in place.</summary>
|
||||||
private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply(
|
private (IActorRef Actor, Akka.TestKit.TestProbe Coordinator) SpawnHostAndApply(
|
||||||
|
|||||||
Reference in New Issue
Block a user