Add Event controls

Add Event controls and implement "stream" commands.

PiperOrigin-RevId: 390691702

9 files changed, 122 insertions, 2 deletions

diff --git a/runsc/boot/controller.go b/runsc/boot/controller.go
index b0282604a..ae32b86e6 100644
--- a/runsc/boot/controller.go
+++ b/runsc/boot/controller.go
@@ -121,6 +121,11 @@ const (
 	UsageReduce  = "Usage.Reduce"
 )
 
+// Events related commands (see events.go for more details).
+const (
+	EventsAttachDebugEmitter = "Events.AttachDebugEmitter"
+)
+
 // ControlSocketAddr generates an abstract unix socket name for the given ID.
 func ControlSocketAddr(id string) string {
 	return fmt.Sprintf("\x00runsc-sandbox.%s", id)
@@ -161,6 +166,7 @@ func newController(fd int, l *Loader) (*controller, error) {
 	}
 
 	ctrl.srv.Register(&debug{})
+	ctrl.srv.Register(&control.Events{})
 	ctrl.srv.Register(&control.Logging{})
 	ctrl.srv.Register(&control.Lifecycle{l.k})
 	ctrl.srv.Register(&control.Fs{l.k})
diff --git a/runsc/boot/strace.go b/runsc/boot/strace.go
index c21648a32..cf5be34cd 100644
--- a/runsc/boot/strace.go
+++ b/runsc/boot/strace.go
@@ -35,9 +35,14 @@ func enableStrace(conf *config.Config) error {
 	}
 	strace.LogMaximumSize = max
 
+	sink := strace.SinkTypeLog
+	if conf.StraceEvent {
+		sink = strace.SinkTypeEvent
+	}
+
 	if len(conf.StraceSyscalls) == 0 {
-		strace.EnableAll(strace.SinkTypeLog)
+		strace.EnableAll(sink)
 		return nil
 	}
-	return strace.Enable(strings.Split(conf.StraceSyscalls, ","), strace.SinkTypeLog)
+	return strace.Enable(strings.Split(conf.StraceSyscalls, ","), sink)
 }
diff --git a/runsc/cmd/events.go b/runsc/cmd/events.go
index c1d029d7f..08246e543 100644
--- a/runsc/cmd/events.go
+++ b/runsc/cmd/events.go
@@ -33,6 +33,10 @@ type Events struct {
 	intervalSec int
 	// If true, events will print a single group of stats and exit.
 	stats bool
+	// If true, events will dump all filtered events to stdout.
+	stream bool
+	// filters for streamed events.
+	filters stringSlice
 }
 
 // Name implements subcommands.Command.Name.
@@ -62,6 +66,8 @@ OPTIONS:
 func (evs *Events) SetFlags(f *flag.FlagSet) {
 	f.IntVar(&evs.intervalSec, "interval", 5, "set the stats collection interval, in seconds")
 	f.BoolVar(&evs.stats, "stats", false, "display the container's stats then exit")
+	f.BoolVar(&evs.stream, "stream", false, "dump all filtered events to stdout")
+	f.Var(&evs.filters, "filters", "only display matching events")
 }
 
 // Execute implements subcommands.Command.Execute.
@@ -79,6 +85,13 @@ func (evs *Events) Execute(ctx context.Context, f *flag.FlagSet, args ...interfa
 		Fatalf("loading sandbox: %v", err)
 	}
 
