Joseph Doherty
bc548e1447
Adds DeploymentStateQuery request/response contracts (Commons), a site-side handler (SiteRuntime), a CommunicationService query method (Communication), and reconciliation in DeploymentService: when a prior record is InProgress or Failed-on-timeout, query the site; if it already holds the target revision hash mark the record Success without re-sending; on query failure fall through to a normal deploy (site-side stale-rejection is the safety net).
906 lines
36 KiB
C#
906 lines
36 KiB
C#
using Akka.Actor;
|
|
using Microsoft.Extensions.Logging;
|
|
using ScadaLink.Commons.Messages.Artifacts;
|
|
using ScadaLink.Commons.Messages.DebugView;
|
|
using ScadaLink.Commons.Messages.Deployment;
|
|
using ScadaLink.Commons.Messages.InboundApi;
|
|
using ScadaLink.Commons.Messages.Instance;
|
|
using ScadaLink.Commons.Messages.Lifecycle;
|
|
using ScadaLink.Commons.Messages.ScriptExecution;
|
|
using ScadaLink.Commons.Types.Enums;
|
|
using ScadaLink.HealthMonitoring;
|
|
using ScadaLink.SiteRuntime.Messages;
|
|
using ScadaLink.SiteRuntime.Persistence;
|
|
using ScadaLink.SiteRuntime.Scripts;
|
|
using ScadaLink.SiteRuntime.Streaming;
|
|
|
|
namespace ScadaLink.SiteRuntime.Actors;
|
|
|
|
/// <summary>
|
|
/// Site-side Deployment Manager — runs as a cluster singleton within the site cluster.
|
|
/// On startup, reads all deployed configs from SQLite and creates Instance Actors
|
|
/// for enabled instances in staggered batches.
|
|
///
|
|
/// Handles: DeployInstance, DisableInstance, EnableInstance, DeleteInstance.
|
|
///
|
|
/// Supervision strategy: OneForOneStrategy with Resume for Instance Actors
|
|
/// so that a single instance failure does not cascade to siblings.
|
|
/// </summary>
|
|
public class DeploymentManagerActor : ReceiveActor, IWithTimers
|
|
{
|
|
private readonly SiteStorageService _storage;
|
|
private readonly ScriptCompilationService _compilationService;
|
|
private readonly SharedScriptLibrary _sharedScriptLibrary;
|
|
private readonly SiteStreamManager? _streamManager;
|
|
private readonly SiteRuntimeOptions _options;
|
|
private readonly ILogger<DeploymentManagerActor> _logger;
|
|
private readonly IActorRef? _dclManager;
|
|
private readonly IActorRef? _replicationActor;
|
|
private readonly ISiteHealthCollector? _healthCollector;
|
|
private readonly IServiceProvider? _serviceProvider;
|
|
private readonly Dictionary<string, IActorRef> _instanceActors = new();
|
|
/// <summary>
|
|
/// Tracks Instance Actors that are terminating as part of a redeployment, keyed by
|
|
/// the terminating actor ref. The buffered command is applied once <see cref="Terminated"/>
|
|
/// confirms the child has fully stopped (SiteRuntime-003).
|
|
/// </summary>
|
|
private readonly Dictionary<IActorRef, PendingRedeploy> _pendingRedeploys = new();
|
|
private int _totalDeployedCount;
|
|
|
|
public ITimerScheduler Timers { get; set; } = null!;
|
|
|
|
public DeploymentManagerActor(
|
|
SiteStorageService storage,
|
|
ScriptCompilationService compilationService,
|
|
SharedScriptLibrary sharedScriptLibrary,
|
|
SiteStreamManager? streamManager,
|
|
SiteRuntimeOptions options,
|
|
ILogger<DeploymentManagerActor> logger,
|
|
IActorRef? dclManager = null,
|
|
IActorRef? replicationActor = null,
|
|
ISiteHealthCollector? healthCollector = null,
|
|
IServiceProvider? serviceProvider = null)
|
|
{
|
|
_storage = storage;
|
|
_compilationService = compilationService;
|
|
_sharedScriptLibrary = sharedScriptLibrary;
|
|
_streamManager = streamManager;
|
|
_options = options;
|
|
_dclManager = dclManager;
|
|
_replicationActor = replicationActor;
|
|
_healthCollector = healthCollector;
|
|
_serviceProvider = serviceProvider;
|
|
_logger = logger;
|
|
|
|
// Lifecycle commands
|
|
Receive<DeployInstanceCommand>(HandleDeploy);
|
|
Receive<DisableInstanceCommand>(HandleDisable);
|
|
Receive<EnableInstanceCommand>(HandleEnable);
|
|
Receive<DeleteInstanceCommand>(HandleDelete);
|
|
|
|
// DeploymentManager-006: query-the-site-before-redeploy idempotency.
|
|
// Central asks for the instance's currently-applied deployment identity
|
|
// before re-sending a deployment whose prior record is stuck InProgress
|
|
// or Failed due to a timeout.
|
|
Receive<DeploymentStateQueryRequest>(HandleDeploymentStateQuery);
|
|
|
|
// WP-33: Handle system-wide artifact deployment
|
|
Receive<DeployArtifactsCommand>(HandleDeployArtifacts);
|
|
|
|
// Debug View — route to Instance Actors
|
|
Receive<SubscribeDebugViewRequest>(RouteDebugViewSubscribe);
|
|
Receive<UnsubscribeDebugViewRequest>(RouteDebugViewUnsubscribe);
|
|
Receive<DebugSnapshotRequest>(RouteDebugSnapshot);
|
|
|
|
// Inbound API Route.To().Call() — route to Instance Actors
|
|
Receive<RouteToCallRequest>(RouteInboundApiCall);
|
|
Receive<RouteToGetAttributesRequest>(RouteInboundApiGetAttributes);
|
|
Receive<RouteToSetAttributesRequest>(RouteInboundApiSetAttributes);
|
|
|
|
// Internal startup messages
|
|
Receive<StartupConfigsLoaded>(HandleStartupConfigsLoaded);
|
|
Receive<StartNextBatch>(HandleStartNextBatch);
|
|
|
|
// Internal enable result
|
|
Receive<EnableResult>(HandleEnableResult);
|
|
|
|
// Internal deploy persistence result
|
|
Receive<DeployPersistenceResult>(HandleDeployPersistenceResult);
|
|
|
|
// Terminated signal — drains a buffered redeployment once the previous
|
|
// Instance Actor has fully stopped (SiteRuntime-003).
|
|
Receive<Terminated>(HandleTerminated);
|
|
}
|
|
|
|
protected override void PreStart()
|
|
{
|
|
base.PreStart();
|
|
_healthCollector?.SetActiveNode(true);
|
|
_logger.LogInformation("DeploymentManagerActor starting — loading deployed configs from SQLite...");
|
|
|
|
// Load all configs asynchronously and pipe to self
|
|
_storage.GetAllDeployedConfigsAsync().ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
return new StartupConfigsLoaded(t.Result, null);
|
|
return new StartupConfigsLoaded([], t.Exception?.GetBaseException().Message);
|
|
}).PipeTo(Self);
|
|
}
|
|
|
|
protected override void PostStop()
|
|
{
|
|
_healthCollector?.SetActiveNode(false);
|
|
base.PostStop();
|
|
}
|
|
|
|
/// <summary>
|
|
/// OneForOneStrategy: Resume on exceptions to preserve Instance Actor state,
|
|
/// Stop only on ActorInitializationException (actor failed to start).
|
|
/// </summary>
|
|
protected override SupervisorStrategy SupervisorStrategy()
|
|
{
|
|
return new OneForOneStrategy(
|
|
maxNrOfRetries: -1,
|
|
withinTimeRange: TimeSpan.FromMinutes(1),
|
|
decider: Decider.From(ex =>
|
|
{
|
|
if (ex is ActorInitializationException)
|
|
{
|
|
_logger.LogError(ex, "Instance Actor failed to initialize, stopping");
|
|
return Directive.Stop;
|
|
}
|
|
|
|
_logger.LogWarning(ex, "Instance Actor threw exception, resuming");
|
|
return Directive.Resume;
|
|
}));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Processes the loaded configs from SQLite and begins staggered Instance Actor creation.
|
|
/// </summary>
|
|
private void HandleStartupConfigsLoaded(StartupConfigsLoaded msg)
|
|
{
|
|
if (msg.Error != null)
|
|
{
|
|
_logger.LogError("Failed to load deployed configs: {Error}", msg.Error);
|
|
return;
|
|
}
|
|
|
|
// Load and compile shared scripts from SQLite before creating Instance Actors
|
|
LoadSharedScriptsFromStorage();
|
|
|
|
var enabledConfigs = msg.Configs.Where(c => c.IsEnabled).ToList();
|
|
_totalDeployedCount = msg.Configs.Count;
|
|
_logger.LogInformation(
|
|
"Loaded {Total} deployed configs ({Enabled} enabled) from SQLite",
|
|
msg.Configs.Count, enabledConfigs.Count);
|
|
UpdateInstanceCounts();
|
|
|
|
if (enabledConfigs.Count == 0)
|
|
return;
|
|
|
|
// Start the first batch immediately
|
|
var batchState = new BatchState(enabledConfigs, 0);
|
|
Self.Tell(new StartNextBatch(batchState));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates Instance Actors in batches with a configurable delay between batches
|
|
/// to prevent reconnection storms on failover.
|
|
/// </summary>
|
|
private void HandleStartNextBatch(StartNextBatch msg)
|
|
{
|
|
var state = msg.State;
|
|
var batchSize = _options.StartupBatchSize;
|
|
var startIdx = state.NextIndex;
|
|
var endIdx = Math.Min(startIdx + batchSize, state.Configs.Count);
|
|
|
|
_logger.LogDebug(
|
|
"Creating Instance Actors batch [{Start}..{End}) of {Total}",
|
|
startIdx, endIdx, state.Configs.Count);
|
|
|
|
for (var i = startIdx; i < endIdx; i++)
|
|
{
|
|
var config = state.Configs[i];
|
|
EnsureDclConnections(config.ConfigJson);
|
|
CreateInstanceActor(config.InstanceUniqueName, config.ConfigJson);
|
|
}
|
|
|
|
UpdateInstanceCounts();
|
|
|
|
// Schedule next batch if there are more, using Timers (IWithTimers)
|
|
if (endIdx < state.Configs.Count)
|
|
{
|
|
var nextState = new BatchState(state.Configs, endIdx);
|
|
Timers.StartSingleTimer(
|
|
"startup-batch",
|
|
new StartNextBatch(nextState),
|
|
TimeSpan.FromMilliseconds(_options.StartupBatchDelayMs));
|
|
}
|
|
else
|
|
{
|
|
_logger.LogInformation(
|
|
"All {Count} Instance Actors created", state.Configs.Count);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles a new deployment: stores config in SQLite, clears previous static overrides,
|
|
/// and creates or replaces the Instance Actor.
|
|
///
|
|
/// Redeployment of an already-running instance must wait for the previous Instance
|
|
/// Actor to fully terminate (including PostStop on its descendants) before the
|
|
/// replacement is created — otherwise <see cref="Context.ActorOf"/> can collide on
|
|
/// the still-registered child name. Instead of guessing with a fixed timer, the
|
|
/// terminating child is watched and the in-flight command is buffered until the
|
|
/// <see cref="Terminated"/> signal arrives.
|
|
/// </summary>
|
|
private void HandleDeploy(DeployInstanceCommand command)
|
|
{
|
|
var instanceName = command.InstanceUniqueName;
|
|
_logger.LogInformation(
|
|
"Deploying instance {Instance}, deploymentId={DeploymentId}",
|
|
instanceName, command.DeploymentId);
|
|
|
|
// Redeployment replaces a running instance. Watch + stop the existing actor
|
|
// and buffer this command until its Terminated signal confirms the child
|
|
// (and its whole subtree) has fully stopped and freed its actor name.
|
|
if (_instanceActors.TryGetValue(instanceName, out var existing))
|
|
{
|
|
_instanceActors.Remove(instanceName);
|
|
_pendingRedeploys[existing] = new PendingRedeploy(command, Sender);
|
|
Context.Watch(existing);
|
|
Context.Stop(existing);
|
|
UpdateInstanceCounts();
|
|
return;
|
|
}
|
|
|
|
// Fresh deployment — no existing actor to replace.
|
|
ApplyDeployment(command, Sender, isRedeploy: false);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Recreates an Instance Actor once its predecessor has fully terminated during a
|
|
/// redeployment, draining the buffered <see cref="DeployInstanceCommand"/>.
|
|
/// </summary>
|
|
private void HandleTerminated(Terminated terminated)
|
|
{
|
|
if (!_pendingRedeploys.Remove(terminated.ActorRef, out var pending))
|
|
return;
|
|
|
|
ApplyDeployment(pending.Command, pending.OriginalSender, isRedeploy: true);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates the Instance Actor, persists the config, and replies to the deployer.
|
|
/// A redeployment is an update of an existing instance, so the deployed-instance
|
|
/// counter is only incremented for genuinely new deployments.
|
|
/// </summary>
|
|
private void ApplyDeployment(DeployInstanceCommand command, IActorRef sender, bool isRedeploy)
|
|
{
|
|
var instanceName = command.InstanceUniqueName;
|
|
|
|
// Ensure DCL connections exist for any data-sourced attributes
|
|
EnsureDclConnections(command.FlattenedConfigurationJson);
|
|
|
|
// Create the Instance Actor immediately
|
|
CreateInstanceActor(instanceName, command.FlattenedConfigurationJson);
|
|
if (!isRedeploy)
|
|
_totalDeployedCount++;
|
|
UpdateInstanceCounts();
|
|
|
|
// Persist to SQLite and clear static overrides asynchronously
|
|
Task.Run(async () =>
|
|
{
|
|
await _storage.StoreDeployedConfigAsync(
|
|
instanceName,
|
|
command.FlattenedConfigurationJson,
|
|
command.DeploymentId,
|
|
command.RevisionHash,
|
|
isEnabled: true);
|
|
|
|
// Static overrides are reset on redeployment per design decision
|
|
await _storage.ClearStaticOverridesAsync(instanceName);
|
|
|
|
// Replicate to standby node
|
|
_replicationActor?.Tell(new ReplicateConfigDeploy(
|
|
instanceName, command.FlattenedConfigurationJson,
|
|
command.DeploymentId, command.RevisionHash, true));
|
|
|
|
return new DeployPersistenceResult(command.DeploymentId, instanceName, true, null, sender);
|
|
}).ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
return t.Result;
|
|
return new DeployPersistenceResult(
|
|
command.DeploymentId, instanceName, false,
|
|
t.Exception?.GetBaseException().Message, sender);
|
|
}).PipeTo(Self);
|
|
|
|
// Reply immediately — deployment is applied (actor is running)
|
|
sender.Tell(new DeploymentStatusResponse(
|
|
command.DeploymentId,
|
|
instanceName,
|
|
DeploymentStatus.Success,
|
|
null,
|
|
DateTimeOffset.UtcNow));
|
|
}
|
|
|
|
private void HandleDeployPersistenceResult(DeployPersistenceResult result)
|
|
{
|
|
if (!result.Success)
|
|
{
|
|
_logger.LogError(
|
|
"Failed to persist deployment {DeploymentId} for {Instance}: {Error}",
|
|
result.DeploymentId, result.InstanceName, result.Error);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Disables an instance: stops the actor and marks as disabled in SQLite.
|
|
/// </summary>
|
|
private void HandleDisable(DisableInstanceCommand command)
|
|
{
|
|
var instanceName = command.InstanceUniqueName;
|
|
|
|
if (_instanceActors.TryGetValue(instanceName, out var actor))
|
|
{
|
|
Context.Stop(actor);
|
|
_instanceActors.Remove(instanceName);
|
|
}
|
|
|
|
UpdateInstanceCounts();
|
|
|
|
var sender = Sender;
|
|
_storage.SetInstanceEnabledAsync(instanceName, false).ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
_replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, false));
|
|
|
|
return new InstanceLifecycleResponse(
|
|
command.CommandId,
|
|
instanceName,
|
|
t.IsCompletedSuccessfully,
|
|
t.Exception?.GetBaseException().Message,
|
|
DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
|
|
_logger.LogInformation("Instance {Instance} disabled", instanceName);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Enables an instance: marks as enabled in SQLite and re-creates the Instance Actor
|
|
/// from the stored config.
|
|
/// </summary>
|
|
private void HandleEnable(EnableInstanceCommand command)
|
|
{
|
|
var instanceName = command.InstanceUniqueName;
|
|
var sender = Sender;
|
|
|
|
Task.Run(async () =>
|
|
{
|
|
await _storage.SetInstanceEnabledAsync(instanceName, true);
|
|
_replicationActor?.Tell(new ReplicateConfigSetEnabled(instanceName, true));
|
|
var configs = await _storage.GetAllDeployedConfigsAsync();
|
|
var config = configs.FirstOrDefault(c => c.InstanceUniqueName == instanceName);
|
|
return new EnableResult(command, config, null, sender);
|
|
}).ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
return t.Result;
|
|
return new EnableResult(command, null, t.Exception?.GetBaseException().Message, sender);
|
|
}).PipeTo(Self);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Processes enable result in the actor context (thread-safe).
|
|
/// </summary>
|
|
private void HandleEnableResult(EnableResult result)
|
|
{
|
|
var instanceName = result.Command.InstanceUniqueName;
|
|
|
|
if (result.Error != null || result.Config == null)
|
|
{
|
|
var error = result.Error ?? $"No deployed config found for {instanceName}";
|
|
result.OriginalSender.Tell(new InstanceLifecycleResponse(
|
|
result.Command.CommandId, instanceName, false, error, DateTimeOffset.UtcNow));
|
|
return;
|
|
}
|
|
|
|
if (!_instanceActors.ContainsKey(instanceName))
|
|
{
|
|
CreateInstanceActor(instanceName, result.Config.ConfigJson);
|
|
}
|
|
UpdateInstanceCounts();
|
|
|
|
result.OriginalSender.Tell(new InstanceLifecycleResponse(
|
|
result.Command.CommandId, instanceName, true, null, DateTimeOffset.UtcNow));
|
|
|
|
_logger.LogInformation("Instance {Instance} enabled", instanceName);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Deletes an instance: stops the actor and removes config from SQLite.
|
|
/// Note: store-and-forward messages are NOT cleared per design decision.
|
|
/// </summary>
|
|
private void HandleDelete(DeleteInstanceCommand command)
|
|
{
|
|
var instanceName = command.InstanceUniqueName;
|
|
|
|
if (_instanceActors.TryGetValue(instanceName, out var actor))
|
|
{
|
|
Context.Stop(actor);
|
|
_instanceActors.Remove(instanceName);
|
|
}
|
|
_totalDeployedCount = Math.Max(0, _totalDeployedCount - 1);
|
|
UpdateInstanceCounts();
|
|
|
|
var sender = Sender;
|
|
_storage.RemoveDeployedConfigAsync(instanceName).ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
_replicationActor?.Tell(new ReplicateConfigRemove(instanceName));
|
|
|
|
return new InstanceLifecycleResponse(
|
|
command.CommandId,
|
|
instanceName,
|
|
t.IsCompletedSuccessfully,
|
|
t.Exception?.GetBaseException().Message,
|
|
DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
|
|
_logger.LogInformation("Instance {Instance} deleted", instanceName);
|
|
}
|
|
|
|
/// <summary>
|
|
/// DeploymentManager-006: answers a central query for the instance's
|
|
/// currently-applied deployment identity. The site's deployed-config store
|
|
/// (SQLite) is the authoritative record — it covers both enabled and
|
|
/// disabled instances, and survives node restart/failover. If the instance
|
|
/// has no stored config, the response reports <c>IsDeployed = false</c> with
|
|
/// null identity so central falls through to a normal deploy.
|
|
/// </summary>
|
|
private void HandleDeploymentStateQuery(DeploymentStateQueryRequest request)
|
|
{
|
|
var sender = Sender;
|
|
var instanceName = request.InstanceUniqueName;
|
|
|
|
_storage.GetAllDeployedConfigsAsync().ContinueWith(t =>
|
|
{
|
|
if (!t.IsCompletedSuccessfully)
|
|
{
|
|
_logger.LogError(
|
|
t.Exception?.GetBaseException(),
|
|
"Failed to read deployed configs for deployment state query of {Instance}",
|
|
instanceName);
|
|
// Treat a storage read failure as "unknown" — central falls
|
|
// through to a normal deploy and relies on site-side
|
|
// stale-rejection as the safety net.
|
|
return new DeploymentStateQueryResponse(
|
|
request.CorrelationId, instanceName, false, null, null, DateTimeOffset.UtcNow);
|
|
}
|
|
|
|
var config = t.Result.FirstOrDefault(c => c.InstanceUniqueName == instanceName);
|
|
return config == null
|
|
? new DeploymentStateQueryResponse(
|
|
request.CorrelationId, instanceName, false, null, null, DateTimeOffset.UtcNow)
|
|
: new DeploymentStateQueryResponse(
|
|
request.CorrelationId, instanceName, true,
|
|
config.DeploymentId, config.RevisionHash, DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
}
|
|
|
|
// ── DCL connection management ──
|
|
|
|
private readonly HashSet<string> _createdConnections = new();
|
|
|
|
/// <summary>
|
|
/// Sets up DCL connections from the flattened config (idempotent: tracks created connections).
|
|
/// </summary>
|
|
private void EnsureDclConnections(string configJson)
|
|
{
|
|
if (_dclManager == null) return;
|
|
|
|
try
|
|
{
|
|
var config = System.Text.Json.JsonSerializer.Deserialize<Commons.Types.Flattening.FlattenedConfiguration>(configJson);
|
|
if (config?.Connections == null) return;
|
|
|
|
foreach (var (name, connConfig) in config.Connections)
|
|
{
|
|
if (_createdConnections.Contains(name))
|
|
continue;
|
|
|
|
var primaryDetails = FlattenConnectionConfig(connConfig.Protocol, connConfig.ConfigurationJson);
|
|
var backupDetails = string.IsNullOrEmpty(connConfig.BackupConfigurationJson)
|
|
? null
|
|
: FlattenConnectionConfig(connConfig.Protocol, connConfig.BackupConfigurationJson);
|
|
|
|
_dclManager.Tell(new Commons.Messages.DataConnection.CreateConnectionCommand(
|
|
name, connConfig.Protocol, primaryDetails, backupDetails, connConfig.FailoverRetryCount));
|
|
|
|
_createdConnections.Add(name);
|
|
_logger.LogInformation(
|
|
"Created DCL connection {Connection} (protocol={Protocol})",
|
|
name, connConfig.Protocol);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Failed to parse flattened config for DCL connections");
|
|
}
|
|
}
|
|
|
|
private static IDictionary<string, string> FlattenConnectionConfig(string protocol, string? json)
|
|
{
|
|
if (string.IsNullOrEmpty(json))
|
|
return new Dictionary<string, string>();
|
|
|
|
if (string.Equals(protocol, "OpcUa", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
var (config, _) = Commons.Serialization.OpcUaEndpointConfigSerializer.Deserialize(json);
|
|
return Commons.Serialization.OpcUaEndpointConfigSerializer.ToFlatDict(config);
|
|
}
|
|
|
|
// Fallback: assume legacy flat-dict shape for any future / unknown protocol.
|
|
try
|
|
{
|
|
var dict = new Dictionary<string, string>();
|
|
using var doc = System.Text.Json.JsonDocument.Parse(json);
|
|
foreach (var prop in doc.RootElement.EnumerateObject())
|
|
dict[prop.Name] = prop.Value.ToString();
|
|
return dict;
|
|
}
|
|
catch
|
|
{
|
|
return new Dictionary<string, string>();
|
|
}
|
|
}
|
|
|
|
// ── Shared Script Loading ──
|
|
|
|
private void LoadSharedScriptsFromStorage()
|
|
{
|
|
try
|
|
{
|
|
var scripts = _storage.GetAllSharedScriptsAsync().GetAwaiter().GetResult();
|
|
var compiled = 0;
|
|
foreach (var script in scripts)
|
|
{
|
|
if (_sharedScriptLibrary.CompileAndRegister(script.Name, script.Code))
|
|
compiled++;
|
|
}
|
|
_logger.LogInformation(
|
|
"Loaded {Compiled}/{Total} shared scripts from SQLite",
|
|
compiled, scripts.Count);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Failed to load shared scripts from SQLite");
|
|
}
|
|
}
|
|
|
|
// ── Debug View routing ──
|
|
|
|
private void RouteDebugViewSubscribe(SubscribeDebugViewRequest request)
|
|
{
|
|
if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
instanceActor.Forward(request);
|
|
}
|
|
else
|
|
{
|
|
_logger.LogWarning(
|
|
"Debug view subscribe for unknown instance {Instance}", request.InstanceUniqueName);
|
|
Sender.Tell(new DebugViewSnapshot(
|
|
request.InstanceUniqueName, Array.Empty<Commons.Messages.Streaming.AttributeValueChanged>(),
|
|
Array.Empty<Commons.Messages.Streaming.AlarmStateChanged>(), DateTimeOffset.UtcNow));
|
|
}
|
|
}
|
|
|
|
private void RouteDebugViewUnsubscribe(UnsubscribeDebugViewRequest request)
|
|
{
|
|
if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
instanceActor.Forward(request);
|
|
}
|
|
}
|
|
|
|
private void RouteDebugSnapshot(DebugSnapshotRequest request)
|
|
{
|
|
if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
instanceActor.Forward(request);
|
|
}
|
|
else
|
|
{
|
|
_logger.LogWarning(
|
|
"Debug snapshot for unknown instance {Instance}", request.InstanceUniqueName);
|
|
Sender.Tell(new DebugViewSnapshot(
|
|
request.InstanceUniqueName, Array.Empty<Commons.Messages.Streaming.AttributeValueChanged>(),
|
|
Array.Empty<Commons.Messages.Streaming.AlarmStateChanged>(), DateTimeOffset.UtcNow));
|
|
}
|
|
}
|
|
|
|
// ── Inbound API routing ──
|
|
|
|
private void RouteInboundApiCall(RouteToCallRequest request)
|
|
{
|
|
if (_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
// Convert to ScriptCallRequest and Ask the Instance Actor
|
|
var scriptCall = new ScriptCallRequest(
|
|
request.ScriptName, request.Parameters, 0, request.CorrelationId);
|
|
var sender = Sender;
|
|
instanceActor.Ask<ScriptCallResult>(scriptCall, TimeSpan.FromSeconds(30))
|
|
.ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
{
|
|
var result = t.Result;
|
|
return new RouteToCallResponse(
|
|
request.CorrelationId, result.Success, result.ReturnValue,
|
|
result.ErrorMessage, DateTimeOffset.UtcNow);
|
|
}
|
|
return new RouteToCallResponse(
|
|
request.CorrelationId, false, null,
|
|
t.Exception?.GetBaseException().Message ?? "Script call timed out",
|
|
DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
}
|
|
else
|
|
{
|
|
Sender.Tell(new RouteToCallResponse(
|
|
request.CorrelationId, false, null,
|
|
$"Instance '{request.InstanceUniqueName}' not found on this site.",
|
|
DateTimeOffset.UtcNow));
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Reads attribute values from a deployed instance for a Route.To().GetAttribute(s)
|
|
/// call (or a central Test Run bound to the instance). Asks the Instance Actor
|
|
/// per attribute and combines the results.
|
|
/// </summary>
|
|
private void RouteInboundApiGetAttributes(RouteToGetAttributesRequest request)
|
|
{
|
|
if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
Sender.Tell(new RouteToGetAttributesResponse(
|
|
request.CorrelationId, new Dictionary<string, object?>(), false,
|
|
$"Instance '{request.InstanceUniqueName}' not found on this site.",
|
|
DateTimeOffset.UtcNow));
|
|
return;
|
|
}
|
|
|
|
var sender = Sender;
|
|
var names = request.AttributeNames;
|
|
var asks = names
|
|
.Select(name => instanceActor.Ask<GetAttributeResponse>(
|
|
new GetAttributeRequest(
|
|
request.CorrelationId, request.InstanceUniqueName, name, DateTimeOffset.UtcNow),
|
|
TimeSpan.FromSeconds(30)))
|
|
.ToArray();
|
|
|
|
Task.WhenAll(asks).ContinueWith(t =>
|
|
{
|
|
if (t.IsCompletedSuccessfully)
|
|
{
|
|
var values = new Dictionary<string, object?>();
|
|
for (var i = 0; i < names.Count; i++)
|
|
values[names[i]] = t.Result[i].Found ? t.Result[i].Value : null;
|
|
return new RouteToGetAttributesResponse(
|
|
request.CorrelationId, values, true, null, DateTimeOffset.UtcNow);
|
|
}
|
|
return new RouteToGetAttributesResponse(
|
|
request.CorrelationId, new Dictionary<string, object?>(), false,
|
|
t.Exception?.GetBaseException().Message ?? "Attribute read timed out",
|
|
DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Writes attribute values on a deployed instance for a Route.To().SetAttribute(s)
|
|
/// call (or a central Test Run bound to the instance). Each write is Ask'd to the
|
|
/// Instance Actor, which routes data-sourced attributes through the DCL and static
|
|
/// attributes to a persisted override. The response reflects the real per-attribute
|
|
/// outcome (a non-existent attribute or a failed device write reports failure),
|
|
/// rather than an unconditional optimistic ack.
|
|
/// </summary>
|
|
private void RouteInboundApiSetAttributes(RouteToSetAttributesRequest request)
|
|
{
|
|
if (!_instanceActors.TryGetValue(request.InstanceUniqueName, out var instanceActor))
|
|
{
|
|
Sender.Tell(new RouteToSetAttributesResponse(
|
|
request.CorrelationId, false,
|
|
$"Instance '{request.InstanceUniqueName}' not found on this site.",
|
|
DateTimeOffset.UtcNow));
|
|
return;
|
|
}
|
|
|
|
var sender = Sender;
|
|
var correlationId = request.CorrelationId;
|
|
var asks = request.AttributeValues
|
|
.Select(kvp => instanceActor.Ask<SetStaticAttributeResponse>(
|
|
new SetStaticAttributeCommand(
|
|
correlationId, request.InstanceUniqueName, kvp.Key, kvp.Value, DateTimeOffset.UtcNow),
|
|
TimeSpan.FromSeconds(30)))
|
|
.ToArray();
|
|
|
|
Task.WhenAll(asks).ContinueWith(t =>
|
|
{
|
|
if (!t.IsCompletedSuccessfully)
|
|
return new RouteToSetAttributesResponse(
|
|
correlationId, false,
|
|
t.Exception?.GetBaseException().Message ?? "Attribute write timed out",
|
|
DateTimeOffset.UtcNow);
|
|
|
|
var failures = t.Result
|
|
.Where(r => !r.Success)
|
|
.Select(r => $"{r.AttributeName}: {r.ErrorMessage}")
|
|
.ToArray();
|
|
|
|
return failures.Length == 0
|
|
? new RouteToSetAttributesResponse(correlationId, true, null, DateTimeOffset.UtcNow)
|
|
: new RouteToSetAttributesResponse(
|
|
correlationId, false, string.Join("; ", failures), DateTimeOffset.UtcNow);
|
|
}).PipeTo(sender);
|
|
}
|
|
|
|
/// <summary>
|
|
/// WP-33: Handles system-wide artifact deployment (shared scripts, external systems, etc.).
|
|
/// Persists artifacts to SiteStorageService and recompiles shared scripts.
|
|
/// </summary>
|
|
private void HandleDeployArtifacts(DeployArtifactsCommand command)
|
|
{
|
|
_logger.LogInformation(
|
|
"Deploying system artifacts, deploymentId={DeploymentId}", command.DeploymentId);
|
|
|
|
var sender = Sender;
|
|
|
|
Task.Run(async () =>
|
|
{
|
|
try
|
|
{
|
|
// WP-33: Store shared scripts and recompile
|
|
if (command.SharedScripts != null)
|
|
{
|
|
foreach (var script in command.SharedScripts)
|
|
{
|
|
await _storage.StoreSharedScriptAsync(script.Name, script.Code,
|
|
script.ParameterDefinitions, script.ReturnDefinition);
|
|
|
|
// WP-33: Shared scripts recompiled on update
|
|
_sharedScriptLibrary.CompileAndRegister(script.Name, script.Code);
|
|
}
|
|
}
|
|
|
|
// WP-33: Store external system definitions
|
|
if (command.ExternalSystems != null)
|
|
{
|
|
foreach (var es in command.ExternalSystems)
|
|
{
|
|
await _storage.StoreExternalSystemAsync(es.Name, es.EndpointUrl,
|
|
es.AuthType, es.AuthConfiguration, es.MethodDefinitionsJson);
|
|
}
|
|
}
|
|
|
|
// WP-33: Store database connection definitions
|
|
if (command.DatabaseConnections != null)
|
|
{
|
|
foreach (var db in command.DatabaseConnections)
|
|
{
|
|
await _storage.StoreDatabaseConnectionAsync(db.Name, db.ConnectionString,
|
|
db.MaxRetries, db.RetryDelay);
|
|
}
|
|
}
|
|
|
|
// WP-33: Store notification lists
|
|
if (command.NotificationLists != null)
|
|
{
|
|
foreach (var nl in command.NotificationLists)
|
|
{
|
|
await _storage.StoreNotificationListAsync(nl.Name, nl.RecipientEmails);
|
|
}
|
|
}
|
|
|
|
// Store data connection definitions (OPC UA endpoints, etc.)
|
|
if (command.DataConnections != null)
|
|
{
|
|
foreach (var dc in command.DataConnections)
|
|
{
|
|
await _storage.StoreDataConnectionDefinitionAsync(
|
|
dc.Name, dc.Protocol, dc.PrimaryConfigurationJson,
|
|
dc.BackupConfigurationJson, dc.FailoverRetryCount);
|
|
}
|
|
}
|
|
|
|
// Store SMTP configurations
|
|
if (command.SmtpConfigurations != null)
|
|
{
|
|
foreach (var smtp in command.SmtpConfigurations)
|
|
{
|
|
await _storage.StoreSmtpConfigurationAsync(
|
|
smtp.Name, smtp.Server, smtp.Port, smtp.AuthMode,
|
|
smtp.FromAddress, smtp.Username, smtp.Password, smtp.OAuthConfig);
|
|
}
|
|
}
|
|
|
|
// Replicate artifacts to standby node
|
|
_replicationActor?.Tell(new ReplicateArtifacts(command));
|
|
|
|
return new ArtifactDeploymentResponse(
|
|
command.DeploymentId, "", true, null, DateTimeOffset.UtcNow);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
return new ArtifactDeploymentResponse(
|
|
command.DeploymentId, "", false, ex.Message, DateTimeOffset.UtcNow);
|
|
}
|
|
}).PipeTo(sender);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates a child Instance Actor with the given name and configuration JSON.
|
|
/// </summary>
|
|
internal void CreateInstanceActor(string instanceName, string configJson)
|
|
{
|
|
if (_instanceActors.ContainsKey(instanceName))
|
|
{
|
|
_logger.LogWarning("Instance Actor {Instance} already exists, skipping creation", instanceName);
|
|
return;
|
|
}
|
|
|
|
var loggerFactory = new LoggerFactory();
|
|
var props = Props.Create(() => new InstanceActor(
|
|
instanceName,
|
|
configJson,
|
|
_storage,
|
|
_compilationService,
|
|
_sharedScriptLibrary,
|
|
_streamManager,
|
|
_options,
|
|
loggerFactory.CreateLogger<InstanceActor>(),
|
|
_dclManager,
|
|
_healthCollector,
|
|
_serviceProvider));
|
|
|
|
var actorRef = Context.ActorOf(props, instanceName);
|
|
_instanceActors[instanceName] = actorRef;
|
|
|
|
_logger.LogDebug("Created Instance Actor for {Instance}", instanceName);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Gets the count of active Instance Actors (for testing/diagnostics).
|
|
/// </summary>
|
|
internal int InstanceActorCount => _instanceActors.Count;
|
|
|
|
/// <summary>
|
|
/// Updates the health collector with current instance counts.
|
|
/// Total deployed = _totalDeployedCount, enabled = running actors, disabled = difference.
|
|
/// </summary>
|
|
private void UpdateInstanceCounts()
|
|
{
|
|
_healthCollector?.SetInstanceCounts(
|
|
deployed: _totalDeployedCount,
|
|
enabled: _instanceActors.Count,
|
|
disabled: _totalDeployedCount - _instanceActors.Count);
|
|
}
|
|
|
|
// ── Internal messages ──
|
|
|
|
internal record StartupConfigsLoaded(List<DeployedInstance> Configs, string? Error);
|
|
internal record StartNextBatch(BatchState State);
|
|
internal record BatchState(List<DeployedInstance> Configs, int NextIndex);
|
|
internal record EnableResult(
|
|
EnableInstanceCommand Command, DeployedInstance? Config, string? Error, IActorRef OriginalSender);
|
|
internal record DeployPersistenceResult(
|
|
string DeploymentId, string InstanceName, bool Success, string? Error, IActorRef OriginalSender);
|
|
|
|
/// <summary>
|
|
/// A redeployment command buffered until the previous Instance Actor terminates.
|
|
/// </summary>
|
|
internal record PendingRedeploy(DeployInstanceCommand Command, IActorRef OriginalSender);
|
|
}
|