From 0ae98218be7447081c3cb2deacd80deee7df78ea Mon Sep 17 00:00:00 2001 From: Fabricio Voznika Date: Wed, 22 Jun 2022 19:20:32 -0700 Subject: [PATCH] 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 | 3 +- pkg/sentry/seccheck/checkers/null/null.go | 8 +- pkg/sentry/seccheck/checkers/remote/BUILD | 1 + pkg/sentry/seccheck/checkers/remote/remote.go | 104 ++++++++++++++++-- .../seccheck/checkers/remote/remote_test.go | 79 +++++++++++++ pkg/sentry/seccheck/config.go | 13 ++- pkg/sentry/seccheck/seccheck.go | 26 ++++- pkg/sentry/seccheck/seccheck_test.go | 5 + runsc/cmd/trace/list.go | 3 + 9 files changed, 227 insertions(+), 15 deletions(-) diff --git a/examples/seccheck/pod_init.json b/examples/seccheck/pod_init.json index a08b26c54..8c26221a4 100644 --- a/examples/seccheck/pod_init.json +++ b/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 } diff --git a/pkg/sentry/seccheck/checkers/null/null.go b/pkg/sentry/seccheck/checkers/null/null.go index 9837952cd..d6c4d9b9b 100644 --- a/pkg/sentry/seccheck/checkers/null/null.go +++ b/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 +} diff --git a/pkg/sentry/seccheck/checkers/remote/BUILD b/pkg/sentry/seccheck/checkers/remote/BUILD index 9a86ba389..0606195b2 100644 --- a/pkg/sentry/seccheck/checkers/remote/BUILD +++ b/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", diff --git a/pkg/sentry/seccheck/checkers/remote/remote.go b/pkg/sentry/seccheck/checkers/remote/remote.go index aadbd40f9..1945e5be8 100644 --- a/pkg/sentry/seccheck/checkers/remote/remote.go +++ b/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 + } } } diff --git a/pkg/sentry/seccheck/checkers/remote/remote_test.go b/pkg/sentry/seccheck/checkers/remote/remote_test.go index d7c3bd69e..977f5b89e 100644 --- a/pkg/sentry/seccheck/checkers/remote/remote_test.go +++ b/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) diff --git a/pkg/sentry/seccheck/config.go b/pkg/sentry/seccheck/config.go index 5f9fe4fe3..003d7864d 100644 --- a/pkg/sentry/seccheck/config.go +++ b/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) } } diff --git a/pkg/sentry/seccheck/seccheck.go b/pkg/sentry/seccheck/seccheck.go index 64310e3b7..3aec5fecb 100644 --- a/pkg/sentry/seccheck/seccheck.go +++ b/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() {} diff --git a/pkg/sentry/seccheck/seccheck_test.go b/pkg/sentry/seccheck/seccheck_test.go index 109fbfb2e..d987f1eb5 100644 --- a/pkg/sentry/seccheck/seccheck_test.go +++ b/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 { diff --git a/runsc/cmd/trace/list.go b/runsc/cmd/trace/list.go index e45d1919b..37f16ec39 100644 --- a/runsc/cmd/trace/list.go +++ b/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 }