feat(management): parquet audit export and audit archive list/verify (ScadaBridge#42)
GET /api/audit/export?format=parquet returned 501. It now streams the same 24 columns as csv/jsonl (AuditExportRow, CsvHeader order) as one Parquet file, one 10,000-row row group at a time. Each row group is encoded into a small non-seekable staging buffer and copied to the response asynchronously, because Kestrel rejects the synchronous writes the Parquet writer issues. New management commands, with CLI parity: - ListAuditArchives -> `audit archive list` (AuditArchiveListResult: enabled flag + every manifest). - VerifyAuditArchive(Name) -> `audit archive verify --name` (files, chain, purge anchor); exit 0 verified, 3 not verified, 1 error, 2 unauthorized. Both carry verify-chain's gate: OperationalAudit roles (Administrator, Viewer), site-scoped non-Administrator refused because an archive spans every site. VerifyAuditArchive uses the long-running Ask timeout. Error codes ARCHIVE_NOT_CONFIGURED, INVALID_ARGUMENT, NOT_FOUND. RequiredRoleMatrixTests freezes both gates; ManagementSurfaceParityTests now probes `audit archive list|verify` alongside `audit verify-chain`. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,219 @@
|
||||
using Parquet;
|
||||
using Parquet.Schema;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.ManagementService;
|
||||
|
||||
/// <summary>
|
||||
/// <c>GET /api/audit/export?format=parquet</c> (ScadaBridge#42): the same 24 columns as the
|
||||
/// csv / jsonl export (<see cref="AuditExportRow"/>, in <see cref="AuditEndpoints.CsvHeader"/>
|
||||
/// order) as one Parquet file, one row group per <see cref="RowGroupRows"/> rows.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Types: ids are <c>D</c>-format strings, timestamps are UTC Parquet timestamps
|
||||
/// (microseconds), enums are their names, <c>HttpStatus</c>/<c>DurationMs</c> are int32,
|
||||
/// <c>PayloadTruncated</c> is boolean, everything else is a UTF-8 string.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Streaming.</b> Kestrel forbids synchronous writes to the response body and the Parquet
|
||||
/// writer issues some, so each row group is encoded into a small in-memory buffer and then
|
||||
/// copied to the response asynchronously. Memory is bounded by one row group, and the client
|
||||
/// starts receiving bytes after the first one. A Parquet file's footer is written last, so a
|
||||
/// response cut off mid-stream is not a readable file — the CLI treats it as a failed export.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// This is the ad-hoc export. The retention archive (alog.md §15.2) is a different layout
|
||||
/// that carries the hash-chain columns; see <c>AuditArchiveParquet</c>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public static class AuditExportParquet
|
||||
{
|
||||
/// <summary>Rows per Parquet row group.</summary>
|
||||
public const int RowGroupRows = 10_000;
|
||||
|
||||
private static readonly DataField<string> EventId = new(nameof(AuditExportRow.EventId), false);
|
||||
private static readonly DateTimeDataField OccurredAtUtc = Timestamp(nameof(AuditExportRow.OccurredAtUtc), false);
|
||||
private static readonly DateTimeDataField IngestedAtUtc = Timestamp(nameof(AuditExportRow.IngestedAtUtc), true);
|
||||
private static readonly DataField<string> Channel = new(nameof(AuditExportRow.Channel), false);
|
||||
private static readonly DataField<string> Kind = new(nameof(AuditExportRow.Kind), false);
|
||||
private static readonly DataField<string> CorrelationId = Text(nameof(AuditExportRow.CorrelationId));
|
||||
private static readonly DataField<string> ExecutionId = Text(nameof(AuditExportRow.ExecutionId));
|
||||
private static readonly DataField<string> ParentExecutionId = Text(nameof(AuditExportRow.ParentExecutionId));
|
||||
private static readonly DataField<string> SourceSiteId = Text(nameof(AuditExportRow.SourceSiteId));
|
||||
private static readonly DataField<string> SourceNode = Text(nameof(AuditExportRow.SourceNode));
|
||||
private static readonly DataField<string> SourceInstanceId = Text(nameof(AuditExportRow.SourceInstanceId));
|
||||
private static readonly DataField<string> SourceScript = Text(nameof(AuditExportRow.SourceScript));
|
||||
private static readonly DataField<string> Actor = Text(nameof(AuditExportRow.Actor));
|
||||
private static readonly DataField<string> Target = Text(nameof(AuditExportRow.Target));
|
||||
private static readonly DataField<string> Status = new(nameof(AuditExportRow.Status), false);
|
||||
private static readonly DataField<int> HttpStatus = new(nameof(AuditExportRow.HttpStatus), true);
|
||||
private static readonly DataField<int> DurationMs = new(nameof(AuditExportRow.DurationMs), true);
|
||||
private static readonly DataField<string> ErrorMessage = Text(nameof(AuditExportRow.ErrorMessage));
|
||||
private static readonly DataField<string> ErrorDetail = Text(nameof(AuditExportRow.ErrorDetail));
|
||||
private static readonly DataField<string> RequestSummary = Text(nameof(AuditExportRow.RequestSummary));
|
||||
private static readonly DataField<string> ResponseSummary = Text(nameof(AuditExportRow.ResponseSummary));
|
||||
private static readonly DataField<bool> PayloadTruncated = new(nameof(AuditExportRow.PayloadTruncated), false);
|
||||
private static readonly DataField<string> Extra = Text(nameof(AuditExportRow.Extra));
|
||||
private static readonly DataField<string> ForwardState = Text(nameof(AuditExportRow.ForwardState));
|
||||
|
||||
/// <summary>The export schema; column names and order equal <see cref="AuditEndpoints.CsvHeader"/>.</summary>
|
||||
public static ParquetSchema Schema { get; } = new(
|
||||
EventId, OccurredAtUtc, IngestedAtUtc, Channel, Kind, CorrelationId, ExecutionId, ParentExecutionId,
|
||||
SourceSiteId, SourceNode, SourceInstanceId, SourceScript, Actor, Target, Status, HttpStatus, DurationMs,
|
||||
ErrorMessage, ErrorDetail, RequestSummary, ResponseSummary, PayloadTruncated, Extra, ForwardState);
|
||||
|
||||
/// <summary>Writes every row of <paramref name="pages"/> to <paramref name="output"/> as one Parquet file.</summary>
|
||||
/// <param name="pages">Export rows, page by page.</param>
|
||||
/// <param name="output">The destination (e.g. the HTTP response body); written asynchronously only.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>The number of rows written.</returns>
|
||||
public static async Task<long> WriteAsync(
|
||||
IAsyncEnumerable<IReadOnlyList<AuditExportRow>> pages, Stream output, CancellationToken ct = default)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(pages);
|
||||
ArgumentNullException.ThrowIfNull(output);
|
||||
|
||||
// Non-seekable on purpose: the writer then tracks file offsets itself, so emptying the
|
||||
// buffer after each drain cannot corrupt the column-chunk offsets in the footer.
|
||||
var staging = new StagingStream();
|
||||
long total = 0;
|
||||
var buffer = new List<AuditExportRow>(RowGroupRows);
|
||||
await using (var writer = await ParquetWriter.CreateAsync(Schema, staging, cancellationToken: ct).ConfigureAwait(false))
|
||||
{
|
||||
await foreach (var page in pages.WithCancellation(ct).ConfigureAwait(false))
|
||||
{
|
||||
foreach (var row in page)
|
||||
{
|
||||
buffer.Add(row);
|
||||
if (buffer.Count == RowGroupRows)
|
||||
{
|
||||
total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
|
||||
await DrainAsync(staging, output, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
total += await WriteGroupAsync(writer, buffer, ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// The footer is written when the writer is disposed.
|
||||
await DrainAsync(staging, output, ct).ConfigureAwait(false);
|
||||
return total;
|
||||
}
|
||||
|
||||
private static async Task<int> WriteGroupAsync(ParquetWriter writer, List<AuditExportRow> rows, CancellationToken ct)
|
||||
{
|
||||
if (rows.Count == 0)
|
||||
{
|
||||
return 0;
|
||||
}
|
||||
|
||||
using (var group = writer.CreateRowGroup())
|
||||
{
|
||||
await group.WriteAsync(EventId, Map(rows, r => r.EventId.ToString("D"))).ConfigureAwait(false);
|
||||
await group.WriteAsync<DateTime>(OccurredAtUtc, rows.Select(r => DateTime.SpecifyKind(r.OccurredAtUtc, DateTimeKind.Utc)).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync<DateTime>(IngestedAtUtc, rows.Select(r => r.IngestedAtUtc is { } i ? DateTime.SpecifyKind(i, DateTimeKind.Utc) : (DateTime?)null).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(Channel, Map(rows, r => r.Channel.ToString())).ConfigureAwait(false);
|
||||
await group.WriteAsync(Kind, Map(rows, r => r.Kind.ToString())).ConfigureAwait(false);
|
||||
await group.WriteAsync(CorrelationId, Map(rows, r => r.CorrelationId?.ToString("D"))!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ExecutionId, Map(rows, r => r.ExecutionId?.ToString("D"))!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ParentExecutionId, Map(rows, r => r.ParentExecutionId?.ToString("D"))!).ConfigureAwait(false);
|
||||
await group.WriteAsync(SourceSiteId, Map(rows, r => r.SourceSiteId)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(SourceNode, Map(rows, r => r.SourceNode)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(SourceInstanceId, Map(rows, r => r.SourceInstanceId)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(SourceScript, Map(rows, r => r.SourceScript)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(Actor, Map(rows, r => r.Actor)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(Target, Map(rows, r => r.Target)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(Status, Map(rows, r => r.Status.ToString())).ConfigureAwait(false);
|
||||
await group.WriteAsync<int>(HttpStatus, rows.Select(r => r.HttpStatus).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync<int>(DurationMs, rows.Select(r => r.DurationMs).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(ErrorMessage, Map(rows, r => r.ErrorMessage)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ErrorDetail, Map(rows, r => r.ErrorDetail)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(RequestSummary, Map(rows, r => r.RequestSummary)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ResponseSummary, Map(rows, r => r.ResponseSummary)!).ConfigureAwait(false);
|
||||
await group.WriteAsync<bool>(PayloadTruncated, rows.Select(r => r.PayloadTruncated).ToArray(), cancellationToken: ct).ConfigureAwait(false);
|
||||
await group.WriteAsync(Extra, Map(rows, r => r.Extra)!).ConfigureAwait(false);
|
||||
await group.WriteAsync(ForwardState, Map(rows, r => r.ForwardState?.ToString())!).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var count = rows.Count;
|
||||
rows.Clear();
|
||||
return count;
|
||||
}
|
||||
|
||||
private static async Task DrainAsync(StagingStream staging, Stream output, CancellationToken ct)
|
||||
{
|
||||
var pending = staging.Pending;
|
||||
if (pending.Length == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await output.WriteAsync(pending, ct).ConfigureAwait(false);
|
||||
await output.FlushAsync(ct).ConfigureAwait(false);
|
||||
staging.Clear();
|
||||
}
|
||||
|
||||
private static string?[] Map(List<AuditExportRow> rows, Func<AuditExportRow, string?> select)
|
||||
=> rows.Select(select).ToArray();
|
||||
|
||||
private static DataField<string> Text(string name) => new(name, true);
|
||||
|
||||
private static DateTimeDataField Timestamp(string name, bool nullable)
|
||||
=> new(name, DateTimeFormat.Timestamp, isAdjustedToUTC: true, unit: DateTimeTimeUnit.Micros, isNullable: nullable);
|
||||
|
||||
/// <summary>
|
||||
/// Forward-only, non-seekable buffer the Parquet writer writes into (synchronously or
|
||||
/// not); <see cref="DrainAsync"/> moves its bytes to the real output asynchronously.
|
||||
/// </summary>
|
||||
private sealed class StagingStream : Stream
|
||||
{
|
||||
private readonly MemoryStream _buffer = new();
|
||||
|
||||
public ReadOnlyMemory<byte> Pending => _buffer.GetBuffer().AsMemory(0, (int)_buffer.Length);
|
||||
|
||||
public override bool CanRead => false;
|
||||
|
||||
public override bool CanSeek => false;
|
||||
|
||||
public override bool CanWrite => true;
|
||||
|
||||
public override long Length => throw new NotSupportedException();
|
||||
|
||||
public override long Position
|
||||
{
|
||||
get => throw new NotSupportedException();
|
||||
set => throw new NotSupportedException();
|
||||
}
|
||||
|
||||
public void Clear() => _buffer.SetLength(0);
|
||||
|
||||
public override void Write(byte[] buffer, int offset, int count) => _buffer.Write(buffer, offset, count);
|
||||
|
||||
public override void Write(ReadOnlySpan<byte> buffer) => _buffer.Write(buffer);
|
||||
|
||||
public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
|
||||
{
|
||||
_buffer.Write(buffer.Span);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
|
||||
{
|
||||
_buffer.Write(buffer, offset, count);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public override void Flush()
|
||||
{
|
||||
}
|
||||
|
||||
public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask;
|
||||
|
||||
public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException();
|
||||
|
||||
public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException();
|
||||
|
||||
public override void SetLength(long value) => throw new NotSupportedException();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user