diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 17b9ec2c..98603529 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -230,6 +230,28 @@ func (mirrors) Claim(key string, r holder.MirrorRequest, revision uint64) (bool, return err == nil, err } func (mirrors) Delete(key string) error { return stdio.State(holder.MirrorsState).Delete(key) } +func (mirrors) List() (map[string]holder.MirrorRequest, error) { + keys, err := stdio.State(holder.MirrorsState).Keys() + if err != nil { + return nil, err + } + out := map[string]holder.MirrorRequest{} + for _, k := range keys { + e, err := stdio.State(holder.MirrorsState).Get(k) + if err != nil { + return nil, err + } + var r holder.MirrorRequest + if e != nil && json.Unmarshal(e.Value, &r) == nil { + out[k] = r + } + } + return out, nil +} +func (mirrors) Put(key string, r holder.MirrorRequest) error { + _, err := stdio.State(holder.MirrorsState).Put(key, r) + return err +} // followMirrors runs every copy `mirror` took on, on the state's watch: the asks kept when the bundle starts (a copy a // restart cut) and each new one, one at a time. Each is claimed first (Verbs.Take), so a second follower copies none @@ -447,6 +469,32 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder. used, _ := v.Objects.Used() return map[string]any{"transfers": all, "used": used, "cap": v.CapNow(), "dropped_idle": idle}, err }}, + stdio.Tool{Name: "registry_mirrors", Description: "The copies of outside images mirror took on and has not ended (the state mirrors): each by its key, its source, digest, who asked and the follower that claimed it (none: it waits). With again, asks one again by hand, its claim let go so a follower takes it on; with drop, drops one by hand, saying refused on the image seat's events so its waiter ends (each a hand act, why required).", + Input: obj(map[string]any{"again": s("a kept copy's key to ask again"), "drop": s("a kept copy's key to drop"), "why": s("why (a hand act)")}), + Run: func(args map[string]any) (any, error) { + by, why := stdio.CallerOf(args), str(args, "why") + switch { + case str(args, "again") != "": + if why == "" { + return nil, errors.New("asking a copy again is a hand act: say why") + } + r, err := v.MirrorAgain(str(args, "again")) + if err != nil { + return nil, err + } + logf("the copy %s asked again by hand by %s: %s", str(args, "again"), by, why) + return map[string]any{"asked_again": str(args, "again"), "copy": r}, nil + case str(args, "drop") != "": + r, err := v.MirrorDrop(str(args, "drop"), by, why) + if err != nil { + return nil, err + } + logf("the copy %s dropped by hand by %s: %s", str(args, "drop"), by, why) + return map[string]any{"dropped": str(args, "drop"), "copy": r}, nil + } + all, err := v.MirrorsNow() + return map[string]any{"mirrors": all}, err + }}, stdio.Tool{Name: "registry_holds", Description: "What is held, since when and for whom; with release, lets one go through the seat's release as the operator's session (a hand act).", Input: obj(map[string]any{"seat": s("the seat"), "release": s("the name to release"), "digest": s("its digest"), "why": s("why")}), Run: func(args map[string]any) (any, error) { diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 0b384a03..f3f11868 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -159,6 +159,14 @@ func (m memMirrors) Claim(k string, r MirrorRequest, rev uint64) (bool, error) { return true, nil } func (m memMirrors) Delete(k string) error { delete(m, k); return nil } +func (m memMirrors) List() (map[string]MirrorRequest, error) { + out := map[string]MirrorRequest{} + for k, r := range m { + out[k] = r + } + return out, nil +} +func (m memMirrors) Put(k string, r MirrorRequest) error { m[k] = r; return nil } // mirrorRev is a kept ask's revision in memMirrors: 1 unclaimed, 2 once claimed. func mirrorRev(r MirrorRequest) uint64 { @@ -646,6 +654,24 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { t.Fatalf("two followers on one ask said %+v (ask id %q), asks %v: want one copy, said once under the ask's id", b.said[copies:], ask.ID, asked) } b.said = b.said[:copies] + // registry_mirrors (seams review note 1): a kept copy is read, asked again with its claim let go, and dropped + // by hand, which says refused under the ask's id with who and why. + _, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d}) + asked[key] = MirrorRequest{Source: asked[key].Source, Digest: d, ID: asked[key].ID, Claim: "cut-follower"} + if now, err := b.v.MirrorsNow(); err != nil || now[key].Claim != "cut-follower" { + t.Fatalf("MirrorsNow = %v, %v; want the claimed ask", now, err) + } + if r, err := b.v.MirrorAgain(key); err != nil || r.Claim != "" || asked[key].Claim != "" { + t.Errorf("MirrorAgain = %+v, %v (kept %+v); want the claim let go", r, err, asked[key]) + } + if _, err := b.v.MirrorDrop(key, "person.jochen", ""); err == nil { + t.Error("a drop with no why is taken") + } + if _, err := b.v.MirrorDrop(key, "person.jochen", "upstream gone"); err != nil || len(asked) != 0 || + b.said[copies].event != EventRefused || b.said[copies].body.ID == "" || !strings.Contains(b.said[copies].body.Reason, "dropped by hand by person.jochen: upstream gone") { + t.Errorf("MirrorDrop = %v, said %+v, asks %v; want refused said under the ask's id and the ask gone", err, b.said[copies:], asked) + } + b.said = b.said[:copies] b.cat.mirrorErr = errors.New("the upstream said 404") _, _ = b.v.Mirror(ImageSeat, "novox.mesh-build-image", MirrorRequest{Source: "nats", Digest: d}) if err := b.v.Mirrored(asked[key]); err != nil || b.said[1].event != EventRefused || !strings.Contains(b.said[1].body.Reason, "404") || len(asked) != 0 { diff --git a/modules/distribution/internal/holder/verbs.go b/modules/distribution/internal/holder/verbs.go index 31af720b..095da21d 100644 --- a/modules/distribution/internal/holder/verbs.go +++ b/modules/distribution/internal/holder/verbs.go @@ -46,6 +46,10 @@ type Mirrors interface { // writer wrote the key first. Claim(key string, r MirrorRequest, revision uint64) (bool, error) Delete(key string) error + // List is every ask kept, by its key: what registry_mirrors reads. + List() (map[string]MirrorRequest, error) + // Put writes r under key whatever it held: an ask asked again by hand, its claim let go. + Put(key string, r MirrorRequest) error } // HoldState is distribution's own state holds (bus state, template state; ADR 0617 ยง2). @@ -353,6 +357,55 @@ func (v *Verbs) Take(key string, r MirrorRequest, revision uint64, current bool, return v.Mirrored(r) } +// MirrorsNow lists every copy `mirror` took on and has not ended, by its key, for registry_mirrors: what waits (no +// claim) and what a follower took on (its claim). +func (v *Verbs) MirrorsNow() (map[string]MirrorRequest, error) { return v.Mirrors.List() } + +// MirrorAgain asks a kept copy again by hand (registry_mirrors, a hand act): its claim is let go, so the state's watch +// hands it to a follower, which takes it on as a new ask. A copy whose follower was cut, or that a follower holds and +// does not end, goes on this way without a restart. +func (v *Verbs) MirrorAgain(key string) (MirrorRequest, error) { + all, err := v.Mirrors.List() + if err != nil { + return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err) + } + r, ok := all[key] + if !ok { + return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key) + } + r.Claim = "" + if err := v.Mirrors.Put(key, r); err != nil { + return MirrorRequest{}, fmt.Errorf("mirror: %s is not asked again: %w", key, err) + } + return r, nil +} + +// MirrorDrop drops a kept copy by hand (registry_mirrors, a hand act with who and why): it says refused on the image +// seat's events with that reason, under the ask's id, so a waiter on the copy ends, and the ask leaves the state. +func (v *Verbs) MirrorDrop(key, by, why string) (MirrorRequest, error) { + if strings.TrimSpace(why) == "" { + return MirrorRequest{}, errors.New("mirror: a drop is a hand act: say why") + } + all, err := v.Mirrors.List() + if err != nil { + return MirrorRequest{}, fmt.Errorf("mirror: the asks cannot be read: %w", err) + } + r, ok := all[key] + if !ok { + return MirrorRequest{}, fmt.Errorf("mirror: no copy is kept under %s", key) + } + src, err := ParseSource(r.Source) + if err != nil { + return MirrorRequest{}, err + } + e := Event{Name: src.Name(), Kind: KindUpstream, Digest: r.Digest, Holder: Self, By: by, ID: r.ID, + Reason: fmt.Sprintf("%s@%s was not copied: dropped by hand by %s: %s", r.Source, r.Digest, by, why)} + if err := v.Say(ImageSeat, EventRefused, e); err != nil { + return MirrorRequest{}, fmt.Errorf("mirror: the drop of %s cannot be said, so the ask stays: %w", key, err) + } + return r, v.Mirrors.Delete(key) +} + // HeldNow lists the holds, for registry_holds. func (v *Verbs) HeldNow() ([]Hold, error) { all, err := v.Holds.List() diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 8f7e20dc..bf8c4f4b 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -48,6 +48,7 @@ "registry_health", "registry_holds", "registry_incoming", + "registry_mirrors", "registry_old_name" ], "invokes": [