diff --git a/hack/mirror/main.go b/hack/mirror/main.go index 8b6d0e6..8104764 100644 --- a/hack/mirror/main.go +++ b/hack/mirror/main.go @@ -24,6 +24,12 @@ // manifest bytes, so the digest the operator records (sha256 of the served // bytes) is the sha256 of this stripped manifest; the prefetch Job verifies // against the same bytes, so resolve and pull stay consistent. +// GET /v2///manifests/sha256: -> the manifest whose SERVED +// (urls-stripped) bytes hash to , independent of which tag it was baked +// under. Ollaya 0.13.0 pulls a pinned revision by digest (spike 011); the +// prefetch Job on 0.13.0 pulls @sha256:, so the mirror +// must answer the digest route with the exact bytes the operator recorded. +// An unknown digest is a 404, as for an unknown tag. // GET /v2///blobs/sha256: -> the blob, with Range support // (http.ServeContent): `ollaya pull` downloads large blobs with parallel // byte-range requests and rejects a server that ignores Range ("corrupt data: @@ -35,7 +41,10 @@ package main import ( + "crypto/sha256" + "encoding/hex" "encoding/json" + "errors" "log" "net/http" "os" @@ -44,6 +53,13 @@ import ( "time" ) +// errBadManifest / errReMarshal are returned by strippedManifest for a file that +// is not a parseable schemaVersion-2 manifest, or that fails to re-marshal. +var ( + errBadManifest = errors.New("bad manifest") + errReMarshal = errors.New("re-marshal manifest") +) + func main() { store := envOr("MIRROR_STORE", "/store/models") addr := envOr("MIRROR_ADDR", ":8080") @@ -63,6 +79,15 @@ func main() { fault := faultStatus(faultEnv) faultReset := faultEnv == "reset" + // MIRROR_TAG_MOVED, when "1"/"true", makes every tag request (not a digest + // request) answer with a byte-different-but-valid manifest: the baked manifest + // with an extra top-level field. It simulates an upstream tag that was + // repackaged (a new revision) while the digest route still serves the original + // bytes. The e2e flips it at test time with `kubectl set env` + a rollout, the + // same mechanism as MIRROR_FAULT, so a store recovery after the flip must pull + // the recorded revision by digest, not whatever the tag now points at. + tagMoved := os.Getenv("MIRROR_TAG_MOVED") == "1" || os.Getenv("MIRROR_TAG_MOVED") == "true" + // The store keeps manifests under a single directory (the host // the model was pulled from at bake time). Discover it so the served repo path // is independent of that bake-time host. @@ -71,7 +96,13 @@ func main() { log.Fatalf("mirror: no manifests found under %s/manifests/*", store) } host := filepath.Base(hostDirs[0]) - log.Printf("mirror: store=%s manifestHost=%s addr=%s fault=%d reset=%v", store, host, addr, fault, faultReset) + // Index every baked manifest by the sha256 of the bytes we SERVE for it (urls + // stripped), so a digest request (Ollaya 0.13.0 pulls a pinned revision by + // digest) is answered with the exact bytes the operator recorded. Keyed by bare + // hex, value is the on-disk manifest file path. + digestIndex := buildDigestIndex(store) + log.Printf("mirror: store=%s manifestHost=%s addr=%s fault=%d reset=%v tagMoved=%v digests=%d", + store, host, addr, fault, faultReset, tagMoved, len(digestIndex)) mux := http.NewServeMux() mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { @@ -106,8 +137,10 @@ func main() { return } switch { + case strings.Contains(p, "/manifests/sha256:"): + serveManifestByDigest(w, r, digestIndex) case strings.Contains(p, "/manifests/"): - serveManifest(w, r, store, host) + serveManifest(w, r, store, host, tagMoved) case strings.Contains(p, "/blobs/"): serveBlob(w, r, store) case p == "/v2/" || p == "/v2": @@ -141,7 +174,7 @@ func faultStatus(v string) int { } } -func serveManifest(w http.ResponseWriter, r *http.Request, store, host string) { +func serveManifest(w http.ResponseWriter, r *http.Request, store, host string, tagMoved bool) { p := r.URL.Path idx := strings.Index(p, "/manifests/") repo := strings.TrimPrefix(p[:idx], "/v2/") @@ -152,12 +185,93 @@ func serveManifest(w http.ResponseWriter, r *http.Request, store, host string) { log.Printf("mirror: 404 manifest %s/%s (%v)", repo, tag, err) return } + out, err := strippedManifest(raw) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + // MIRROR_TAG_MOVED simulates an upstream repackage: the tag now serves a + // byte-different (still valid) manifest, so a tag pull would get the new + // revision. A digest pull is unaffected (serveManifestByDigest serves the + // original bytes). The marker is an added top-level field, which changes the + // bytes and therefore the sha256 the client would record for the tag. + if tagMoved { + var m map[string]any + if err := json.Unmarshal(out, &m); err == nil { + m["ollayaMirrorMoved"] = "true" + if moved, err := json.Marshal(m); err == nil { + out = moved + } + } + w.Header().Set("Content-Type", "application/vnd.docker.distribution.manifest.v2+json") + _, _ = w.Write(out) + log.Printf("mirror: 200 manifest %s/%s (%d bytes, urls stripped, TAG MOVED)", repo, tag, len(out)) + return + } + w.Header().Set("Content-Type", "application/vnd.docker.distribution.manifest.v2+json") + _, _ = w.Write(out) + log.Printf("mirror: 200 manifest %s/%s (%d bytes, urls stripped)", repo, tag, len(out)) +} + +// serveManifestByDigest answers GET .../manifests/sha256: with the baked +// manifest whose SERVED (urls-stripped) bytes hash to . Ollaya 0.13.0 pulls a +// pinned revision by digest (spike 011), so the digest route must return the exact +// bytes the operator recorded, regardless of where the tag now points. An unknown +// digest is a 404, matching an unknown tag and the real registry's behaviour. +func serveManifestByDigest(w http.ResponseWriter, r *http.Request, index map[string][]byte) { + p := r.URL.Path + i := strings.Index(p, "sha256:") + hexDigest := p[i+len("sha256:"):] + out, ok := index[hexDigest] + if !ok { + http.NotFound(w, r) + log.Printf("mirror: 404 manifest by digest sha256:%s (not baked)", hexDigest) + return + } + w.Header().Set("Content-Type", "application/vnd.docker.distribution.manifest.v2+json") + _, _ = w.Write(out) + log.Printf("mirror: 200 manifest by digest sha256:%s (%d bytes)", hexDigest, len(out)) +} + +// buildDigestIndex walks every baked manifest file and maps the sha256 of its +// SERVED bytes (urls stripped) to those bytes. The served bytes, not the on-disk +// bytes, are what the operator records as the digest, so the index must key on the +// same stripping serveManifest applies. +func buildDigestIndex(store string) map[string][]byte { + index := make(map[string][]byte) + root := filepath.Join(store, "manifests") + _ = filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil || info.IsDir() { + return nil + } + raw, err := os.ReadFile(path) // #nosec G304 - walking a fixed store root + if err != nil { + return nil + } + out, err := strippedManifest(raw) + if err != nil { + return nil // not a manifest (e.g. a stray file); skip + } + sum := sha256.Sum256(out) + index[hex.EncodeToString(sum[:])] = out + return nil + }) + return index +} + +// strippedManifest parses a baked manifest and removes every config/layer "urls" +// field so `ollaya pull` fetches blobs from this registry, not upstream. The +// returned bytes are what the mirror serves and what the operator records as the +// digest. It is the single source of the stripping both the tag and digest routes +// use, so their digests stay consistent. +func strippedManifest(raw []byte) ([]byte, error) { var m map[string]any if err := json.Unmarshal(raw, &m); err != nil { - http.Error(w, "bad manifest", http.StatusInternalServerError) - return + return nil, errBadManifest + } + if _, ok := m["schemaVersion"]; !ok { + return nil, errBadManifest } - // Strip "urls" so `ollaya pull` fetches blobs from this registry, not upstream. if c, ok := m["config"].(map[string]any); ok { delete(c, "urls") } @@ -170,12 +284,9 @@ func serveManifest(w http.ResponseWriter, r *http.Request, store, host string) { } out, err := json.Marshal(m) if err != nil { - http.Error(w, "re-marshal manifest", http.StatusInternalServerError) - return + return nil, errReMarshal } - w.Header().Set("Content-Type", "application/vnd.docker.distribution.manifest.v2+json") - _, _ = w.Write(out) - log.Printf("mirror: 200 manifest %s/%s (%d bytes, urls stripped)", repo, tag, len(out)) + return out, nil } func serveBlob(w http.ResponseWriter, r *http.Request, store string) { diff --git a/test/e2e/digest_restore_test.go b/test/e2e/digest_restore_test.go new file mode 100644 index 0000000..1d4afea --- /dev/null +++ b/test/e2e/digest_restore_test.go @@ -0,0 +1,292 @@ +/* +Copyright 2026 maks3201. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package e2e + +import ( + "fmt" + "os" + "os/exec" + "strings" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/maks3201/decision-model-operator/test/utils" +) + +// digestRestoreMirrorNS holds this spec's own registry mirror; digestRestoreNS holds +// the DecisionModel under test. The spec runs its own mirror (not the chaos/mirror +// one) so it can flip MIRROR_TAG_MOVED without racing another container's teardown. +const digestRestoreMirrorNS = "dmo-e2e-digest-restore-mirror" +const digestRestoreNS = "dmo-e2e-digest-restore" + +// digestRestoreMirrorHost is the in-cluster DNS name of this spec's mirror Service. +var digestRestoreMirrorHost = fmt.Sprintf("mirror.%s.svc", digestRestoreMirrorNS) + +// This spike-011 scenario proves the value of pulling by digest on Ollaya 0.13.0: +// after a model's upstream tag moves to a different revision, a lost store is +// restored to the exact PINNED revision the operator recorded, not whatever the tag +// now points at. The recovery prefetch pulls @sha256:, so a +// moved tag can neither substitute a new revision (no silent swap) nor fail the +// recovery (no UpstreamTagMoved). Nightly-only: it loads laya:en and drives a store +// recovery. Runs through this spec's in-cluster mirror so the tag can be moved while +// the digest route keeps serving the original bytes. +var _ = Describe("Digest restore: a moved tag does not change a recovered revision", + Label("nightly", "chaos"), Ordered, func() { + const dm = "digest-restore-router" + + BeforeAll(func() { + if testInstall == "helm" { + Skip("helm job runs lifecycle specs only") + } + if os.Getenv("E2E_MIRROR") == "" { + Skip("set E2E_MIRROR to build the mirror image and run this spec") + } + if !modelIsLaya() { + Skip("laya-specific: the registry mirror image serves laya:en") + } + + installAndDeploy() + deployDigestRestoreMirror() + + By("pointing the operator at this spec's mirror") + patchManagerArgs( + "--ollaya-registry=http://"+digestRestoreMirrorHost, + "--allowed-registries="+digestRestoreMirrorHost, + "--allow-insecure-registries", + ) + + _, _ = utils.Kubectl("create", "ns", digestRestoreNS) + By("bringing up a stable revision through the mirror") + applyYAML(fmt.Sprintf(`apiVersion: decisionmodel.io/v1alpha1 +kind: DecisionModel +metadata: {name: %s, namespace: %s} +spec: + engine: ollaya + model: laya:en + device: %s + replicas: 1 + resources: {requests: {cpu: 250m, memory: 1Gi}, limits: {memory: 4Gi}} +`, dm, digestRestoreNS, testDevice)) + Eventually(func() (string, error) { + return utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, "{.status.phase}") + }, 8*time.Minute, 5*time.Second).Should(Equal("Ready"), + "the stable did not reach Ready through the mirror") + }) + + AfterAll(func() { + _, _ = utils.Kubectl("delete", "ns", digestRestoreNS, "--ignore-not-found") + _, _ = utils.Kubectl("delete", "ns", digestRestoreMirrorNS, "--ignore-not-found") + scaleManager(1) + undeploy() + cluster := envOr("KIND_CLUSTER", "kind") + _, _ = utils.Run(exec.Command("docker", "exec", cluster+"-control-plane", + "crictl", "rmi", "docker.io/library/"+mirrorImage)) + }) + + AfterEach(func() { dumpDiag(digestRestoreNS, dm) }) + + It("restores the pinned revision after the tag moved, not the tag's new target", func() { + By("recording the pinned digest of the stable revision") + pinnedDigest, err := utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, + "{.status.stableRevision.digest}") + Expect(err).NotTo(HaveOccurred()) + Expect(pinnedDigest).To(MatchRegexp(`^[0-9a-f]{64}$`), + "stableRevision.digest is not a bare-hex sha256: %q", pinnedDigest) + stableHash, err := utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, + "{.status.stableRevision.hash}") + Expect(err).NotTo(HaveOccurred()) + Expect(stableHash).NotTo(BeEmpty()) + + // Deleting the store inside the post-promotion stabilization window would + // make the controller roll back to the prior revision instead of recovering + // the store. The first revision has no previous one, so previousRevision is + // empty from the start; poll to be explicit and robust if an earlier step + // ever promotes. + By("confirming no stabilization window is open (previousRevision empty)") + Eventually(func() (string, error) { + return utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, "{.status.previousRevision}") + }, stabilizationWait, 5*time.Second).Should(BeEmpty()) + + By("moving the mirror's laya:en tag to a different manifest (upstream repackage)") + // A tag pull from now on returns a byte-different revision; the digest route + // still serves the pinned bytes. The flag is read at startup, so a rollout + // is needed for it to take effect (same mechanism as MIRROR_FAULT). + setDigestRestoreTagMoved(true) + + By("deleting the stable store PVC and its serving Pod to force a store recovery") + pvc := dm + "-store-" + stableHash + _, _ = utils.Kubectl("delete", "pvc", pvc, "-n", digestRestoreNS, "--wait=false") + _, _ = utils.Kubectl("delete", "pods", + "-l", "decisionmodel.io/name="+dm+",decisionmodel.io/revision="+stableHash, + "-n", digestRestoreNS, "--grace-period=0", "--force") + + By("the DM returns Ready on the SAME stable revision (store recovery succeeded)") + Eventually(func() (string, error) { + return utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, "{.status.phase}") + }, 12*time.Minute, 10*time.Second).Should(Equal("Ready"), + "store recovery did not return the DM to Ready") + gotHash, err := utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, + "{.status.stableRevision.hash}") + Expect(err).NotTo(HaveOccurred()) + Expect(gotHash).To(Equal(stableHash), + "store recovery must restore the same stable revision, not a new one") + + By("the recovered stable still records the PINNED digest (unchanged by the moved tag)") + gotDigest, err := utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, + "{.status.stableRevision.digest}") + Expect(err).NotTo(HaveOccurred()) + Expect(gotDigest).To(Equal(pinnedDigest), + "the recovered revision's digest must equal the originally pinned digest") + + By("a recovered serving Pod has the model-ready gate True (api/ps shows the pinned digest+device)") + // The readiness gate is True only when the runtime's /api/ps reports the + // expected digest on the expected device; a True gate is the operator's own + // proof that /api/ps serves the pinned revision. + Eventually(func(g Gomega) { + out, err := utils.Kubectl("get", "pods", + "-l", servingPodSelector(dm), "-n", digestRestoreNS, + "-o", fmt.Sprintf( + "jsonpath={.items[*].status.conditions[?(@.type=='%s')].status}", + modelReadyGate)) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(out).To(ContainSubstring("True"), + "no recovered Pod reported %s=True (got %q)", modelReadyGate, out) + }, 3*time.Minute, 5*time.Second).Should(Succeed()) + + By("querying /api/ps on the recovered Pod directly: it reports the ORIGINAL digest") + Expect(apiPSDigest(dm)).To(Equal(pinnedDigest), + "/api/ps on the recovered Pod must report the originally pinned digest") + + By("no revision was marked failed and no moved-tag / prefetch failure was recorded") + failed, err := utils.KubectlJSONPath(digestRestoreNS, "decisionmodel", dm, "{.status.failedRevision}") + Expect(err).NotTo(HaveOccurred()) + Expect(failed).To(BeEmpty(), "store recovery by digest must not fail the revision") + Expect(digestRestoreEventCount("UpstreamTagMoved")).To(Equal(0), + "a digest pull must never report UpstreamTagMoved") + Expect(digestRestoreEventCount("StorePrefetchFailed")).To(Equal(0), + "the recovery prefetch by digest must not fail") + }) + }) + +// deployDigestRestoreMirror brings up this spec's own registry mirror (the same +// baked image as the chaos/mirror spec) in digestRestoreMirrorNS and waits for it to +// be Available. Kept separate from deployMirror so flipping MIRROR_TAG_MOVED here +// never races the chaos container's mirror lifecycle. +func deployDigestRestoreMirror() { + _, _ = utils.Kubectl("create", "ns", digestRestoreMirrorNS) + applyYAML(fmt.Sprintf(`apiVersion: apps/v1 +kind: Deployment +metadata: {name: mirror, namespace: %s} +spec: + replicas: 1 + selector: {matchLabels: {app: mirror}} + template: + metadata: {labels: {app: mirror}} + spec: + securityContext: + runAsNonRoot: true + runAsUser: 65532 + seccompProfile: {type: RuntimeDefault} + containers: + - name: mirror + image: %s + imagePullPolicy: Never + ports: [{containerPort: 8080}] + readinessProbe: + httpGet: {path: /healthz, port: 8080} + periodSeconds: 2 + securityContext: + allowPrivilegeEscalation: false + readOnlyRootFilesystem: true + runAsNonRoot: true + capabilities: {drop: [ALL]} +--- +apiVersion: v1 +kind: Service +metadata: {name: mirror, namespace: %s} +spec: + selector: {app: mirror} + ports: [{port: 80, targetPort: 8080}] +`, digestRestoreMirrorNS, mirrorImage, digestRestoreMirrorNS)) + _, err := utils.Kubectl("wait", "deployment/mirror", "-n", digestRestoreMirrorNS, + "--for=condition=Available", "--timeout=2m") + Expect(err).NotTo(HaveOccurred(), "digest-restore mirror did not become Available") +} + +// setDigestRestoreTagMoved flips the mirror's MIRROR_TAG_MOVED mode and waits for the +// rollout, so every subsequent tag pull returns a byte-different manifest while the +// digest route keeps serving the original bytes. +func setDigestRestoreTagMoved(on bool) { + val := "MIRROR_TAG_MOVED-" + if on { + val = "MIRROR_TAG_MOVED=1" + } + _, err := utils.Kubectl("set", "env", "deployment/mirror", "-n", digestRestoreMirrorNS, val) + Expect(err).NotTo(HaveOccurred()) + _, err = utils.Kubectl("rollout", "status", "deployment/mirror", + "-n", digestRestoreMirrorNS, "--timeout=2m") + Expect(err).NotTo(HaveOccurred(), "mirror did not roll out with MIRROR_TAG_MOVED") +} + +// apiPSDigest port-forwards to the DecisionModel's serving Service and returns the +// bare-hex digest the runtime reports for the loaded model at /api/ps. It is the +// direct, runtime-side confirmation that the recovered Pod serves the pinned +// revision (the readiness gate asserts the same thing operator-side). +func apiPSDigest(dm string) string { + stop := make(chan struct{}) + defer close(stop) + local := portForward(dm, digestRestoreNS, 11435, stop) + + var digest string + Eventually(func(g Gomega) { + out, err := utils.Run(exec.Command("curl", "-sS", "--max-time", "15", + fmt.Sprintf("http://%s/api/ps", local))) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(out).To(ContainSubstring(`"digest"`), "unexpected /api/ps body: %s", out) + digest = firstJSONStringField(out, "digest") + g.Expect(digest).To(MatchRegexp(`^[0-9a-f]{64}$`), "no bare-hex digest in /api/ps: %s", out) + }, 1*time.Minute, 3*time.Second).Should(Succeed()) + return digest +} + +// firstJSONStringField extracts the value of the first `"":""` pair +// from a JSON body without a full unmarshal (the /api/ps shape is fixed and small, +// and the only string digest in it is the model digest). Returns "" if absent. +func firstJSONStringField(body, field string) string { + key := `"` + field + `":"` + i := strings.Index(body, key) + if i < 0 { + return "" + } + rest := body[i+len(key):] + j := strings.IndexByte(rest, '"') + if j < 0 { + return "" + } + return rest[:j] +} + +// digestRestoreEventCount counts Events on the DecisionModel in this spec's +// namespace by reason. +func digestRestoreEventCount(reason string) int { + out, _ := utils.Kubectl("get", "events", "-n", digestRestoreNS, + "--field-selector", "involvedObject.name="+"digest-restore-router"+",reason="+reason, + "-o", "jsonpath={.items[*].reason}") + return len(strings.Fields(out)) +}