Move stream to interface

debug/log/default.go

@@ -33,13 +33,14 @@ func NewLog(opts ...Option) Log {
 }
 
 // Write writes logs into logger
-func (l *defaultLog) Write(r Record) {
+func (l *defaultLog) Write(r Record) error {
 	golog.Print(r.Value)
 	l.Buffer.Put(fmt.Sprint(r.Value))
+	return nil
 }
 
 // Read reads logs and returns them
-func (l *defaultLog) Read(opts ...ReadOption) []Record {
+func (l *defaultLog) Read(opts ...ReadOption) ([]Record, error) {
 	options := ReadOptions{}
 	// initialize the read options
 	for _, o := range opts {
@@ -78,12 +79,12 @@ func (l *defaultLog) Read(opts ...ReadOption) []Record {
 		records = append(records, record)
 	}
 
-	return records
+	return records, nil
 }
 
 // Stream returns channel for reading log records
 // along with a stop channel, close it when done
-func (l *defaultLog) Stream() (<-chan Record, chan bool) {
+func (l *defaultLog) Stream() (Stream, error) {
 	// get stream channel from ring buffer
 	stream, stop := l.Buffer.Stream()
 	// make a buffered channel
@@ -111,5 +112,8 @@ func (l *defaultLog) Stream() (<-chan Record, chan bool) {
 		}
 	}()
 
-	return records, stop
+	return &logStream{
+		stream: records,
+		stop:   stop,
+	}, nil
 }

debug/log/default_test.go

@@ -23,7 +23,7 @@ func TestLogger(t *testing.T) {
 
 	// Check if the logs are stored in the logger ring buffer
 	expected := []string{"foobar", "foo bar"}
-	entries := DefaultLog.Read(Count(len(expected)))
+	entries, _ := DefaultLog.Read(Count(len(expected)))
 	for i, entry := range entries {
 		if !reflect.DeepEqual(entry.Value, expected[i]) {
 			t.Errorf("expected %s, got %s", expected[i], entry.Value)

debug/log/level.go

@@ -0,0 +1,35 @@
+// Package log provides debug logging
+package log
+
+import (
+	"os"
+)
+
+// level is a log level
+type Level int
+
+const (
+	LevelFatal Level = iota
+	LevelError
+	LevelInfo
+	LevelWarn
+	LevelDebug
+	LevelTrace
+)
+
+func init() {
+	switch os.Getenv("MICRO_LOG_LEVEL") {
+	case "trace":
+		DefaultLevel = LevelTrace
+	case "debug":
+		DefaultLevel = LevelDebug
+	case "warn":
+		DefaultLevel = LevelWarn
+	case "info":
+		DefaultLevel = LevelInfo
+	case "error":
+		DefaultLevel = LevelError
+	case "fatal":
+		DefaultLevel = LevelFatal
+	}
+}

debug/log/log.go

@@ -2,8 +2,6 @@
 package log
 
 import (
-	"fmt"
-	"os"
 	"time"
 )
 
@@ -19,11 +17,11 @@ var (
 // Log is event log
 type Log interface {
 	// Read reads log entries from the logger
-	Read(...ReadOption) []Record
+	Read(...ReadOption) ([]Record, error)
 	// Write writes records to log
-	Write(Record)
+	Write(Record) error
 	// Stream log records
-	Stream() (<-chan Record, chan bool)
+	Stream() (Stream, error)
 }
 
 // Record is log record entry
@@ -36,144 +34,7 @@ type Record struct {
 	Metadata map[string]string
 }
 
-// level is a log level
-type Level int
-
-const (
-	LevelFatal Level = iota
-	LevelError
-	LevelInfo
-	LevelWarn
-	LevelDebug
-	LevelTrace
-)
-
-func init() {
-	switch os.Getenv("MICRO_LOG_LEVEL") {
-	case "trace":
-		DefaultLevel = LevelTrace
-	case "debug":
-		DefaultLevel = LevelDebug
-	case "warn":
-		DefaultLevel = LevelWarn
-	case "info":
-		DefaultLevel = LevelInfo
-	case "error":
-		DefaultLevel = LevelError
-	case "fatal":
-		DefaultLevel = LevelFatal
-	}
-}
-
-func log(v ...interface{}) {
-	if len(prefix) > 0 {
-		DefaultLog.Write(Record{Value: fmt.Sprint(append([]interface{}{prefix, " "}, v...)...)})
-		return
-	}
-	DefaultLog.Write(Record{Value: fmt.Sprint(v...)})
-}
-
-func logf(format string, v ...interface{}) {
-	if len(prefix) > 0 {
-		format = prefix + " " + format
-	}
-	DefaultLog.Write(Record{Value: fmt.Sprintf(format, v...)})
-}
-
-// WithLevel logs with the level specified
-func WithLevel(l Level, v ...interface{}) {
-	if l > DefaultLevel {
-		return
-	}
-	log(v...)
-}
-
-// WithLevel logs with the level specified
-func WithLevelf(l Level, format string, v ...interface{}) {
-	if l > DefaultLevel {
-		return
-	}
-	logf(format, v...)
-}
-
-// Trace provides trace level logging
-func Trace(v ...interface{}) {
-	WithLevel(LevelTrace, v...)
-}
-
-// Tracef provides trace level logging
-func Tracef(format string, v ...interface{}) {
-	WithLevelf(LevelTrace, format, v...)
-}
-
-// Debug provides debug level logging
-func Debug(v ...interface{}) {
-	WithLevel(LevelDebug, v...)
-}
-
-// Debugf provides debug level logging
-func Debugf(format string, v ...interface{}) {
-	WithLevelf(LevelDebug, format, v...)
-}
-
-// Warn provides warn level logging
-func Warn(v ...interface{}) {
-	WithLevel(LevelWarn, v...)
-}
-
-// Warnf provides warn level logging
-func Warnf(format string, v ...interface{}) {
-	WithLevelf(LevelWarn, format, v...)
-}
-
-// Info provides info level logging
-func Info(v ...interface{}) {
-	WithLevel(LevelInfo, v...)
-}
-
-// Infof provides info level logging
-func Infof(format string, v ...interface{}) {
-	WithLevelf(LevelInfo, format, v...)
-}
-
-// Error provides warn level logging
-func Error(v ...interface{}) {
-	WithLevel(LevelError, v...)
-}
-
-// Errorf provides warn level logging
-func Errorf(format string, v ...interface{}) {
-	WithLevelf(LevelError, format, v...)
-}
-
-// Fatal logs with Log and then exits with os.Exit(1)
-func Fatal(v ...interface{}) {
-	WithLevel(LevelFatal, v...)
-	os.Exit(1)
-}
-
-// Fatalf logs with Logf and then exits with os.Exit(1)
-func Fatalf(format string, v ...interface{}) {
-	WithLevelf(LevelFatal, format, v...)
-	os.Exit(1)
-}
-
-// SetLevel sets the log level
-func SetLevel(l Level) {
-	DefaultLevel = l
-}
-
-// GetLevel returns the current level
-func GetLevel() Level {
-	return DefaultLevel
-}
-
-// Set a prefix for the logger
-func SetPrefix(p string) {
-	prefix = p
-}
-
-// Set service name
-func SetName(name string) {
-	prefix = fmt.Sprintf("[%s]", name)
+type Stream interface {
+	Chan() <-chan Record
+	Stop() error
 }

debug/log/logger.go

@@ -0,0 +1,120 @@
+// Package log provides debug logging
+package log
+
+import (
+	"fmt"
+	"os"
+)
+
+func log(v ...interface{}) {
+	if len(prefix) > 0 {
+		DefaultLog.Write(Record{Value: fmt.Sprint(append([]interface{}{prefix, " "}, v...)...)})
+		return
+	}
+	DefaultLog.Write(Record{Value: fmt.Sprint(v...)})
+}
+
+func logf(format string, v ...interface{}) {
+	if len(prefix) > 0 {
+		format = prefix + " " + format
+	}
+	DefaultLog.Write(Record{Value: fmt.Sprintf(format, v...)})
+}
+
+// WithLevel logs with the level specified
+func WithLevel(l Level, v ...interface{}) {
+	if l > DefaultLevel {
+		return
+	}
+	log(v...)
+}
+
+// WithLevel logs with the level specified
+func WithLevelf(l Level, format string, v ...interface{}) {
+	if l > DefaultLevel {
+		return
+	}
+	logf(format, v...)
+}
+
+// Trace provides trace level logging
+func Trace(v ...interface{}) {
+	WithLevel(LevelTrace, v...)
+}
+
+// Tracef provides trace level logging
+func Tracef(format string, v ...interface{}) {
+	WithLevelf(LevelTrace, format, v...)
+}
+
+// Debug provides debug level logging
+func Debug(v ...interface{}) {
+	WithLevel(LevelDebug, v...)
+}
+
+// Debugf provides debug level logging
+func Debugf(format string, v ...interface{}) {
+	WithLevelf(LevelDebug, format, v...)
+}
+
+// Warn provides warn level logging
+func Warn(v ...interface{}) {
+	WithLevel(LevelWarn, v...)
+}
+
+// Warnf provides warn level logging
+func Warnf(format string, v ...interface{}) {
+	WithLevelf(LevelWarn, format, v...)
+}
+
+// Info provides info level logging
+func Info(v ...interface{}) {
+	WithLevel(LevelInfo, v...)
+}
+
+// Infof provides info level logging
+func Infof(format string, v ...interface{}) {
+	WithLevelf(LevelInfo, format, v...)
+}
+
+// Error provides warn level logging
+func Error(v ...interface{}) {
+	WithLevel(LevelError, v...)
+}
+
+// Errorf provides warn level logging
+func Errorf(format string, v ...interface{}) {
+	WithLevelf(LevelError, format, v...)
+}
+
+// Fatal logs with Log and then exits with os.Exit(1)
+func Fatal(v ...interface{}) {
+	WithLevel(LevelFatal, v...)
+	os.Exit(1)
+}
+
+// Fatalf logs with Logf and then exits with os.Exit(1)
+func Fatalf(format string, v ...interface{}) {
+	WithLevelf(LevelFatal, format, v...)
+	os.Exit(1)
+}
+
+// SetLevel sets the log level
+func SetLevel(l Level) {
+	DefaultLevel = l
+}
+
+// GetLevel returns the current level
+func GetLevel() Level {
+	return DefaultLevel
+}
+
+// Set a prefix for the logger
+func SetPrefix(p string) {
+	prefix = p
+}
+
+// Set service name
+func SetName(name string) {
+	prefix = fmt.Sprintf("[%s]", name)
+}

debug/log/options.go

@@ -60,10 +60,3 @@ func Count(c int) ReadOption {
 		o.Count = c
 	}
 }
-
-// Stream requests continuous log stream
-func Stream(s bool) ReadOption {
-	return func(o *ReadOptions) {
-		o.Stream = s
-	}
-}

debug/log/stream.go

@@ -0,0 +1,20 @@
+package log
+
+type logStream struct {
+	stream <-chan Record
+	stop   chan bool
+}
+
+func (l *logStream) Chan() <-chan Record {
+	return l.stream
+}
+
+func (l *logStream) Stop() error {
+	select {
+	case <-l.stop:
+		return nil
+	default:
+		close(l.stop)
+	}
+	return nil
+}

debug/service/handler/debug.go

@@ -71,10 +71,13 @@ func (d *Debug) Log(ctx context.Context, stream server.Stream) error {
 		// the connection stays open until some timeout expires
 		// or something like that; that means the map of streams
 		// might end up leaking memory if not cleaned up properly
-		records, stop := d.log.Stream()
-		defer close(stop)
+		lgStream, err := d.log.Stream()
+		if err != nil {
+			return err
+		}
+		defer lgStream.Stop()
 
-		for record := range records {
+		for record := range lgStream.Chan() {
 			if err := d.sendRecord(record, stream); err != nil {
 				return err
 			}
@@ -85,7 +88,10 @@ func (d *Debug) Log(ctx context.Context, stream server.Stream) error {
 	}
 
 	// get the log records
-	records := d.log.Read(options...)
+	records, err := d.log.Read(options...)
+	if err != nil {
+		return err
+	}
 
 	// send all the logs downstream
 	for _, record := range records {

debug/service/log.go

@@ -1,6 +1,8 @@
 package service
 
 import (
+	"time"
+
 	"github.com/micro/go-micro/debug"
 	"github.com/micro/go-micro/debug/log"
 )
@@ -10,50 +12,36 @@ type serviceLog struct {
 }
 
 // Read reads log entries from the logger
-func (s *serviceLog) Read(opts ...log.ReadOption) []log.Record {
-	// TODO: parse opts
-	stream, err := s.Client.Log(opts...)
-	if err != nil {
-		return nil
+func (s *serviceLog) Read(opts ...log.ReadOption) ([]log.Record, error) {
+	var options log.ReadOptions
+	for _, o := range opts {
+		o(&options)
 	}
+
+	stream, err := s.Client.Log(options.Since, options.Count, false)
+	if err != nil {
+		return nil, err
+	}
+	defer stream.Stop()
+
 	// stream the records until nothing is left
 	var records []log.Record
-	for record := range stream {
+
+	for record := range stream.Chan() {
 		records = append(records, record)
 	}
-	return records
+
+	return records, nil
 }
 
 // There is no write support
-func (s *serviceLog) Write(r log.Record) {
-	return
+func (s *serviceLog) Write(r log.Record) error {
+	return nil
 }
 
 // Stream log records
-func (s *serviceLog) Stream() (<-chan log.Record, chan bool) {
-	stop := make(chan bool)
-	stream, err := s.Client.Log(log.Stream(true))
-	if err != nil {
-		// return a closed stream
-		deadStream := make(chan log.Record)
-		close(deadStream)
-		return deadStream, stop
-	}
-
-	newStream := make(chan log.Record, 128)
-
-	go func() {
-		for {
-			select {
-			case rec := <-stream:
-				newStream <- rec
-			case <-stop:
-				return
-			}
-		}
-	}()
-
-	return newStream, stop
+func (s *serviceLog) Stream() (log.Stream, error) {
+	return s.Client.Log(time.Time{}, 0, true)
 }
 
 // NewLog returns a new log interface

debug/service/service.go

@@ -27,42 +27,40 @@ func NewClient(name string) *debugClient {
 	}
 }
 
-// Logs queries the service logs and returns a channel to read the logs from
-func (d *debugClient) Log(opts ...log.ReadOption) (<-chan log.Record, error) {
-	var options log.ReadOptions
-	// initialize the read options
-	for _, o := range opts {
-		o(&options)
-	}
-
+// Logs queries the services logs and returns a channel to read the logs from
+func (d *debugClient) Log(since time.Time, count int, stream bool) (log.Stream, error) {
 	req := &pb.LogRequest{}
-	if !options.Since.IsZero() {
-		req.Since = options.Since.Unix()
+	if !since.IsZero() {
+		req.Since = since.Unix()
 	}
 
-	if options.Count > 0 {
-		req.Count = int64(options.Count)
+	if count > 0 {
+		req.Count = int64(count)
 	}
 
-	req.Stream = options.Stream
+	// set whether to stream
+	req.Stream = stream
 
 	// get the log stream
-	stream, err := d.Client.Log(context.Background(), req)
+	serverStream, err := d.Client.Log(context.Background(), req)
 	if err != nil {
 		return nil, fmt.Errorf("failed getting log stream: %s", err)
 	}
 
-	// log channel for streaming logs
-	logChan := make(chan log.Record)
-
-	// go stream logs
-	go d.streamLogs(logChan, stream)
-
-	return logChan, nil
+	lg := &logStream{
+		stream: make(chan log.Record),
+		stop:   make(chan bool),
 	}
 
-func (d *debugClient) streamLogs(logChan chan log.Record, stream pb.Debug_LogService) {
+	// go stream logs
+	go d.streamLogs(lg, serverStream)
+
+	return lg, nil
+}
+
+func (d *debugClient) streamLogs(lg *logStream, stream pb.Debug_LogService) {
 	defer stream.Close()
+	defer lg.Stop()
 
 	for {
 		resp, err := stream.Recv()
@@ -81,8 +79,10 @@ func (d *debugClient) streamLogs(logChan chan log.Record, stream pb.Debug_LogSer
 			Metadata:  metadata,
 		}
 
-		logChan <- record
+		select {
+		case <-lg.stop:
+			return
+		case lg.stream <- record:
+		}
 	}
-
-	close(logChan)
 }

debug/service/stream.go

@@ -0,0 +1,25 @@
+package service
+
+import (
+	"github.com/micro/go-micro/debug/log"
+)
+
+type logStream struct {
+	stream chan log.Record
+	stop   chan bool
+}
+
+func (l *logStream) Chan() <-chan log.Record {
+	return l.stream
+}
+
+func (l *logStream) Stop() error {
+	select {
+	case <-l.stop:
+		return nil
+	default:
+		close(l.stream)
+		close(l.stop)
+	}
+	return nil
+}