Add async write to remote checker

When writing a trace point fails, the remote checker can retry with
configurable exponential backoff. After the number of retries is
exceeded, the point is dropped. The number of dropped events are
reported in the message header and also in `runsc trace list`
command.

Updates #4805

PiperOrigin-RevId: 456662539

examples/seccheck/pod_init.json

@@ -46,7 +46,8 @@
       {
         "name": "remote",
         "config": {
-          "endpoint": "/tmp/gvisor_events.sock"
+          "endpoint": "/tmp/gvisor_events.sock",
+          "retries": 3
         },
         "ignore_setup_error": true
       }

pkg/sentry/seccheck/checkers/null/null.go

@@ -21,9 +21,11 @@ import (
 	"gvisor.dev/gvisor/pkg/sentry/seccheck"
 )
 
+const name = "null"
+
 func init() {
 	seccheck.RegisterSink(seccheck.SinkDesc{
-		Name: "null",
+		Name: name,
 		New:  new,
 	})
 }
@@ -38,3 +40,7 @@ var _ seccheck.Checker = (*null)(nil)
 func new(_ map[string]interface{}, _ *fd.FD) (seccheck.Checker, error) {
 	return &null{}, nil
 }
+
+func (*null) Name() string {
+	return name
+}

pkg/sentry/seccheck/checkers/remote/BUILD

