Repository: vmware-tanzu/octant
Pull request #1425 (2020-09-29 00:46:02)

Create a streamer interface for logger
**What this PR does / why we need it**:

This change adds a streamer to Octant's logger. The default streamer is a ring buffer with 1500 slots that will allow multiple clients to get access to the latest 1500 log entries.

**Special notes for your reviewer**:

This PR is part of a series of changes that will allow Octant to show its logs in the web interface.

**Release note**:
```
release-note
```



Patch:
diff --git a/cmd/octant-electron/main.go b/cmd/octant-electron/main.go
index 72da0eec..58a5a7af 100644
--- a/cmd/octant-electron/main.go
+++ b/cmd/octant-electron/main.go
@@ -33,17 +33,13 @@ var (
 )
 
 func main() {
-	z, err := log.Init(0)
+	logger, err := log.Init(0)
 	if err != nil {
 		golog.Printf("unable to initialize logger: %v", err)
 		os.Exit(1)
 	}
 
-	defer func() {
-		_ = z.Sync()
-	}()
-
-	logger := log.Wrap(z.Sugar())
+	defer logger.Close()
 	ctx := log.WithLoggerContext(context.Background(), logger)
 
 	if err := run(ctx); err != nil {
diff --git a/internal/commands/dash.go b/internal/commands/dash.go
index 9dda6041..0683a3e7 100644
--- a/internal/commands/dash.go
+++ b/internal/commands/dash.go
@@ -16,8 +16,6 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
-	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog"
 
@@ -46,19 +44,13 @@ func newOctantCmd(version string, gitCommit string, buildTime string) *cobra.Com
 				logLevel = 1
 			}
 
-			z, err := log.Init(logLevel)
+			logger, err := log.Init(logLevel)
 			if err != nil {
 				golog.Printf("unable to initialize logger: %v", err)
 				os.Exit(1)
 			}
 
-			defer func() {
-				// this fails, but it should be safe to ignore according
-				// to https://github.com/uber-go/zap/issues/328
-				_ = z.Sync()
-			}()
-
-			logger := log.Wrap(z.Sugar())
+			defer logger.Close()
 
 			sigCh := make(chan os.Signal, 1)
 			signal.Notify(sigCh, os.Interrupt)
@@ -195,25 +187,3 @@ func bindViper(cmd *cobra.Command) error {
 
 	return nil
 }
-
-// Returns a new zap logger, setting level according to the provided
-// verbosity level as an offset of the base level, Info.
-// i.e. verboseLevel==0, level==Info
-//      verboseLevel==1, level==Debug
-func newZapLogger(verboseLevel int) (*zap.Logger, error) {
-	level := zapcore.InfoLevel - zapcore.Level(verboseLevel)
-	if level < zapcore.DebugLevel || level > zapcore.FatalLevel {
-		level = zapcore.DebugLevel
-	}
-
-	cfg := zap.Config{
-		Level:            zap.NewAtomicLevelAt(level),
-		Development:      true,
-		Encoding:         "console",
-		EncoderConfig:    zap.NewDevelopmentEncoderConfig(),
-		OutputPaths:      []string{"stderr"},
-		ErrorOutputPaths: []string{"stderr"},
-	}
-
-	return cfg.Build()
-}
diff --git a/internal/log/logger.go b/internal/log/logger.go
index 091f448f..6d3ea608 100644
--- a/internal/log/logger.go
+++ b/internal/log/logger.go
@@ -8,10 +8,12 @@ package log
 import (
 	"context"
 	"fmt"
+	"net/url"
 
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
+	"github.com/vmware-tanzu/octant/pkg/event"
 	"github.com/vmware-tanzu/octant/pkg/log"
 )
 
@@ -19,28 +21,64 @@ type key string
 
 var contextKey = key("com.heptio.logger")
 
+type sugaredLogWrapperOption func(wrapper *sugaredLogWrapper)
+
+func sugaredLogStreamer(s EventStreamer) sugaredLogWrapperOption {
+	return func(wrapper *sugaredLogWrapper) {
+		wrapper.streamer = s
+	}
+}
+
 // sugaredLogWrapper adapts a zap.SugaredLogger to the Logger interface
 type sugaredLogWrapper struct {
 	*zap.SugaredLogger
+
+	streamer EventStreamer
+}
+
+var _ log.LoggerCloser = &sugaredLogWrapper{}
+
+func newSugaredLogWrapper(sl *zap.SugaredLogger, options ...sugaredLogWrapperOption) *sugaredLogWrapper {
+	slw := &sugaredLogWrapper{
+		SugaredLogger: sl,
+	}
+
+	for _, option := range options {
+		option(slw)
+	}
+
+	return slw
+}
+
+func (s *sugaredLogWrapper) fork(sl *zap.SugaredLogger, options ...sugaredLogWrapperOption) *sugaredLogWrapper {
+	return newSugaredLogWrapper(sl, append([]sugaredLogWrapperOption{sugaredLogStreamer(s.streamer)}, options...)...)
 }
 
 func (s *sugaredLogWrapper) WithErr(err error) log.Logger {
-	return &sugaredLogWrapper{s.SugaredLogger.With("err", err.Error())}
+	return s.fork(s.SugaredLogger.With("err", err.Error()))
 }
 
 func (s *sugaredLogWrapper) With(args ...interface{}) log.Logger {
-	return &sugaredLogWrapper{s.SugaredLogger.With(args...)}
+	return s.fork(s.SugaredLogger.With(args...))
 }
 
 func (s *sugaredLogWrapper) Named(name string) log.Logger {
-	return &sugaredLogWrapper{s.SugaredLogger.Named(name)}
+	return s.fork(s.SugaredLogger.Named(name))
 }
 
-var _ log.Logger = (*sugaredLogWrapper)(nil)
+func (s *sugaredLogWrapper) Close() {
+	// this fails, but it should be safe to ignore according
+	// to https://github.com/uber-go/zap/issues/328
+	_ = s.SugaredLogger.Sync()
+}
+
+func (s *sugaredLogWrapper) Stream(readyCh <-chan struct{}) (<-chan event.Event, func()) {
+	return s.streamer.Stream(readyCh)
+}
 
 // Wrap zap.SugaredLogger as Logger interface
-func Wrap(z *zap.SugaredLogger) log.Logger {
-	return &sugaredLogWrapper{z}
+func Wrap(z *zap.SugaredLogger) log.LoggerCloser {
+	return newSugaredLogWrapper(z)
 }
 
 // NopLogger constructs a nop logger
@@ -70,13 +108,26 @@ func From(ctx context.Context) log.Logger {
 type InitOption func(config zap.Config) zap.Config
 
 // Init initializes a logger with options.
-func Init(logLevel int, options ...InitOption) (*zap.Logger, error) {
+func Init(logLevel int, options ...InitOption) (log.LoggerCloser, error) {
+	octantSink := NewOctantSink()
+
+	err := zap.RegisterSink("octant-stream", func(url *url.URL) (zap.Sink, error) {
+		return octantSink, nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("register octant log sink: %w", err)
+	}
+
 	z, err := newZapLogger(logLevel, options...)
 	if err != nil {
 		return nil, fmt.Errorf("create zap logger: %w", err)
 	}
 
-	return z, nil
+	logger := newSugaredLogWrapper(z.Sugar(), func(wrapper *sugaredLogWrapper) {
+		wrapper.streamer = NewStreamer(octantSink)
+	})
+
+	return logger, nil
 
 }
 
@@ -95,8 +146,8 @@ func newZapLogger(verboseLevel int, options ...InitOption) (*zap.Logger, error)
 		Development:      true,
 		Encoding:         "console",
 		EncoderConfig:    zap.NewDevelopmentEncoderConfig(),
-		OutputPaths:      []string{"stderr"},
-		ErrorOutputPaths: []string{"stderr"},
+		OutputPaths:      []string{"stderr", "octant-stream://output"},
+		ErrorOutputPaths: []string{"stderr", "octant-stream://output"},
 	}
 
 	for _, option := range options {
diff --git a/internal/log/message_id_generator.go b/internal/log/message_id_generator.go
new file mode 100644
index 00000000..cbb685e0
--- /dev/null
+++ b/internal/log/message_id_generator.go
@@ -0,0 +1,23 @@
+/*
+ * Copyright (c) 2020 the Octant contributors. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package log
+
+import "github.com/google/uuid"
+
+// MessageIDGenerator is an interface for generating message IDs.
+type MessageIDGenerator interface {
+	Generate() string
+}
+
+// UUIDMessageIDGenerator generates messages IDs as UUIDs.
+type UUIDMessageIDGenerator struct{}
+
+var _ MessageIDGenerator = &UUIDMessageIDGenerator{}
+
+// Generate generates a UUID as a string.
+func (d UUIDMessageIDGenerator) Generate() string {
+	return uuid.New().String()
+}
diff --git a/internal/log/options.go b/internal/log/options.go
new file mode 100644
index 00000000..049daeca
--- /dev/null
+++ b/internal/log/options.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2020 the Octant contributors. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package log
+
+type options struct {
+	messageIDGenerator MessageIDGenerator
+}
+
+func makeDefaultOptions(list ...Option) options {
+	opts := options{
+		messageIDGenerator: &UUIDMessageIDGenerator{},
+	}
+
+	for _, o := range list {
+		o(&opts)
+	}
+
+	return opts
+}
+
+// Option is a log option.
+type Option func(o *options)
+
+// WithIDGenerator sets the message id generator option.
+func WithIDGenerator(g MessageIDGenerator) Option {
+	return func(o *options) {
+		o.messageIDGenerator = g
+	}
+}
diff --git a/internal/log/sink.go b/internal/log/sink.go
index 57fe3a3a..3d245cf9 100644
--- a/internal/log/sink.go
+++ b/internal/log/sink.go
@@ -6,7 +6,6 @@
 package log
 
 import (
-	"errors"
 	"fmt"
 	"strings"
 	"sync"
@@ -18,16 +17,20 @@ import (
 
 // Message is an Octant log message.
 type Message struct {
+	// ID is the identifier for this log message.
+	ID string
 	// Date is the seconds since epoch.
-	Date int64
+	Date int64 `json:"date"`
 	// LogLevel is the log level.
-	LogLevel string
+	LogLevel string `json:"logLevel"`
 	// Location is the source location.
-	Location string
+	Location string `json:"location"`
 	// Text is the actual message.
-	Text string
+	Text string `json:"text"`
 	// JSON is the JSON payload.
-	JSON string
+	JSON string `json:"json"`
+	// StackTrace is an optional stack trace.
+	StackTrace string `json:"stackTrace"`
 }
 
 // ListenCancelFunc is a function for canceling a sink listener.
@@ -40,7 +43,7 @@ type OctantSinkOption func(o *OctantSink)
 // allows multiple loggers to listen to message.
 type OctantSink struct {
 	listeners map[string]chan Message
-	converter func(b []byte) (Message, error)
+	converter func(b []byte, options ...Option) (Message, error)
 
 	mu sync.RWMutex
 }
@@ -121,12 +124,15 @@ func (o *OctantSink) Listen() (<-chan Message, ListenCancelFunc) {
 }
 
 // ConvertBytesToMessage converts a zap message string to a Message instance.
-func ConvertBytesToMessage(b []byte) (Message, error) {
-	parts := strings.Split(strings.TrimSpace(string(b)), "\t")
+func ConvertBytesToMessage(b []byte, options ...Option) (Message, error) {
+	opts := makeDefaultOptions(options...)
+
+	lines := strings.SplitN(strings.TrimSpace(string(b)), "\n", 2)
+	parts := strings.SplitN(strings.TrimSpace(lines[0]), "\t", 5)
 	pLen := len(parts)
 
 	if pLen < 4 || pLen > 5 {
-		return Message{}, errors.New("unknown log message format")
+		return Message{}, fmt.Errorf("unknown log message format: %q", string(b))
 	}
 
 	t, err := time.Parse("2006-01-02T15:04:05.000Z0700", parts[0])
@@ -135,12 +141,17 @@ func ConvertBytesToMessage(b []byte) (Message, error) {
 	}
 
 	m := Message{
-		Date:     t.Unix(),
+		ID:       opts.messageIDGenerator.Generate(),
+		Date:     t.UnixNano(),
 		LogLevel: parts[1],
 		Location: parts[2],
 		Text:     parts[3],
 	}
 
+	if len(lines) > 1 {
+		m.StackTrace = strings.Join(lines[1:], "\n")
+	}
+
 	if pLen > 4 {
 		m.JSON = parts[4]
 	}
diff --git a/internal/log/streamer.go b/internal/log/streamer.go
new file mode 100644
index 00000000..bb2732bf
--- /dev/null
+++ b/internal/log/streamer.go
@@ -0,0 +1,137 @@
+/*
+ * Copyright (c) 2020 the Octant contributors. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package log
+
+import (
+	"container/ring"
+	"sort"
+	"sync"
+
+	"k8s.io/apimachinery/pkg/util/rand"
+
+	"github.com/vmware-tanzu/octant/pkg/event"
+)
+
+// MessageListenerFactory is a factory which generates message listeners.
+type MessageListenerFactory interface {
+	// Listen creates a listener.
+	Listen() (<-chan Message, ListenCancelFunc)
+}
+
+// EventStreamer is an interface for streaming.
+type EventStreamer interface {
+	// Stream streams events.
+	Stream(ready <-chan struct{}) (<-chan event.Event, func())
+	// Close closes the streamer. No more events will be generated
+	// one the streamer is closed.
+	Close()
+}
+
+// Streamer streams events using a ring buffer.
+type Streamer struct {
+	messageCh     <-chan Message
+	messageCancel ListenCancelFunc
+	r             *ring.Ring
+	eventCh       chan event.Event
+
+	mu        sync.RWMutex
+	listeners map[string]chan event.Event
+}
+
+var _ EventStreamer = &Streamer{}
+
+// NewStreamer creates an instance of Streamer.
+func NewStreamer(messageListenerFactory MessageListenerFactory) *Streamer {
+	messageCh, messageCancel := messageListenerFactory.Listen()
+
+	lm := &Streamer{
+		messageCh:     messageCh,
+		r:             ring.New(1500),
+		eventCh:       make(chan event.Event),
+		listeners:     make(map[string]chan event.Event),
+		messageCancel: messageCancel,
+	}
+
+	go func() {
+		for m := range messageCh {
+			lm.handle(m)
+		}
+	}()
+
+	return lm
+}
+
+// Close closes the streamer.
+func (lm *Streamer) Close() {
+	lm.messageCancel()
+}
+
+func (lm *Streamer) handle(m Message) {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	lm.r.Value = m
+	lm.r = lm.r.Next()
+
+	e := lm.createEvent()
+
+	for _, ch := range lm.listeners {
+		ch <- e
+	}
+
+}
+
+func (lm *Streamer) createEvent() event.Event {
+	var messages []Message
+	lm.r.Do(func(i interface{}) {
+		if i != nil {
+			m := i.(Message)
+			messages = append(messages, m)
+		}
+	})
+
+	sort.Slice(messages, func(i, j int) bool {
+		return messages[j].Date < messages[i].Date
+	})
+
+	e := event.Event{
+		Type: event.EventTypeAppLogs,
+		Data: messages,
+	}
+
+	return e
+}
+
+// Stream creates a channel which streams messages and a cancel function.
+func (lm *Streamer) Stream(readyCh <-chan struct{}) (<-chan event.Event, func()) {
+	lm.mu.Lock()
+	defer lm.mu.Unlock()
+
+	id := rand.String(6)
+	ch := make(chan event.Event, 10)
+
+	go func() {
+		lm.mu.RLock()
+		defer lm.mu.RUnlock()
+
+		if _, ok := lm.listeners[id]; !ok {
+			return
+		}
+
+		<-readyCh
+		ch <- lm.createEvent()
+	}()
+
+	lm.listeners[id] = ch
+
+	return ch, func() {
+		lm.mu.Lock()
+		defer lm.mu.Unlock()
+
+		close(ch)
+		delete(lm.listeners, id)
+	}
+}
diff --git a/pkg/event/event.go b/pkg/event/event.go
index 769ccadc..bc76fe90 100644
--- a/pkg/event/event.go
+++ b/pkg/event/event.go
@@ -9,8 +9,6 @@ package event
 
 import (
 	"fmt"
-
-	"github.com/vmware-tanzu/octant/pkg/action"
 )
 
 const (
@@ -59,6 +57,9 @@ const (
 	// EventTypeLoading is a loading event.
 	EventTypeLoading EventType = "event.octant.dev/loading"
 
+	// EventTypeAppLogs is an app logs event.
+	EventTypeAppLogs EventType = "event.octant.dev/app-logs"
+
 	// EventTypeTerminalFormat is a string with format specifiers to assist in generating
 	// a terminal event type.
 	EventTypeTerminalFormat string = "event.octant.dev/terminals/namespace/%s/pod/%s/container/%s"
@@ -89,7 +90,7 @@ type Event struct {
 	Err  error
 }
 
-func CreateEvent(eventType EventType, fields action.Payload) Event {
+func CreateEvent(eventType EventType, fields map[string]interface{}) Event {
 	return Event{
 		Type: eventType,
 		Data: fields,
diff --git a/pkg/log/logger.go b/pkg/log/logger.go
index a1f3c6d9..bc4c71db 100644
--- a/pkg/log/logger.go
+++ b/pkg/log/logger.go
@@ -5,6 +5,8 @@ SPDX-License-Identifier: Apache-2.0
 
 package log
 
+import "github.com/vmware-tanzu/octant/pkg/event"
+
 // Logger is an interface for logging
 type Logger interface {
 	// Debugf uses fmt.Sprintf to log a templated message.
@@ -24,4 +26,14 @@ type Logger interface {
 	WithErr(err error) Logger
 
 	Named(name string) Logger
+
+	Stream(readyCh <-chan struct{}) (<-chan event.Event, func())
+}
+
+// LoggerCloser is an interface that wraps a Logger and a close function.
+type LoggerCloser interface {
+	Logger
+
+	// Close closes the logger.
+	Close()
 }


Test patch:
diff --git a/internal/log/message_id_generator_test.go b/internal/log/message_id_generator_test.go
new file mode 100644
index 00000000..160c43f7
--- /dev/null
+++ b/internal/log/message_id_generator_test.go
@@ -0,0 +1,24 @@
+/*
+ * Copyright (c) 2020 the Octant contributors. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package log
+
+import (
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUUIDMessageIDGenerator_Generate(t *testing.T) {
+	gen := &UUIDMessageIDGenerator{}
+	got := gen.Generate()
+	require.True(t, isValidUUID(got))
+}
+
+func isValidUUID(u string) bool {
+	_, err := uuid.Parse(u)
+	return err == nil
+}
diff --git a/internal/log/sink_test.go b/internal/log/sink_test.go
index c6e5161d..35ca5562 100644
--- a/internal/log/sink_test.go
+++ b/internal/log/sink_test.go
@@ -16,11 +16,11 @@ import (
 )
 
 func TestOctantSink(t *testing.T) {
-	validConvert := func(b []byte) (Message, error) {
+	validConvert := func(b []byte, options ...Option) (Message, error) {
 		return Message{}, nil
 	}
 
-	invalidConvert := func(b []byte) (Message, error) {
+	invalidConvert := func(b []byte, options ...Option) (Message, error) {
 		return Message{}, errors.New("invalid")
 	}
 
@@ -85,6 +85,25 @@ func TestOctantSink(t *testing.T) {
 	}
 }
 
+func TestOctantSink_Close(t *testing.T) {
+	s := NewOctantSink()
+	ch, _ := s.Listen()
+
+	done := make(chan struct{}, 1)
+
+	closed := false
+	go func() {
+		<-ch
+		closed = true
+		done <- struct{}{}
+	}()
+
+	require.NoError(t, s.Close())
+
+	<-done
+	require.True(t, closed)
+}
+
 func TestConvertBytesToMessage(t *testing.T) {
 	type args struct {
 		bytes []byte
@@ -96,7 +115,7 @@ func TestConvertBytesToMessage(t *testing.T) {
 		wantErr bool
 	}{
 		{
-			name: "in general with JSON",
+			name: "info message with JSON",
 			args: args{
 				bytes: []byte(strings.Join([]string{
 					"2020-09-03T14:39:51.115-0400",
@@ -107,7 +126,8 @@ func TestConvertBytesToMessage(t *testing.T) {
 				}, "\t") + "\n"),
 			},
 			want: Message{
-				Date:     1599158391,
+				ID:       "12345",
+				Date:     1599158391115000000,
 				LogLevel: "INFO",
 				Location: "file.go:50",
 				Text:     "message",
@@ -115,7 +135,7 @@ func TestConvertBytesToMessage(t *testing.T) {
 			},
 		},
 		{
-			name: "in general without JSON",
+			name: "info message without JSON",
 			args: args{
 				bytes: []byte(strings.Join([]string{
 					"2020-09-03T14:39:51.115-0400",
@@ -125,12 +145,35 @@ func TestConvertBytesToMessage(t *testing.T) {
 				}, "\t") + "\n"),
 			},
 			want: Message{
-				Date:     1599158391,
+				ID:       "12345",
+				Date:     1599158391115000000,
 				LogLevel: "INFO",
 				Location: "file.go:50",
 				Text:     "message",
 			},
 		},
+		{
+			name: "error message with JSON",
+			args: args{
+				bytes: []byte("2020-09-23T09:10:52.181-0400\tERROR\tapi/content_manager.go:154\tgenerate content\t{\"client-id\": \"2c3d1670-fd9e-11ea-b34c-e450ebbc2e8c\", \"err\": \"generate content: calling contentHandler: check access to list CacheKey[Namespace='default', APIVersion='cluster.x-k8s.io/v1alpha3', Kind='Cluster']: unable to get resource for group kind Cluster.cluster.x-k8s.io: no matches for kind \\\"Cluster\\\" in group \\\"cluster.x-k8s.io\\\"\", \"content-path\": \"octant-clusterapi-plugin\"}\ngithub.com/vmware-tanzu/octant/internal/api.(*ContentManager).runUpdate.func1\n\t/Users/bryan/Development/projects/octant/internal/api/content_manager.go:154\ngithub.com/vmware-tanzu/octant/internal/api.(*InterruptiblePoller).Run.func1\n\t/Users/bryan/Development/projects/octant/internal/api/poller.go:86\ngithub.com/vmware-tanzu/octant/internal/api.(*InterruptiblePoller).Run\n\t/Users/bryan/Development/projects/octant/internal/api/poller.go:95\ngithub.com/vmware-tanzu/octant/internal/api.(*ContentManager).Start\n\t/Users/bryan/Development/projects/octant/internal/api/content_manager.go:128\n"),
+			},
+			want: Message{
+				ID:       "12345",
+				Date:     1600866652181000000,
+				LogLevel: "ERROR",
+				Location: "api/content_manager.go:154",
+				Text:     "generate content",
+				JSON:     "{\"client-id\": \"2c3d1670-fd9e-11ea-b34c-e450ebbc2e8c\", \"err\": \"generate content: calling contentHandler: check access to list CacheKey[Namespace='default', APIVersion='cluster.x-k8s.io/v1alpha3', Kind='Cluster']: unable to get resource for group kind Cluster.cluster.x-k8s.io: no matches for kind \\\"Cluster\\\" in group \\\"cluster.x-k8s.io\\\"\", \"content-path\": \"octant-clusterapi-plugin\"}",
+				StackTrace: "github.com/vmware-tanzu/octant/internal/api.(*ContentManager).runUpdate.func1\n" +
+					"\t/Users/bryan/Development/projects/octant/internal/api/content_manager.go:154\n" +
+					"github.com/vmware-tanzu/octant/internal/api.(*InterruptiblePoller).Run.func1\n" +
+					"\t/Users/bryan/Development/projects/octant/internal/api/poller.go:86\n" +
+					"github.com/vmware-tanzu/octant/internal/api.(*InterruptiblePoller).Run\n" +
+					"\t/Users/bryan/Development/projects/octant/internal/api/poller.go:95\n" +
+					"github.com/vmware-tanzu/octant/internal/api.(*ContentManager).Start\n" +
+					"\t/Users/bryan/Development/projects/octant/internal/api/content_manager.go:128",
+			},
+		},
 		{
 			name: "invalid format (too short)",
 			args: args{
@@ -169,10 +212,24 @@ func TestConvertBytesToMessage(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := ConvertBytesToMessage(tt.args.bytes)
+			got, err := ConvertBytesToMessage(tt.args.bytes, WithIDGenerator(newTestIDGen("12345")))
 			testutil.RequireErrorOrNot(t, tt.wantErr, err, func() {
 				require.Equal(t, tt.want, got)
 			})
 		})
 	}
 }
+
+type testIDGen struct {
+	id string
+}
+
+var _ MessageIDGenerator = &testIDGen{}
+
+func newTestIDGen(id string) *testIDGen {
+	return &testIDGen{id: id}
+}
+
+func (i testIDGen) Generate() string {
+	return i.id
+}
diff --git a/internal/log/streamer_test.go b/internal/log/streamer_test.go
new file mode 100644
index 00000000..d1bac3d9
--- /dev/null
+++ b/internal/log/streamer_test.go
@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2020 the Octant contributors. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package log
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/vmware-tanzu/octant/pkg/event"
+)
+
+func TestStreamer_Close(t *testing.T) {
+	f := newStubMessageListenerFactory()
+	s := NewStreamer(f)
+
+	s.Close()
+	require.True(t, f.isCanceled)
+}
+
+func TestStreamer_Stream(t *testing.T) {
+	tests := []struct {
+		name string
+	}{
+		{
+			name: "in general",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			message := Message{
+				ID: "1",
+			}
+
+			f := newStubMessageListenerFactory()
+			s := NewStreamer(f)
+
+			readyCh := make(chan struct{}, 1)
+			close(readyCh)
+
+			ch, cancel := s.Stream(readyCh)
+
+			done := make(chan struct{}, 1)
+			go func() {
+				e := <-ch
+				want := event.Event{
+					Type: event.EventTypeAppLogs,
+					Data: []Message{message},
+					Err:  nil,
+				}
+				require.Equal(t, want, e)
+				close(done)
+			}()
+
+			f.ch <- message
+
+			<-done
+			cancel()
+
+		})
+	}
+}
+
+type stubMessageListenerFactory struct {
+	ch         chan Message
+	cancel     func()
+	isCanceled bool
+}
+
+func newStubMessageListenerFactory() *stubMessageListenerFactory {
+	f := &stubMessageListenerFactory{
+		ch: make(chan Message, 1),
+	}
+
+	f.cancel = func() {
+		if f.isCanceled {
+			panic("already canceled")
+		}
+		f.isCanceled = true
+	}
+
+	return f
+}
+
+func (f *stubMessageListenerFactory) Listen() (<-chan Message, ListenCancelFunc) {
+	return f.ch, f.cancel
+}
+
+var _ MessageListenerFactory = &stubMessageListenerFactory{}
diff --git a/internal/log/testing.go b/internal/log/testing.go
deleted file mode 100644
index 4b166c2e..00000000
--- a/internal/log/testing.go
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
-Copyright (c) 2019 the Octant contributors. All Rights Reserved.
-SPDX-License-Identifier: Apache-2.0
-*/
-
-package log
-
-import (
-	"testing"
-
-	"github.com/vmware-tanzu/octant/pkg/log"
-)
-
-type testingLogger struct {
-	t *testing.T
-}
-
-// TestLogger returns a logger for tests
-func TestLogger(t *testing.T) log.Logger {
-	return &testingLogger{t: t}
-}
-
-func (t *testingLogger) Debugf(format string, args ...interface{}) {
-	t.t.Logf(format, args...)
-}
-func (t *testingLogger) Infof(format string, args ...interface{}) {
-	t.t.Logf(format, args...)
-}
-func (t *testingLogger) Warnf(format string, args ...interface{}) {
-	t.t.Logf(format, args...)
-}
-func (t *testingLogger) Errorf(format string, args ...interface{}) {
-	t.t.Errorf(format, args...)
-}
-func (t *testingLogger) With(args ...interface{}) log.Logger {
-	return t
-}
-func (t *testingLogger) WithErr(err error) log.Logger {
-	return t.With("err", err)
-}
-func (t *testingLogger) Name() string {
-	return t.Name()
-}
-func (t *testingLogger) Named(string) log.Logger {
-	return t
-}