Start with a live resource view, then add editing only where the page needs it. This guide wires the currently supported gateway stream into an existing Go application. The library owns stream recovery and browser reconciliation; your application owns identity, authorization policy, credentials and writes.
Choose scope, projection and sharing separately using watching resources. Choose the source for the guarantees the page needs: native access retains original resources through an existing host proxy; the gateway adds projected views, redaction, suppression and optional sharing, delivered over SSE. Both connectors share one fetch-based frontend lifecycle.
For native access, a host that already proxies Kubernetes needs no gateway: point
connectNativeWatch at a collection behind that proxy, as in the README
and the native viewer example. A reconnect resumes the watch,
and only an expired history lists again; to edit through the same proxy, see native editing.
The steps below wire the gateway source.
Each browser stream acts as its caller: the backend is built from that user's credential, so Kubernetes RBAC stays the boundary. The session cookie identifies the caller, and the request ends when the session does.
import (
"context"
"log/slog"
"net/http"
"time"
"k8s.io/client-go/rest"
"github.com/ConfigButler/krm-stream/gateway"
"github.com/ConfigButler/krm-stream/gateway/kube"
)
func mount(mux *http.ServeMux, restConfigFor func(*User) *rest.Config) {
stream := gateway.Handler(gateway.Options{
// Your session cookie -> your application user. krm-stream never sees a token.
Principal: func(r *http.Request) (gateway.Principal, error) {
return userFromSession(r)
},
Scopes: gateway.ScopePolicy{
Targets: []string{"production"},
Resources: []gateway.GroupResource{
{Group: "", Resource: "configmaps", Scope: gateway.ResourceScopeNamespaced},
{Group: "apps", Resource: "deployments", Scope: gateway.ResourceScopeNamespaced},
},
},
// Settings a stream needs however it is served; a Gateway embeds the same StreamConfig.
StreamConfig: gateway.StreamConfig{
// Deny before opening a watch; rechecked on every snapshot cycle.
Authorizer: gateway.AuthorizerFunc(func(ctx context.Context, p gateway.Principal, s gateway.Scope) error {
user := p.(*User)
if s.Target != user.Target || s.Namespace != user.Namespace {
return gateway.Forbidden("scope is not available to this user")
}
return nil
}),
// A backend acting as this user.
Clients: func(_ context.Context, _ string, p gateway.Principal) (gateway.Backend, error) {
return kube.NewBackendForConfig(restConfigFor(p.(*User)))
},
// One view for everyone. Nil means the same; a ProjectionPolicy can choose per caller.
Projections: gateway.StaticProjection(gateway.ProjectionFull),
WriteTimeout: 10 * time.Second, // bound each write to a browser that stops reading
Diagnostics: func(d gateway.Diagnostic) {
slog.Warn("stream error", "code", d.Code, "resource", d.Scope.Resource, "err", redact(d.Err))
},
},
})
mux.Handle("/resource-stream/v1", endWithSession(stream))
}
// endWithSession ends a stream when the caller's session expires.
func endWithSession(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if expiry, ok := sessionExpiry(r); ok {
ctx, cancel := context.WithDeadline(r.Context(), expiry)
defer cancel()
r = r.WithContext(ctx)
}
next.ServeHTTP(w, r)
})
}WriteTimeout needs a response writer that supports flushing and write deadlines. Check your mounted
middleware with the
capability test;
an unsupported writer aborts the request before streaming. The
shared-host example shows a complete session
resolver that uses the earlier of session and token expiry.
An empty namespace has two Kubernetes meanings: a cluster-scoped resource, or every namespace of a
namespaced resource. ScopePolicy makes the host choose explicitly.
| Resource kind | GroupResource |
Caller namespace | Result |
|---|---|---|---|
| one namespace of ConfigMaps | Scope: ResourceScopeNamespaced |
app |
watches only app |
| all namespaces of ConfigMaps | Scope: ResourceScopeNamespaced, AllowAllNamespaces: true |
empty | watches all namespaces |
| Namespaces | Scope: ResourceScopeCluster |
empty | watches cluster-scoped objects |
Namespaces with namespace=app |
Scope: ResourceScopeCluster |
app |
refused |
Keep all-namespaces access rare: it changes the size, disclosure risk and operating cost of a stream.
Use the Authorizer to pin a user to a namespace or target before any watch opens.
import {
LiveResourceStore,
readOnlyPolicy,
applyStreamEvent,
connectResourceStream,
resourceStreamURL,
} from "@configbutler/krm-stream";
const store = new LiveResourceStore(readOnlyPolicy); // list or viewer
const url = resourceStreamURL("/resource-stream/v1", {
target: "production",
version: "v1",
resource: "configmaps",
namespace: "app",
projection: "krm-full/v1",
});
const stopRendering = store.subscribe(() => render(store));
const connection = connectResourceStream(url, event => applyStreamEvent(store, event));
renderConnection(connection.state.status);
const stopConnection = connection.subscribe(state => renderConnection(state.status));
connection.closed.catch(reportApplicationError); // host callback failures stop the stream
// On view disposal:
stopRendering();
stopConnection();
connection.close();Fetch sends the same-origin session cookie, and the connection recovers on its own after network failures and sequence gaps.
For an editor, choose new LiveResourceStore() with the default policy or a form-specific policy.
Render the draft, edit through the store methods and resolve conflicts explicitly; see the
editor state model. Saving is the host's: see saving.
A quiet stream can still hold an older write version;
see why a quiet stream can reject a save.
Every snapshot cycle reauthorizes, but a quiet stream can go a long time without one. To bound how
long a revoked caller keeps it, opt into timed checks, and make the Authorizer check that the
caller's session is still valid, since the principal is the one captured when the stream opened:
options.ReauthorizationInterval = 30 * time.Second
options.ReauthorizationTimeout = 5 * time.Second
options.WriteTimeout = 10 * time.Second // required with timed checks; Handler panics without itPlan with the revocation budget.
SharedBackend opens one upstream watch per scope for every caller, as one service identity. Your
Authorizer is then the only check between a caller and the cached objects, so pair it with
kube.SubjectAccessReviewAuthorizer, which asks Kubernetes whether each caller may list and watch
the scope:
shared := gateway.NewSharedBackend(serviceAccountBackend, gateway.SharedOptions{
QueueDepth: 512,
Observer: metrics,
})
options.Authorizer = kube.SubjectAccessReviewAuthorizer(clientset, subjectFromUser)
options.Clients = func(context.Context, string, gateway.Principal) (gateway.Backend, error) { return shared, nil }Read shared-watch authorization first. The shared ConfigMap host is a tested composition with session expiry, timed checks and a write timeout.
Next: saving for the host-owned write path and operations for monitoring and limits.
A StreamError you return keeps its Message; its Cause never reaches the wire. Any other error
reaches the browser as INTERNAL with the message internal error. StreamConfig.Diagnostics receives
the raw error with its principal and scope; log what your policy allows.
Principal may return gateway.Unauthenticated when signing in again will help, or
gateway.UpstreamUnavailable when your session store is down. Any other Principal error is sent as
UNAUTHENTICATED "not authenticated".
ScopeFromQuery and ScopePolicy.Validate return error whose concrete value is a *StreamError;
reach its code with errors.As.
An endpoint that should expose whatever Kubernetes lets each user read, as a /k8s proxy does, can
replace the resource list with AnyResource:
Scopes: gateway.ScopePolicy{
Targets: []string{""},
AnyResource: true, // admission is left to the upstream's own authorization
},Any resource the scope syntax accepts then passes, with or without a namespace, and the API server
decides. The target allowlist, AllowLabelSelector, the refusal of API-server addresses and the
Authorizer still apply.
AnyResource is safe only when Clients returns a backend built from the caller's own credential.
The library cannot check that: with a service account's client or a hidden SharedBackend, every
resource that identity can read is open to everyone your Authorizer admits. Prove the binding in
your own tests.
A rest.Config.Host may include a path prefix, such as a kcp workspace URL
(https://kcp.example/clusters/root:org:ws); the dynamic client preserves it.
kube.NewBackendForConfig(cfg) includes the endpoint in upstream errors, which reach
StreamConfig.Diagnostics and never the browser. To use kube.NewBackend(dynamicClient) instead, build
the client on kube.HTTPClientFor(cfg) (dynamic.NewForConfigAndClient(cfg, httpClient)), which
refuses redirects so a user's token is not carried along.
connectResourceStream(url, consume, { headers: { Authorization: ... } }) suits a non-browser
client or an intentionally token-bearing browser application; the same-origin cookie route is the
safer browser default. The connector delegates transport to fetch and enforces no credential policy,
so the host must require HTTPS for the resolved destination, including redirects, before supplying
credentials.
A host may opt into OpenAPI-declared associative-list merging without exposing schemas to the browser:
const store = new LiveResourceStore(withOpenAPIKeyedLists(defaultPolicy, deploymentSchema));deploymentSchema is the structural schema for that exact GroupVersionKind. Only lists marked
x-kubernetes-list-type: map with x-kubernetes-list-map-keys are merged by key; every other list
stays atomic.