Stream logs from nats

Try and stream log from nats and fallback to polling case of an error.

- [ ] Streaming from nats  currently blocks forever.
- [ ] Need to make a better unified format.
- [ ] Handle errors

cmd/deploy.go

@@ -449,8 +449,8 @@ func watchDeployment(ctx context.Context, cmdCtx *cmdctx.CmdContext) error {
 				}
 
 				cmdCtx.Status("deploy", cmdctx.STITLE, "Recent Logs")
-				logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
-				logPresenter.FPrint(cmdCtx.Out, cmdCtx.OutputJSON(), alloc.RecentLogs)
+				// logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
+				// logPresenter.FPrint(cmdCtx.Out, cmdCtx.OutputJSON(), alloc.RecentLogs)
 			}
 
 		}

cmd/logs.go

@@ -1,9 +1,11 @@
 package cmd
 
 import (
+	"github.com/superfly/flyctl/cmd/presenters"
 	"github.com/superfly/flyctl/cmdctx"
 	"github.com/superfly/flyctl/internal/client"
-	"github.com/superfly/flyctl/internal/monitor"
+	"github.com/superfly/flyctl/pkg/logs"
+	"github.com/superfly/flyctl/terminal"
 
 	"github.com/superfly/flyctl/docstrings"
 )
@@ -27,12 +29,34 @@ func newLogsCommand(client *client.Client) *Command {
 	return cmd
 }
 
-func runLogs(ctx *cmdctx.CmdContext) error {
-	err := monitor.WatchLogs(ctx, ctx.Out, monitor.LogOptions{
-		AppName:    ctx.AppName,
-		VMID:       ctx.Config.GetString("instance"),
-		RegionCode: ctx.Config.GetString("region"),
-	})
+func runLogs(cc *cmdctx.CmdContext) error {
+	ctx := createCancellableContext()
+
+	client := cc.Client.API()
+
+	opts := &logs.LogOptions{
+		AppName:    cc.Config.GetString("app"),
+		RegionCode: cc.Config.GetString("region"),
+		VMID:       cc.Config.GetString("instance"),
+	}
+
+	stream, err := logs.NewNatsStream(client, opts)
+
+	if err != nil {
+		terminal.Debugf("could not connect to wireguard tunnel, err: %v\n", err)
+		terminal.Debug("Falling back to log polling...")
+
+		stream, err = logs.NewPollingStream(client)
+		if err != nil {
+			return err
+		}
+	}
+
+	presenter := presenters.LogPresenter{}
+
+	for entry := range stream.Stream(ctx, opts) {
+		presenter.FPrint(cc.Out, false, entry)
+	}
 
-	return err
+	return nil
 }

cmd/monitor.go

@@ -112,8 +112,8 @@ func monitorDeployment(ctx context.Context, commandContext *cmdctx.CmdContext) e
 				}
 
 				commandContext.Status("monitor", cmdctx.STITLE, "Recent Logs")
-				logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
-				logPresenter.FPrint(commandContext.Out, commandContext.OutputJSON(), alloc.RecentLogs)
+				// logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
+				// logPresenter.FPrint(commandContext.Out, commandContext.OutputJSON(), alloc.RecentLogs)
 
 			}
 

cmd/presenters/logs.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/logrusorgru/aurora"
-	"github.com/superfly/flyctl/api"
+	"github.com/superfly/flyctl/pkg/logs"
 )
 
 type LogPresenter struct {
@@ -16,16 +16,14 @@ type LogPresenter struct {
 	HideAllocID    bool
 }
 
-func (lp *LogPresenter) FPrint(w io.Writer, asJSON bool, entries []api.LogEntry) {
-	for _, entry := range entries {
-		lp.printEntry(w, asJSON, entry)
-	}
+func (lp *LogPresenter) FPrint(w io.Writer, asJSON bool, entry logs.LogEntry) {
+	lp.printEntry(w, asJSON, entry)
 }
 
 var newLineReplacer = strings.NewReplacer("\r\n", aurora.Faint("↩︎").String(), "\n", aurora.Faint("↩︎").String())
 var newline = []byte("\n")
 
-func (lp *LogPresenter) printEntry(w io.Writer, asJSON bool, entry api.LogEntry) {
+func (lp *LogPresenter) printEntry(w io.Writer, asJSON bool, entry logs.LogEntry) {
 	if asJSON {
 		outBuf, _ := json.MarshalIndent(entry, "", "    ")
 		fmt.Fprintln(w, string(outBuf))

cmd/status.go

@@ -199,8 +199,8 @@ func runAllocStatus(ctx *cmdctx.CmdContext) error {
 		p = ctx.Out
 	}
 
-	logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
-	logPresenter.FPrint(p, ctx.OutputJSON(), alloc.RecentLogs)
+	// logPresenter := presenters.LogPresenter{HideAllocID: true, HideRegion: true, RemoveNewlines: true}
+	// logPresenter.FPrint(p, ctx.OutputJSON(), alloc.RecentLogs)
 
 	if p != ctx.Out {
 		_ = pw.Flush()

internal/monitor/logs.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/jpillora/backoff"
 	"github.com/superfly/flyctl/api"
-	"github.com/superfly/flyctl/cmd/presenters"
 	"github.com/superfly/flyctl/cmdctx"
 	"github.com/superfly/flyctl/terminal"
 )
@@ -37,7 +36,7 @@ func WatchLogs(cc *cmdctx.CmdContext, w io.Writer, opts LogOptions) error {
 
 	nextToken := ""
 
-	logPresenter := presenters.LogPresenter{}
+	// logPresenter := presenters.LogPresenter{}
 
 	for {
 		entries, token, err := cc.Client.API().GetAppLogs(opts.AppName, nextToken, opts.RegionCode, opts.VMID)
@@ -63,7 +62,7 @@ func WatchLogs(cc *cmdctx.CmdContext, w io.Writer, opts LogOptions) error {
 		} else {
 			b.Reset()
 
-			logPresenter.FPrint(w, false, entries)
+			// logPresenter.FPrint(w, false, entries)
 
 			if token != "" {
 				nextToken = token

pkg/agent/start.go

@@ -35,7 +35,7 @@ func StartDaemon(ctx context.Context, api *api.Client, command string) (*Client,
 		return nil, err
 	}
 	agentPid := cmd.Process.Pid
-	terminal.Debugf("started agent process %d", agentPid)
+	terminal.Debug("started agent process ", agentPid)
 
 	// read stdout and stderr from the daemon process. If it
 	// includes "[pid] OK" we know it started successfully, and