mirror of
https://github.com/vee1e/krply.git
synced 2026-09-01 17:57:03 +00:00
lakshit verma
44fbd878a1
Storage: fixed-width millisecond timestamp format with a one-time migration so lexicographic comparisons stay chronological; snapshot auto-IDs get nanosecond precision; Close is synchronized; commit failures roll back; DSN paths are URI-escaped; the dedup index is no longer rebuilt every startup; the in-memory store now dedups by event_id and honors Limit/Offset/SinceSeq. Event/discovery: stream IDs URL-escape the selector so slash-containing label selectors round-trip and cannot collide; alias resolution fills empty fields only and no longer clobbers explicit group/version/kind; rc maps to replicationcontrollers and rs to replicasets. Watch: the label selector is now actually applied to List and Watch; the 410 relist path is backoff-throttled, backoff resets after a healthy watch, an empty list resourceVersion backs off instead of looping forever, non-410 watch errors reconnect without writing a spurious permanent gap, an idle watch timer forces reconnects, and Run cancels sibling streams and recovers from panics. Synthetic relist events carry a distinct event_id so unchanged objects re-listed after a gap survive dedup. Materialize: a baseline now resets object state (objects deleted during a gap disappear) and heals open gaps; diff ignores server metadata by path, not by key name, so user fields named status/uid are kept; nested add/remove carry Added/Removed flags; dotted paths are escaped; Diff validates the window and treats a zero before as empty; Snapshot records per-stream watermarks and writes a TypeSnapshot journal record. Replay: apply requires a successful dry run; gvrFor covers every policy kind so Include* toggles work; unsupported kinds are reported as skipped instead of silently dropped; NodePort services lose clusterIP and nodePort while headless keeps clusterIP: None; targetNS is always honored; namespace mapping collisions are detected; plan IDs and field managers are collision-safe and plans are mutex-guarded against concurrent dry-runs/applies. API: cursor pagination pushes ingest_seq into the SQL filter (export --server no longer truncates); gap records expose their payload; /v1/diff requires a cluster_id; dry-run/apply no longer accept a client-supplied kubeconfig; plans carry target_context; HTTP server gets timeouts. Metrics: the watch collector and replay planner now bump the registered counters, and the server refreshes store-derived gauges on a ticker. CLI: replay apply works locally and refuses to apply after a failed dry run; coverage --server follows cursors and shows gap details; timeline auto-detects namespaces and resolves cluster-scoped objects; diff validates the window; export writes 0600 files; PVCs are no longer treated as cluster-scoped; krply-server reports the build version.
106 lines
2.7 KiB
Go
106 lines
2.7 KiB
Go
package materialize
|
|
|
|
import (
|
|
"context"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/krply/krply/internal/event"
|
|
"github.com/krply/krply/internal/storage"
|
|
)
|
|
|
|
// Snapshot is a materialized view of a cluster at a point in time.
|
|
type Snapshot struct {
|
|
ID string
|
|
ClusterID string
|
|
Name string
|
|
At time.Time
|
|
Objects []ObjectState
|
|
Streams []string
|
|
Watermarks []StreamWatermark
|
|
Complete bool
|
|
Missing []string
|
|
Warning string
|
|
}
|
|
|
|
// StreamWatermark is the durable boundary of one stream at snapshot time.
|
|
type StreamWatermark struct {
|
|
StreamID string
|
|
LastObservedAt time.Time
|
|
LastResourceVersion string
|
|
}
|
|
|
|
// Snapshot materializes every watched object of the cluster at at, records
|
|
// per-stream coverage and watermarks, persists a TypeSnapshot journal entry,
|
|
// and stores the snapshot metadata.
|
|
func (m *Materializer) Snapshot(ctx context.Context, clusterID string, at time.Time, name string) (*Snapshot, error) {
|
|
streams, err := m.store.Streams(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
snap := &Snapshot{
|
|
ID: "snap-" + uuid.NewString()[:8] + "-" + name,
|
|
ClusterID: clusterID,
|
|
Name: name,
|
|
At: at,
|
|
Complete: true,
|
|
}
|
|
|
|
var missing []string
|
|
for _, s := range streams {
|
|
if s.ClusterID != clusterID {
|
|
continue
|
|
}
|
|
ss, err := m.StreamState(ctx, s.StreamID, at)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
snap.Objects = append(snap.Objects, ss.Objects...)
|
|
snap.Streams = append(snap.Streams, s.StreamID)
|
|
snap.Watermarks = append(snap.Watermarks, StreamWatermark{
|
|
StreamID: s.StreamID,
|
|
LastObservedAt: ss.LastObservedAt,
|
|
LastResourceVersion: ss.LastResourceVersion,
|
|
})
|
|
|
|
if !ss.HasBaseline || ss.HasGaps {
|
|
missing = append(missing, s.StreamID)
|
|
}
|
|
}
|
|
|
|
sort.Strings(snap.Streams)
|
|
sortObjects(snap.Objects)
|
|
sort.Slice(snap.Watermarks, func(i, j int) bool { return snap.Watermarks[i].StreamID < snap.Watermarks[j].StreamID })
|
|
sort.Strings(missing)
|
|
|
|
if len(missing) > 0 {
|
|
snap.Complete = false
|
|
snap.Missing = missing
|
|
snap.Warning = "coverage incomplete for stream(s): " + strings.Join(missing, ", ")
|
|
}
|
|
|
|
if err := m.store.SaveSnapshot(ctx, &storage.SnapshotRef{
|
|
ID: snap.ID,
|
|
ClusterID: clusterID,
|
|
Name: name,
|
|
At: at,
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Persist a TypeSnapshot journal record so the snapshot is reconstructable
|
|
// and provable from the event stream, not only from the metadata table.
|
|
if _, err := m.store.Append(ctx, &event.Record{
|
|
ClusterID: clusterID,
|
|
ObservedAt: at,
|
|
Type: event.TypeSnapshot,
|
|
Snapshot: &event.SnapshotInfo{Name: name},
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return snap, nil
|
|
}
|