feat: Generic endpoint to get logs from Admin Service

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.proto

@@ -1,9 +1,10 @@
 syntax = "proto3";
 
 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,17 +197,17 @@
   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;
 }
 
 // AdminService is the service that provides and updates admin data. For example,
 // it is used to encapsulate configuration and secrets.
 service AdminService {
   rpc Ping(ftl.v1.PingRequest) returns (ftl.v1.PingResponse) {
     option idempotency_level = NO_SIDE_EFFECTS;
   }
@@ -283,5 +284,5 @@
   // Upload an artefact to the server.
   rpc UploadArtefact(stream UploadArtefactRequest) returns (UploadArtefactResponse);
 
-  rpc StreamChangesetLogs(StreamChangesetLogsRequest) returns (stream StreamChangesetLogsResponse);
+  rpc StreamLogs(StreamLogsRequest) returns (stream StreamLogsResponse);
 }