Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7b86bab705 | |||
| 56886c3b4e |
@@ -18,7 +18,6 @@ starting `MxGateway.Worker.exe` or loading MXAccess COM. The harness scripts:
|
|||||||
- `WorkerHello` and `WorkerReady` startup,
|
- `WorkerHello` and `WorkerReady` startup,
|
||||||
- command replies with matching correlation ids,
|
- command replies with matching correlation ids,
|
||||||
- ordered `WorkerEvent` frames,
|
- ordered `WorkerEvent` frames,
|
||||||
- `WorkerHeartbeat` frames,
|
|
||||||
- `WorkerFault` frames,
|
- `WorkerFault` frames,
|
||||||
- shutdown acknowledgements,
|
- shutdown acknowledgements,
|
||||||
- malformed protobuf payloads and oversized frame headers,
|
- malformed protobuf payloads and oversized frame headers,
|
||||||
@@ -44,8 +43,6 @@ event streaming behavior:
|
|||||||
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~FakeWorkerHarnessTests
|
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~FakeWorkerHarnessTests
|
||||||
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~SessionWorkerClientFactoryFakeWorkerTests
|
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~SessionWorkerClientFactoryFakeWorkerTests
|
||||||
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~GatewayEndToEndFakeWorkerSmokeTests
|
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~GatewayEndToEndFakeWorkerSmokeTests
|
||||||
dotnet test src/MxGateway.Tests/MxGateway.Tests.csproj --filter FullyQualifiedName~WorkerClientTests
|
|
||||||
dotnet test src/MxGateway.Worker.Tests/MxGateway.Worker.Tests.csproj -p:Platform=x86 --filter FullyQualifiedName~WorkerPipeSessionTests
|
|
||||||
```
|
```
|
||||||
|
|
||||||
Run the gateway test project after shared gateway test infrastructure changes:
|
Run the gateway test project after shared gateway test infrastructure changes:
|
||||||
|
|||||||
@@ -34,9 +34,13 @@ SignalR circuit. Bootstrap is sufficient for a basic dashboard.
|
|||||||
|
|
||||||
## Hosting Model
|
## Hosting Model
|
||||||
|
|
||||||
The dashboard is hosted by `MxGateway.Server` alongside the gRPC API.
|
The dashboard is hosted by `MxGateway.Server` alongside the gRPC API. When
|
||||||
|
`MxGateway:Dashboard:Enabled` is `true`, `MapGatewayDashboard()` maps the
|
||||||
|
configured `Dashboard:PathBase` to the Blazor Server app and maps the login,
|
||||||
|
logout, and access-denied HTTP endpoints beside it. When dashboard hosting is
|
||||||
|
disabled, those routes are not mapped.
|
||||||
|
|
||||||
Suggested endpoint layout:
|
Endpoint layout:
|
||||||
|
|
||||||
```text
|
```text
|
||||||
/dashboard
|
/dashboard
|
||||||
@@ -45,7 +49,7 @@ Suggested endpoint layout:
|
|||||||
/dashboard/workers
|
/dashboard/workers
|
||||||
/dashboard/events
|
/dashboard/events
|
||||||
/dashboard/settings
|
/dashboard/settings
|
||||||
/_blazor
|
/dashboard/_blazor
|
||||||
```
|
```
|
||||||
|
|
||||||
The app should redirect `/` to `/dashboard` only if the deployment wants the
|
The app should redirect `/` to `/dashboard` only if the deployment wants the
|
||||||
@@ -59,9 +63,10 @@ MxGateway.Server
|
|||||||
Components/
|
Components/
|
||||||
App.razor
|
App.razor
|
||||||
Routes.razor
|
Routes.razor
|
||||||
|
DashboardPageBase.cs
|
||||||
|
DashboardDisplay.cs
|
||||||
Layout/
|
Layout/
|
||||||
DashboardLayout.razor
|
DashboardLayout.razor
|
||||||
NavMenu.razor
|
|
||||||
Pages/
|
Pages/
|
||||||
DashboardHome.razor
|
DashboardHome.razor
|
||||||
SessionsPage.razor
|
SessionsPage.razor
|
||||||
@@ -69,26 +74,21 @@ MxGateway.Server
|
|||||||
WorkersPage.razor
|
WorkersPage.razor
|
||||||
EventsPage.razor
|
EventsPage.razor
|
||||||
SettingsPage.razor
|
SettingsPage.razor
|
||||||
Components/
|
Shared/
|
||||||
MetricCard.razor
|
MetricCard.razor
|
||||||
SessionTable.razor
|
StatusBadge.razor
|
||||||
WorkerTable.razor
|
|
||||||
EventRatePanel.razor
|
|
||||||
FaultList.razor
|
FaultList.razor
|
||||||
Services/
|
DashboardSnapshotService.cs
|
||||||
DashboardSnapshotService.cs
|
DashboardAuthorizationHandler.cs
|
||||||
DashboardUpdateHub.cs
|
DashboardAuthenticator.cs
|
||||||
DashboardAuthorization.cs
|
DashboardSnapshot.cs
|
||||||
Models/
|
DashboardSessionSummary.cs
|
||||||
DashboardSnapshot.cs
|
DashboardWorkerSummary.cs
|
||||||
SessionSummary.cs
|
DashboardMetricSummary.cs
|
||||||
WorkerSummary.cs
|
|
||||||
MetricSummary.cs
|
|
||||||
```
|
```
|
||||||
|
|
||||||
`DashboardUpdateHub` here means an internal application update service, not a
|
Blazor Server provides the SignalR circuit for UI updates. The implementation
|
||||||
separate public SignalR hub unless implementation proves one is needed. Blazor
|
does not add a separate public dashboard hub.
|
||||||
Server already uses SignalR for UI circuits.
|
|
||||||
|
|
||||||
## Dashboard Data Source
|
## Dashboard Data Source
|
||||||
|
|
||||||
@@ -137,7 +137,7 @@ gateway internals.
|
|||||||
|
|
||||||
Use Blazor Server component state updates for real-time dashboard refresh.
|
Use Blazor Server component state updates for real-time dashboard refresh.
|
||||||
|
|
||||||
Recommended pattern:
|
Implemented pattern:
|
||||||
|
|
||||||
1. Page/component subscribes to `WatchSnapshotsAsync`.
|
1. Page/component subscribes to `WatchSnapshotsAsync`.
|
||||||
2. Snapshot service emits updates from a bounded channel or timer.
|
2. Snapshot service emits updates from a bounded channel or timer.
|
||||||
@@ -147,10 +147,8 @@ Recommended pattern:
|
|||||||
|
|
||||||
Default update cadence:
|
Default update cadence:
|
||||||
|
|
||||||
- immediate update on session create/close/fault,
|
|
||||||
- immediate update on worker fault,
|
|
||||||
- periodic metrics refresh every 1 second,
|
- periodic metrics refresh every 1 second,
|
||||||
- event-rate windows updated every 1 second.
|
- event counters update on the next snapshot tick.
|
||||||
|
|
||||||
Avoid pushing every MXAccess data-change event to the dashboard. Aggregate event
|
Avoid pushing every MXAccess data-change event to the dashboard. Aggregate event
|
||||||
counts and rates instead.
|
counts and rates instead.
|
||||||
@@ -320,7 +318,9 @@ Suggested configuration:
|
|||||||
|
|
||||||
## Styling
|
## Styling
|
||||||
|
|
||||||
Use Bootstrap utility classes and a small local stylesheet.
|
The dashboard serves Bootstrap 5.3.3 assets from
|
||||||
|
`src/MxGateway.Server/wwwroot/lib/bootstrap/` and local layout/status styling
|
||||||
|
from `src/MxGateway.Server/wwwroot/css/dashboard.css`.
|
||||||
|
|
||||||
Recommended visual language:
|
Recommended visual language:
|
||||||
|
|
||||||
@@ -361,15 +361,18 @@ Integration tests should verify:
|
|||||||
|
|
||||||
## Initial Implementation Slice
|
## Initial Implementation Slice
|
||||||
|
|
||||||
The first dashboard slice should implement:
|
The first dashboard slice implements:
|
||||||
|
|
||||||
1. Blazor Server hosting in `MxGateway.Server`.
|
1. Blazor Server hosting in `MxGateway.Server`.
|
||||||
2. Bootstrap static assets.
|
2. local Bootstrap static assets.
|
||||||
3. dashboard configuration binding.
|
3. dashboard configuration binding.
|
||||||
4. dashboard auth using API key login and HTTP-only cookie.
|
4. dashboard auth using API key login and HTTP-only cookie.
|
||||||
5. read-only `DashboardSnapshotService`.
|
5. read-only `DashboardSnapshotService`.
|
||||||
6. home page with metric cards.
|
6. home page with metric cards.
|
||||||
7. sessions page with active session table.
|
7. sessions page with active session table and session details.
|
||||||
8. workers page with worker table.
|
8. workers page with worker table.
|
||||||
9. 1-second realtime refresh through Blazor Server.
|
9. events page with aggregate counters.
|
||||||
10. redaction tests for secrets.
|
10. settings page with redacted effective configuration.
|
||||||
|
11. periodic realtime refresh through Blazor Server.
|
||||||
|
12. route-mapping tests, disabled-dashboard tests, auth tests, and snapshot
|
||||||
|
projection/redaction tests.
|
||||||
|
|||||||
@@ -576,19 +576,13 @@ Do not drop or coalesce events in v1.
|
|||||||
|
|
||||||
## Heartbeat And Watchdog
|
## Heartbeat And Watchdog
|
||||||
|
|
||||||
`WorkerPipeSession` starts the heartbeat loop after the gateway validates
|
The worker heartbeat should prove that:
|
||||||
`WorkerHello` and receives `WorkerReady`. Heartbeats continue until
|
|
||||||
`WorkerShutdown`, cancellation, or a pipe/protocol failure stops the session.
|
|
||||||
The loop uses `WorkerPipeSessionOptions.HeartbeatInterval`; the default matches
|
|
||||||
the gateway worker heartbeat interval.
|
|
||||||
|
|
||||||
The worker heartbeat proves that:
|
|
||||||
|
|
||||||
- pipe writer is alive,
|
- pipe writer is alive,
|
||||||
- worker host is alive,
|
- worker host is alive,
|
||||||
- STA has recently pumped or completed work.
|
- STA has recently pumped or completed work.
|
||||||
|
|
||||||
Heartbeat payload includes:
|
Heartbeat payload should include:
|
||||||
|
|
||||||
- worker process id,
|
- worker process id,
|
||||||
- session id,
|
- session id,
|
||||||
@@ -599,19 +593,13 @@ Heartbeat payload includes:
|
|||||||
- event sequence,
|
- event sequence,
|
||||||
- current command correlation id if any.
|
- current command correlation id if any.
|
||||||
|
|
||||||
`MxAccessStaSession.CaptureHeartbeat()` reads `StaRuntime.LastActivityUtc` and
|
The STA watchdog should warn when:
|
||||||
`StaCommandDispatcher` queue state without touching the raw MXAccess COM object
|
|
||||||
outside the STA. Event queue depth and event sequence are reported as zero until
|
|
||||||
the event queue implementation owns those counters.
|
|
||||||
|
|
||||||
The STA watchdog currently emits a `WorkerFault` with
|
- one command exceeds its expected duration,
|
||||||
`WorkerFaultCategory.StaHung` when `LastStaActivityUtc` is older than
|
- the STA has not pumped messages within the heartbeat grace period,
|
||||||
`WorkerPipeSessionOptions.HeartbeatGrace`. The fault includes the current
|
- event queue depth remains high.
|
||||||
command correlation id when a command is active. Command duration and high event
|
|
||||||
queue depth remain observable through heartbeat fields until dedicated
|
The worker can report the problem, but the gateway owns the final kill decision.
|
||||||
thresholds own those warnings. The worker reports stale STA activity, but the
|
|
||||||
gateway owns the final kill decision through its existing heartbeat and worker
|
|
||||||
lifecycle policy.
|
|
||||||
|
|
||||||
## Shutdown
|
## Shutdown
|
||||||
|
|
||||||
|
|||||||
+10
-1
@@ -109,13 +109,22 @@ histograms through .NET `Meter` and a snapshot API that dashboard services can
|
|||||||
project without binding to a metrics exporter.
|
project without binding to a metrics exporter.
|
||||||
`DashboardSnapshotService` projects sessions, workers, metrics, faults, and
|
`DashboardSnapshotService` projects sessions, workers, metrics, faults, and
|
||||||
effective configuration into immutable DTOs for read-only dashboard rendering.
|
effective configuration into immutable DTOs for read-only dashboard rendering.
|
||||||
|
The Blazor Server dashboard renders those snapshots at `/dashboard`,
|
||||||
|
`/dashboard/sessions`, `/dashboard/workers`, `/dashboard/events`, and
|
||||||
|
`/dashboard/settings`. Components subscribe to
|
||||||
|
`IDashboardSnapshotService.WatchSnapshotsAsync()` and update on the configured
|
||||||
|
snapshot interval without mutating session or worker state. The dashboard uses
|
||||||
|
local Bootstrap CSS and JavaScript plus a small local stylesheet; it does not
|
||||||
|
use a Blazor UI component library.
|
||||||
|
|
||||||
Dashboard routes use the same API-key verifier as gRPC. `/dashboard/login`
|
Dashboard routes use the same API-key verifier as gRPC. `/dashboard/login`
|
||||||
accepts the API key in a form body, validates the configured `admin` scope,
|
accepts the API key in a form body, validates the configured `admin` scope,
|
||||||
and issues an HTTP-only secure cookie for subsequent dashboard requests.
|
and issues an HTTP-only secure cookie for subsequent dashboard requests.
|
||||||
`/dashboard/logout` clears that cookie. Login and logout posts validate
|
`/dashboard/logout` clears that cookie. Login and logout posts validate
|
||||||
anti-forgery tokens, and API keys are never accepted through query strings.
|
anti-forgery tokens, and API keys are never accepted through query strings.
|
||||||
`Dashboard:AllowAnonymousLocalhost` can bypass the cookie requirement for
|
`Dashboard:AllowAnonymousLocalhost` can bypass the cookie requirement for
|
||||||
loopback requests only when explicitly enabled.
|
loopback requests only when explicitly enabled. Setting
|
||||||
|
`MxGateway:Dashboard:Enabled` to `false` leaves the dashboard routes unmapped.
|
||||||
|
|
||||||
### Worker Process
|
### Worker Process
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,34 @@
|
|||||||
|
@inject IOptions<GatewayOptions> GatewayOptions
|
||||||
|
|
||||||
|
<!doctype html>
|
||||||
|
<html lang="en">
|
||||||
|
<head>
|
||||||
|
<meta charset="utf-8" />
|
||||||
|
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||||
|
<base href="@DashboardBaseHref" />
|
||||||
|
<link rel="stylesheet" href="/lib/bootstrap/css/bootstrap.min.css" />
|
||||||
|
<link rel="stylesheet" href="/css/dashboard.css" />
|
||||||
|
<HeadOutlet @rendermode="InteractiveServer" />
|
||||||
|
</head>
|
||||||
|
<body class="dashboard-body">
|
||||||
|
<Routes @rendermode="InteractiveServer" />
|
||||||
|
<script src="/lib/bootstrap/js/bootstrap.bundle.min.js"></script>
|
||||||
|
<script src="_framework/blazor.web.js"></script>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
|
||||||
|
@code {
|
||||||
|
private string DashboardBaseHref
|
||||||
|
{
|
||||||
|
get
|
||||||
|
{
|
||||||
|
string pathBase = GatewayOptions.Value.Dashboard.PathBase.TrimEnd('/');
|
||||||
|
if (string.IsNullOrWhiteSpace(pathBase))
|
||||||
|
{
|
||||||
|
pathBase = "/dashboard";
|
||||||
|
}
|
||||||
|
|
||||||
|
return $"{pathBase}/";
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
namespace MxGateway.Server.Dashboard.Components;
|
||||||
|
|
||||||
|
public static class DashboardDisplay
|
||||||
|
{
|
||||||
|
public static string DateTime(DateTimeOffset? value)
|
||||||
|
{
|
||||||
|
return value.HasValue
|
||||||
|
? value.Value.UtcDateTime.ToString("yyyy-MM-dd HH:mm:ss 'UTC'", System.Globalization.CultureInfo.InvariantCulture)
|
||||||
|
: "-";
|
||||||
|
}
|
||||||
|
|
||||||
|
public static string Duration(TimeSpan value)
|
||||||
|
{
|
||||||
|
return value.TotalDays >= 1
|
||||||
|
? value.ToString(@"d\.hh\:mm\:ss", System.Globalization.CultureInfo.InvariantCulture)
|
||||||
|
: value.ToString(@"hh\:mm\:ss", System.Globalization.CultureInfo.InvariantCulture);
|
||||||
|
}
|
||||||
|
|
||||||
|
public static string Text(string? value)
|
||||||
|
{
|
||||||
|
return string.IsNullOrWhiteSpace(value) ? "-" : value;
|
||||||
|
}
|
||||||
|
|
||||||
|
public static long MetricValue(DashboardSnapshot snapshot, string name, string? dimension = null)
|
||||||
|
{
|
||||||
|
return snapshot.Metrics.FirstOrDefault(metric =>
|
||||||
|
string.Equals(metric.Name, name, StringComparison.Ordinal)
|
||||||
|
&& string.Equals(metric.Dimension, dimension, StringComparison.Ordinal))?.Value ?? 0;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,48 @@
|
|||||||
|
using Microsoft.AspNetCore.Components;
|
||||||
|
|
||||||
|
namespace MxGateway.Server.Dashboard.Components;
|
||||||
|
|
||||||
|
public abstract class DashboardPageBase : ComponentBase, IAsyncDisposable
|
||||||
|
{
|
||||||
|
private readonly CancellationTokenSource _disposeCancellation = new();
|
||||||
|
private Task? _watchTask;
|
||||||
|
|
||||||
|
[Inject]
|
||||||
|
protected IDashboardSnapshotService SnapshotService { get; set; } = null!;
|
||||||
|
|
||||||
|
protected DashboardSnapshot? Snapshot { get; private set; }
|
||||||
|
|
||||||
|
protected override void OnInitialized()
|
||||||
|
{
|
||||||
|
_watchTask = WatchSnapshotsAsync();
|
||||||
|
}
|
||||||
|
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
await _disposeCancellation.CancelAsync().ConfigureAwait(false);
|
||||||
|
if (_watchTask is not null)
|
||||||
|
{
|
||||||
|
await _watchTask.ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
|
||||||
|
_disposeCancellation.Dispose();
|
||||||
|
GC.SuppressFinalize(this);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task WatchSnapshotsAsync()
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await foreach (DashboardSnapshot snapshot in SnapshotService
|
||||||
|
.WatchSnapshotsAsync(_disposeCancellation.Token)
|
||||||
|
.ConfigureAwait(false))
|
||||||
|
{
|
||||||
|
Snapshot = snapshot;
|
||||||
|
await InvokeAsync(StateHasChanged).ConfigureAwait(false);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (_disposeCancellation.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,53 @@
|
|||||||
|
@inherits LayoutComponentBase
|
||||||
|
@inject IOptions<GatewayOptions> GatewayOptions
|
||||||
|
|
||||||
|
<div class="dashboard-shell">
|
||||||
|
<nav class="navbar navbar-expand-lg bg-body border-bottom dashboard-navbar">
|
||||||
|
<div class="container-fluid">
|
||||||
|
<a class="navbar-brand" href="">MXAccess Gateway</a>
|
||||||
|
<button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#dashboardNav"
|
||||||
|
aria-controls="dashboardNav" aria-expanded="false" aria-label="Toggle navigation">
|
||||||
|
<span class="navbar-toggler-icon"></span>
|
||||||
|
</button>
|
||||||
|
<div class="collapse navbar-collapse" id="dashboardNav">
|
||||||
|
<ul class="navbar-nav me-auto mb-2 mb-lg-0">
|
||||||
|
<li class="nav-item">
|
||||||
|
<NavLink class="nav-link" href="" Match="NavLinkMatch.All">Overview</NavLink>
|
||||||
|
</li>
|
||||||
|
<li class="nav-item">
|
||||||
|
<NavLink class="nav-link" href="sessions">Sessions</NavLink>
|
||||||
|
</li>
|
||||||
|
<li class="nav-item">
|
||||||
|
<NavLink class="nav-link" href="workers">Workers</NavLink>
|
||||||
|
</li>
|
||||||
|
<li class="nav-item">
|
||||||
|
<NavLink class="nav-link" href="events">Events</NavLink>
|
||||||
|
</li>
|
||||||
|
<li class="nav-item">
|
||||||
|
<NavLink class="nav-link" href="settings">Settings</NavLink>
|
||||||
|
</li>
|
||||||
|
</ul>
|
||||||
|
<form method="post" action="@DashboardPath("/logout")" class="d-flex">
|
||||||
|
<AntiforgeryToken />
|
||||||
|
<button class="btn btn-outline-secondary btn-sm" type="submit">Sign out</button>
|
||||||
|
</form>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
</nav>
|
||||||
|
<main class="container-fluid dashboard-content">
|
||||||
|
@Body
|
||||||
|
</main>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
@code {
|
||||||
|
private string DashboardPath(string relativePath)
|
||||||
|
{
|
||||||
|
string pathBase = GatewayOptions.Value.Dashboard.PathBase.TrimEnd('/');
|
||||||
|
if (string.IsNullOrWhiteSpace(pathBase))
|
||||||
|
{
|
||||||
|
pathBase = "/dashboard";
|
||||||
|
}
|
||||||
|
|
||||||
|
return $"{pathBase}{relativePath}";
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,37 @@
|
|||||||
|
@page "/"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>MXAccess Gateway Dashboard</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading dashboard snapshot.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Overview</h1>
|
||||||
|
<div class="text-secondary">Generated @DashboardDisplay.DateTime(Snapshot.GeneratedAt)</div>
|
||||||
|
</div>
|
||||||
|
<StatusBadge Text="@Snapshot.GatewayStatus" />
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="metric-grid">
|
||||||
|
<MetricCard Label="Uptime" Value="@DashboardDisplay.Duration(Snapshot.GatewayUptime)" Detail="@Snapshot.GatewayVersion" />
|
||||||
|
<MetricCard Label="Open Sessions" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.sessions.open").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Workers Running" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.workers.running").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Event Queue Depth" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.queue.depth").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Commands Failed" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.commands.failed").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Events Received" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.received").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Faults" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.faults").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Queue Overflows" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.queues.overflows").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
</section>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<div class="section-heading">
|
||||||
|
<h2>Recent Faults</h2>
|
||||||
|
</div>
|
||||||
|
<FaultList Faults="@Snapshot.Faults" />
|
||||||
|
</section>
|
||||||
|
}
|
||||||
@@ -0,0 +1,64 @@
|
|||||||
|
@page "/events"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>Dashboard Events</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading event diagnostics.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Events</h1>
|
||||||
|
<div class="text-secondary">Generated @DashboardDisplay.DateTime(Snapshot.GeneratedAt)</div>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="metric-grid compact">
|
||||||
|
<MetricCard Label="Events Received" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.received").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Event Queue Depth" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.events.queue.depth").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Queue Overflows" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.queues.overflows").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
<MetricCard Label="Stream Disconnects" Value="@DashboardDisplay.MetricValue(Snapshot, "mxgateway.grpc.streams.disconnected").ToString(System.Globalization.CultureInfo.InvariantCulture)" />
|
||||||
|
</section>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<div class="section-heading">
|
||||||
|
<h2>Event Families</h2>
|
||||||
|
</div>
|
||||||
|
@if (EventFamilyMetrics.Count == 0)
|
||||||
|
{
|
||||||
|
<div class="empty-state">No event family counters recorded.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm dashboard-table">
|
||||||
|
<thead>
|
||||||
|
<tr>
|
||||||
|
<th scope="col">Family</th>
|
||||||
|
<th scope="col">Count</th>
|
||||||
|
</tr>
|
||||||
|
</thead>
|
||||||
|
<tbody>
|
||||||
|
@foreach (DashboardMetricSummary metric in EventFamilyMetrics)
|
||||||
|
{
|
||||||
|
<tr>
|
||||||
|
<td>@metric.Dimension</td>
|
||||||
|
<td>@metric.Value</td>
|
||||||
|
</tr>
|
||||||
|
}
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
}
|
||||||
|
</section>
|
||||||
|
}
|
||||||
|
|
||||||
|
@code {
|
||||||
|
private IReadOnlyList<DashboardMetricSummary> EventFamilyMetrics => Snapshot?.Metrics
|
||||||
|
.Where(metric => metric.Name == "mxgateway.events.received" && metric.Dimension is not null)
|
||||||
|
.OrderBy(metric => metric.Dimension, StringComparer.OrdinalIgnoreCase)
|
||||||
|
.ToArray() ?? [];
|
||||||
|
}
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
@page "/sessions/{SessionId}"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>Dashboard Session</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading session.</div>
|
||||||
|
}
|
||||||
|
else if (CurrentSession is null)
|
||||||
|
{
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<h1 class="h4 mb-3">Session Not Found</h1>
|
||||||
|
<p class="mb-0">The session is not present in the current snapshot.</p>
|
||||||
|
</section>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Session Details</h1>
|
||||||
|
<div class="text-secondary"><code>@CurrentSession.SessionId</code></div>
|
||||||
|
</div>
|
||||||
|
<StatusBadge Text="@CurrentSession.State.ToString()" />
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<div class="section-heading">
|
||||||
|
<h2>Session</h2>
|
||||||
|
</div>
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm dashboard-table details-table">
|
||||||
|
<tbody>
|
||||||
|
<tr><th scope="row">Backend</th><td>@CurrentSession.BackendName</td></tr>
|
||||||
|
<tr><th scope="row">Client identity</th><td>@DashboardDisplay.Text(CurrentSession.ClientIdentity)</td></tr>
|
||||||
|
<tr><th scope="row">Client session</th><td>@DashboardDisplay.Text(CurrentSession.ClientSessionName)</td></tr>
|
||||||
|
<tr><th scope="row">Client correlation</th><td>@DashboardDisplay.Text(CurrentSession.ClientCorrelationId)</td></tr>
|
||||||
|
<tr><th scope="row">Opened</th><td>@DashboardDisplay.DateTime(CurrentSession.OpenedAt)</td></tr>
|
||||||
|
<tr><th scope="row">Last activity</th><td>@DashboardDisplay.DateTime(CurrentSession.LastClientActivityAt)</td></tr>
|
||||||
|
<tr><th scope="row">Lease expires</th><td>@DashboardDisplay.DateTime(CurrentSession.LeaseExpiresAt)</td></tr>
|
||||||
|
<tr><th scope="row">Last fault</th><td>@DashboardDisplay.Text(CurrentSession.LastFault)</td></tr>
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
</section>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<div class="section-heading">
|
||||||
|
<h2>Worker</h2>
|
||||||
|
</div>
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm dashboard-table details-table">
|
||||||
|
<tbody>
|
||||||
|
<tr><th scope="row">Process id</th><td>@(CurrentSession.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td></tr>
|
||||||
|
<tr><th scope="row">State</th><td><StatusBadge Text="@(CurrentSession.WorkerState?.ToString() ?? "-")" /></td></tr>
|
||||||
|
<tr><th scope="row">Last heartbeat</th><td>@DashboardDisplay.DateTime(CurrentSession.LastWorkerHeartbeatAt)</td></tr>
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
</section>
|
||||||
|
}
|
||||||
|
|
||||||
|
@code {
|
||||||
|
[Parameter]
|
||||||
|
public string SessionId { get; set; } = string.Empty;
|
||||||
|
|
||||||
|
private DashboardSessionSummary? CurrentSession => Snapshot?.Sessions.FirstOrDefault(session =>
|
||||||
|
string.Equals(session.SessionId, SessionId, StringComparison.Ordinal));
|
||||||
|
}
|
||||||
@@ -0,0 +1,67 @@
|
|||||||
|
@page "/sessions"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>Dashboard Sessions</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading sessions.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Sessions</h1>
|
||||||
|
<div class="text-secondary">@Snapshot.Sessions.Count session rows</div>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
@if (Snapshot.Sessions.Count == 0)
|
||||||
|
{
|
||||||
|
<div class="empty-state">No sessions are active or retained.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm align-middle dashboard-table">
|
||||||
|
<thead>
|
||||||
|
<tr>
|
||||||
|
<th scope="col">Session</th>
|
||||||
|
<th scope="col">State</th>
|
||||||
|
<th scope="col">Client</th>
|
||||||
|
<th scope="col">Backend</th>
|
||||||
|
<th scope="col">Worker</th>
|
||||||
|
<th scope="col">Opened</th>
|
||||||
|
<th scope="col">Activity</th>
|
||||||
|
<th scope="col">Heartbeat</th>
|
||||||
|
<th scope="col">Fault</th>
|
||||||
|
</tr>
|
||||||
|
</thead>
|
||||||
|
<tbody>
|
||||||
|
@foreach (DashboardSessionSummary session in Snapshot.Sessions)
|
||||||
|
{
|
||||||
|
<tr>
|
||||||
|
<td><NavLink href="@($"sessions/{Uri.EscapeDataString(session.SessionId)}")"><code>@session.SessionId</code></NavLink></td>
|
||||||
|
<td><StatusBadge Text="@session.State.ToString()" /></td>
|
||||||
|
<td>@DashboardDisplay.Text(session.ClientIdentity)</td>
|
||||||
|
<td>@session.BackendName</td>
|
||||||
|
<td>
|
||||||
|
@(session.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")
|
||||||
|
@if (session.WorkerState is not null)
|
||||||
|
{
|
||||||
|
<span class="ms-1"><StatusBadge Text="@session.WorkerState.ToString()" /></span>
|
||||||
|
}
|
||||||
|
</td>
|
||||||
|
<td>@DashboardDisplay.DateTime(session.OpenedAt)</td>
|
||||||
|
<td>@DashboardDisplay.DateTime(session.LastClientActivityAt)</td>
|
||||||
|
<td>@DashboardDisplay.DateTime(session.LastWorkerHeartbeatAt)</td>
|
||||||
|
<td>@DashboardDisplay.Text(session.LastFault)</td>
|
||||||
|
</tr>
|
||||||
|
}
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
}
|
||||||
|
</section>
|
||||||
|
}
|
||||||
@@ -0,0 +1,47 @@
|
|||||||
|
@page "/settings"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>Dashboard Settings</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading settings.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Settings</h1>
|
||||||
|
<div class="text-secondary">Effective gateway configuration</div>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm dashboard-table details-table">
|
||||||
|
<tbody>
|
||||||
|
<tr><th scope="row">Authentication mode</th><td>@Snapshot.Configuration.Authentication.Mode</td></tr>
|
||||||
|
<tr><th scope="row">Auth database</th><td><code>@Snapshot.Configuration.Authentication.SqlitePath</code></td></tr>
|
||||||
|
<tr><th scope="row">Pepper secret</th><td>@Snapshot.Configuration.Authentication.PepperSecretName</td></tr>
|
||||||
|
<tr><th scope="row">Run migrations</th><td>@Snapshot.Configuration.Authentication.RunMigrationsOnStartup</td></tr>
|
||||||
|
<tr><th scope="row">Worker executable</th><td><code>@Snapshot.Configuration.Worker.ExecutablePath</code></td></tr>
|
||||||
|
<tr><th scope="row">Worker architecture</th><td>@Snapshot.Configuration.Worker.RequiredArchitecture</td></tr>
|
||||||
|
<tr><th scope="row">Startup timeout</th><td>@Snapshot.Configuration.Worker.StartupTimeoutSeconds seconds</td></tr>
|
||||||
|
<tr><th scope="row">Shutdown timeout</th><td>@Snapshot.Configuration.Worker.ShutdownTimeoutSeconds seconds</td></tr>
|
||||||
|
<tr><th scope="row">Heartbeat grace</th><td>@Snapshot.Configuration.Worker.HeartbeatGraceSeconds seconds</td></tr>
|
||||||
|
<tr><th scope="row">Default command timeout</th><td>@Snapshot.Configuration.Sessions.DefaultCommandTimeoutSeconds seconds</td></tr>
|
||||||
|
<tr><th scope="row">Max sessions</th><td>@Snapshot.Configuration.Sessions.MaxSessions</td></tr>
|
||||||
|
<tr><th scope="row">Event queue capacity</th><td>@Snapshot.Configuration.Events.QueueCapacity</td></tr>
|
||||||
|
<tr><th scope="row">Backpressure policy</th><td>@Snapshot.Configuration.Events.BackpressurePolicy</td></tr>
|
||||||
|
<tr><th scope="row">Dashboard enabled</th><td>@Snapshot.Configuration.Dashboard.Enabled</td></tr>
|
||||||
|
<tr><th scope="row">Dashboard path</th><td>@Snapshot.Configuration.Dashboard.PathBase</td></tr>
|
||||||
|
<tr><th scope="row">Require admin scope</th><td>@Snapshot.Configuration.Dashboard.RequireAdminScope</td></tr>
|
||||||
|
<tr><th scope="row">Anonymous localhost</th><td>@Snapshot.Configuration.Dashboard.AllowAnonymousLocalhost</td></tr>
|
||||||
|
<tr><th scope="row">Snapshot interval</th><td>@Snapshot.Configuration.Dashboard.SnapshotIntervalMilliseconds ms</td></tr>
|
||||||
|
<tr><th scope="row">Show tag values</th><td>@Snapshot.Configuration.Dashboard.ShowTagValues</td></tr>
|
||||||
|
<tr><th scope="row">Worker protocol</th><td>@Snapshot.Configuration.Protocol.WorkerProtocolVersion</td></tr>
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
</section>
|
||||||
|
}
|
||||||
@@ -0,0 +1,53 @@
|
|||||||
|
@page "/workers"
|
||||||
|
@inherits DashboardPageBase
|
||||||
|
|
||||||
|
<PageTitle>Dashboard Workers</PageTitle>
|
||||||
|
|
||||||
|
@if (Snapshot is null)
|
||||||
|
{
|
||||||
|
<div class="empty-state">Loading workers.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="dashboard-page-header">
|
||||||
|
<div>
|
||||||
|
<h1>Workers</h1>
|
||||||
|
<div class="text-secondary">@Snapshot.Workers.Count worker rows</div>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<section class="dashboard-section">
|
||||||
|
@if (Snapshot.Workers.Count == 0)
|
||||||
|
{
|
||||||
|
<div class="empty-state">No worker processes are attached.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm align-middle dashboard-table">
|
||||||
|
<thead>
|
||||||
|
<tr>
|
||||||
|
<th scope="col">Process</th>
|
||||||
|
<th scope="col">State</th>
|
||||||
|
<th scope="col">Session</th>
|
||||||
|
<th scope="col">Heartbeat</th>
|
||||||
|
<th scope="col">Fault</th>
|
||||||
|
</tr>
|
||||||
|
</thead>
|
||||||
|
<tbody>
|
||||||
|
@foreach (DashboardWorkerSummary worker in Snapshot.Workers)
|
||||||
|
{
|
||||||
|
<tr>
|
||||||
|
<td>@(worker.ProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td>
|
||||||
|
<td><StatusBadge Text="@worker.State.ToString()" /></td>
|
||||||
|
<td><NavLink href="@($"sessions/{Uri.EscapeDataString(worker.SessionId)}")"><code>@worker.SessionId</code></NavLink></td>
|
||||||
|
<td>@DashboardDisplay.DateTime(worker.LastHeartbeatAt)</td>
|
||||||
|
<td>@DashboardDisplay.Text(worker.LastFault)</td>
|
||||||
|
</tr>
|
||||||
|
}
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
}
|
||||||
|
</section>
|
||||||
|
}
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
<Router AppAssembly="@typeof(Routes).Assembly">
|
||||||
|
<Found Context="routeData">
|
||||||
|
<RouteView RouteData="@routeData" DefaultLayout="@typeof(DashboardLayout)" />
|
||||||
|
<FocusOnNavigate RouteData="@routeData" Selector="h1" />
|
||||||
|
</Found>
|
||||||
|
<NotFound>
|
||||||
|
<LayoutView Layout="@typeof(DashboardLayout)">
|
||||||
|
<PageTitle>Dashboard - Not Found</PageTitle>
|
||||||
|
<section class="dashboard-section">
|
||||||
|
<h1 class="h4 mb-3">Not Found</h1>
|
||||||
|
<p class="mb-0">The requested dashboard page does not exist.</p>
|
||||||
|
</section>
|
||||||
|
</LayoutView>
|
||||||
|
</NotFound>
|
||||||
|
</Router>
|
||||||
@@ -0,0 +1,39 @@
|
|||||||
|
@if (Faults.Count == 0)
|
||||||
|
{
|
||||||
|
<div class="empty-state">No faults recorded.</div>
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
<div class="table-responsive">
|
||||||
|
<table class="table table-sm align-middle dashboard-table">
|
||||||
|
<thead>
|
||||||
|
<tr>
|
||||||
|
<th scope="col">Observed</th>
|
||||||
|
<th scope="col">Source</th>
|
||||||
|
<th scope="col">Session</th>
|
||||||
|
<th scope="col">Worker</th>
|
||||||
|
<th scope="col">State</th>
|
||||||
|
<th scope="col">Message</th>
|
||||||
|
</tr>
|
||||||
|
</thead>
|
||||||
|
<tbody>
|
||||||
|
@foreach (DashboardFaultSummary fault in Faults)
|
||||||
|
{
|
||||||
|
<tr>
|
||||||
|
<td>@DashboardDisplay.DateTime(fault.ObservedAt)</td>
|
||||||
|
<td>@fault.Source</td>
|
||||||
|
<td><code>@DashboardDisplay.Text(fault.SessionId)</code></td>
|
||||||
|
<td>@(fault.WorkerProcessId?.ToString(System.Globalization.CultureInfo.InvariantCulture) ?? "-")</td>
|
||||||
|
<td><StatusBadge Text="@fault.State" /></td>
|
||||||
|
<td>@fault.Message</td>
|
||||||
|
</tr>
|
||||||
|
}
|
||||||
|
</tbody>
|
||||||
|
</table>
|
||||||
|
</div>
|
||||||
|
}
|
||||||
|
|
||||||
|
@code {
|
||||||
|
[Parameter]
|
||||||
|
public IReadOnlyList<DashboardFaultSummary> Faults { get; set; } = [];
|
||||||
|
}
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
<div class="card metric-card h-100">
|
||||||
|
<div class="card-body">
|
||||||
|
<div class="metric-label">@Label</div>
|
||||||
|
<div class="metric-value">@Value</div>
|
||||||
|
@if (!string.IsNullOrWhiteSpace(Detail))
|
||||||
|
{
|
||||||
|
<div class="metric-detail">@Detail</div>
|
||||||
|
}
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
@code {
|
||||||
|
[Parameter]
|
||||||
|
public string Label { get; set; } = string.Empty;
|
||||||
|
|
||||||
|
[Parameter]
|
||||||
|
public string Value { get; set; } = string.Empty;
|
||||||
|
|
||||||
|
[Parameter]
|
||||||
|
public string? Detail { get; set; }
|
||||||
|
}
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
<span class="badge @CssClass">@Text</span>
|
||||||
|
|
||||||
|
@code {
|
||||||
|
[Parameter]
|
||||||
|
public string? Text { get; set; }
|
||||||
|
|
||||||
|
private string CssClass => Text switch
|
||||||
|
{
|
||||||
|
"Ready" or "Healthy" => "text-bg-success",
|
||||||
|
"Creating" or "StartingWorker" or "WaitingForPipe" or "InitializingWorker" or "Closing" => "text-bg-info",
|
||||||
|
"Closed" => "text-bg-secondary",
|
||||||
|
"Faulted" => "text-bg-danger",
|
||||||
|
_ => "text-bg-light text-dark border"
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
@using Microsoft.AspNetCore.Components.Authorization
|
||||||
|
@using Microsoft.AspNetCore.Components.Forms
|
||||||
|
@using Microsoft.AspNetCore.Components.Routing
|
||||||
|
@using Microsoft.AspNetCore.Components.Web
|
||||||
|
@using Microsoft.Extensions.Options
|
||||||
|
@using MxGateway.Contracts.Proto
|
||||||
|
@using MxGateway.Server.Configuration
|
||||||
|
@using MxGateway.Server.Dashboard
|
||||||
|
@using MxGateway.Server.Dashboard.Components.Layout
|
||||||
|
@using MxGateway.Server.Dashboard.Components.Shared
|
||||||
|
@using MxGateway.Server.Workers
|
||||||
|
@using static Microsoft.AspNetCore.Components.Web.RenderMode
|
||||||
@@ -3,6 +3,7 @@ using Microsoft.AspNetCore.Antiforgery;
|
|||||||
using Microsoft.AspNetCore.Authentication;
|
using Microsoft.AspNetCore.Authentication;
|
||||||
using Microsoft.AspNetCore.Http.HttpResults;
|
using Microsoft.AspNetCore.Http.HttpResults;
|
||||||
using MxGateway.Server.Configuration;
|
using MxGateway.Server.Configuration;
|
||||||
|
using MxGateway.Server.Dashboard.Components;
|
||||||
|
|
||||||
namespace MxGateway.Server.Dashboard;
|
namespace MxGateway.Server.Dashboard;
|
||||||
|
|
||||||
@@ -22,13 +23,6 @@ public static class DashboardEndpointRouteBuilderExtensions
|
|||||||
string pathBase = NormalizePathBase(dashboardSection["PathBase"] ?? new DashboardOptions().PathBase);
|
string pathBase = NormalizePathBase(dashboardSection["PathBase"] ?? new DashboardOptions().PathBase);
|
||||||
RouteGroupBuilder dashboard = endpoints.MapGroup(pathBase);
|
RouteGroupBuilder dashboard = endpoints.MapGroup(pathBase);
|
||||||
|
|
||||||
dashboard.MapGet(
|
|
||||||
"/",
|
|
||||||
(HttpContext httpContext, IAntiforgery antiforgery, IDashboardSnapshotService snapshotService) =>
|
|
||||||
GetDashboardHomeAsync(httpContext, antiforgery, snapshotService, pathBase))
|
|
||||||
.RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy)
|
|
||||||
.WithName("DashboardHome");
|
|
||||||
|
|
||||||
dashboard.MapGet(
|
dashboard.MapGet(
|
||||||
"/login",
|
"/login",
|
||||||
(HttpContext httpContext, IAntiforgery antiforgery) => GetLoginAsync(httpContext, antiforgery, pathBase))
|
(HttpContext httpContext, IAntiforgery antiforgery) => GetLoginAsync(httpContext, antiforgery, pathBase))
|
||||||
@@ -54,36 +48,13 @@ public static class DashboardEndpointRouteBuilderExtensions
|
|||||||
.AllowAnonymous()
|
.AllowAnonymous()
|
||||||
.WithName("DashboardAccessDenied");
|
.WithName("DashboardAccessDenied");
|
||||||
|
|
||||||
|
dashboard.MapRazorComponents<App>()
|
||||||
|
.AddInteractiveServerRenderMode()
|
||||||
|
.RequireAuthorization(DashboardAuthenticationDefaults.AuthorizationPolicy);
|
||||||
|
|
||||||
return endpoints;
|
return endpoints;
|
||||||
}
|
}
|
||||||
|
|
||||||
private static ContentHttpResult GetDashboardHomeAsync(
|
|
||||||
HttpContext httpContext,
|
|
||||||
IAntiforgery antiforgery,
|
|
||||||
IDashboardSnapshotService snapshotService,
|
|
||||||
string pathBase)
|
|
||||||
{
|
|
||||||
AntiforgeryTokenSet tokens = antiforgery.GetAndStoreTokens(httpContext);
|
|
||||||
DashboardSnapshot snapshot = snapshotService.GetSnapshot();
|
|
||||||
string requestToken = tokens.RequestToken ?? string.Empty;
|
|
||||||
string body = $"""
|
|
||||||
<form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/logout")}" class="mb-3">
|
|
||||||
<input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
|
|
||||||
<button type="submit">Sign out</button>
|
|
||||||
</form>
|
|
||||||
<dl>
|
|
||||||
<dt>Open sessions</dt>
|
|
||||||
<dd>{snapshot.Sessions.Count}</dd>
|
|
||||||
<dt>Workers</dt>
|
|
||||||
<dd>{snapshot.Workers.Count}</dd>
|
|
||||||
<dt>Faults</dt>
|
|
||||||
<dd>{snapshot.Faults.Count}</dd>
|
|
||||||
</dl>
|
|
||||||
""";
|
|
||||||
|
|
||||||
return TypedResults.Content(RenderPage("MXAccess Gateway Dashboard", body), "text/html");
|
|
||||||
}
|
|
||||||
|
|
||||||
private static Task<ContentHttpResult> GetLoginAsync(
|
private static Task<ContentHttpResult> GetLoginAsync(
|
||||||
HttpContext httpContext,
|
HttpContext httpContext,
|
||||||
IAntiforgery antiforgery,
|
IAntiforgery antiforgery,
|
||||||
@@ -159,14 +130,20 @@ public static class DashboardEndpointRouteBuilderExtensions
|
|||||||
: $"<p role=\"alert\">{HtmlEncoder.Default.Encode(failureMessage)}</p>";
|
: $"<p role=\"alert\">{HtmlEncoder.Default.Encode(failureMessage)}</p>";
|
||||||
|
|
||||||
string body = $"""
|
string body = $"""
|
||||||
{alert}
|
<section class="dashboard-login">
|
||||||
<form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/login")}">
|
{alert}
|
||||||
<input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
|
<form method="post" action="{HtmlEncoder.Default.Encode(pathBase + "/login")}" class="card login-card">
|
||||||
<input name="returnUrl" type="hidden" value="{HtmlEncoder.Default.Encode(returnUrl)}" />
|
<div class="card-body">
|
||||||
<label for="apiKey">API key</label>
|
<input name="{tokens.FormFieldName}" type="hidden" value="{HtmlEncoder.Default.Encode(requestToken)}" />
|
||||||
<input id="apiKey" name="apiKey" type="password" autocomplete="off" />
|
<input name="returnUrl" type="hidden" value="{HtmlEncoder.Default.Encode(returnUrl)}" />
|
||||||
<button type="submit">Sign in</button>
|
<div class="mb-3">
|
||||||
</form>
|
<label for="apiKey" class="form-label">API key</label>
|
||||||
|
<input id="apiKey" name="apiKey" type="password" autocomplete="off" class="form-control" />
|
||||||
|
</div>
|
||||||
|
<button type="submit" class="btn btn-primary">Sign in</button>
|
||||||
|
</div>
|
||||||
|
</form>
|
||||||
|
</section>
|
||||||
""";
|
""";
|
||||||
|
|
||||||
return RenderPage("Dashboard Sign In", body);
|
return RenderPage("Dashboard Sign In", body);
|
||||||
@@ -181,12 +158,15 @@ public static class DashboardEndpointRouteBuilderExtensions
|
|||||||
<meta charset="utf-8" />
|
<meta charset="utf-8" />
|
||||||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||||
<title>{HtmlEncoder.Default.Encode(title)}</title>
|
<title>{HtmlEncoder.Default.Encode(title)}</title>
|
||||||
|
<link rel="stylesheet" href="/lib/bootstrap/css/bootstrap.min.css" />
|
||||||
|
<link rel="stylesheet" href="/css/dashboard.css" />
|
||||||
</head>
|
</head>
|
||||||
<body>
|
<body class="dashboard-body">
|
||||||
<main>
|
<main class="container py-5">
|
||||||
<h1>{HtmlEncoder.Default.Encode(title)}</h1>
|
<h1 class="h3 mb-4">{HtmlEncoder.Default.Encode(title)}</h1>
|
||||||
{body}
|
{body}
|
||||||
</main>
|
</main>
|
||||||
|
<script src="/lib/bootstrap/js/bootstrap.bundle.min.js"></script>
|
||||||
</body>
|
</body>
|
||||||
</html>
|
</html>
|
||||||
""";
|
""";
|
||||||
|
|||||||
@@ -13,6 +13,9 @@ public static class DashboardServiceCollectionExtensions
|
|||||||
services.AddSingleton<IDashboardAuthenticator, DashboardAuthenticator>();
|
services.AddSingleton<IDashboardAuthenticator, DashboardAuthenticator>();
|
||||||
services.AddHttpContextAccessor();
|
services.AddHttpContextAccessor();
|
||||||
services.AddAntiforgery();
|
services.AddAntiforgery();
|
||||||
|
services.AddCascadingAuthenticationState();
|
||||||
|
services.AddRazorComponents()
|
||||||
|
.AddInteractiveServerComponents();
|
||||||
services
|
services
|
||||||
.AddAuthentication(DashboardAuthenticationDefaults.AuthenticationScheme)
|
.AddAuthentication(DashboardAuthenticationDefaults.AuthenticationScheme)
|
||||||
.AddCookie(DashboardAuthenticationDefaults.AuthenticationScheme);
|
.AddCookie(DashboardAuthenticationDefaults.AuthenticationScheme);
|
||||||
|
|||||||
@@ -19,8 +19,10 @@ public static class GatewayApplication
|
|||||||
WebApplication app = builder.Build();
|
WebApplication app = builder.Build();
|
||||||
|
|
||||||
app.UseGatewayRequestLoggingScope();
|
app.UseGatewayRequestLoggingScope();
|
||||||
|
app.UseStaticFiles();
|
||||||
app.UseAuthentication();
|
app.UseAuthentication();
|
||||||
app.UseAuthorization();
|
app.UseAuthorization();
|
||||||
|
app.UseAntiforgery();
|
||||||
app.MapGatewayEndpoints();
|
app.MapGatewayEndpoints();
|
||||||
|
|
||||||
return app;
|
return app;
|
||||||
|
|||||||
@@ -0,0 +1,14 @@
|
|||||||
|
{
|
||||||
|
"version": "1.0",
|
||||||
|
"defaultProvider": "cdnjs",
|
||||||
|
"libraries": [
|
||||||
|
{
|
||||||
|
"library": "bootstrap@5.3.3",
|
||||||
|
"destination": "wwwroot/lib/bootstrap/",
|
||||||
|
"files": [
|
||||||
|
"css/bootstrap.min.css",
|
||||||
|
"js/bootstrap.bundle.min.js"
|
||||||
|
]
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
@@ -0,0 +1,138 @@
|
|||||||
|
:root {
|
||||||
|
--mxgw-surface: #f7f8fa;
|
||||||
|
--mxgw-border: #d8dee6;
|
||||||
|
--mxgw-ink-muted: #667085;
|
||||||
|
--mxgw-accent: #146c64;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-body {
|
||||||
|
background: var(--mxgw-surface);
|
||||||
|
color: #1f2933;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-navbar {
|
||||||
|
min-height: 3.5rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-content {
|
||||||
|
padding: 1.25rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-page-header {
|
||||||
|
align-items: center;
|
||||||
|
display: flex;
|
||||||
|
gap: 1rem;
|
||||||
|
justify-content: space-between;
|
||||||
|
margin-bottom: 1rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-page-header h1,
|
||||||
|
.section-heading h2 {
|
||||||
|
font-size: 1.35rem;
|
||||||
|
font-weight: 650;
|
||||||
|
letter-spacing: 0;
|
||||||
|
margin: 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
.section-heading {
|
||||||
|
margin-bottom: .75rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-section {
|
||||||
|
background: #fff;
|
||||||
|
border-top: 1px solid var(--mxgw-border);
|
||||||
|
margin-top: 1rem;
|
||||||
|
padding: 1rem 0 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-grid {
|
||||||
|
display: grid;
|
||||||
|
gap: .75rem;
|
||||||
|
grid-template-columns: repeat(auto-fit, minmax(12rem, 1fr));
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-grid.compact {
|
||||||
|
grid-template-columns: repeat(auto-fit, minmax(10rem, 1fr));
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-card {
|
||||||
|
border-color: var(--mxgw-border);
|
||||||
|
border-radius: .375rem;
|
||||||
|
box-shadow: none;
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-label {
|
||||||
|
color: var(--mxgw-ink-muted);
|
||||||
|
font-size: .78rem;
|
||||||
|
font-weight: 650;
|
||||||
|
letter-spacing: 0;
|
||||||
|
text-transform: uppercase;
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-value {
|
||||||
|
color: var(--mxgw-accent);
|
||||||
|
font-size: 1.7rem;
|
||||||
|
font-weight: 700;
|
||||||
|
letter-spacing: 0;
|
||||||
|
line-height: 1.25;
|
||||||
|
overflow-wrap: anywhere;
|
||||||
|
}
|
||||||
|
|
||||||
|
.metric-detail {
|
||||||
|
color: var(--mxgw-ink-muted);
|
||||||
|
font-size: .85rem;
|
||||||
|
overflow-wrap: anywhere;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-table {
|
||||||
|
--bs-table-bg: #fff;
|
||||||
|
border-color: var(--mxgw-border);
|
||||||
|
margin-bottom: 0;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-table th {
|
||||||
|
color: #344054;
|
||||||
|
font-weight: 650;
|
||||||
|
white-space: nowrap;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-table td {
|
||||||
|
max-width: 24rem;
|
||||||
|
overflow-wrap: anywhere;
|
||||||
|
}
|
||||||
|
|
||||||
|
.details-table th {
|
||||||
|
width: 14rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.empty-state {
|
||||||
|
background: #fff;
|
||||||
|
border: 1px dashed var(--mxgw-border);
|
||||||
|
border-radius: .375rem;
|
||||||
|
color: var(--mxgw-ink-muted);
|
||||||
|
padding: 1rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-login {
|
||||||
|
max-width: 28rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.login-card {
|
||||||
|
border-color: var(--mxgw-border);
|
||||||
|
border-radius: .375rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
@media (max-width: 700px) {
|
||||||
|
.dashboard-content {
|
||||||
|
padding: .75rem;
|
||||||
|
}
|
||||||
|
|
||||||
|
.dashboard-page-header {
|
||||||
|
align-items: flex-start;
|
||||||
|
flex-direction: column;
|
||||||
|
}
|
||||||
|
|
||||||
|
.details-table th {
|
||||||
|
width: 9rem;
|
||||||
|
}
|
||||||
|
}
|
||||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -33,6 +33,37 @@ public sealed class GatewayApplicationTests
|
|||||||
Assert.NotNull(metrics);
|
Assert.NotNull(metrics);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Build_WhenDashboardEnabled_MapsBlazorDashboardAndAuthEndpoints()
|
||||||
|
{
|
||||||
|
WebApplication app = GatewayApplication.Build([]);
|
||||||
|
IReadOnlyList<RouteEndpoint> endpoints = GetRouteEndpoints(app);
|
||||||
|
|
||||||
|
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/");
|
||||||
|
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/sessions");
|
||||||
|
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/workers");
|
||||||
|
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/events");
|
||||||
|
Assert.Contains(endpoints, endpoint => endpoint.RoutePattern.RawText == "/dashboard/settings");
|
||||||
|
Assert.Contains(endpoints, endpoint =>
|
||||||
|
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName == "DashboardLogin");
|
||||||
|
Assert.Contains(endpoints, endpoint =>
|
||||||
|
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName == "DashboardLogout");
|
||||||
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void Build_WhenDashboardDisabled_DoesNotMapDashboardRoutes()
|
||||||
|
{
|
||||||
|
WebApplication app = GatewayApplication.Build(["--MxGateway:Dashboard:Enabled=false"]);
|
||||||
|
IReadOnlyList<RouteEndpoint> endpoints = GetRouteEndpoints(app);
|
||||||
|
|
||||||
|
Assert.DoesNotContain(endpoints, endpoint =>
|
||||||
|
endpoint.RoutePattern.RawText?.StartsWith("/dashboard", StringComparison.Ordinal) == true);
|
||||||
|
Assert.DoesNotContain(endpoints, endpoint =>
|
||||||
|
endpoint.Metadata.GetMetadata<IEndpointNameMetadata>()?.EndpointName?.StartsWith(
|
||||||
|
"Dashboard",
|
||||||
|
StringComparison.Ordinal) == true);
|
||||||
|
}
|
||||||
|
|
||||||
[Theory]
|
[Theory]
|
||||||
[InlineData(
|
[InlineData(
|
||||||
"MxGateway:Worker:ExecutablePath",
|
"MxGateway:Worker:ExecutablePath",
|
||||||
@@ -65,4 +96,12 @@ public sealed class GatewayApplicationTests
|
|||||||
exception.Failures,
|
exception.Failures,
|
||||||
failure => failure.Contains(expectedFailure, StringComparison.Ordinal));
|
failure => failure.Contains(expectedFailure, StringComparison.Ordinal));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static IReadOnlyList<RouteEndpoint> GetRouteEndpoints(WebApplication app)
|
||||||
|
{
|
||||||
|
return ((IEndpointRouteBuilder)app).DataSources
|
||||||
|
.SelectMany(dataSource => dataSource.Endpoints)
|
||||||
|
.OfType<RouteEndpoint>()
|
||||||
|
.ToArray();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -105,25 +105,6 @@ public sealed class FakeWorkerHarnessTests
|
|||||||
Assert.Equal(WorkerClientState.Faulted, client.State);
|
Assert.Equal(WorkerClientState.Faulted, client.State);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task SendHeartbeatAsync_UpdatesClientHeartbeatState()
|
|
||||||
{
|
|
||||||
await using FakeWorkerHarness fakeWorker = await FakeWorkerHarness.CreateConnectedPairAsync();
|
|
||||||
await using WorkerClient client = fakeWorker.CreateClient();
|
|
||||||
await StartClientAsync(fakeWorker, client);
|
|
||||||
DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;
|
|
||||||
|
|
||||||
await Task.Delay(TimeSpan.FromMilliseconds(20));
|
|
||||||
await fakeWorker.SendHeartbeatAsync(
|
|
||||||
configureHeartbeat: heartbeat => heartbeat.WorkerProcessId = 2468);
|
|
||||||
|
|
||||||
await WaitUntilAsync(
|
|
||||||
() => client.ProcessId == 2468 && client.LastHeartbeatAt > previousHeartbeat,
|
|
||||||
TestTimeout);
|
|
||||||
|
|
||||||
Assert.Equal(WorkerClientState.Ready, client.State);
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task InvokeAsync_WithHungWorker_TimesOutPendingCommand()
|
public async Task InvokeAsync_WithHungWorker_TimesOutPendingCommand()
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -284,26 +284,6 @@ public sealed class FakeWorkerHarness : IAsyncDisposable
|
|||||||
cancellationToken).ConfigureAwait(false);
|
cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
public async Task SendHeartbeatAsync(
|
|
||||||
WorkerState state = WorkerState.Ready,
|
|
||||||
CancellationToken cancellationToken = default,
|
|
||||||
Action<WorkerHeartbeat>? configureHeartbeat = null)
|
|
||||||
{
|
|
||||||
WorkerHeartbeat heartbeat = new()
|
|
||||||
{
|
|
||||||
WorkerProcessId = DefaultWorkerProcessId,
|
|
||||||
State = state,
|
|
||||||
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
||||||
};
|
|
||||||
configureHeartbeat?.Invoke(heartbeat);
|
|
||||||
|
|
||||||
await _writer.WriteAsync(
|
|
||||||
CreateEnvelope(
|
|
||||||
correlationId: string.Empty,
|
|
||||||
envelope => envelope.WorkerHeartbeat = heartbeat),
|
|
||||||
cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
public async Task SendShutdownAckAsync(
|
public async Task SendShutdownAckAsync(
|
||||||
ProtocolStatusCode statusCode = ProtocolStatusCode.Ok,
|
ProtocolStatusCode statusCode = ProtocolStatusCode.Ok,
|
||||||
CancellationToken cancellationToken = default)
|
CancellationToken cancellationToken = default)
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
using System.IO.Pipes;
|
using System.IO.Pipes;
|
||||||
using Google.Protobuf.WellKnownTypes;
|
|
||||||
using MxGateway.Contracts;
|
using MxGateway.Contracts;
|
||||||
using MxGateway.Contracts.Proto;
|
using MxGateway.Contracts.Proto;
|
||||||
using MxGateway.Server.Workers;
|
using MxGateway.Server.Workers;
|
||||||
@@ -152,24 +151,6 @@ public sealed class WorkerClientTests
|
|||||||
Assert.Equal(WorkerClientState.Faulted, client.State);
|
Assert.Equal(WorkerClientState.Faulted, client.State);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task ReadLoop_WhenHeartbeatArrives_UpdatesLastHeartbeatAndWorkerProcess()
|
|
||||||
{
|
|
||||||
await using PipePair pipePair = await PipePair.CreateAsync();
|
|
||||||
await using WorkerClient client = CreateClient(pipePair);
|
|
||||||
await CompleteHandshakeAsync(client, pipePair);
|
|
||||||
DateTimeOffset previousHeartbeat = client.LastHeartbeatAt;
|
|
||||||
|
|
||||||
await Task.Delay(TimeSpan.FromMilliseconds(20));
|
|
||||||
await pipePair.WorkerWriter.WriteAsync(CreateHeartbeatEnvelope(workerProcessId: 9876));
|
|
||||||
|
|
||||||
await WaitUntilAsync(
|
|
||||||
() => client.ProcessId == 9876 && client.LastHeartbeatAt > previousHeartbeat,
|
|
||||||
TestTimeout);
|
|
||||||
|
|
||||||
Assert.Equal(WorkerClientState.Ready, client.State);
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public async Task HeartbeatMonitor_WhenHeartbeatExpires_FaultsClient()
|
public async Task HeartbeatMonitor_WhenHeartbeatExpires_FaultsClient()
|
||||||
{
|
{
|
||||||
@@ -295,21 +276,6 @@ public sealed class WorkerClientTests
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
private static WorkerEnvelope CreateHeartbeatEnvelope(int workerProcessId)
|
|
||||||
{
|
|
||||||
return CreateWorkerEnvelope(
|
|
||||||
correlationId: string.Empty,
|
|
||||||
sequence: 20,
|
|
||||||
envelope => envelope.WorkerHeartbeat = new WorkerHeartbeat
|
|
||||||
{
|
|
||||||
WorkerProcessId = workerProcessId,
|
|
||||||
State = WorkerState.Ready,
|
|
||||||
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
||||||
PendingCommandCount = 0,
|
|
||||||
OutboundEventQueueDepth = 0,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerEnvelope CreateWorkerEnvelope(
|
private static WorkerEnvelope CreateWorkerEnvelope(
|
||||||
string correlationId,
|
string correlationId,
|
||||||
ulong sequence,
|
ulong sequence,
|
||||||
|
|||||||
@@ -30,7 +30,7 @@ public sealed class WorkerApplicationTests
|
|||||||
Assert.Equal("mxaccess-gateway-123-session-1", entry.Fields["pipe_name"]);
|
Assert.Equal("mxaccess-gateway-123-session-1", entry.Fields["pipe_name"]);
|
||||||
Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, entry.Fields["protocol_version"]);
|
Assert.Equal(GatewayContractInfo.WorkerProtocolVersion, entry.Fields["protocol_version"]);
|
||||||
Assert.Equal("[redacted]", entry.Fields["nonce"]);
|
Assert.Equal("[redacted]", entry.Fields["nonce"]);
|
||||||
Assert.Equal("WorkerPipeSessionCompleted", logger.Entries[1].EventName);
|
Assert.Equal("WorkerPipeHandshakeSucceeded", logger.Entries[1].EventName);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
@@ -1,15 +1,10 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.IO;
|
|
||||||
using System.IO.Pipes;
|
using System.IO.Pipes;
|
||||||
using System.Threading;
|
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Google.Protobuf.WellKnownTypes;
|
|
||||||
using MxGateway.Contracts;
|
using MxGateway.Contracts;
|
||||||
using MxGateway.Contracts.Proto;
|
using MxGateway.Contracts.Proto;
|
||||||
using MxGateway.Worker.Bootstrap;
|
using MxGateway.Worker.Bootstrap;
|
||||||
using MxGateway.Worker.Ipc;
|
using MxGateway.Worker.Ipc;
|
||||||
using MxGateway.Worker.MxAccess;
|
|
||||||
using MxGateway.Worker.Sta;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.Tests.Ipc;
|
namespace MxGateway.Worker.Tests.Ipc;
|
||||||
|
|
||||||
@@ -33,9 +28,7 @@ public sealed class WorkerPipeClientTests
|
|||||||
PipeTransmissionMode.Byte,
|
PipeTransmissionMode.Byte,
|
||||||
PipeOptions.Asynchronous);
|
PipeOptions.Asynchronous);
|
||||||
|
|
||||||
WorkerPipeClient client = new(
|
WorkerPipeClient client = new(connectTimeoutMilliseconds: 5000);
|
||||||
connectTimeoutMilliseconds: 5000,
|
|
||||||
(stream, options) => CreateSession(stream, options));
|
|
||||||
Task clientTask = client.RunAsync(workerOptions);
|
Task clientTask = client.RunAsync(workerOptions);
|
||||||
|
|
||||||
await Task.Factory.FromAsync(server.BeginWaitForConnection, server.EndWaitForConnection, null);
|
await Task.Factory.FromAsync(server.BeginWaitForConnection, server.EndWaitForConnection, null);
|
||||||
@@ -63,86 +56,6 @@ public sealed class WorkerPipeClientTests
|
|||||||
WorkerEnvelope ready = await reader.ReadAsync();
|
WorkerEnvelope ready = await reader.ReadAsync();
|
||||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase);
|
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase);
|
||||||
|
|
||||||
await writer.WriteAsync(new WorkerEnvelope
|
|
||||||
{
|
|
||||||
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
|
|
||||||
SessionId = "session-1",
|
|
||||||
Sequence = 2,
|
|
||||||
WorkerShutdown = new WorkerShutdown
|
|
||||||
{
|
|
||||||
GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(1)),
|
|
||||||
Reason = "test-complete",
|
|
||||||
},
|
|
||||||
});
|
|
||||||
|
|
||||||
WorkerEnvelope shutdownAck = await reader.ReadAsync();
|
|
||||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerShutdownAck, shutdownAck.BodyCase);
|
|
||||||
await clientTask;
|
await clientTask;
|
||||||
}
|
}
|
||||||
|
|
||||||
private static WorkerPipeSession CreateSession(
|
|
||||||
Stream stream,
|
|
||||||
WorkerFrameProtocolOptions options)
|
|
||||||
{
|
|
||||||
return new WorkerPipeSession(
|
|
||||||
new WorkerFrameReader(stream, options),
|
|
||||||
new WorkerFrameWriter(stream, options),
|
|
||||||
options,
|
|
||||||
() => 1234,
|
|
||||||
new WorkerPipeSessionOptions
|
|
||||||
{
|
|
||||||
HeartbeatInterval = TimeSpan.FromSeconds(30),
|
|
||||||
HeartbeatGrace = TimeSpan.FromSeconds(30),
|
|
||||||
},
|
|
||||||
() => new FakeRuntimeSession());
|
|
||||||
}
|
|
||||||
|
|
||||||
private sealed class FakeRuntimeSession : IWorkerRuntimeSession
|
|
||||||
{
|
|
||||||
public Task<WorkerReady> StartAsync(
|
|
||||||
int workerProcessId,
|
|
||||||
CancellationToken cancellationToken = default)
|
|
||||||
{
|
|
||||||
return Task.FromResult(new WorkerReady
|
|
||||||
{
|
|
||||||
WorkerProcessId = workerProcessId,
|
|
||||||
MxaccessProgid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.ProgId,
|
|
||||||
MxaccessClsid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.Clsid,
|
|
||||||
ReadyTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
public Task<MxCommandReply> DispatchAsync(StaCommand command)
|
|
||||||
{
|
|
||||||
return Task.FromResult(new MxCommandReply
|
|
||||||
{
|
|
||||||
SessionId = command.SessionId,
|
|
||||||
CorrelationId = command.CorrelationId,
|
|
||||||
Kind = command.Kind,
|
|
||||||
ProtocolStatus = new ProtocolStatus
|
|
||||||
{
|
|
||||||
Code = ProtocolStatusCode.Ok,
|
|
||||||
Message = "OK",
|
|
||||||
},
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
|
|
||||||
{
|
|
||||||
return new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset.UtcNow,
|
|
||||||
pendingCommandCount: 0,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
currentCommandCorrelationId: string.Empty);
|
|
||||||
}
|
|
||||||
|
|
||||||
public void RequestShutdown()
|
|
||||||
{
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Dispose()
|
|
||||||
{
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,16 +1,11 @@
|
|||||||
using System;
|
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
using System.IO;
|
using System.IO;
|
||||||
using System.IO.Pipes;
|
|
||||||
using System.Runtime.InteropServices;
|
using System.Runtime.InteropServices;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using Google.Protobuf.WellKnownTypes;
|
|
||||||
using MxGateway.Contracts;
|
using MxGateway.Contracts;
|
||||||
using MxGateway.Contracts.Proto;
|
using MxGateway.Contracts.Proto;
|
||||||
using MxGateway.Worker.Ipc;
|
using MxGateway.Worker.Ipc;
|
||||||
using MxGateway.Worker.MxAccess;
|
|
||||||
using MxGateway.Worker.Sta;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.Tests.Ipc;
|
namespace MxGateway.Worker.Tests.Ipc;
|
||||||
|
|
||||||
@@ -152,127 +147,6 @@ public sealed class WorkerPipeSessionTests
|
|||||||
Assert.Equal(ProtocolStatusCode.WorkerUnavailable, written[1].WorkerFault.ProtocolStatus.Code);
|
Assert.Equal(ProtocolStatusCode.WorkerUnavailable, written[1].WorkerFault.ProtocolStatus.Code);
|
||||||
}
|
}
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task RunAsync_SendsHeartbeatPayloadFromRuntimeSnapshot()
|
|
||||||
{
|
|
||||||
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
|
|
||||||
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
|
|
||||||
FakeRuntimeSession runtime = new();
|
|
||||||
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset.UtcNow,
|
|
||||||
pendingCommandCount: 2,
|
|
||||||
outboundEventQueueDepth: 3,
|
|
||||||
lastEventSequence: 42,
|
|
||||||
currentCommandCorrelationId: "current-command"));
|
|
||||||
WorkerPipeSession session = CreatePipeSession(
|
|
||||||
pipePair.WorkerStream,
|
|
||||||
runtime,
|
|
||||||
new WorkerPipeSessionOptions
|
|
||||||
{
|
|
||||||
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
|
|
||||||
HeartbeatGrace = TimeSpan.FromSeconds(5),
|
|
||||||
});
|
|
||||||
Task runTask = session.RunAsync(cancellation.Token);
|
|
||||||
|
|
||||||
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
|
|
||||||
await ThrowIfCompletedAsync(runTask);
|
|
||||||
|
|
||||||
WorkerEnvelope heartbeat = await ReadUntilAsync(
|
|
||||||
pipePair.GatewayReader,
|
|
||||||
WorkerEnvelope.BodyOneofCase.WorkerHeartbeat,
|
|
||||||
cancellation.Token);
|
|
||||||
|
|
||||||
Assert.Equal(WorkerState.ExecutingCommand, heartbeat.WorkerHeartbeat.State);
|
|
||||||
Assert.Equal(1234, heartbeat.WorkerHeartbeat.WorkerProcessId);
|
|
||||||
Assert.Equal(2u, heartbeat.WorkerHeartbeat.PendingCommandCount);
|
|
||||||
Assert.Equal(3u, heartbeat.WorkerHeartbeat.OutboundEventQueueDepth);
|
|
||||||
Assert.Equal(42UL, heartbeat.WorkerHeartbeat.LastEventSequence);
|
|
||||||
Assert.Equal("current-command", heartbeat.WorkerHeartbeat.CurrentCommandCorrelationId);
|
|
||||||
|
|
||||||
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task RunAsync_WhenCommandIsExecuting_HeartbeatReportsCurrentCorrelation()
|
|
||||||
{
|
|
||||||
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
|
|
||||||
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
|
|
||||||
FakeRuntimeSession runtime = new()
|
|
||||||
{
|
|
||||||
BlockDispatch = true,
|
|
||||||
};
|
|
||||||
WorkerPipeSession session = CreatePipeSession(
|
|
||||||
pipePair.WorkerStream,
|
|
||||||
runtime,
|
|
||||||
new WorkerPipeSessionOptions
|
|
||||||
{
|
|
||||||
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
|
|
||||||
HeartbeatGrace = TimeSpan.FromSeconds(5),
|
|
||||||
});
|
|
||||||
Task runTask = session.RunAsync(cancellation.Token);
|
|
||||||
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
|
|
||||||
|
|
||||||
await pipePair.GatewayWriter.WriteAsync(
|
|
||||||
CreateCommandEnvelope("command-1"),
|
|
||||||
cancellation.Token);
|
|
||||||
|
|
||||||
Assert.True(runtime.DispatchStarted.Wait(TimeSpan.FromSeconds(2)));
|
|
||||||
WorkerEnvelope heartbeat = await ReadUntilAsync(
|
|
||||||
pipePair.GatewayReader,
|
|
||||||
WorkerEnvelope.BodyOneofCase.WorkerHeartbeat,
|
|
||||||
envelope => envelope.WorkerHeartbeat.CurrentCommandCorrelationId == "command-1",
|
|
||||||
cancellation.Token);
|
|
||||||
|
|
||||||
Assert.Equal("command-1", heartbeat.WorkerHeartbeat.CurrentCommandCorrelationId);
|
|
||||||
Assert.Equal(WorkerState.ExecutingCommand, heartbeat.WorkerHeartbeat.State);
|
|
||||||
|
|
||||||
runtime.ReleaseDispatch();
|
|
||||||
WorkerEnvelope reply = await ReadUntilAsync(
|
|
||||||
pipePair.GatewayReader,
|
|
||||||
WorkerEnvelope.BodyOneofCase.WorkerCommandReply,
|
|
||||||
cancellation.Token);
|
|
||||||
|
|
||||||
Assert.Equal("command-1", reply.CorrelationId);
|
|
||||||
Assert.Equal(ProtocolStatusCode.Ok, reply.WorkerCommandReply.Reply.ProtocolStatus.Code);
|
|
||||||
|
|
||||||
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
|
|
||||||
}
|
|
||||||
|
|
||||||
[Fact]
|
|
||||||
public async Task RunAsync_WhenStaActivityIsStale_WritesWatchdogFault()
|
|
||||||
{
|
|
||||||
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5));
|
|
||||||
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
|
|
||||||
FakeRuntimeSession runtime = new();
|
|
||||||
runtime.SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset.UtcNow - TimeSpan.FromSeconds(5),
|
|
||||||
pendingCommandCount: 0,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
currentCommandCorrelationId: "stuck-command"));
|
|
||||||
WorkerPipeSession session = CreatePipeSession(
|
|
||||||
pipePair.WorkerStream,
|
|
||||||
runtime,
|
|
||||||
new WorkerPipeSessionOptions
|
|
||||||
{
|
|
||||||
HeartbeatInterval = TimeSpan.FromMilliseconds(20),
|
|
||||||
HeartbeatGrace = TimeSpan.FromMilliseconds(50),
|
|
||||||
});
|
|
||||||
Task runTask = session.RunAsync(cancellation.Token);
|
|
||||||
await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token);
|
|
||||||
|
|
||||||
WorkerEnvelope fault = await ReadUntilAsync(
|
|
||||||
pipePair.GatewayReader,
|
|
||||||
WorkerEnvelope.BodyOneofCase.WorkerFault,
|
|
||||||
cancellation.Token);
|
|
||||||
|
|
||||||
Assert.Equal(WorkerFaultCategory.StaHung, fault.WorkerFault.Category);
|
|
||||||
Assert.Equal("stuck-command", fault.WorkerFault.CommandMethod);
|
|
||||||
Assert.Contains("STA activity is stale", fault.WorkerFault.DiagnosticMessage);
|
|
||||||
|
|
||||||
await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerPipeSession CreateSession(
|
private static WorkerPipeSession CreateSession(
|
||||||
Stream inbound,
|
Stream inbound,
|
||||||
Stream outbound,
|
Stream outbound,
|
||||||
@@ -285,21 +159,6 @@ public sealed class WorkerPipeSessionTests
|
|||||||
() => 1234);
|
() => 1234);
|
||||||
}
|
}
|
||||||
|
|
||||||
private static WorkerPipeSession CreatePipeSession(
|
|
||||||
Stream stream,
|
|
||||||
FakeRuntimeSession runtime,
|
|
||||||
WorkerPipeSessionOptions sessionOptions)
|
|
||||||
{
|
|
||||||
WorkerFrameProtocolOptions options = CreateOptions();
|
|
||||||
return new WorkerPipeSession(
|
|
||||||
new WorkerFrameReader(stream, options),
|
|
||||||
new WorkerFrameWriter(stream, options),
|
|
||||||
options,
|
|
||||||
() => 1234,
|
|
||||||
sessionOptions,
|
|
||||||
() => runtime);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerFrameProtocolOptions CreateOptions()
|
private static WorkerFrameProtocolOptions CreateOptions()
|
||||||
{
|
{
|
||||||
return new WorkerFrameProtocolOptions(
|
return new WorkerFrameProtocolOptions(
|
||||||
@@ -326,119 +185,6 @@ public sealed class WorkerPipeSessionTests
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
private static WorkerEnvelope CreateCommandEnvelope(string correlationId)
|
|
||||||
{
|
|
||||||
return new WorkerEnvelope
|
|
||||||
{
|
|
||||||
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
|
|
||||||
SessionId = SessionId,
|
|
||||||
Sequence = 2,
|
|
||||||
CorrelationId = correlationId,
|
|
||||||
WorkerCommand = new WorkerCommand
|
|
||||||
{
|
|
||||||
Command = new MxCommand
|
|
||||||
{
|
|
||||||
Kind = MxCommandKind.Ping,
|
|
||||||
Ping = new PingCommand
|
|
||||||
{
|
|
||||||
Message = "ping",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
EnqueueTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerEnvelope CreateShutdownEnvelope()
|
|
||||||
{
|
|
||||||
return new WorkerEnvelope
|
|
||||||
{
|
|
||||||
ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion,
|
|
||||||
SessionId = SessionId,
|
|
||||||
Sequence = 3,
|
|
||||||
WorkerShutdown = new WorkerShutdown
|
|
||||||
{
|
|
||||||
GracePeriod = Duration.FromTimeSpan(TimeSpan.FromSeconds(1)),
|
|
||||||
Reason = "test-complete",
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
private static async Task CompleteGatewayHandshakeAsync(
|
|
||||||
PipePair pipePair,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
await pipePair.GatewayWriter
|
|
||||||
.WriteAsync(CreateGatewayHelloEnvelope(), cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
|
|
||||||
WorkerEnvelope hello = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
WorkerEnvelope ready = await pipePair.GatewayReader.ReadAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
|
|
||||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerHello, hello.BodyCase);
|
|
||||||
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerReady, ready.BodyCase);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static async Task SendShutdownAndWaitAsync(
|
|
||||||
PipePair pipePair,
|
|
||||||
Task runTask,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
await pipePair.GatewayWriter
|
|
||||||
.WriteAsync(CreateShutdownEnvelope(), cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
|
|
||||||
WorkerEnvelope shutdownAck = await ReadUntilAsync(
|
|
||||||
pipePair.GatewayReader,
|
|
||||||
WorkerEnvelope.BodyOneofCase.WorkerShutdownAck,
|
|
||||||
cancellationToken);
|
|
||||||
|
|
||||||
Assert.Equal(ProtocolStatusCode.Ok, shutdownAck.WorkerShutdownAck.Status.Code);
|
|
||||||
Task completedTask = await Task
|
|
||||||
.WhenAny(runTask, Task.Delay(TimeSpan.FromSeconds(2), cancellationToken))
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
|
|
||||||
Assert.Same(runTask, completedTask);
|
|
||||||
await runTask.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static async Task ThrowIfCompletedAsync(Task task)
|
|
||||||
{
|
|
||||||
await Task.Delay(TimeSpan.FromMilliseconds(100));
|
|
||||||
if (task.IsCompleted)
|
|
||||||
{
|
|
||||||
await task;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private static Task<WorkerEnvelope> ReadUntilAsync(
|
|
||||||
WorkerFrameReader reader,
|
|
||||||
WorkerEnvelope.BodyOneofCase expectedBody,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
return ReadUntilAsync(
|
|
||||||
reader,
|
|
||||||
expectedBody,
|
|
||||||
_ => true,
|
|
||||||
cancellationToken);
|
|
||||||
}
|
|
||||||
|
|
||||||
private static async Task<WorkerEnvelope> ReadUntilAsync(
|
|
||||||
WorkerFrameReader reader,
|
|
||||||
WorkerEnvelope.BodyOneofCase expectedBody,
|
|
||||||
Func<WorkerEnvelope, bool> predicate,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
while (true)
|
|
||||||
{
|
|
||||||
WorkerEnvelope envelope = await reader.ReadAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
if (envelope.BodyCase == expectedBody && predicate(envelope))
|
|
||||||
{
|
|
||||||
return envelope;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerEnvelope[] ReadWrittenFrames(
|
private static WorkerEnvelope[] ReadWrittenFrames(
|
||||||
MemoryStream stream,
|
MemoryStream stream,
|
||||||
WorkerFrameProtocolOptions options)
|
WorkerFrameProtocolOptions options)
|
||||||
@@ -473,157 +219,4 @@ public sealed class WorkerPipeSessionTests
|
|||||||
buffer[2] = (byte)(value >> 16);
|
buffer[2] = (byte)(value >> 16);
|
||||||
buffer[3] = (byte)(value >> 24);
|
buffer[3] = (byte)(value >> 24);
|
||||||
}
|
}
|
||||||
|
|
||||||
private sealed class FakeRuntimeSession : IWorkerRuntimeSession
|
|
||||||
{
|
|
||||||
private readonly ManualResetEventSlim releaseDispatch = new(false);
|
|
||||||
private readonly object gate = new();
|
|
||||||
private WorkerRuntimeHeartbeatSnapshot snapshot = new(
|
|
||||||
DateTimeOffset.UtcNow,
|
|
||||||
pendingCommandCount: 0,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
currentCommandCorrelationId: string.Empty);
|
|
||||||
|
|
||||||
public ManualResetEventSlim DispatchStarted { get; } = new(false);
|
|
||||||
|
|
||||||
public bool BlockDispatch { get; set; }
|
|
||||||
|
|
||||||
public Task<WorkerReady> StartAsync(
|
|
||||||
int workerProcessId,
|
|
||||||
CancellationToken cancellationToken = default)
|
|
||||||
{
|
|
||||||
return Task.FromResult(new WorkerReady
|
|
||||||
{
|
|
||||||
WorkerProcessId = workerProcessId,
|
|
||||||
MxaccessProgid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.ProgId,
|
|
||||||
MxaccessClsid = MxGateway.Worker.MxAccess.MxAccessInteropInfo.Clsid,
|
|
||||||
ReadyTimestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
public Task<MxCommandReply> DispatchAsync(StaCommand command)
|
|
||||||
{
|
|
||||||
return Task.Run(
|
|
||||||
() =>
|
|
||||||
{
|
|
||||||
SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset.UtcNow,
|
|
||||||
pendingCommandCount: 0,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
command.CorrelationId));
|
|
||||||
DispatchStarted.Set();
|
|
||||||
|
|
||||||
if (BlockDispatch)
|
|
||||||
{
|
|
||||||
releaseDispatch.Wait(TimeSpan.FromSeconds(5));
|
|
||||||
}
|
|
||||||
|
|
||||||
SetSnapshot(new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset.UtcNow,
|
|
||||||
pendingCommandCount: 0,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
currentCommandCorrelationId: string.Empty));
|
|
||||||
|
|
||||||
return new MxCommandReply
|
|
||||||
{
|
|
||||||
SessionId = command.SessionId,
|
|
||||||
CorrelationId = command.CorrelationId,
|
|
||||||
Kind = command.Kind,
|
|
||||||
ProtocolStatus = new ProtocolStatus
|
|
||||||
{
|
|
||||||
Code = ProtocolStatusCode.Ok,
|
|
||||||
Message = "OK",
|
|
||||||
},
|
|
||||||
};
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
|
|
||||||
{
|
|
||||||
lock (gate)
|
|
||||||
{
|
|
||||||
return snapshot;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public void RequestShutdown()
|
|
||||||
{
|
|
||||||
releaseDispatch.Set();
|
|
||||||
}
|
|
||||||
|
|
||||||
public void ReleaseDispatch()
|
|
||||||
{
|
|
||||||
releaseDispatch.Set();
|
|
||||||
}
|
|
||||||
|
|
||||||
public void SetSnapshot(WorkerRuntimeHeartbeatSnapshot value)
|
|
||||||
{
|
|
||||||
lock (gate)
|
|
||||||
{
|
|
||||||
snapshot = value;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Dispose()
|
|
||||||
{
|
|
||||||
releaseDispatch.Set();
|
|
||||||
releaseDispatch.Dispose();
|
|
||||||
DispatchStarted.Dispose();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private sealed class PipePair : IDisposable
|
|
||||||
{
|
|
||||||
private readonly NamedPipeServerStream gatewayStream;
|
|
||||||
|
|
||||||
private PipePair(
|
|
||||||
NamedPipeServerStream gatewayStream,
|
|
||||||
NamedPipeClientStream workerStream)
|
|
||||||
{
|
|
||||||
this.gatewayStream = gatewayStream;
|
|
||||||
WorkerStream = workerStream;
|
|
||||||
WorkerFrameProtocolOptions options = CreateOptions();
|
|
||||||
GatewayReader = new WorkerFrameReader(gatewayStream, options);
|
|
||||||
GatewayWriter = new WorkerFrameWriter(gatewayStream, options);
|
|
||||||
}
|
|
||||||
|
|
||||||
public Stream WorkerStream { get; }
|
|
||||||
|
|
||||||
public WorkerFrameReader GatewayReader { get; }
|
|
||||||
|
|
||||||
public WorkerFrameWriter GatewayWriter { get; }
|
|
||||||
|
|
||||||
public static async Task<PipePair> CreateAsync(CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
string pipeName = $"mxaccessgw-worker-session-tests-{Guid.NewGuid():N}";
|
|
||||||
NamedPipeServerStream gatewayStream = new(
|
|
||||||
pipeName,
|
|
||||||
PipeDirection.InOut,
|
|
||||||
maxNumberOfServerInstances: 1,
|
|
||||||
PipeTransmissionMode.Byte,
|
|
||||||
PipeOptions.Asynchronous);
|
|
||||||
NamedPipeClientStream workerStream = new(
|
|
||||||
".",
|
|
||||||
pipeName,
|
|
||||||
PipeDirection.InOut,
|
|
||||||
PipeOptions.Asynchronous);
|
|
||||||
|
|
||||||
Task waitForConnectionTask = gatewayStream.WaitForConnectionAsync();
|
|
||||||
await Task
|
|
||||||
.Run(() => workerStream.Connect(5000), cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
await waitForConnectionTask.ConfigureAwait(false);
|
|
||||||
|
|
||||||
return new PipePair(gatewayStream, workerStream);
|
|
||||||
}
|
|
||||||
|
|
||||||
public void Dispose()
|
|
||||||
{
|
|
||||||
WorkerStream.Dispose();
|
|
||||||
gatewayStream.Dispose();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,4 @@
|
|||||||
using System;
|
using System;
|
||||||
using System.IO;
|
|
||||||
using System.IO.Pipes;
|
using System.IO.Pipes;
|
||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
@@ -12,7 +11,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
|
|||||||
public const int DefaultConnectTimeoutMilliseconds = 30000;
|
public const int DefaultConnectTimeoutMilliseconds = 30000;
|
||||||
|
|
||||||
private readonly int _connectTimeoutMilliseconds;
|
private readonly int _connectTimeoutMilliseconds;
|
||||||
private readonly Func<Stream, WorkerFrameProtocolOptions, WorkerPipeSession> _sessionFactory;
|
|
||||||
|
|
||||||
public WorkerPipeClient()
|
public WorkerPipeClient()
|
||||||
: this(DefaultConnectTimeoutMilliseconds)
|
: this(DefaultConnectTimeoutMilliseconds)
|
||||||
@@ -20,15 +18,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
|
|||||||
}
|
}
|
||||||
|
|
||||||
public WorkerPipeClient(int connectTimeoutMilliseconds)
|
public WorkerPipeClient(int connectTimeoutMilliseconds)
|
||||||
: this(
|
|
||||||
connectTimeoutMilliseconds,
|
|
||||||
(stream, frameOptions) => new WorkerPipeSession(stream, frameOptions))
|
|
||||||
{
|
|
||||||
}
|
|
||||||
|
|
||||||
public WorkerPipeClient(
|
|
||||||
int connectTimeoutMilliseconds,
|
|
||||||
Func<Stream, WorkerFrameProtocolOptions, WorkerPipeSession> sessionFactory)
|
|
||||||
{
|
{
|
||||||
if (connectTimeoutMilliseconds <= 0)
|
if (connectTimeoutMilliseconds <= 0)
|
||||||
{
|
{
|
||||||
@@ -37,7 +26,6 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
|
|||||||
"Worker pipe connect timeout must be greater than zero.");
|
"Worker pipe connect timeout must be greater than zero.");
|
||||||
}
|
}
|
||||||
|
|
||||||
_sessionFactory = sessionFactory ?? throw new ArgumentNullException(nameof(sessionFactory));
|
|
||||||
_connectTimeoutMilliseconds = connectTimeoutMilliseconds;
|
_connectTimeoutMilliseconds = connectTimeoutMilliseconds;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -60,8 +48,8 @@ public sealed class WorkerPipeClient : IWorkerPipeClient
|
|||||||
|
|
||||||
await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false);
|
await ConnectAsync(pipe, cancellationToken).ConfigureAwait(false);
|
||||||
|
|
||||||
WorkerPipeSession session = _sessionFactory(pipe, frameOptions);
|
WorkerPipeSession session = new(pipe, frameOptions);
|
||||||
await session.RunAsync(cancellationToken).ConfigureAwait(false);
|
await session.CompleteStartupHandshakeAsync(cancellationToken).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
|
|
||||||
private Task ConnectAsync(
|
private Task ConnectAsync(
|
||||||
|
|||||||
@@ -6,7 +6,6 @@ using System.Threading.Tasks;
|
|||||||
using Google.Protobuf.WellKnownTypes;
|
using Google.Protobuf.WellKnownTypes;
|
||||||
using MxGateway.Contracts.Proto;
|
using MxGateway.Contracts.Proto;
|
||||||
using MxGateway.Worker.MxAccess;
|
using MxGateway.Worker.MxAccess;
|
||||||
using MxGateway.Worker.Sta;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.Ipc;
|
namespace MxGateway.Worker.Ipc;
|
||||||
|
|
||||||
@@ -14,14 +13,10 @@ public sealed class WorkerPipeSession
|
|||||||
{
|
{
|
||||||
private readonly WorkerFrameProtocolOptions _options;
|
private readonly WorkerFrameProtocolOptions _options;
|
||||||
private readonly Func<int> _processIdProvider;
|
private readonly Func<int> _processIdProvider;
|
||||||
private readonly Func<IWorkerRuntimeSession> _runtimeSessionFactory;
|
|
||||||
private readonly WorkerPipeSessionOptions _sessionOptions;
|
|
||||||
private readonly WorkerFrameReader _reader;
|
private readonly WorkerFrameReader _reader;
|
||||||
private readonly WorkerFrameWriter _writer;
|
private readonly WorkerFrameWriter _writer;
|
||||||
private IWorkerRuntimeSession? _runtimeSession;
|
private MxAccessStaSession? _mxAccessStaSession;
|
||||||
private long _nextSequence;
|
private long _nextSequence;
|
||||||
private WorkerState _state = WorkerState.Starting;
|
|
||||||
private bool _watchdogFaultSent;
|
|
||||||
|
|
||||||
public WorkerPipeSession(
|
public WorkerPipeSession(
|
||||||
Stream stream,
|
Stream stream,
|
||||||
@@ -39,49 +34,11 @@ public sealed class WorkerPipeSession
|
|||||||
WorkerFrameWriter writer,
|
WorkerFrameWriter writer,
|
||||||
WorkerFrameProtocolOptions options,
|
WorkerFrameProtocolOptions options,
|
||||||
Func<int> processIdProvider)
|
Func<int> processIdProvider)
|
||||||
: this(
|
|
||||||
reader,
|
|
||||||
writer,
|
|
||||||
options,
|
|
||||||
processIdProvider,
|
|
||||||
new WorkerPipeSessionOptions(),
|
|
||||||
() => new MxAccessStaSession())
|
|
||||||
{
|
|
||||||
}
|
|
||||||
|
|
||||||
public WorkerPipeSession(
|
|
||||||
WorkerFrameReader reader,
|
|
||||||
WorkerFrameWriter writer,
|
|
||||||
WorkerFrameProtocolOptions options,
|
|
||||||
Func<int> processIdProvider,
|
|
||||||
WorkerPipeSessionOptions sessionOptions,
|
|
||||||
Func<IWorkerRuntimeSession> runtimeSessionFactory)
|
|
||||||
{
|
{
|
||||||
_reader = reader ?? throw new ArgumentNullException(nameof(reader));
|
_reader = reader ?? throw new ArgumentNullException(nameof(reader));
|
||||||
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
|
_writer = writer ?? throw new ArgumentNullException(nameof(writer));
|
||||||
_options = options ?? throw new ArgumentNullException(nameof(options));
|
_options = options ?? throw new ArgumentNullException(nameof(options));
|
||||||
_processIdProvider = processIdProvider ?? throw new ArgumentNullException(nameof(processIdProvider));
|
_processIdProvider = processIdProvider ?? throw new ArgumentNullException(nameof(processIdProvider));
|
||||||
_sessionOptions = sessionOptions ?? throw new ArgumentNullException(nameof(sessionOptions));
|
|
||||||
_runtimeSessionFactory = runtimeSessionFactory ?? throw new ArgumentNullException(nameof(runtimeSessionFactory));
|
|
||||||
_sessionOptions.Validate();
|
|
||||||
}
|
|
||||||
|
|
||||||
public async Task RunAsync(CancellationToken cancellationToken = default)
|
|
||||||
{
|
|
||||||
_runtimeSession = _runtimeSessionFactory();
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await CompleteStartupHandshakeAsync(
|
|
||||||
token => _runtimeSession.StartAsync(_processIdProvider(), token),
|
|
||||||
cancellationToken).ConfigureAwait(false);
|
|
||||||
await RunMessageLoopAsync(cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
finally
|
|
||||||
{
|
|
||||||
_runtimeSession?.Dispose();
|
|
||||||
_runtimeSession = null;
|
|
||||||
_state = WorkerState.Stopped;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public Task CompleteStartupHandshakeAsync(CancellationToken cancellationToken = default)
|
public Task CompleteStartupHandshakeAsync(CancellationToken cancellationToken = default)
|
||||||
@@ -119,14 +76,11 @@ public sealed class WorkerPipeSession
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
|
WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
|
||||||
_state = WorkerState.Handshaking;
|
|
||||||
ValidateGatewayHello(envelope);
|
ValidateGatewayHello(envelope);
|
||||||
|
|
||||||
await WriteWorkerHelloAsync(cancellationToken).ConfigureAwait(false);
|
await WriteWorkerHelloAsync(cancellationToken).ConfigureAwait(false);
|
||||||
_state = WorkerState.InitializingSta;
|
|
||||||
WorkerReady ready = await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false);
|
WorkerReady ready = await initializeMxAccessAsync(cancellationToken).ConfigureAwait(false);
|
||||||
await WriteWorkerReadyAsync(ready, cancellationToken).ConfigureAwait(false);
|
await WriteWorkerReadyAsync(ready, cancellationToken).ConfigureAwait(false);
|
||||||
_state = WorkerState.Ready;
|
|
||||||
}
|
}
|
||||||
catch (WorkerFrameProtocolException exception)
|
catch (WorkerFrameProtocolException exception)
|
||||||
{
|
{
|
||||||
@@ -186,174 +140,6 @@ public sealed class WorkerPipeSession
|
|||||||
return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken);
|
return _writer.WriteAsync(CreateEnvelope(ready), cancellationToken);
|
||||||
}
|
}
|
||||||
|
|
||||||
private async Task RunMessageLoopAsync(CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
using CancellationTokenSource heartbeatCancellation = CancellationTokenSource
|
|
||||||
.CreateLinkedTokenSource(cancellationToken);
|
|
||||||
Task heartbeatTask = RunHeartbeatLoopAsync(heartbeatCancellation.Token);
|
|
||||||
|
|
||||||
try
|
|
||||||
{
|
|
||||||
while (!cancellationToken.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
Task<WorkerEnvelope> readTask = _reader.ReadAsync(cancellationToken);
|
|
||||||
Task completedTask = await Task.WhenAny(readTask, heartbeatTask).ConfigureAwait(false);
|
|
||||||
if (completedTask == heartbeatTask)
|
|
||||||
{
|
|
||||||
await heartbeatTask.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
WorkerEnvelope envelope = await readTask.ConfigureAwait(false);
|
|
||||||
bool keepReading = await DispatchGatewayEnvelopeAsync(envelope, cancellationToken).ConfigureAwait(false);
|
|
||||||
if (!keepReading)
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
finally
|
|
||||||
{
|
|
||||||
heartbeatCancellation.Cancel();
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await heartbeatTask.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
catch (OperationCanceledException)
|
|
||||||
{
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task<bool> DispatchGatewayEnvelopeAsync(
|
|
||||||
WorkerEnvelope envelope,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
switch (envelope.BodyCase)
|
|
||||||
{
|
|
||||||
case WorkerEnvelope.BodyOneofCase.WorkerCommand:
|
|
||||||
_ = ProcessCommandAsync(envelope, cancellationToken);
|
|
||||||
return true;
|
|
||||||
case WorkerEnvelope.BodyOneofCase.WorkerShutdown:
|
|
||||||
await ShutdownAsync(envelope.WorkerShutdown, cancellationToken).ConfigureAwait(false);
|
|
||||||
return false;
|
|
||||||
case WorkerEnvelope.BodyOneofCase.WorkerCancel:
|
|
||||||
return true;
|
|
||||||
default:
|
|
||||||
throw new WorkerFrameProtocolException(
|
|
||||||
WorkerFrameProtocolErrorCode.UnexpectedEnvelopeBody,
|
|
||||||
$"Worker received unexpected gateway envelope body {envelope.BodyCase}.");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task ProcessCommandAsync(
|
|
||||||
WorkerEnvelope envelope,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
IWorkerRuntimeSession runtimeSession = _runtimeSession
|
|
||||||
?? throw new InvalidOperationException("Worker runtime session has not been initialized.");
|
|
||||||
WorkerCommand workerCommand = envelope.WorkerCommand;
|
|
||||||
MxCommand command = workerCommand.Command;
|
|
||||||
StaCommand staCommand = new(
|
|
||||||
_options.SessionId,
|
|
||||||
envelope.CorrelationId,
|
|
||||||
command,
|
|
||||||
workerCommand.EnqueueTimestamp,
|
|
||||||
cancellationToken);
|
|
||||||
|
|
||||||
try
|
|
||||||
{
|
|
||||||
MxCommandReply reply = await runtimeSession.DispatchAsync(staCommand).ConfigureAwait(false);
|
|
||||||
await _writer
|
|
||||||
.WriteAsync(
|
|
||||||
CreateEnvelope(new WorkerCommandReply
|
|
||||||
{
|
|
||||||
Reply = reply,
|
|
||||||
CompletedTimestamp = Timestamp.FromDateTime(DateTime.UtcNow),
|
|
||||||
}),
|
|
||||||
cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
catch (Exception exception) when (exception is not OperationCanceledException)
|
|
||||||
{
|
|
||||||
_state = WorkerState.Faulted;
|
|
||||||
await TryWriteFaultAsync(
|
|
||||||
CreateFault(
|
|
||||||
WorkerFaultCategory.MxaccessCommandFailed,
|
|
||||||
staCommand.MethodName,
|
|
||||||
exception),
|
|
||||||
cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task ShutdownAsync(
|
|
||||||
WorkerShutdown shutdown,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
_state = WorkerState.ShuttingDown;
|
|
||||||
_runtimeSession?.RequestShutdown();
|
|
||||||
|
|
||||||
await _writer
|
|
||||||
.WriteAsync(
|
|
||||||
CreateEnvelope(
|
|
||||||
new WorkerShutdownAck
|
|
||||||
{
|
|
||||||
Status = new ProtocolStatus
|
|
||||||
{
|
|
||||||
Code = ProtocolStatusCode.Ok,
|
|
||||||
Message = string.IsNullOrWhiteSpace(shutdown.Reason)
|
|
||||||
? "Worker shutdown accepted."
|
|
||||||
: $"Worker shutdown accepted: {shutdown.Reason}",
|
|
||||||
},
|
|
||||||
}),
|
|
||||||
cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task RunHeartbeatLoopAsync(CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
while (!cancellationToken.IsCancellationRequested)
|
|
||||||
{
|
|
||||||
await Task.Delay(_sessionOptions.HeartbeatInterval, cancellationToken).ConfigureAwait(false);
|
|
||||||
IWorkerRuntimeSession? runtimeSession = _runtimeSession;
|
|
||||||
if (runtimeSession is null)
|
|
||||||
{
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
|
|
||||||
WorkerRuntimeHeartbeatSnapshot snapshot = runtimeSession.CaptureHeartbeat();
|
|
||||||
await _writer
|
|
||||||
.WriteAsync(CreateEnvelope(CreateHeartbeat(snapshot)), cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
|
|
||||||
await ReportWatchdogFaultIfNeededAsync(snapshot, cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task ReportWatchdogFaultIfNeededAsync(
|
|
||||||
WorkerRuntimeHeartbeatSnapshot snapshot,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
TimeSpan staleFor = DateTimeOffset.UtcNow - snapshot.LastStaActivityUtc;
|
|
||||||
if (staleFor <= _sessionOptions.HeartbeatGrace)
|
|
||||||
{
|
|
||||||
_watchdogFaultSent = false;
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (_watchdogFaultSent)
|
|
||||||
{
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
_watchdogFaultSent = true;
|
|
||||||
await TryWriteFaultAsync(
|
|
||||||
CreateFault(
|
|
||||||
WorkerFaultCategory.StaHung,
|
|
||||||
snapshot.CurrentCommandCorrelationId,
|
|
||||||
$"STA activity is stale by {staleFor}."),
|
|
||||||
cancellationToken).ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
|
|
||||||
private async Task TryWriteFaultAsync(
|
private async Task TryWriteFaultAsync(
|
||||||
WorkerFrameProtocolException exception,
|
WorkerFrameProtocolException exception,
|
||||||
CancellationToken cancellationToken)
|
CancellationToken cancellationToken)
|
||||||
@@ -392,25 +178,6 @@ public sealed class WorkerPipeSession
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private async Task TryWriteFaultAsync(
|
|
||||||
WorkerFault fault,
|
|
||||||
CancellationToken cancellationToken)
|
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
|
||||||
await _writer
|
|
||||||
.WriteAsync(CreateEnvelope(fault), cancellationToken)
|
|
||||||
.ConfigureAwait(false);
|
|
||||||
}
|
|
||||||
catch (Exception faultWriteException) when (
|
|
||||||
faultWriteException is IOException
|
|
||||||
|| faultWriteException is ObjectDisposedException
|
|
||||||
|| faultWriteException is WorkerFrameProtocolException)
|
|
||||||
{
|
|
||||||
// The runtime fault remains observable through worker exit or pipe closure.
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateEnvelope(WorkerHello hello)
|
private WorkerEnvelope CreateEnvelope(WorkerHello hello)
|
||||||
{
|
{
|
||||||
return CreateBaseEnvelope(hello);
|
return CreateBaseEnvelope(hello);
|
||||||
@@ -426,21 +193,6 @@ public sealed class WorkerPipeSession
|
|||||||
return CreateBaseEnvelope(fault);
|
return CreateBaseEnvelope(fault);
|
||||||
}
|
}
|
||||||
|
|
||||||
private WorkerEnvelope CreateEnvelope(WorkerCommandReply reply)
|
|
||||||
{
|
|
||||||
return CreateBaseEnvelope(reply);
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateEnvelope(WorkerShutdownAck shutdownAck)
|
|
||||||
{
|
|
||||||
return CreateBaseEnvelope(shutdownAck);
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateEnvelope(WorkerHeartbeat heartbeat)
|
|
||||||
{
|
|
||||||
return CreateBaseEnvelope(heartbeat);
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateBaseEnvelope(WorkerHello body)
|
private WorkerEnvelope CreateBaseEnvelope(WorkerHello body)
|
||||||
{
|
{
|
||||||
WorkerEnvelope envelope = CreateBaseEnvelope();
|
WorkerEnvelope envelope = CreateBaseEnvelope();
|
||||||
@@ -462,28 +214,6 @@ public sealed class WorkerPipeSession
|
|||||||
return envelope;
|
return envelope;
|
||||||
}
|
}
|
||||||
|
|
||||||
private WorkerEnvelope CreateBaseEnvelope(WorkerCommandReply body)
|
|
||||||
{
|
|
||||||
WorkerEnvelope envelope = CreateBaseEnvelope();
|
|
||||||
envelope.CorrelationId = body.Reply?.CorrelationId ?? string.Empty;
|
|
||||||
envelope.WorkerCommandReply = body;
|
|
||||||
return envelope;
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateBaseEnvelope(WorkerShutdownAck body)
|
|
||||||
{
|
|
||||||
WorkerEnvelope envelope = CreateBaseEnvelope();
|
|
||||||
envelope.WorkerShutdownAck = body;
|
|
||||||
return envelope;
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateBaseEnvelope(WorkerHeartbeat body)
|
|
||||||
{
|
|
||||||
WorkerEnvelope envelope = CreateBaseEnvelope();
|
|
||||||
envelope.WorkerHeartbeat = body;
|
|
||||||
return envelope;
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerEnvelope CreateBaseEnvelope()
|
private WorkerEnvelope CreateBaseEnvelope()
|
||||||
{
|
{
|
||||||
return new WorkerEnvelope
|
return new WorkerEnvelope
|
||||||
@@ -501,39 +231,21 @@ public sealed class WorkerPipeSession
|
|||||||
|
|
||||||
private async Task<WorkerReady> InitializeMxAccessAsync(CancellationToken cancellationToken)
|
private async Task<WorkerReady> InitializeMxAccessAsync(CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
_runtimeSession = new MxAccessStaSession();
|
_mxAccessStaSession = new MxAccessStaSession();
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
return await _runtimeSession
|
return await _mxAccessStaSession
|
||||||
.StartAsync(_processIdProvider(), cancellationToken)
|
.StartAsync(_processIdProvider(), cancellationToken)
|
||||||
.ConfigureAwait(false);
|
.ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
catch
|
catch
|
||||||
{
|
{
|
||||||
_runtimeSession.Dispose();
|
_mxAccessStaSession.Dispose();
|
||||||
_runtimeSession = null;
|
_mxAccessStaSession = null;
|
||||||
throw;
|
throw;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private WorkerHeartbeat CreateHeartbeat(WorkerRuntimeHeartbeatSnapshot snapshot)
|
|
||||||
{
|
|
||||||
WorkerState state = string.IsNullOrWhiteSpace(snapshot.CurrentCommandCorrelationId)
|
|
||||||
? _state
|
|
||||||
: WorkerState.ExecutingCommand;
|
|
||||||
|
|
||||||
return new WorkerHeartbeat
|
|
||||||
{
|
|
||||||
WorkerProcessId = _processIdProvider(),
|
|
||||||
State = state,
|
|
||||||
LastStaActivityTimestamp = Timestamp.FromDateTimeOffset(snapshot.LastStaActivityUtc),
|
|
||||||
PendingCommandCount = snapshot.PendingCommandCount,
|
|
||||||
OutboundEventQueueDepth = snapshot.OutboundEventQueueDepth,
|
|
||||||
LastEventSequence = snapshot.LastEventSequence,
|
|
||||||
CurrentCommandCorrelationId = snapshot.CurrentCommandCorrelationId,
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
private WorkerReady CreateWorkerReady()
|
private WorkerReady CreateWorkerReady()
|
||||||
{
|
{
|
||||||
return new WorkerReady
|
return new WorkerReady
|
||||||
@@ -583,42 +295,6 @@ public sealed class WorkerPipeSession
|
|||||||
return fault;
|
return fault;
|
||||||
}
|
}
|
||||||
|
|
||||||
private static WorkerFault CreateFault(
|
|
||||||
WorkerFaultCategory category,
|
|
||||||
string commandMethod,
|
|
||||||
Exception exception)
|
|
||||||
{
|
|
||||||
WorkerFault fault = CreateFault(
|
|
||||||
category,
|
|
||||||
commandMethod,
|
|
||||||
exception.Message);
|
|
||||||
fault.ExceptionType = exception.GetType().FullName ?? string.Empty;
|
|
||||||
fault.ProtocolStatus = new ProtocolStatus
|
|
||||||
{
|
|
||||||
Code = ProtocolStatusCode.WorkerUnavailable,
|
|
||||||
Message = exception.Message,
|
|
||||||
};
|
|
||||||
return fault;
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerFault CreateFault(
|
|
||||||
WorkerFaultCategory category,
|
|
||||||
string commandMethod,
|
|
||||||
string diagnosticMessage)
|
|
||||||
{
|
|
||||||
return new WorkerFault
|
|
||||||
{
|
|
||||||
Category = category,
|
|
||||||
CommandMethod = commandMethod ?? string.Empty,
|
|
||||||
DiagnosticMessage = diagnosticMessage,
|
|
||||||
ProtocolStatus = new ProtocolStatus
|
|
||||||
{
|
|
||||||
Code = ProtocolStatusCode.WorkerUnavailable,
|
|
||||||
Message = diagnosticMessage,
|
|
||||||
},
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
private static WorkerFaultCategory MapFaultCategory(WorkerFrameProtocolErrorCode errorCode)
|
private static WorkerFaultCategory MapFaultCategory(WorkerFrameProtocolErrorCode errorCode)
|
||||||
{
|
{
|
||||||
return errorCode switch
|
return errorCode switch
|
||||||
|
|||||||
@@ -1,36 +0,0 @@
|
|||||||
using System;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.Ipc;
|
|
||||||
|
|
||||||
public sealed class WorkerPipeSessionOptions
|
|
||||||
{
|
|
||||||
public static readonly TimeSpan DefaultHeartbeatInterval = TimeSpan.FromSeconds(5);
|
|
||||||
public static readonly TimeSpan DefaultHeartbeatGrace = TimeSpan.FromSeconds(15);
|
|
||||||
|
|
||||||
public WorkerPipeSessionOptions()
|
|
||||||
{
|
|
||||||
HeartbeatInterval = DefaultHeartbeatInterval;
|
|
||||||
HeartbeatGrace = DefaultHeartbeatGrace;
|
|
||||||
}
|
|
||||||
|
|
||||||
public TimeSpan HeartbeatInterval { get; set; }
|
|
||||||
|
|
||||||
public TimeSpan HeartbeatGrace { get; set; }
|
|
||||||
|
|
||||||
public void Validate()
|
|
||||||
{
|
|
||||||
if (HeartbeatInterval <= TimeSpan.Zero)
|
|
||||||
{
|
|
||||||
throw new ArgumentOutOfRangeException(
|
|
||||||
nameof(HeartbeatInterval),
|
|
||||||
"Worker heartbeat interval must be greater than zero.");
|
|
||||||
}
|
|
||||||
|
|
||||||
if (HeartbeatGrace <= TimeSpan.Zero)
|
|
||||||
{
|
|
||||||
throw new ArgumentOutOfRangeException(
|
|
||||||
nameof(HeartbeatGrace),
|
|
||||||
"Worker heartbeat grace must be greater than zero.");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,20 +0,0 @@
|
|||||||
using System;
|
|
||||||
using System.Threading;
|
|
||||||
using System.Threading.Tasks;
|
|
||||||
using MxGateway.Contracts.Proto;
|
|
||||||
using MxGateway.Worker.Sta;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.MxAccess;
|
|
||||||
|
|
||||||
public interface IWorkerRuntimeSession : IDisposable
|
|
||||||
{
|
|
||||||
Task<WorkerReady> StartAsync(
|
|
||||||
int workerProcessId,
|
|
||||||
CancellationToken cancellationToken = default);
|
|
||||||
|
|
||||||
Task<MxCommandReply> DispatchAsync(StaCommand command);
|
|
||||||
|
|
||||||
WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat();
|
|
||||||
|
|
||||||
void RequestShutdown();
|
|
||||||
}
|
|
||||||
@@ -7,7 +7,7 @@ using MxGateway.Worker.Sta;
|
|||||||
|
|
||||||
namespace MxGateway.Worker.MxAccess;
|
namespace MxGateway.Worker.MxAccess;
|
||||||
|
|
||||||
public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
public sealed class MxAccessStaSession : IDisposable
|
||||||
{
|
{
|
||||||
private readonly IMxAccessComObjectFactory factory;
|
private readonly IMxAccessComObjectFactory factory;
|
||||||
private readonly IMxAccessEventSink eventSink;
|
private readonly IMxAccessEventSink eventSink;
|
||||||
@@ -68,30 +68,6 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
|||||||
return commandDispatcher.DispatchAsync(command);
|
return commandDispatcher.DispatchAsync(command);
|
||||||
}
|
}
|
||||||
|
|
||||||
public WorkerRuntimeHeartbeatSnapshot CaptureHeartbeat()
|
|
||||||
{
|
|
||||||
uint pendingCommandCount = 0;
|
|
||||||
string currentCommandCorrelationId = string.Empty;
|
|
||||||
|
|
||||||
if (commandDispatcher is not null)
|
|
||||||
{
|
|
||||||
pendingCommandCount = (uint)commandDispatcher.PendingCommandCount;
|
|
||||||
currentCommandCorrelationId = commandDispatcher.CurrentCommandCorrelationId;
|
|
||||||
}
|
|
||||||
|
|
||||||
return new WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
staRuntime.LastActivityUtc,
|
|
||||||
pendingCommandCount,
|
|
||||||
outboundEventQueueDepth: 0,
|
|
||||||
lastEventSequence: 0,
|
|
||||||
currentCommandCorrelationId);
|
|
||||||
}
|
|
||||||
|
|
||||||
public void RequestShutdown()
|
|
||||||
{
|
|
||||||
commandDispatcher?.RequestShutdown();
|
|
||||||
}
|
|
||||||
|
|
||||||
public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync(
|
public Task<IReadOnlyList<RegisteredServerHandle>> GetRegisteredServerHandlesAsync(
|
||||||
CancellationToken cancellationToken = default)
|
CancellationToken cancellationToken = default)
|
||||||
{
|
{
|
||||||
@@ -125,7 +101,7 @@ public sealed class MxAccessStaSession : IWorkerRuntimeSession
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
RequestShutdown();
|
commandDispatcher?.RequestShutdown();
|
||||||
|
|
||||||
if (session is not null)
|
if (session is not null)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -1,30 +0,0 @@
|
|||||||
using System;
|
|
||||||
|
|
||||||
namespace MxGateway.Worker.MxAccess;
|
|
||||||
|
|
||||||
public sealed class WorkerRuntimeHeartbeatSnapshot
|
|
||||||
{
|
|
||||||
public WorkerRuntimeHeartbeatSnapshot(
|
|
||||||
DateTimeOffset lastStaActivityUtc,
|
|
||||||
uint pendingCommandCount,
|
|
||||||
uint outboundEventQueueDepth,
|
|
||||||
ulong lastEventSequence,
|
|
||||||
string currentCommandCorrelationId)
|
|
||||||
{
|
|
||||||
LastStaActivityUtc = lastStaActivityUtc;
|
|
||||||
PendingCommandCount = pendingCommandCount;
|
|
||||||
OutboundEventQueueDepth = outboundEventQueueDepth;
|
|
||||||
LastEventSequence = lastEventSequence;
|
|
||||||
CurrentCommandCorrelationId = currentCommandCorrelationId ?? string.Empty;
|
|
||||||
}
|
|
||||||
|
|
||||||
public DateTimeOffset LastStaActivityUtc { get; }
|
|
||||||
|
|
||||||
public uint PendingCommandCount { get; }
|
|
||||||
|
|
||||||
public uint OutboundEventQueueDepth { get; }
|
|
||||||
|
|
||||||
public ulong LastEventSequence { get; }
|
|
||||||
|
|
||||||
public string CurrentCommandCorrelationId { get; }
|
|
||||||
}
|
|
||||||
@@ -84,7 +84,7 @@ public static class WorkerApplication
|
|||||||
|
|
||||||
pipeClient.RunAsync(options).GetAwaiter().GetResult();
|
pipeClient.RunAsync(options).GetAwaiter().GetResult();
|
||||||
|
|
||||||
logger.Information("WorkerPipeSessionCompleted", new Dictionary<string, object?>
|
logger.Information("WorkerPipeHandshakeSucceeded", new Dictionary<string, object?>
|
||||||
{
|
{
|
||||||
["session_id"] = options.SessionId,
|
["session_id"] = options.SessionId,
|
||||||
["pipe_name"] = options.PipeName,
|
["pipe_name"] = options.PipeName,
|
||||||
|
|||||||
Reference in New Issue
Block a user