diff --git a/docs/metrics.md b/docs/metrics.md index cf7e54e..3e48560 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -6,8 +6,8 @@ defaults to `10250` and is set by `VK_KUBELET_PORT`. ## `:10250/stats/summary` — kubelet stats API Consumed by metrics-server and `kubectl top`. Reports per-pod CPU -(cumulative nanoseconds from `/proc//stat`) and memory (RSS from -`/proc//status`), plus per-pod network I/O from the TAP device +(cumulative nanoseconds) and memory (RSS field 24) from one +`/proc//stat` read, plus per-pod network I/O from the TAP device inside each VM's network namespace (`/proc//net/dev`). Node-level CPU and memory are read from `/proc/stat` and `/proc/meminfo`. @@ -55,12 +55,13 @@ Prometheus endpoint with vk-cocoon-specific metrics: | `cocoon_vk_hibernate_evidence_total{verdict}` | Counter | Fresh boots intercepted by hibernate evidence (`verdict=restored\|image_conflict\|source_conflict\|unavailable`) | | `cocoon_vk_startup_resume_total{op}` | Counter | Interrupted operations re-dispatched by startup reconcile (`op=hibernate\|post_clone\|ready_wait\|classify_drop_nic`) | -All per-VM stats are read from `/proc` using the hypervisor PID tracked -in memory — no shell-out to `cocoon` on each scrape. The tracking table -is snapshot-copied under RLock and `/proc` reads happen outside the lock -to avoid blocking CreatePod/DeletePod. When a VM is restarted in-place -(event watcher → `cocoon vm start`), the PID is re-inspected and -refreshed. +All three metrics surfaces share one complete sample for two seconds, +so reported values may trail `/proc` by up to that interval. Per-VM +stats use the hypervisor PID tracked in memory — no shell-out to `cocoon` +on each scrape. The tracking table is snapshot-copied under RLock and +`/proc` reads happen outside the lock to avoid blocking +CreatePod/DeletePod. When a VM is restarted in-place (event watcher → +`cocoon vm start`), the PID is re-inspected and refreshed. ## Kubernetes Events diff --git a/go.mod b/go.mod index dd5913b..8eb758d 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/cocoonstack/vk-cocoon go 1.26.5 require ( - github.com/cocoonstack/cocoon-common v0.2.8 + github.com/cocoonstack/cocoon-common v0.2.9-0.20260728170736-b0a8b954c944 github.com/google/go-containerregistry v0.21.7 github.com/projecteru2/core v0.0.0-20241016125006-ff909eefe04c github.com/prometheus/client_golang v1.23.2 diff --git a/go.sum b/go.sum index 9675a26..83a9e12 100644 --- a/go.sum +++ b/go.sum @@ -43,8 +43,8 @@ github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZe github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= github.com/cockroachdb/redact v1.1.3 h1:AKZds10rFSIj7qADf0g46UixK8NNLwWTNdCIGS5wfSQ= github.com/cockroachdb/redact v1.1.3/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= -github.com/cocoonstack/cocoon-common v0.2.8 h1:xpfNVTBmP/VGD/XB4oDzCI7bO1Ygb8Yd9Gnq8U/VJh8= -github.com/cocoonstack/cocoon-common v0.2.8/go.mod h1:VSfgYiWxoHRnWybzQNaxmK4kVoLT9ffiMKll5/i92CM= +github.com/cocoonstack/cocoon-common v0.2.9-0.20260728170736-b0a8b954c944 h1:Jru/8LxzwHZKVjFklwClA/UU6oRV9wVVTIJxLSeMBlg= +github.com/cocoonstack/cocoon-common v0.2.9-0.20260728170736-b0a8b954c944/go.mod h1:VSfgYiWxoHRnWybzQNaxmK4kVoLT9ffiMKll5/i92CM= github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM= github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE= github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk= diff --git a/provider/cocoon/provider.go b/provider/cocoon/provider.go index 5a278bf..43e8e3f 100644 --- a/provider/cocoon/provider.go +++ b/provider/cocoon/provider.go @@ -12,6 +12,7 @@ import ( "time" "github.com/projecteru2/core/log" + "golang.org/x/sync/errgroup" "golang.org/x/sync/singleflight" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -54,6 +55,12 @@ const ( // a single CLI hiccup the deferred recheck takes over. inlineInspectAttempts = 2 + // startupFanOut bounds the boot-gating fan-outs (stale creates, first + // probe starts); statusReconcileFanOut bounds the steady-state status + // drift loop against the apiserver. Equal today, tuned separately. + startupFanOut = 8 + statusReconcileFanOut = 8 + // Default tunables for the recheck path. Overridable via Provider // fields so tests can shrink them without racing on package globals. defaultInlineInspectBaseDelay = 200 * time.Millisecond @@ -104,6 +111,12 @@ type Provider struct { lifecycleIntent map[string]meta.LifecycleStatus lifecycleFlushed map[string]string + // Shared scrape sample; see sampleStats. + statsMu sync.Mutex + statsAt time.Time + statsVMs []vmSample + statsNode provider.NodeStats + // Recheck tunables. Zero values fall back to the defaultXxx // constants, so production code never sets them; tests shrink them // before exercising handleVMGone. @@ -231,24 +244,24 @@ func (p *Provider) reconcilePodStatuses(ctx context.Context) { return } logger := log.WithFunc("Provider.reconcilePodStatuses") - for _, pod := range pods { + fanOut(statusReconcileFanOut, pods, func(pod *corev1.Pod) { current, err := p.Clientset.CoreV1().Pods(pod.Namespace).Get(ctx, pod.Name, metav1.GetOptions{}) if err != nil { logger.Errorf(ctx, err, "get pod %s/%s for status reconciliation", pod.Namespace, pod.Name) - continue + return } status, err := p.GetPodStatus(ctx, pod.Namespace, pod.Name) if err != nil { logger.Errorf(ctx, err, "derive pod %s/%s status", pod.Namespace, pod.Name) - continue + return } if podStatusMatches(current.Status, *status) { - continue + return } current.Status = *status logger.Infof(ctx, "republishing drifted status for pod %s/%s", pod.Namespace, pod.Name) p.notify(current) - } + }) } func (p *Provider) notify(pod *corev1.Pod) { @@ -639,6 +652,16 @@ func (p *Provider) recheckBackoff() (delay, maxDelay, budget time.Duration) { cmp.Or(p.deferredRecheckBudget, defaultDeferredRecheckBudget) } +// fanOut runs f over items with bounded concurrency; f logs its own failures. +func fanOut[T any](limit int, items []T, f func(T)) { + var g errgroup.Group + g.SetLimit(limit) + for _, item := range items { + g.Go(func() error { f(item); return nil }) + } + _ = g.Wait() +} + // podForVMMatch returns the pod and tracked-VM ID for a pod that matches // the given id or (optionally) name. name may be empty to restrict the // match to id only. Used by handleVMGone (match on id OR name from a diff --git a/provider/cocoon/reconcile.go b/provider/cocoon/reconcile.go index d02fadf..8e9ac1b 100644 --- a/provider/cocoon/reconcile.go +++ b/provider/cocoon/reconcile.go @@ -20,8 +20,6 @@ import ( "github.com/cocoonstack/vk-cocoon/vm" ) -const staleCreateConcurrency = 8 - // StartupReconcile rebuilds the in-memory tables from K8s pods and // cocoon VMs so restarts don't leak VMs or lose pod associations. // Unmatched VMs are handled per OrphanPolicy. @@ -65,6 +63,7 @@ func (p *Provider) StartupReconcile(ctx context.Context) error { } } matched := make(map[string]bool, len(vms)) + var probePods []*corev1.Pod for i := range podItems(pods) { pod := &pods.Items[i] @@ -72,6 +71,7 @@ func (p *Provider) StartupReconcile(ctx context.Context) error { if runtime.VMID == "" { if v := p.adoptByVMName(ctx, pod, vmByName); v != nil { matched[v.ID] = true + probePods = append(probePods, pod) continue } p.reconcileNoVMID(ctx, pod) @@ -91,8 +91,11 @@ func (p *Provider) StartupReconcile(ctx context.Context) error { p.trackPod(pod, v) p.seedLifecycleIntentFromPod(pod) matched[v.ID] = true - p.startProbeIfEnabled(pod) + probePods = append(probePods, pod) } + // First probes run synchronously (3s worst case each) and this path + // gates node registration — start them bounded-parallel. + fanOut(startupFanOut, probePods, p.startProbeIfEnabled) for i := range vms { if matched[vms[i].ID] { @@ -112,7 +115,7 @@ func (p *Provider) reconcileStaleCreates(ctx context.Context, vms []vm.VM) []vm. logger := log.WithFunc("Provider.reconcileStaleCreates") keep := make([]*vm.VM, len(vms)) var g errgroup.Group - g.SetLimit(staleCreateConcurrency) + g.SetLimit(startupFanOut) for i := range vms { v := &vms[i] if v.State != vm.StateCreating { @@ -228,7 +231,6 @@ func (p *Provider) adoptByVMName(ctx context.Context, pod *corev1.Pod, idx map[s p.applyRuntime(ctx, pod, v) p.trackPod(pod, v) p.seedLifecycleIntentFromPod(pod) - p.startProbeIfEnabled(pod) metrics.ReconcileAdoptByNameTotal.Inc() return v } diff --git a/provider/cocoon/restore_test.go b/provider/cocoon/restore_test.go index 0357121..2bd3728 100644 --- a/provider/cocoon/restore_test.go +++ b/provider/cocoon/restore_test.go @@ -208,7 +208,7 @@ func TestCreatePodEvidenceFailClosedOnRegistryError(t *testing.T) { rt := &fakeRuntime{} p := newTestProvider(t) p.Runtime = rt - p.Registry = wakeVerifyRegistry{hasManifestErr: errors.New("registry down")} + p.Registry = wakeVerifyRegistry{manifestErr: errors.New("registry down")} pod := newPodWithSpec(meta.VMSpec{VMName: "vk-ns-demo-0", Image: "snapshot-repo:latest", Mode: "clone"}) err := p.CreatePod(t.Context(), pod) diff --git a/provider/cocoon/resume_test.go b/provider/cocoon/resume_test.go index d302d2f..585db22 100644 --- a/provider/cocoon/resume_test.go +++ b/provider/cocoon/resume_test.go @@ -3,6 +3,7 @@ package cocoon import ( "context" "errors" + "fmt" "strings" "sync" "testing" @@ -14,6 +15,7 @@ import ( "k8s.io/client-go/kubernetes/fake" "github.com/cocoonstack/cocoon-common/meta" + "github.com/cocoonstack/cocoon-common/snapshot" "github.com/cocoonstack/vk-cocoon/vm" ) @@ -380,29 +382,29 @@ func awaitLifecycle(t *testing.T, p *Provider, namespace, name string, want meta t.Fatalf("lifecycle never reached %q (pod: %v, err: %v)", want, pod, err) } -// flakyEvidenceRegistry errors HasManifest a set number of times, then -// reports no hibernate tag. +// flakyEvidenceRegistry errors the manifest fetch a set number of times, +// then reports no hibernate tag. type flakyEvidenceRegistry struct { fakeRegistry mu sync.Mutex fails int } -func (r *flakyEvidenceRegistry) HasManifest(context.Context, string, string) (bool, error) { +func (r *flakyEvidenceRegistry) GetManifest(context.Context, string, string) ([]byte, string, error) { r.mu.Lock() defer r.mu.Unlock() if r.fails > 0 { r.fails-- - return false, errors.New("registry down") + return nil, "", errors.New("registry down") } - return false, nil + return nil, "", fmt.Errorf("get manifest: %w", snapshot.ErrManifestNotFound) } // blockingEvidenceRegistry accepts the lookup and never answers until the // caller's context dies. type blockingEvidenceRegistry struct{ fakeRegistry } -func (blockingEvidenceRegistry) HasManifest(ctx context.Context, _, _ string) (bool, error) { +func (blockingEvidenceRegistry) GetManifest(ctx context.Context, _, _ string) ([]byte, string, error) { <-ctx.Done() - return false, ctx.Err() + return nil, "", ctx.Err() } diff --git a/provider/cocoon/stats.go b/provider/cocoon/stats.go index 9c79ecd..aaf3bb2 100644 --- a/provider/cocoon/stats.go +++ b/provider/cocoon/stats.go @@ -8,6 +8,8 @@ import ( "strings" "time" + "github.com/cocoonstack/vk-cocoon/vm" + dto "github.com/prometheus/client_model/go" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" statsv1alpha1 "k8s.io/kubelet/pkg/apis/stats/v1alpha1" @@ -17,6 +19,8 @@ import ( "github.com/cocoonstack/vk-cocoon/provider" ) +const statsSampleTTL = 2 * time.Second + // vmSnapshot is a minimal copy of VM state taken under lock so /proc // reads happen outside the critical section. type vmSnapshot struct { @@ -30,15 +34,25 @@ type vmSnapshot struct { VMName string } +// vmSample is one cached stats reading for a tracked VM. +type vmSample struct { + vmSnapshot + cpuSeconds float64 + memBytes int64 + diskCOW int64 + rxBytes uint64 + txBytes uint64 +} + // metrics-server and kubectl top consume this endpoint. func (p *Provider) GetStatsSummary(_ context.Context) (*statsv1alpha1.Summary, error) { now := metav1.Now() - nodeCPU, nodeMemory := cpuMemStats(readNodeCPUSeconds(), readNodeMemoryWorkingSet()) + samples, node := p.sampleStats() + nodeCPU, nodeMemory := cpuMemStats(node.CPUSeconds, node.MemoryUsedBytes) - snapshots := p.snapshotTrackedVMs() - podStats := make([]statsv1alpha1.PodStats, 0, len(snapshots)) - for _, s := range snapshots { - cpu, mem := cpuMemStats(readProcessCPUSeconds(s.PID), readProcessMemoryWorkingSet(s.PID)) + podStats := make([]statsv1alpha1.PodStats, 0, len(samples)) + for _, s := range samples { + cpu, mem := cpuMemStats(s.cpuSeconds, s.memBytes) ps := statsv1alpha1.PodStats{ PodRef: statsv1alpha1.PodReference{Name: s.PodName, Namespace: s.Namespace}, StartTime: now, @@ -46,7 +60,7 @@ func (p *Provider) GetStatsSummary(_ context.Context) (*statsv1alpha1.Summary, e Name: containerName, StartTime: now, CPU: cpu, Memory: mem, }}, } - if net := buildNetworkStats(s.PID, s.Tap); net != nil { + if net := buildNetworkStats(s); net != nil { ps.Network = net } podStats = append(podStats, ps) @@ -65,21 +79,21 @@ func (p *Provider) GetStatsSummary(_ context.Context) (*statsv1alpha1.Summary, e func (p *Provider) GetMetricsResource(_ context.Context) ([]*dto.MetricFamily, error) { nowMs := time.Now().UnixMilli() + samples, node := p.sampleStats() families := []*dto.MetricFamily{ newCounterFamily("node_cpu_usage_seconds_total", "Cumulative cpu time consumed by the node in core-seconds", - newCounter(readNodeCPUSeconds(), nowMs, nil)), + newCounter(node.CPUSeconds, nowMs, nil)), newGaugeFamily("node_memory_working_set_bytes", "Current working set of the node in bytes", - newGauge(float64(readNodeMemoryWorkingSet()), nowMs, nil)), + newGauge(float64(node.MemoryUsedBytes), nowMs, nil)), } - snapshots := p.snapshotTrackedVMs() var containerCPU, containerMem, podCPU, podMem []*dto.Metric - for _, s := range snapshots { - cpuSec := readProcessCPUSeconds(s.PID) - memBytes := float64(readProcessMemoryWorkingSet(s.PID)) + for _, s := range samples { + cpuSec := s.cpuSeconds + memBytes := float64(s.memBytes) containerLabels := []*dto.LabelPair{ {Name: new("namespace"), Value: new(s.Namespace)}, @@ -114,6 +128,36 @@ func (p *Provider) GetMetricsResource(_ context.Context) ([]*dto.MetricFamily, e return families, nil } +// sampleStats returns the shared short-TTL scrape sample: three consumers +// (stats summary, metrics resource, Prometheus) land independently on it. +func (p *Provider) sampleStats() ([]vmSample, provider.NodeStats) { + p.statsMu.Lock() + defer p.statsMu.Unlock() + if time.Since(p.statsAt) < statsSampleTTL { + return p.statsVMs, p.statsNode + } + snaps := p.snapshotTrackedVMs() + vms := make([]vmSample, 0, len(snaps)) + for _, s := range snaps { + cpu, rss := readProcStatCPURSS(s.PID) + sample := vmSample{ + vmSnapshot: s, cpuSeconds: cpu, memBytes: rss, + diskCOW: vm.COWSize(provider.CocoonRootDir(), s.Hypervisor, s.ID), + } + if s.Tap != "" { + sample.rxBytes, sample.txBytes = readProcNetDev(s.PID, s.Tap) + } + vms = append(vms, sample) + } + node := provider.NodeStats{ + CPUSeconds: readNodeCPUSeconds(), + MemoryUsedBytes: readNodeMemoryWorkingSet(), + } + node.StorageTotal, node.StorageAvailable = provider.StorageBytes() + p.statsVMs, p.statsNode, p.statsAt = vms, node, time.Now() + return vms, node +} + // snapshotTrackedVMs copies the minimal VM data needed for stats under // RLock, then releases it so /proc reads don't block CreatePod/DeletePod. func (p *Provider) snapshotTrackedVMs() []vmSnapshot { @@ -141,17 +185,14 @@ func (p *Provider) snapshotTrackedVMs() []vmSnapshot { return out } -func buildNetworkStats(pid int, tap string) *statsv1alpha1.NetworkStats { - if tap == "" || pid == 0 { - return nil - } - rx, tx := readProcNetDev(pid, tap) - if rx == 0 && tx == 0 { +func buildNetworkStats(s vmSample) *statsv1alpha1.NetworkStats { + if s.rxBytes == 0 && s.txBytes == 0 { return nil } + rx, tx := s.rxBytes, s.txBytes return &statsv1alpha1.NetworkStats{ InterfaceStats: statsv1alpha1.InterfaceStats{ - Name: tap, RxBytes: &rx, TxBytes: &tx, + Name: s.Tap, RxBytes: &rx, TxBytes: &tx, }, } } @@ -200,31 +241,29 @@ func cpuMemStats(cpuSeconds float64, memBytes int64) (*statsv1alpha1.CPUStats, * &statsv1alpha1.MemoryStats{WorkingSetBytes: &mem} } -func readProcessCPUSeconds(pid int) float64 { +func readProcStatCPURSS(pid int) (cpuSeconds float64, rssBytes int64) { data, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat") if err != nil { - return 0 + return 0, 0 } - s := string(data) + return parseProcStat(string(data), os.Getpagesize()) +} + +// parseProcStat extracts utime+stime and RSS from a /proc//stat line +// (fields 14, 15, 24; split after the parenthesized comm, which may contain spaces). +func parseProcStat(s string, pageSize int) (cpuSeconds float64, rssBytes int64) { idx := strings.LastIndex(s, ")") if idx < 0 || idx+2 >= len(s) { - return 0 + return 0, 0 } fields := strings.Fields(s[idx+2:]) - if len(fields) < 13 { - return 0 + if len(fields) < 22 { + return 0, 0 } utime, _ := strconv.ParseInt(fields[11], 10, 64) stime, _ := strconv.ParseInt(fields[12], 10, 64) - return float64(utime+stime) / 100 // CLK_TCK -} - -func readProcessMemoryWorkingSet(pid int) int64 { - fields, err := provider.ReadKeyedProcFile("/proc/"+strconv.Itoa(pid)+"/status", "VmRSS") - if err != nil { - return 0 - } - return fields["VmRSS"] * 1024 + rssPages, _ := strconv.ParseInt(fields[21], 10, 64) + return float64(utime+stime) / 100, rssPages * int64(pageSize) // CLK_TCK } func newCounterFamily(name, help string, metrics ...*dto.Metric) *dto.MetricFamily { diff --git a/provider/cocoon/stats_collector.go b/provider/cocoon/stats_collector.go index 8203aa4..9c16224 100644 --- a/provider/cocoon/stats_collector.go +++ b/provider/cocoon/stats_collector.go @@ -2,38 +2,25 @@ package cocoon import ( "github.com/cocoonstack/vk-cocoon/provider" - "github.com/cocoonstack/vk-cocoon/vm" ) // CollectVMStats returns per-VM and node-level stats for the Prometheus // collector. Called on every scrape from the metrics endpoint. func (p *Provider) CollectVMStats() ([]provider.VMStats, provider.NodeStats) { - snapshots := p.snapshotTrackedVMs() - - out := make([]provider.VMStats, 0, len(snapshots)) - for _, s := range snapshots { - var rxBytes, txBytes uint64 - if s.Tap != "" { - rxBytes, txBytes = readProcNetDev(s.PID, s.Tap) - } + samples, node := p.sampleStats() + out := make([]provider.VMStats, 0, len(samples)) + for _, s := range samples { out = append(out, provider.VMStats{ VMName: s.VMName, PodName: s.PodName, Namespace: s.Namespace, Backend: s.Backend, - CPUSeconds: readProcessCPUSeconds(s.PID), - MemoryRSS: readProcessMemoryWorkingSet(s.PID), - DiskCOW: vm.COWSize(provider.CocoonRootDir(), s.Hypervisor, s.ID), - NetRxBytes: rxBytes, - NetTxBytes: txBytes, + CPUSeconds: s.cpuSeconds, + MemoryRSS: s.memBytes, + DiskCOW: s.diskCOW, + NetRxBytes: s.rxBytes, + NetTxBytes: s.txBytes, }) } - - node := provider.NodeStats{ - CPUSeconds: readNodeCPUSeconds(), - MemoryUsedBytes: readNodeMemoryWorkingSet(), - } - node.StorageTotal, node.StorageAvailable = provider.StorageBytes() - return out, node } diff --git a/provider/cocoon/stats_test.go b/provider/cocoon/stats_test.go new file mode 100644 index 0000000..d04c986 --- /dev/null +++ b/provider/cocoon/stats_test.go @@ -0,0 +1,45 @@ +package cocoon + +import ( + "testing" + "time" + + "github.com/cocoonstack/vk-cocoon/provider" +) + +func TestParseProcStat(t *testing.T) { + // comm carries spaces and parens; fields after it: state=R, then 20+ numeric. + line := "1234 (cloud (hv) proc) R 1 1 1 0 -1 4194560 100 0 0 0 250 150 0 0 20 0 4 0 12345 999424 512 18446744073709551615" + cpu, rss := parseProcStat(line, 4096) + if cpu != 4.0 { + t.Errorf("cpu = %v, want 4.0 ((250+150)/100)", cpu) + } + if rss != 512*4096 { + t.Errorf("rss = %d, want %d", rss, 512*4096) + } +} + +func TestParseProcStatMalformed(t *testing.T) { + for _, s := range []string{"", "no comm here", "1 (x) R 1 2 3"} { + if cpu, rss := parseProcStat(s, 4096); cpu != 0 || rss != 0 { + t.Errorf("parseProcStat(%q) = %v,%d, want zeros", s, cpu, rss) + } + } +} + +func TestSampleStatsServesCachedWithinTTL(t *testing.T) { + p := newTestProvider(t) + seeded := []vmSample{{vmSnapshot: vmSnapshot{VMName: "vk-ns-demo-0"}, cpuSeconds: 7}} + p.statsVMs, p.statsNode, p.statsAt = seeded, provider.NodeStats{CPUSeconds: 42}, time.Now() + + vms, node := p.sampleStats() + if len(vms) != 1 || vms[0].cpuSeconds != 7 || node.CPUSeconds != 42 { + t.Fatalf("within TTL must serve the cached sample, got %+v node %+v", vms, node) + } + + p.statsAt = time.Now().Add(-2 * statsSampleTTL) + vms, _ = p.sampleStats() + if len(vms) != 0 { + t.Fatalf("expired TTL must resample (no tracked VMs), got %+v", vms) + } +} diff --git a/provider/cocoon/update.go b/provider/cocoon/update.go index 1fad492..6a21b03 100644 --- a/provider/cocoon/update.go +++ b/provider/cocoon/update.go @@ -434,17 +434,14 @@ func (p *Provider) verifyLocalSnapshot(ctx context.Context, vmName string, local } // fetchHibernateManifest returns vmName's parsed hibernate-tag manifest; -// ok=false means the registry has no such tag. +// ok=false is the registry's authoritative no-such-tag (typed 404), and any +// other error keeps the evidence checks failing closed. func (p *Provider) fetchHibernateManifest(ctx context.Context, vmName string) (*manifest.OCIManifest, bool, error) { - exists, err := p.Registry.HasManifest(ctx, vmName, meta.HibernateSnapshotTag) - if err != nil { - return nil, false, fmt.Errorf("check hibernate tag: %w", err) - } - if !exists { - return nil, false, nil - } raw, _, err := p.Registry.GetManifest(ctx, vmName, meta.HibernateSnapshotTag) - if err != nil { + switch { + case errors.Is(err, commonsnapshot.ErrManifestNotFound): + return nil, false, nil + case err != nil: return nil, false, fmt.Errorf("get hibernate manifest: %w", err) } m, err := manifest.Parse(raw) diff --git a/provider/cocoon/wakeverify_test.go b/provider/cocoon/wakeverify_test.go index 77da2b8..9b1ac13 100644 --- a/provider/cocoon/wakeverify_test.go +++ b/provider/cocoon/wakeverify_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "slices" "strings" @@ -12,6 +13,7 @@ import ( "github.com/cocoonstack/cocoon-common/manifest" "github.com/cocoonstack/cocoon-common/ociutil" + "github.com/cocoonstack/cocoon-common/snapshot" "github.com/cocoonstack/vk-cocoon/vm" ) @@ -79,7 +81,7 @@ func TestResolveWakeSourceRegistryErrorFailsClosed(t *testing.T) { }} p := newTestProvider(t) p.Runtime = rt - p.Registry = wakeVerifyRegistry{tagExists: false, hasManifestErr: errors.New("registry down")} + p.Registry = wakeVerifyRegistry{manifestErr: errors.New("registry down")} _, _, err := p.resolveWakeSource(t.Context(), "vk-ns-demo-0") if err == nil || !strings.Contains(err.Error(), "registry down") { @@ -94,10 +96,10 @@ func TestResolveWakeSourceRegistryErrorFailsClosed(t *testing.T) { // blob) so resolveWakeSource's local-cache verification can run against it. type wakeVerifyRegistry struct { fakeRegistry - tagExists bool - hasManifestErr error - manifestRaw []byte - blobs map[string][]byte + tagExists bool + manifestErr error + manifestRaw []byte + blobs map[string][]byte } func newWakeVerifyRegistry(t *testing.T, snapshotID string) wakeVerifyRegistry { @@ -111,11 +113,13 @@ func newWakeVerifyRegistryWithImage(t *testing.T, snapshotID, baseImage string) return wakeVerifyRegistry{tagExists: true, manifestRaw: raw, blobs: blobs} } -func (r wakeVerifyRegistry) HasManifest(context.Context, string, string) (bool, error) { - return r.tagExists, r.hasManifestErr -} - func (r wakeVerifyRegistry) GetManifest(context.Context, string, string) ([]byte, string, error) { + if r.manifestErr != nil { + return nil, "", r.manifestErr + } + if !r.tagExists { + return nil, "", fmt.Errorf("get manifest: %w", snapshot.ErrManifestNotFound) + } return r.manifestRaw, "", nil }