From 79946e8cca96a1dbf2b8cd944d763800ab88078b Mon Sep 17 00:00:00 2001 From: Jonathan Ogilvie Date: Tue, 3 Nov 2026 18:21:21 -0500 Subject: [PATCH] fix(simulate): give each run its own diff servers and reclaim dead runs Diff server containers were always named by package digest, and a run reused any container with that name. With the default Remove policy, two concurrent `resource simulate` runs on the same provider shared one container, and the first to exit removed it under the other, whose plans then failed with connection refused. Now only an Orphan container keeps the deterministic name crossplane- diff-server-, so runs that ask to keep a server still find and share it, and none removes it. The container sets its label domain, so concurrent Orphan runs that race to create it share the winner's rather than failing on the name conflict. Every other container gets a random 8-hex suffix, so no other run finds it by name. The Manager also takes part in crash reclamation through containerruntime.Run, as render's Session does: with WithLeaseDir it leases a generated session ID before starting its first container (a run of only Development providers touches no Docker and takes no lease), reclaims what dead runs left, labels its containers with the run ID, and releases the lease in Stop. resource simulate defers Stop before starting servers, so an interrupted startup stops what it started. The lease file is kept for a later run when a container could not be removed, including one created by a Start that then failed. A supplied --session-id is not leased, as in render. resource simulate passes containerruntime.DefaultLeaseDir(). WithDockerClient now takes a DockerClient, the container client plus RunClient, and the Manager's Run uses it too, so one caller-supplied client serves everything the Manager does on Docker. The Manager never closes it. Behaviour change (resource simulate is unreleased): a run without the Orphan annotation no longer cleans up a server an earlier run kept. The help now gives the docker rm filter for kept servers, describes reclamation, and the render help names simulate among the runs that reclaim. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Jonathan Ogilvie --- cmd/crossplane/render/op/help/render.md | 7 +- cmd/crossplane/render/xr/help/render.md | 7 +- cmd/crossplane/simulate/help/simulate.md | 25 +- cmd/crossplane/simulate/simulate.go | 26 +- design/one-pager-cli-simulate.md | 26 +- internal/diffserver/manager.go | 180 +++++++++++-- internal/diffserver/manager_test.go | 328 ++++++++++++++++++++++- 7 files changed, 534 insertions(+), 65 deletions(-) diff --git a/cmd/crossplane/render/op/help/render.md b/cmd/crossplane/render/op/help/render.md index 21ceb9e4..3a82aeb7 100644 --- a/cmd/crossplane/render/op/help/render.md +++ b/cmd/crossplane/render/op/help/render.md @@ -34,9 +34,10 @@ reference. ## Cleaning up after a crashed run If a run is killed before it cleans up (for example by `kill -9`), the next -`composition render` or `operation render` run cleans up what it left behind, -following each container's `runtime-docker-cleanup` policy. This doesn't apply -to runs started with `--session-id`, to runs on Windows, or to a different +`composition render` or `operation render` run, or `resource simulate` run +that starts a diff server container, cleans up what it left behind, following +each container's `runtime-docker-cleanup` policy. This doesn't apply to runs +started with `--session-id`, to runs on Windows, or to a different `DOCKER_HOST`. If a container named by `runtime-docker-name` already exists: diff --git a/cmd/crossplane/render/xr/help/render.md b/cmd/crossplane/render/xr/help/render.md index 322802ee..686dc940 100644 --- a/cmd/crossplane/render/xr/help/render.md +++ b/cmd/crossplane/render/xr/help/render.md @@ -37,9 +37,10 @@ reference. ## Cleaning up after a crashed run If a run is killed before it cleans up (for example by `kill -9`), the next -`composition render` or `operation render` run cleans up what it left behind, -following each container's `runtime-docker-cleanup` policy. This doesn't apply -to runs started with `--session-id`, to runs on Windows, or to a different +`composition render` or `operation render` run, or `resource simulate` run +that starts a diff server container, cleans up what it left behind, following +each container's `runtime-docker-cleanup` policy. This doesn't apply to runs +started with `--session-id`, to runs on Windows, or to a different `DOCKER_HOST`. If a container named by `runtime-docker-name` already exists: diff --git a/cmd/crossplane/simulate/help/simulate.md b/cmd/crossplane/simulate/help/simulate.md index de1b0e1d..6eda1f36 100644 --- a/cmd/crossplane/simulate/help/simulate.md +++ b/cmd/crossplane/simulate/help/simulate.md @@ -46,13 +46,28 @@ Nothing simulate does writes to the cluster or the cloud: it reads the cluster with your kubeconfig and RBAC, and the diff servers run locally with no credentials and no cluster access. +Each run starts its own diff servers, named +`crossplane-diff-server-DIGEST-SUFFIX`, and removes them when it exits. + Diff server startup takes seconds per provider image. In an edit-simulate-edit loop, annotate a provider manifest with -`simulate.crossplane.io/runtime-docker-cleanup: Orphan` (like render's -per-function cleanup annotation) to leave its container running so the next -run reuses it. The containers have names like `crossplane-diff-server-DIGEST` -and carry `simulate.crossplane.io/managed-by` and session labels; any later -run without the annotation cleans them up, or remove them with docker rm. +`simulate.crossplane.io/runtime-docker-cleanup: Orphan` to leave its +container, named `crossplane-diff-server-DIGEST`, running for every later run +with the annotation to share. No run removes it, including a later run without +the annotation, which starts its own server instead. Remove kept containers +with: + +```shell +docker rm -f $(docker ps -aq --filter label=simulate.crossplane.io/cleanup=Orphan) +``` + +If a run is killed before it cleans up (for example by `kill -9`), the next +`crossplane resource simulate` run that starts a diff server container, or +`crossplane composition render` or `crossplane operation render` run, removes +what it left, except kept containers. A run whose providers are all +Development starts no containers, so it doesn't. This doesn't apply to runs +started with `--session-id`, to runs on Windows, or to a different +`DOCKER_HOST`. The command exits with one of these statuses, so a CI pipeline can gate on the result: diff --git a/cmd/crossplane/simulate/simulate.go b/cmd/crossplane/simulate/simulate.go index ef582965..ef6cfc8c 100644 --- a/cmd/crossplane/simulate/simulate.go +++ b/cmd/crossplane/simulate/simulate.go @@ -38,6 +38,7 @@ import ( "github.com/crossplane/crossplane/apis/v2/pkg" + "github.com/crossplane/cli/v2/internal/containerruntime" "github.com/crossplane/cli/v2/internal/diffserver" "github.com/crossplane/cli/v2/internal/docker" "github.com/crossplane/cli/v2/internal/kube" @@ -57,14 +58,14 @@ var helpDetail string type Cmd struct { Filename string `help:"YAML files or directory with managed resources, or '-' for stdin (comma-separated)." predictor:"yaml_file_or_directory" required:"" short:"f"` - Providers string `help:"YAML file or directory with Provider manifests to run as diff servers, overriding discovery from the cluster. Supports per-provider runtime annotations." name:"providers" optional:"" predictor:"yaml_file_or_directory" type:"path"` - ImageConfigs string `help:"YAML file with ImageConfig manifests that rewrite provider image paths for local pulls. Defaults to the project's imageConfigs when a project file is present." name:"image-configs" optional:"" predictor:"yaml_file" type:"existingfile"` - ProjectFile string `help:"Path to a project definition file to read imageConfigs from. Defaults to ${project_file} when present." name:"project-file" optional:""` - SessionID string `help:"Session ID to apply as the simulate.crossplane.io/session label on diff server containers. Defaults to a random per-process ID." placeholder:"ID"` - Context string `default:"" help:"Kubernetes context." name:"context" predictor:"context" short:"c"` - CacheDir string `default:"~/.crossplane/cache" help:"Cache directory for downloaded provider packages." name:"cache-dir" predictor:"directory"` - Timeout time.Duration `default:"2m" help:"Per-resource plan timeout."` - Output string `default:"plan" enum:"plan,json,yaml" help:"Output format. One of: plan, json, yaml." name:"output" short:"o"` + Providers string `help:"YAML file or directory with Provider manifests to run as diff servers, overriding discovery from the cluster. Supports per-provider runtime annotations." name:"providers" optional:"" predictor:"yaml_file_or_directory" type:"path"` + ImageConfigs string `help:"YAML file with ImageConfig manifests that rewrite provider image paths for local pulls. Defaults to the project's imageConfigs when a project file is present." name:"image-configs" optional:"" predictor:"yaml_file" type:"existingfile"` + ProjectFile string `help:"Path to a project definition file to read imageConfigs from. Defaults to ${project_file} when present." name:"project-file" optional:""` + SessionID string `help:"Session ID to apply as the simulate.crossplane.io/session label on diff server containers. Defaults to a random per-run ID. A supplied ID isn't leased, so its containers aren't reclaimed after a crash." placeholder:"ID"` + Context string `default:"" help:"Kubernetes context." name:"context" predictor:"context" short:"c"` + CacheDir string `default:"~/.crossplane/cache" help:"Cache directory for downloaded provider packages." name:"cache-dir" predictor:"directory"` + Timeout time.Duration `default:"2m" help:"Per-resource plan timeout."` + Output string `default:"plan" enum:"plan,json,yaml" help:"Output format. One of: plan, json, yaml." name:"output" short:"o"` Impersonation kube.ImpersonationFlags `embed:""` } @@ -154,11 +155,16 @@ func (c *Cmd) Run(ctx context.Context, k *kong.Context, logger logging.Logger, s return errors.Wrap(err, "cannot create package client") } - mgr := diffserver.NewManager(pkgClient, logger, diffserver.WithSessionID(c.SessionID)) + mgr := diffserver.NewManager(pkgClient, logger, + diffserver.WithSessionID(c.SessionID), + diffserver.WithLeaseDir(containerruntime.DefaultLeaseDir()), + ) + // Stop before EnsureServers returns, so an interrupted startup still + // stops the servers it started. Stop is safe when none started. + defer isimulate.StopServers(ctx, k.Stderr, mgr) if err := isimulate.EnsureServers(ctx, sp, k.Stderr, mgr, inputs); err != nil { return err } - defer isimulate.StopServers(ctx, k.Stderr, mgr) plans := make([]*plan.ResourcePlan, 0, len(resources)) planner := isimulate.NewPlanner(kubeClient, rmapper, defaultNamespace, mgr, c.Timeout) diff --git a/design/one-pager-cli-simulate.md b/design/one-pager-cli-simulate.md index 97ad8aa1..80448f33 100644 --- a/design/one-pager-cli-simulate.md +++ b/design/one-pager-cli-simulate.md @@ -354,17 +354,21 @@ read. Startup is the one cost that hurts across runs: a diff server takes a few seconds to come up, and an edit-simulate-edit loop pays that on every run -if containers are torn down at exit. So containers get deterministic -names derived from the provider image digest, and a provider manifest -annotated `simulate.crossplane.io/runtime-docker-cleanup: Orphan`, the same -cleanup policy render's function containers take through their annotation, -leaves its container running when the command exits: the next run finds a -live server with a matching name and reuses it instead of starting a new -one. Because the name carries the digest, a changed image never matches a -stale server, and leftovers are easy to find and remove since they all -follow one naming scheme and carry render-style managed-by and session -labels. The default stays clean, start and tear down per run; Orphan is -for the loops where those seconds add up. +if containers are torn down at exit. So a provider manifest annotated +`simulate.crossplane.io/runtime-docker-cleanup: Orphan`, the same cleanup +policy render's function containers take through their annotation, leaves +its container running when the command exits, under a deterministic +`crossplane-diff-server-` name derived from the provider image +digest: the next run finds a live server with a matching name and reuses it +instead of starting a new one. Because the name carries the digest, a +changed image never matches a stale server. Only Orphan servers are shared +this way; every other server gets a per-run suffix on that name, so +concurrent runs never stop each other's servers, and a run that dies before +cleaning up has its servers reclaimed by the next run through the lease +mechanism render uses. All of them carry render-style managed-by and session +labels, so leftovers are easy to find and remove. The default stays clean, +start and tear down per run; Orphan is for the loops where those seconds +add up. ### From Response to Printed Diff diff --git a/internal/diffserver/manager.go b/internal/diffserver/manager.go index 6e143c36..4a9e960c 100644 --- a/internal/diffserver/manager.go +++ b/internal/diffserver/manager.go @@ -28,6 +28,7 @@ import ( "context" "fmt" "net/netip" + "slices" "strings" "sync" "time" @@ -64,6 +65,10 @@ const ( // orphaned leftovers are easy to find. containerPrefix = "crossplane-diff-server-" + // runSuffixLength is the length of the random suffix that makes a + // container with the Remove policy this run's alone. + runSuffixLength = 8 + readyPollInterval = 500 * time.Millisecond defaultReadyTimeout = 60 * time.Second ) @@ -77,7 +82,9 @@ const ( CleanupRemove = containerruntime.CleanupRemove // CleanupOrphan leaves the container running, so a later run finds it by - // its deterministic name and reuses it without paying startup. Render's + // its deterministic name and reuses it without paying startup. Only an + // Orphan container gets that name: a Remove one's name is unique to its + // run, so no other run uses it, and its run can remove it. Render's // function runtimes support a third policy, Stop; a stopped diff server // serves nothing and restarting costs nearly as much as starting, so // simulate omits it. @@ -116,7 +123,8 @@ const ( // LabelKeySession identifies the run that created a container. Orphaned // containers outlive their session by design; the label records which run - // created them. + // created them. A later run reclaims the containers of a run that died + // without cleaning up by this label, unless they are Orphan. LabelKeySession = labelDomain + "/session" // LabelKeyCleanup records the effective cleanup policy a container was @@ -182,25 +190,44 @@ type Input struct { // diff servers run outside Docker, and routes resources to them. type Manager struct { pkgs runtimexpkg.Client - cli containerruntime.Client + cli DockerClient log logging.Logger readyTimeout time.Duration publishAddr netip.Addr session string - - mu sync.Mutex - routes map[schema.GroupKind]*Server - groups map[string]*Server - servers []*Server + leaseDir string + + // run leases this run's containers, and reclaims what dead runs left. + run *containerruntime.Run + + mu sync.Mutex + // leaked is set when a container Start may have created was not + // removed, so Stop keeps the run's lease for a later run to reclaim it. + leaked bool + // containers maps each container this run starts to the image of the + // first input that asked for it, so inputs whose packages share a digest + // share one diff server. + containers map[containerKey]string + routes map[schema.GroupKind]*Server + groups map[string]*Server + servers []*Server } // Option configures a Manager. type Option func(*Manager) -// WithDockerClient replaces the Docker client the manager runs containers -// with, for testing. Nil (the default) means a real client built from the -// environment. -func WithDockerClient(cli containerruntime.Client) Option { +// DockerClient is the Docker client a Manager runs its containers and +// reclaims dead runs' containers with. +type DockerClient interface { + containerruntime.Client + containerruntime.RunClient +} + +// WithDockerClient makes the manager do everything it does on Docker through +// cli: run and clean up its containers, and reclaim what dead runs left +// behind. The manager never closes cli. Nil (the default) means clients +// built from the environment. +func WithDockerClient(cli DockerClient) Option { return func(m *Manager) { m.cli = cli } } @@ -211,28 +238,57 @@ func WithReadyTimeout(d time.Duration) Option { // WithSessionID sets the session label value for the containers the manager // creates, so callers orchestrating a larger run can scope cleanup to it. -// Empty means a random per-process ID, matching render's session labels. +// Empty means a random ID generated for this manager, which it leases (see +// WithLeaseDir). A supplied ID is never leased, since other processes may +// share it, so what a run with one leaves after a crash is not reclaimed. func WithSessionID(session string) Option { return func(m *Manager) { m.session = session } } +// WithLeaseDir makes the manager lease its run in dir, and reclaim what dead +// runs leased there left behind, before it starts its first container. The +// default, "", disables both. See containerruntime.Run. +func WithLeaseDir(dir string) Option { + return func(m *Manager) { m.leaseDir = dir } +} + // NewManager returns a Manager that fetches provider packages with pkgs and -// runs their images as local diff server containers. +// runs their images as local diff server containers. NewManager does no I/O. func NewManager(pkgs runtimexpkg.Client, log logging.Logger, opts ...Option) *Manager { m := &Manager{ pkgs: pkgs, log: log, readyTimeout: defaultReadyTimeout, publishAddr: netip.MustParseAddr("127.0.0.1"), + containers: map[containerKey]string{}, routes: map[schema.GroupKind]*Server{}, groups: map[string]*Server{}, } for _, opt := range opts { opt(m) } + + leased := m.session == "" + if leased { + m.session = containerruntime.NewSessionID() + } + var runOpts []containerruntime.RunOption + if m.cli != nil { + runOpts = append(runOpts, containerruntime.WithRunClient(func() (containerruntime.RunClient, error) { + return sharedRunClient{m.cli}, nil + })) + } + m.run = containerruntime.NewRun(m.leaseDir, m.session, leased, log, runOpts...) return m } +// sharedRunClient is the manager's Docker client as its Run uses it. It hides +// the client's Close, so the Run, which closes the clients it gets, leaves the +// caller's open. +type sharedRunClient struct { + containerruntime.RunClient +} + // Start inspects each provider image and runs a diff server container for // every package that declares the DiffServer capability, and connects to the // already running server of every Development provider, in parallel and best @@ -243,6 +299,21 @@ func NewManager(pkgs runtimexpkg.Client, log logging.Logger, opts ...Option) *Ma // reporting), with the package pull and the server start as separate steps // because the pull dominates a cold start. func (m *Manager) Start(ctx context.Context, inputs []Input, events async.EventChannel) ([]Skipped, error) { + // Only a run that starts containers touches Docker: a Development + // provider's server is the user's. + if slices.ContainsFunc(inputs, func(in Input) bool { return in.Runtime != RuntimeDevelopment }) { + // Idempotent: only the first call does anything. Best effort: a + // failure only risks leaking this run's containers if it dies. It + // holds mu, so Servers and Stop wait for it, up to the reap timeout, + // on purpose: reclaiming must finish before any container starts. + m.mu.Lock() + err := m.run.Acquire(ctx) + m.mu.Unlock() + if err != nil { + m.log.Info("Problem managing this run's diff server containers", "error", err) + } + } + var ( wg sync.WaitGroup mu sync.Mutex @@ -255,11 +326,12 @@ func (m *Manager) Start(ctx context.Context, inputs []Input, events async.EventC s, skip := m.startOne(ctx, in, events) mu.Lock() defer mu.Unlock() - if skip != nil { + switch { + case skip != nil: skipped = append(skipped, *skip) - return + case s != nil: + m.register(s) } - m.register(s) }(in) } wg.Wait() @@ -298,6 +370,8 @@ func (m *Manager) register(s *Server) { } } +// startOne returns neither a Server nor a Skipped for an input whose diff +// server another input of this run already starts. func (m *Manager) startOne(ctx context.Context, in Input, events async.EventChannel) (*Server, *Skipped) { if in.Runtime == RuntimeDevelopment { return m.connectDevelopment(ctx, in, events) @@ -347,13 +421,26 @@ func (m *Manager) startOne(ctx context.Context, in Input, events async.EventChan cleanup = CleanupRemove } + // Two inputs can name one package, e.g. by tag and by digest. Its + // server serves both: a second would only be started and torn down. + // Without a digest the packages cannot be told apart, so each starts. + if pkg.Digest != "" { + if first, ok := m.claim(containerKey{digest: pkg.Digest, cleanup: cleanup, publishAddr: publishAddr, target: in.Target}, image); !ok { + m.log.Debug("Sharing the diff server of another provider with the same package", "image", image, "shared-with", first) + return nil, nil + } + } + ctr := &containerruntime.Container{ Image: runImage, - Name: containerName(pkg.Digest), + Name: containerName(pkg.Digest, cleanup), // --debug precedes the subcommand: it is the provider's global flag. - Cmd: []string{"--debug", "internal", "diff-server", "--address", fmt.Sprintf(":%d", containerPort)}, - Port: containerPort, - Labels: containerruntime.ContainerLabels(labelDomain, m.session, cleanup), + Cmd: []string{"--debug", "internal", "diff-server", "--address", fmt.Sprintf(":%d", containerPort)}, + Port: containerPort, + Labels: containerruntime.ContainerLabels(labelDomain, m.run.ID(), cleanup), + // With a Domain, a name conflict with a concurrent run's container + // is judged by its labels, so concurrent Orphan runs share one. + Domain: labelDomain, BindAddress: publishAddr.String(), Target: in.Target, Log: m.log, @@ -369,6 +456,11 @@ func (m *Manager) startOne(ctx context.Context, in Input, events async.EventChan running, err := ctr.Start(ctx) if err != nil { events.SendEvent(start, async.EventStatusFailure) + // Start may have created the container before failing, and no run + // finds a Remove container by its name again. + if cleanup != CleanupOrphan { + m.markLeaked() + } return nil, &Skipped{Image: image, Reason: fmt.Sprintf("cannot start diff server container: %s", err)} } @@ -379,6 +471,7 @@ func (m *Manager) startOne(ctx context.Context, in Input, events async.EventChan if !running.Reused && cleanup != CleanupOrphan { if serr := running.Stop(ctx, CleanupRemove); serr != nil { m.log.Info("Cannot clean up failed diff server container", "container", running.ID, "err", serr) + m.markLeaked() } } return nil, &Skipped{Image: image, Reason: err.Error()} @@ -390,6 +483,34 @@ func (m *Manager) startOne(ctx context.Context, in Input, events async.EventChan return server, nil } +// containerKey is what makes two inputs' diff server containers the same. +type containerKey struct { + digest string + cleanup Cleanup + publishAddr netip.Addr + target string +} + +// claim records that image's input starts the container for k, and reports +// true, unless another input already did, in which case it returns that +// input's image and false. +func (m *Manager) claim(k containerKey, image string) (string, bool) { + m.mu.Lock() + defer m.mu.Unlock() + if first, ok := m.containers[k]; ok { + return first, false + } + m.containers[k] = image + return image, true +} + +// markLeaked records that this run may have left a container behind. +func (m *Manager) markLeaked() { + m.mu.Lock() + defer m.mu.Unlock() + m.leaked = true +} + // connectDevelopment connects to a diff server the user already runs, from // source or under a debugger. Nothing is fetched and nothing starts, so the // routes come from the API groups the provider declared. @@ -502,7 +623,9 @@ func (m *Manager) Servers() []*Server { // Stop closes connections and tears down containers according to each // server's cleanup policy: Remove tears the container down, Orphan leaves it // running so the next run finds it by name and skips startup. A development -// server is the user's to stop, so only its connection is closed. +// server is the user's to stop, so only its connection is closed. Stop then +// gives up the run's lease, keeping its file for a later run to reclaim what +// is left when a container could not be torn down. func (m *Manager) Stop(ctx context.Context) error { m.mu.Lock() defer m.mu.Unlock() @@ -519,13 +642,22 @@ func (m *Manager) Stop(ctx context.Context) error { errs = append(errs, errors.Wrapf(err, "cannot stop diff server container %s", s.ContainerID)) } } + // Only render creates networks, so there are none to sweep. + errs = append(errs, m.run.Release(ctx, len(errs) == 0 && !m.leaked, "")) return errors.Join(errs...) } -func containerName(digest string) string { +// containerName names the container for the package with digest. An Orphan +// container's name is the digest's alone, so later runs find and reuse it. +// Any other's has a random suffix, so no other run ever finds it by name and +// uses a container this run removes when it exits. +func containerName(digest string, cleanup Cleanup) string { d := strings.TrimPrefix(digest, "sha256:") if len(d) > 12 { d = d[:12] } - return containerPrefix + d + if cleanup == CleanupOrphan { + return containerPrefix + d + } + return containerPrefix + d + "-" + containerruntime.NewSessionID()[:runSuffixLength] } diff --git a/internal/diffserver/manager_test.go b/internal/diffserver/manager_test.go index c690f97f..38e4dc7e 100644 --- a/internal/diffserver/manager_test.go +++ b/internal/diffserver/manager_test.go @@ -22,6 +22,11 @@ import ( "io" "net" "net/netip" + "os" + "path/filepath" + "regexp" + "runtime" + "slices" "strings" "sync" "testing" @@ -37,10 +42,12 @@ import ( "google.golang.org/grpc/status" "k8s.io/apimachinery/pkg/runtime/schema" + "github.com/crossplane/crossplane-runtime/v2/pkg/errors" "github.com/crossplane/crossplane-runtime/v2/pkg/logging" runtimexpkg "github.com/crossplane/crossplane-runtime/v2/pkg/xpkg" "github.com/crossplane/crossplane-runtime/v2/pkg/xpkg/parser" + "github.com/crossplane/cli/v2/internal/containerruntime" "github.com/crossplane/cli/v2/internal/containerruntime/fake" ) @@ -163,28 +170,59 @@ func (f *fakePkgClient) ListVersions(_ context.Context, _ string, _ ...runtimexp type fakeDocker struct { mu sync.Mutex addr string // host:port every container's port binds to - existing map[string]string // name -> id of pre-existing containers + existing map[string]string // name -> id of containers, pre-existing or created known map[string]bool // ids inspect resolves port bindings for created []string // names passed to ContainerCreate stopped []string // ids passed to ContainerStop removed []string // ids passed to ContainerRemove + // labels holds each created container's labels, by name. + labels map[string]map[string]string + // listed are the containers ContainerList matches label filters against, + // as left behind by dead runs. + listed []container.Summary + // removeErr fails ContainerRemove; startErr fails ContainerStart. + removeErr, startErr error + // conflict names a container another run creates, with an Orphan + // policy, just before this run's ContainerCreate for that name. + conflict string } func newFakeDocker(addr string, existing map[string]string) *fakeDocker { - f := &fakeDocker{addr: addr, existing: existing, known: map[string]bool{}} + if existing == nil { + existing = map[string]string{} + } + f := &fakeDocker{addr: addr, existing: existing, known: map[string]bool{}, labels: map[string]map[string]string{}} for _, id := range existing { f.known[id] = true } return f } +// session returns the session label of the container created with a name +// starting with prefix. +func (f *fakeDocker) session(t *testing.T, prefix string) string { + t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + for name, l := range f.labels { + if strings.HasPrefix(name, prefix) { + return l[LabelKeySession] + } + } + t.Fatalf("no container created with a name starting with %q, got %v", prefix, f.created) + return "" +} + func (f *fakeDocker) client() *fake.Client { return &fake.Client{ MockContainerInspect: func(_ context.Context, ref string, _ client.ContainerInspectOptions) (client.ContainerInspectResult, error) { f.mu.Lock() defer f.mu.Unlock() if id, ok := f.existing[ref]; ok { - return client.ContainerInspectResult{Container: container.InspectResponse{ID: id}}, nil + return client.ContainerInspectResult{Container: container.InspectResponse{ + ID: id, + Config: &container.Config{Labels: f.labels[ref]}, + }}, nil } if f.known[ref] { host, port, err := net.SplitHostPort(f.addr) @@ -207,13 +245,21 @@ func (f *fakeDocker) client() *fake.Client { MockContainerCreate: func(_ context.Context, options client.ContainerCreateOptions) (client.ContainerCreateResult, error) { f.mu.Lock() defer f.mu.Unlock() + if options.Name == f.conflict { + f.existing[options.Name] = "winner-id" + f.known["winner-id"] = true + f.labels[options.Name] = containerruntime.ContainerLabels(labelDomain, "0000000000000001", CleanupOrphan) + return client.ContainerCreateResult{}, errdefs.ErrConflict + } id := "new-" + options.Name f.created = append(f.created, options.Name) + f.existing[options.Name] = id + f.labels[options.Name] = options.Config.Labels f.known[id] = true return client.ContainerCreateResult{ID: id}, nil }, MockContainerStart: func(_ context.Context, _ string, _ client.ContainerStartOptions) (client.ContainerStartResult, error) { - return client.ContainerStartResult{}, nil + return client.ContainerStartResult{}, f.startErr }, MockContainerStop: func(_ context.Context, id string, _ client.ContainerStopOptions) (client.ContainerStopResult, error) { f.mu.Lock() @@ -224,12 +270,41 @@ func (f *fakeDocker) client() *fake.Client { MockContainerRemove: func(_ context.Context, id string, _ client.ContainerRemoveOptions) (client.ContainerRemoveResult, error) { f.mu.Lock() defer f.mu.Unlock() + if f.removeErr != nil { + return client.ContainerRemoveResult{}, f.removeErr + } f.removed = append(f.removed, id) return client.ContainerRemoveResult{}, nil }, + MockContainerList: func(_ context.Context, options client.ContainerListOptions) (client.ContainerListResult, error) { + f.mu.Lock() + defer f.mu.Unlock() + var res client.ContainerListResult + for _, c := range f.listed { + if matchesLabels(options.Filters, c.Labels) { + res.Items = append(res.Items, c) + } + } + return res, nil + }, + MockNetworkList: func(context.Context, client.NetworkListOptions) (client.NetworkListResult, error) { + return client.NetworkListResult{}, nil + }, } } +// matchesLabels reports whether labels carry every label filter, as Docker +// matches them. +func matchesLabels(filters client.Filters, labels map[string]string) bool { + for v := range filters["label"] { + k, want, _ := strings.Cut(v, "=") + if labels[k] != want { + return false + } + } + return true +} + // readyServer serves a real PlanService on a loopback listener that answers // every request with INVALID_ARGUMENT, which is all readiness needs. type readyServer struct { @@ -303,9 +378,8 @@ func TestManagerStart(t *testing.T) { t.Errorf("Route(Bucket.s3): want no route for a kind the package does not ship") } - wantName := containerPrefix + "0123456789ab" - if len(fd.created) != 1 || fd.created[0] != wantName { - t.Errorf("Start(...): want container %q started, got %v", wantName, fd.created) + if len(fd.created) != 1 || !runName.MatchString(fd.created[0]) { + t.Errorf("Start(...): want one container named like %s started, got %v", runName, fd.created) } if err := m.Stop(ctx); err != nil { @@ -316,13 +390,13 @@ func TestManagerStart(t *testing.T) { } }) - t.Run("ReusesContainerByName", func(t *testing.T) { + t.Run("OrphanReusesContainerByName", func(t *testing.T) { fd := newFakeDocker(servePlanService(t), map[string]string{ containerPrefix + "0123456789ab": "existing-id", }) m := NewManager(pkgs, log, WithDockerClient(fd.client())) - skipped, err := m.Start(ctx, Images([]string{"xpkg.example.org/provider-aws-ec2:v2.9.0"}), nil) + skipped, err := m.Start(ctx, []Input{{Image: "xpkg.example.org/provider-aws-ec2:v2.9.0", Cleanup: CleanupOrphan}}, nil) if err != nil || len(skipped) != 0 { t.Fatalf("Start(...): unexpected error or skips: %v %+v", err, skipped) } @@ -331,6 +405,99 @@ func TestManagerStart(t *testing.T) { } }) + t.Run("RemoveNeverReusesContainerByName", func(t *testing.T) { + fd := newFakeDocker(servePlanService(t), map[string]string{ + containerPrefix + "0123456789ab": "kept-id", + }) + m := NewManager(pkgs, log, WithDockerClient(fd.client())) + + if _, err := m.Start(ctx, Images([]string{"xpkg.example.org/provider-aws-ec2:v2.9.0"}), nil); err != nil { + t.Fatalf("Start(...): unexpected error: %v", err) + } + if len(fd.created) != 1 || !runName.MatchString(fd.created[0]) { + t.Errorf("Start(...): want a container of this run's own, named like %s, got %v", runName, fd.created) + } + if err := m.Stop(ctx); err != nil { + t.Fatalf("Stop(...): unexpected error: %v", err) + } + if slices.Contains(fd.removed, "kept-id") { + t.Errorf("Stop(...): removed the kept container another run left, got removes %v", fd.removed) + } + }) + + t.Run("ConcurrentRemoveRunsDoNotShareAServer", func(t *testing.T) { + // Two runs on one Docker daemon serving the same provider: the + // first to exit must not remove the server the other still uses. + fd := newFakeDocker(servePlanService(t), nil) + img := Images([]string{"xpkg.example.org/provider-aws-ec2:v2.9.0"}) + m1 := NewManager(pkgs, log, WithDockerClient(fd.client())) + m2 := NewManager(pkgs, log, WithDockerClient(fd.client())) + + if _, err := m1.Start(ctx, img, nil); err != nil { + t.Fatalf("m1.Start(...): unexpected error: %v", err) + } + if _, err := m2.Start(ctx, img, nil); err != nil { + t.Fatalf("m2.Start(...): unexpected error: %v", err) + } + if len(fd.created) != 2 || fd.created[0] == fd.created[1] { + t.Fatalf("Start(...): want each run to create its own container, got %v", fd.created) + } + + if err := m1.Stop(ctx); err != nil { + t.Fatalf("m1.Stop(...): unexpected error: %v", err) + } + if want := []string{m1.Servers()[0].ContainerID}; !slices.Equal(fd.removed, want) { + t.Errorf("m1.Stop(...): want only m1's container removed %v, got %v", want, fd.removed) + } + }) + + t.Run("InputsOfOnePackageShareAServer", func(t *testing.T) { + // Two Provider manifests name one package, by tag and by digest. + const byTag, byDigest = "xpkg.example.org/provider-aws-ec2:v2.9.0", "xpkg.example.org/provider-aws-ec2@sha256:0123456789abcdef" + pkgs := &fakePkgClient{pkgs: map[string]*runtimexpkg.Package{ + byTag: pkgs.pkgs[byTag], + byDigest: pkgs.pkgs[byTag], + }} + fd := newFakeDocker(servePlanService(t), nil) + m := NewManager(pkgs, log, WithDockerClient(fd.client())) + + skipped, err := m.Start(ctx, Images([]string{byTag, byDigest}), nil) + if err != nil || len(skipped) != 0 { + t.Fatalf("Start(...): unexpected error or skips: %v %+v", err, skipped) + } + if _, ok := m.Route(schema.GroupKind{Group: "ec2.aws.upbound.io", Kind: "VPC"}); !ok { + t.Errorf("Route(VPC.ec2): want a route from the shared server") + } + if err := m.Stop(ctx); err != nil { + t.Fatalf("Stop(...): unexpected error: %v", err) + } + if len(fd.created) != 1 || !slices.Equal(fd.removed, []string{"new-" + fd.created[0]}) { + t.Errorf("Start and Stop: want one container started and removed, got created %v, removed %v", fd.created, fd.removed) + } + }) + + t.Run("ConcurrentOrphanRunsShareTheWinnersServer", func(t *testing.T) { + // Another run creates the Orphan container between this run's + // lookup and its create. + fd := newFakeDocker(servePlanService(t), nil) + fd.conflict = containerPrefix + "0123456789ab" + m := NewManager(pkgs, log, WithDockerClient(fd.client())) + + skipped, err := m.Start(ctx, []Input{{Image: "xpkg.example.org/provider-aws-ec2:v2.9.0", Cleanup: CleanupOrphan}}, nil) + if err != nil || len(skipped) != 0 { + t.Fatalf("Start(...): unexpected error or skips: %v %+v", err, skipped) + } + if got := m.Servers()[0].ContainerID; got != "winner-id" { + t.Errorf("Start(...): want the winner's container shared, got %q", got) + } + if err := m.Stop(ctx); err != nil { + t.Fatalf("Stop(...): unexpected error: %v", err) + } + if len(fd.stopped) != 0 || len(fd.removed) != 0 { + t.Errorf("Stop(...): want the shared Orphan container left running, got stops %v removes %v", fd.stopped, fd.removed) + } + }) + t.Run("OrphanCleanupLeavesContainersRunning", func(t *testing.T) { fd := newFakeDocker(servePlanService(t), nil) m := NewManager(pkgs, log, WithDockerClient(fd.client())) @@ -480,3 +647,146 @@ func TestManagerStart(t *testing.T) { } }) } + +// runName matches the name of a diff server container that is one run's +// alone: the package digest, then a random suffix. +var runName = regexp.MustCompile(`^` + containerPrefix + `0123456789ab-[0-9a-f]{8}$`) + +func TestManagerLease(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("run leases are not supported on Windows") + } + + ctx := context.Background() + log := logging.NewNopLogger() + const ( + img = "xpkg.example.org/provider-aws-ec2:v2.9.0" + deadID = "dead00000000000a" + liveID = "1111111111111111" + ) + pkgs := &fakePkgClient{pkgs: map[string]*runtimexpkg.Package{ + img: parsePackage(t, pkgWithCapability, "sha256:0123456789abcdef"), + }} + leases := func(t *testing.T, dir string) []string { + t.Helper() + got, err := filepath.Glob(filepath.Join(dir, "*.lock")) + if err != nil { + t.Fatal(err) + } + for i := range got { + got[i] = strings.TrimSuffix(filepath.Base(got[i]), ".lock") + } + return got + } + + t.Run("LeasesTheRunUntilACleanStop", func(t *testing.T) { + dir := t.TempDir() + fd := newFakeDocker(servePlanService(t), nil) + m := NewManager(pkgs, log, WithDockerClient(fd.client()), WithLeaseDir(dir)) + + if _, err := m.Start(ctx, Images([]string{img}), nil); err != nil { + t.Fatalf("Start(...): unexpected error: %v", err) + } + if got, want := leases(t, dir), []string{fd.session(t, containerPrefix)}; !slices.Equal(got, want) { + t.Errorf("Start(...): want a lease for the run its containers are labelled with %v, got %v", want, got) + } + if err := m.Stop(ctx); err != nil { + t.Fatalf("Stop(...): unexpected error: %v", err) + } + if got := leases(t, dir); len(got) != 0 { + t.Errorf("Stop(...): want the lease removed after a clean stop, got %v", got) + } + }) + + t.Run("KeepsTheLeaseWhenAContainerIsLeft", func(t *testing.T) { + cases := map[string]struct { + removeErr, startErr error + wantSkips int + wantStopErr bool + }{ + // Stop cannot remove the run's container. + "RemoveFails": {removeErr: errors.New("daemon busy"), wantStopErr: true}, + // The container was created, then failed to start: nothing in + // this run knows its ID to remove it. + "StartFails": {startErr: errors.New("port is already allocated"), wantSkips: 1}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + dir := t.TempDir() + fd := newFakeDocker(servePlanService(t), nil) + fd.removeErr, fd.startErr = tc.removeErr, tc.startErr + m := NewManager(pkgs, log, WithDockerClient(fd.client()), WithLeaseDir(dir)) + + if skipped, err := m.Start(ctx, Images([]string{img}), nil); err != nil || len(skipped) != tc.wantSkips { + t.Fatalf("Start(...): want %d skips and no error, got %+v, %v", tc.wantSkips, skipped, err) + } + if err := m.Stop(ctx); (err != nil) != tc.wantStopErr { + t.Fatalf("Stop(...): want error %t, got %v", tc.wantStopErr, err) + } + if got := leases(t, dir); len(got) != 1 { + t.Errorf("Stop(...): want the lease kept for a later run to reclaim, got %v", got) + } + }) + } + }) + + t.Run("ReclaimsDeadRunsBeforeStarting", func(t *testing.T) { + dir := t.TempDir() + // A lease no process holds: its run died. + if err := os.WriteFile(filepath.Join(dir, deadID+".lock"), nil, 0o600); err != nil { + t.Fatal(err) + } + fd := newFakeDocker(servePlanService(t), nil) + fd.listed = []container.Summary{ + {ID: "dead-server", Labels: containerruntime.ContainerLabels(labelDomain, deadID, CleanupRemove)}, + // A run with no dead lease here, such as a live one: left alone. + {ID: "live-server", Labels: containerruntime.ContainerLabels(labelDomain, liveID, CleanupRemove)}, + {ID: "kept-server", Labels: containerruntime.ContainerLabels(labelDomain, deadID, CleanupOrphan)}, + } + m := NewManager(pkgs, log, WithDockerClient(fd.client()), WithLeaseDir(dir)) + + if _, err := m.Start(ctx, Images([]string{img}), nil); err != nil { + t.Fatalf("Start(...): unexpected error: %v", err) + } + if want := []string{"dead-server"}; !slices.Equal(fd.removed, want) { + t.Errorf("Start(...): want only the dead run's Remove container removed %v, got %v", want, fd.removed) + } + if slices.Contains(leases(t, dir), deadID) { + t.Errorf("Start(...): want the dead run's lease removed once reclaimed") + } + }) + + t.Run("SuppliedIDIsNotLeased", func(t *testing.T) { + dir := t.TempDir() + fd := newFakeDocker(servePlanService(t), nil) + m := NewManager(pkgs, log, WithDockerClient(fd.client()), WithLeaseDir(dir), WithSessionID("0123456789abcdef")) + + if _, err := m.Start(ctx, Images([]string{img}), nil); err != nil { + t.Fatalf("Start(...): unexpected error: %v", err) + } + if got := leases(t, dir); len(got) != 0 { + t.Errorf("Start(...): want no lease for a supplied session ID, got %v", got) + } + if got := fd.session(t, containerPrefix); got != "0123456789abcdef" { + t.Errorf("Start(...): want the supplied session label, got %q", got) + } + }) + + t.Run("DevelopmentOnlyRunTakesNoLease", func(t *testing.T) { + dir := t.TempDir() + m := NewManager(&fakePkgClient{}, log, WithLeaseDir(dir)) + + _, err := m.Start(ctx, []Input{{ + Runtime: RuntimeDevelopment, + Name: "provider-aws-ec2", + DevelopmentTarget: servePlanService(t), + APIGroups: []string{"ec2.aws.upbound.io"}, + }}, nil) + if err != nil { + t.Fatalf("Start(...): unexpected error: %v", err) + } + if got := leases(t, dir); len(got) != 0 { + t.Errorf("Start(...): want no lease for a run that starts no containers, got %v", got) + } + }) +}