feat: improved deploy progress reporting

See toolshed/abra#478

cli/app/logs.go

@@ -3,23 +3,14 @@ package app
 import (
 	"context"
 	"fmt"
-	"io"
-	"os"
-	"slices"
-	"sync"
-	"time"
 
 	"coopcloud.tech/abra/cli/internal"
 	appPkg "coopcloud.tech/abra/pkg/app"
 	"coopcloud.tech/abra/pkg/autocomplete"
 	"coopcloud.tech/abra/pkg/client"
 	"coopcloud.tech/abra/pkg/log"
+	"coopcloud.tech/abra/pkg/logs"
 	"coopcloud.tech/abra/pkg/upstream/stack"
-	"github.com/docker/docker/api/types"
-	containerTypes "github.com/docker/docker/api/types/container"
-	"github.com/docker/docker/api/types/filters"
-	"github.com/docker/docker/api/types/swarm"
-	dockerClient "github.com/docker/docker/client"
 	"github.com/spf13/cobra"
 )
 
@@ -73,80 +64,25 @@ var AppLogsCommand = &cobra.Command{
 			serviceNames = []string{args[1]}
 		}
 
-		if err = tailLogs(cl, app, serviceNames); err != nil {
+		f, err := app.Filters(true, false, serviceNames...)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		opts := logs.TailOpts{
+			AppName:  app.Name,
+			Services: serviceNames,
+			StdErr:   stdErr,
+			Since:    sinceLogs,
+			Filters:  f,
+		}
+
+		if err := logs.TailLogs(cl, opts); err != nil {
 			log.Fatal(err)
 		}
 	},
 }
 
-// tailLogs prints logs for the given app with optional service names to be
-// filtered on. It also checks if the latest task is not runnning and then
-// prints the past tasks.
-func tailLogs(cl *dockerClient.Client, app appPkg.App, serviceNames []string) error {
-	f, err := app.Filters(true, false, serviceNames...)
-	if err != nil {
-		return err
-	}
-
-	services, err := cl.ServiceList(context.Background(), types.ServiceListOptions{Filters: f})
-	if err != nil {
-		return err
-	}
-
-	var wg sync.WaitGroup
-	for _, service := range services {
-		filters := filters.NewArgs()
-		filters.Add("name", service.Spec.Name)
-		tasks, err := cl.TaskList(context.Background(), types.TaskListOptions{Filters: f})
-		if err != nil {
-			return err
-		}
-		if len(tasks) > 0 {
-			// Need to sort the tasks by the CreatedAt field in the inverse order.
-			// Otherwise they are in the reversed order and not sorted properly.
-			slices.SortFunc[[]swarm.Task](tasks, func(t1, t2 swarm.Task) int {
-				return int(t2.Meta.CreatedAt.Unix() - t1.Meta.CreatedAt.Unix())
-			})
-			lastTask := tasks[0].Status
-			if lastTask.State != swarm.TaskStateRunning {
-				for _, task := range tasks {
-					log.Errorf("[%s] %s State %s: %s", service.Spec.Name, task.Meta.CreatedAt.Format(time.RFC3339), task.Status.State, task.Status.Err)
-				}
-			}
-		}
-
-		// Collect the logs in a go routine, so the logs from all services are
-		// collected in parallel.
-		wg.Add(1)
-		go func(serviceID string) {
-			logs, err := cl.ServiceLogs(context.Background(), serviceID, containerTypes.LogsOptions{
-				ShowStderr: true,
-				ShowStdout: !stdErr,
-				Since:      sinceLogs,
-				Until:      "",
-				Timestamps: true,
-				Follow:     true,
-				Tail:       "20",
-				Details:    false,
-			})
-			if err != nil {
-				log.Fatal(err)
-			}
-			defer logs.Close()
-
-			_, err = io.Copy(os.Stdout, logs)
-			if err != nil && err != io.EOF {
-				log.Fatal(err)
-			}
-		}(service.ID)
-	}
-
-	// Wait for all log streams to be closed.
-	wg.Wait()
-
-	return nil
-}
-
 var (
 	stdErr    bool
 	sinceLogs string