diff --git a/gardener/gardener.go b/gardener/gardener.go index 0592d4372..f72f7b934 100644 --- a/gardener/gardener.go +++ b/gardener/gardener.go @@ -50,6 +50,7 @@ type SysInfoProvider interface { type Containerizer interface { Create(log lager.Logger, desiredContainerSpec spec.DesiredContainerSpec) error + Start(log lager.Logger, handle string) error Handles() ([]string, error) StreamIn(log lager.Logger, handle string, streamInSpec garden.StreamInSpec) error @@ -320,6 +321,10 @@ func (g *Gardener) Create(containerSpec garden.ContainerSpec) (ctr garden.Contai return nil, err } + if err = g.Containerizer.Start(log, containerSpec.Handle); err != nil { + return nil, err + } + if containerSpec.GraceTime != 0 { if err := container.SetGraceTime(containerSpec.GraceTime); err != nil { return nil, err diff --git a/gardener/gardenerfakes/fake_containerizer.go b/gardener/gardenerfakes/fake_containerizer.go index 085ebb29c..fee0b8edf 100644 --- a/gardener/gardenerfakes/fake_containerizer.go +++ b/gardener/gardenerfakes/fake_containerizer.go @@ -130,6 +130,18 @@ type FakeContainerizer struct { shutdownReturnsOnCall map[int]struct { result1 error } + StartStub func(lager.Logger, string) error + startMutex sync.RWMutex + startArgsForCall []struct { + arg1 lager.Logger + arg2 string + } + startReturns struct { + result1 error + } + startReturnsOnCall map[int]struct { + result1 error + } StopStub func(lager.Logger, string, bool) error stopMutex sync.RWMutex stopArgsForCall []struct { @@ -745,6 +757,68 @@ func (fake *FakeContainerizer) ShutdownReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeContainerizer) Start(arg1 lager.Logger, arg2 string) error { + fake.startMutex.Lock() + ret, specificReturn := fake.startReturnsOnCall[len(fake.startArgsForCall)] + fake.startArgsForCall = append(fake.startArgsForCall, struct { + arg1 lager.Logger + arg2 string + }{arg1, arg2}) + stub := fake.StartStub + fakeReturns := fake.startReturns + fake.recordInvocation("Start", []interface{}{arg1, arg2}) + fake.startMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeContainerizer) StartCallCount() int { + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() + return len(fake.startArgsForCall) +} + +func (fake *FakeContainerizer) StartCalls(stub func(lager.Logger, string) error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = stub +} + +func (fake *FakeContainerizer) StartArgsForCall(i int) (lager.Logger, string) { + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() + argsForCall := fake.startArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeContainerizer) StartReturns(result1 error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = nil + fake.startReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeContainerizer) StartReturnsOnCall(i int, result1 error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = nil + if fake.startReturnsOnCall == nil { + fake.startReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.startReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeContainerizer) Stop(arg1 lager.Logger, arg2 string, arg3 bool) error { fake.stopMutex.Lock() ret, specificReturn := fake.stopReturnsOnCall[len(fake.stopArgsForCall)] @@ -1019,6 +1093,8 @@ func (fake *FakeContainerizer) Invocations() map[string][][]interface{} { defer fake.runMutex.RUnlock() fake.shutdownMutex.RLock() defer fake.shutdownMutex.RUnlock() + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() fake.stopMutex.RLock() defer fake.stopMutex.RUnlock() fake.streamInMutex.RLock() diff --git a/guardiancmd/command.go b/guardiancmd/command.go index 78e930ca6..6177eb458 100644 --- a/guardiancmd/command.go +++ b/guardiancmd/command.go @@ -191,6 +191,7 @@ type CommonCommand struct { Containerd struct { Socket string `long:"containerd-socket" description:"Path to a containerd socket."` UseContainerdForProcesses bool `long:"use-containerd-for-processes" description:"Use containerd to run processes in containers."` + RuntimeType string `long:"containerd-runtime-type" description:"Containerd runtime type (e.g. io.containerd.runc.v2 or io.containerd.runsc.v1)."` } `group:"Containerd"` CPUThrottling struct { @@ -666,15 +667,16 @@ func (cmd *CommonCommand) wireContainerizer( peaCleaner := cmd.wirePeaCleaner(factory, volumizer, ociRuntime, peaPidGetter) peaCreator = &peas.PeaCreator{ - Volumizer: volumizer, - PidGetter: containersPidGetter, - PrivilegedGetter: privilegeChecker, - NetworkDepot: networkDepot, - BundleGenerator: template, - ProcessBuilder: processBuilder, - BundleSaver: bundleSaver, - ExecRunner: peasExecRunner, - PeaCleaner: peaCleaner, + Volumizer: volumizer, + PidGetter: containersPidGetter, + PrivilegedGetter: privilegeChecker, + NetworkDepot: networkDepot, + BundleGenerator: template, + ProcessBuilder: processBuilder, + BundleSaver: bundleSaver, + ExecRunner: peasExecRunner, + PeaCleaner: peaCleaner, + SkipUserNamespace: cmd.Containerd.RuntimeType != "" && cmd.Containerd.RuntimeType != "io.containerd.runc.v2", } peaUsernameResolver := &peas.PeaUsernameResolver{ @@ -692,7 +694,15 @@ func (cmd *CommonCommand) wireContainerizer( return nil, nil, err } - return rundmc.New(depot, template, ociRuntime, nstar, processesStopper, eventStore, stateStore, peaCreator, peaUsernameResolver, cpuEntitlementPerShare, runtimeStopper, cpuCgrouper), peaCleaner, nil + var streamer rundmc.Streamer + if cmd.Containerd.RuntimeType != "" && cmd.Containerd.RuntimeType != "io.containerd.runc.v2" { + streamer = rundmc.NewExecStreamer(ociRuntime) + } else { + streamer = rundmc.NewNstarStreamer(ociRuntime, nstar) + } + + execPeas := cmd.Containerd.RuntimeType != "" && cmd.Containerd.RuntimeType != "io.containerd.runc.v2" + return rundmc.New(depot, template, ociRuntime, streamer, processesStopper, eventStore, stateStore, peaCreator, peaUsernameResolver, cpuEntitlementPerShare, runtimeStopper, cpuCgrouper, execPeas), peaCleaner, nil } func (cmd *CommonCommand) useContainerd() bool { diff --git a/guardiancmd/command_linux.go b/guardiancmd/command_linux.go index edfa49b72..55a21942f 100644 --- a/guardiancmd/command_linux.go +++ b/guardiancmd/command_linux.go @@ -140,18 +140,28 @@ func (f *LinuxFactory) WireResolvConfigurer() kawasaki.DnsResolvConfigurer { } func (f *LinuxFactory) WireContainerd(processBuilder *processes.ProcBuilder, userLookupper users.UserLookupper, wireExecer func(pidGetter runrunc.PidGetter) *runrunc.Execer, statser runcontainerd.Statser, log lager.Logger, volumizer peas.Volumizer, peaHandlesGetter runcontainerd.PeaHandlesGetter, metricsProvider *metrics.MetricsProvider) (*runcontainerd.RunContainerd, *runcontainerd.RunContainerPea, *runcontainerd.PidGetter, *privchecker.PrivilegeChecker, peas.BundleLoader, error) { - containerdClient, err := client.New(f.config.Containerd.Socket, client.WithDefaultRuntime(plugins.RuntimeRuncV2)) + runtimeType := f.config.Containerd.RuntimeType + if runtimeType == "" { + runtimeType = plugins.RuntimeRuncV2 + } + containerdClient, err := client.New(f.config.Containerd.Socket, client.WithDefaultRuntime(runtimeType)) if err != nil { return nil, nil, nil, nil, nil, err } ctx := namespaces.WithNamespace(context.Background(), containerdNamespace) ctx = leases.WithLease(ctx, "lease-is-off") - nerd := nerdpkg.New(containerdClient, ctx, filepath.Join(containerdRuncRoot(), "fifo"), metricsProvider) + nerd := nerdpkg.New(containerdClient, ctx, filepath.Join(containerdRuncRoot(), "fifo"), metricsProvider, runtimeType) nerdStopper := nerdpkg.NewNerdStopper(containerdClient) pidGetter := &runcontainerd.PidGetter{Nerd: nerd} cgroupManager := runcontainerd.NewCgroupManager(containerdRuncRoot(), containerdNamespace) + // For non-runc runtimes (e.g. gVisor/runsc), use containerd task API for metrics + // instead of shelling out to runc which has no knowledge of the container. + if runtimeType != plugins.RuntimeRuncV2 { + statser = runcontainerd.NewContainerdStatser(nerd) + } + containerdManager := runcontainerd.New(nerd, nerd, processBuilder, userLookupper, wireExecer(pidGetter), statser, f.config.Containerd.UseContainerdForProcesses, cgroupManager, f.WireMkdirer(), peaHandlesGetter, f.config.Containers.CleanupProcessDirsOnWait, nerdStopper) peaRunner := runcontainerd.NewRunContainerPea(containerdManager, nerd, volumizer, f.config.Containers.CleanupProcessDirsOnWait) diff --git a/netplugin/external_networker.go b/netplugin/external_networker.go index ae5cf4484..99013fb14 100644 --- a/netplugin/external_networker.go +++ b/netplugin/external_networker.go @@ -14,6 +14,8 @@ import ( "code.cloudfoundry.org/guardian/gardener" "code.cloudfoundry.org/guardian/kawasaki" "code.cloudfoundry.org/lager/v3" + "github.com/vishvananda/netlink" + "github.com/vishvananda/netns" ) const NetworkPropertyPrefix = "network." @@ -111,6 +113,15 @@ func (p *externalBinaryNetworker) Network(log lager.Logger, containerSpec garden return err } + // Ensure loopback interface is up in the container's network namespace. + // External network plugins (e.g. silk CNI) typically only create a veth pair. + // Runtimes like gVisor that build their own netstack by scraping the netns + // need lo to be present for applications binding to 127.0.0.1. + if err := setupLoopback(log, pid); err != nil { + log.Error("setup-loopback", err) + // Non-fatal: runc brings up lo on its own, only gVisor needs this. + } + for k, v := range outputs.Properties { p.configStore.Set(containerSpec.Handle, k, v) } @@ -151,6 +162,36 @@ func (p *externalBinaryNetworker) Network(log lager.Logger, containerSpec garden return nil } +// setupLoopback enters the container's network namespace and brings up the +// loopback interface. This is needed for runtimes like gVisor that scrape the +// netns for interfaces rather than creating lo themselves. +func setupLoopback(log lager.Logger, pid int) error { + containerNS, err := netns.GetFromPid(pid) + if err != nil { + return fmt.Errorf("getting netns for pid %d: %w", pid, err) + } + defer containerNS.Close() + + // Create a netlink handle in the container's network namespace. + nlh, err := netlink.NewHandleAt(containerNS) + if err != nil { + return fmt.Errorf("creating netlink handle in container netns: %w", err) + } + defer nlh.Close() + + lo, err := nlh.LinkByName("lo") + if err != nil { + return fmt.Errorf("finding lo interface: %w", err) + } + + if err := nlh.LinkSetUp(lo); err != nil { + return fmt.Errorf("bringing up lo: %w", err) + } + + log.Info("setup-loopback-success", lager.Data{"pid": pid}) + return nil +} + func (p *externalBinaryNetworker) Destroy(log lager.Logger, handle string) error { err := p.exec(log, "down", handle, nil, nil) if err != nil { diff --git a/rundmc/containerizer.go b/rundmc/containerizer.go index 910b7bcf3..c7aea215a 100644 --- a/rundmc/containerizer.go +++ b/rundmc/containerizer.go @@ -53,6 +53,7 @@ type State struct { type OCIRuntime interface { Create(log lager.Logger, id string, bundle goci.Bndl, io garden.ProcessIO) error + Start(log lager.Logger, id string) error Exec(log lager.Logger, id string, spec garden.ProcessSpec, io garden.ProcessIO) (garden.Process, error) Attach(log lager.Logger, id, processId string, io garden.ProcessIO) (garden.Process, error) Delete(log lager.Logger, id string) error @@ -108,7 +109,7 @@ type Containerizer struct { bundler BundleGenerator runtime OCIRuntime processesStopper ProcessesStopper - nstar NstarRunner + streamer Streamer events EventStore states StateStore peaCreator PeaCreator @@ -116,13 +117,14 @@ type Containerizer struct { cpuEntitlementPerShare float64 runtimeStopper RuntimeStopper cpuCgrouper CPUCgrouper + execPeas bool } func New( depot Depot, bundler BundleGenerator, runtime OCIRuntime, - nstarRunner NstarRunner, + streamer Streamer, processesStopper ProcessesStopper, events EventStore, states StateStore, @@ -131,12 +133,13 @@ func New( cpuEntitlementPerShare float64, runtimeStopper RuntimeStopper, cpuCgrouper CPUCgrouper, + execPeas bool, ) *Containerizer { containerizer := &Containerizer{ depot: depot, bundler: bundler, runtime: runtime, - nstar: nstarRunner, + streamer: streamer, processesStopper: processesStopper, events: events, states: states, @@ -145,6 +148,7 @@ func New( cpuEntitlementPerShare: cpuEntitlementPerShare, runtimeStopper: runtimeStopper, cpuCgrouper: cpuCgrouper, + execPeas: execPeas, } return containerizer } @@ -192,6 +196,11 @@ func (c *Containerizer) Create(log lager.Logger, spec spec.DesiredContainerSpec) return nil } +func (c *Containerizer) Start(log lager.Logger, handle string) error { + log = log.Session("containerizer-start", lager.Data{"handle": handle}) + return c.runtime.Start(log, handle) +} + // Run runs a process inside a running container func (c *Containerizer) Run(log lager.Logger, handle string, spec garden.ProcessSpec, io garden.ProcessIO) (garden.Process, error) { log = log.Session("run", lager.Data{"handle": handle, "path": spec.Path}) @@ -200,6 +209,16 @@ func (c *Containerizer) Run(log lager.Logger, handle string, spec garden.Process defer log.Info("finished") if isPea(spec) { + if c.execPeas { + // On gVisor, peas (sidecar containers) cannot share namespaces across + // sandboxes. Run the process as exec inside the existing container instead. + // The required binaries (healthcheck, etc.) are already bind-mounted in. + log.Info("exec-pea-as-process", lager.Data{"path": spec.Path}) + spec.Image = garden.ImageRef{} + spec.BindMounts = nil + return c.runtime.Exec(log, handle, spec, io) + } + if shouldResolveUsername(spec.User) { resolvedUID, resolvedGID, err := c.peaUsernameResolver.ResolveUser(log, handle, spec.Image, spec.User) if err != nil { @@ -248,18 +267,7 @@ func (c *Containerizer) StreamIn(log lager.Logger, handle string, spec garden.St _ = metrics.SendValue("StreamInDuration", float64(time.Since(startedAt).Nanoseconds()), "nanos") }(time.Now()) - state, err := c.runtime.State(log, handle) - if err != nil { - log.Error("check-pid-failed", err) - return fmt.Errorf("stream-in: pid not found for container") - } - - if err := c.nstar.StreamIn(log, state.Pid, spec.Path, spec.User, spec.TarStream); err != nil { - log.Error("nstar-failed", err) - return fmt.Errorf("stream-in: nstar: %s", err) - } - - return nil + return c.streamer.StreamIn(log, handle, spec) } // StreamOut stream files from the container @@ -269,19 +277,7 @@ func (c *Containerizer) StreamOut(log lager.Logger, handle string, spec garden.S log.Info("started") defer log.Info("finished") - state, err := c.runtime.State(log, handle) - if err != nil { - log.Error("check-pid-failed", err) - return nil, fmt.Errorf("stream-out: pid not found for container") - } - - stream, err := c.nstar.StreamOut(log, state.Pid, spec.Path, spec.User) - if err != nil { - log.Error("nstar-failed", err) - return nil, fmt.Errorf("stream-out: nstar: %s", err) - } - - return stream, nil + return c.streamer.StreamOut(log, handle, spec) } // Stop stops all the processes other than the init process in the container diff --git a/rundmc/containerizer_test.go b/rundmc/containerizer_test.go index fd7da03d2..a1d635892 100644 --- a/rundmc/containerizer_test.go +++ b/rundmc/containerizer_test.go @@ -64,7 +64,7 @@ var _ = Describe("Rundmc", func() { fakeDepot, fakeBundleGenerator, fakeOCIRuntime, - fakeNstarRunner, + rundmc.NewNstarStreamer(fakeOCIRuntime, fakeNstarRunner), fakeProcessesStopper, fakeEventStore, fakeStateStore, @@ -655,7 +655,7 @@ var _ = Describe("Rundmc", func() { fakeDepot, fakeBundleGenerator, fakeOCIRuntime, - fakeNstarRunner, + rundmc.NewNstarStreamer(fakeOCIRuntime, fakeNstarRunner), fakeProcessesStopper, fakeEventStore, fakeStateStore, diff --git a/rundmc/peas/pea_creator.go b/rundmc/peas/pea_creator.go index 39426d0be..dbbf95faa 100644 --- a/rundmc/peas/pea_creator.go +++ b/rundmc/peas/pea_creator.go @@ -59,15 +59,16 @@ type ExecRunner interface { } type PeaCreator struct { - Volumizer Volumizer - PidGetter PidGetter - PrivilegedGetter PrivilegedGetter - NetworkDepot NetworkDepot - BundleGenerator BundleGenerator - BundleSaver depot.BundleSaver - ProcessBuilder runrunc.ProcessBuilder - ExecRunner ExecRunner - PeaCleaner gardener.PeaCleaner + Volumizer Volumizer + PidGetter PidGetter + PrivilegedGetter PrivilegedGetter + NetworkDepot NetworkDepot + BundleGenerator BundleGenerator + BundleSaver depot.BundleSaver + ProcessBuilder runrunc.ProcessBuilder + ExecRunner ExecRunner + PeaCleaner gardener.PeaCleaner + SkipUserNamespace bool } func (p *PeaCreator) CreatePea(log lager.Logger, processSpec garden.ProcessSpec, procIO garden.ProcessIO, sandboxHandle string) (_ garden.Process, theErr error) { @@ -181,7 +182,7 @@ func (p *PeaCreator) linuxNamespaces(log lager.Logger, sandboxHandle string, pri linuxNamespaces["pid"] = fmt.Sprintf("/proc/%d/ns/pid", originalCtrInitPid) linuxNamespaces["uts"] = fmt.Sprintf("/proc/%d/ns/uts", originalCtrInitPid) - if !privileged { + if !privileged && !p.SkipUserNamespace { linuxNamespaces["user"] = fmt.Sprintf("/proc/%d/ns/user", originalCtrInitPid) } diff --git a/rundmc/runcontainerd/cgroup_manager.go b/rundmc/runcontainerd/cgroup_manager.go index ce0b35dec..17e9bc51f 100644 --- a/rundmc/runcontainerd/cgroup_manager.go +++ b/rundmc/runcontainerd/cgroup_manager.go @@ -38,6 +38,9 @@ func NewCgroupManager(runcRoot, namespace string) CgroupManager { func (m cgroupManager) SetUseMemoryHierarchy(handle string) error { statePath := filepath.Join(m.runcRoot, m.namespace, handle, "state.json") + if _, statErr := os.Stat(statePath); os.IsNotExist(statErr) { + return nil + } stateFile, err := os.Open(statePath) if err != nil { return err diff --git a/rundmc/runcontainerd/cgroup_manager_test.go b/rundmc/runcontainerd/cgroup_manager_test.go index d4c9cfea1..8bae3bf70 100644 --- a/rundmc/runcontainerd/cgroup_manager_test.go +++ b/rundmc/runcontainerd/cgroup_manager_test.go @@ -72,8 +72,8 @@ var _ = Describe("CgroupManager", func() { containerHandle = "foo" }) - It("returns an error", func() { - Expect(setErr).To(HaveOccurred()) + It("returns nil (tolerates missing state for gVisor)", func() { + Expect(setErr).NotTo(HaveOccurred()) }) }) diff --git a/rundmc/runcontainerd/containerd_statser.go b/rundmc/runcontainerd/containerd_statser.go new file mode 100644 index 000000000..f51706b9c --- /dev/null +++ b/rundmc/runcontainerd/containerd_statser.go @@ -0,0 +1,122 @@ +//go:build !windows + +package runcontainerd + +import ( + "fmt" + "time" + + "code.cloudfoundry.org/garden" + "code.cloudfoundry.org/guardian/gardener" + "code.cloudfoundry.org/lager/v3" + cgroup2stats "github.com/containerd/cgroups/v3/cgroup2/stats" + "github.com/containerd/typeurl/v2" +) + +// ContainerMetricsGetter abstracts the containerd task metrics retrieval. +type ContainerMetricsGetter interface { + TaskMetrics(log lager.Logger, containerID string) (*cgroup2stats.Metrics, time.Time, error) +} + +// ContainerdStatser implements the Statser interface using containerd task +// metrics API instead of shelling out to runc. This is required for non-runc +// runtimes (e.g. gVisor/runsc) where "runc events --stats" does not work +// because runc has no knowledge of the container. +type ContainerdStatser struct { + metricsGetter ContainerMetricsGetter +} + +func NewContainerdStatser(metricsGetter ContainerMetricsGetter) *ContainerdStatser { + return &ContainerdStatser{metricsGetter: metricsGetter} +} + +func (s *ContainerdStatser) Stats(log lager.Logger, id string) (gardener.StatsContainerMetrics, error) { + log = log.Session("containerd-statser", lager.Data{"id": id}) + + metrics, createdAt, err := s.metricsGetter.TaskMetrics(log, id) + if err != nil { + return gardener.StatsContainerMetrics{}, fmt.Errorf("containerd task metrics: %w", err) + } + + var memoryStat garden.ContainerMemoryStat + if metrics.Memory != nil { + memoryStat = garden.ContainerMemoryStat{ + Anon: metrics.Memory.Anon, + File: metrics.Memory.File, + ActiveAnon: metrics.Memory.ActiveAnon, + InactiveAnon: metrics.Memory.InactiveAnon, + ActiveFile: metrics.Memory.ActiveFile, + InactiveFile: metrics.Memory.InactiveFile, + Unevictable: metrics.Memory.Unevictable, + MappedFile: metrics.Memory.FileMapped, + Pgfault: metrics.Memory.Pgfault, + Pgmajfault: metrics.Memory.Pgmajfault, + SwapCached: 0, // cgroupsv2 does not expose swapcached separately + // cgroupsv2 does not have hierarchical_memory_limit; + // usage_limit is the closest equivalent + HierarchicalMemoryLimit: metrics.Memory.UsageLimit, + // Populate TotalXxx with the same values (cgroupsv2 unified hierarchy + // means there is no separate "total" vs "local" distinction) + TotalActiveAnon: metrics.Memory.ActiveAnon, + TotalActiveFile: metrics.Memory.ActiveFile, + TotalInactiveAnon: metrics.Memory.InactiveAnon, + TotalInactiveFile: metrics.Memory.InactiveFile, + TotalUnevictable: metrics.Memory.Unevictable, + TotalMappedFile: metrics.Memory.FileMapped, + TotalPgfault: metrics.Memory.Pgfault, + TotalPgmajfault: metrics.Memory.Pgmajfault, + // For cgroupsv2, "Rss" is essentially Anon + swap-backed anon + Rss: metrics.Memory.Anon, + TotalRss: metrics.Memory.Anon, + // Cache = file-backed pages + Cache: metrics.Memory.File, + TotalCache: metrics.Memory.File, + // Swap + Swap: metrics.Memory.SwapUsage, + TotalSwap: metrics.Memory.SwapUsage, + } + // Calculate TotalUsageTowardLimit the same way garden does for cgroupsv2: + // File + Anon + SwapCached - InactiveFile + totalMemoryUsage := memoryStat.File + memoryStat.Anon + memoryStat.SwapCached + if memoryStat.InactiveFile > totalMemoryUsage { + memoryStat.TotalUsageTowardLimit = 0 + } else { + memoryStat.TotalUsageTowardLimit = totalMemoryUsage - memoryStat.InactiveFile + } + } + + var cpuStat garden.ContainerCPUStat + if metrics.CPU != nil { + // cgroupsv2 reports CPU in microseconds; garden expects nanoseconds + cpuStat = garden.ContainerCPUStat{ + Usage: metrics.CPU.UsageUsec * 1000, + User: metrics.CPU.UserUsec * 1000, + System: metrics.CPU.SystemUsec * 1000, + } + } + + var pidStat garden.ContainerPidStat + if metrics.Pids != nil { + pidStat = garden.ContainerPidStat{ + Current: metrics.Pids.Current, + Max: metrics.Pids.Limit, + } + } + + return gardener.StatsContainerMetrics{ + CPU: cpuStat, + Memory: memoryStat, + Pid: pidStat, + Age: time.Since(createdAt), + }, nil +} + +// UnmarshalContainerdMetrics unmarshals the protobuf Any from containerd task +// metrics response into cgroupsv2 Metrics. +func UnmarshalContainerdMetrics(data typeurl.Any) (*cgroup2stats.Metrics, error) { + var m cgroup2stats.Metrics + if err := typeurl.UnmarshalTo(data, &m); err != nil { + return nil, fmt.Errorf("unmarshal cgroupsv2 metrics: %w", err) + } + return &m, nil +} diff --git a/rundmc/runcontainerd/nerd/nerd.go b/rundmc/runcontainerd/nerd/nerd.go index 02a8eebe6..a517004d9 100644 --- a/rundmc/runcontainerd/nerd/nerd.go +++ b/rundmc/runcontainerd/nerd/nerd.go @@ -19,32 +19,39 @@ import ( apievents "github.com/containerd/containerd/api/events" v2types "github.com/containerd/containerd/api/types/runc/options" "github.com/containerd/containerd/v2/client" + "github.com/containerd/containerd/v2/plugins" ctrdevents "github.com/containerd/containerd/v2/core/events" "github.com/containerd/containerd/v2/pkg/cio" "github.com/containerd/errdefs" + cgroup2stats "github.com/containerd/cgroups/v3/cgroup2/stats" "github.com/containerd/typeurl/v2" specs "github.com/opencontainers/runtime-spec/specs-go" ) type Nerd struct { - client *client.Client - context context.Context - ioFifoDir string - mp *metrics.MetricsProvider + client *client.Client + context context.Context + ioFifoDir string + mp *metrics.MetricsProvider + runtimeType string } -func New(client *client.Client, context context.Context, ioFifoDir string, mp *metrics.MetricsProvider) *Nerd { +func New(client *client.Client, context context.Context, ioFifoDir string, mp *metrics.MetricsProvider, runtimeType string) *Nerd { return &Nerd{ - client: client, - context: context, - ioFifoDir: ioFifoDir, - mp: mp, + client: client, + context: context, + ioFifoDir: ioFifoDir, + mp: mp, + runtimeType: runtimeType, } } func (n *Nerd) Create(log lager.Logger, containerID string, spec *specs.Spec, hostUID, hostGID uint32, pio func() (io.Reader, io.Writer, io.Writer)) error { log.Debug("creating-container", lager.Data{"containerID": containerID}) + if n.runtimeType != "" && n.runtimeType != plugins.RuntimeRuncV2 { + spec.Windows = nil // runsc rejects the Windows section + } container, err := n.client.NewContainer(n.context, containerID, client.WithSpec(spec)) if err != nil { return err @@ -58,14 +65,23 @@ func (n *Nerd) Create(log lager.Logger, containerID string, spec *specs.Spec, ho } log.Debug("creating-task", lager.Data{"containerID": containerID}) - task, err := container.NewTask(n.context, cio.NewCreator(withProcessIO(noTTYProcessIO(pio), n.ioFifoDir)), client.WithNoNewKeyring, WithUIDAndGID(hostUID, hostGID)) + var taskOpts []client.NewTaskOpts + if n.runtimeType == "" || n.runtimeType == plugins.RuntimeRuncV2 { + // runsc manages its own keyring and UID/GID via the OCI spec + taskOpts = append(taskOpts, client.WithNoNewKeyring, WithUIDAndGID(hostUID, hostGID)) + } + _, err = container.NewTask(n.context, cio.NewCreator(withProcessIO(noTTYProcessIO(pio), n.ioFifoDir)), taskOpts...) + return err +} + +func (n *Nerd) Start(log lager.Logger, containerID string) error { + _, task, err := n.loadContainerAndTask(log, containerID) if err != nil { return err } log.Debug("starting-task", lager.Data{"containerID": containerID}) - err = task.Start(n.context) - if err != nil { + if err := task.Start(n.context); err != nil { return err } @@ -451,3 +467,38 @@ func updateTaskInfoCreateOptions(taskInfo *client.TaskInfo, updateCreateOptions return updateCreateOptions(opts) } + +// TaskMetrics retrieves cgroup metrics for a container via the containerd task +// API. Returns cgroupsv2 Metrics and the container creation time (used for Age). +func (n *Nerd) TaskMetrics(log lager.Logger, containerID string) (*cgroup2stats.Metrics, time.Time, error) { + log = log.Session("task-metrics", lager.Data{"containerID": containerID}) + + container, task, err := n.loadContainerAndTask(log, containerID) + if err != nil { + return nil, time.Time{}, err + } + + metric, err := task.Metrics(n.context) + if err != nil { + return nil, time.Time{}, fmt.Errorf("task.Metrics: %w", err) + } + + if metric.Data == nil { + return nil, time.Time{}, fmt.Errorf("no metrics data for container %s", containerID) + } + + cgMetrics, err := runcontainerd.UnmarshalContainerdMetrics(metric.Data) + if err != nil { + return nil, time.Time{}, err + } + + // Get creation time from container info + info, err := container.Info(n.context) + if err != nil { + // Non-fatal: use zero time (Age will be large but metrics still report) + log.Error("get-info-for-creation-time", err) + return cgMetrics, time.Time{}, nil + } + + return cgMetrics, info.CreatedAt, nil +} diff --git a/rundmc/runcontainerd/runcontainerd.go b/rundmc/runcontainerd/runcontainerd.go index 1d534c483..7f02f58c4 100644 --- a/rundmc/runcontainerd/runcontainerd.go +++ b/rundmc/runcontainerd/runcontainerd.go @@ -4,9 +4,7 @@ import ( "fmt" "io" "os" - "path/filepath" "regexp" - "strconv" "code.cloudfoundry.org/guardian/rundmc" @@ -25,6 +23,7 @@ import ( //counterfeiter:generate . ContainerManager type ContainerManager interface { Create(log lager.Logger, containerID string, spec *specs.Spec, containerRootUID, containerRootGID uint32, processIO func() (io.Reader, io.Writer, io.Writer)) error + Start(log lager.Logger, containerID string) error Delete(log lager.Logger, containerID string) error Exec(log lager.Logger, containerID, processID string, spec *specs.Process, processIO func() (io.Reader, io.Writer, io.Writer, bool)) (BackingProcess, error) State(log lager.Logger, containerID string) (int, string, error) @@ -151,6 +150,10 @@ func (r *RunContainerd) Create(log lager.Logger, id string, bundle goci.Bndl, pi return nil } +func (r *RunContainerd) Start(log lager.Logger, id string) error { + return r.containerManager.Start(log, id) +} + func updateAnnotationsIfNeeded(bundle *goci.Bndl) { if _, ok := bundle.Spec.Annotations["container-type"]; !ok { if bundle.Spec.Annotations == nil { @@ -170,18 +173,13 @@ func (r *RunContainerd) Exec(log lager.Logger, containerID string, gardenProcess return r.execer.ExecWithBndl(log, containerID, bundle, gardenProcessSpec, gardenIO) } - containerPid, err := r.containerManager.GetContainerPID(log, containerID) - if err != nil { - return nil, err - } + rootfsPath := bundle.Spec.Root.Path - resolvedUser, err := r.userLookupper.Lookup(fmt.Sprintf("/proc/%d/root", containerPid), gardenProcessSpec.User) + resolvedUser, err := r.userLookupper.Lookup(rootfsPath, gardenProcessSpec.User) if err != nil { return nil, err } - rootfsPath := filepath.Join("/proc", strconv.FormatInt(int64(containerPid), 10), "root") - hostUID := idmapper.MappingList(bundle.Spec.Linux.UIDMappings).Map(resolvedUser.Uid) hostGID := idmapper.MappingList(bundle.Spec.Linux.GIDMappings).Map(resolvedUser.Gid) diff --git a/rundmc/runcontainerd/runcontainerd_test.go b/rundmc/runcontainerd/runcontainerd_test.go index d8190910a..536c5dde6 100644 --- a/rundmc/runcontainerd/runcontainerd_test.go +++ b/rundmc/runcontainerd/runcontainerd_test.go @@ -5,7 +5,6 @@ import ( "errors" "io" "os" - "path/filepath" "code.cloudfoundry.org/guardian/rundmc" @@ -299,6 +298,7 @@ var _ = Describe("Runcontainerd", func() { bundle = goci.Bndl{ Spec: specs.Spec{ Hostname: "test-hostname", + Root: &specs.Root{Path: "/test/rootfs"}, Linux: &specs.Linux{}, }, }.WithUIDMappings(specs.LinuxIDMapping{ @@ -417,13 +417,10 @@ var _ = Describe("Runcontainerd", func() { }) It("creates the process with the resolved user", func() { - _, actualContainerId := containerManager.GetContainerPIDArgsForCall(0) - Expect(actualContainerId).To(Equal("container-id")) - Expect(userLookupper.LookupCallCount()).To(Equal(1)) passedRootfs, passedUserId := userLookupper.LookupArgsForCall(0) Expect(passedUserId).To(Equal("testuser")) - Expect(passedRootfs).To(Equal("/proc/1234/root")) + Expect(passedRootfs).To(Equal("/test/rootfs")) Expect(processBuilder.BuildProcessCallCount()).To(Equal(1)) _, _, user := processBuilder.BuildProcessArgsForCall(0) @@ -435,7 +432,7 @@ var _ = Describe("Runcontainerd", func() { It("sets up the working directory", func() { Expect(mkdirer.MkdirAsCallCount()).To(Equal(1)) rootfsPath, hostUID, hostGID, mode, shouldRecreate, workDir := mkdirer.MkdirAsArgsForCall(0) - Expect(rootfsPath).To(Equal(filepath.Join("/proc", "1234", "root"))) + Expect(rootfsPath).To(Equal("/test/rootfs")) Expect(hostUID).To(Equal(2000)) Expect(hostGID).To(Equal(2001)) Expect(mode).To(Equal(os.FileMode(0755))) @@ -505,16 +502,6 @@ var _ = Describe("Runcontainerd", func() { }) }) - Context("when getting the container PID fails", func() { - BeforeEach(func() { - containerManager.GetContainerPIDReturns(0, errors.New("get-container-pid-failure")) - }) - - It("returns the error", func() { - Expect(execErr).To(MatchError("get-container-pid-failure")) - }) - }) - Context("when containerManager returns an error", func() { BeforeEach(func() { containerManager.ExecReturns(nil, errors.New("error-execing")) diff --git a/rundmc/rundmcfakes/fake_ociruntime.go b/rundmc/rundmcfakes/fake_ociruntime.go index c88770eca..bc8dedee4 100644 --- a/rundmc/rundmcfakes/fake_ociruntime.go +++ b/rundmc/rundmcfakes/fake_ociruntime.go @@ -138,6 +138,18 @@ type FakeOCIRuntime struct { removeBundleReturnsOnCall map[int]struct { result1 error } + StartStub func(lager.Logger, string) error + startMutex sync.RWMutex + startArgsForCall []struct { + arg1 lager.Logger + arg2 string + } + startReturns struct { + result1 error + } + startReturnsOnCall map[int]struct { + result1 error + } StateStub func(lager.Logger, string) (rundmc.State, error) stateMutex sync.RWMutex stateArgsForCall []struct { @@ -745,6 +757,68 @@ func (fake *FakeOCIRuntime) RemoveBundleReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeOCIRuntime) Start(arg1 lager.Logger, arg2 string) error { + fake.startMutex.Lock() + ret, specificReturn := fake.startReturnsOnCall[len(fake.startArgsForCall)] + fake.startArgsForCall = append(fake.startArgsForCall, struct { + arg1 lager.Logger + arg2 string + }{arg1, arg2}) + stub := fake.StartStub + fakeReturns := fake.startReturns + fake.recordInvocation("Start", []interface{}{arg1, arg2}) + fake.startMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeOCIRuntime) StartCallCount() int { + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() + return len(fake.startArgsForCall) +} + +func (fake *FakeOCIRuntime) StartCalls(stub func(lager.Logger, string) error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = stub +} + +func (fake *FakeOCIRuntime) StartArgsForCall(i int) (lager.Logger, string) { + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() + argsForCall := fake.startArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeOCIRuntime) StartReturns(result1 error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = nil + fake.startReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeOCIRuntime) StartReturnsOnCall(i int, result1 error) { + fake.startMutex.Lock() + defer fake.startMutex.Unlock() + fake.StartStub = nil + if fake.startReturnsOnCall == nil { + fake.startReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.startReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeOCIRuntime) State(arg1 lager.Logger, arg2 string) (rundmc.State, error) { fake.stateMutex.Lock() ret, specificReturn := fake.stateReturnsOnCall[len(fake.stateArgsForCall)] @@ -896,6 +970,8 @@ func (fake *FakeOCIRuntime) Invocations() map[string][][]interface{} { defer fake.execMutex.RUnlock() fake.removeBundleMutex.RLock() defer fake.removeBundleMutex.RUnlock() + fake.startMutex.RLock() + defer fake.startMutex.RUnlock() fake.stateMutex.RLock() defer fake.stateMutex.RUnlock() fake.statsMutex.RLock() diff --git a/rundmc/runrunc/runrunc.go b/rundmc/runrunc/runrunc.go index 7385e0058..de990b986 100644 --- a/rundmc/runrunc/runrunc.go +++ b/rundmc/runrunc/runrunc.go @@ -60,3 +60,7 @@ func New( BundleManager: bundleManager, } } + +func (r *RunRunc) Start(_ lager.Logger, _ string) error { + return nil +} diff --git a/rundmc/streamer.go b/rundmc/streamer.go new file mode 100644 index 000000000..c55f4f2d1 --- /dev/null +++ b/rundmc/streamer.go @@ -0,0 +1,145 @@ +package rundmc + +import ( + "bytes" + "fmt" + "io" + "os" + "path/filepath" + "strings" + + "code.cloudfoundry.org/garden" + "code.cloudfoundry.org/lager/v3" +) + +//counterfeiter:generate . Streamer +type Streamer interface { + StreamIn(log lager.Logger, handle string, spec garden.StreamInSpec) error + StreamOut(log lager.Logger, handle string, spec garden.StreamOutSpec) (io.ReadCloser, error) +} + +type NstarStreamer struct { + runtime OCIRuntime + nstar NstarRunner +} + +func NewNstarStreamer(runtime OCIRuntime, nstar NstarRunner) *NstarStreamer { + return &NstarStreamer{runtime: runtime, nstar: nstar} +} + +func (s *NstarStreamer) StreamIn(log lager.Logger, handle string, spec garden.StreamInSpec) error { + state, err := s.runtime.State(log, handle) + if err != nil { + return fmt.Errorf("stream-in: pid not found for container") + } + if err := s.nstar.StreamIn(log, state.Pid, spec.Path, spec.User, spec.TarStream); err != nil { + return fmt.Errorf("stream-in: nstar: %s", err) + } + return nil +} + +func (s *NstarStreamer) StreamOut(log lager.Logger, handle string, spec garden.StreamOutSpec) (io.ReadCloser, error) { + state, err := s.runtime.State(log, handle) + if err != nil { + return nil, fmt.Errorf("stream-out: pid not found for container") + } + stream, err := s.nstar.StreamOut(log, state.Pid, spec.Path, spec.User) + if err != nil { + return nil, fmt.Errorf("stream-out: nstar: %s", err) + } + return stream, nil +} + +type ExecStreamer struct { + runtime OCIRuntime +} + +func NewExecStreamer(runtime OCIRuntime) *ExecStreamer { + return &ExecStreamer{runtime: runtime} +} + +func (s *ExecStreamer) StreamIn(log lager.Logger, handle string, spec garden.StreamInSpec) error { + user := spec.User + if user == "" { + user = "root" + } + + if err := s.execAndWait(log, handle, garden.ProcessSpec{ + Path: "/bin/mkdir", + Args: []string{"-p", spec.Path}, + User: user, + }, garden.ProcessIO{Stdout: io.Discard, Stderr: io.Discard}); err != nil { + return fmt.Errorf("stream-in: mkdir: %s", err) + } + + var stderrBuf bytes.Buffer + if err := s.execAndWait(log, handle, garden.ProcessSpec{ + Path: "/bin/tar", + Args: []string{"-xf", "-", "-C", spec.Path}, + User: user, + }, garden.ProcessIO{Stdin: spec.TarStream, Stdout: io.Discard, Stderr: &stderrBuf}); err != nil { + if stderrBuf.Len() > 0 { + log.Error("stream-in-tar-stderr", fmt.Errorf("%s", stderrBuf.String())) + } + return fmt.Errorf("stream-in: tar: %s", err) + } + + return nil +} + +func (s *ExecStreamer) StreamOut(log lager.Logger, handle string, spec garden.StreamOutSpec) (io.ReadCloser, error) { + path := spec.Path + sourcePath := filepath.Dir(path) + compressPath := filepath.Base(path) + if strings.HasSuffix(path, "/") { + sourcePath = path + compressPath = "." + } + + reader, writer, err := os.Pipe() + if err != nil { + return nil, err + } + + user := spec.User + if user == "" { + user = "root" + } + + var stderrBuf bytes.Buffer + process, err := s.runtime.Exec(log, handle, garden.ProcessSpec{ + Path: "/bin/tar", + Args: []string{"-cf", "-", "-C", sourcePath, compressPath}, + User: user, + }, garden.ProcessIO{Stdout: writer, Stderr: &stderrBuf}) + if err != nil { + writer.Close() + reader.Close() + return nil, fmt.Errorf("stream-out: exec tar: %s", err) + } + + go func() { + exitCode, waitErr := process.Wait() + if waitErr != nil || exitCode != 0 { + log.Error("stream-out-tar-failed", fmt.Errorf("exit=%d stderr=%s", exitCode, stderrBuf.String())) + } + writer.Close() + }() + + return reader, nil +} + +func (s *ExecStreamer) execAndWait(log lager.Logger, handle string, spec garden.ProcessSpec, pio garden.ProcessIO) error { + process, err := s.runtime.Exec(log, handle, spec, pio) + if err != nil { + return err + } + exitCode, err := process.Wait() + if err != nil { + return err + } + if exitCode != 0 { + return fmt.Errorf("exited %d", exitCode) + } + return nil +}