Add Event controls

Add Event controls and implement "stream" commands.

PiperOrigin-RevId: 390691702

5 files changed, 203 insertions, 0 deletions

diff --git a/pkg/eventchannel/BUILD b/pkg/eventchannel/BUILD
index ad15d3672..56399232a 100644
--- a/pkg/eventchannel/BUILD
+++ b/pkg/eventchannel/BUILD
@@ -7,6 +7,7 @@ go_library(
     srcs = [
         "event.go",
         "event_any.go",
+        "processor.go",
         "rate.go",
     ],
     visibility = ["//:sandbox"],
diff --git a/pkg/eventchannel/event_any.go b/pkg/eventchannel/event_any.go
index 13f300061..b708937a4 100644
--- a/pkg/eventchannel/event_any.go
+++ b/pkg/eventchannel/event_any.go
@@ -26,3 +26,8 @@ import (
 func newAny(m proto.Message) (*anypb.Any, error) {
 	return anypb.New(m)
 }
+
+func emptyAny() *anypb.Any {
+	var any anypb.Any
+	return &any
+}
diff --git a/pkg/eventchannel/processor.go b/pkg/eventchannel/processor.go
new file mode 100644
index 000000000..e765c10d1
--- /dev/null
+++ b/pkg/eventchannel/processor.go
@@ -0,0 +1,130 @@
+// Copyright 2021 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package eventchannel
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+	pb "gvisor.dev/gvisor/pkg/eventchannel/eventchannel_go_proto"
+)
+
+// eventProcessor carries display state across multiple events.
+type eventProcessor struct {
+	filtering bool
+	// filtered is the number of events omitted since printing the last matching
+	// event. Only meaningful when filtering == true.
+	filtered uint64
+	// allowlist is the set of event names to display. If empty, all events are
+	// displayed.
+	allowlist map[string]bool
+}
+
+// newEventProcessor creates a new EventProcessor with filters.
+func newEventProcessor(filters []string) *eventProcessor {
+	e := &eventProcessor{
+		filtering: len(filters) > 0,
+		allowlist: make(map[string]bool),
+	}
+	for _, f := range filters {
+		e.allowlist[f] = true
+	}
+	return e
+}
+
+// processOne reads, parses and displays a single event from the event channel.
+//
+// The event channel is a stream of (msglen, payload) packets; this function
+// processes a single such packet. The msglen is a uvarint-encoded length for
+// the associated payload. The payload is a binary-encoded 'Any' protobuf, which
+// in turn encodes an arbitrary event protobuf.
+func (e *eventProcessor) processOne(src io.Reader, out *os.File) error {
+	// Read and parse the msglen.
+	lenbuf := make([]byte, binary.MaxVarintLen64)
+	if _, err := io.ReadFull(src, lenbuf); err != nil {
+		return err
+	}
+	msglen, consumed := binary.Uvarint(lenbuf)
+	if consumed <= 0 {
+		return fmt.Errorf("couldn't parse the message length")
+	}
+
+	// Read the payload.
+	buf := make([]byte, msglen)
+	// Copy any unused bytes from the len buffer into the payload buffer. These
+	// bytes are actually part of the payload.
+	extraBytes := copy(buf, lenbuf[consumed:])
+	if _, err := io.ReadFull(src, buf[extraBytes:]); err != nil {
+		return err
+	}
+
+	// Unmarshal the payload into an "Any" protobuf, which encodes the actual
+	// event.
+	encodedEv := emptyAny()
+	if err := proto.Unmarshal(buf, encodedEv); err != nil {
+		return fmt.Errorf("failed to unmarshal 'any' protobuf message: %v", err)
+	}
+
+	var ev pb.DebugEvent
+	if err := (encodedEv).UnmarshalTo(&ev); err != nil {
+		return fmt.Errorf("failed to decode 'any' protobuf message: %v", err)
+	}
+
+	if e.filtering && e.allowlist[ev.Name] {
+		e.filtered++
+		return nil
+	}
+
+	if e.filtering && e.filtered > 0 {
+		if e.filtered == 1 {
+			fmt.Fprintf(out, "... filtered %d event ...\n\n", e.filtered)
+		} else {
+			fmt.Fprintf(out, "... filtered %d events ...\n\n", e.filtered)
+		}
+		e.filtered = 0
+	}
+
+	// Extract the inner event and display it. Example:
+	//
+	//   2017-10-04 14:35:05.316180374 -0700 PDT m=+1.132485846
+	//   cloud_gvisor.MemoryUsage {
+	//   total: 23822336
+	//   }
+	fmt.Fprintf(out, "%v\n%v {\n", time.Now(), ev.Name)
+	fmt.Fprintf(out, "%v", ev.Text)
+	fmt.Fprintf(out, "}\n\n")
+
+	return nil
+}
+
+// ProcessAll reads, parses and displays all events from src. The events are
+// displayed to out.
+func ProcessAll(src io.Reader, filters []string, out *os.File) error {
+	ep := newEventProcessor(filters)
+	for {
+		switch err := ep.processOne(src, out); err {
+		case nil:
+			continue
+		case io.EOF:
+			return nil
+		default:
+			return err
+		}
+	}
+}
diff --git a/pkg/sentry/control/BUILD b/pkg/sentry/control/BUILD
index fa3fe47c1..a4934a565 100644
--- a/pkg/sentry/control/BUILD
+++ b/pkg/sentry/control/BUILD
@@ -6,6 +6,7 @@ go_library(
     name = "control",
     srcs = [
         "control.go",
+        "events.go",
         "fs.go",
         "lifecycle.go",
         "logging.go",
@@ -20,6 +21,7 @@ go_library(
     deps = [
         "//pkg/abi/linux",
         "//pkg/context",
+        "//pkg/eventchannel",
         "//pkg/fd",
         "//pkg/log",
         "//pkg/sentry/fdimport",
diff --git a/pkg/sentry/control/events.go b/pkg/sentry/control/events.go
new file mode 100644
index 000000000..92e437ae7
--- /dev/null
+++ b/pkg/sentry/control/events.go
@@ -0,0 +1,65 @@
+// Copyright 2021 The gVisor Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package control
+
+import (
+	"errors"
+	"fmt"
+
+	"gvisor.dev/gvisor/pkg/eventchannel"
+	"gvisor.dev/gvisor/pkg/urpc"
+)
+
+// EventsOpts are the arguments for eventchannel-related commands.
+type EventsOpts struct {
+	urpc.FilePayload
+}
+
+// Events is the control server state for eventchannel-related commands.
+type Events struct {
+	emitter eventchannel.Emitter
+}
+
+// AttachDebugEmitter receives a connected unix domain socket FD from the client
+// and establishes it as a new emitter for the sentry eventchannel. Any existing
+// emitters are replaced on a subsequent attach.
+func (e *Events) AttachDebugEmitter(o *EventsOpts, _ *struct{}) error {
+	if len(o.FilePayload.Files) < 1 {
+		return errors.New("no output writer provided")
+	}
+
+	sock, err := o.ReleaseFD(0)
+	if err != nil {
+		return err
+	}
+	sockFD := sock.Release()
+
+	// SocketEmitter takes ownership of sockFD.
+	emitter, err := eventchannel.SocketEmitter(sockFD)
+	if err != nil {
+		return fmt.Errorf("failed to create SocketEmitter for FD %d: %v", sockFD, err)
+	}
+
+	// If there is already a debug emitter, close the old one.
+	if e.emitter != nil {
+		e.emitter.Close()
+	}
+
+	e.emitter = eventchannel.DebugEmitterFrom(emitter)
+
+	// Register the new stream destination.
+	eventchannel.AddEmitter(e.emitter)
+	return nil
+}