chore: convert admin log endpoint to a generic endpoint

Author: jvmakine

backend/admin/service.go

@@ -618,18 +618,9 @@ func (s *Service) FailChangeset(ctx context.Context, c *connect.Request[ftlv1.Fa
 	return connect.NewResponse(res.Msg), nil
 }
 
-func (s *Service) StreamChangesetLogs(ctx context.Context, req *connect.Request[adminpb.StreamChangesetLogsRequest], resp *connect.ServerStream[adminpb.StreamChangesetLogsResponse]) error {
+func (s *Service) StreamLogs(ctx context.Context, req *connect.Request[adminpb.StreamLogsRequest], resp *connect.ServerStream[adminpb.StreamLogsResponse]) error {
 	timeline, err := s.timelineClient.StreamTimeline(ctx, connect.NewRequest(&timelinepb.StreamTimelineRequest{
-		Query: &timelinepb.GetTimelineRequest{
-			Limit: 100000,
-			Filters: []*timelinepb.GetTimelineRequest_Filter{{
-				Filter: &timelinepb.GetTimelineRequest_Filter_Changesets{
-					Changesets: &timelinepb.GetTimelineRequest_ChangesetFilter{
-						Changesets: []string{req.Msg.ChangesetKey},
-					},
-				},
-			}},
-		},
+		Query: req.Msg.Query,
 	}))
 	if err != nil {
 		return fmt.Errorf("failed to get timeline: %w", err)
@@ -643,7 +634,7 @@ func (s *Service) StreamChangesetLogs(ctx context.Context, req *connect.Request[
 				logs = append(logs, log)
 			}
 		}
-		if err := resp.Send(&adminpb.StreamChangesetLogsResponse{
+		if err := resp.Send(&adminpb.StreamLogsResponse{
 			Logs: logs,
 		}); err != nil {
 			return fmt.Errorf("failed to send logs: %w", err)

backend/protos/xyz/block/ftl/admin/v1/admin.pb.go

@@ -4,6 +4,7 @@ package xyz.block.ftl.admin.v1;
 
 import "xyz/block/ftl/schema/v1/schema.proto";
 import "xyz/block/ftl/timeline/v1/event.proto";
+import "xyz/block/ftl/timeline/v1/timeline.proto";
 import "xyz/block/ftl/v1/ftl.proto";
 import "xyz/block/ftl/v1/schemaservice.proto";
 
@@ -196,11 +197,11 @@ message ClusterInfoResponse {
   string arch = 2;
 }
 
-message StreamChangesetLogsRequest {
-  string changeset_key = 1;
+message StreamLogsRequest {
+  timeline.v1.TimelineQuery query = 1;
 }
 
-message StreamChangesetLogsResponse {
+message StreamLogsResponse {
   repeated timeline.v1.LogEvent logs = 1;
 }
 
@@ -283,5 +284,5 @@ service AdminService {
   // Upload an artefact to the server.
   rpc UploadArtefact(stream UploadArtefactRequest) returns (UploadArtefactResponse);
 
-  rpc StreamChangesetLogs(StreamChangesetLogsRequest) returns (stream StreamChangesetLogsResponse);
+  rpc StreamLogs(StreamLogsRequest) returns (stream StreamLogsResponse);
 }