Quixio.PodEventRelay.Client 0.2.2

Quixio.PodEventRelay.Client

.NET 8 client SDK for the Quix Pod Event Relay API.

It covers the relay's REST endpoints (pod listing, pod/relay event history, auth sessions, health) and the Server-Sent Events stream, including gap-less, hash-based resumption with automatic reconnection.

Install

dotnet add package Quixio.PodEventRelay.Client

Quick start

using Quixio.PodEventRelay.Client;

using var httpClient = new HttpClient();
var client = new PodEventRelayClient(httpClient, new PodEventRelayClientOptions
{
    BaseUrl = "http://localhost:8080",
});

var pods = await client.ListPodsAsync();
Console.WriteLine($"pods: {pods.Count}");

await foreach (var podEvent in client.StreamPodEventsAsync(
    new StreamPodEventsOptions { ClientId = "my-app" }))
{
    Console.WriteLine($"[{podEvent.Type}] {podEvent.Pod?.Namespace}/{podEvent.Pod?.Name}");
}

Authenticated streams

When API authentication is enabled, supply a Kubernetes ServiceAccount token and mint a fresh one-time session before each (re)connect. Sessions are single-use, so use SessionRefresher() rather than a fixed session ID:

var client = new PodEventRelayClient(httpClient, new PodEventRelayClientOptions
{
    BaseUrl = "http://localhost:8080",
    Token = serviceAccountToken,
});

await foreach (var podEvent in client.StreamPodEventsAsync(new StreamPodEventsOptions
{
    ClientId = "my-controller",
    NewSessionAsync = client.SessionRefresher(), // omit when auth is disabled
}))
{
    // ...
}

By default the SDK sends Token in both Authorization and X-Relay-Authorization. That works for direct relay access and for Kubernetes API service proxy access, where the Kubernetes API server consumes Authorization and forwards X-Relay-Authorization to the relay:

var client = new PodEventRelayClient(httpClient, new PodEventRelayClientOptions
{
    BaseUrl = "https://k8s-api/api/v1/namespaces/quix/services/http:quix-pod-event-relay:8080/proxy",
    Token = serviceAccountToken,
});

Set SendTokenInAuthorizationHeader = false if the caller needs to provide different Kubernetes API proxy credentials on the HttpClient.

Gap-less reconnection

StreamPodEventsAsync advances the since_hash cursor as events arrive and resumes from it on reconnect. Persist the last processed hash after your business logic succeeds and pass it back as SinceHash on restart:

await foreach (var podEvent in client.StreamPodEventsAsync(new StreamPodEventsOptions
{
    ClientId = "my-app",
    SinceHash = savedHash,
}))
{
    Process(podEvent);
    savedHash = podEvent.Hash; // persist for resumption
}

If the relay reports a gap (expired cursor or a slow consumer overflowing its buffer), reseed current state from ListPodsAsync() before trusting subsequent events. Use the lower-level StreamAsync to observe gap, connected, and catchup_complete messages directly.

Dependency injection

services.AddPodEventRelayClient("http://pod-event-relay:8080");
// or
services.AddPodEventRelayClient(options =>
{
    options.BaseUrl = "http://pod-event-relay:8080";
    options.Token = token;
});

Then inject IPodEventRelayClient.

API surface

StreamPodStateAsync / PodStateCache exist for consumers that need the full set of pods (e.g. to group by a label and compute an aggregate) rather than individual events — for example, replacing a Kubernetes polling loop:

await foreach (var snapshot in client.StreamPodStateAsync(new StreamPodStateOptions
{
    ClientId = "my-monitor",
}))
{
    foreach (var (deploymentId, pods) in snapshot.Pods.GroupByLabel("run"))
    {
        // recompute aggregate state for this deployment from all its pods
    }
}

Use StreamPodStateWithReplayAsync when current state and historical replay must be handled differently. It emits:

• CurrentSeed from /pods immediately.
• retained events as Backfill.
• CatchupComplete when replay and live-buffer drain are complete.
• another CurrentSeed from /pods before live processing.
• live events as Live.
• GapReseed after a gap, without implying skipped events were processed.
• ReconnectReseed before reconnecting after a dropped stream.

Backfill does not mutate the current-state cache by default. Set ApplyBackfillToCurrentStateCache = true only when historical replay should be folded into the current snapshot.

Persist AcknowledgementHash only after processing succeeds:

await foreach (var snapshot in client.StreamPodStateWithReplayAsync(new StreamPodStateReplayOptions
{
    ClientId = "deployment-service",
    SinceHash = await cursorStore.Load(),
    Namespaces = new[] { "prod" },
    Labels = new Dictionary<string, string> { ["app"] = "worker" },
}))
{
    await Process(snapshot);

    if (snapshot.AcknowledgementHash is not null)
    {
        await cursorStore.Save(snapshot.AcknowledgementHash);
    }
}

For multiple relays (e.g. one per cluster) whose endpoints/tokens are only known at runtime, register AddPodEventRelayClientFactory() and call IPodEventRelayClientFactory.Create(baseUrl, token) per relay.

To access the relay through the Kubernetes API server service proxy, set BaseUrl to the proxied service endpoint, for example:

options.BaseUrl = "https://kubernetes.default.svc/api/v1/namespaces/relay-system/services/http:quix-pod-event-relay:8080/proxy";

Versioning

The package version is defined in code via <Version> in the project file, not derived from a git tag. Bump it in a commit, then trigger the manual Publish .NET Client GitHub workflow.

License

Apache-2.0.