diff options
author | Chong Cai <chongc@google.com> | 2021-08-13 14:17:56 -0700 |
---|---|---|
committer | gVisor bot <gvisor-bot@google.com> | 2021-08-13 14:20:12 -0700 |
commit | 6eb8596f72f3c889de3f826b82319d41ac655829 (patch) | |
tree | 6dcae39e3fbb4bdf078dd5d77e19ef5626712d8d /pkg/eventchannel | |
parent | a7b59445db6b76611ea384659cff8f0dfa75cb00 (diff) |
Add Event controls
Add Event controls and implement "stream" commands.
PiperOrigin-RevId: 390691702
Diffstat (limited to 'pkg/eventchannel')
-rw-r--r-- | pkg/eventchannel/BUILD | 1 | ||||
-rw-r--r-- | pkg/eventchannel/event_any.go | 5 | ||||
-rw-r--r-- | pkg/eventchannel/processor.go | 130 |
3 files changed, 136 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 + } + } +} |