+	if evs.stream {
+		if err := c.Stream(evs.filters, os.Stdout); err != nil {
+			Fatalf("Stream failed: %v", err)
+		}
+		return subcommands.ExitSuccess
+	}
+
 	// Repeatedly get stats from the container.
 	for {
 		// Get the event and print it as JSON.
diff --git a/runsc/config/config.go b/runsc/config/config.go
index cc4650180..b811a170a 100644
--- a/runsc/config/config.go
+++ b/runsc/config/config.go
@@ -117,6 +117,10 @@ type Config struct {
 	// StraceLogSize is the max size of data blobs to display.
 	StraceLogSize uint `flag:"strace-log-size"`
 
+	// StraceEvent indicates sending strace to events if true. Strace is
+	// sent to log if false.
+	StraceEvent bool `flag:"strace-event"`
+
 	// DisableSeccomp indicates whether seccomp syscall filters should be
 	// disabled. Pardon the double negation, but default to enabled is important.
 	DisableSeccomp bool
diff --git a/runsc/config/flags.go b/runsc/config/flags.go
index 6f1b5927a..8fde31167 100644
--- a/runsc/config/flags.go
+++ b/runsc/config/flags.go
@@ -56,6 +56,7 @@ func RegisterFlags() {
 		flag.Bool("strace", false, "enable strace.")
 		flag.String("strace-syscalls", "", "comma-separated list of syscalls to trace. If --strace is true and this list is empty, then all syscalls will be traced.")
 		flag.Uint("strace-log-size", 1024, "default size (in bytes) to log data argument blobs.")
+		flag.Bool("strace-event", false, "send strace to event.")
 
 		// Flags that control sandbox runtime behavior.
 		flag.String("platform", "ptrace", "specifies which platform to use: ptrace (default), kvm.")
diff --git a/runsc/container/container.go b/runsc/container/container.go
index 381f57574..50b0dd5e7 100644
--- a/runsc/container/container.go
+++ b/runsc/container/container.go
@@ -670,6 +670,12 @@ func (c *Container) Reduce(wait bool) error {
 	return c.Sandbox.Reduce(c.ID, wait)
 }
 
+// Stream dumps all events to out.
+func (c *Container) Stream(filters []string, out *os.File) error {
+	log.Debugf("Stream in container, cid: %s", c.ID)
+	return c.Sandbox.Stream(c.ID, filters, out)
+}
+
 // State returns the metadata of the container.
 func (c *Container) State() specs.State {
 	return specs.State{
diff --git a/runsc/container/container_test.go b/runsc/container/container_test.go
index 8f612db8f..681f5c1a9 100644
--- a/runsc/container/container_test.go
+++ b/runsc/container/container_test.go
@@ -2779,3 +2779,53 @@ func TestReduce(t *testing.T) {
 		t.Fatalf("error reduce from container: %v", err)
 	}
 }
+
+// TestStream checks that Stream dumps expected events.
+func TestStream(t *testing.T) {
+	spec, conf := sleepSpecConf(t)
+	conf.Strace = true
+	conf.StraceEvent = true
+	conf.StraceSyscalls = ""
+
+	_, bundleDir, cleanup, err := testutil.SetupContainer(spec, conf)
+	if err != nil {
+		t.Fatalf("error setting up container: %v", err)
+	}
+	defer cleanup()
+
+	args := Args{
+		ID:        testutil.RandomContainerID(),
+		Spec:      spec,
+		BundleDir: bundleDir,
+	}
+
+	cont, err := New(conf, args)
+	if err != nil {
+		t.Fatalf("Creating container: %v", err)
+	}
+	defer cont.Destroy()
+
+	if err := cont.Start(conf); err != nil {
+		t.Fatalf("starting container: %v", err)
+	}
+
+	r, w, err := os.Pipe()
+	if err != nil {
+		t.Fatalf("os.Create(): %v", err)
+	}
+
+	// Spawn a new thread to Stream events as it blocks indefinitely.
+	go func() {
+		cont.Stream(nil, w)
+	}()
+
+	buf := make([]byte, 1024)
+	if _, err := r.Read(buf); err != nil {
+		t.Fatalf("Read out: %v", err)
+	}
+
+	// A syscall strace event includes "Strace".
+	if got, want := string(buf), "Strace"; !strings.Contains(got, want) {
+		t.Errorf("out got %s, want include %s", buf, want)
+	}
+}
diff --git a/runsc/sandbox/BUILD b/runsc/sandbox/BUILD
index bc4a3fa32..d625230dd 100644
--- a/runsc/sandbox/BUILD
+++ b/runsc/sandbox/BUILD
@@ -17,12 +17,14 @@ go_library(
         "//pkg/control/client",
         "//pkg/control/server",
         "//pkg/coverage",
+        "//pkg/eventchannel",
         "//pkg/log",
         "//pkg/sentry/control",
         "//pkg/sentry/platform",
         "//pkg/sync",
         "//pkg/tcpip/header",
         "//pkg/tcpip/stack",
+        "//pkg/unet",
         "//pkg/urpc",
         "//runsc/boot",
         "//runsc/boot/platforms",
diff --git a/runsc/sandbox/sandbox.go b/runsc/sandbox/sandbox.go
index c8c6886c7..9fbce6bd6 100644
--- a/runsc/sandbox/sandbox.go
+++ b/runsc/sandbox/sandbox.go
@@ -35,10 +35,12 @@ import (
 	"gvisor.dev/gvisor/pkg/control/client"
 	"gvisor.dev/gvisor/pkg/control/server"
 	"gvisor.dev/gvisor/pkg/coverage"
+	"gvisor.dev/gvisor/pkg/eventchannel"
 	"gvisor.dev/gvisor/pkg/log"
 	"gvisor.dev/gvisor/pkg/sentry/control"
 	"gvisor.dev/gvisor/pkg/sentry/platform"
 	"gvisor.dev/gvisor/pkg/sync"
+	"gvisor.dev/gvisor/pkg/unet"
 	"gvisor.dev/gvisor/pkg/urpc"
 	"gvisor.dev/gvisor/runsc/boot"
 	"gvisor.dev/gvisor/runsc/boot/platforms"
@@ -1073,6 +1075,37 @@ func (s *Sandbox) Reduce(cid string, wait bool) error {
 	}, nil)
 }
 
+// Stream sends the AttachDebugEmitter call for a container in the sandbox, and
+// dumps filtered events to out.
+func (s *Sandbox) Stream(cid string, filters []string, out *os.File) error {
+	log.Debugf("Stream sandbox %q", s.ID)
+	conn, err := s.sandboxConnect()
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	r, w, err := unet.SocketPair(false)
+	if err != nil {
+		return err
+	}
+
+	wfd, err := w.Release()
+	if err != nil {
+		return fmt.Errorf("failed to release write socket FD: %v", err)
+	}
+
+	if err := conn.Call(boot.EventsAttachDebugEmitter, &control.EventsOpts{
+		FilePayload: urpc.FilePayload{Files: []*os.File{
+			os.NewFile(uintptr(wfd), "event sink"),
+		}},
+	}, nil); err != nil {
+		return fmt.Errorf("AttachDebugEmitter failed: %v", err)
+	}
+
+	return eventchannel.ProcessAll(r, filters, out)
+}
+
 // IsRunning returns true if the sandbox or gofer process is running.
 func (s *Sandbox) IsRunning() bool {
 	if s.Pid != 0 {