wip

Author: lionello

src/cmd/cli/command/compose.go

@@ -423,6 +423,7 @@ func makeComposeLogsCmd() *cobra.Command {
 			var utc, _ = cmd.Flags().GetBool("utc")
 			var verbose, _ = cmd.Flags().GetBool("verbose")
 			var filter, _ = cmd.Flags().GetString("filter")
+			var until, _ = cmd.Flags().GetString("until")
 
 			if !cmd.Flags().Changed("verbose") {
 				verbose = true // default verbose for explicit tail command
@@ -432,17 +433,27 @@ func makeComposeLogsCmd() *cobra.Command {
 				os.Setenv("TZ", "") // used by Go's "time" package, see https://pkg.go.dev/time#Location
 			}
 
-			ts, err := cli.ParseTimeOrDuration(since, time.Now())
+			now := time.Now()
+			sinceTs, err := cli.ParseTimeOrDuration(since, now)
 			if err != nil {
-				return fmt.Errorf("invalid duration or time: %w", err)
+				return fmt.Errorf("invalid 'since' duration or time: %w", err)
 			}
+			sinceTs = sinceTs.UTC()
+			untilTs, err := cli.ParseTimeOrDuration(until, now)
+			if err != nil {
+				return fmt.Errorf("invalid 'until' duration or time: %w", err)
+			}
+			untilTs = untilTs.UTC()
 
-			ts = ts.UTC()
-			sinceStr := ""
-			if pkg.IsValidTime(ts) {
-				sinceStr = " since " + ts.Format(time.RFC3339Nano) + " "
+			rangeStr := ""
+			if pkg.IsValidTime(sinceTs) {
+				rangeStr = " since " + sinceTs.Format(time.RFC3339Nano)
 			}
-			term.Infof("Showing logs%s; press Ctrl+C to stop:", sinceStr)
+			if pkg.IsValidTime(untilTs) {
+				rangeStr += " until " + untilTs.Format(time.RFC3339Nano)
+			}
+			term.Infof("Showing logs%s; press Ctrl+C to stop:", rangeStr)
+
 			services := args
 			if len(name) > 0 {
 				services = append(args, strings.Split(name, ",")...) // backwards compat
@@ -465,7 +476,8 @@ func makeComposeLogsCmd() *cobra.Command {
 				LogType:  logType,
 				Raw:      raw,
 				Services: services,
-				Since:    ts,
+				Since:    sinceTs,
+				Until:    untilTs,
 				Verbose:  verbose,
 			}
 
@@ -478,7 +490,8 @@ func makeComposeLogsCmd() *cobra.Command {
 	logsCmd.Flags().Bool("follow", false, "follow log output") // NOTE: -f is already used by --file
 	logsCmd.Flags().MarkHidden("follow")                       // TODO: implement this
 	logsCmd.Flags().BoolP("raw", "r", false, "show raw (unparsed) logs")
-	logsCmd.Flags().StringP("since", "S", "", "show logs since duration/time")
+	logsCmd.Flags().String("since", "", "show logs since duration/time")
+	logsCmd.Flags().String("until", "", "show logs until duration/time")
 	logsCmd.Flags().Bool("utc", false, "show logs in UTC timezone (ie. TZ=UTC)")
 	logsCmd.Flags().Var(&logType, "type", fmt.Sprintf(`show logs of type; one of %v`, logs.AllLogTypes))
 	logsCmd.Flags().String("filter", "", "only show logs containing given text; case-insensitive")

src/pkg/cli/client/byoc/aws/byoc.go

@@ -634,7 +634,7 @@ func (b *ByocAws) Query(ctx context.Context, req *defangv1.DebugRequest) error {
 			}
 			close(lgEvtChan)
 		}(lgi)
-		evtsChan = mergeLogEventChan(evtsChan, lgEvtChan) // Merge sort the log events base one timestamp
+		evtsChan = mergeLogEventChan(evtsChan, lgEvtChan) // Merge sort the log events based on timestamp
 	}
 
 loop:
@@ -678,36 +678,40 @@ loop:
 }
 
 // Inspired by https://dev.to/vinaygo/concurrency-merge-sort-using-channels-and-goroutines-in-golang-35f7
-func mergeLogEventChan(a, b chan ecs.LogEvent) chan ecs.LogEvent {
-	if a == nil {
-		return b
+func mergech[T any](left chan T, right chan T, c chan T, less func(T, T) bool) {
+	defer close(c)
+	val, ok := <-left
+	val2, ok2 := <-right
+	for ok && ok2 {
+		if less(val, val2) {
+			c <- val
+			val, ok = <-left
+		} else {
+			c <- val2
+			val2, ok2 = <-right
+		}
+	}
+	for ok {
+		c <- val
+		val, ok = <-left
+	}
+	for ok2 {
+		c <- val2
+		val2, ok2 = <-right
+	}
+}
+
+func mergeLogEventChan(left, right chan ecs.LogEvent) chan ecs.LogEvent {
+	if left == nil {
+		return right
 	}
-	if b == nil {
-		return a
+	if right == nil {
+		return left
 	}
 	out := make(chan ecs.LogEvent)
-	go func() {
-		defer close(out)
-		valA, okA := <-a
-		valB, okB := <-b
-		for okA && okB {
-			if *valA.Timestamp < *valB.Timestamp {
-				out <- valA
-				valA, okA = <-a
-			} else {
-				out <- valB
-				valB, okB = <-b
-			}
-		}
-		for okA {
-			out <- valA
-			valA, okA = <-a
-		}
-		for okB {
-			out <- valB
-			valB, okB = <-b
-		}
-	}()
+	go mergech(left, right, out, func(i1, i2 ecs.LogEvent) bool {
+		return *i1.Timestamp < *i2.Timestamp
+	})
 	return out
 }
 

src/pkg/cli/debug.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/AlecAivazis/survey/v2"
+	"github.com/DefangLabs/defang/src/pkg"
 	"github.com/DefangLabs/defang/src/pkg/cli/client"
 	"github.com/DefangLabs/defang/src/pkg/cli/compose"
 	"github.com/DefangLabs/defang/src/pkg/term"
@@ -35,6 +36,7 @@ type DebugConfig struct {
 	Project        *compose.Project
 	Provider       client.Provider
 	Since          time.Time
+	Until          time.Time
 }
 
 func InteractiveDebugDeployment(ctx context.Context, client client.FabricClient, debugConfig DebugConfig) error {
@@ -95,16 +97,20 @@ func DebugDeployment(ctx context.Context, client client.FabricClient, debugConfi
 		return ErrDryRun
 	}
 
-	var sinceTime *timestamppb.Timestamp = nil
-	if !debugConfig.Since.IsZero() {
+	var sinceTime, untilTime *timestamppb.Timestamp
+	if pkg.IsValidTime(debugConfig.Since) {
 		sinceTime = timestamppb.New(debugConfig.Since)
 	}
+	if pkg.IsValidTime(debugConfig.Until) {
+		untilTime = timestamppb.New(debugConfig.Until)
+	}
 	req := defangv1.DebugRequest{
 		Etag:     debugConfig.Etag,
 		Files:    files,
 		Project:  debugConfig.Project.Name,
 		Services: debugConfig.FailedServices,
 		Since:    sinceTime,
+		Until:    untilTime,
 	}
 	err := debugConfig.Provider.Query(ctx, &req)
 	if err != nil {

src/pkg/cli/tail.go

@@ -67,21 +67,25 @@ type EndLogConditional struct {
 type TailDetectStopEventFunc func(services []string, host string, eventlog string) bool
 
 type TailOptions struct {
-	Services           []string
+	EndEventDetectFunc TailDetectStopEventFunc // Deprecated: use Subscribe instead #851
 	Etag               types.ETag
-	Since              time.Time
+	Filter             string
+	LogType            logs.LogType
 	Raw                bool
-	EndEventDetectFunc TailDetectStopEventFunc // Deprecated: use Subscribe instead #851
+	Services           []string
+	Since              time.Time
+	Until              time.Time
 	Verbose            bool
-	LogType            logs.LogType
-	Filter             string
 }
 
 func (to TailOptions) String() string {
 	cmd := "tail --since=" + to.Since.UTC().Format(time.RFC3339Nano)
 	if len(to.Services) > 0 {
 		cmd += " --name=" + strings.Join(to.Services, ",")
 	}
+	if !to.Until.IsZero() {
+		cmd += " --until=" + to.Until.UTC().Format(time.RFC3339Nano)
+	}
 	if to.Etag != "" {
 		cmd += " --etag=" + to.Etag
 	}
@@ -208,11 +212,18 @@ func isTransientError(err error) bool {
 }
 
 func tail(ctx context.Context, provider client.Provider, projectName string, options TailOptions) error {
-	var since *timestamppb.Timestamp
+	var since, until *timestamppb.Timestamp
 	if pkg.IsValidTime(options.Since) {
 		since = timestamppb.New(options.Since)
-	} else {
-		options.Since = time.Now() // this is used to continue from the last timestamp
+	}
+	if pkg.IsValidTime(options.Until) {
+		until = timestamppb.New(options.Until)
+		// If the user specifies a deadline in the future, we should respect it
+		if options.Until.After(time.Now()) {
+			var cancel context.CancelFunc
+			ctx, cancel = context.WithDeadline(ctx, options.Until)
+			defer cancel()
+		}
 	}
 
 	tailRequest := &defangv1.TailRequest{
@@ -221,7 +232,8 @@ func tail(ctx context.Context, provider client.Provider, projectName string, opt
 		Pattern:  options.Filter,
 		Project:  projectName,
 		Services: options.Services,
-		Since:    since,
+		Since:    since, // this is also used to continue from the last timestamp
+		Until:    until,
 	}
 
 	serverStream, err := provider.Follow(ctx, tailRequest)