Merge pull request #20 from datacol-io/19_streaming_logs

adding streaming logs for cli

api/controller/services.proto

@@ -265,7 +265,7 @@ service ProviderService {
     };
   }
 
-  rpc LogStream(LogStreamReq) returns (LogStreamResponse) {
+  rpc LogStream(LogStreamReq) returns (stream LogStreamResponse) {
     option (google.api.http) = {
       get: "/v1/logs/{name}"
     };

api/controller/services.swagger.json

@@ -340,7 +340,7 @@
         "operationId": "LogStream",
         "responses": {
           "200": {
-            "description": "",
+            "description": "(streaming responses)",
             "schema": {
               "$ref": "#/definitions/controllerLogStreamResponse"
             }

api/server.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"bytes"
 	"cloud.google.com/go/compute/metadata"
 	"fmt"
 	pbs "github.com/dinesh/datacol/api/controller"
@@ -298,21 +297,35 @@ func (s *Server) ResourceUnlink(ctx context.Context, req *pbs.AppResourceReq) (*
 	return ret, nil
 }
 
-func (s *Server) LogStream(ctx context.Context, req *pbs.LogStreamReq) (*pbs.LogStreamResponse, error) {
-	buf := new(bytes.Buffer)
+func (s *Server) LogStream(req *pbs.LogStreamReq, stream pbs.ProviderService_LogStreamServer) error {
 	since, err := ptypes.Duration(req.Since)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	if err := s.Provider.LogStream(req.Name, buf, pb.LogStreamOptions{
+	reader, close, err := s.Provider.LogStream(req.Name, pb.LogStreamOptions{
 		Follow: req.Follow,
 		Since:  since,
-	}); err != nil {
-		return nil, err
+	})
+
+	if err != nil {
+		return err
 	}
+	defer close()
+
+	for {
+		line, err := reader.ReadBytes('\n')
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return err
+		}
 
-	return &pbs.LogStreamResponse{Data: buf.Bytes()}, nil
+		if err := stream.Send(&pbs.LogStreamResponse{Data: line}); err != nil {
+			return err
+		}
+	}
 }
 
 func (s *Server) unaryInterceptor(

client/apps.go

@@ -1,7 +1,6 @@
 package client
 
 import (
-	"bytes"
 	pbs "github.com/dinesh/datacol/api/controller"
 	pb "github.com/dinesh/datacol/api/models"
 	"github.com/golang/protobuf/ptypes"
@@ -36,7 +35,7 @@ func (c *Client) RestartApp(name string) error {
 }
 
 func (c *Client) StreamAppLogs(name string, follow bool, since time.Duration, out io.Writer) error {
-	ret, err := c.ProviderServiceClient.LogStream(ctx, &pbs.LogStreamReq{
+	stream, err := c.ProviderServiceClient.LogStream(ctx, &pbs.LogStreamReq{
 		Name:   name,
 		Since:  ptypes.DurationProto(since),
 		Follow: follow,
@@ -45,11 +44,20 @@ func (c *Client) StreamAppLogs(name string, follow bool, since time.Duration, ou
 		return err
 	}
 
-	if _, err = io.Copy(out, bytes.NewBuffer(ret.Data)); err != nil {
-		return err
-	}
+	defer stream.CloseSend()
 
-	return nil
+	for {
+		ret, err := stream.Recv()
+		if err != nil {
+			if err == io.EOF {
+				return nil
+			}
+			return err
+		}
+		if _, err := out.Write(ret.Data); err != nil {
+			return err
+		}
+	}
 }
 
 func (c *Client) RunProcess(name string, args []string) (*pbs.CmdResponse, error) {

cloud/google/provider.go

@@ -1,6 +1,7 @@
 package google
 
 import (
+	"bufio"
 	"cloud.google.com/go/datastore"
 	"context"
 	"fmt"
@@ -98,16 +99,16 @@ func runningPods(ns, app string, c *kubernetes.Clientset) (string, error) {
 	return podNames[0], nil
 }
 
-func (g *GCPCloud) LogStream(app string, out io.Writer, opts pb.LogStreamOptions) error {
+func (g *GCPCloud) LogStream(app string, opts pb.LogStreamOptions) (*bufio.Reader, func() error, error) {
 	ns := g.DeploymentName
 	c, err := getKubeClientset(ns)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 
 	pod, err := runningPods(ns, app, c)
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 
 	log.Debugf("Getting logs from pod %s", pod)
@@ -127,12 +128,10 @@ func (g *GCPCloud) LogStream(app string, out io.Writer, opts pb.LogStreamOptions
 
 	rc, err := req.Stream()
 	if err != nil {
-		return err
+		return nil, nil, err
 	}
 
-	defer rc.Close()
-	_, err = io.Copy(out, rc)
-	return err
+	return bufio.NewReader(rc), rc.Close, nil
 }
 
 func (g *GCPCloud) storage() *storage.Service {

cloud/provider.go

@@ -1,6 +1,7 @@
 package cloud
 
 import (
+	"bufio"
 	pb "github.com/dinesh/datacol/api/models"
 	"io"
 )
@@ -25,7 +26,7 @@ type Provider interface {
 	ReleaseList(string, int) (pb.Releases, error)
 	ReleaseDelete(string, string) error
 
-	LogStream(app string, w io.Writer, opts pb.LogStreamOptions) error
+	LogStream(app string, opts pb.LogStreamOptions) (*bufio.Reader, func() error, error)
 
 	ResourceList() (pb.Resources, error)
 	ResourceCreate(name, kind string, params map[string]string) (*pb.Resource, error)

cmd/logs.go

@@ -14,7 +14,8 @@ func init() {
 		Action: cmdAppLogStream,
 		Flags: []cli.Flag{
 			&cli.BoolFlag{
-				Name:  "follow, f",
+				Name:  "follow",
+				Aliases: []string{"f"},
 				Usage: "keep streaming new log output (default)",
 			},
 			&cli.DurationFlag{