add StreamLogs() to collector interface

connector/collector/docker.go

@@ -1,6 +1,9 @@
 package collector
 
 import (
+	"bufio"
+	"io"
+
 	"github.com/bcicen/ctop/models"
 	api "github.com/fsouza/go-dockerclient"
 )
@@ -65,6 +68,31 @@ func (c *Docker) Stream() chan models.Metrics {
 	return c.stream
 }
 
+func (c *Docker) StreamLogs() (chan string, error) {
+	r, w := io.Pipe()
+	logCh := make(chan string)
+
+	opts := api.LogsOptions{
+		Container:    c.id,
+		OutputStream: w,
+		ErrorStream:  w,
+		Stdout:       true,
+		Stderr:       true,
+		Tail:         "10",
+		Follow:       true,
+		Timestamps:   true,
+	}
+
+	go tailLogs(r, logCh)
+	go func() {
+		err := c.client.Logs(opts)
+		if err != nil {
+			log.Errorf("error reading container logs: %s", err)
+		}
+	}()
+	return logCh, nil
+}
+
 // Stop collector
 func (c *Docker) Stop() {
 	c.done <- true
@@ -111,3 +139,10 @@ func (c *Docker) ReadIO(stats *api.Stats) {
 	}
 	c.IOBytesRead, c.IOBytesWrite = read, write
 }
+
+func tailLogs(reader io.Reader, ch chan string) {
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		ch <- scanner.Text()
+	}
+}

connector/collector/main.go

@@ -4,10 +4,19 @@ import (
 	"math"
 
 	"github.com/bcicen/ctop/logging"
+	"github.com/bcicen/ctop/models"
 )
 
 var log = logging.Init()
 
+type Collector interface {
+	Stream() chan models.Metrics
+	StreamLogs() (chan string, error)
+	Running() bool
+	Start()
+	Stop()
+}
+
 func round(num float64) int {
 	return int(num + math.Copysign(0.5, num))
 }

connector/collector/mock.go

@@ -9,6 +9,8 @@ import (
 	"github.com/bcicen/ctop/models"
 )
 
+const mockLog = "Cura ob pro qui tibi inveni dum qua fit donec amare illic mea, regem falli contexo pro peregrinorum heremo absconditi araneae meminerim deliciosas actionibus facere modico dura sonuerunt psalmi contra rerum, tempus mala anima volebant dura quae o modis."
+
 // Mock collector
 type Mock struct {
 	models.Metrics
@@ -45,6 +47,17 @@ func (c *Mock) Stream() chan models.Metrics {
 	return c.stream
 }
 
+func (c *Mock) StreamLogs() (chan string, error) {
+	logCh := make(chan string)
+	go func() {
+		for {
+			logCh <- mockLog
+			time.Sleep(250 * time.Millisecond)
+		}
+	}()
+	return logCh, nil
+}
+
 func (c *Mock) run() {
 	c.running = true
 	rand.Seed(int64(time.Now().Nanosecond()))

connector/collector/runc.go

@@ -3,6 +3,7 @@
 package collector
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/bcicen/ctop/models"
@@ -51,6 +52,10 @@ func (c *Runc) Stream() chan models.Metrics {
 	return c.stream
 }
 
+func (c *Runc) StreamLogs() (chan string, error) {
+	return nil, fmt.Errorf("log streaming unavailable for runc collector")
+}
+
 func (c *Runc) run() {
 	c.running = true
 	defer close(c.stream)