From ca5c8280f322c00f05813e09465a6ccfd83c890d Mon Sep 17 00:00:00 2001 From: zhengyilei Date: Fri, 3 Jul 2026 11:12:38 +0800 Subject: [PATCH] fix(template): probe envd readiness before annotation Probe envd /health during template snapshot flows before reporting an envd version from Cubelet. This makes the envd capability annotation mean that the template has a usable envd service, not just an envd binary that may be stopped or still starting. Apply the same readiness check to both AppSnapshot and CommitSandbox. Only persist cube.master.components.envd.version when all READY replicas report the same valid, ready envd version. Otherwise, omit the annotation so later sandbox create flows do not rely on an optimistic capability signal. Signed-off-by: zhengyilei --- CubeMaster/pkg/templatecenter/store.go | 59 +++-- CubeMaster/pkg/templatecenter/store_test.go | 173 ++++++++++++- Cubelet/services/cubebox/appsnapshot.go | 5 +- Cubelet/services/cubebox/envd_version.go | 211 +++++++++++++++- Cubelet/services/cubebox/envd_version_test.go | 235 ++++++++++++++++++ Cubelet/services/cubebox/service.go | 7 - Cubelet/services/cubebox/template_ops.go | 8 +- 7 files changed, 655 insertions(+), 43 deletions(-) diff --git a/CubeMaster/pkg/templatecenter/store.go b/CubeMaster/pkg/templatecenter/store.go index 8eb73d482..124625cd5 100644 --- a/CubeMaster/pkg/templatecenter/store.go +++ b/CubeMaster/pkg/templatecenter/store.go @@ -482,6 +482,7 @@ func createTemplateReplicasOnNodes(ctx context.Context, templateID string, req * replicas = append(replicas, replica) envdVersions = append(envdVersions, nodeEnvdVersion{ NodeID: target.ID(), + Ready: replica.Status == ReplicaStatusReady, Version: envdVersion, }) lock.Unlock() @@ -495,49 +496,58 @@ func createTemplateReplicasOnNodes(ctx context.Context, templateID string, req * }() } wg.Wait() - // Converge per-node envd versions into a single template value and persist it - // once to the definition annotation (idempotent; covers create and redo). - if envdVersion := convergeEnvdVersion(ctx, envdVersions); envdVersion != "" { - if err := persistTemplateEnvdVersion(ctx, templateID, envdVersion); err != nil { - log.G(ctx).Warnf("persist template envd version fail, template=%s err=%v", templateID, err) - } + // Converge per-node envd versions into a single template value and reconcile + // the definition annotation once (idempotent; covers create and redo). + if err := reconcileTemplateEnvdVersion(ctx, templateID, convergeEnvdVersion(ctx, envdVersions)); err != nil { + log.G(ctx).Warnf("reconcile template envd version fail, template=%s err=%v", templateID, err) } return replicas, persistErr } type nodeEnvdVersion struct { NodeID string + Ready bool Version string } -// convergeEnvdVersion picks a single template-level envd version from the -// per-node collection results: the first valid semver wins, and any divergence -// across nodes is logged but not treated as an error. +// convergeEnvdVersion only returns a template-level envd version when every +// READY replica reported a valid and consistent value. Failed/non-ready replicas +// do not participate in this capability contract. func convergeEnvdVersion(ctx context.Context, versions []nodeEnvdVersion) string { chosen := "" + readyCount := 0 for _, item := range versions { + if !item.Ready { + continue + } + readyCount++ v := sanitizeEnvdVersion(item.Version) if v == "" { - continue + log.G(ctx).Warnf("envd version missing on ready node=%s; suppressing template capability annotation", item.NodeID) + return "" } if chosen == "" { chosen = v continue } if v != chosen { - log.G(ctx).Warnf("envd version mismatch across nodes: keeping=%s saw=%s node=%s", chosen, v, item.NodeID) + log.G(ctx).Warnf("envd version mismatch across ready nodes: keeping=%s saw=%s node=%s; suppressing template capability annotation", chosen, v, item.NodeID) + return "" } } + if readyCount == 0 { + return "" + } return chosen } -// persistTemplateEnvdVersion writes the converged envd version into the template -// definition's request_json annotation exactly once (read-modify-write), then -// invalidates the cached request so new sandboxes inherit the annotation. It is -// idempotent: a no-op when the annotation already holds the same value. -func persistTemplateEnvdVersion(ctx context.Context, templateID, version string) error { +// reconcileTemplateEnvdVersion aligns the stored template envd annotation with +// the latest convergence result. A non-empty version is persisted; an empty +// version clears any previously persisted annotation so redo cannot leave a +// stale capability signal behind. +func reconcileTemplateEnvdVersion(ctx context.Context, templateID, version string) error { version = sanitizeEnvdVersion(version) - if templateID == "" || version == "" { + if templateID == "" { return nil } // Serialize the request_json read-modify-write against concurrent template @@ -554,10 +564,17 @@ func persistTemplateEnvdVersion(ctx context.Context, templateID, version string) if req.Annotations == nil { req.Annotations = make(map[string]string) } - if req.Annotations[constants.CubeAnnotationComponentEnvdVersion] == version { + current := strings.TrimSpace(req.Annotations[constants.CubeAnnotationComponentEnvdVersion]) + if version == "" { + if current == "" { + return nil + } + delete(req.Annotations, constants.CubeAnnotationComponentEnvdVersion) + } else if current == version { return nil + } else { + req.Annotations[constants.CubeAnnotationComponentEnvdVersion] = version } - req.Annotations[constants.CubeAnnotationComponentEnvdVersion] = version payload, err := json.Marshal(req) if err != nil { return err @@ -572,6 +589,10 @@ func persistTemplateEnvdVersion(ctx context.Context, templateID, version string) }) } +func persistTemplateEnvdVersion(ctx context.Context, templateID, version string) error { + return reconcileTemplateEnvdVersion(ctx, templateID, version) +} + func createReplicaOnNode(ctx context.Context, target *node.Node, req *sandboxtypes.CreateCubeSandboxReq, opts replicaRunOptions) (ReplicaStatus, string) { replica := ReplicaStatus{ NodeID: target.ID(), diff --git a/CubeMaster/pkg/templatecenter/store_test.go b/CubeMaster/pkg/templatecenter/store_test.go index b8e2ddb3e..8160f14ae 100644 --- a/CubeMaster/pkg/templatecenter/store_test.go +++ b/CubeMaster/pkg/templatecenter/store_test.go @@ -62,14 +62,175 @@ func TestNormalizeStoredTemplateRequestStripsPhysicalAnnotations(t *testing.T) { assert.Empty(t, out.SnapshotDir) } -func TestConvergeEnvdVersionUsesNodeCollectionResults(t *testing.T) { - got := convergeEnvdVersion(context.Background(), []nodeEnvdVersion{ - {NodeID: "node-a", Version: ""}, - {NodeID: "node-b", Version: "envd version 0.5.11"}, - {NodeID: "node-c", Version: "0.6.0"}, +func TestConvergeEnvdVersionRequiresAllReadyReplicasToAgree(t *testing.T) { + tests := []struct { + name string + versions []nodeEnvdVersion + want string + }{ + { + name: "all ready replicas report same version", + versions: []nodeEnvdVersion{ + {NodeID: "node-a", Ready: true, Version: "envd version 0.5.11"}, + {NodeID: "node-b", Ready: true, Version: "0.5.11"}, + {NodeID: "node-c", Ready: false, Version: ""}, + }, + want: "0.5.11", + }, + { + name: "missing version on ready replica suppresses annotation", + versions: []nodeEnvdVersion{ + {NodeID: "node-a", Ready: true, Version: "0.5.11"}, + {NodeID: "node-b", Ready: true, Version: ""}, + }, + want: "", + }, + { + name: "version mismatch across ready replicas suppresses annotation", + versions: []nodeEnvdVersion{ + {NodeID: "node-a", Ready: true, Version: "0.5.11"}, + {NodeID: "node-b", Ready: true, Version: "0.6.0"}, + }, + want: "", + }, + { + name: "failed replicas do not participate", + versions: []nodeEnvdVersion{ + {NodeID: "node-a", Ready: false, Version: ""}, + {NodeID: "node-b", Ready: true, Version: "0.5.11"}, + }, + want: "0.5.11", + }, + { + name: "no ready replicas means no annotation", + versions: []nodeEnvdVersion{ + {NodeID: "node-a", Ready: false, Version: "0.5.11"}, + {NodeID: "node-b", Ready: false, Version: ""}, + }, + want: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := convergeEnvdVersion(context.Background(), tt.versions) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestCreateTemplateReplicasOnNodesClearsEnvdAnnotationWhenReadyReplicaHasNoEnvdVersion(t *testing.T) { + patches := gomonkey.NewPatches() + defer patches.Reset() + + req := &sandboxtypes.CreateCubeSandboxReq{InstanceType: "cubebox"} + targets := []*node.Node{{InsID: "node-a", IP: "10.0.0.1", Healthy: true}} + + patches.ApplyFunc(createReplicaOnNode, func(ctx context.Context, target *node.Node, req *sandboxtypes.CreateCubeSandboxReq, opts replicaRunOptions) (ReplicaStatus, string) { + return ReplicaStatus{ + NodeID: target.ID(), + NodeIP: target.IP, + InstanceType: req.InstanceType, + Status: ReplicaStatusReady, + Phase: ReplicaPhaseReady, + }, "" + }) + patches.ApplyFunc(UpsertReplica, func(ctx context.Context, templateID, instanceType string, replica ReplicaStatus) error { + return nil + }) + + reconcileCalled := false + reconciledVersion := "unexpected" + patches.ApplyFunc(reconcileTemplateEnvdVersion, func(ctx context.Context, templateID, version string) error { + reconcileCalled = true + reconciledVersion = version + return nil + }) + + replicas, err := createTemplateReplicasOnNodes(context.Background(), "tpl-no-envd-annotation", req, targets, replicaRunOptions{}) + require.NoError(t, err) + require.Len(t, replicas, 1) + assert.Equal(t, ReplicaStatusReady, replicas[0].Status) + assert.True(t, reconcileCalled, "template envd annotation should still be reconciled when convergence is suppressed") + assert.Equal(t, "", reconciledVersion, "missing envd version from a READY replica must clear any stale template annotation") +} + +func TestCreateTemplateReplicasOnNodesPersistsConvergedEnvdVersion(t *testing.T) { + patches := gomonkey.NewPatches() + defer patches.Reset() + + req := &sandboxtypes.CreateCubeSandboxReq{InstanceType: "cubebox"} + targets := []*node.Node{ + {InsID: "node-a", IP: "10.0.0.1", Healthy: true}, + {InsID: "node-b", IP: "10.0.0.2", Healthy: true}, + } + + patches.ApplyFunc(createReplicaOnNode, func(ctx context.Context, target *node.Node, req *sandboxtypes.CreateCubeSandboxReq, opts replicaRunOptions) (ReplicaStatus, string) { + return ReplicaStatus{ + NodeID: target.ID(), + NodeIP: target.IP, + InstanceType: req.InstanceType, + Status: ReplicaStatusReady, + Phase: ReplicaPhaseReady, + }, "0.5.11" + }) + patches.ApplyFunc(UpsertReplica, func(ctx context.Context, templateID, instanceType string, replica ReplicaStatus) error { + return nil + }) + + var ( + reconciledTemplateID string + reconciledVersion string + ) + patches.ApplyFunc(reconcileTemplateEnvdVersion, func(ctx context.Context, templateID, version string) error { + reconciledTemplateID = templateID + reconciledVersion = version + return nil + }) + + replicas, err := createTemplateReplicasOnNodes(context.Background(), "tpl-envd-annotation", req, targets, replicaRunOptions{}) + require.NoError(t, err) + require.Len(t, replicas, 2) + assert.Equal(t, "tpl-envd-annotation", reconciledTemplateID) + assert.Equal(t, "0.5.11", reconciledVersion) +} + +func TestReconcileTemplateEnvdVersionClearsStaleAnnotation(t *testing.T) { + patches := gomonkey.NewPatches() + defer patches.Reset() + + const templateID = "tpl-stale-envd" + def := &models.TemplateDefinition{ + TemplateID: templateID, + RequestJSON: `{ + "annotations": { + "` + constants.CubeAnnotationComponentEnvdVersion + `": "0.5.11", + "keep": "value" + } + }`, + } + + patches.ApplyFunc(withTemplateWriteLock, func(templateID string, fn func() error) error { + return fn() + }) + patches.ApplyFunc(GetDefinition, func(ctx context.Context, gotTemplateID string) (*models.TemplateDefinition, error) { + assert.Equal(t, templateID, gotTemplateID) + return def, nil + }) + + var updatedRequestJSON string + patches.ApplyFunc(updateDefinitionFields, func(ctx context.Context, gotTemplateID string, values map[string]any) error { + assert.Equal(t, templateID, gotTemplateID) + payload, ok := values["request_json"].(string) + require.True(t, ok) + updatedRequestJSON = payload + return nil }) - assert.Equal(t, "0.5.11", got) + require.NoError(t, reconcileTemplateEnvdVersion(context.Background(), templateID, "")) + assert.NotEmpty(t, updatedRequestJSON) + assert.NotContains(t, updatedRequestJSON, constants.CubeAnnotationComponentEnvdVersion) + assert.Contains(t, updatedRequestJSON, `"keep":"value"`) } func TestResolveTemplateNodesFiltersRequestedHealthyNodes(t *testing.T) { diff --git a/Cubelet/services/cubebox/appsnapshot.go b/Cubelet/services/cubebox/appsnapshot.go index 7af252283..f73bd887b 100644 --- a/Cubelet/services/cubebox/appsnapshot.go +++ b/Cubelet/services/cubebox/appsnapshot.go @@ -166,6 +166,7 @@ func (s *service) AppSnapshot(ctx context.Context, req *cubebox.AppSnapshotReque rsp.SandboxID = sandboxID stepLog = stepLog.WithFields(CubeLog.Fields{"sandboxID": sandboxID}) stepLog.Infof("Cubebox created successfully: %s", sandboxID) + envdVersion := s.collectReadyEnvdVersion(ctx, sandboxID) snapshotSuccess := false temporaryCubeboxDestroyed := false @@ -276,10 +277,6 @@ func (s *service) AppSnapshot(ctx context.Context, req *cubebox.AppSnapshotReque return rsp, nil } - // collectEnvdVersion uses containerd Exec, which must run before - // cube-runtime marks the guest as app-snapshotting and disables exec. - envdVersion := s.collectEnvdVersion(ctx, sandboxID) - stepLog.Info("Step 4: Executing cube-runtime snapshot...") // AppSnapshot builds a brand-new template from a fresh sandbox: there is // no base memory blob to overlay onto, so we always ask for a full memory diff --git a/Cubelet/services/cubebox/envd_version.go b/Cubelet/services/cubebox/envd_version.go index e08708447..ceb26422e 100644 --- a/Cubelet/services/cubebox/envd_version.go +++ b/Cubelet/services/cubebox/envd_version.go @@ -9,7 +9,10 @@ import ( "context" "errors" "fmt" + "io" + "net/http" "regexp" + "strings" "sync" "syscall" "time" @@ -19,7 +22,9 @@ import ( "github.com/google/uuid" containerd "github.com/containerd/containerd/v2/client" + cubeboxv1 "github.com/tencentcloud/CubeSandbox/Cubelet/api/services/cubebox/v1" "github.com/tencentcloud/CubeSandbox/Cubelet/pkg/log" + "github.com/tencentcloud/CubeSandbox/Cubelet/pkg/telnet" ) const ( @@ -27,6 +32,15 @@ const ( // envdVersionExecTimeout caps the in-guest `envd --version` probe so a hung // or unresponsive guest can never stall snapshot/commit. envdVersionExecTimeout = 5 * time.Second + envdDefaultPort = 49983 + // Template/snapshot creation is already a slow path, so after we have proven + // the envd binary exists we can afford a longer bounded wait for the service + // itself to come up and pass `/health`. + envdReadinessTotalTimeout = 10 * time.Second + envdReadinessAttemptTimeout = 500 * time.Millisecond + envdReadinessPeriod = 500 * time.Millisecond + envdReadinessMaxAttempts = 10 + envdReadinessPath = "/health" // envdVersionOutputLimit bounds the captured stdout/stderr to defend against // an image that floods the probe with output. envdVersionOutputLimit = 4 << 10 // 4 KiB @@ -129,6 +143,196 @@ func abortProbe(process containerd.Process, statusCh <-chan containerd.ExitStatu killProbeProcess(process, statusCh, logger) } +func buildEnvdReadinessProbe(ctx context.Context, sandboxIP string, port int) (*telnet.ProbeConfig, error) { + path := envdReadinessPath + req, err := NewRequestForHTTPGetAction(ctx, &cubeboxv1.HTTPGetAction{ + Path: &path, + Port: int32(port), + }, sandboxIP) + if err != nil { + return nil, err + } + return &telnet.ProbeConfig{ + Addr: sandboxIP, + Port: int32(port), + Timeout: envdReadinessTotalTimeout, + Period: envdReadinessPeriod, + SuccessThreshold: 1, + FailureThreshold: envdReadinessMaxAttempts, + ProbeTimeout: envdReadinessAttemptTimeout, + Action: telnet.ActionHTTPGet, + HttpGetRequest: req, + InstanceType: "cubebox", + }, nil +} + +func probeEnvdReadiness(ctx context.Context, sandboxIP string, port int) error { + return (&local{}).probeEnvdReadiness(ctx, sandboxIP, port) +} + +func getEnvdReadinessHTTPClient(l *local) *http.Client { + if l != nil && l.envdHTTPClient != nil { + return l.envdHTTPClient + } + return newEnvdReadinessHTTPClient() +} + +func newEnvdReadinessHTTPClient() *http.Client { + return &http.Client{ + CheckRedirect: func(req *http.Request, via []*http.Request) error { + return http.ErrUseLastResponse + }, + Transport: &http.Transport{ + DisableKeepAlives: true, + }, + } +} + +func (l *local) probeEnvdReadiness(ctx context.Context, sandboxIP string, port int) error { + sandboxIP = strings.TrimSpace(sandboxIP) + if sandboxIP == "" || sandboxIP == "" { + return fmt.Errorf("sandbox IP is empty") + } + if port <= 0 { + port = envdDefaultPort + } + cfg, err := buildEnvdReadinessProbe(ctx, sandboxIP, port) + if err != nil { + return err + } + return probeEnvdReadinessWithClient(ctx, cfg, getEnvdReadinessHTTPClient(l)) +} + +func probeEnvdReadinessWithClient(ctx context.Context, cfg *telnet.ProbeConfig, client *http.Client) error { + if cfg == nil || cfg.HttpGetRequest == nil { + return fmt.Errorf("envd readiness probe config is incomplete") + } + if client == nil { + client = newEnvdReadinessHTTPClient() + } + if cfg.Timeout <= 0 { + cfg.Timeout = envdReadinessTotalTimeout + } + if cfg.ProbeTimeout <= 0 { + cfg.ProbeTimeout = envdReadinessAttemptTimeout + } + if cfg.Period <= 0 { + cfg.Period = envdReadinessPeriod + } + if cfg.FailureThreshold <= 0 { + cfg.FailureThreshold = 1 + } + + probeCtx, cancel := context.WithTimeout(ctx, cfg.Timeout) + defer cancel() + + var lastErr error + for attempt := int32(0); attempt < cfg.FailureThreshold; attempt++ { + if err := probeCtx.Err(); err != nil { + if lastErr != nil { + return lastErr + } + return err + } + + attemptCtx, attemptCancel := context.WithTimeout(probeCtx, cfg.ProbeTimeout) + req := cfg.HttpGetRequest.Clone(attemptCtx) + req.Header = cfg.HttpGetRequest.Header.Clone() + req.Host = cfg.HttpGetRequest.Host + + resp, err := client.Do(req) + if err == nil { + _, _ = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode >= http.StatusOK && resp.StatusCode < http.StatusMultipleChoices { + attemptCancel() + return nil + } + lastErr = fmt.Errorf("statuscode:%d", resp.StatusCode) + } else { + lastErr = err + } + attemptCancel() + + if attempt+1 >= cfg.FailureThreshold { + break + } + + timer := time.NewTimer(cfg.Period) + select { + case <-timer.C: + case <-probeCtx.Done(): + if !timer.Stop() { + <-timer.C + } + if lastErr != nil { + return lastErr + } + return probeCtx.Err() + } + } + + if lastErr != nil { + return lastErr + } + return context.DeadlineExceeded +} + +// collectReadyEnvdVersion first proves the envd binary exists via +// `envd --version`; only then does it spend extra time waiting for the envd +// service to come up and pass `/health`. This keeps non-envd templates fast +// while still giving envd-backed templates a generous bounded readiness window. +func (s *service) collectReadyEnvdVersion(ctx context.Context, sandboxID string) string { + if s == nil || s.cubeboxMgr == nil { + return "" + } + return s.cubeboxMgr.collectReadyEnvdVersion(ctx, sandboxID) +} + +func finalizeReadyEnvdVersion( + ctx context.Context, + sandboxID string, + version string, + logger *log.CubeWrapperLogEntry, + lookupSandboxIP func(context.Context, string) (string, error), + probeReadiness func(context.Context, string, int) error, +) string { + if version == "" { + return "" + } + if logger == nil { + logger = log.G(ctx).WithField("sandboxID", sandboxID) + } + sandboxIP, err := lookupSandboxIP(ctx, sandboxID) + if err != nil { + logger.Warnf("collect envd version: get cubebox for readiness probe failed: %v", err) + return "" + } + if err := probeReadiness(ctx, sandboxIP, envdDefaultPort); err != nil { + logger.Warnf("collect envd version: envd readiness probe failed on %s:%d: %v", sandboxIP, envdDefaultPort, err) + return "" + } + logger.Infof("collect envd version: readiness confirmed, version=%s", version) + return version +} + +func (l *local) lookupSandboxIP(ctx context.Context, sandboxID string) (string, error) { + cb, err := l.cubeboxManger.Get(ctx, sandboxID) + if err != nil { + return "", err + } + if cb == nil { + return "", fmt.Errorf("cubebox %s not found", sandboxID) + } + return strings.TrimSpace(cb.IP), nil +} + +func (l *local) collectReadyEnvdVersion(ctx context.Context, sandboxID string) string { + logger := log.G(ctx).WithField("sandboxID", sandboxID) + version := l.collectEnvdVersion(ctx, sandboxID) + return finalizeReadyEnvdVersion(ctx, sandboxID, version, logger, l.lookupSandboxIP, l.probeEnvdReadiness) +} + // collectEnvdVersion runs `envd --version` inside the running guest of sandboxID // via containerd task.Exec and returns the parsed semantic version. // @@ -139,7 +343,7 @@ func abortProbe(process containerd.Process, statusCh <-chan containerd.ExitStatu // Security: the command always executes inside the microVM guest (task.Exec), // never on the host, so an untrusted custom-image binary stays confined to the // sandbox. -func (s *service) collectEnvdVersion(ctx context.Context, sandboxID string) (version string) { +func (l *local) collectEnvdVersion(ctx context.Context, sandboxID string) (version string) { logger := log.G(ctx).WithField("sandboxID", sandboxID) // Self-contained panic guard: this runs inside the AppSnapshot/CommitSandbox @@ -152,7 +356,7 @@ func (s *service) collectEnvdVersion(ctx context.Context, sandboxID string) (ver } }() - cb, err := s.cubeboxMgr.cubeboxManger.Get(ctx, sandboxID) + cb, err := l.cubeboxManger.Get(ctx, sandboxID) if err != nil { logger.Warnf("collect envd version: get cubebox failed: %v", err) return "" @@ -164,7 +368,7 @@ func (s *service) collectEnvdVersion(ctx context.Context, sandboxID string) (ver execCtx, cancel := context.WithTimeout(namespaces.WithNamespace(ctx, ns), envdVersionExecTimeout) defer cancel() - container, err := s.cubeboxMgr.client.LoadContainer(execCtx, sandboxID) + container, err := l.client.LoadContainer(execCtx, sandboxID) if err != nil { logger.Warnf("collect envd version: load container failed: %v", err) return "" @@ -242,6 +446,5 @@ func (s *service) collectEnvdVersion(ctx context.Context, sandboxID string) (ver logger.Warnf("collect envd version: no semver in output") return "" } - logger.Infof("collect envd version: %s", version) return version } diff --git a/Cubelet/services/cubebox/envd_version_test.go b/Cubelet/services/cubebox/envd_version_test.go index 78cfa9b56..1baf676c1 100644 --- a/Cubelet/services/cubebox/envd_version_test.go +++ b/Cubelet/services/cubebox/envd_version_test.go @@ -7,7 +7,13 @@ package cubebox import ( "context" "errors" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" "sync" + "sync/atomic" "syscall" "testing" "time" @@ -169,3 +175,232 @@ func TestKillProbeProcessNilStatusCh(t *testing.T) { killProbeProcess(process, nil, testProbeLogger()) assert.Equal(t, 1, process.killCount()) } + +func TestProbeEnvdReadinessSuccess(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, envdReadinessPath, r.URL.Path) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + host, port := testServerHostPort(t, srv) + err := probeEnvdReadiness(context.Background(), host, port) + require.NoError(t, err) +} + +func TestProbeEnvdReadinessRetriesUntilReady(t *testing.T) { + t.Parallel() + + var attempts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if attempts.Add(1) < 3 { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + host, port := testServerHostPort(t, srv) + err := probeEnvdReadiness(context.Background(), host, port) + require.NoError(t, err) + assert.EqualValues(t, 3, attempts.Load()) +} + +func TestProbeEnvdReadinessRejectsNonOK(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + host, port := testServerHostPort(t, srv) + err := probeEnvdReadiness(context.Background(), host, port) + require.Error(t, err) + assert.Contains(t, err.Error(), "503") +} + +func TestFinalizeReadyEnvdVersionSkipsReadinessWhenVersionMissing(t *testing.T) { + t.Parallel() + + lookupCalled := false + probeCalled := false + version := finalizeReadyEnvdVersion( + context.Background(), + "sandbox-1", + "", + testProbeLogger(), + func(context.Context, string) (string, error) { + lookupCalled = true + return "172.31.0.2", nil + }, + func(context.Context, string, int) error { + probeCalled = true + return nil + }, + ) + + assert.Empty(t, version) + assert.False(t, lookupCalled) + assert.False(t, probeCalled) +} + +func TestFinalizeReadyEnvdVersionReturnsEmptyWhenSandboxLookupFails(t *testing.T) { + t.Parallel() + + probeCalled := false + version := finalizeReadyEnvdVersion( + context.Background(), + "sandbox-lookup-fail", + "0.5.11", + testProbeLogger(), + func(context.Context, string) (string, error) { + return "", errors.New("lookup failed") + }, + func(context.Context, string, int) error { + probeCalled = true + return nil + }, + ) + + assert.Empty(t, version) + assert.False(t, probeCalled) +} + +func TestFinalizeReadyEnvdVersionReturnsEmptyWhenReadinessFails(t *testing.T) { + t.Parallel() + + version := finalizeReadyEnvdVersion( + context.Background(), + "sandbox-not-ready", + "0.5.11", + testProbeLogger(), + func(context.Context, string) (string, error) { + return "172.31.0.3", nil + }, + func(context.Context, string, int) error { + return errors.New("connection refused") + }, + ) + + assert.Empty(t, version) +} + +func TestFinalizeReadyEnvdVersionReturnsVersionWhenReadinessSucceeds(t *testing.T) { + t.Parallel() + + version := finalizeReadyEnvdVersion( + context.Background(), + "sandbox-ready", + "0.5.11", + testProbeLogger(), + func(context.Context, string) (string, error) { + return "172.31.0.4", nil + }, + func(context.Context, string, int) error { + return nil + }, + ) + + assert.Equal(t, "0.5.11", version) +} + +func TestProbeEnvdReadinessReturnsWhenContextCanceled(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-r.Context().Done() + })) + defer srv.Close() + + host, port := testServerHostPort(t, srv) + cfg, err := buildEnvdReadinessProbe(context.Background(), host, port) + require.NoError(t, err) + cfg.FailureThreshold = 10 + cfg.Period = 5 * time.Millisecond + cfg.ProbeTimeout = 100 * time.Millisecond + cfg.Timeout = time.Second + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + start := time.Now() + err = probeEnvdReadinessWithClient(ctx, cfg, newEnvdReadinessHTTPClient()) + elapsed := time.Since(start) + + require.Error(t, err) + assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Less(t, elapsed, 300*time.Millisecond) +} + +func TestProbeEnvdReadinessReturnsConnectionRefused(t *testing.T) { + t.Parallel() + + l, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + host, port := testListenerHostPort(t, l) + require.NoError(t, l.Close()) + + cfg, err := buildEnvdReadinessProbe(context.Background(), host, port) + require.NoError(t, err) + cfg.FailureThreshold = 1 + cfg.Period = 0 + cfg.ProbeTimeout = 100 * time.Millisecond + cfg.Timeout = 200 * time.Millisecond + + err = probeEnvdReadinessWithClient(context.Background(), cfg, newEnvdReadinessHTTPClient()) + require.Error(t, err) + assert.Contains(t, strings.ToLower(err.Error()), "refused") +} + +func TestProbeEnvdReadinessRejectsRedirect(t *testing.T) { + t.Parallel() + + var redirected atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + redirected.Add(1) + w.WriteHeader(http.StatusOK) + })) + defer target.Close() + + redirect := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target.URL, http.StatusFound) + })) + defer redirect.Close() + + host, port := testServerHostPort(t, redirect) + cfg, err := buildEnvdReadinessProbe(context.Background(), host, port) + require.NoError(t, err) + cfg.FailureThreshold = 1 + cfg.Period = 0 + cfg.ProbeTimeout = 100 * time.Millisecond + cfg.Timeout = 200 * time.Millisecond + + err = probeEnvdReadinessWithClient(context.Background(), cfg, newEnvdReadinessHTTPClient()) + require.Error(t, err) + assert.Contains(t, err.Error(), "302") + assert.Zero(t, redirected.Load()) +} + +func testServerHostPort(t *testing.T, srv *httptest.Server) (string, int) { + t.Helper() + + host, portText, err := net.SplitHostPort(srv.Listener.Addr().String()) + require.NoError(t, err) + port, err := strconv.Atoi(portText) + require.NoError(t, err) + return host, port +} + +func testListenerHostPort(t *testing.T, l net.Listener) (string, int) { + t.Helper() + + host, portText, err := net.SplitHostPort(l.Addr().String()) + require.NoError(t, err) + port, err := strconv.Atoi(portText) + require.NoError(t, err) + return host, port +} diff --git a/Cubelet/services/cubebox/service.go b/Cubelet/services/cubebox/service.go index 33b53e5a5..0751112aa 100644 --- a/Cubelet/services/cubebox/service.go +++ b/Cubelet/services/cubebox/service.go @@ -290,8 +290,6 @@ func (s *service) Create(ctx context.Context, req *cubebox.RunCubeSandboxRequest rsp.SandboxID = createInfo.SandboxID rsp.SandboxIP = getSandboxIp(createInfo) rsp.PortMappings = getAllocatedPort(createInfo) - - setCubeExtKey(rsp, createInfo) }() or, e := s.cubeboxMgr.getSandboxRuntime(req) @@ -353,11 +351,6 @@ func SetRunCubeSandboxRequestDefaultValue(req *cubebox.RunCubeSandboxRequest) { } } -func setCubeExtKey(rsp *cubebox.RunCubeSandboxResponse, createInfo *workflow.CreateContext) { - _ = rsp - _ = createInfo -} - func dealCreateInnerMetric(rsp *cubebox.RunCubeSandboxResponse, createInfo *workflow.CreateContext) { shimMetric := time.Duration(0) diff --git a/Cubelet/services/cubebox/template_ops.go b/Cubelet/services/cubebox/template_ops.go index 0a1188abd..3e0fcb456 100644 --- a/Cubelet/services/cubebox/template_ops.go +++ b/Cubelet/services/cubebox/template_ops.go @@ -181,6 +181,10 @@ func (s *service) CommitSandbox(ctx context.Context, req *cubebox.CommitSandboxR } _ = os.RemoveAll(tmpSnapshotPath) // NOCC:Path Traversal() + // Probe envd before taking the snapshot so any reported capability reflects + // the runtime state captured into this snapshot, not a later post-snapshot + // recovery of the service. + envdVersion := s.collectReadyEnvdVersion(ctx, rsp.SandboxID) // CommitSandbox snapshots a running sandbox whose memory artifact has // just been prepared above. snapshotTypeForCmd carries the right type // for the path we took: soft-dirty when reflink-cloning a base, or full @@ -245,9 +249,7 @@ func (s *service) CommitSandbox(ctx context.Context, req *cubebox.CommitSandboxR rsp.GuestImageVersion = versions.GuestImage rsp.AgentVersion = versions.Agent rsp.KernelVersion = versions.Kernel - // The source sandbox stays running through commit, so probe its real envd - // version in-guest after the memory snapshot. Best-effort: empty on failure. - rsp.EnvdVersion = s.collectEnvdVersion(ctx, rsp.SandboxID) + rsp.EnvdVersion = envdVersion if err := storage.WriteSnapshotCatalog(&storage.SnapshotCatalogEntry{ SnapshotID: rsp.TemplateID, InstanceType: "cubebox",