test(opcua): package cross-platform lifecycle memory controls
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,208 @@
|
||||
using System.Diagnostics;
|
||||
using System.Collections;
|
||||
using System.Reflection;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Runtime;
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Client;
|
||||
using Opc.Ua.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
using ZB.MOM.WW.OtOpcUa.OpcUaServer.Security;
|
||||
var root = Path.GetFullPath(args[0]);
|
||||
Directory.CreateDirectory(root);
|
||||
var seconds = int.Parse(args[1]);
|
||||
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(seconds + 120));
|
||||
var ct = timeout.Token;
|
||||
var clientPki = Path.Combine(root, "client-pki");
|
||||
var telemetry = DefaultTelemetry.Create(_ => { });
|
||||
var config = new ApplicationConfiguration(telemetry);
|
||||
config.ApplicationName = "NativeLifecycleControl";
|
||||
config.ApplicationUri = "urn:NativeLifecycleControl:" + Guid.NewGuid();
|
||||
config.ApplicationType = ApplicationType.Client;
|
||||
config.SecurityConfiguration = new SecurityConfiguration
|
||||
{
|
||||
ApplicationCertificate = new CertificateIdentifier { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientPki, "own"), SubjectName = "CN=NativeLifecycleControl" },
|
||||
TrustedIssuerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientPki, "issuer") },
|
||||
TrustedPeerCertificates = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientPki, "trusted") },
|
||||
RejectedCertificateStore = new CertificateTrustList { StoreType = CertificateStoreType.Directory, StorePath = Path.Combine(clientPki, "rejected") },
|
||||
AutoAcceptUntrustedCertificates = true
|
||||
};
|
||||
config.ClientConfiguration = new ClientConfiguration { DefaultSessionTimeout = 15000 };
|
||||
var app = new ApplicationInstance(telemetry) { ApplicationConfiguration = config };
|
||||
await config.ValidateAsync(ApplicationType.Client, ct);
|
||||
await app.CheckApplicationInstanceCertificatesAsync(true, null, ct);
|
||||
int port;
|
||||
using (var listener = new TcpListener(IPAddress.Loopback, 0))
|
||||
{
|
||||
listener.Start();
|
||||
port = ((IPEndPoint)listener.LocalEndpoint).Port;
|
||||
}
|
||||
var options = new OpcUaApplicationHostOptions { ApplicationName = "NativeLifecycleServer", ApplicationUri = "urn:NativeLifecycleServer:" + Guid.NewGuid(), OpcUaPort = port, PublicHostname = "127.0.0.1", PkiStoreRoot = Path.Combine(root, "server-pki"), AllowAnonymous = false, AutoAcceptUntrustedClientCertificates = true, EnabledSecurityProfiles = [OpcUaSecurityProfile.Basic256Sha256SignAndEncrypt] };
|
||||
OpcUaApplicationHost host = null!;
|
||||
OtOpcUaSdkServer server = null!;
|
||||
async Task Boot()
|
||||
{
|
||||
server = new OtOpcUaSdkServer();
|
||||
host = new OpcUaApplicationHost(options, NullLogger<OpcUaApplicationHost>.Instance, new Auth());
|
||||
await host.StartAsync(server, ct);
|
||||
var n = server.NodeManager!;
|
||||
n.NodeWriteGateway = new Gateway();
|
||||
n.EnsureFolder("Control", null, "Control", AddressSpaceRealm.Raw);
|
||||
n.EnsureFolder("Equipment", null, "Equipment", AddressSpaceRealm.Uns);
|
||||
n.EnsureVariable("Control/Scalar", "Control", "Scalar", "Int32", true, AddressSpaceRealm.Raw);
|
||||
n.EnsureVariable("Control/Array", "Control", "Array", "Int16", true, AddressSpaceRealm.Raw, isArray: true, arrayLength: 4);
|
||||
n.EnsureVariable("Equipment/Scalar", "Equipment", "Scalar", "Int32", true, AddressSpaceRealm.Uns);
|
||||
for (var i = 0; i < 16; i++) n.EnsureVariable("Control/Browse" + i, "Control", "Browse" + i, "Int32", false, AddressSpaceRealm.Raw);
|
||||
n.MaterialiseAlarmCondition("Control/Alarm", "Control", "Alarm", "OffNormalAlarm", 500, AddressSpaceRealm.Raw, isNative: true);
|
||||
n.WriteValue("Control/Scalar", 0, 0, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
n.WriteValue("Control/Array", new short[] { 0, 1, 2, 3 }, 0, DateTime.UtcNow, AddressSpaceRealm.Raw);
|
||||
n.WriteValue("Equipment/Scalar", 0, 0, DateTime.UtcNow, AddressSpaceRealm.Uns);
|
||||
}
|
||||
async Task<ISession> Connect()
|
||||
{
|
||||
var t = DefaultTelemetry.Create(_ => { });
|
||||
var endpoint = await CoreClientUtils.SelectEndpointAsync(config, $"opc.tcp://127.0.0.1:{port}/OtOpcUa", true, t, ct);
|
||||
var factory = new DefaultSessionFactory(t);
|
||||
var identity = new UserIdentity("control", Encoding.UTF8.GetBytes(FixtureCredential.Value));
|
||||
return await factory.CreateAsync(config, new ConfiguredEndpoint(null, endpoint ?? throw new Exception("Missing endpoint"), EndpointConfiguration.Create(config)), false, false, "NativeLifecycleControl", 15000, identity, default, ct);
|
||||
}
|
||||
static DataValue Value(object v)
|
||||
{
|
||||
#if SDK2
|
||||
if (!VariantHelper.TryCastFrom(v, out var variant)) throw new Exception("Unsupported control value");
|
||||
return new DataValue(variant);
|
||||
#else
|
||||
return new DataValue(new Variant(v));
|
||||
#endif
|
||||
}
|
||||
static object? Clr(DataValue v)
|
||||
{
|
||||
#if SDK2
|
||||
return v.WrappedValue.AsBoxedObject(Variant.BoxingBehavior.Legacy);
|
||||
#else
|
||||
return v.Value;
|
||||
#endif
|
||||
}
|
||||
object? CertificateStats()
|
||||
{
|
||||
#if SDK2
|
||||
return new { created = Opc.Ua.Security.Certificates.Certificate.InstancesCreated, disposed = Opc.Ua.Security.Certificates.Certificate.InstancesDisposed, outstanding = Opc.Ua.Security.Certificates.Certificate.InstancesLeaked };
|
||||
#else
|
||||
return null;
|
||||
#endif
|
||||
}
|
||||
void Sample(int cycle, string kind)
|
||||
{
|
||||
var live = GC.GetTotalMemory(true);
|
||||
using var p = Process.GetCurrentProcess();
|
||||
p.Refresh();
|
||||
var m = GC.GetGCMemoryInfo();
|
||||
Console.WriteLine(JsonSerializer.Serialize(new { kind, cycle, pid = Environment.ProcessId, live, rss = p.WorkingSet64, gcCommitted = m.TotalCommittedBytes, gcHeap = m.HeapSizeBytes, fragmented = m.FragmentedBytes, jit = JitInfo.GetCompiledMethodCount(), jitIl = JitInfo.GetCompiledILBytes(), sdk = typeof(NodeId).Assembly.GetName().Version?.ToString(), scope = "matched encrypted public lifecycle subset; no history/revocation/proxy", assemblies = AppDomain.CurrentDomain.GetAssemblies().Length, dynamicAssemblies = AppDomain.CurrentDomain.GetAssemblies().Count(a => a.IsDynamic), certificate = CertificateStats(), timers = Timer.ActiveCount, threads = p.Threads.Count, sessions = server.CurrentInstance.SessionManager.GetSessions().Count, subscriptions = server.CurrentInstance.SubscriptionManager.GetSubscriptions().Count }));
|
||||
Console.Out.Flush();
|
||||
}
|
||||
int cycles = 0, restarts = 0;
|
||||
await Boot();
|
||||
var watch = Stopwatch.StartNew();
|
||||
Sample(0, "start");
|
||||
try
|
||||
{
|
||||
while (watch.Elapsed.TotalSeconds < seconds)
|
||||
{
|
||||
using (var owner = await Connect()) using (var other = await Connect())
|
||||
{
|
||||
var ns = (ushort)owner.NamespaceUris.GetIndex("https://zb.com/otopcua/raw");
|
||||
var scalar = new NodeId("Control/Scalar", ns);
|
||||
var array = new NodeId("Control/Array", ns);
|
||||
using var subscription = new Subscription(owner.DefaultSubscription) { PublishingInterval = 100 };
|
||||
owner.AddSubscription(subscription);
|
||||
await subscription.CreateAsync(ct);
|
||||
subscription.AddItem(new MonitoredItem(subscription.DefaultItem) { StartNodeId = scalar, AttributeId = Attributes.Value, SamplingInterval = 50 });
|
||||
subscription.AddItem(new MonitoredItem(subscription.DefaultItem) { StartNodeId = array, AttributeId = Attributes.Value, SamplingInterval = 50 });
|
||||
await subscription.ApplyChangesAsync(ct);
|
||||
var response = await owner.WriteAsync(null, [new WriteValue { NodeId = scalar, AttributeId = Attributes.Value, Value = Value(++cycles) }, new WriteValue { NodeId = array, AttributeId = Attributes.Value, Value = Value(new short[] { 1, 2, 3, 4 }) }], ct);
|
||||
if (response.Results.ToList().Any(s => StatusCode.IsBad(s))) throw new Exception("Control write denied");
|
||||
if (!Equals(Clr(await owner.ReadValueAsync(scalar, ct)), cycles)) throw new Exception("Scalar changed");
|
||||
if (!((short[])Clr(await owner.ReadValueAsync(array, ct))!).SequenceEqual(new short[] { 1, 2, 3, 4 })) throw new Exception("Array changed");
|
||||
var browse = await owner.BrowseAsync(null, null, 1, [new BrowseDescription { NodeId = new NodeId("Control", ns), BrowseDirection = BrowseDirection.Forward, ReferenceTypeId = ReferenceTypeIds.HierarchicalReferences, IncludeSubtypes = true, ResultMask = (uint)BrowseResultMask.All }], ct);
|
||||
await Task.Delay(1000, ct);
|
||||
await subscription.DeleteAsync(true, ct);
|
||||
await owner.CloseAsync(ct);
|
||||
await other.CloseAsync(ct);
|
||||
}
|
||||
if (server.CurrentInstance.SessionManager.GetSessions().Count != 0 || server.CurrentInstance.SubscriptionManager.GetSubscriptions().Count != 0) throw new Exception("Control retained sessions/subscriptions");
|
||||
Sample(cycles, "sample");
|
||||
if (cycles % 4 == 0)
|
||||
{
|
||||
using var parked = await Connect();
|
||||
using var sub = new Subscription(parked.DefaultSubscription) { PublishingInterval = 100 };
|
||||
parked.AddSubscription(sub);
|
||||
await sub.CreateAsync(ct);
|
||||
sub.AddItem(new MonitoredItem(sub.DefaultItem) { StartNodeId = new NodeId("Control/Scalar", (ushort)parked.NamespaceUris.GetIndex("https://zb.com/otopcua/raw")), AttributeId = Attributes.Value, SamplingInterval = 50 });
|
||||
await sub.ApplyChangesAsync(ct);
|
||||
await host.DisposeAsync();
|
||||
await Boot();
|
||||
restarts++;
|
||||
}
|
||||
}
|
||||
Console.WriteLine(JsonSerializer.Serialize(new { kind = "complete", cycles, restarts, seconds = watch.Elapsed.TotalSeconds, fullGate = false }));
|
||||
}
|
||||
finally
|
||||
{
|
||||
await host.DisposeAsync();
|
||||
#if SDK2
|
||||
await app.DisposeAsync();
|
||||
#endif
|
||||
}
|
||||
#if SDK2
|
||||
GC.Collect();
|
||||
GC.WaitForPendingFinalizers();
|
||||
GC.Collect();
|
||||
var certificateType = typeof(Opc.Ua.Security.Certificates.Certificate);
|
||||
var report = new Dictionary<string, object>();
|
||||
var trackingEnabled = Environment.GetEnvironmentVariable("OPCUA_CERTIFICATE_LEAK_TRACKING") == "1";
|
||||
if (trackingEnabled)
|
||||
{
|
||||
foreach (var name in new[] { "EnumerateLiveCertificates", "EnumerateUnreachableUndisposedCertificates" })
|
||||
{
|
||||
var method = certificateType.GetMethod(name, BindingFlags.Static | BindingFlags.NonPublic)
|
||||
?? throw new InvalidOperationException("SDK2 diagnostic method missing: " + name);
|
||||
var values = method.Invoke(null, null) as IEnumerable
|
||||
?? throw new InvalidOperationException("SDK2 diagnostic method return type changed: " + name);
|
||||
var stacks = new List<string>();
|
||||
foreach (var value in values)
|
||||
{
|
||||
var field = value!.GetType().GetField(name.Contains("Unreachable") ? "Item2" : "Item4")
|
||||
?? throw new InvalidOperationException("SDK2 diagnostic tuple field missing: " + name);
|
||||
if (field.FieldType != typeof(string))
|
||||
throw new InvalidOperationException("SDK2 diagnostic stack field type changed");
|
||||
// Deliberately omit tuple Item1 (certificate identity).
|
||||
stacks.Add((string)field.GetValue(value)!);
|
||||
}
|
||||
report[name] = new
|
||||
{
|
||||
count = stacks.Count,
|
||||
allocations = stacks.GroupBy(x => x).OrderByDescending(g => g.Count()).Take(15)
|
||||
.Select(g => new { count = g.Count(), stack = g.Key }).ToArray()
|
||||
};
|
||||
}
|
||||
}
|
||||
Console.WriteLine(JsonSerializer.Serialize(new { kind = "certificate-final-diagnostics", trackingEnabled, stats = CertificateStats(), report }));
|
||||
#endif
|
||||
class Auth : IOpcUaUserAuthenticator
|
||||
{
|
||||
public Task<OpcUaUserAuthResult> AuthenticateUserNameAsync(string user, string password, CancellationToken ct) => Task.FromResult(user == "control" && password == FixtureCredential.Value ? OpcUaUserAuthResult.Allow("control", [OpcUaDataPlaneRoles.WriteOperate]) : OpcUaUserAuthResult.Deny("Control credential refused"));
|
||||
}
|
||||
class Gateway : IOpcUaNodeWriteGateway
|
||||
{
|
||||
public Task<NodeWriteOutcome> WriteAsync(string n, object? v, AddressSpaceRealm r, CancellationToken ct) => Task.FromResult(new NodeWriteOutcome(true, null));
|
||||
}
|
||||
|
||||
static class FixtureCredential
|
||||
{
|
||||
internal static string Value { get; } = Environment.GetEnvironmentVariable("OPCUA_MEMORY_PASSWORD") ?? throw new InvalidOperationException("Set OPCUA_MEMORY_PASSWORD for the task-owned fixture.");
|
||||
}
|
||||
Reference in New Issue
Block a user