support --until end time

Author: lionello

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

@@ -709,7 +709,14 @@ func (b *ByocAws) Follow(ctx context.Context, req *defangv1.TailRequest) (client
 		if len(req.Services) == 1 {
 			service = req.Services[0]
 		}
-		eventStream, err = ecs.TailLogGroups(ctx, req.Since.AsTime(), b.getLogGroupInputs(etag, req.Project, service, req.Pattern, logType)...)
+		var start, end time.Time
+		if req.Since.IsValid() {
+			start = req.Since.AsTime()
+		}
+		if req.Until.IsValid() {
+			end = req.Until.AsTime()
+		}
+		eventStream, err = ecs.QueryAndTailLogGroups(ctx, start, end, b.getLogGroupInputs(etag, req.Project, service, req.Pattern, logType)...)
 		taskArn = b.cdTaskArn
 	}
 	if err != nil {

src/pkg/clouds/aws/ecs/logs.go

@@ -25,7 +25,7 @@ func getLogGroupIdentifier(arnOrId string) string {
 	return strings.TrimSuffix(arnOrId, ":*")
 }
 
-func TailLogGroups(ctx context.Context, start time.Time, logGroups ...LogGroupInput) (EventStream, error) {
+func QueryAndTailLogGroups(ctx context.Context, start, end time.Time, logGroups ...LogGroupInput) (EventStream, error) {
 	ctx, cancel := context.WithCancel(ctx)
 
 	e := &eventStream{
@@ -34,13 +34,14 @@ func TailLogGroups(ctx context.Context, start time.Time, logGroups ...LogGroupIn
 	}
 
 	for _, lgi := range logGroups {
-		es, err := QueryAndTailLogGroup(ctx, lgi, start)
+		es, err := QueryAndTailLogGroup(ctx, lgi, start, end)
 		if err != nil {
 			cancel()
 			return nil, err
 		}
 		go func() {
 			defer es.Close()
+			// FIXME: this should *merge* the events from all log groups
 			e.pipeEvents(ctx, es)
 		}()
 	}
@@ -56,7 +57,7 @@ type LogGroupInput struct {
 	LogEventFilterPattern string
 }
 
-func TailLogGroup(ctx context.Context, input LogGroupInput) (EventStream, error) {
+func TailLogGroup(ctx context.Context, input LogGroupInput) (cloudwatchlogs.StartLiveTailResponseStreamReader, error) {
 	var pattern *string
 	if input.LogEventFilterPattern != "" {
 		pattern = &input.LogEventFilterPattern
@@ -172,13 +173,7 @@ func newCloudWatchLogsClient(ctx context.Context, region aws.Region) (*cloudwatc
 type LogEvent = types.LiveTailSessionLogEvent
 
 // EventStream is an interface that represents a stream of events from a call to StartLiveTail.
-type EventStream interface {
-	Close() error
-	Events() <-chan types.StartLiveTailResponseStream
-	Err() error
-}
-
-var _ EventStream = (*cloudwatchlogs.StartLiveTailEventStream)(nil)
+type EventStream = cloudwatchlogs.StartLiveTailResponseStreamReader
 
 func GetLogEvents(e types.StartLiveTailResponseStream) ([]LogEvent, error) {
 	switch ev := e.(type) {

src/pkg/clouds/aws/ecs/merge.go

@@ -1,7 +1,7 @@
 package ecs
 
 // Inspired by https://dev.to/vinaygo/concurrency-merge-sort-using-channels-and-goroutines-in-golang-35f7
-func mergech[T any](left chan T, right chan T, c chan T, less func(T, T) bool) {
+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
@@ -32,7 +32,7 @@ func mergeLogEventChan(left, right chan LogEvent) chan LogEvent {
 		return left
 	}
 	out := make(chan LogEvent)
-	go mergech(left, right, out, func(i1, i2 LogEvent) bool {
+	go Mergech(left, right, out, func(i1, i2 LogEvent) bool {
 		return *i1.Timestamp < *i2.Timestamp
 	})
 	return out

src/pkg/clouds/aws/ecs/stream.go

@@ -5,44 +5,58 @@ import (
 	"errors"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs/types"
 )
 
 // QueryAndTailLogGroup queries the log group from the give start time and initiates a Live Tail session.
 // This function also handles the case where the log group does not exist yet.
 // The caller should call `Close()` on the returned EventStream when done.
-func QueryAndTailLogGroup(ctx context.Context, lgi LogGroupInput, start time.Time) (EventStream, error) {
+func QueryAndTailLogGroup(ctx context.Context, lgi LogGroupInput, start, end time.Time) (EventStream, error) {
 	ctx, cancel := context.WithCancel(ctx)
 
 	es := &eventStream{
 		cancel: cancel,
 		ch:     make(chan types.StartLiveTailResponseStream),
 	}
 
-	// First call TailLogGroup once to check if the log group exists or we have another error
-	eventStream, err := TailLogGroup(ctx, lgi)
-	if err != nil {
-		var resourceNotFound *types.ResourceNotFoundException
-		if !errors.As(err, &resourceNotFound) {
-			return nil, err
+	doTail := !end.IsZero()
+
+	var tailStream cloudwatchlogs.StartLiveTailResponseStreamReader
+	if doTail {
+		// First call TailLogGroup once to check if the log group exists or we have another error
+		var err error
+		tailStream, err = TailLogGroup(ctx, lgi)
+		if err != nil {
+			var resourceNotFound *types.ResourceNotFoundException
+			if !errors.As(err, &resourceNotFound) {
+				return nil, err
+			}
+			// Doesn't exist yet, continue to poll for it
 		}
 	}
 
 	// Start goroutine to wait for the log group to be created and then tail it
 	go func() {
 		defer close(es.ch)
 
-		if eventStream == nil {
+		if doTail {
 			// If the log group does not exist yet, poll until it does
-			eventStream, err = pollTailLogGroup(ctx, lgi)
-			if err != nil {
-				es.err = err
-				return
+			if tailStream == nil {
+				var err error
+				tailStream, err = pollTailLogGroup(ctx, lgi)
+				if err != nil {
+					es.err = err
+					return
+				}
 			}
+			defer tailStream.Close()
 		}
-		defer eventStream.Close()
 
 		if !start.IsZero() {
+			if end.IsZero() {
+				end = time.Now()
+			}
 			// Query the logs between the start time and now
 			if err := QueryLogGroup(ctx, lgi, start, time.Now(), func(events []LogEvent) error {
 				es.ch <- &types.StartLiveTailResponseStreamMemberSessionUpdate{
@@ -55,14 +69,17 @@ func QueryAndTailLogGroup(ctx context.Context, lgi LogGroupInput, start time.Tim
 			}
 		}
 
-		es.pipeEvents(ctx, eventStream)
+		if doTail {
+			// Pipe the events from the tail stream to the internal channel
+			es.pipeEvents(ctx, tailStream)
+		}
 	}()
 
 	return es, nil
 }
 
 // pollTailLogGroup polls the log group and starts the Live Tail session once it's available
-func pollTailLogGroup(ctx context.Context, lgi LogGroupInput) (EventStream, error) {
+func pollTailLogGroup(ctx context.Context, lgi LogGroupInput) (cloudwatchlogs.StartLiveTailResponseStreamReader, error) {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 

src/pkg/clouds/aws/ecs/stream_test.go

@@ -15,7 +15,7 @@ func TestPendingStream(t *testing.T) {
 
 	ps, _ := QueryAndTailLogGroup(context.Background(), LogGroupInput{
 		LogGroupARN: "arn:aws:logs:us-west-2:532501343364:log-group:/ecs/lio/logss:*",
-	}, time.Now().Add(-time.Minute))
+	}, time.Now().Add(-time.Minute), time.Time{})
 
 	go func() {
 		time.Sleep(5 * time.Second)