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