support --until end time

lionello · lionello · commit c99f6f45f071 · 2025-03-06T09:28:46.000-08:00
diff --git a/src/pkg/cli/client/byoc/aws/byoc.go b/src/pkg/cli/client/byoc/aws/byoc.go
@@ -596,7 +596,7 @@ func (b *ByocAws) CreateUploadURL(ctx context.Context, req *defangv1.UploadURLRe
 	}, nil
 }
 
-func (b *ByocAws) Query(ctx context.Context, req *defangv1.DebugRequest) error {
+func (b *ByocAws) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {
 	// tailRequest := &defangv1.TailRequest{
 	// 	Etag:     req.Etag,
 	// 	Project:  req.Project,
@@ -695,11 +695,11 @@ func (b *ByocAws) Follow(ctx context.Context, req *defangv1.TailRequest) (client
 	//  * Etag, service:		tail that task/service
 	var err error
 	var taskArn ecs.TaskArn
-	var eventStream ecs.EventStream
+	var tailStream ecs.LiveTailStream
 	stopWhenCDTaskDone := false
 	logType := logs.LogType(req.LogType)
 	if etag != "" && !pkg.IsValidRandomID(etag) { // Assume invalid "etag" is a task ID
-		eventStream, err = b.driver.TailTaskID(ctx, etag)
+		tailStream, err = b.driver.TailTaskID(ctx, etag)
 		taskArn, _ = b.driver.GetTaskArn(etag)
 		term.Debugf("Tailing task %s", *taskArn)
 		etag = "" // no need to filter by etag
@@ -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()
+		}
+		tailStream, err = ecs.QueryAndTailLogGroups(ctx, start, end, b.getLogGroupInputs(etag, req.Project, service, req.Pattern, logType)...)
 		taskArn = b.cdTaskArn
 	}
 	if err != nil {
@@ -731,7 +738,7 @@ func (b *ByocAws) Follow(ctx context.Context, req *defangv1.TailRequest) (client
 		}()
 	}
 
-	return newByocServerStream(ctx, eventStream, etag, req.GetServices(), b), nil
+	return newByocServerStream(ctx, tailStream, etag, req.GetServices(), b), nil
 }
 
 func (b *ByocAws) makeLogGroupARN(name string) string {
diff --git a/src/pkg/cli/client/byoc/aws/stream.go b/src/pkg/cli/client/byoc/aws/stream.go
@@ -24,12 +24,12 @@ type byocServerStream struct {
 	etag     string
 	response *defangv1.TailResponse
 	services []string
-	stream   ecs.EventStream
+	stream   ecs.LiveTailStream
 
 	ecsEventsHandler ECSEventHandler
 }
 
-func newByocServerStream(ctx context.Context, stream ecs.EventStream, etag string, services []string, ecsEventHandler ECSEventHandler) *byocServerStream {
+func newByocServerStream(ctx context.Context, stream ecs.LiveTailStream, etag string, services []string, ecsEventHandler ECSEventHandler) *byocServerStream {
 	return &byocServerStream{
 		ctx:      ctx,
 		etag:     etag,
diff --git a/src/pkg/cli/client/byoc/do/byoc.go b/src/pkg/cli/client/byoc/do/byoc.go
@@ -598,7 +598,7 @@ func (s *subscribeStream) Close() error {
 	return nil
 }
 
-func (b *ByocDo) Query(ctx context.Context, req *defangv1.DebugRequest) error {
+func (b *ByocDo) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {
 	return client.ErrNotImplemented("AI debugging is not yet supported for DO BYOC")
 }
 
diff --git a/src/pkg/cli/client/byoc/gcp/byoc.go b/src/pkg/cli/client/byoc/gcp/byoc.go
@@ -735,7 +735,7 @@ func (b *ByocGcp) query(ctx context.Context, query string) ([]*loggingpb.LogEntr
 	return entries, nil
 }
 
-func (b *ByocGcp) Query(ctx context.Context, req *defangv1.DebugRequest) error {
+func (b *ByocGcp) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {
 	logEntries, err := b.query(ctx, b.createDeploymentLogQuery(req))
 	if err != nil {
 		return annotateGcpError(err)
diff --git a/src/pkg/cli/client/playground.go b/src/pkg/cli/client/playground.go
@@ -121,7 +121,7 @@ func (g *PlaygroundProvider) AccountInfo(ctx context.Context) (AccountInfo, erro
 	return PlaygroundAccountInfo{}, nil
 }
 
-func (g *PlaygroundProvider) Query(ctx context.Context, req *defangv1.DebugRequest) error {
+func (g *PlaygroundProvider) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {
 	return nil
 }
 
diff --git a/src/pkg/cli/client/provider.go b/src/pkg/cli/client/provider.go
@@ -142,7 +142,7 @@ type Provider interface {
 	GetService(context.Context, *defangv1.GetRequest) (*defangv1.ServiceInfo, error)
 	GetServices(context.Context, *defangv1.GetServicesRequest) (*defangv1.GetServicesResponse, error)
 	ListConfig(context.Context, *defangv1.ListConfigsRequest) (*defangv1.Secrets, error)
-	Query(context.Context, *defangv1.DebugRequest) error
+	QueryForDebug(context.Context, *defangv1.DebugRequest) error
 	Preview(context.Context, *defangv1.DeployRequest) (*defangv1.DeployResponse, error)
 	PutConfig(context.Context, *defangv1.PutConfigRequest) error
 	RemoteProjectName(context.Context) (string, error)
diff --git a/src/pkg/cli/debug.go b/src/pkg/cli/debug.go
@@ -112,7 +112,7 @@ func DebugDeployment(ctx context.Context, client client.FabricClient, debugConfi
 		Since:    sinceTime,
 		Until:    untilTime,
 	}
-	err := debugConfig.Provider.Query(ctx, &req)
+	err := debugConfig.Provider.QueryForDebug(ctx, &req)
 	if err != nil {
 		return err
 	}
diff --git a/src/pkg/cli/debug_test.go b/src/pkg/cli/debug_test.go
@@ -41,7 +41,7 @@ type MustHaveProjectNameQueryProvider struct {
 	client.Provider
 }
 
-func (m MustHaveProjectNameQueryProvider) Query(ctx context.Context, req *defangv1.DebugRequest) error {
+func (m MustHaveProjectNameQueryProvider) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {
 	if req.Project == "" {
 		return errors.New("project name is missing")
 	}
diff --git a/src/pkg/clouds/aws/ecs/logs.go b/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) (LiveTailStream, 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) (LiveTailStream, error) {
 	var pattern *string
 	if input.LogEventFilterPattern != "" {
 		pattern = &input.LogEventFilterPattern
@@ -171,14 +172,15 @@ 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 {
+// EventStream is a generic interface that represents a stream of events
+type EventStream[T any] interface {
+	Events() <-chan T
 	Close() error
-	Events() <-chan types.StartLiveTailResponseStream
 	Err() error
 }
 
-var _ EventStream = (*cloudwatchlogs.StartLiveTailEventStream)(nil)
+// Deprecated: LiveTailStream is a stream of events from a call to StartLiveTail
+type LiveTailStream = EventStream[types.StartLiveTailResponseStream]
 
 func GetLogEvents(e types.StartLiveTailResponseStream) ([]LogEvent, error) {
 	switch ev := e.(type) {
diff --git a/src/pkg/clouds/aws/ecs/merge.go b/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
diff --git a/src/pkg/clouds/aws/ecs/stream.go b/src/pkg/clouds/aws/ecs/stream.go
@@ -11,38 +11,51 @@ import (
 // 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) (LiveTailStream, 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 LiveTailStream
+	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 +68,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) (LiveTailStream, error) {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 
@@ -88,7 +104,7 @@ type eventStream struct {
 	err    error
 }
 
-var _ EventStream = (*eventStream)(nil)
+var _ LiveTailStream = (*eventStream)(nil)
 
 func (es *eventStream) Close() error {
 	es.cancel()
@@ -105,13 +121,13 @@ func (es *eventStream) Events() <-chan types.StartLiveTailResponseStream {
 
 // pipeEvents copies events from the given EventStream to the internal channel,
 // until the context is canceled or an error occurs in the given EventStream.
-func (es *eventStream) pipeEvents(ctx context.Context, eventStream EventStream) {
+func (es *eventStream) pipeEvents(ctx context.Context, tailStream LiveTailStream) {
 	for {
 		// Double select to make sure context cancellation is not blocked by either the receive or send
 		// See: https://stackoverflow.com/questions/60030756/what-does-it-mean-when-one-channel-uses-two-arrows-to-write-to-another-channel
 		select {
-		case event := <-eventStream.Events(): // blocking
-			if err := eventStream.Err(); err != nil {
+		case event := <-tailStream.Events(): // blocking
+			if err := tailStream.Err(); err != nil {
 				es.err = err
 				// Don't return, but continue to send any events to the caller so they can see the error
 			}
diff --git a/src/pkg/clouds/aws/ecs/stream_test.go b/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)
diff --git a/src/pkg/clouds/aws/ecs/tail.go b/src/pkg/clouds/aws/ecs/tail.go
@@ -57,7 +57,7 @@ func (a *AwsEcs) GetTaskArn(taskID string) (TaskArn, error) {
 	return &taskArn, nil
 }
 
-func (a *AwsEcs) TailTaskID(ctx context.Context, taskID string) (EventStream, error) {
+func (a *AwsEcs) TailTaskID(ctx context.Context, taskID string) (LiveTailStream, error) {
 	if taskID == "" {
 		return nil, errors.New("taskID is empty")
 	}

Original file line number	Diff line number	Diff line change
`@@ -598,7 +598,7 @@ func (s *subscribeStream) Close() error {`
`598`	`598`	`return nil`
`599`	`599`	`}`
`600`	`600`
`601`		`-func (b ByocDo) Query(ctx context.Context, req defangv1.DebugRequest) error {`
	`601`	`+func (b ByocDo) QueryForDebug(ctx context.Context, req defangv1.DebugRequest) error {`
`602`	`602`	`return client.ErrNotImplemented("AI debugging is not yet supported for DO BYOC")`
`603`	`603`	`}`
`604`	`604`
Original file line number	Diff line number	Diff line change
`@@ -735,7 +735,7 @@ func (b ByocGcp) query(ctx context.Context, query string) ([]loggingpb.LogEntr`
`735`	`735`	`return entries, nil`
`736`	`736`	`}`
`737`	`737`
`738`		`-func (b ByocGcp) Query(ctx context.Context, req defangv1.DebugRequest) error {`
	`738`	`+func (b ByocGcp) QueryForDebug(ctx context.Context, req defangv1.DebugRequest) error {`
`739`	`739`	`logEntries, err := b.query(ctx, b.createDeploymentLogQuery(req))`
`740`	`740`	`if err != nil {`
`741`	`741`	`return annotateGcpError(err)`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,7 @@ func (g *PlaygroundProvider) AccountInfo(ctx context.Context) (AccountInfo, erro`
`121`	`121`	`return PlaygroundAccountInfo{}, nil`
`122`	`122`	`}`
`123`	`123`
`124`		`-func (g PlaygroundProvider) Query(ctx context.Context, req defangv1.DebugRequest) error {`
	`124`	`+func (g PlaygroundProvider) QueryForDebug(ctx context.Context, req defangv1.DebugRequest) error {`
`125`	`125`	`return nil`
`126`	`126`	`}`
`127`	`127`
Original file line number	Diff line number	Diff line change
`@@ -112,7 +112,7 @@ func DebugDeployment(ctx context.Context, client client.FabricClient, debugConfi`
`112`	`112`	`Since: sinceTime,`
`113`	`113`	`Until: untilTime,`
`114`	`114`	`}`
`115`		`- err := debugConfig.Provider.Query(ctx, &req)`
	`115`	`+ err := debugConfig.Provider.QueryForDebug(ctx, &req)`
`116`	`116`	`if err != nil {`
`117`	`117`	`return err`
`118`	`118`	`}`
Original file line number	Diff line number	Diff line change
`@@ -41,7 +41,7 @@ type MustHaveProjectNameQueryProvider struct {`
`41`	`41`	`client.Provider`
`42`	`42`	`}`
`43`	`43`
`44`		`-func (m MustHaveProjectNameQueryProvider) Query(ctx context.Context, req *defangv1.DebugRequest) error {`
	`44`	`+func (m MustHaveProjectNameQueryProvider) QueryForDebug(ctx context.Context, req *defangv1.DebugRequest) error {`
`45`	`45`	`if req.Project == "" {`
`46`	`46`	`return errors.New("project name is missing")`
`47`	`47`	`}`