Add node-scoped provisioning job channel
This commit is contained in:
parent
8620ade42e
commit
2757881a41
6 changed files with 142 additions and 2 deletions
|
|
@ -113,7 +113,7 @@
|
||||||
- [x] создание, чтение, подтверждение и отмена через Operator API;
|
- [x] создание, чтение, подтверждение и отмена через Operator API;
|
||||||
- [ ] выполнение, retry, verification и rollback в полной state machine;
|
- [ ] выполнение, retry, verification и rollback в полной state machine;
|
||||||
- [x] обязательные idempotency key, audit reason и job TTL;
|
- [x] обязательные idempotency key, audit reason и job TTL;
|
||||||
- [ ] Agent job channel только для собственного `node_id`;
|
- [x] атомарный Agent job channel только для собственного `node_id`;
|
||||||
- [x] начальные строгие JSON schemas v1 для `preflight` и `system.base-install`;
|
- [x] начальные строгие JSON schemas v1 для `preflight` и `system.base-install`;
|
||||||
- [ ] versioned JSON schemas для остальных action type;
|
- [ ] versioned JSON schemas для остальных action type;
|
||||||
- [ ] restricted root helper через Unix socket;
|
- [ ] restricted root helper через Unix socket;
|
||||||
|
|
|
||||||
|
|
@ -377,6 +377,20 @@ agents.MapPost("/heartbeat", async (
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
agents.MapGet("/provisioning/jobs/next", async (
|
||||||
|
HttpContext context,
|
||||||
|
ControlStore controlStore,
|
||||||
|
CancellationToken cancellationToken) =>
|
||||||
|
{
|
||||||
|
var nodeId = context.User.FindFirstValue(ClaimTypes.NameIdentifier);
|
||||||
|
if (string.IsNullOrWhiteSpace(nodeId) || !NodeIdValidator.IsValid(nodeId))
|
||||||
|
{
|
||||||
|
return Results.Forbid();
|
||||||
|
}
|
||||||
|
|
||||||
|
var job = await controlStore.ClaimNextProvisioningJobAsync(nodeId, cancellationToken);
|
||||||
|
return job is null ? Results.NoContent() : Results.Ok(job);
|
||||||
|
});
|
||||||
|
|
||||||
var control = app.MapGroup("/api/v1/control").RequireAuthorization("Operator");
|
var control = app.MapGroup("/api/v1/control").RequireAuthorization("Operator");
|
||||||
control.MapGet("/agents", async (ControlStore controlStore, CancellationToken cancellationToken) =>
|
control.MapGet("/agents", async (ControlStore controlStore, CancellationToken cancellationToken) =>
|
||||||
|
|
|
||||||
|
|
@ -96,6 +96,56 @@ public sealed partial class ControlStore
|
||||||
return await reader.ReadAsync(cancellationToken) ? ReadProvisioningJob(reader) : null;
|
return await reader.ReadAsync(cancellationToken) ? ReadProvisioningJob(reader) : null;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task<ProvisioningJob?> ClaimNextProvisioningJobAsync(
|
||||||
|
string nodeId,
|
||||||
|
CancellationToken cancellationToken = default)
|
||||||
|
{
|
||||||
|
await using var connection = await OpenAsync(cancellationToken);
|
||||||
|
await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync(cancellationToken);
|
||||||
|
var now = DateTimeOffset.UtcNow;
|
||||||
|
var command = connection.CreateCommand();
|
||||||
|
command.Transaction = transaction;
|
||||||
|
command.CommandText = """
|
||||||
|
UPDATE provisioning_jobs SET
|
||||||
|
state = $preflight,
|
||||||
|
updated_at = $now,
|
||||||
|
version = version + 1
|
||||||
|
WHERE id = (
|
||||||
|
SELECT id FROM provisioning_jobs
|
||||||
|
WHERE node_id = $node
|
||||||
|
AND state = $queued
|
||||||
|
AND expires_at > $now
|
||||||
|
ORDER BY created_at, id
|
||||||
|
LIMIT 1)
|
||||||
|
AND state = $queued
|
||||||
|
RETURNING *;
|
||||||
|
""";
|
||||||
|
command.Parameters.AddWithValue("$node", nodeId);
|
||||||
|
command.Parameters.AddWithValue("$queued", ProvisioningJobStates.Queued);
|
||||||
|
command.Parameters.AddWithValue("$preflight", ProvisioningJobStates.Preflight);
|
||||||
|
command.Parameters.AddWithValue("$now", now.ToString("O"));
|
||||||
|
ProvisioningJob? job;
|
||||||
|
await using (var reader = await command.ExecuteReaderAsync(cancellationToken))
|
||||||
|
{
|
||||||
|
job = await reader.ReadAsync(cancellationToken) ? ReadProvisioningJob(reader) : null;
|
||||||
|
}
|
||||||
|
if (job is null)
|
||||||
|
{
|
||||||
|
await transaction.CommitAsync(cancellationToken);
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
await WriteProvisioningEventAsync(
|
||||||
|
connection, transaction, job.Id, "job.claimed", job.State,
|
||||||
|
"Provisioning job claimed by its assigned Node.", now, cancellationToken);
|
||||||
|
await WriteAuditAsync(
|
||||||
|
connection, transaction, nodeId, "provisioning.job.claimed", job.Id,
|
||||||
|
JsonSerializer.Serialize(new { job.NodeId, job.ActionType, job.SchemaVersion }),
|
||||||
|
cancellationToken);
|
||||||
|
await transaction.CommitAsync(cancellationToken);
|
||||||
|
return job;
|
||||||
|
}
|
||||||
|
|
||||||
public Task<ProvisioningJob?> ConfirmProvisioningJobAsync(
|
public Task<ProvisioningJob?> ConfirmProvisioningJobAsync(
|
||||||
string id,
|
string id,
|
||||||
ProvisioningJobCommandRequest request,
|
ProvisioningJobCommandRequest request,
|
||||||
|
|
@ -148,7 +198,9 @@ public sealed partial class ControlStore
|
||||||
}
|
}
|
||||||
var allowed = requiredState is not null
|
var allowed = requiredState is not null
|
||||||
? current.State == requiredState
|
? current.State == requiredState
|
||||||
: current.State is ProvisioningJobStates.Queued or ProvisioningJobStates.AwaitingConfirmation;
|
: current.State is ProvisioningJobStates.Queued
|
||||||
|
or ProvisioningJobStates.Preflight
|
||||||
|
or ProvisioningJobStates.AwaitingConfirmation;
|
||||||
if (!allowed || current.ExpiresAt <= DateTimeOffset.UtcNow)
|
if (!allowed || current.ExpiresAt <= DateTimeOffset.UtcNow)
|
||||||
{
|
{
|
||||||
throw new ProvisioningTransitionException(current.State, targetState);
|
throw new ProvisioningTransitionException(current.State, targetState);
|
||||||
|
|
|
||||||
|
|
@ -176,6 +176,7 @@ public sealed record ProvisioningJob(
|
||||||
public static class ProvisioningJobStates
|
public static class ProvisioningJobStates
|
||||||
{
|
{
|
||||||
public const string Queued = "Queued";
|
public const string Queued = "Queued";
|
||||||
|
public const string Preflight = "Preflight";
|
||||||
public const string AwaitingConfirmation = "AwaitingConfirmation";
|
public const string AwaitingConfirmation = "AwaitingConfirmation";
|
||||||
public const string Cancelled = "Cancelled";
|
public const string Cancelled = "Cancelled";
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -118,6 +118,50 @@ public sealed class ControlApiTests : IAsyncDisposable
|
||||||
$"/api/v1/control/provisioning/jobs/{job.Id}", cancellationToken)).StatusCode);
|
$"/api/v1/control/provisioning/jobs/{job.Id}", cancellationToken)).StatusCode);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task AgentReceivesOnlyItsOwnProvisioningJob()
|
||||||
|
{
|
||||||
|
var nodeId = $"node-{Guid.NewGuid():N}"[..13];
|
||||||
|
var store = _factory.Services.GetRequiredService<ServerMonitorManager.Control.ControlStore>();
|
||||||
|
var cancellationToken = TestContext.Current.CancellationToken;
|
||||||
|
var token = await store.CreateEnrollmentTokenAsync(nodeId, TimeSpan.FromMinutes(10), cancellationToken);
|
||||||
|
await store.EnrollAsync(
|
||||||
|
new ServerMonitorManager.Core.EnrollmentRequest(
|
||||||
|
nodeId, token, "csr", Guid.NewGuid().ToString()),
|
||||||
|
() => new ServerMonitorManager.Control.IssuedCertificate(
|
||||||
|
"certificate", "ca", Guid.NewGuid().ToString("N"), DateTimeOffset.UtcNow.AddDays(1)),
|
||||||
|
cancellationToken);
|
||||||
|
using var parameters = System.Text.Json.JsonDocument.Parse("{}");
|
||||||
|
var created = await store.CreateProvisioningJobAsync(
|
||||||
|
nodeId,
|
||||||
|
new ServerMonitorManager.Core.ProvisioningJobCreateRequest(
|
||||||
|
"preflight", 1, parameters.RootElement.Clone(), 60,
|
||||||
|
"API Agent isolation test", Guid.NewGuid().ToString()),
|
||||||
|
"windows-pc",
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
using var other = _factory.CreateClient();
|
||||||
|
other.DefaultRequestHeaders.Add("X-Test-Identity", "other-node");
|
||||||
|
other.DefaultRequestHeaders.Add("X-Test-Role", "Agent");
|
||||||
|
Assert.Equal(HttpStatusCode.NoContent, (await other.GetAsync(
|
||||||
|
"/api/v1/agents/provisioning/jobs/next", cancellationToken)).StatusCode);
|
||||||
|
|
||||||
|
using var assigned = _factory.CreateClient();
|
||||||
|
assigned.DefaultRequestHeaders.Add("X-Test-Identity", nodeId);
|
||||||
|
assigned.DefaultRequestHeaders.Add("X-Test-Role", "Agent");
|
||||||
|
var response = await assigned.GetAsync(
|
||||||
|
"/api/v1/agents/provisioning/jobs/next", cancellationToken);
|
||||||
|
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
|
||||||
|
var claimed = await response.Content.ReadFromJsonAsync<ServerMonitorManager.Core.ProvisioningJob>(
|
||||||
|
cancellationToken);
|
||||||
|
Assert.NotNull(claimed);
|
||||||
|
Assert.Equal(created.Id, claimed.Id);
|
||||||
|
Assert.Equal(nodeId, claimed.NodeId);
|
||||||
|
Assert.Equal(ServerMonitorManager.Core.ProvisioningJobStates.Preflight, claimed.State);
|
||||||
|
Assert.Equal(HttpStatusCode.NoContent, (await assigned.GetAsync(
|
||||||
|
"/api/v1/agents/provisioning/jobs/next", cancellationToken)).StatusCode);
|
||||||
|
}
|
||||||
|
|
||||||
public async ValueTask DisposeAsync() => await _factory.DisposeAsync();
|
public async ValueTask DisposeAsync() => await _factory.DisposeAsync();
|
||||||
|
|
||||||
private sealed class ControlApiFactory : WebApplicationFactory<controlapp::Program>
|
private sealed class ControlApiFactory : WebApplicationFactory<controlapp::Program>
|
||||||
|
|
|
||||||
|
|
@ -83,6 +83,35 @@ public sealed class ControlStoreTests : IAsyncDisposable
|
||||||
Assert.Equal(ProvisioningJobStates.AwaitingConfirmation, next.State);
|
Assert.Equal(ProvisioningJobStates.AwaitingConfirmation, next.State);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public async Task ProvisioningJobCanBeClaimedOnlyOnceByAssignedNode()
|
||||||
|
{
|
||||||
|
var cancellationToken = TestContext.Current.CancellationToken;
|
||||||
|
var store = CreateStore();
|
||||||
|
await store.InitializeAsync(cancellationToken);
|
||||||
|
await EnrollAgentAsync(store, "home", "C3D4", cancellationToken);
|
||||||
|
await EnrollAgentAsync(store, "other", "E5F6", cancellationToken);
|
||||||
|
using var parameters = JsonDocument.Parse("{}");
|
||||||
|
var created = await store.CreateProvisioningJobAsync(
|
||||||
|
"home",
|
||||||
|
new ProvisioningJobCreateRequest(
|
||||||
|
"preflight", 1, parameters.RootElement.Clone(), 60,
|
||||||
|
"Inspect server", Guid.NewGuid().ToString()),
|
||||||
|
"operator",
|
||||||
|
cancellationToken);
|
||||||
|
|
||||||
|
Assert.Null(await store.ClaimNextProvisioningJobAsync("other", cancellationToken));
|
||||||
|
var claims = await Task.WhenAll(
|
||||||
|
store.ClaimNextProvisioningJobAsync("home", cancellationToken),
|
||||||
|
store.ClaimNextProvisioningJobAsync("home", cancellationToken));
|
||||||
|
|
||||||
|
var claimed = Assert.Single(claims, job => job is not null)!;
|
||||||
|
Assert.Equal(created.Id, claimed.Id);
|
||||||
|
Assert.Equal("home", claimed.NodeId);
|
||||||
|
Assert.Equal(ProvisioningJobStates.Preflight, claimed.State);
|
||||||
|
Assert.Null(Assert.Single(claims, job => job is null));
|
||||||
|
}
|
||||||
|
|
||||||
[Fact]
|
[Fact]
|
||||||
public void DiagnosticsExportOmitsRawIdentitiesAndNormalizesStates()
|
public void DiagnosticsExportOmitsRawIdentitiesAndNormalizesStates()
|
||||||
{
|
{
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue