From 23232e0c3c8806dd0749910a6d169a10dd72b63e Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Fri, 15 May 2026 09:25:37 +0200 Subject: [PATCH 1/4] Regenerate proto --- protogen/datasets/v1/collections_grpc.pb.go | 2 +- protogen/datasets/v1/data_access_grpc.pb.go | 2 +- .../datasets/v1/data_ingestion_grpc.pb.go | 2 +- protogen/datasets/v1/datasets_grpc.pb.go | 2 +- protogen/tilebox/v1/query.pb.go | 79 +++++++-- protogen/workflows/v1/automation_grpc.pb.go | 2 +- protogen/workflows/v1/diagram_grpc.pb.go | 2 +- protogen/workflows/v1/job.pb.go | 94 ++++++----- protogen/workflows/v1/job_grpc.pb.go | 2 +- protogen/workflows/v1/task_grpc.pb.go | 2 +- protogen/workflows/v1/task_release_grpc.pb.go | 2 +- protogen/workflows/v1/telemetry.pb.go | 158 ++++++------------ protogen/workflows/v1/telemetry_grpc.pb.go | 2 +- protogen/workflows/v1/workflows_grpc.pb.go | 2 +- 14 files changed, 186 insertions(+), 167 deletions(-) diff --git a/protogen/datasets/v1/collections_grpc.pb.go b/protogen/datasets/v1/collections_grpc.pb.go index 75c7a0e..4a1d092 100644 --- a/protogen/datasets/v1/collections_grpc.pb.go +++ b/protogen/datasets/v1/collections_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: datasets/v1/collections.proto diff --git a/protogen/datasets/v1/data_access_grpc.pb.go b/protogen/datasets/v1/data_access_grpc.pb.go index 71289e9..74a58e4 100644 --- a/protogen/datasets/v1/data_access_grpc.pb.go +++ b/protogen/datasets/v1/data_access_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: datasets/v1/data_access.proto diff --git a/protogen/datasets/v1/data_ingestion_grpc.pb.go b/protogen/datasets/v1/data_ingestion_grpc.pb.go index a4f65a4..882f2c3 100644 --- a/protogen/datasets/v1/data_ingestion_grpc.pb.go +++ b/protogen/datasets/v1/data_ingestion_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: datasets/v1/data_ingestion.proto diff --git a/protogen/datasets/v1/datasets_grpc.pb.go b/protogen/datasets/v1/datasets_grpc.pb.go index 94db713..8d18369 100644 --- a/protogen/datasets/v1/datasets_grpc.pb.go +++ b/protogen/datasets/v1/datasets_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: datasets/v1/datasets.proto diff --git a/protogen/tilebox/v1/query.pb.go b/protogen/tilebox/v1/query.pb.go index ff9c944..ebc0d7d 100644 --- a/protogen/tilebox/v1/query.pb.go +++ b/protogen/tilebox/v1/query.pb.go @@ -22,6 +22,54 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) +// SortDirection specifies the direction in which to sort log entries when querying logs. It can be used in combination +// with the cursor field in TelemetryPagination to paginate through log entries in either ascending or descending order. +type SortDirection int32 + +const ( + SortDirection_SORT_DIRECTION_UNSPECIFIED SortDirection = 0 + // Sort in ascending order (oldest entries first). + SortDirection_SORT_DIRECTION_ASCENDING SortDirection = 1 + // Sort in descending order (newest entries first). + SortDirection_SORT_DIRECTION_DESCENDING SortDirection = 2 +) + +// Enum value maps for SortDirection. +var ( + SortDirection_name = map[int32]string{ + 0: "SORT_DIRECTION_UNSPECIFIED", + 1: "SORT_DIRECTION_ASCENDING", + 2: "SORT_DIRECTION_DESCENDING", + } + SortDirection_value = map[string]int32{ + "SORT_DIRECTION_UNSPECIFIED": 0, + "SORT_DIRECTION_ASCENDING": 1, + "SORT_DIRECTION_DESCENDING": 2, + } +) + +func (x SortDirection) Enum() *SortDirection { + p := new(SortDirection) + *p = x + return p +} + +func (x SortDirection) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (SortDirection) Descriptor() protoreflect.EnumDescriptor { + return file_tilebox_v1_query_proto_enumTypes[0].Descriptor() +} + +func (SortDirection) Type() protoreflect.EnumType { + return &file_tilebox_v1_query_proto_enumTypes[0] +} + +func (x SortDirection) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + // A time interval type TimeInterval struct { state protoimpl.MessageState `protogen:"opaque.v1"` @@ -405,26 +453,32 @@ const file_tilebox_v1_query_proto_rawDesc = "" + "\n" + "Pagination\x12\"\n" + "\x05limit\x18\x01 \x01(\x03B\f\xbaH\x04\"\x02(\x01\xaa\x01\x02\b\x01R\x05limit\x12<\n" + - "\x0estarting_after\x18\x02 \x01(\v2\x0e.tilebox.v1.IDB\x05\xaa\x01\x02\b\x01R\rstartingAfterB\xa7\x01\n" + + "\x0estarting_after\x18\x02 \x01(\v2\x0e.tilebox.v1.IDB\x05\xaa\x01\x02\b\x01R\rstartingAfter*l\n" + + "\rSortDirection\x12\x1e\n" + + "\x1aSORT_DIRECTION_UNSPECIFIED\x10\x00\x12\x1c\n" + + "\x18SORT_DIRECTION_ASCENDING\x10\x01\x12\x1d\n" + + "\x19SORT_DIRECTION_DESCENDING\x10\x02B\xa7\x01\n" + "\x0ecom.tilebox.v1B\n" + "QueryProtoP\x01Z;github.com/tilebox/tilebox-go/protogen/tilebox/v1;tileboxv1\xa2\x02\x03TXX\xaa\x02\n" + "Tilebox.V1\xca\x02\n" + "Tilebox\\V1\xe2\x02\x16Tilebox\\V1\\GPBMetadata\xea\x02\vTilebox::V1\x92\x03\x02\b\x02b\beditionsp\xe8\a" +var file_tilebox_v1_query_proto_enumTypes = make([]protoimpl.EnumInfo, 1) var file_tilebox_v1_query_proto_msgTypes = make([]protoimpl.MessageInfo, 3) var file_tilebox_v1_query_proto_goTypes = []any{ - (*TimeInterval)(nil), // 0: tilebox.v1.TimeInterval - (*IDInterval)(nil), // 1: tilebox.v1.IDInterval - (*Pagination)(nil), // 2: tilebox.v1.Pagination - (*timestamppb.Timestamp)(nil), // 3: google.protobuf.Timestamp - (*ID)(nil), // 4: tilebox.v1.ID + (SortDirection)(0), // 0: tilebox.v1.SortDirection + (*TimeInterval)(nil), // 1: tilebox.v1.TimeInterval + (*IDInterval)(nil), // 2: tilebox.v1.IDInterval + (*Pagination)(nil), // 3: tilebox.v1.Pagination + (*timestamppb.Timestamp)(nil), // 4: google.protobuf.Timestamp + (*ID)(nil), // 5: tilebox.v1.ID } var file_tilebox_v1_query_proto_depIdxs = []int32{ - 3, // 0: tilebox.v1.TimeInterval.start_time:type_name -> google.protobuf.Timestamp - 3, // 1: tilebox.v1.TimeInterval.end_time:type_name -> google.protobuf.Timestamp - 4, // 2: tilebox.v1.IDInterval.start_id:type_name -> tilebox.v1.ID - 4, // 3: tilebox.v1.IDInterval.end_id:type_name -> tilebox.v1.ID - 4, // 4: tilebox.v1.Pagination.starting_after:type_name -> tilebox.v1.ID + 4, // 0: tilebox.v1.TimeInterval.start_time:type_name -> google.protobuf.Timestamp + 4, // 1: tilebox.v1.TimeInterval.end_time:type_name -> google.protobuf.Timestamp + 5, // 2: tilebox.v1.IDInterval.start_id:type_name -> tilebox.v1.ID + 5, // 3: tilebox.v1.IDInterval.end_id:type_name -> tilebox.v1.ID + 5, // 4: tilebox.v1.Pagination.starting_after:type_name -> tilebox.v1.ID 5, // [5:5] is the sub-list for method output_type 5, // [5:5] is the sub-list for method input_type 5, // [5:5] is the sub-list for extension type_name @@ -443,13 +497,14 @@ func file_tilebox_v1_query_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_tilebox_v1_query_proto_rawDesc), len(file_tilebox_v1_query_proto_rawDesc)), - NumEnums: 0, + NumEnums: 1, NumMessages: 3, NumExtensions: 0, NumServices: 0, }, GoTypes: file_tilebox_v1_query_proto_goTypes, DependencyIndexes: file_tilebox_v1_query_proto_depIdxs, + EnumInfos: file_tilebox_v1_query_proto_enumTypes, MessageInfos: file_tilebox_v1_query_proto_msgTypes, }.Build() File_tilebox_v1_query_proto = out.File diff --git a/protogen/workflows/v1/automation_grpc.pb.go b/protogen/workflows/v1/automation_grpc.pb.go index b85c6d2..6c39e5d 100644 --- a/protogen/workflows/v1/automation_grpc.pb.go +++ b/protogen/workflows/v1/automation_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/automation.proto diff --git a/protogen/workflows/v1/diagram_grpc.pb.go b/protogen/workflows/v1/diagram_grpc.pb.go index 58b424b..b67835e 100644 --- a/protogen/workflows/v1/diagram_grpc.pb.go +++ b/protogen/workflows/v1/diagram_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/diagram.proto diff --git a/protogen/workflows/v1/job.pb.go b/protogen/workflows/v1/job.pb.go index 6ac77fc..69bd05b 100644 --- a/protogen/workflows/v1/job.pb.go +++ b/protogen/workflows/v1/job.pb.go @@ -894,11 +894,12 @@ func (b0 QueryFilters_builder) Build() *QueryFilters { // QueryJobsRequest requests a list of jobs. type QueryJobsRequest struct { - state protoimpl.MessageState `protogen:"opaque.v1"` - xxx_hidden_Filters *QueryFilters `protobuf:"bytes,1,opt,name=filters"` - xxx_hidden_Page *v1.Pagination `protobuf:"bytes,2,opt,name=page"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_Filters *QueryFilters `protobuf:"bytes,1,opt,name=filters"` + xxx_hidden_Page *v1.Pagination `protobuf:"bytes,2,opt,name=page"` + xxx_hidden_SortDirection v1.SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=tilebox.v1.SortDirection"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *QueryJobsRequest) Reset() { @@ -940,6 +941,13 @@ func (x *QueryJobsRequest) GetPage() *v1.Pagination { return nil } +func (x *QueryJobsRequest) GetSortDirection() v1.SortDirection { + if x != nil { + return x.xxx_hidden_SortDirection + } + return v1.SortDirection(0) +} + func (x *QueryJobsRequest) SetFilters(v *QueryFilters) { x.xxx_hidden_Filters = v } @@ -948,6 +956,10 @@ func (x *QueryJobsRequest) SetPage(v *v1.Pagination) { x.xxx_hidden_Page = v } +func (x *QueryJobsRequest) SetSortDirection(v v1.SortDirection) { + x.xxx_hidden_SortDirection = v +} + func (x *QueryJobsRequest) HasFilters() bool { if x == nil { return false @@ -977,6 +989,8 @@ type QueryJobsRequest_builder struct { Filters *QueryFilters // The pagination parameters for this request. Page *v1.Pagination + // The sort order for jobs (by submission date). Defaults to descending order (newest jobs first). + SortDirection v1.SortDirection } func (b0 QueryJobsRequest_builder) Build() *QueryJobsRequest { @@ -985,6 +999,7 @@ func (b0 QueryJobsRequest_builder) Build() *QueryJobsRequest { _, _ = b, x x.xxx_hidden_Filters = b.Filters x.xxx_hidden_Page = b.Page + x.xxx_hidden_SortDirection = b.SortDirection return m0 } @@ -1361,10 +1376,11 @@ const file_workflows_v1_job_proto_rawDesc = "" + "\vtask_states\x18\x06 \x03(\x0e2\x17.workflows.v1.TaskStateR\n" + "taskStates:#\xbaH \"\x1e\n" + "\rtime_interval\n" + - "\vid_interval\x10\x00\"{\n" + + "\vid_interval\x10\x00\"\xbd\x01\n" + "\x10QueryJobsRequest\x124\n" + "\afilters\x18\x01 \x01(\v2\x1a.workflows.v1.QueryFiltersR\afilters\x121\n" + - "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\"v\n" + + "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12@\n" + + "\x0esort_direction\x18\x03 \x01(\x0e2\x19.tilebox.v1.SortDirectionR\rsortDirection\"v\n" + "\x11QueryJobsResponse\x12%\n" + "\x04jobs\x18\x01 \x03(\v2\x11.workflows.v1.JobR\x04jobs\x12:\n" + "\tnext_page\x18\x03 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\bnextPage\"G\n" + @@ -1425,8 +1441,9 @@ var file_workflows_v1_job_proto_goTypes = []any{ (JobState)(0), // 21: workflows.v1.JobState (TaskState)(0), // 22: workflows.v1.TaskState (*v1.Pagination)(nil), // 23: tilebox.v1.Pagination - (*Job)(nil), // 24: workflows.v1.Job - (*Diagram)(nil), // 25: workflows.v1.Diagram + (v1.SortDirection)(0), // 24: tilebox.v1.SortDirection + (*Job)(nil), // 25: workflows.v1.Job + (*Diagram)(nil), // 26: workflows.v1.Diagram } var file_workflows_v1_job_proto_depIdxs = []int32{ 15, // 0: workflows.v1.SubmitJobRequest.tasks:type_name -> workflows.v1.TaskSubmissions @@ -1446,35 +1463,36 @@ var file_workflows_v1_job_proto_depIdxs = []int32{ 22, // 14: workflows.v1.QueryFilters.task_states:type_name -> workflows.v1.TaskState 9, // 15: workflows.v1.QueryJobsRequest.filters:type_name -> workflows.v1.QueryFilters 23, // 16: workflows.v1.QueryJobsRequest.page:type_name -> tilebox.v1.Pagination - 24, // 17: workflows.v1.QueryJobsResponse.jobs:type_name -> workflows.v1.Job - 23, // 18: workflows.v1.QueryJobsResponse.next_page:type_name -> tilebox.v1.Pagination - 16, // 19: workflows.v1.GetJobPrototypeRequest.job_id:type_name -> tilebox.v1.ID - 17, // 20: workflows.v1.GetJobPrototypeResponse.root_tasks:type_name -> workflows.v1.SingleTaskSubmission - 16, // 21: workflows.v1.CloneJobRequest.job_id:type_name -> tilebox.v1.ID - 17, // 22: workflows.v1.CloneJobRequest.root_tasks_overrides:type_name -> workflows.v1.SingleTaskSubmission - 1, // 23: workflows.v1.JobService.SubmitJob:input_type -> workflows.v1.SubmitJobRequest - 2, // 24: workflows.v1.JobService.GetJob:input_type -> workflows.v1.GetJobRequest - 3, // 25: workflows.v1.JobService.GetJobProgress:input_type -> workflows.v1.GetJobProgressRequest - 4, // 26: workflows.v1.JobService.RetryJob:input_type -> workflows.v1.RetryJobRequest - 6, // 27: workflows.v1.JobService.CancelJob:input_type -> workflows.v1.CancelJobRequest - 8, // 28: workflows.v1.JobService.VisualizeJob:input_type -> workflows.v1.VisualizeJobRequest - 10, // 29: workflows.v1.JobService.QueryJobs:input_type -> workflows.v1.QueryJobsRequest - 12, // 30: workflows.v1.JobService.GetJobPrototype:input_type -> workflows.v1.GetJobPrototypeRequest - 14, // 31: workflows.v1.JobService.CloneJob:input_type -> workflows.v1.CloneJobRequest - 24, // 32: workflows.v1.JobService.SubmitJob:output_type -> workflows.v1.Job - 24, // 33: workflows.v1.JobService.GetJob:output_type -> workflows.v1.Job - 24, // 34: workflows.v1.JobService.GetJobProgress:output_type -> workflows.v1.Job - 5, // 35: workflows.v1.JobService.RetryJob:output_type -> workflows.v1.RetryJobResponse - 7, // 36: workflows.v1.JobService.CancelJob:output_type -> workflows.v1.CancelJobResponse - 25, // 37: workflows.v1.JobService.VisualizeJob:output_type -> workflows.v1.Diagram - 11, // 38: workflows.v1.JobService.QueryJobs:output_type -> workflows.v1.QueryJobsResponse - 13, // 39: workflows.v1.JobService.GetJobPrototype:output_type -> workflows.v1.GetJobPrototypeResponse - 24, // 40: workflows.v1.JobService.CloneJob:output_type -> workflows.v1.Job - 32, // [32:41] is the sub-list for method output_type - 23, // [23:32] is the sub-list for method input_type - 23, // [23:23] is the sub-list for extension type_name - 23, // [23:23] is the sub-list for extension extendee - 0, // [0:23] is the sub-list for field type_name + 24, // 17: workflows.v1.QueryJobsRequest.sort_direction:type_name -> tilebox.v1.SortDirection + 25, // 18: workflows.v1.QueryJobsResponse.jobs:type_name -> workflows.v1.Job + 23, // 19: workflows.v1.QueryJobsResponse.next_page:type_name -> tilebox.v1.Pagination + 16, // 20: workflows.v1.GetJobPrototypeRequest.job_id:type_name -> tilebox.v1.ID + 17, // 21: workflows.v1.GetJobPrototypeResponse.root_tasks:type_name -> workflows.v1.SingleTaskSubmission + 16, // 22: workflows.v1.CloneJobRequest.job_id:type_name -> tilebox.v1.ID + 17, // 23: workflows.v1.CloneJobRequest.root_tasks_overrides:type_name -> workflows.v1.SingleTaskSubmission + 1, // 24: workflows.v1.JobService.SubmitJob:input_type -> workflows.v1.SubmitJobRequest + 2, // 25: workflows.v1.JobService.GetJob:input_type -> workflows.v1.GetJobRequest + 3, // 26: workflows.v1.JobService.GetJobProgress:input_type -> workflows.v1.GetJobProgressRequest + 4, // 27: workflows.v1.JobService.RetryJob:input_type -> workflows.v1.RetryJobRequest + 6, // 28: workflows.v1.JobService.CancelJob:input_type -> workflows.v1.CancelJobRequest + 8, // 29: workflows.v1.JobService.VisualizeJob:input_type -> workflows.v1.VisualizeJobRequest + 10, // 30: workflows.v1.JobService.QueryJobs:input_type -> workflows.v1.QueryJobsRequest + 12, // 31: workflows.v1.JobService.GetJobPrototype:input_type -> workflows.v1.GetJobPrototypeRequest + 14, // 32: workflows.v1.JobService.CloneJob:input_type -> workflows.v1.CloneJobRequest + 25, // 33: workflows.v1.JobService.SubmitJob:output_type -> workflows.v1.Job + 25, // 34: workflows.v1.JobService.GetJob:output_type -> workflows.v1.Job + 25, // 35: workflows.v1.JobService.GetJobProgress:output_type -> workflows.v1.Job + 5, // 36: workflows.v1.JobService.RetryJob:output_type -> workflows.v1.RetryJobResponse + 7, // 37: workflows.v1.JobService.CancelJob:output_type -> workflows.v1.CancelJobResponse + 26, // 38: workflows.v1.JobService.VisualizeJob:output_type -> workflows.v1.Diagram + 11, // 39: workflows.v1.JobService.QueryJobs:output_type -> workflows.v1.QueryJobsResponse + 13, // 40: workflows.v1.JobService.GetJobPrototype:output_type -> workflows.v1.GetJobPrototypeResponse + 25, // 41: workflows.v1.JobService.CloneJob:output_type -> workflows.v1.Job + 33, // [33:42] is the sub-list for method output_type + 24, // [24:33] is the sub-list for method input_type + 24, // [24:24] is the sub-list for extension type_name + 24, // [24:24] is the sub-list for extension extendee + 0, // [0:24] is the sub-list for field type_name } func init() { file_workflows_v1_job_proto_init() } diff --git a/protogen/workflows/v1/job_grpc.pb.go b/protogen/workflows/v1/job_grpc.pb.go index e79d7d3..2a61694 100644 --- a/protogen/workflows/v1/job_grpc.pb.go +++ b/protogen/workflows/v1/job_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/job.proto diff --git a/protogen/workflows/v1/task_grpc.pb.go b/protogen/workflows/v1/task_grpc.pb.go index 1d9a425..21c6c5a 100644 --- a/protogen/workflows/v1/task_grpc.pb.go +++ b/protogen/workflows/v1/task_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/task.proto diff --git a/protogen/workflows/v1/task_release_grpc.pb.go b/protogen/workflows/v1/task_release_grpc.pb.go index 959ba0b..4eb7976 100644 --- a/protogen/workflows/v1/task_release_grpc.pb.go +++ b/protogen/workflows/v1/task_release_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/task_release.proto diff --git a/protogen/workflows/v1/telemetry.pb.go b/protogen/workflows/v1/telemetry.pb.go index 93c0334..5ab6160 100644 --- a/protogen/workflows/v1/telemetry.pb.go +++ b/protogen/workflows/v1/telemetry.pb.go @@ -25,61 +25,13 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// SortDirection specifies the direction in which to sort log entries when querying logs. It can be used in combination -// with the cursor field in TelemetryPagination to paginate through log entries in either ascending or descending order. -type SortDirection int32 - -const ( - SortDirection_SORT_DIRECTION_UNSPECIFIED SortDirection = 0 - // Sort in ascending order (oldest entries first). - SortDirection_SORT_DIRECTION_ASCENDING SortDirection = 1 - // Sort in descending order (newest entries first). - SortDirection_SORT_DIRECTION_DESCENDING SortDirection = 2 -) - -// Enum value maps for SortDirection. -var ( - SortDirection_name = map[int32]string{ - 0: "SORT_DIRECTION_UNSPECIFIED", - 1: "SORT_DIRECTION_ASCENDING", - 2: "SORT_DIRECTION_DESCENDING", - } - SortDirection_value = map[string]int32{ - "SORT_DIRECTION_UNSPECIFIED": 0, - "SORT_DIRECTION_ASCENDING": 1, - "SORT_DIRECTION_DESCENDING": 2, - } -) - -func (x SortDirection) Enum() *SortDirection { - p := new(SortDirection) - *p = x - return p -} - -func (x SortDirection) String() string { - return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) -} - -func (SortDirection) Descriptor() protoreflect.EnumDescriptor { - return file_workflows_v1_telemetry_proto_enumTypes[0].Descriptor() -} - -func (SortDirection) Type() protoreflect.EnumType { - return &file_workflows_v1_telemetry_proto_enumTypes[0] -} - -func (x SortDirection) Number() protoreflect.EnumNumber { - return protoreflect.EnumNumber(x) -} - // QueryJobLogsRequest is the request message for querying logs of a specific job, in ascending or descending order, // with pagination support. type QueryJobLogsRequest struct { state protoimpl.MessageState `protogen:"opaque.v1"` xxx_hidden_JobId *v1.ID `protobuf:"bytes,1,opt,name=job_id,json=jobId"` xxx_hidden_Page *v1.Pagination `protobuf:"bytes,2,opt,name=page"` - xxx_hidden_SortDirection SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=workflows.v1.SortDirection"` + xxx_hidden_SortDirection v1.SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=tilebox.v1.SortDirection"` XXX_raceDetectHookData protoimpl.RaceDetectHookData XXX_presence [1]uint32 unknownFields protoimpl.UnknownFields @@ -125,13 +77,13 @@ func (x *QueryJobLogsRequest) GetPage() *v1.Pagination { return nil } -func (x *QueryJobLogsRequest) GetSortDirection() SortDirection { +func (x *QueryJobLogsRequest) GetSortDirection() v1.SortDirection { if x != nil { if protoimpl.X.Present(&(x.XXX_presence[0]), 2) { return x.xxx_hidden_SortDirection } } - return SortDirection_SORT_DIRECTION_UNSPECIFIED + return v1.SortDirection(0) } func (x *QueryJobLogsRequest) SetJobId(v *v1.ID) { @@ -142,7 +94,7 @@ func (x *QueryJobLogsRequest) SetPage(v *v1.Pagination) { x.xxx_hidden_Page = v } -func (x *QueryJobLogsRequest) SetSortDirection(v SortDirection) { +func (x *QueryJobLogsRequest) SetSortDirection(v v1.SortDirection) { x.xxx_hidden_SortDirection = v protoimpl.X.SetPresent(&(x.XXX_presence[0]), 2, 3) } @@ -178,7 +130,7 @@ func (x *QueryJobLogsRequest) ClearPage() { func (x *QueryJobLogsRequest) ClearSortDirection() { protoimpl.X.ClearPresent(&(x.XXX_presence[0]), 2) - x.xxx_hidden_SortDirection = SortDirection_SORT_DIRECTION_UNSPECIFIED + x.xxx_hidden_SortDirection = v1.SortDirection_SORT_DIRECTION_UNSPECIFIED } type QueryJobLogsRequest_builder struct { @@ -189,7 +141,7 @@ type QueryJobLogsRequest_builder struct { // The pagination parameters for this request. Page *v1.Pagination // The direction in which to sort log entries. If not specified, defaults to ascending order (oldest entries first). - SortDirection *SortDirection + SortDirection *v1.SortDirection } func (b0 QueryJobLogsRequest_builder) Build() *QueryJobLogsRequest { @@ -211,7 +163,7 @@ type QueryLogsInIntervalRequest struct { state protoimpl.MessageState `protogen:"opaque.v1"` xxx_hidden_TimeInterval *v1.TimeInterval `protobuf:"bytes,1,opt,name=time_interval,json=timeInterval"` xxx_hidden_Page *v1.Pagination `protobuf:"bytes,2,opt,name=page"` - xxx_hidden_SortDirection SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=workflows.v1.SortDirection"` + xxx_hidden_SortDirection v1.SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=tilebox.v1.SortDirection"` XXX_raceDetectHookData protoimpl.RaceDetectHookData XXX_presence [1]uint32 unknownFields protoimpl.UnknownFields @@ -257,13 +209,13 @@ func (x *QueryLogsInIntervalRequest) GetPage() *v1.Pagination { return nil } -func (x *QueryLogsInIntervalRequest) GetSortDirection() SortDirection { +func (x *QueryLogsInIntervalRequest) GetSortDirection() v1.SortDirection { if x != nil { if protoimpl.X.Present(&(x.XXX_presence[0]), 2) { return x.xxx_hidden_SortDirection } } - return SortDirection_SORT_DIRECTION_UNSPECIFIED + return v1.SortDirection(0) } func (x *QueryLogsInIntervalRequest) SetTimeInterval(v *v1.TimeInterval) { @@ -274,7 +226,7 @@ func (x *QueryLogsInIntervalRequest) SetPage(v *v1.Pagination) { x.xxx_hidden_Page = v } -func (x *QueryLogsInIntervalRequest) SetSortDirection(v SortDirection) { +func (x *QueryLogsInIntervalRequest) SetSortDirection(v v1.SortDirection) { x.xxx_hidden_SortDirection = v protoimpl.X.SetPresent(&(x.XXX_presence[0]), 2, 3) } @@ -310,7 +262,7 @@ func (x *QueryLogsInIntervalRequest) ClearPage() { func (x *QueryLogsInIntervalRequest) ClearSortDirection() { protoimpl.X.ClearPresent(&(x.XXX_presence[0]), 2) - x.xxx_hidden_SortDirection = SortDirection_SORT_DIRECTION_UNSPECIFIED + x.xxx_hidden_SortDirection = v1.SortDirection_SORT_DIRECTION_UNSPECIFIED } type QueryLogsInIntervalRequest_builder struct { @@ -321,7 +273,7 @@ type QueryLogsInIntervalRequest_builder struct { // The pagination parameters for this request. Page *v1.Pagination // The direction in which to sort log entries. If not specified, defaults to descending order (newest entries first). - SortDirection *SortDirection + SortDirection *v1.SortDirection } func (b0 QueryLogsInIntervalRequest_builder) Build() *QueryLogsInIntervalRequest { @@ -431,7 +383,7 @@ type QueryJobSpansRequest struct { state protoimpl.MessageState `protogen:"opaque.v1"` xxx_hidden_JobId *v1.ID `protobuf:"bytes,1,opt,name=job_id,json=jobId"` xxx_hidden_Page *v1.Pagination `protobuf:"bytes,2,opt,name=page"` - xxx_hidden_SortDirection SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=workflows.v1.SortDirection"` + xxx_hidden_SortDirection v1.SortDirection `protobuf:"varint,3,opt,name=sort_direction,json=sortDirection,enum=tilebox.v1.SortDirection"` XXX_raceDetectHookData protoimpl.RaceDetectHookData XXX_presence [1]uint32 unknownFields protoimpl.UnknownFields @@ -477,13 +429,13 @@ func (x *QueryJobSpansRequest) GetPage() *v1.Pagination { return nil } -func (x *QueryJobSpansRequest) GetSortDirection() SortDirection { +func (x *QueryJobSpansRequest) GetSortDirection() v1.SortDirection { if x != nil { if protoimpl.X.Present(&(x.XXX_presence[0]), 2) { return x.xxx_hidden_SortDirection } } - return SortDirection_SORT_DIRECTION_UNSPECIFIED + return v1.SortDirection(0) } func (x *QueryJobSpansRequest) SetJobId(v *v1.ID) { @@ -494,7 +446,7 @@ func (x *QueryJobSpansRequest) SetPage(v *v1.Pagination) { x.xxx_hidden_Page = v } -func (x *QueryJobSpansRequest) SetSortDirection(v SortDirection) { +func (x *QueryJobSpansRequest) SetSortDirection(v v1.SortDirection) { x.xxx_hidden_SortDirection = v protoimpl.X.SetPresent(&(x.XXX_presence[0]), 2, 3) } @@ -530,7 +482,7 @@ func (x *QueryJobSpansRequest) ClearPage() { func (x *QueryJobSpansRequest) ClearSortDirection() { protoimpl.X.ClearPresent(&(x.XXX_presence[0]), 2) - x.xxx_hidden_SortDirection = SortDirection_SORT_DIRECTION_UNSPECIFIED + x.xxx_hidden_SortDirection = v1.SortDirection_SORT_DIRECTION_UNSPECIFIED } type QueryJobSpansRequest_builder struct { @@ -541,7 +493,7 @@ type QueryJobSpansRequest_builder struct { // The pagination parameters for this request. Page *v1.Pagination // The direction in which to sort log entries. If not specified, defaults to ascending order (oldest entries first). - SortDirection *SortDirection + SortDirection *v1.SortDirection } func (b0 QueryJobSpansRequest_builder) Build() *QueryJobSpansRequest { @@ -649,70 +601,65 @@ var File_workflows_v1_telemetry_proto protoreflect.FileDescriptor const file_workflows_v1_telemetry_proto_rawDesc = "" + "\n" + - "\x1cworkflows/v1/telemetry.proto\x12\fworkflows.v1\x1a\x1bbuf/validate/validate.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&opentelemetry/proto/logs/v1/logs.proto\x1a(opentelemetry/proto/trace/v1/trace.proto\x1a\x13tilebox/v1/id.proto\x1a\x16tilebox/v1/query.proto\"\xbb\x01\n" + + "\x1cworkflows/v1/telemetry.proto\x12\fworkflows.v1\x1a\x1bbuf/validate/validate.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a&opentelemetry/proto/logs/v1/logs.proto\x1a(opentelemetry/proto/trace/v1/trace.proto\x1a\x13tilebox/v1/id.proto\x1a\x16tilebox/v1/query.proto\"\xb9\x01\n" + "\x13QueryJobLogsRequest\x12-\n" + "\x06job_id\x18\x01 \x01(\v2\x0e.tilebox.v1.IDB\x06\xbaH\x03\xc8\x01\x01R\x05jobId\x121\n" + - "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12B\n" + - "\x0esort_direction\x18\x03 \x01(\x0e2\x1b.workflows.v1.SortDirectionR\rsortDirection\"\xd2\x01\n" + + "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12@\n" + + "\x0esort_direction\x18\x03 \x01(\x0e2\x19.tilebox.v1.SortDirectionR\rsortDirection\"\xd0\x01\n" + "\x1aQueryLogsInIntervalRequest\x12=\n" + "\rtime_interval\x18\x01 \x01(\v2\x18.tilebox.v1.TimeIntervalR\ftimeInterval\x121\n" + - "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12B\n" + - "\x0esort_direction\x18\x03 \x01(\x0e2\x1b.workflows.v1.SortDirectionR\rsortDirection\"\x9f\x01\n" + + "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12@\n" + + "\x0esort_direction\x18\x03 \x01(\x0e2\x19.tilebox.v1.SortDirectionR\rsortDirection\"\x9f\x01\n" + "\x11PaginatedLogsData\x12N\n" + "\rresource_logs\x18\x01 \x03(\v2).opentelemetry.proto.logs.v1.ResourceLogsR\fresourceLogs\x12:\n" + - "\tnext_page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\bnextPage\"\xbc\x01\n" + + "\tnext_page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\bnextPage\"\xba\x01\n" + "\x14QueryJobSpansRequest\x12-\n" + "\x06job_id\x18\x01 \x01(\v2\x0e.tilebox.v1.IDB\x06\xbaH\x03\xc8\x01\x01R\x05jobId\x121\n" + - "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12B\n" + - "\x0esort_direction\x18\x03 \x01(\x0e2\x1b.workflows.v1.SortDirectionR\rsortDirection\"\xa4\x01\n" + + "\x04page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\x04page\x12@\n" + + "\x0esort_direction\x18\x03 \x01(\x0e2\x19.tilebox.v1.SortDirectionR\rsortDirection\"\xa4\x01\n" + "\x12PaginatedSpansData\x12R\n" + "\x0eresource_spans\x18\x01 \x03(\v2+.opentelemetry.proto.trace.v1.ResourceSpansR\rresourceSpans\x12:\n" + - "\tnext_page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\bnextPage*l\n" + - "\rSortDirection\x12\x1e\n" + - "\x1aSORT_DIRECTION_UNSPECIFIED\x10\x00\x12\x1c\n" + - "\x18SORT_DIRECTION_ASCENDING\x10\x01\x12\x1d\n" + - "\x19SORT_DIRECTION_DESCENDING\x10\x022\xa4\x02\n" + + "\tnext_page\x18\x02 \x01(\v2\x16.tilebox.v1.PaginationB\x05\xaa\x01\x02\b\x01R\bnextPage2\xa4\x02\n" + "\x15TelemetryQueryService\x12R\n" + "\fQueryJobLogs\x12!.workflows.v1.QueryJobLogsRequest\x1a\x1f.workflows.v1.PaginatedLogsData\x12`\n" + "\x13QueryLogsInInterval\x12(.workflows.v1.QueryLogsInIntervalRequest\x1a\x1f.workflows.v1.PaginatedLogsData\x12U\n" + "\rQueryJobSpans\x12\".workflows.v1.QueryJobSpansRequest\x1a .workflows.v1.PaginatedSpansDataB\xb4\x01\n" + "\x10com.workflows.v1B\x0eTelemetryProtoP\x01Z?github.com/tilebox/tilebox-go/protogen/workflows/v1;workflowsv1\xa2\x02\x03WXX\xaa\x02\fWorkflows.V1\xca\x02\fWorkflows\\V1\xe2\x02\x18Workflows\\V1\\GPBMetadata\xea\x02\rWorkflows::V1b\beditionsp\xe8\a" -var file_workflows_v1_telemetry_proto_enumTypes = make([]protoimpl.EnumInfo, 1) var file_workflows_v1_telemetry_proto_msgTypes = make([]protoimpl.MessageInfo, 5) var file_workflows_v1_telemetry_proto_goTypes = []any{ - (SortDirection)(0), // 0: workflows.v1.SortDirection - (*QueryJobLogsRequest)(nil), // 1: workflows.v1.QueryJobLogsRequest - (*QueryLogsInIntervalRequest)(nil), // 2: workflows.v1.QueryLogsInIntervalRequest - (*PaginatedLogsData)(nil), // 3: workflows.v1.PaginatedLogsData - (*QueryJobSpansRequest)(nil), // 4: workflows.v1.QueryJobSpansRequest - (*PaginatedSpansData)(nil), // 5: workflows.v1.PaginatedSpansData - (*v1.ID)(nil), // 6: tilebox.v1.ID - (*v1.Pagination)(nil), // 7: tilebox.v1.Pagination + (*QueryJobLogsRequest)(nil), // 0: workflows.v1.QueryJobLogsRequest + (*QueryLogsInIntervalRequest)(nil), // 1: workflows.v1.QueryLogsInIntervalRequest + (*PaginatedLogsData)(nil), // 2: workflows.v1.PaginatedLogsData + (*QueryJobSpansRequest)(nil), // 3: workflows.v1.QueryJobSpansRequest + (*PaginatedSpansData)(nil), // 4: workflows.v1.PaginatedSpansData + (*v1.ID)(nil), // 5: tilebox.v1.ID + (*v1.Pagination)(nil), // 6: tilebox.v1.Pagination + (v1.SortDirection)(0), // 7: tilebox.v1.SortDirection (*v1.TimeInterval)(nil), // 8: tilebox.v1.TimeInterval (*v11.ResourceLogs)(nil), // 9: opentelemetry.proto.logs.v1.ResourceLogs (*v12.ResourceSpans)(nil), // 10: opentelemetry.proto.trace.v1.ResourceSpans } var file_workflows_v1_telemetry_proto_depIdxs = []int32{ - 6, // 0: workflows.v1.QueryJobLogsRequest.job_id:type_name -> tilebox.v1.ID - 7, // 1: workflows.v1.QueryJobLogsRequest.page:type_name -> tilebox.v1.Pagination - 0, // 2: workflows.v1.QueryJobLogsRequest.sort_direction:type_name -> workflows.v1.SortDirection + 5, // 0: workflows.v1.QueryJobLogsRequest.job_id:type_name -> tilebox.v1.ID + 6, // 1: workflows.v1.QueryJobLogsRequest.page:type_name -> tilebox.v1.Pagination + 7, // 2: workflows.v1.QueryJobLogsRequest.sort_direction:type_name -> tilebox.v1.SortDirection 8, // 3: workflows.v1.QueryLogsInIntervalRequest.time_interval:type_name -> tilebox.v1.TimeInterval - 7, // 4: workflows.v1.QueryLogsInIntervalRequest.page:type_name -> tilebox.v1.Pagination - 0, // 5: workflows.v1.QueryLogsInIntervalRequest.sort_direction:type_name -> workflows.v1.SortDirection + 6, // 4: workflows.v1.QueryLogsInIntervalRequest.page:type_name -> tilebox.v1.Pagination + 7, // 5: workflows.v1.QueryLogsInIntervalRequest.sort_direction:type_name -> tilebox.v1.SortDirection 9, // 6: workflows.v1.PaginatedLogsData.resource_logs:type_name -> opentelemetry.proto.logs.v1.ResourceLogs - 7, // 7: workflows.v1.PaginatedLogsData.next_page:type_name -> tilebox.v1.Pagination - 6, // 8: workflows.v1.QueryJobSpansRequest.job_id:type_name -> tilebox.v1.ID - 7, // 9: workflows.v1.QueryJobSpansRequest.page:type_name -> tilebox.v1.Pagination - 0, // 10: workflows.v1.QueryJobSpansRequest.sort_direction:type_name -> workflows.v1.SortDirection + 6, // 7: workflows.v1.PaginatedLogsData.next_page:type_name -> tilebox.v1.Pagination + 5, // 8: workflows.v1.QueryJobSpansRequest.job_id:type_name -> tilebox.v1.ID + 6, // 9: workflows.v1.QueryJobSpansRequest.page:type_name -> tilebox.v1.Pagination + 7, // 10: workflows.v1.QueryJobSpansRequest.sort_direction:type_name -> tilebox.v1.SortDirection 10, // 11: workflows.v1.PaginatedSpansData.resource_spans:type_name -> opentelemetry.proto.trace.v1.ResourceSpans - 7, // 12: workflows.v1.PaginatedSpansData.next_page:type_name -> tilebox.v1.Pagination - 1, // 13: workflows.v1.TelemetryQueryService.QueryJobLogs:input_type -> workflows.v1.QueryJobLogsRequest - 2, // 14: workflows.v1.TelemetryQueryService.QueryLogsInInterval:input_type -> workflows.v1.QueryLogsInIntervalRequest - 4, // 15: workflows.v1.TelemetryQueryService.QueryJobSpans:input_type -> workflows.v1.QueryJobSpansRequest - 3, // 16: workflows.v1.TelemetryQueryService.QueryJobLogs:output_type -> workflows.v1.PaginatedLogsData - 3, // 17: workflows.v1.TelemetryQueryService.QueryLogsInInterval:output_type -> workflows.v1.PaginatedLogsData - 5, // 18: workflows.v1.TelemetryQueryService.QueryJobSpans:output_type -> workflows.v1.PaginatedSpansData + 6, // 12: workflows.v1.PaginatedSpansData.next_page:type_name -> tilebox.v1.Pagination + 0, // 13: workflows.v1.TelemetryQueryService.QueryJobLogs:input_type -> workflows.v1.QueryJobLogsRequest + 1, // 14: workflows.v1.TelemetryQueryService.QueryLogsInInterval:input_type -> workflows.v1.QueryLogsInIntervalRequest + 3, // 15: workflows.v1.TelemetryQueryService.QueryJobSpans:input_type -> workflows.v1.QueryJobSpansRequest + 2, // 16: workflows.v1.TelemetryQueryService.QueryJobLogs:output_type -> workflows.v1.PaginatedLogsData + 2, // 17: workflows.v1.TelemetryQueryService.QueryLogsInInterval:output_type -> workflows.v1.PaginatedLogsData + 4, // 18: workflows.v1.TelemetryQueryService.QueryJobSpans:output_type -> workflows.v1.PaginatedSpansData 16, // [16:19] is the sub-list for method output_type 13, // [13:16] is the sub-list for method input_type 13, // [13:13] is the sub-list for extension type_name @@ -730,14 +677,13 @@ func file_workflows_v1_telemetry_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_workflows_v1_telemetry_proto_rawDesc), len(file_workflows_v1_telemetry_proto_rawDesc)), - NumEnums: 1, + NumEnums: 0, NumMessages: 5, NumExtensions: 0, NumServices: 1, }, GoTypes: file_workflows_v1_telemetry_proto_goTypes, DependencyIndexes: file_workflows_v1_telemetry_proto_depIdxs, - EnumInfos: file_workflows_v1_telemetry_proto_enumTypes, MessageInfos: file_workflows_v1_telemetry_proto_msgTypes, }.Build() File_workflows_v1_telemetry_proto = out.File diff --git a/protogen/workflows/v1/telemetry_grpc.pb.go b/protogen/workflows/v1/telemetry_grpc.pb.go index 85e96fa..a483288 100644 --- a/protogen/workflows/v1/telemetry_grpc.pb.go +++ b/protogen/workflows/v1/telemetry_grpc.pb.go @@ -1,6 +1,6 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/telemetry.proto diff --git a/protogen/workflows/v1/workflows_grpc.pb.go b/protogen/workflows/v1/workflows_grpc.pb.go index 25ae4a7..ba7720e 100644 --- a/protogen/workflows/v1/workflows_grpc.pb.go +++ b/protogen/workflows/v1/workflows_grpc.pb.go @@ -2,7 +2,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: -// - protoc-gen-go-grpc v1.6.1 +// - protoc-gen-go-grpc v1.6.2 // - protoc (unknown) // source: workflows/v1/workflows.proto From 24dcb841f47d38c5cc097f8945373829452d08f2 Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Fri, 15 May 2026 09:26:06 +0200 Subject: [PATCH 2/4] Add sort direction param to job query --- workflows/v1/job/options.go | 22 ++++++++++++++ workflows/v1/job/options_test.go | 9 ++++++ workflows/v1/jobs.go | 23 +++++++++++--- workflows/v1/jobs_test.go | 52 ++++++++++++++++++++++++++------ workflows/v1/service.go | 17 ++++++----- 5 files changed, 100 insertions(+), 23 deletions(-) diff --git a/workflows/v1/job/options.go b/workflows/v1/job/options.go index 3d44546..29b4fee 100644 --- a/workflows/v1/job/options.go +++ b/workflows/v1/job/options.go @@ -56,10 +56,23 @@ type QueryOptions struct { AutomationIDs []uuid.UUID States []State Name string + SortDirection SortDirection } type QueryOption func(*QueryOptions) +// SortDirection specifies the sort direction for job query results. +type SortDirection int32 + +// SortDirection values. +const ( + _ SortDirection = iota + // Ascending sorts jobs oldest first. + Ascending + // Descending sorts jobs newest first. + Descending +) + // WithTemporalExtent specifies the time interval for which jobs should be queried. // Right now, a temporal extent is required for every query. func WithTemporalExtent(temporalExtent query.TemporalExtent) QueryOption { @@ -90,3 +103,12 @@ func WithJobName(name string) QueryOption { cfg.Name = name } } + +// WithSortDirection sets the sort direction for job query results. +// +// Defaults to the server default. +func WithSortDirection(direction SortDirection) QueryOption { + return func(cfg *QueryOptions) { + cfg.SortDirection = direction + } +} diff --git a/workflows/v1/job/options_test.go b/workflows/v1/job/options_test.go index 3cfaf2e..88a566e 100644 --- a/workflows/v1/job/options_test.go +++ b/workflows/v1/job/options_test.go @@ -99,6 +99,15 @@ func Test_QueryOptions(t *testing.T) { Name: "my-job", }, }, + { + name: "with sort direction", + options: []QueryOption{ + WithSortDirection(Descending), + }, + want: QueryOptions{ + SortDirection: Descending, + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/workflows/v1/jobs.go b/workflows/v1/jobs.go index cc62da2..2b66d39 100644 --- a/workflows/v1/jobs.go +++ b/workflows/v1/jobs.go @@ -233,6 +233,7 @@ type JobClient interface { // automation will be returned. (Optional) // - job.WithJobStates: specifies the job state to filter jobs by. Only jobs with the specified state will be returned. (Optional) // - job.WithJobName: specifies the job name to filter jobs by. Only jobs with the specified name will be returned. (Optional) + // - job.WithSortDirection: specifies the direction to sort jobs by submission date. (Optional) // // The jobs are lazily loaded and returned as a sequence. // The output sequence can be transformed into a slice using Collect. @@ -322,6 +323,7 @@ func (c jobClient) Query(ctx context.Context, options ...job.QueryOption) iter.S return func(yield func(*Job, error) bool) { var page *tileboxv1.Pagination // nil for the first request + sortDirection := jobQuerySortDirectionToProto(opts.SortDirection) // we already validated that TemporalExtent is not nil timeInterval := opts.TemporalExtent.ToProtoTimeInterval() @@ -349,7 +351,7 @@ func (c jobClient) Query(ctx context.Context, options ...job.QueryOption) iter.S }.Build() for { - jobsMessage, err := c.service.QueryJobs(ctx, filters, page) + jobsMessage, err := c.service.QueryJobs(ctx, filters, page, sortDirection) if err != nil { yield(nil, err) return @@ -574,13 +576,24 @@ func mergeTelemetryQueryOptions(options []TelemetryQueryOption) TelemetryQueryOp return opts } -func (direction SortDirection) protoSortDirection() *workflowsv1.SortDirection { - var protoDirection workflowsv1.SortDirection +func jobQuerySortDirectionToProto(direction job.SortDirection) tileboxv1.SortDirection { + switch direction { + case job.Ascending: + return tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING + case job.Descending: + return tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING + default: + return tileboxv1.SortDirection_SORT_DIRECTION_UNSPECIFIED + } +} + +func (direction SortDirection) protoSortDirection() *tileboxv1.SortDirection { + var protoDirection tileboxv1.SortDirection switch direction { case Ascending: - protoDirection = workflowsv1.SortDirection_SORT_DIRECTION_ASCENDING + protoDirection = tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING case Descending: - protoDirection = workflowsv1.SortDirection_SORT_DIRECTION_DESCENDING + protoDirection = tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING default: return nil } diff --git a/workflows/v1/jobs_test.go b/workflows/v1/jobs_test.go index 98b24d0..a679494 100644 --- a/workflows/v1/jobs_test.go +++ b/workflows/v1/jobs_test.go @@ -21,7 +21,11 @@ import ( type mockJobService struct { JobService - reqs []*workflowsv1.SubmitJobRequest + reqs []*workflowsv1.SubmitJobRequest + queryPages []*workflowsv1.QueryJobsResponse + queryFilters []*workflowsv1.QueryFilters + queryPageReqs []*tileboxv1.Pagination + querySortReqs []tileboxv1.SortDirection } type mockTelemetryService struct { @@ -31,8 +35,8 @@ type mockTelemetryService struct { spanPages []*workflowsv1.PaginatedSpansData logPageReqs []*tileboxv1.Pagination spanPageReqs []*tileboxv1.Pagination - logSortReqs []*workflowsv1.SortDirection - spanSortReqs []*workflowsv1.SortDirection + logSortReqs []*tileboxv1.SortDirection + spanSortReqs []*tileboxv1.SortDirection queryLogJobID uuid.UUID querySpanJobID uuid.UUID } @@ -45,7 +49,19 @@ func (m *mockJobService) SubmitJob(_ context.Context, req *workflowsv1.SubmitJob }.Build(), nil } -func (m *mockTelemetryService) QueryJobLogs(_ context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) { +func (m *mockJobService) QueryJobs(_ context.Context, filters *workflowsv1.QueryFilters, page *tileboxv1.Pagination, sortDirection tileboxv1.SortDirection) (*workflowsv1.QueryJobsResponse, error) { + m.queryFilters = append(m.queryFilters, filters) + m.queryPageReqs = append(m.queryPageReqs, page) + m.querySortReqs = append(m.querySortReqs, sortDirection) + + pageIndex := len(m.queryPageReqs) - 1 + if pageIndex >= len(m.queryPages) { + return workflowsv1.QueryJobsResponse_builder{}.Build(), nil + } + return m.queryPages[pageIndex], nil +} + +func (m *mockTelemetryService) QueryJobLogs(_ context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) { m.queryLogJobID = jobID m.logPageReqs = append(m.logPageReqs, page) m.logSortReqs = append(m.logSortReqs, sortDirection) @@ -54,7 +70,7 @@ func (m *mockTelemetryService) QueryJobLogs(_ context.Context, jobID uuid.UUID, return m.logPages[pageIndex], nil } -func (m *mockTelemetryService) QueryJobSpans(_ context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) { +func (m *mockTelemetryService) QueryJobSpans(_ context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) { m.querySpanJobID = jobID m.spanPageReqs = append(m.spanPageReqs, page) m.spanSortReqs = append(m.spanSortReqs, sortDirection) @@ -169,6 +185,22 @@ func Test_jobClient_Query(t *testing.T) { assert.Equal(t, job.Completed, firstJob.State) } +func Test_jobClient_Query_WithSortDirection(t *testing.T) { + service := &mockJobService{} + client := jobClient{service: service} + timeInterval := query.NewTimeInterval( + time.Date(2024, time.February, 1, 0, 0, 0, 0, time.UTC), + time.Date(2025, time.February, 1, 0, 0, 0, 0, time.UTC), + ) + + jobs, err := Collect(client.Query(context.Background(), job.WithTemporalExtent(timeInterval), job.WithSortDirection(job.Ascending))) + require.NoError(t, err) + assert.Empty(t, jobs) + + require.Len(t, service.querySortReqs, 1) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING, service.querySortReqs[0]) +} + func Test_jobClient_QueryLogs(t *testing.T) { jobID := uuid.MustParse("01994da3-779b-7d01-a570-2a04d0ac163b") nextPage := tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b"))}.Build() @@ -193,7 +225,7 @@ func Test_jobClient_QueryLogs(t *testing.T) { require.Len(t, service.logPageReqs, 2) assert.Nil(t, service.logPageReqs[0]) assert.Equal(t, nextPage, service.logPageReqs[1]) - assert.Equal(t, []*workflowsv1.SortDirection{nil, nil}, service.logSortReqs) + assert.Equal(t, []*tileboxv1.SortDirection{nil, nil}, service.logSortReqs) assert.Equal(t, "first log", logs[0].Body) assert.Equal(t, 10, logs[0].SeverityNumber) assert.Equal(t, "INFO", logs[0].SeverityText) @@ -228,8 +260,8 @@ func Test_jobClient_QueryLogs_WithOptions(t *testing.T) { assert.Equal(t, int64(1), service.logPageReqs[1].GetLimit()) assert.Equal(t, nextPage.GetStartingAfter(), service.logPageReqs[1].GetStartingAfter()) require.Len(t, service.logSortReqs, 2) - assert.Equal(t, workflowsv1.SortDirection_SORT_DIRECTION_DESCENDING, *service.logSortReqs[0]) - assert.Equal(t, workflowsv1.SortDirection_SORT_DIRECTION_DESCENDING, *service.logSortReqs[1]) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING, *service.logSortReqs[0]) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING, *service.logSortReqs[1]) assert.Equal(t, "first log", logs[0].Body) assert.Equal(t, "second log", logs[1].Body) } @@ -252,7 +284,7 @@ func Test_jobClient_QuerySpans(t *testing.T) { assert.Equal(t, jobID, service.querySpanJobID) require.Len(t, service.spanPageReqs, 1) assert.Nil(t, service.spanPageReqs[0]) - assert.Equal(t, []*workflowsv1.SortDirection{nil}, service.spanSortReqs) + assert.Equal(t, []*tileboxv1.SortDirection{nil}, service.spanSortReqs) assert.Equal(t, "task/run", spans[0].Name) assert.Equal(t, "010203", spans[0].TraceID) assert.Equal(t, "040506", spans[0].SpanID) @@ -285,7 +317,7 @@ func Test_jobClient_QuerySpans_WithOptions(t *testing.T) { require.Len(t, service.spanPageReqs, 1) assert.Equal(t, int64(1), service.spanPageReqs[0].GetLimit()) require.Len(t, service.spanSortReqs, 1) - assert.Equal(t, workflowsv1.SortDirection_SORT_DIRECTION_ASCENDING, *service.spanSortReqs[0]) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING, *service.spanSortReqs[0]) } func testResourceLogs(body string, severity logsv1.SeverityNumber) *logsv1.ResourceLogs { diff --git a/workflows/v1/service.go b/workflows/v1/service.go index 9654316..2cb8007 100644 --- a/workflows/v1/service.go +++ b/workflows/v1/service.go @@ -153,12 +153,12 @@ type JobService interface { GetJob(ctx context.Context, jobID uuid.UUID) (*workflowsv1.Job, error) RetryJob(ctx context.Context, jobID uuid.UUID) (*workflowsv1.RetryJobResponse, error) CancelJob(ctx context.Context, jobID uuid.UUID) error - QueryJobs(ctx context.Context, filters *workflowsv1.QueryFilters, page *tileboxv1.Pagination) (*workflowsv1.QueryJobsResponse, error) + QueryJobs(ctx context.Context, filters *workflowsv1.QueryFilters, page *tileboxv1.Pagination, sortDirection tileboxv1.SortDirection) (*workflowsv1.QueryJobsResponse, error) } type TelemetryService interface { - QueryJobLogs(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) - QueryJobSpans(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) + QueryJobLogs(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) + QueryJobSpans(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) } var _ JobService = &jobService{} @@ -228,11 +228,12 @@ func (s *jobService) CancelJob(ctx context.Context, jobID uuid.UUID) error { }) } -func (s *jobService) QueryJobs(ctx context.Context, filters *workflowsv1.QueryFilters, page *tileboxv1.Pagination) (*workflowsv1.QueryJobsResponse, error) { +func (s *jobService) QueryJobs(ctx context.Context, filters *workflowsv1.QueryFilters, page *tileboxv1.Pagination, sortDirection tileboxv1.SortDirection) (*workflowsv1.QueryJobsResponse, error) { return observability.WithSpanResult(ctx, s.tracer, "workflows/jobs/query", func(ctx context.Context) (*workflowsv1.QueryJobsResponse, error) { res, err := s.jobClient.QueryJobs(ctx, connect.NewRequest(workflowsv1.QueryJobsRequest_builder{ - Filters: filters, - Page: page, + Filters: filters, + Page: page, + SortDirection: sortDirection, }.Build())) if err != nil { return nil, fmt.Errorf("failed to query jobs: %w", err) @@ -256,7 +257,7 @@ func newTelemetryService(telemetryClient workflowsv1connect.TelemetryQueryServic } } -func (s *telemetryService) QueryJobLogs(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) { +func (s *telemetryService) QueryJobLogs(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedLogsData, error) { return observability.WithSpanResult(ctx, s.tracer, "workflows/jobs/query_logs", func(ctx context.Context) (*workflowsv1.PaginatedLogsData, error) { res, err := s.telemetryClient.QueryJobLogs(ctx, connect.NewRequest(workflowsv1.QueryJobLogsRequest_builder{ JobId: tileboxv1.NewUUID(jobID), @@ -271,7 +272,7 @@ func (s *telemetryService) QueryJobLogs(ctx context.Context, jobID uuid.UUID, pa }) } -func (s *telemetryService) QueryJobSpans(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *workflowsv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) { +func (s *telemetryService) QueryJobSpans(ctx context.Context, jobID uuid.UUID, page *tileboxv1.Pagination, sortDirection *tileboxv1.SortDirection) (*workflowsv1.PaginatedSpansData, error) { return observability.WithSpanResult(ctx, s.tracer, "workflows/jobs/query_spans", func(ctx context.Context) (*workflowsv1.PaginatedSpansData, error) { res, err := s.telemetryClient.QueryJobSpans(ctx, connect.NewRequest(workflowsv1.QueryJobSpansRequest_builder{ JobId: tileboxv1.NewUUID(jobID), From d1f204985e30fb16631ad9cbb48160202251e8fb Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Wed, 20 May 2026 15:41:54 +0200 Subject: [PATCH 3/4] Manual pagination query endpoints for jobs, logs and traces --- CHANGELOG.md | 16 +- README.md | 35 ++ .../workflows/observability/query/main.go | 3 +- workflows/v1/job/options.go | 269 ++++++++++- workflows/v1/job/options_test.go | 146 +++++- workflows/v1/jobs.go | 445 ++++++++++++------ workflows/v1/jobs_test.go | 187 +++++++- 7 files changed, 931 insertions(+), 170 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a1ccdbd..ce0c06b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [0.6.0] - 2026-05-20 + +### Added + +- `workflows`: Added cursor and limit query options plus `QueryPage`, `QueryLogsPage`, and `QuerySpansPage` for manual pagination of job, log, and span queries. +- `workflows`: Added `job.WithTaskStates(...)` to filter job queries by task state. +- `workflows`: Added string and JSON representations for job and task states. + +### Changed + +- `workflows`: Changed job, log, and span queries to support cursor-based pagination while keeping existing sequence APIs auto-paginated. +- `workflows`: Moved telemetry query options to the `workflows/v1/job` package so job and telemetry queries share `job.WithLimit`, `job.WithCursor`, and `job.WithSortDirection`. + ## [0.5.0] - 2026-05-11 ### Added @@ -90,7 +103,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added support for Tilebox Observability, including logging and tracing helpers. - Added examples for using the library. -[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.5.0...HEAD +[Unreleased]: https://github.com/tilebox/tilebox-go/compare/v0.6.0...HEAD +[0.6.0]: https://github.com/tilebox/tilebox-go/compare/v0.5.0...v0.6.0 [0.5.0]: https://github.com/tilebox/tilebox-go/compare/v0.4.0...v0.5.0 [0.4.0]: https://github.com/tilebox/tilebox-go/compare/v0.3.2...v0.4.0 [0.3.2]: https://github.com/tilebox/tilebox-go/compare/v0.3.1...v0.3.2 diff --git a/README.md b/README.md index 58d23a1..80fd07b 100644 --- a/README.md +++ b/README.md @@ -162,6 +162,41 @@ func main() { } ``` +### Querying Job Telemetry + +After a job has been executed, we can query runner logs associated with it. + +```go +package main + +import ( + "context" + "log/slog" + + "github.com/google/uuid" + "github.com/tilebox/tilebox-go/workflows/v1" + "github.com/tilebox/tilebox-go/workflows/v1/job" +) + +func main() { + ctx := context.Background() + client := workflows.NewClient() + jobID := uuid.MustParse("019e070c-63ba-1c7e-5f1d-65be3e22d52a") + + for logRecord, err := range client.Jobs.QueryLogs(ctx, jobID, + job.WithLimit(100), + job.WithSortDirection(job.Ascending), + ) { + if err != nil { + slog.ErrorContext(ctx, "failed to query job logs", slog.Any("error", err)) + return + } + + slog.InfoContext(ctx, "job log", slog.String("body", logRecord.Body)) + } +} +``` + ## License Distributed under the MIT License (`The MIT License`). diff --git a/examples/workflows/observability/query/main.go b/examples/workflows/observability/query/main.go index 5e193e3..0976d25 100644 --- a/examples/workflows/observability/query/main.go +++ b/examples/workflows/observability/query/main.go @@ -8,6 +8,7 @@ import ( "github.com/google/uuid" "github.com/tilebox/tilebox-go/workflows/v1" + "github.com/tilebox/tilebox-go/workflows/v1/job" ) func main() { @@ -18,7 +19,7 @@ func main() { fmt.Println("Log messages:") //nolint:forbidigo // This example intentionally prints query results to stdout. fmt.Printf("%-8s %-30s %s\n", "LEVEL", "TIME", "BODY") //nolint:forbidigo // This example intentionally prints query results to stdout. - for logRecord, err := range client.Jobs.QueryLogs(ctx, jobID, workflows.WithSortDirection(workflows.Ascending)) { + for logRecord, err := range client.Jobs.QueryLogs(ctx, jobID, job.WithSortDirection(job.Ascending)) { if err != nil { slog.ErrorContext(ctx, "failed to query job logs", slog.Any("error", err)) return diff --git a/workflows/v1/job/options.go b/workflows/v1/job/options.go index 29b4fee..8490733 100644 --- a/workflows/v1/job/options.go +++ b/workflows/v1/job/options.go @@ -1,6 +1,9 @@ package job // import "github.com/tilebox/tilebox-go/workflows/v1/job" import ( + "encoding/json" + "fmt" + "github.com/google/uuid" "github.com/tilebox/tilebox-go/query" ) @@ -50,65 +53,285 @@ const ( Canceled State = 6 ) +func (state State) String() string { + switch state { + case Submitted: + return "submitted" + case Running: + return "running" + case Started: + return "started" + case Completed: + return "completed" + case Failed: + return "failed" + case Canceled: + return "canceled" + default: + return "unspecified" + } +} + +func (state State) MarshalJSON() ([]byte, error) { + return json.Marshal(state.String()) +} + +// TaskState is a task state value for use in query filters. +type TaskState int32 + +// TaskState values for use in query filters. +const ( + // TaskQueued means the task is queued and waiting to be run. + TaskQueued TaskState = 1 + // TaskRunning means the task is currently running on some task runner. + TaskRunning TaskState = 2 + // TaskComputed means the task has been computed and the output is available. + TaskComputed TaskState = 3 + // TaskFailed means the task has failed. + TaskFailed TaskState = 4 + // TaskSkipped means the task has been skipped. + TaskSkipped TaskState = 5 + // TaskFailedOptional means the task has failed, but was marked as optional. + TaskFailedOptional TaskState = 6 +) + +func (state TaskState) String() string { + switch state { + case TaskQueued: + return "queued" + case TaskRunning: + return "running" + case TaskComputed: + return "computed" + case TaskFailed: + return "failed" + case TaskSkipped: + return "skipped" + case TaskFailedOptional: + return "failed_optional" + default: + return "unspecified" + } +} + +func (state TaskState) MarshalJSON() ([]byte, error) { + return json.Marshal(state.String()) +} + // QueryOptions contains the configuration for a Query request. type QueryOptions struct { + // TemporalExtent is the time or ID interval for which jobs should be queried. + // Leave unset to query jobs without a temporal extent filter. TemporalExtent query.TemporalExtent - AutomationIDs []uuid.UUID - States []State - Name string - SortDirection SortDirection + // Cursor starts the query after a cursor returned by a previous page. + Cursor *Cursor + // AutomationIDs filters jobs by the automations that submitted them. + AutomationIDs []uuid.UUID + // States filters jobs by their states. + States []State + // TaskStates filters jobs by the states of their tasks. + TaskStates []TaskState + // Name filters jobs by their name. + Name string + // Limit is the maximum number of jobs to return. + // Leave unset or set to 0 to paginate through and return all jobs. + Limit int64 + // SortDirection is the direction in which jobs should be sorted by submission date. + // Leave unset to let the server choose its default sort direction. + SortDirection SortDirection +} + +type QueryOption interface { + ApplyQueryOption(*QueryOptions) +} + +type queryOptionFunc func(*QueryOptions) + +func (f queryOptionFunc) ApplyQueryOption(cfg *QueryOptions) { + f(cfg) +} + +// TelemetryQueryOptions contains options for querying job observability. +type TelemetryQueryOptions struct { + // Cursor starts the query after a cursor returned by a previous page. + Cursor *Cursor + // Limit is the maximum number of observability records to return. + // Leave unset or set to 0 to paginate through and return all records. + Limit int64 + // SortDirection is the direction in which observability records should be sorted. + // Leave unset to let the server choose its default sort direction. + SortDirection SortDirection +} + +type TelemetryQueryOption interface { + ApplyTelemetryQueryOption(*TelemetryQueryOptions) +} + +type SortDirectionOption interface { + QueryOption + TelemetryQueryOption +} + +type LimitOption interface { + QueryOption + TelemetryQueryOption +} + +type CursorOption interface { + QueryOption + TelemetryQueryOption +} + +// Cursor identifies where to continue a paginated query. +// +// Cursors are returned as NextCursor values from page query methods and should only be reused with the same endpoint, +// filters and sort direction that produced them. +type Cursor struct { + startingAfter uuid.UUID +} + +// NewCursor creates a cursor that starts after the entry with the given ID. +func NewCursor(startingAfter uuid.UUID) *Cursor { + return &Cursor{startingAfter: startingAfter} +} + +// ParseCursor parses a cursor string returned by Cursor.String. +func ParseCursor(value string) (*Cursor, error) { + startingAfter, err := uuid.Parse(value) + if err != nil { + return nil, fmt.Errorf("invalid cursor: %w", err) + } + return NewCursor(startingAfter), nil +} + +// String returns the cursor as a string that can be passed to ParseCursor. +func (c *Cursor) String() string { + if c == nil { + return "" + } + return c.startingAfter.String() +} + +// StartingAfter returns the ID after which the next page should start. +func (c *Cursor) StartingAfter() uuid.UUID { + if c == nil { + return uuid.Nil + } + return c.startingAfter +} + +type cursorOption struct { + cursor *Cursor +} + +func (o cursorOption) ApplyQueryOption(cfg *QueryOptions) { + cfg.Cursor = o.cursor +} + +func (o cursorOption) ApplyTelemetryQueryOption(cfg *TelemetryQueryOptions) { + cfg.Cursor = o.cursor +} + +type limitOption struct { + limit int64 +} + +func (o limitOption) ApplyQueryOption(cfg *QueryOptions) { + if o.limit > 0 { + cfg.Limit = o.limit + } } -type QueryOption func(*QueryOptions) +func (o limitOption) ApplyTelemetryQueryOption(cfg *TelemetryQueryOptions) { + if o.limit > 0 { + cfg.Limit = o.limit + } +} + +type sortDirectionOption struct { + direction SortDirection +} -// SortDirection specifies the sort direction for job query results. +func (o sortDirectionOption) ApplyQueryOption(cfg *QueryOptions) { + if o.direction != 0 { + cfg.SortDirection = o.direction + } +} + +func (o sortDirectionOption) ApplyTelemetryQueryOption(cfg *TelemetryQueryOptions) { + if o.direction != 0 { + cfg.SortDirection = o.direction + } +} + +// SortDirection specifies the sort direction for job and observability query results. type SortDirection int32 // SortDirection values. const ( _ SortDirection = iota - // Ascending sorts jobs oldest first. + // Ascending sorts query results oldest first. Ascending - // Descending sorts jobs newest first. + // Descending sorts query results newest first. Descending ) -// WithTemporalExtent specifies the time interval for which jobs should be queried. -// Right now, a temporal extent is required for every query. +// WithTemporalExtent specifies the time or ID interval for which jobs should be queried. func WithTemporalExtent(temporalExtent query.TemporalExtent) QueryOption { - return func(cfg *QueryOptions) { + return queryOptionFunc(func(cfg *QueryOptions) { cfg.TemporalExtent = temporalExtent - } + }) } // WithAutomationIDs specifies multiple automation IDs to filter jobs by. // Only jobs submitted by any of the specified automations will be returned. func WithAutomationIDs(automationIDs ...uuid.UUID) QueryOption { - return func(cfg *QueryOptions) { + return queryOptionFunc(func(cfg *QueryOptions) { cfg.AutomationIDs = append(cfg.AutomationIDs, automationIDs...) - } + }) } // WithJobStates filters jobs by their state. // Only jobs in any of the specified states will be returned. func WithJobStates(states ...State) QueryOption { - return func(cfg *QueryOptions) { + return queryOptionFunc(func(cfg *QueryOptions) { cfg.States = append(cfg.States, states...) - } + }) +} + +// WithTaskStates filters jobs by task state. +// Only jobs with at least one task in any of the specified states will be returned. +func WithTaskStates(states ...TaskState) QueryOption { + return queryOptionFunc(func(cfg *QueryOptions) { + cfg.TaskStates = append(cfg.TaskStates, states...) + }) } // WithJobName filters jobs by name. func WithJobName(name string) QueryOption { - return func(cfg *QueryOptions) { + return queryOptionFunc(func(cfg *QueryOptions) { cfg.Name = name - } + }) } -// WithSortDirection sets the sort direction for job query results. +// WithCursor starts the query after a cursor returned by a previous page. +func WithCursor(cursor *Cursor) CursorOption { + return cursorOption{cursor: cursor} +} + +// WithLimit limits the number of query results returned. +// +// For auto-paginated query methods, the limit applies to the total number of results yielded. For page query methods, +// the limit applies to the single page returned. +// +// Defaults to unlimited. +func WithLimit(limit int64) LimitOption { + return limitOption{limit: limit} +} + +// WithSortDirection sets the sort direction for job and observability query results. // // Defaults to the server default. -func WithSortDirection(direction SortDirection) QueryOption { - return func(cfg *QueryOptions) { - cfg.SortDirection = direction - } +func WithSortDirection(direction SortDirection) SortDirectionOption { + return sortDirectionOption{direction: direction} } diff --git a/workflows/v1/job/options_test.go b/workflows/v1/job/options_test.go index 88a566e..f7676cc 100644 --- a/workflows/v1/job/options_test.go +++ b/workflows/v1/job/options_test.go @@ -1,14 +1,70 @@ package job import ( + "encoding/json" "testing" "time" "github.com/google/uuid" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/tilebox/tilebox-go/query" ) +func TestStateStringAndMarshalJSON(t *testing.T) { + tests := []struct { + name string + state State + want string + }{ + {name: "submitted", state: Submitted, want: "submitted"}, + {name: "running", state: Running, want: "running"}, + {name: "started", state: Started, want: "started"}, + {name: "completed", state: Completed, want: "completed"}, + {name: "failed", state: Failed, want: "failed"}, + {name: "canceled", state: Canceled, want: "canceled"}, + {name: "unspecified", state: 0, want: "unspecified"}, + {name: "unknown", state: 99, want: "unspecified"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, test.state.String()) + + data, err := json.Marshal(test.state) + require.NoError(t, err) + assert.JSONEq(t, `"`+test.want+`"`, string(data)) + }) + } +} + +func TestTaskStateStringAndMarshalJSON(t *testing.T) { + tests := []struct { + name string + state TaskState + want string + }{ + {name: "queued", state: TaskQueued, want: "queued"}, + {name: "running", state: TaskRunning, want: "running"}, + {name: "computed", state: TaskComputed, want: "computed"}, + {name: "failed", state: TaskFailed, want: "failed"}, + {name: "skipped", state: TaskSkipped, want: "skipped"}, + {name: "failed optional", state: TaskFailedOptional, want: "failed_optional"}, + {name: "unspecified", state: 0, want: "unspecified"}, + {name: "unknown", state: 99, want: "unspecified"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, test.state.String()) + + data, err := json.Marshal(test.state) + require.NoError(t, err) + assert.JSONEq(t, `"`+test.want+`"`, string(data)) + }) + } +} + func Test_SubmitJobOptions(t *testing.T) { tests := []struct { name string @@ -49,6 +105,7 @@ func Test_QueryOptions(t *testing.T) { now := time.Now() automationID := uuid.New() anotherAutomationID := uuid.New() + cursor := NewCursor(uuid.New()) tests := []struct { name string @@ -90,6 +147,16 @@ func Test_QueryOptions(t *testing.T) { States: []State{Submitted, Running, Started}, }, }, + { + name: "with task states", + options: []QueryOption{ + WithTaskStates(TaskQueued, TaskRunning), + WithTaskStates(TaskComputed), + }, + want: QueryOptions{ + TaskStates: []TaskState{TaskQueued, TaskRunning, TaskComputed}, + }, + }, { name: "with job name", options: []QueryOption{ @@ -99,6 +166,24 @@ func Test_QueryOptions(t *testing.T) { Name: "my-job", }, }, + { + name: "with cursor", + options: []QueryOption{ + WithCursor(cursor), + }, + want: QueryOptions{ + Cursor: cursor, + }, + }, + { + name: "with limit", + options: []QueryOption{ + WithLimit(10), + }, + want: QueryOptions{ + Limit: 10, + }, + }, { name: "with sort direction", options: []QueryOption{ @@ -113,9 +198,68 @@ func Test_QueryOptions(t *testing.T) { t.Run(tt.name, func(t *testing.T) { var options QueryOptions for _, option := range tt.options { - option(&options) + option.ApplyQueryOption(&options) + } + assert.Equal(t, tt.want, options) + }) + } +} + +func Test_TelemetryQueryOptions(t *testing.T) { + cursor := NewCursor(uuid.New()) + + tests := []struct { + name string + options []TelemetryQueryOption + want TelemetryQueryOptions + }{ + { + name: "with cursor", + options: []TelemetryQueryOption{ + WithCursor(cursor), + }, + want: TelemetryQueryOptions{ + Cursor: cursor, + }, + }, + { + name: "with limit", + options: []TelemetryQueryOption{ + WithLimit(10), + }, + want: TelemetryQueryOptions{ + Limit: 10, + }, + }, + { + name: "with sort direction", + options: []TelemetryQueryOption{ + WithSortDirection(Ascending), + }, + want: TelemetryQueryOptions{ + SortDirection: Ascending, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var options TelemetryQueryOptions + for _, option := range tt.options { + option.ApplyTelemetryQueryOption(&options) } assert.Equal(t, tt.want, options) }) } } + +func TestCursor(t *testing.T) { + id := uuid.New() + cursor := NewCursor(id) + + assert.Equal(t, id, cursor.StartingAfter()) + assert.Equal(t, id.String(), cursor.String()) + + parsed, err := ParseCursor(cursor.String()) + assert.NoError(t, err) + assert.Equal(t, id, parsed.StartingAfter()) +} diff --git a/workflows/v1/jobs.go b/workflows/v1/jobs.go index 2b66d39..c5efd65 100644 --- a/workflows/v1/jobs.go +++ b/workflows/v1/jobs.go @@ -154,52 +154,64 @@ type SpanEvent struct { Attributes map[string]any } -// SortDirection specifies the sort direction for observability query results. -type SortDirection int32 - -// SortDirection values. -const ( - _ SortDirection = iota - // Ascending sorts observability query results oldest first. - Ascending - // Descending sorts observability query results newest first. - Descending -) - -// TelemetryQueryOption contains options for querying job observability. -type TelemetryQueryOption struct { - // Limit is the maximum number of observability records to return. - // Leave unset or set to 0 to paginate through and return all records. - Limit int64 - // SortDirection is the direction in which observability records should be sorted. - // Leave unset to let the server choose its default sort direction. - SortDirection SortDirection +// JobPage is a single page of jobs returned by a manual page query. +type JobPage struct { + // Jobs is the list of jobs in this page. + Jobs []*Job + // NextCursor can be used to request the next page. Nil means there are no more pages. + NextCursor *job.Cursor } -// WithLimit limits the total number of observability records returned. -// -// Defaults to unlimited. -func WithLimit(limit int64) TelemetryQueryOption { - return TelemetryQueryOption{Limit: limit} +// LogPage is a single page of log records returned by a manual page query. +type LogPage struct { + // Logs is the list of log records in this page. + Logs []*LogRecord + // NextCursor can be used to request the next page. Nil means there are no more pages. + NextCursor *job.Cursor } -// WithSortDirection sets the sort direction for observability query results. -// -// Defaults to the server default. -func WithSortDirection(direction SortDirection) TelemetryQueryOption { - return TelemetryQueryOption{SortDirection: direction} +// SpanPage is a single page of spans returned by a manual page query. +type SpanPage struct { + // Spans is the list of spans in this page. + Spans []*Span + // NextCursor can be used to request the next page. Nil means there are no more pages. + NextCursor *job.Cursor } // TaskState values. const ( - _ TaskState = iota - TaskQueued // The task is queued and waiting to be run. - TaskRunning // The task is currently running on some task runner. - TaskComputed // The task has been computed and the output is available. - TaskFailed // The task has failed. - TaskCancelled // The task has been cancelled due to user request. + _ TaskState = iota + TaskQueued // The task is queued and waiting to be run. + TaskRunning // The task is currently running on some task runner. + TaskComputed // The task has been computed and the output is available. + TaskFailed // The task has failed. + TaskSkipped // The task has been skipped. + TaskFailedOptional // The task has failed, but was marked as optional. ) +func (state TaskState) String() string { + switch state { + case TaskQueued: + return "queued" + case TaskRunning: + return "running" + case TaskComputed: + return "computed" + case TaskFailed: + return "failed" + case TaskSkipped: + return "skipped" + case TaskFailedOptional: + return "failed_optional" + default: + return "unspecified" + } +} + +func (state TaskState) MarshalJSON() ([]byte, error) { + return json.Marshal(state.String()) +} + type JobClient interface { // Submit submits a job to a cluster. // @@ -225,39 +237,76 @@ type JobClient interface { // Documentation: https://docs.tilebox.com/workflows/concepts/jobs#cancellation Cancel(ctx context.Context, jobID uuid.UUID) error - // Query returns a list of all jobs within the given interval. + // Query returns a list of jobs matching the provided options. // // Options: - // - job.WithTemporalExtent: specifies the time or ID interval for which jobs should be queried (Required) + // - job.WithTemporalExtent: specifies the time or ID interval for which jobs should be queried. (Optional) // - job.WithAutomationIDs: specifies the automation IDs to filter jobs by. Only jobs submitted by the specified // automation will be returned. (Optional) // - job.WithJobStates: specifies the job state to filter jobs by. Only jobs with the specified state will be returned. (Optional) + // - job.WithTaskStates: specifies task states to filter jobs by. Only jobs with at least one task in one of the + // specified states will be returned. (Optional) // - job.WithJobName: specifies the job name to filter jobs by. Only jobs with the specified name will be returned. (Optional) + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the total number of jobs returned. Defaults to unlimited. // - job.WithSortDirection: specifies the direction to sort jobs by submission date. (Optional) // - // The jobs are lazily loaded and returned as a sequence. + // Pagination is handled automatically under the hood. The jobs are lazily loaded and returned as a sequence. // The output sequence can be transformed into a slice using Collect. Query(ctx context.Context, options ...job.QueryOption) iter.Seq2[*Job, error] + // QueryPage returns a single page of jobs matching the provided options. + // + // Options: + // - job.WithTemporalExtent: specifies the time or ID interval for which jobs should be queried. (Optional) + // - job.WithAutomationIDs: specifies the automation IDs to filter jobs by. Only jobs submitted by the specified + // automation will be returned. (Optional) + // - job.WithJobStates: specifies the job state to filter jobs by. Only jobs with the specified state will be returned. (Optional) + // - job.WithTaskStates: specifies task states to filter jobs by. Only jobs with at least one task in one of the + // specified states will be returned. (Optional) + // - job.WithJobName: specifies the job name to filter jobs by. Only jobs with the specified name will be returned. (Optional) + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the number of jobs returned in this page. Defaults to the server default. + // - job.WithSortDirection: specifies the direction to sort jobs by submission date. (Optional) + QueryPage(ctx context.Context, options ...job.QueryOption) (*JobPage, error) + // QueryLogs returns the logs emitted while running a job. // // Options: - // - TelemetryQueryOption.Limit: limits the total number of records returned. Defaults to unlimited. - // - TelemetryQueryOption.SortDirection: sorts records ascending or descending. Defaults to the server default. + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the total number of records returned. Defaults to unlimited. + // - job.WithSortDirection: sorts records ascending or descending. Defaults to the server default. // - // The logs are lazily loaded and returned as a sequence. + // Pagination is handled automatically under the hood. The logs are lazily loaded and returned as a sequence. // The output sequence can be transformed into a slice using Collect. - QueryLogs(ctx context.Context, jobID uuid.UUID, options ...TelemetryQueryOption) iter.Seq2[*LogRecord, error] + QueryLogs(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*LogRecord, error] + + // QueryLogsPage returns a single page of logs emitted while running a job. + // + // Options: + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the number of records returned in this page. Defaults to the server default. + // - job.WithSortDirection: sorts records ascending or descending. Defaults to the server default. + QueryLogsPage(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) (*LogPage, error) // QuerySpans returns the spans emitted while running a job. // // Options: - // - TelemetryQueryOption.Limit: limits the total number of records returned. Defaults to unlimited. - // - TelemetryQueryOption.SortDirection: sorts records ascending or descending. Defaults to the server default. + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the total number of records returned. Defaults to unlimited. + // - job.WithSortDirection: sorts records ascending or descending. Defaults to the server default. // - // The spans are lazily loaded and returned as a sequence. + // Pagination is handled automatically under the hood. The spans are lazily loaded and returned as a sequence. // The output sequence can be transformed into a slice using Collect. - QuerySpans(ctx context.Context, jobID uuid.UUID, options ...TelemetryQueryOption) iter.Seq2[*Span, error] + QuerySpans(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*Span, error] + + // QuerySpansPage returns a single page of spans emitted while running a job. + // + // Options: + // - job.WithCursor: starts the query after a cursor returned by a previous page. (Optional) + // - job.WithLimit: limits the number of records returned in this page. Defaults to the server default. + // - job.WithSortDirection: sorts records ascending or descending. Defaults to the server default. + QuerySpansPage(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) (*SpanPage, error) } var _ JobClient = &jobClient{} @@ -311,92 +360,139 @@ func (c jobClient) Cancel(ctx context.Context, jobID uuid.UUID) error { func (c jobClient) Query(ctx context.Context, options ...job.QueryOption) iter.Seq2[*Job, error] { opts := &job.QueryOptions{} for _, option := range options { - option(opts) - } - - if opts.TemporalExtent == nil { - return func(yield func(*Job, error) bool) { - // right now we return an error, in the future we might want to support queries without a temporal extent - yield(nil, errors.New("temporal extent is required")) - } + option.ApplyQueryOption(opts) } return func(yield func(*Job, error) bool) { - var page *tileboxv1.Pagination // nil for the first request - sortDirection := jobQuerySortDirectionToProto(opts.SortDirection) - - // we already validated that TemporalExtent is not nil - timeInterval := opts.TemporalExtent.ToProtoTimeInterval() - idInterval := opts.TemporalExtent.ToProtoIDInterval() - - if timeInterval == nil && idInterval == nil { - yield(nil, errors.New("invalid temporal extent")) - return - } - - automationIDs := lo.Map(opts.AutomationIDs, func(id uuid.UUID, _ int) *tileboxv1.ID { - return tileboxv1.NewUUID(id) - }) - - states := lo.Map(opts.States, func(state job.State, _ int) workflowsv1.JobState { - return workflowsv1.JobState(state) - }) - - filters := workflowsv1.QueryFilters_builder{ - TimeInterval: timeInterval, - IdInterval: idInterval, - AutomationIds: automationIDs, - States: states, - Name: opts.Name, - }.Build() + cursor := opts.Cursor + remaining := opts.Limit for { - jobsMessage, err := c.service.QueryJobs(ctx, filters, page, sortDirection) + pageOpts := *opts + pageOpts.Cursor = cursor + if opts.Limit > 0 { + if remaining == 0 { + break + } + pageOpts.Limit = remaining + } + + jobsPage, err := c.queryPage(ctx, &pageOpts) if err != nil { yield(nil, err) return } - for _, jobMessage := range jobsMessage.GetJobs() { - if !yield(protoToJob(jobMessage), nil) { + for _, result := range jobsPage.Jobs { + if opts.Limit > 0 && remaining == 0 { + return + } + + if !yield(result, nil) { return } + if opts.Limit > 0 { + remaining-- + } } - page = jobsMessage.GetNextPage() - if page == nil { + cursor = jobsPage.NextCursor + if cursor == nil || (opts.Limit > 0 && remaining == 0) { break } } } } -func (c jobClient) QueryLogs(ctx context.Context, jobID uuid.UUID, options ...TelemetryQueryOption) iter.Seq2[*LogRecord, error] { - opts := mergeTelemetryQueryOptions(options) - sortDirection := opts.SortDirection.protoSortDirection() +func (c jobClient) QueryPage(ctx context.Context, options ...job.QueryOption) (*JobPage, error) { + opts := &job.QueryOptions{} + for _, option := range options { + option.ApplyQueryOption(opts) + } + return c.queryPage(ctx, opts) +} + +func (c jobClient) queryPage(ctx context.Context, opts *job.QueryOptions) (*JobPage, error) { + sortDirection := jobQuerySortDirectionToProto(opts.SortDirection) + + var timeInterval *tileboxv1.TimeInterval + var idInterval *tileboxv1.IDInterval + if opts.TemporalExtent != nil { + timeInterval = opts.TemporalExtent.ToProtoTimeInterval() + idInterval = opts.TemporalExtent.ToProtoIDInterval() + + if timeInterval == nil && idInterval == nil { + return nil, errors.New("invalid temporal extent") + } + } + + automationIDs := lo.Map(opts.AutomationIDs, func(id uuid.UUID, _ int) *tileboxv1.ID { + return tileboxv1.NewUUID(id) + }) + + states := lo.Map(opts.States, func(state job.State, _ int) workflowsv1.JobState { + return workflowsv1.JobState(state) + }) + taskStates := lo.Map(opts.TaskStates, func(state job.TaskState, _ int) workflowsv1.TaskState { + return workflowsv1.TaskState(state) + }) + + filters := workflowsv1.QueryFilters_builder{ + TimeInterval: timeInterval, + IdInterval: idInterval, + AutomationIds: automationIDs, + States: states, + TaskStates: taskStates, + Name: opts.Name, + }.Build() + + jobsMessage, err := c.service.QueryJobs(ctx, filters, paginationFromOptions(opts.Limit, opts.Cursor), sortDirection) + if err != nil { + return nil, err + } + + jobs := make([]*Job, 0, len(jobsMessage.GetJobs())) + for _, jobMessage := range jobsMessage.GetJobs() { + jobs = append(jobs, protoToJob(jobMessage)) + } + + return &JobPage{ + Jobs: jobs, + NextCursor: cursorFromPagination(jobsMessage.GetNextPage()), + }, nil +} + +func (c jobClient) QueryLogs(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*LogRecord, error] { + opts := &job.TelemetryQueryOptions{} + for _, option := range options { + option.ApplyTelemetryQueryOption(opts) + } return func(yield func(*LogRecord, error) bool) { - var page *tileboxv1.Pagination // nil for the first request + cursor := opts.Cursor remaining := opts.Limit for { - page = telemetryQueryPage(page, remaining) - logsMessage, err := c.telemetryService.QueryJobLogs(ctx, jobID, page, sortDirection) + pageOpts := *opts + pageOpts.Cursor = cursor + if opts.Limit > 0 { + if remaining == 0 { + break + } + pageOpts.Limit = remaining + } + + logsPage, err := c.queryLogsPage(ctx, jobID, &pageOpts) if err != nil { yield(nil, err) return } - for _, resourceLogs := range logsMessage.GetResourceLogs() { + for _, logRecord := range logsPage.Logs { if opts.Limit > 0 && remaining == 0 { return } - logRecord, err := protoToLogRecord(resourceLogs) - if err != nil { - yield(nil, err) - return - } if !yield(logRecord, nil) { return } @@ -405,40 +501,74 @@ func (c jobClient) QueryLogs(ctx context.Context, jobID uuid.UUID, options ...Te } } - page = logsMessage.GetNextPage() - if page == nil || (opts.Limit > 0 && remaining == 0) { + cursor = logsPage.NextCursor + if cursor == nil || (opts.Limit > 0 && remaining == 0) { break } } } } -func (c jobClient) QuerySpans(ctx context.Context, jobID uuid.UUID, options ...TelemetryQueryOption) iter.Seq2[*Span, error] { - opts := mergeTelemetryQueryOptions(options) - sortDirection := opts.SortDirection.protoSortDirection() +func (c jobClient) QueryLogsPage(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) (*LogPage, error) { + opts := &job.TelemetryQueryOptions{} + for _, option := range options { + option.ApplyTelemetryQueryOption(opts) + } + return c.queryLogsPage(ctx, jobID, opts) +} + +func (c jobClient) queryLogsPage(ctx context.Context, jobID uuid.UUID, opts *job.TelemetryQueryOptions) (*LogPage, error) { + logsMessage, err := c.telemetryService.QueryJobLogs(ctx, jobID, paginationFromOptions(opts.Limit, opts.Cursor), sortDirectionToProto(opts.SortDirection)) + if err != nil { + return nil, err + } + + logs := make([]*LogRecord, 0, len(logsMessage.GetResourceLogs())) + for _, resourceLogs := range logsMessage.GetResourceLogs() { + logRecord, err := protoToLogRecord(resourceLogs) + if err != nil { + return nil, err + } + logs = append(logs, logRecord) + } + + return &LogPage{ + Logs: logs, + NextCursor: cursorFromPagination(logsMessage.GetNextPage()), + }, nil +} + +func (c jobClient) QuerySpans(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*Span, error] { + opts := &job.TelemetryQueryOptions{} + for _, option := range options { + option.ApplyTelemetryQueryOption(opts) + } return func(yield func(*Span, error) bool) { - var page *tileboxv1.Pagination // nil for the first request + cursor := opts.Cursor remaining := opts.Limit for { - page = telemetryQueryPage(page, remaining) - spansMessage, err := c.telemetryService.QueryJobSpans(ctx, jobID, page, sortDirection) + pageOpts := *opts + pageOpts.Cursor = cursor + if opts.Limit > 0 { + if remaining == 0 { + break + } + pageOpts.Limit = remaining + } + + spansPage, err := c.querySpansPage(ctx, jobID, &pageOpts) if err != nil { yield(nil, err) return } - for _, resourceSpans := range spansMessage.GetResourceSpans() { + for _, span := range spansPage.Spans { if opts.Limit > 0 && remaining == 0 { return } - span, err := protoToSpan(resourceSpans) - if err != nil { - yield(nil, err) - return - } if !yield(span, nil) { return } @@ -447,14 +577,43 @@ func (c jobClient) QuerySpans(ctx context.Context, jobID uuid.UUID, options ...T } } - page = spansMessage.GetNextPage() - if page == nil || (opts.Limit > 0 && remaining == 0) { + cursor = spansPage.NextCursor + if cursor == nil || (opts.Limit > 0 && remaining == 0) { break } } } } +func (c jobClient) QuerySpansPage(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) (*SpanPage, error) { + opts := &job.TelemetryQueryOptions{} + for _, option := range options { + option.ApplyTelemetryQueryOption(opts) + } + return c.querySpansPage(ctx, jobID, opts) +} + +func (c jobClient) querySpansPage(ctx context.Context, jobID uuid.UUID, opts *job.TelemetryQueryOptions) (*SpanPage, error) { + spansMessage, err := c.telemetryService.QueryJobSpans(ctx, jobID, paginationFromOptions(opts.Limit, opts.Cursor), sortDirectionToProto(opts.SortDirection)) + if err != nil { + return nil, err + } + + spans := make([]*Span, 0, len(spansMessage.GetResourceSpans())) + for _, resourceSpans := range spansMessage.GetResourceSpans() { + span, err := protoToSpan(resourceSpans) + if err != nil { + return nil, err + } + spans = append(spans, span) + } + + return &SpanPage{ + Spans: spans, + NextCursor: cursorFromPagination(spansMessage.GetNextPage()), + }, nil +} + func validateJob(jobName string, clusterSlug string, maxRetries int64, tasks ...Task) (*workflowsv1.SubmitJobRequest, error) { if len(tasks) == 0 { return nil, errors.New("no tasks to submit") @@ -563,36 +722,20 @@ func protoToProgressIndicator(p *workflowsv1.Progress) *ProgressIndicator { } } -func mergeTelemetryQueryOptions(options []TelemetryQueryOption) TelemetryQueryOption { - var opts TelemetryQueryOption - for _, option := range options { - if option.Limit > 0 { - opts.Limit = option.Limit - } - if option.SortDirection != 0 { - opts.SortDirection = option.SortDirection - } - } - return opts -} - func jobQuerySortDirectionToProto(direction job.SortDirection) tileboxv1.SortDirection { - switch direction { - case job.Ascending: - return tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING - case job.Descending: - return tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING - default: + protoDirection := sortDirectionToProto(direction) + if protoDirection == nil { return tileboxv1.SortDirection_SORT_DIRECTION_UNSPECIFIED } + return *protoDirection } -func (direction SortDirection) protoSortDirection() *tileboxv1.SortDirection { +func sortDirectionToProto(direction job.SortDirection) *tileboxv1.SortDirection { var protoDirection tileboxv1.SortDirection switch direction { - case Ascending: + case job.Ascending: protoDirection = tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING - case Descending: + case job.Descending: protoDirection = tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING default: return nil @@ -600,17 +743,35 @@ func (direction SortDirection) protoSortDirection() *tileboxv1.SortDirection { return &protoDirection } -func telemetryQueryPage(page *tileboxv1.Pagination, remaining int64) *tileboxv1.Pagination { - if remaining <= 0 { - return page +func paginationFromOptions(limit int64, cursor *job.Cursor) *tileboxv1.Pagination { + if limit <= 0 && cursor == nil { + return nil + } + + var startingAfter *tileboxv1.ID + if cursor != nil { + startingAfter = tileboxv1.NewUUID(cursor.StartingAfter()) + } + + if limit <= 0 { + return tileboxv1.Pagination_builder{ + StartingAfter: startingAfter, + }.Build() } return tileboxv1.Pagination_builder{ - Limit: &remaining, - StartingAfter: page.GetStartingAfter(), + Limit: &limit, + StartingAfter: startingAfter, }.Build() } +func cursorFromPagination(page *tileboxv1.Pagination) *job.Cursor { + if page == nil || page.GetStartingAfter() == nil { + return nil + } + return job.NewCursor(page.GetStartingAfter().AsUUID()) +} + func protoToLogRecord(resourceLogs *logsv1.ResourceLogs) (*LogRecord, error) { if len(resourceLogs.GetScopeLogs()) == 0 || len(resourceLogs.GetScopeLogs()[0].GetLogRecords()) == 0 { return nil, errors.New("resource logs does not contain a log record") diff --git a/workflows/v1/jobs_test.go b/workflows/v1/jobs_test.go index a679494..2ad64bd 100644 --- a/workflows/v1/jobs_test.go +++ b/workflows/v1/jobs_test.go @@ -2,6 +2,7 @@ package workflows import ( "context" + "encoding/json" "testing" "time" @@ -79,6 +80,33 @@ func (m *mockTelemetryService) QueryJobSpans(_ context.Context, jobID uuid.UUID, return m.spanPages[pageIndex], nil } +func TestTaskStateString(t *testing.T) { + tests := []struct { + name string + state TaskState + want string + }{ + {name: "queued", state: TaskQueued, want: "queued"}, + {name: "running", state: TaskRunning, want: "running"}, + {name: "computed", state: TaskComputed, want: "computed"}, + {name: "failed", state: TaskFailed, want: "failed"}, + {name: "skipped", state: TaskSkipped, want: "skipped"}, + {name: "failed optional", state: TaskFailedOptional, want: "failed_optional"}, + {name: "unspecified", state: 0, want: "unspecified"}, + {name: "unknown", state: 99, want: "unspecified"}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, test.state.String()) + + data, err := json.Marshal(test.state) + require.NoError(t, err) + assert.JSONEq(t, `"`+test.want+`"`, string(data)) + }) + } +} + func TestJobService_Submit(t *testing.T) { type args struct { jobName string @@ -201,6 +229,105 @@ func Test_jobClient_Query_WithSortDirection(t *testing.T) { assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING, service.querySortReqs[0]) } +func Test_jobClient_Query_WithoutTemporalExtent(t *testing.T) { + service := &mockJobService{} + client := jobClient{service: service} + + jobs, err := Collect(client.Query(context.Background())) + require.NoError(t, err) + assert.Empty(t, jobs) + + require.Len(t, service.queryFilters, 1) + assert.Nil(t, service.queryFilters[0].GetTimeInterval()) + assert.Nil(t, service.queryFilters[0].GetIdInterval()) +} + +func Test_jobClient_Query_WithTaskStates(t *testing.T) { + service := &mockJobService{} + client := jobClient{service: service} + timeInterval := query.NewTimeInterval( + time.Date(2024, time.February, 1, 0, 0, 0, 0, time.UTC), + time.Date(2025, time.February, 1, 0, 0, 0, 0, time.UTC), + ) + + jobs, err := Collect(client.Query(context.Background(), job.WithTemporalExtent(timeInterval), job.WithTaskStates(job.TaskRunning, job.TaskFailed))) + require.NoError(t, err) + assert.Empty(t, jobs) + + require.Len(t, service.queryFilters, 1) + assert.Equal(t, []workflowsv1.TaskState{workflowsv1.TaskState_TASK_STATE_RUNNING, workflowsv1.TaskState_TASK_STATE_FAILED}, service.queryFilters[0].GetTaskStates()) +} + +func Test_jobClient_Query_WithLimit(t *testing.T) { + nextPage := tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b"))}.Build() + service := &mockJobService{ + queryPages: []*workflowsv1.QueryJobsResponse{ + workflowsv1.QueryJobsResponse_builder{ + Jobs: []*workflowsv1.Job{ + workflowsv1.Job_builder{Name: "first", Id: tileboxv1.NewUUID(uuid.New())}.Build(), + }, + NextPage: nextPage, + }.Build(), + workflowsv1.QueryJobsResponse_builder{ + Jobs: []*workflowsv1.Job{ + workflowsv1.Job_builder{Name: "second", Id: tileboxv1.NewUUID(uuid.New())}.Build(), + workflowsv1.Job_builder{Name: "third", Id: tileboxv1.NewUUID(uuid.New())}.Build(), + }, + }.Build(), + }, + } + client := jobClient{service: service} + timeInterval := query.NewTimeInterval( + time.Date(2024, time.February, 1, 0, 0, 0, 0, time.UTC), + time.Date(2025, time.February, 1, 0, 0, 0, 0, time.UTC), + ) + + jobs, err := Collect(client.Query(context.Background(), job.WithTemporalExtent(timeInterval), job.WithLimit(2))) + require.NoError(t, err) + require.Len(t, jobs, 2) + + require.Len(t, service.queryPageReqs, 2) + assert.Equal(t, int64(2), service.queryPageReqs[0].GetLimit()) + assert.Nil(t, service.queryPageReqs[0].GetStartingAfter()) + assert.Equal(t, int64(1), service.queryPageReqs[1].GetLimit()) + assert.Equal(t, nextPage.GetStartingAfter(), service.queryPageReqs[1].GetStartingAfter()) + assert.Equal(t, "first", jobs[0].Name) + assert.Equal(t, "second", jobs[1].Name) +} + +func Test_jobClient_QueryPage(t *testing.T) { + cursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b") + nextCursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75c") + service := &mockJobService{ + queryPages: []*workflowsv1.QueryJobsResponse{ + workflowsv1.QueryJobsResponse_builder{ + Jobs: []*workflowsv1.Job{ + workflowsv1.Job_builder{Name: "first", Id: tileboxv1.NewUUID(uuid.New())}.Build(), + }, + NextPage: tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(nextCursorID)}.Build(), + }.Build(), + }, + } + client := jobClient{service: service} + timeInterval := query.NewTimeInterval( + time.Date(2024, time.February, 1, 0, 0, 0, 0, time.UTC), + time.Date(2025, time.February, 1, 0, 0, 0, 0, time.UTC), + ) + + page, err := client.QueryPage(context.Background(), job.WithTemporalExtent(timeInterval), job.WithCursor(job.NewCursor(cursorID)), job.WithLimit(10), job.WithSortDirection(job.Descending)) + require.NoError(t, err) + require.Len(t, page.Jobs, 1) + require.NotNil(t, page.NextCursor) + + assert.Equal(t, "first", page.Jobs[0].Name) + assert.Equal(t, nextCursorID, page.NextCursor.StartingAfter()) + require.Len(t, service.queryPageReqs, 1) + assert.Equal(t, int64(10), service.queryPageReqs[0].GetLimit()) + assert.Equal(t, cursorID, service.queryPageReqs[0].GetStartingAfter().AsUUID()) + require.Len(t, service.querySortReqs, 1) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING, service.querySortReqs[0]) +} + func Test_jobClient_QueryLogs(t *testing.T) { jobID := uuid.MustParse("01994da3-779b-7d01-a570-2a04d0ac163b") nextPage := tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b"))}.Build() @@ -250,7 +377,7 @@ func Test_jobClient_QueryLogs_WithOptions(t *testing.T) { } client := jobClient{telemetryService: service} - logs, err := Collect(client.QueryLogs(context.Background(), jobID, WithLimit(2), WithSortDirection(Descending))) + logs, err := Collect(client.QueryLogs(context.Background(), jobID, job.WithLimit(2), job.WithSortDirection(job.Descending))) require.NoError(t, err) require.Len(t, logs, 2) @@ -266,6 +393,34 @@ func Test_jobClient_QueryLogs_WithOptions(t *testing.T) { assert.Equal(t, "second log", logs[1].Body) } +func Test_jobClient_QueryLogsPage(t *testing.T) { + jobID := uuid.MustParse("01994da3-779b-7d01-a570-2a04d0ac163b") + cursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b") + nextCursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75c") + service := &mockTelemetryService{ + logPages: []*workflowsv1.PaginatedLogsData{ + workflowsv1.PaginatedLogsData_builder{ + ResourceLogs: []*logsv1.ResourceLogs{testResourceLogs("first log", 10)}, + NextPage: tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(nextCursorID)}.Build(), + }.Build(), + }, + } + client := jobClient{telemetryService: service} + + page, err := client.QueryLogsPage(context.Background(), jobID, job.WithCursor(job.NewCursor(cursorID)), job.WithLimit(10), job.WithSortDirection(job.Descending)) + require.NoError(t, err) + require.Len(t, page.Logs, 1) + require.NotNil(t, page.NextCursor) + + assert.Equal(t, "first log", page.Logs[0].Body) + assert.Equal(t, nextCursorID, page.NextCursor.StartingAfter()) + require.Len(t, service.logPageReqs, 1) + assert.Equal(t, int64(10), service.logPageReqs[0].GetLimit()) + assert.Equal(t, cursorID, service.logPageReqs[0].GetStartingAfter().AsUUID()) + require.Len(t, service.logSortReqs, 1) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_DESCENDING, *service.logSortReqs[0]) +} + func Test_jobClient_QuerySpans(t *testing.T) { jobID := uuid.MustParse("01994da3-779b-7d01-a570-2a04d0ac163b") service := &mockTelemetryService{ @@ -310,7 +465,7 @@ func Test_jobClient_QuerySpans_WithOptions(t *testing.T) { } client := jobClient{telemetryService: service} - spans, err := Collect(client.QuerySpans(context.Background(), jobID, WithLimit(1), WithSortDirection(Ascending))) + spans, err := Collect(client.QuerySpans(context.Background(), jobID, job.WithLimit(1), job.WithSortDirection(job.Ascending))) require.NoError(t, err) require.Len(t, spans, 1) @@ -320,6 +475,34 @@ func Test_jobClient_QuerySpans_WithOptions(t *testing.T) { assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING, *service.spanSortReqs[0]) } +func Test_jobClient_QuerySpansPage(t *testing.T) { + jobID := uuid.MustParse("01994da3-779b-7d01-a570-2a04d0ac163b") + cursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75b") + nextCursorID := uuid.MustParse("01994da4-255e-740d-9df7-b8c1aa41c75c") + service := &mockTelemetryService{ + spanPages: []*workflowsv1.PaginatedSpansData{ + workflowsv1.PaginatedSpansData_builder{ + ResourceSpans: []*tracev1.ResourceSpans{testResourceSpans()}, + NextPage: tileboxv1.Pagination_builder{StartingAfter: tileboxv1.NewUUID(nextCursorID)}.Build(), + }.Build(), + }, + } + client := jobClient{telemetryService: service} + + page, err := client.QuerySpansPage(context.Background(), jobID, job.WithCursor(job.NewCursor(cursorID)), job.WithLimit(10), job.WithSortDirection(job.Ascending)) + require.NoError(t, err) + require.Len(t, page.Spans, 1) + require.NotNil(t, page.NextCursor) + + assert.Equal(t, "task/run", page.Spans[0].Name) + assert.Equal(t, nextCursorID, page.NextCursor.StartingAfter()) + require.Len(t, service.spanPageReqs, 1) + assert.Equal(t, int64(10), service.spanPageReqs[0].GetLimit()) + assert.Equal(t, cursorID, service.spanPageReqs[0].GetStartingAfter().AsUUID()) + require.Len(t, service.spanSortReqs, 1) + assert.Equal(t, tileboxv1.SortDirection_SORT_DIRECTION_ASCENDING, *service.spanSortReqs[0]) +} + func testResourceLogs(body string, severity logsv1.SeverityNumber) *logsv1.ResourceLogs { return &logsv1.ResourceLogs{ Resource: &resourcev1.Resource{Attributes: []*commonv1.KeyValue{stringKeyValue("runner", "test-runner")}}, From 2fc0576ed806884fcc266ec4fef8f95099cf3f93 Mon Sep 17 00:00:00 2001 From: Lukas Bindreiter Date: Wed, 20 May 2026 16:01:21 +0200 Subject: [PATCH 4/4] Fix linter issues --- workflows/v1/job/options.go | 14 ++- workflows/v1/job/options_test.go | 2 +- workflows/v1/jobs.go | 142 +++++++++++++++---------------- 3 files changed, 77 insertions(+), 81 deletions(-) diff --git a/workflows/v1/job/options.go b/workflows/v1/job/options.go index 8490733..74cc8fb 100644 --- a/workflows/v1/job/options.go +++ b/workflows/v1/job/options.go @@ -167,20 +167,16 @@ type TelemetryQueryOption interface { ApplyTelemetryQueryOption(*TelemetryQueryOptions) } -type SortDirectionOption interface { +type sharedQueryOption interface { QueryOption TelemetryQueryOption } -type LimitOption interface { - QueryOption - TelemetryQueryOption -} +type SortDirectionOption = sharedQueryOption -type CursorOption interface { - QueryOption - TelemetryQueryOption -} +type LimitOption = sharedQueryOption + +type CursorOption = sharedQueryOption // Cursor identifies where to continue a paginated query. // diff --git a/workflows/v1/job/options_test.go b/workflows/v1/job/options_test.go index f7676cc..1d7c0f9 100644 --- a/workflows/v1/job/options_test.go +++ b/workflows/v1/job/options_test.go @@ -260,6 +260,6 @@ func TestCursor(t *testing.T) { assert.Equal(t, id.String(), cursor.String()) parsed, err := ParseCursor(cursor.String()) - assert.NoError(t, err) + require.NoError(t, err) assert.Equal(t, id, parsed.StartingAfter()) } diff --git a/workflows/v1/jobs.go b/workflows/v1/jobs.go index c5efd65..64768ef 100644 --- a/workflows/v1/jobs.go +++ b/workflows/v1/jobs.go @@ -412,56 +412,6 @@ func (c jobClient) QueryPage(ctx context.Context, options ...job.QueryOption) (* return c.queryPage(ctx, opts) } -func (c jobClient) queryPage(ctx context.Context, opts *job.QueryOptions) (*JobPage, error) { - sortDirection := jobQuerySortDirectionToProto(opts.SortDirection) - - var timeInterval *tileboxv1.TimeInterval - var idInterval *tileboxv1.IDInterval - if opts.TemporalExtent != nil { - timeInterval = opts.TemporalExtent.ToProtoTimeInterval() - idInterval = opts.TemporalExtent.ToProtoIDInterval() - - if timeInterval == nil && idInterval == nil { - return nil, errors.New("invalid temporal extent") - } - } - - automationIDs := lo.Map(opts.AutomationIDs, func(id uuid.UUID, _ int) *tileboxv1.ID { - return tileboxv1.NewUUID(id) - }) - - states := lo.Map(opts.States, func(state job.State, _ int) workflowsv1.JobState { - return workflowsv1.JobState(state) - }) - taskStates := lo.Map(opts.TaskStates, func(state job.TaskState, _ int) workflowsv1.TaskState { - return workflowsv1.TaskState(state) - }) - - filters := workflowsv1.QueryFilters_builder{ - TimeInterval: timeInterval, - IdInterval: idInterval, - AutomationIds: automationIDs, - States: states, - TaskStates: taskStates, - Name: opts.Name, - }.Build() - - jobsMessage, err := c.service.QueryJobs(ctx, filters, paginationFromOptions(opts.Limit, opts.Cursor), sortDirection) - if err != nil { - return nil, err - } - - jobs := make([]*Job, 0, len(jobsMessage.GetJobs())) - for _, jobMessage := range jobsMessage.GetJobs() { - jobs = append(jobs, protoToJob(jobMessage)) - } - - return &JobPage{ - Jobs: jobs, - NextCursor: cursorFromPagination(jobsMessage.GetNextPage()), - }, nil -} - func (c jobClient) QueryLogs(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*LogRecord, error] { opts := &job.TelemetryQueryOptions{} for _, option := range options { @@ -517,27 +467,6 @@ func (c jobClient) QueryLogsPage(ctx context.Context, jobID uuid.UUID, options . return c.queryLogsPage(ctx, jobID, opts) } -func (c jobClient) queryLogsPage(ctx context.Context, jobID uuid.UUID, opts *job.TelemetryQueryOptions) (*LogPage, error) { - logsMessage, err := c.telemetryService.QueryJobLogs(ctx, jobID, paginationFromOptions(opts.Limit, opts.Cursor), sortDirectionToProto(opts.SortDirection)) - if err != nil { - return nil, err - } - - logs := make([]*LogRecord, 0, len(logsMessage.GetResourceLogs())) - for _, resourceLogs := range logsMessage.GetResourceLogs() { - logRecord, err := protoToLogRecord(resourceLogs) - if err != nil { - return nil, err - } - logs = append(logs, logRecord) - } - - return &LogPage{ - Logs: logs, - NextCursor: cursorFromPagination(logsMessage.GetNextPage()), - }, nil -} - func (c jobClient) QuerySpans(ctx context.Context, jobID uuid.UUID, options ...job.TelemetryQueryOption) iter.Seq2[*Span, error] { opts := &job.TelemetryQueryOptions{} for _, option := range options { @@ -614,6 +543,77 @@ func (c jobClient) querySpansPage(ctx context.Context, jobID uuid.UUID, opts *jo }, nil } +func (c jobClient) queryPage(ctx context.Context, opts *job.QueryOptions) (*JobPage, error) { + sortDirection := jobQuerySortDirectionToProto(opts.SortDirection) + + var timeInterval *tileboxv1.TimeInterval + var idInterval *tileboxv1.IDInterval + if opts.TemporalExtent != nil { + timeInterval = opts.TemporalExtent.ToProtoTimeInterval() + idInterval = opts.TemporalExtent.ToProtoIDInterval() + + if timeInterval == nil && idInterval == nil { + return nil, errors.New("invalid temporal extent") + } + } + + automationIDs := lo.Map(opts.AutomationIDs, func(id uuid.UUID, _ int) *tileboxv1.ID { + return tileboxv1.NewUUID(id) + }) + + states := lo.Map(opts.States, func(state job.State, _ int) workflowsv1.JobState { + return workflowsv1.JobState(state) + }) + taskStates := lo.Map(opts.TaskStates, func(state job.TaskState, _ int) workflowsv1.TaskState { + return workflowsv1.TaskState(state) + }) + + filters := workflowsv1.QueryFilters_builder{ + TimeInterval: timeInterval, + IdInterval: idInterval, + AutomationIds: automationIDs, + States: states, + TaskStates: taskStates, + Name: opts.Name, + }.Build() + + jobsMessage, err := c.service.QueryJobs(ctx, filters, paginationFromOptions(opts.Limit, opts.Cursor), sortDirection) + if err != nil { + return nil, err + } + + jobs := make([]*Job, 0, len(jobsMessage.GetJobs())) + for _, jobMessage := range jobsMessage.GetJobs() { + jobs = append(jobs, protoToJob(jobMessage)) + } + + return &JobPage{ + Jobs: jobs, + NextCursor: cursorFromPagination(jobsMessage.GetNextPage()), + }, nil +} + +func (c jobClient) queryLogsPage(ctx context.Context, jobID uuid.UUID, opts *job.TelemetryQueryOptions) (*LogPage, error) { + logsMessage, err := c.telemetryService.QueryJobLogs(ctx, jobID, paginationFromOptions(opts.Limit, opts.Cursor), sortDirectionToProto(opts.SortDirection)) + if err != nil { + return nil, err + } + + logs := make([]*LogRecord, 0, len(logsMessage.GetResourceLogs())) + for _, resourceLogs := range logsMessage.GetResourceLogs() { + logRecord, err := protoToLogRecord(resourceLogs) + if err != nil { + return nil, err + } + logs = append(logs, logRecord) + } + + return &LogPage{ + Logs: logs, + NextCursor: cursorFromPagination(logsMessage.GetNextPage()), + }, nil +} + func validateJob(jobName string, clusterSlug string, maxRetries int64, tasks ...Task) (*workflowsv1.SubmitJobRequest, error) { if len(tasks) == 0 { return nil, errors.New("no tasks to submit")