Efficiency follow-ups: shared scrape sample, parallel startup probes and status reconcile

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/<pid>/stat`) and memory (RSS from
-`/proc/<pid>/status`), plus per-pod network I/O from the TAP device
+(cumulative nanoseconds) and memory (RSS field 24) from one
+`/proc/<pid>/stat` read, plus per-pod network I/O from the TAP device
 inside each VM's network namespace (`/proc/<pid>/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
 

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

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=

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

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,13 +63,15 @@ 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]
 		runtime := meta.ParseVMRuntime(pod)
 		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
 }

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)

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()
 }