@@ -7,6 +7,7 @@ go_library(
     srcs = ["remote.go"],
     visibility = ["//:sandbox"],
     deps = [
+        "//pkg/atomicbitops",
         "//pkg/cleanup",
         "//pkg/context",
         "//pkg/fd",

pkg/sentry/seccheck/checkers/remote/remote.go

@@ -21,9 +21,11 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"time"
 
 	"golang.org/x/sys/unix"
 	"google.golang.org/protobuf/proto"
+	"gvisor.dev/gvisor/pkg/atomicbitops"
 	"gvisor.dev/gvisor/pkg/cleanup"
 	"gvisor.dev/gvisor/pkg/context"
 	"gvisor.dev/gvisor/pkg/fd"
@@ -33,9 +35,11 @@ import (
 	pb "gvisor.dev/gvisor/pkg/sentry/seccheck/points/points_go_proto"
 )
 
+const name = "remote"
+
 func init() {
 	seccheck.RegisterSink(seccheck.SinkDesc{
-		Name:  "remote",
+		Name:  name,
 		Setup: setupSink,
 		New:   new,
 	})
@@ -48,6 +52,12 @@ func init() {
 // delaying/hanging indefinitely the application.
 type remote struct {
 	endpoint *fd.FD
+
+	droppedCount atomicbitops.Uint32
+
+	retries        int
+	initialBackoff time.Duration
+	maxBackoff     time.Duration
 }
 
 var _ seccheck.Checker = (*remote)(nil)
@@ -115,16 +125,76 @@ func setup(path string) (*os.File, error) {
 		return nil, fmt.Errorf("remote version (%d) is smaller than minimum supported (%d)", hsIn.Version, minSupportedVersion)
 	}
 
+	if err := unix.SetNonblock(int(f.Fd()), true); err != nil {
+		return nil, err
+	}
+
 	cu.Release()
 	return f, nil
 }
 
+func parseDuration(config map[string]interface{}, name string) (bool, time.Duration, error) {
+	opaque, ok := config[name]
+	if !ok {
+		return false, 0, nil
+	}
+	duration, ok := opaque.(string)
+	if !ok {
+		return false, 0, fmt.Errorf("%s %v is not an string", name, opaque)
+	}
+	rv, err := time.ParseDuration(duration)
+	if err != nil {
+		return false, 0, err
+	}
+	return true, rv, nil
+}
+
 // new creates a new Remote checker.
-func new(_ map[string]interface{}, endpoint *fd.FD) (seccheck.Checker, error) {
+func new(config map[string]interface{}, endpoint *fd.FD) (seccheck.Checker, error) {
 	if endpoint == nil {
 		return nil, fmt.Errorf("remote sink requires an endpoint")
 	}
-	return &remote{endpoint: endpoint}, nil
+	r := &remote{
+		endpoint:       endpoint,
+		initialBackoff: 25 * time.Microsecond,
+		maxBackoff:     10 * time.Millisecond,
+	}
+	if retriesOpaque, ok := config["retries"]; ok {
+		retries, ok := retriesOpaque.(float64)
+		if !ok {
+			return nil, fmt.Errorf("retries %q is not an int", retriesOpaque)
+		}
+		r.retries = int(retries)
+		if float64(r.retries) != retries {
+			return nil, fmt.Errorf("retries %q is not an int", retriesOpaque)
+		}
+	}
+	if ok, backoff, err := parseDuration(config, "backoff"); err != nil {
+		return nil, err
+	} else if ok {
+		r.initialBackoff = backoff
+	}
+	if ok, backoff, err := parseDuration(config, "backoff_max"); err != nil {
+		return nil, err
+	} else if ok {
+		r.maxBackoff = backoff
+	}
+	if r.initialBackoff > r.maxBackoff {
+		return nil, fmt.Errorf("initial backoff (%v) cannot be larger than max backoff (%v)", r.initialBackoff, r.maxBackoff)
+	}
+
+	log.Debugf("Remote sink created, endpoint FD: %d, %+v", r.endpoint.FD(), r)
+	return r, nil
+}
+
+func (*remote) Name() string {
+	return name
+}
+
+func (r *remote) Status() seccheck.CheckerStatus {
+	return seccheck.CheckerStatus{
+		DroppedCount: uint64(r.droppedCount.Load()),
+	}
 }
 
 // Stop implements seccheck.Checker.
@@ -143,17 +213,31 @@ func (r *remote) write(msg proto.Message, msgType pb.MessageType) {
 		return
 	}
 	hdr := wire.Header{
-		HeaderSize:  uint16(wire.HeaderStructSize),
-		MessageType: uint16(msgType),
+		HeaderSize:   uint16(wire.HeaderStructSize),
+		DroppedCount: r.droppedCount.Load(),
+		MessageType:  uint16(msgType),
 	}
 	var hdrOut [wire.HeaderStructSize]byte
 	hdr.MarshalUnsafe(hdrOut[:])
 
-	// TODO(gvisor.dev/issue/4805): Change to non-blocking write. Count as dropped
-	// if write fails.
-	if _, err := unix.Writev(r.endpoint.FD(), [][]byte{hdrOut[:], out}); err != nil {
-		log.Debugf("write(%+v, %v): %v", msg, msgType, err)
-		return
+	backoff := r.initialBackoff
+	for i := 0; ; i++ {
+		_, err := unix.Writev(r.endpoint.FD(), [][]byte{hdrOut[:], out})
+		if err == nil {
+			// Write succeeded, we're done!
+			return
+		}
+		if !errors.Is(err, unix.EAGAIN) || i >= r.retries {
+			log.Debugf("Write failed, dropping point: %v", err)
+			r.droppedCount.Add(1)
+			return
+		}
+		log.Debugf("Write failed, retrying (%d/%d) in %v: %v", i+1, r.retries, backoff, err)
+		time.Sleep(backoff)
+		backoff *= 2
+		if r.maxBackoff > 0 && backoff > r.maxBackoff {
+			backoff = r.maxBackoff
+		}
 	}
 }
 

pkg/sentry/seccheck/checkers/remote/remote_test.go

@@ -228,6 +228,85 @@ func TestExample(t *testing.T) {
 	}
 }
 
+func TestConfig(t *testing.T) {
+	for _, tc := range []struct {
+		name   string
+		config map[string]interface{}
+		want   *remote
+		err    string
+	}{
+		{
+			name:   "default",
+			config: map[string]interface{}{},
+			want: &remote{
+				retries:        0,
+				initialBackoff: 25 * time.Microsecond,
+				maxBackoff:     10 * time.Millisecond,
+			},
+		},
+		{
+			name: "all",
+			config: map[string]interface{}{
+				"retries":     float64(10),
+				"backoff":     "1s",
+				"backoff_max": "10s",
+			},
+			want: &remote{
+				retries:        10,
+				initialBackoff: time.Second,
+				maxBackoff:     10 * time.Second,
+			},
+		},
+		{
+			name: "bad-retries",
+			config: map[string]interface{}{
+				"retries": "10",
+			},
+			err: "retries",
+		},
+		{
+			name: "bad-backoff",
+			config: map[string]interface{}{
+				"backoff": "wrong",
+			},
+			err: "invalid duration",
+		},
+		{
+			name: "bad-backoff-max",
+			config: map[string]interface{}{
+				"backoff_max": 10,
+			},
+			err: "is not an string",
+		},
+		{
+			name: "bad-invalid-backoffs",
+			config: map[string]interface{}{
+				"retries":     float64(10),
+				"backoff":     "10s",
+				"backoff_max": "1s",
+			},
+			err: "cannot be larger than max",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var endpoint fd.FD
+			checker, err := new(tc.config, &endpoint)
+			if len(tc.err) == 0 {
+				if err != nil {
+					t.Fatalf("new(%q): %v", tc.config, err)
+				}
+				got := checker.(*remote)
+				got.endpoint = nil
+				if *got != *tc.want {
+					t.Errorf("wrong remote: want: %+v, got: %+v", tc.want, got)
+				}
+			} else if err == nil || !strings.Contains(err.Error(), tc.err) {
+				t.Errorf("wrong error: want: %v, got: %v", tc.err, err)
+			}
+		})
+	}
+}
+
 func BenchmarkSmall(t *testing.B) {
 	// Run server in a separate process just to isolate it as much as possible.
 	server, err := newExampleServer(false)

pkg/sentry/seccheck/config.go

@@ -63,6 +63,8 @@ type SinkConfig struct {
 	// IgnoreSetupError makes errors during sink setup to be ignored. Otherwise,
 	// failures will prevent the container from starting.
 	IgnoreSetupError bool `json:"ignore_setup_error,omitempty"`
+	// Status is the runtime status for the sink.
+	Status CheckerStatus `json:"status,omitempty"`
 	// FD is the endpoint returned from Setup. It may be nil.
 	FD *fd.FD `json:"-"`
 }
@@ -181,9 +183,16 @@ func List(out *[]SessionConfig) {
 	sessionsMu.Lock()
 	defer sessionsMu.Unlock()
 
-	for name := range sessions {
+	for name, state := range sessions {
 		// Only report session name. Consider adding rest of the fields as needed.
-		*out = append(*out, SessionConfig{Name: name})
+		session := SessionConfig{Name: name}
+		for _, checker := range state.getCheckers() {
+			session.Sinks = append(session.Sinks, SinkConfig{
+				Name:   checker.Name(),
+				Status: checker.Status(),
+			})
+		}
+		*out = append(*out, session)
 	}
 }
 

pkg/sentry/seccheck/seccheck.go

@@ -96,6 +96,10 @@ func (fm *FieldMask) Empty() bool {
 // may be missing requested fields in some cases (e.g. if the Checker is
 // registered concurrently with invocations of checkpoints).
 type Checker interface {
+	// Name return the checker name.
+	Name() string
+	// Status returns the checker runtime status.
+	Status() CheckerStatus
 	// Stop requests the checker to stop.
 	Stop()
 
@@ -110,11 +114,31 @@ type Checker interface {
 	RawSyscall(context.Context, FieldSet, *pb.Syscall) error
 }
 
+// CheckerStatus represents stats about each checker instance.
+type CheckerStatus struct {
+	// DroppedCount is the number of trace points dropped.
+	DroppedCount uint64
+}
+
 // CheckerDefaults may be embedded by implementations of Checker to obtain
 // no-op implementations of Checker methods that may be explicitly overridden.
 type CheckerDefaults struct{}
 
-var _ Checker = (*CheckerDefaults)(nil)
+// Add functions missing in CheckerDefaults to make it possible to check for the
+// implementation below to catch missing functions more easily.
+type checkerDefaultsImpl struct {
+	CheckerDefaults
+}
+
+// Name implements Checker.Name.
+func (checkerDefaultsImpl) Name() string { return "" }
+
+var _ Checker = (*checkerDefaultsImpl)(nil)
+
+// Status implements Checker.Status.
+func (CheckerDefaults) Status() CheckerStatus {
+	return CheckerStatus{}
+}
 
 // Stop implements Checker.Stop.
 func (CheckerDefaults) Stop() {}

pkg/sentry/seccheck/seccheck_test.go

@@ -28,6 +28,11 @@ type testChecker struct {
 	onClone func(ctx context.Context, fields FieldSet, info *pb.CloneInfo) error
 }
 
+// Name implements Checker.Name.
+func (c *testChecker) Name() string {
+	return "test-checker"
+}
+
 // Clone implements Checker.Clone.
 func (c *testChecker) Clone(ctx context.Context, fields FieldSet, info *pb.CloneInfo) error {
 	if c.onClone == nil {

runsc/cmd/trace/list.go

@@ -73,6 +73,9 @@ func (l *list) Execute(_ context.Context, f *flag.FlagSet, args ...interface{})
 	fmt.Printf("SESSIONS (%d)\n", len(sessions))
 	for _, session := range sessions {
 		fmt.Printf("%q\n", session.Name)
+		for _, sink := range session.Sinks {
+			fmt.Printf("\tSink: %q, dropped: %d\n", sink.Name, sink.Status.DroppedCount)
+		}
 	}
 	return subcommands.ExitSuccess
 }