From 4010630fe8e97c7237dea876c7e3009983c8818c Mon Sep 17 00:00:00 2001 From: Albert Wu Date: Mon, 6 Jul 2026 17:04:38 -0700 Subject: [PATCH] feat(gateway): expose List API from request summaries --- api/submitqueue/gateway/proto/gateway.proto | 60 +++ api/submitqueue/gateway/protopb/gateway.pb.go | 463 ++++++++++++++++-- .../gateway/protopb/gateway.pb.yarpc.go | 149 ++++-- .../gateway/protopb/gateway_grpc.pb.go | 40 ++ service/submitqueue/gateway/server/main.go | 11 +- submitqueue/entity/request_log.go | 27 + submitqueue/gateway/controller/BUILD.bazel | 2 + submitqueue/gateway/controller/cancel.go | 25 +- submitqueue/gateway/controller/cancel_test.go | 58 ++- submitqueue/gateway/controller/land.go | 13 +- submitqueue/gateway/controller/land_test.go | 9 + submitqueue/gateway/controller/list.go | 261 ++++++++++ submitqueue/gateway/controller/list_test.go | 286 +++++++++++ .../gateway/controller/log/BUILD.bazel | 1 + submitqueue/gateway/controller/log/log.go | 7 +- .../gateway/controller/log/log_test.go | 7 + 16 files changed, 1300 insertions(+), 119 deletions(-) create mode 100644 submitqueue/gateway/controller/list.go create mode 100644 submitqueue/gateway/controller/list_test.go diff --git a/api/submitqueue/gateway/proto/gateway.proto b/api/submitqueue/gateway/proto/gateway.proto index 244f7b43..65b879e9 100644 --- a/api/submitqueue/gateway/proto/gateway.proto +++ b/api/submitqueue/gateway/proto/gateway.proto @@ -100,6 +100,63 @@ message StatusResponse { map metadata = 3; } +// ListSort defines supported deterministic orderings for List. +enum ListSort { + // Server default: ADMITTED_ASC. + SORT_DEFAULT = 0; + // FIFO/admission order. + ADMITTED_ASC = 1; + // Newest admissions first. + ADMITTED_DESC = 2; +} + +// ListRequest defines a request to list queue requests admitted during a time window. +message ListRequest { + // Name of the queue whose requests should be returned. + string queue = 1; + // Inclusive lower bound for admission-time filtering, in Unix epoch milliseconds. + int64 start_time_ms = 2; + // Exclusive upper bound for admission-time filtering, in Unix epoch milliseconds. + int64 end_time_ms = 3; + // Optional current status filters. Values use the same customer-facing strings returned by Status. + repeated string statuses = 4; + // Maximum number of requests to return. Zero means the server default. + int32 page_size = 5; + // Opaque token returned by a previous List response. + string page_token = 6; + // Sort order for deterministic pagination. Unspecified defaults to FIFO/admission order. + ListSort sort = 7; +} + +// RequestSummary is the current gateway-owned view of one request for queue-listing UX. +message RequestSummary { + // Globally unique request ID, as returned by Land. + string sqid = 1; + // Queue the request belongs to. + string queue = 2; + // Change URIs submitted with the request. + repeated string change_uris = 3; + // Current customer-friendly status of the request. + string status = 4; + // Last error message associated with the current status. Empty string if there is no error. + string last_error = 5; + // Free-form key-value metadata associated with the current status. + map metadata = 6; + // Time the request entered SubmitQueue, in Unix epoch milliseconds. + int64 started_at_ms = 7; + // Time the visible summary state last changed, in Unix epoch milliseconds. + int64 updated_at_ms = 8; + // Time the request completed, in Unix epoch milliseconds. Zero if the request is not terminal. + int64 completed_at_ms = 9; +} + +// ListResponse defines a page of queue request summaries. +message ListResponse { + repeated RequestSummary requests = 1; + // Opaque token for the next page. Empty when no further page exists. + string next_page_token = 2; +} + // *************** // Error messages, returned as `google.rpc.Status` messages. // *************** @@ -155,4 +212,7 @@ service SubmitQueueGateway { // Status returns the current status of a previously submitted request, identified by its sqid. // The status is eventually consistent with the request store and reconciled from the append-only request log. rpc Status(StatusRequest) returns (StatusResponse) {} + + // List returns request summaries for one queue admitted during a time window. + rpc List(ListRequest) returns (ListResponse) {} } diff --git a/api/submitqueue/gateway/protopb/gateway.pb.go b/api/submitqueue/gateway/protopb/gateway.pb.go index 20c020f3..25ec1778 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.go @@ -38,6 +38,59 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) +// ListSort defines supported deterministic orderings for List. +type ListSort int32 + +const ( + // Server default: ADMITTED_ASC. + ListSort_SORT_DEFAULT ListSort = 0 + // FIFO/admission order. + ListSort_ADMITTED_ASC ListSort = 1 + // Newest admissions first. + ListSort_ADMITTED_DESC ListSort = 2 +) + +// Enum value maps for ListSort. +var ( + ListSort_name = map[int32]string{ + 0: "SORT_DEFAULT", + 1: "ADMITTED_ASC", + 2: "ADMITTED_DESC", + } + ListSort_value = map[string]int32{ + "SORT_DEFAULT": 0, + "ADMITTED_ASC": 1, + "ADMITTED_DESC": 2, + } +) + +func (x ListSort) Enum() *ListSort { + p := new(ListSort) + *p = x + return p +} + +func (x ListSort) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (ListSort) Descriptor() protoreflect.EnumDescriptor { + return file_gateway_proto_enumTypes[0].Descriptor() +} + +func (ListSort) Type() protoreflect.EnumType { + return &file_gateway_proto_enumTypes[0] +} + +func (x ListSort) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use ListSort.Descriptor instead. +func (ListSort) EnumDescriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{0} +} + // PingRequest is the request for the Ping method type PingRequest struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -479,6 +532,278 @@ func (x *StatusResponse) GetMetadata() map[string]string { return nil } +// ListRequest defines a request to list queue requests admitted during a time window. +type ListRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Name of the queue whose requests should be returned. + Queue string `protobuf:"bytes,1,opt,name=queue,proto3" json:"queue,omitempty"` + // Inclusive lower bound for admission-time filtering, in Unix epoch milliseconds. + StartTimeMs int64 `protobuf:"varint,2,opt,name=start_time_ms,json=startTimeMs,proto3" json:"start_time_ms,omitempty"` + // Exclusive upper bound for admission-time filtering, in Unix epoch milliseconds. + EndTimeMs int64 `protobuf:"varint,3,opt,name=end_time_ms,json=endTimeMs,proto3" json:"end_time_ms,omitempty"` + // Optional current status filters. Values use the same customer-facing strings returned by Status. + Statuses []string `protobuf:"bytes,4,rep,name=statuses,proto3" json:"statuses,omitempty"` + // Maximum number of requests to return. Zero means the server default. + PageSize int32 `protobuf:"varint,5,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Opaque token returned by a previous List response. + PageToken string `protobuf:"bytes,6,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` + // Sort order for deterministic pagination. Unspecified defaults to FIFO/admission order. + Sort ListSort `protobuf:"varint,7,opt,name=sort,proto3,enum=uber.submitqueue.gateway.ListSort" json:"sort,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListRequest) Reset() { + *x = ListRequest{} + mi := &file_gateway_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListRequest) ProtoMessage() {} + +func (x *ListRequest) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ListRequest.ProtoReflect.Descriptor instead. +func (*ListRequest) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{8} +} + +func (x *ListRequest) GetQueue() string { + if x != nil { + return x.Queue + } + return "" +} + +func (x *ListRequest) GetStartTimeMs() int64 { + if x != nil { + return x.StartTimeMs + } + return 0 +} + +func (x *ListRequest) GetEndTimeMs() int64 { + if x != nil { + return x.EndTimeMs + } + return 0 +} + +func (x *ListRequest) GetStatuses() []string { + if x != nil { + return x.Statuses + } + return nil +} + +func (x *ListRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *ListRequest) GetPageToken() string { + if x != nil { + return x.PageToken + } + return "" +} + +func (x *ListRequest) GetSort() ListSort { + if x != nil { + return x.Sort + } + return ListSort_SORT_DEFAULT +} + +// RequestSummary is the current gateway-owned view of one request for queue-listing UX. +type RequestSummary struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Globally unique request ID, as returned by Land. + Sqid string `protobuf:"bytes,1,opt,name=sqid,proto3" json:"sqid,omitempty"` + // Queue the request belongs to. + Queue string `protobuf:"bytes,2,opt,name=queue,proto3" json:"queue,omitempty"` + // Change URIs submitted with the request. + ChangeUris []string `protobuf:"bytes,3,rep,name=change_uris,json=changeUris,proto3" json:"change_uris,omitempty"` + // Current customer-friendly status of the request. + Status string `protobuf:"bytes,4,opt,name=status,proto3" json:"status,omitempty"` + // Last error message associated with the current status. Empty string if there is no error. + LastError string `protobuf:"bytes,5,opt,name=last_error,json=lastError,proto3" json:"last_error,omitempty"` + // Free-form key-value metadata associated with the current status. + Metadata map[string]string `protobuf:"bytes,6,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // Time the request entered SubmitQueue, in Unix epoch milliseconds. + StartedAtMs int64 `protobuf:"varint,7,opt,name=started_at_ms,json=startedAtMs,proto3" json:"started_at_ms,omitempty"` + // Time the visible summary state last changed, in Unix epoch milliseconds. + UpdatedAtMs int64 `protobuf:"varint,8,opt,name=updated_at_ms,json=updatedAtMs,proto3" json:"updated_at_ms,omitempty"` + // Time the request completed, in Unix epoch milliseconds. Zero if the request is not terminal. + CompletedAtMs int64 `protobuf:"varint,9,opt,name=completed_at_ms,json=completedAtMs,proto3" json:"completed_at_ms,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *RequestSummary) Reset() { + *x = RequestSummary{} + mi := &file_gateway_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *RequestSummary) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*RequestSummary) ProtoMessage() {} + +func (x *RequestSummary) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use RequestSummary.ProtoReflect.Descriptor instead. +func (*RequestSummary) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{9} +} + +func (x *RequestSummary) GetSqid() string { + if x != nil { + return x.Sqid + } + return "" +} + +func (x *RequestSummary) GetQueue() string { + if x != nil { + return x.Queue + } + return "" +} + +func (x *RequestSummary) GetChangeUris() []string { + if x != nil { + return x.ChangeUris + } + return nil +} + +func (x *RequestSummary) GetStatus() string { + if x != nil { + return x.Status + } + return "" +} + +func (x *RequestSummary) GetLastError() string { + if x != nil { + return x.LastError + } + return "" +} + +func (x *RequestSummary) GetMetadata() map[string]string { + if x != nil { + return x.Metadata + } + return nil +} + +func (x *RequestSummary) GetStartedAtMs() int64 { + if x != nil { + return x.StartedAtMs + } + return 0 +} + +func (x *RequestSummary) GetUpdatedAtMs() int64 { + if x != nil { + return x.UpdatedAtMs + } + return 0 +} + +func (x *RequestSummary) GetCompletedAtMs() int64 { + if x != nil { + return x.CompletedAtMs + } + return 0 +} + +// ListResponse defines a page of queue request summaries. +type ListResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + Requests []*RequestSummary `protobuf:"bytes,1,rep,name=requests,proto3" json:"requests,omitempty"` + // Opaque token for the next page. Empty when no further page exists. + NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ListResponse) Reset() { + *x = ListResponse{} + mi := &file_gateway_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ListResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ListResponse) ProtoMessage() {} + +func (x *ListResponse) ProtoReflect() protoreflect.Message { + mi := &file_gateway_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ListResponse.ProtoReflect.Descriptor instead. +func (*ListResponse) Descriptor() ([]byte, []int) { + return file_gateway_proto_rawDescGZIP(), []int{10} +} + +func (x *ListResponse) GetRequests() []*RequestSummary { + if x != nil { + return x.Requests + } + return nil +} + +func (x *ListResponse) GetNextPageToken() string { + if x != nil { + return x.NextPageToken + } + return "" +} + // Generic error with metadata. Each custom error type should extend this message. type Error struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -490,7 +815,7 @@ type Error struct { func (x *Error) Reset() { *x = Error{} - mi := &file_gateway_proto_msgTypes[8] + mi := &file_gateway_proto_msgTypes[11] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -502,7 +827,7 @@ func (x *Error) String() string { func (*Error) ProtoMessage() {} func (x *Error) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[8] + mi := &file_gateway_proto_msgTypes[11] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -515,7 +840,7 @@ func (x *Error) ProtoReflect() protoreflect.Message { // Deprecated: Use Error.ProtoReflect.Descriptor instead. func (*Error) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{8} + return file_gateway_proto_rawDescGZIP(), []int{11} } func (x *Error) GetMessage() string { @@ -538,7 +863,7 @@ type UnrecognizedQueueError struct { func (x *UnrecognizedQueueError) Reset() { *x = UnrecognizedQueueError{} - mi := &file_gateway_proto_msgTypes[9] + mi := &file_gateway_proto_msgTypes[12] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -550,7 +875,7 @@ func (x *UnrecognizedQueueError) String() string { func (*UnrecognizedQueueError) ProtoMessage() {} func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[9] + mi := &file_gateway_proto_msgTypes[12] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -563,7 +888,7 @@ func (x *UnrecognizedQueueError) ProtoReflect() protoreflect.Message { // Deprecated: Use UnrecognizedQueueError.ProtoReflect.Descriptor instead. func (*UnrecognizedQueueError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{9} + return file_gateway_proto_rawDescGZIP(), []int{12} } func (x *UnrecognizedQueueError) GetError() *Error { @@ -593,7 +918,7 @@ type RequestNotFoundError struct { func (x *RequestNotFoundError) Reset() { *x = RequestNotFoundError{} - mi := &file_gateway_proto_msgTypes[10] + mi := &file_gateway_proto_msgTypes[13] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -605,7 +930,7 @@ func (x *RequestNotFoundError) String() string { func (*RequestNotFoundError) ProtoMessage() {} func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { - mi := &file_gateway_proto_msgTypes[10] + mi := &file_gateway_proto_msgTypes[13] if x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -618,7 +943,7 @@ func (x *RequestNotFoundError) ProtoReflect() protoreflect.Message { // Deprecated: Use RequestNotFoundError.ProtoReflect.Descriptor instead. func (*RequestNotFoundError) Descriptor() ([]byte, []int) { - return file_gateway_proto_rawDescGZIP(), []int{10} + return file_gateway_proto_rawDescGZIP(), []int{13} } func (x *RequestNotFoundError) GetError() *Error { @@ -666,7 +991,34 @@ const file_gateway_proto_rawDesc = "" + "\bmetadata\x18\x03 \x03(\v26.uber.submitqueue.gateway.StatusResponse.MetadataEntryR\bmetadata\x1a;\n" + "\rMetadataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"!\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xf7\x01\n" + + "\vListRequest\x12\x14\n" + + "\x05queue\x18\x01 \x01(\tR\x05queue\x12\"\n" + + "\rstart_time_ms\x18\x02 \x01(\x03R\vstartTimeMs\x12\x1e\n" + + "\vend_time_ms\x18\x03 \x01(\x03R\tendTimeMs\x12\x1a\n" + + "\bstatuses\x18\x04 \x03(\tR\bstatuses\x12\x1b\n" + + "\tpage_size\x18\x05 \x01(\x05R\bpageSize\x12\x1d\n" + + "\n" + + "page_token\x18\x06 \x01(\tR\tpageToken\x126\n" + + "\x04sort\x18\a \x01(\x0e2\".uber.submitqueue.gateway.ListSortR\x04sort\"\x93\x03\n" + + "\x0eRequestSummary\x12\x12\n" + + "\x04sqid\x18\x01 \x01(\tR\x04sqid\x12\x14\n" + + "\x05queue\x18\x02 \x01(\tR\x05queue\x12\x1f\n" + + "\vchange_uris\x18\x03 \x03(\tR\n" + + "changeUris\x12\x16\n" + + "\x06status\x18\x04 \x01(\tR\x06status\x12\x1d\n" + + "\n" + + "last_error\x18\x05 \x01(\tR\tlastError\x12R\n" + + "\bmetadata\x18\x06 \x03(\v26.uber.submitqueue.gateway.RequestSummary.MetadataEntryR\bmetadata\x12\"\n" + + "\rstarted_at_ms\x18\a \x01(\x03R\vstartedAtMs\x12\"\n" + + "\rupdated_at_ms\x18\b \x01(\x03R\vupdatedAtMs\x12&\n" + + "\x0fcompleted_at_ms\x18\t \x01(\x03R\rcompletedAtMs\x1a;\n" + + "\rMetadataEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"|\n" + + "\fListResponse\x12D\n" + + "\brequests\x18\x01 \x03(\v2(.uber.submitqueue.gateway.RequestSummaryR\brequests\x12&\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"!\n" + "\x05Error\x12\x18\n" + "\amessage\x18\x01 \x01(\tR\amessage\"e\n" + "\x16UnrecognizedQueueError\x125\n" + @@ -674,12 +1026,17 @@ const file_gateway_proto_rawDesc = "" + "\x05queue\x18\x02 \x01(\tR\x05queue\"a\n" + "\x14RequestNotFoundError\x125\n" + "\x05error\x18\x01 \x01(\v2\x1f.uber.submitqueue.gateway.ErrorR\x05error\x12\x12\n" + - "\x04sqid\x18\x02 \x01(\tR\x04sqid2\x84\x03\n" + + "\x04sqid\x18\x02 \x01(\tR\x04sqid*A\n" + + "\bListSort\x12\x10\n" + + "\fSORT_DEFAULT\x10\x00\x12\x10\n" + + "\fADMITTED_ASC\x10\x01\x12\x11\n" + + "\rADMITTED_DESC\x10\x022\xdd\x03\n" + "\x12SubmitQueueGateway\x12W\n" + "\x04Ping\x12%.uber.submitqueue.gateway.PingRequest\x1a&.uber.submitqueue.gateway.PingResponse\"\x00\x12W\n" + "\x04Land\x12%.uber.submitqueue.gateway.LandRequest\x1a&.uber.submitqueue.gateway.LandResponse\"\x00\x12]\n" + "\x06Cancel\x12'.uber.submitqueue.gateway.CancelRequest\x1a(.uber.submitqueue.gateway.CancelResponse\"\x00\x12]\n" + - "\x06Status\x12'.uber.submitqueue.gateway.StatusRequest\x1a(.uber.submitqueue.gateway.StatusResponse\"\x00Bk\n" + + "\x06Status\x12'.uber.submitqueue.gateway.StatusRequest\x1a(.uber.submitqueue.gateway.StatusResponse\"\x00\x12W\n" + + "\x04List\x12%.uber.submitqueue.gateway.ListRequest\x1a&.uber.submitqueue.gateway.ListResponse\"\x00Bk\n" + "\x1ccom.uber.submitqueue.gatewayB\fGatewayProtoP\x01Z;github.com/uber/submitqueue/api/submitqueue/gateway/protopbb\x06proto3" var ( @@ -694,42 +1051,53 @@ func file_gateway_proto_rawDescGZIP() []byte { return file_gateway_proto_rawDescData } -var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 12) +var file_gateway_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_gateway_proto_msgTypes = make([]protoimpl.MessageInfo, 16) var file_gateway_proto_goTypes = []any{ - (*PingRequest)(nil), // 0: uber.submitqueue.gateway.PingRequest - (*PingResponse)(nil), // 1: uber.submitqueue.gateway.PingResponse - (*LandRequest)(nil), // 2: uber.submitqueue.gateway.LandRequest - (*LandResponse)(nil), // 3: uber.submitqueue.gateway.LandResponse - (*CancelRequest)(nil), // 4: uber.submitqueue.gateway.CancelRequest - (*CancelResponse)(nil), // 5: uber.submitqueue.gateway.CancelResponse - (*StatusRequest)(nil), // 6: uber.submitqueue.gateway.StatusRequest - (*StatusResponse)(nil), // 7: uber.submitqueue.gateway.StatusResponse - (*Error)(nil), // 8: uber.submitqueue.gateway.Error - (*UnrecognizedQueueError)(nil), // 9: uber.submitqueue.gateway.UnrecognizedQueueError - (*RequestNotFoundError)(nil), // 10: uber.submitqueue.gateway.RequestNotFoundError - nil, // 11: uber.submitqueue.gateway.StatusResponse.MetadataEntry - (*protopb.Change)(nil), // 12: uber.base.change.Change - (protopb1.Strategy)(0), // 13: uber.base.mergestrategy.Strategy + (ListSort)(0), // 0: uber.submitqueue.gateway.ListSort + (*PingRequest)(nil), // 1: uber.submitqueue.gateway.PingRequest + (*PingResponse)(nil), // 2: uber.submitqueue.gateway.PingResponse + (*LandRequest)(nil), // 3: uber.submitqueue.gateway.LandRequest + (*LandResponse)(nil), // 4: uber.submitqueue.gateway.LandResponse + (*CancelRequest)(nil), // 5: uber.submitqueue.gateway.CancelRequest + (*CancelResponse)(nil), // 6: uber.submitqueue.gateway.CancelResponse + (*StatusRequest)(nil), // 7: uber.submitqueue.gateway.StatusRequest + (*StatusResponse)(nil), // 8: uber.submitqueue.gateway.StatusResponse + (*ListRequest)(nil), // 9: uber.submitqueue.gateway.ListRequest + (*RequestSummary)(nil), // 10: uber.submitqueue.gateway.RequestSummary + (*ListResponse)(nil), // 11: uber.submitqueue.gateway.ListResponse + (*Error)(nil), // 12: uber.submitqueue.gateway.Error + (*UnrecognizedQueueError)(nil), // 13: uber.submitqueue.gateway.UnrecognizedQueueError + (*RequestNotFoundError)(nil), // 14: uber.submitqueue.gateway.RequestNotFoundError + nil, // 15: uber.submitqueue.gateway.StatusResponse.MetadataEntry + nil, // 16: uber.submitqueue.gateway.RequestSummary.MetadataEntry + (*protopb.Change)(nil), // 17: uber.base.change.Change + (protopb1.Strategy)(0), // 18: uber.base.mergestrategy.Strategy } var file_gateway_proto_depIdxs = []int32{ - 12, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change - 13, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy - 11, // 2: uber.submitqueue.gateway.StatusResponse.metadata:type_name -> uber.submitqueue.gateway.StatusResponse.MetadataEntry - 8, // 3: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error - 8, // 4: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error - 0, // 5: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest - 2, // 6: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest - 4, // 7: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest - 6, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Status:input_type -> uber.submitqueue.gateway.StatusRequest - 1, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse - 3, // 10: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse - 5, // 11: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse - 7, // 12: uber.submitqueue.gateway.SubmitQueueGateway.Status:output_type -> uber.submitqueue.gateway.StatusResponse - 9, // [9:13] is the sub-list for method output_type - 5, // [5:9] is the sub-list for method input_type - 5, // [5:5] is the sub-list for extension type_name - 5, // [5:5] is the sub-list for extension extendee - 0, // [0:5] is the sub-list for field type_name + 17, // 0: uber.submitqueue.gateway.LandRequest.change:type_name -> uber.base.change.Change + 18, // 1: uber.submitqueue.gateway.LandRequest.strategy:type_name -> uber.base.mergestrategy.Strategy + 15, // 2: uber.submitqueue.gateway.StatusResponse.metadata:type_name -> uber.submitqueue.gateway.StatusResponse.MetadataEntry + 0, // 3: uber.submitqueue.gateway.ListRequest.sort:type_name -> uber.submitqueue.gateway.ListSort + 16, // 4: uber.submitqueue.gateway.RequestSummary.metadata:type_name -> uber.submitqueue.gateway.RequestSummary.MetadataEntry + 10, // 5: uber.submitqueue.gateway.ListResponse.requests:type_name -> uber.submitqueue.gateway.RequestSummary + 12, // 6: uber.submitqueue.gateway.UnrecognizedQueueError.error:type_name -> uber.submitqueue.gateway.Error + 12, // 7: uber.submitqueue.gateway.RequestNotFoundError.error:type_name -> uber.submitqueue.gateway.Error + 1, // 8: uber.submitqueue.gateway.SubmitQueueGateway.Ping:input_type -> uber.submitqueue.gateway.PingRequest + 3, // 9: uber.submitqueue.gateway.SubmitQueueGateway.Land:input_type -> uber.submitqueue.gateway.LandRequest + 5, // 10: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:input_type -> uber.submitqueue.gateway.CancelRequest + 7, // 11: uber.submitqueue.gateway.SubmitQueueGateway.Status:input_type -> uber.submitqueue.gateway.StatusRequest + 9, // 12: uber.submitqueue.gateway.SubmitQueueGateway.List:input_type -> uber.submitqueue.gateway.ListRequest + 2, // 13: uber.submitqueue.gateway.SubmitQueueGateway.Ping:output_type -> uber.submitqueue.gateway.PingResponse + 4, // 14: uber.submitqueue.gateway.SubmitQueueGateway.Land:output_type -> uber.submitqueue.gateway.LandResponse + 6, // 15: uber.submitqueue.gateway.SubmitQueueGateway.Cancel:output_type -> uber.submitqueue.gateway.CancelResponse + 8, // 16: uber.submitqueue.gateway.SubmitQueueGateway.Status:output_type -> uber.submitqueue.gateway.StatusResponse + 11, // 17: uber.submitqueue.gateway.SubmitQueueGateway.List:output_type -> uber.submitqueue.gateway.ListResponse + 13, // [13:18] is the sub-list for method output_type + 8, // [8:13] is the sub-list for method input_type + 8, // [8:8] is the sub-list for extension type_name + 8, // [8:8] is the sub-list for extension extendee + 0, // [0:8] is the sub-list for field type_name } func init() { file_gateway_proto_init() } @@ -742,13 +1110,14 @@ func file_gateway_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_gateway_proto_rawDesc), len(file_gateway_proto_rawDesc)), - NumEnums: 0, - NumMessages: 12, + NumEnums: 1, + NumMessages: 16, NumExtensions: 0, NumServices: 1, }, GoTypes: file_gateway_proto_goTypes, DependencyIndexes: file_gateway_proto_depIdxs, + EnumInfos: file_gateway_proto_enumTypes, MessageInfos: file_gateway_proto_msgTypes, }.Build() File_gateway_proto = out.File diff --git a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go index a549dbd2..00f3ccee 100644 --- a/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go +++ b/api/submitqueue/gateway/protopb/gateway.pb.yarpc.go @@ -25,6 +25,7 @@ type SubmitQueueGatewayYARPCClient interface { Land(context.Context, *LandRequest, ...yarpc.CallOption) (*LandResponse, error) Cancel(context.Context, *CancelRequest, ...yarpc.CallOption) (*CancelResponse, error) Status(context.Context, *StatusRequest, ...yarpc.CallOption) (*StatusResponse, error) + List(context.Context, *ListRequest, ...yarpc.CallOption) (*ListResponse, error) } func newSubmitQueueGatewayYARPCClient(clientConfig transport.ClientConfig, anyResolver v2.AnyResolver, options ...v2.ClientOption) SubmitQueueGatewayYARPCClient { @@ -49,6 +50,7 @@ type SubmitQueueGatewayYARPCServer interface { Land(context.Context, *LandRequest) (*LandResponse, error) Cancel(context.Context, *CancelRequest) (*CancelResponse, error) Status(context.Context, *StatusRequest) (*StatusResponse, error) + List(context.Context, *ListRequest) (*ListResponse, error) } type buildSubmitQueueGatewayYARPCProceduresParams struct { @@ -102,6 +104,16 @@ func buildSubmitQueueGatewayYARPCProcedures(params buildSubmitQueueGatewayYARPCP }, ), }, + { + MethodName: "List", + Handler: v2.NewUnaryHandler( + v2.UnaryHandlerParams{ + Handle: handler.List, + NewRequest: newSubmitQueueGatewayServiceListYARPCRequest, + AnyResolver: params.AnyResolver, + }, + ), + }, }, OnewayHandlerParams: []v2.BuildProceduresOnewayHandlerParams{}, StreamHandlerParams: []v2.BuildProceduresStreamHandlerParams{}, @@ -262,6 +274,18 @@ func (c *_SubmitQueueGatewayYARPCCaller) Status(ctx context.Context, request *St return response, err } +func (c *_SubmitQueueGatewayYARPCCaller) List(ctx context.Context, request *ListRequest, options ...yarpc.CallOption) (*ListResponse, error) { + responseMessage, err := c.streamClient.Call(ctx, "List", request, newSubmitQueueGatewayServiceListYARPCResponse, options...) + if responseMessage == nil { + return nil, err + } + response, ok := responseMessage.(*ListResponse) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceListYARPCResponse, responseMessage) + } + return response, err +} + type _SubmitQueueGatewayYARPCHandler struct { server SubmitQueueGatewayYARPCServer } @@ -330,6 +354,22 @@ func (h *_SubmitQueueGatewayYARPCHandler) Status(ctx context.Context, requestMes return response, err } +func (h *_SubmitQueueGatewayYARPCHandler) List(ctx context.Context, requestMessage proto.Message) (proto.Message, error) { + var request *ListRequest + var ok bool + if requestMessage != nil { + request, ok = requestMessage.(*ListRequest) + if !ok { + return nil, v2.CastError(emptySubmitQueueGatewayServiceListYARPCRequest, requestMessage) + } + } + response, err := h.server.List(ctx, request) + if response == nil { + return nil, err + } + return response, err +} + func newSubmitQueueGatewayServicePingYARPCRequest() proto.Message { return &PingRequest{} } @@ -362,6 +402,14 @@ func newSubmitQueueGatewayServiceStatusYARPCResponse() proto.Message { return &StatusResponse{} } +func newSubmitQueueGatewayServiceListYARPCRequest() proto.Message { + return &ListRequest{} +} + +func newSubmitQueueGatewayServiceListYARPCResponse() proto.Message { + return &ListResponse{} +} + var ( emptySubmitQueueGatewayServicePingYARPCRequest = &PingRequest{} emptySubmitQueueGatewayServicePingYARPCResponse = &PingResponse{} @@ -371,50 +419,73 @@ var ( emptySubmitQueueGatewayServiceCancelYARPCResponse = &CancelResponse{} emptySubmitQueueGatewayServiceStatusYARPCRequest = &StatusRequest{} emptySubmitQueueGatewayServiceStatusYARPCResponse = &StatusResponse{} + emptySubmitQueueGatewayServiceListYARPCRequest = &ListRequest{} + emptySubmitQueueGatewayServiceListYARPCResponse = &ListResponse{} ) var yarpcFileDescriptorClosuref1a937782ebbded5 = [][]byte{ // gateway.proto []byte{ - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x54, 0x5f, 0x6b, 0xd4, 0x40, - 0x10, 0x6f, 0x2e, 0xed, 0xd9, 0xce, 0xdd, 0x95, 0xb2, 0x94, 0x12, 0x42, 0xc5, 0x76, 0x45, 0x7b, - 0x4f, 0x49, 0x39, 0x51, 0xc4, 0xe2, 0x4b, 0x4b, 0xf5, 0x45, 0x4b, 0x4d, 0x11, 0x41, 0x90, 0xb2, - 0xc9, 0x0d, 0xb9, 0xd0, 0x4b, 0x72, 0x97, 0xdd, 0x54, 0xce, 0x67, 0xfd, 0x06, 0x7e, 0x38, 0x3f, - 0x8e, 0xec, 0x9f, 0xdc, 0x25, 0x60, 0xae, 0x0f, 0x3e, 0x65, 0x67, 0xf2, 0xfb, 0xcd, 0xcc, 0xce, - 0xfc, 0x66, 0x61, 0x10, 0x33, 0x81, 0xdf, 0xd9, 0xc2, 0x9b, 0x15, 0xb9, 0xc8, 0x89, 0x53, 0x86, - 0x58, 0x78, 0xbc, 0x0c, 0xd3, 0x44, 0xcc, 0x4b, 0x2c, 0xd1, 0x33, 0xff, 0x5d, 0xca, 0x66, 0x89, - 0x1f, 0x32, 0x8e, 0x7e, 0x34, 0x61, 0x59, 0x8c, 0xbe, 0x22, 0x18, 0x43, 0xb3, 0xdd, 0xd3, 0x25, - 0x26, 0xc5, 0x22, 0x46, 0x2e, 0x0a, 0x26, 0x30, 0x5e, 0x18, 0x68, 0xc3, 0xa7, 0x19, 0xf4, 0x04, - 0x7a, 0xd7, 0x49, 0x16, 0x07, 0x38, 0x2f, 0x91, 0x0b, 0xe2, 0xc0, 0xa3, 0x14, 0x39, 0x67, 0x31, - 0x3a, 0xd6, 0x91, 0x35, 0xdc, 0x09, 0x2a, 0x93, 0xfe, 0xb2, 0xa0, 0xaf, 0x91, 0x7c, 0x96, 0x67, - 0x1c, 0xdb, 0xa1, 0xe4, 0x18, 0xfa, 0x1c, 0x8b, 0xfb, 0x24, 0xc2, 0xdb, 0x8c, 0xa5, 0xe8, 0x74, - 0xd4, 0xef, 0x9e, 0xf1, 0x5d, 0xb1, 0x14, 0xc9, 0x21, 0xec, 0x88, 0x24, 0x45, 0x2e, 0x58, 0x3a, - 0x73, 0xec, 0x23, 0x6b, 0x68, 0x07, 0x2b, 0x07, 0x71, 0x61, 0x7b, 0x92, 0x73, 0xa1, 0xc8, 0x9b, - 0x8a, 0xbc, 0xb4, 0xe9, 0x6f, 0x0b, 0x7a, 0x1f, 0x58, 0x36, 0xae, 0x2a, 0xde, 0x87, 0x2d, 0xd5, - 0x27, 0x53, 0x84, 0x36, 0xc8, 0x29, 0x74, 0x75, 0x63, 0x54, 0xf2, 0xde, 0xc8, 0xf1, 0x54, 0x5f, - 0x65, 0x6b, 0x3c, 0xd3, 0xb1, 0x0b, 0xf5, 0x09, 0x0c, 0x8e, 0xbc, 0x85, 0xed, 0xaa, 0x35, 0x2a, - 0xe7, 0xee, 0xe8, 0xb8, 0xc6, 0x69, 0xb6, 0xee, 0xc6, 0x1c, 0x82, 0x25, 0x85, 0x52, 0xe8, 0xeb, - 0xaa, 0x4c, 0x77, 0x08, 0x6c, 0xf2, 0x79, 0x32, 0x36, 0x55, 0xa9, 0x33, 0x3d, 0x83, 0xc1, 0x05, - 0xcb, 0x22, 0x9c, 0x56, 0xb5, 0xff, 0x03, 0x44, 0x0e, 0xa0, 0x5b, 0x20, 0xe3, 0x79, 0x66, 0xda, - 0x66, 0x2c, 0xba, 0x07, 0xbb, 0x15, 0x59, 0xa7, 0xa0, 0x4f, 0x61, 0x70, 0x23, 0x98, 0x28, 0xf9, - 0x9a, 0x70, 0xf4, 0x8f, 0x05, 0xbb, 0x15, 0xca, 0x94, 0x76, 0x00, 0x5d, 0xae, 0x3c, 0x06, 0x68, - 0x2c, 0xf2, 0x18, 0x60, 0xca, 0xb8, 0xb8, 0xc5, 0xa2, 0xc8, 0x0b, 0x93, 0x7d, 0x47, 0x7a, 0x2e, - 0xa5, 0x83, 0x04, 0xb0, 0x9d, 0xa2, 0x60, 0x63, 0x26, 0x98, 0x63, 0x1f, 0xd9, 0xc3, 0xde, 0xe8, - 0x95, 0xd7, 0x26, 0x56, 0xaf, 0x99, 0xd2, 0xfb, 0x68, 0x88, 0x97, 0x99, 0x28, 0x16, 0xc1, 0x32, - 0x8e, 0x7b, 0x06, 0x83, 0xc6, 0x2f, 0xb2, 0x07, 0xf6, 0x1d, 0x2e, 0x4c, 0x61, 0xf2, 0x28, 0xe7, - 0x7b, 0xcf, 0xa6, 0x65, 0xa5, 0x22, 0x6d, 0xbc, 0xe9, 0xbc, 0xb6, 0xe8, 0x31, 0x6c, 0xe9, 0xca, - 0xda, 0x45, 0x8b, 0x70, 0xf0, 0x39, 0x2b, 0x30, 0xca, 0xe3, 0x2c, 0xf9, 0x81, 0xe3, 0x4f, 0xb2, - 0x46, 0xcd, 0x79, 0x09, 0x5b, 0xfa, 0x9e, 0x96, 0xd2, 0xc7, 0x93, 0xf6, 0xab, 0x28, 0x7c, 0xa0, - 0xd1, 0x2b, 0xb5, 0x75, 0x6a, 0x6a, 0xa3, 0x0c, 0xf6, 0xcd, 0x0c, 0xae, 0x72, 0xf1, 0x2e, 0x2f, - 0xb3, 0xf1, 0x7f, 0x25, 0xa9, 0xe6, 0xd8, 0x59, 0xcd, 0x71, 0xf4, 0xd3, 0x06, 0x72, 0xa3, 0x88, - 0xea, 0x12, 0xef, 0x35, 0x8f, 0x7c, 0x81, 0x4d, 0xb9, 0x94, 0xe4, 0x59, 0x7b, 0xe8, 0xda, 0x7a, - 0xbb, 0xcf, 0x1f, 0x82, 0x19, 0x69, 0x6d, 0xc8, 0xc0, 0x52, 0xcf, 0xeb, 0x02, 0xd7, 0xb6, 0x70, - 0x5d, 0xe0, 0xfa, 0x5a, 0xd0, 0x0d, 0xf2, 0x0d, 0xba, 0x5a, 0xc7, 0xe4, 0xa4, 0x9d, 0xd3, 0x58, - 0x13, 0x77, 0xf8, 0x30, 0xb0, 0x1e, 0x5e, 0x6b, 0x6f, 0x5d, 0xf8, 0xc6, 0xda, 0xac, 0x0b, 0xdf, - 0x94, 0x31, 0xdd, 0x38, 0xbf, 0x83, 0xc3, 0x28, 0x4f, 0x5b, 0x09, 0xe7, 0x7d, 0x33, 0x98, 0x6b, - 0xf9, 0xb8, 0x5e, 0x5b, 0x5f, 0xcf, 0xe2, 0x44, 0x4c, 0xca, 0xd0, 0x8b, 0xf2, 0xd4, 0x97, 0x24, - 0xbf, 0x46, 0xf2, 0xe5, 0x63, 0x5d, 0xb7, 0x4d, 0x10, 0xfd, 0x5a, 0xcf, 0xc2, 0xb0, 0xab, 0x0e, - 0x2f, 0xfe, 0x06, 0x00, 0x00, 0xff, 0xff, 0x01, 0x08, 0x76, 0xa7, 0x23, 0x06, 0x00, 0x00, + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xa4, 0x56, 0x5d, 0x6f, 0xe3, 0x44, + 0x17, 0xae, 0xf3, 0xd5, 0xe4, 0x24, 0xe9, 0x9b, 0x77, 0xb4, 0xaa, 0xac, 0xb0, 0xb0, 0xa9, 0x11, + 0xbb, 0x11, 0x17, 0xc9, 0xaa, 0x88, 0x15, 0xa2, 0xe2, 0x22, 0xdb, 0x64, 0x11, 0xd2, 0x76, 0x29, + 0x4e, 0x2a, 0x24, 0x24, 0x64, 0x4d, 0xe2, 0xa3, 0xd4, 0x6a, 0x6d, 0xa7, 0x33, 0xe3, 0x85, 0x54, + 0xdc, 0xf2, 0x0b, 0xe0, 0xc7, 0x71, 0xc3, 0xef, 0xe0, 0x16, 0xcd, 0x87, 0x1d, 0x47, 0xaa, 0x5d, + 0xa4, 0xbd, 0x8a, 0xcf, 0x99, 0xe7, 0x7c, 0xcc, 0x33, 0xcf, 0xcc, 0x09, 0x74, 0xd7, 0x54, 0xe0, + 0x2f, 0x74, 0x3b, 0xda, 0xb0, 0x58, 0xc4, 0xc4, 0x4e, 0x96, 0xc8, 0x46, 0x3c, 0x59, 0x86, 0x81, + 0xb8, 0x4b, 0x30, 0xc1, 0x91, 0x59, 0xef, 0x3b, 0x74, 0x13, 0x8c, 0x97, 0x94, 0xe3, 0x78, 0x75, + 0x4d, 0xa3, 0x35, 0x8e, 0x55, 0x80, 0x31, 0x74, 0x74, 0xff, 0x65, 0x86, 0x09, 0x91, 0xad, 0x91, + 0x0b, 0x46, 0x05, 0xae, 0xb7, 0x06, 0xba, 0xe7, 0xd3, 0x11, 0xce, 0x0b, 0x68, 0x5f, 0x06, 0xd1, + 0xda, 0xc5, 0xbb, 0x04, 0xb9, 0x20, 0x36, 0x1c, 0x86, 0xc8, 0x39, 0x5d, 0xa3, 0x6d, 0x0d, 0xac, + 0x61, 0xcb, 0x4d, 0x4d, 0xe7, 0x77, 0x0b, 0x3a, 0x1a, 0xc9, 0x37, 0x71, 0xc4, 0xb1, 0x18, 0x4a, + 0x4e, 0xa0, 0xc3, 0x91, 0xbd, 0x0f, 0x56, 0xe8, 0x45, 0x34, 0x44, 0xbb, 0xa2, 0x96, 0xdb, 0xc6, + 0xf7, 0x8e, 0x86, 0x48, 0x9e, 0x42, 0x4b, 0x04, 0x21, 0x72, 0x41, 0xc3, 0x8d, 0x5d, 0x1d, 0x58, + 0xc3, 0xaa, 0xbb, 0x73, 0x90, 0x3e, 0x34, 0xaf, 0x63, 0x2e, 0x54, 0x70, 0x4d, 0x05, 0x67, 0xb6, + 0xf3, 0xa7, 0x05, 0xed, 0xb7, 0x34, 0xf2, 0xd3, 0x8e, 0x9f, 0x40, 0x5d, 0xf1, 0x64, 0x9a, 0xd0, + 0x06, 0x79, 0x09, 0x0d, 0x4d, 0x8c, 0x2a, 0xde, 0x3e, 0xb5, 0x47, 0x8a, 0x57, 0x49, 0xcd, 0xc8, + 0x30, 0x76, 0xae, 0x7e, 0x5c, 0x83, 0x23, 0xdf, 0x40, 0x33, 0xa5, 0x46, 0xd5, 0x3c, 0x3a, 0x3d, + 0xc9, 0xc5, 0xec, 0x53, 0x37, 0x37, 0x1f, 0x6e, 0x16, 0xe2, 0x38, 0xd0, 0xd1, 0x5d, 0x19, 0x76, + 0x08, 0xd4, 0xf8, 0x5d, 0xe0, 0x9b, 0xae, 0xd4, 0xb7, 0x73, 0x06, 0xdd, 0x73, 0x1a, 0xad, 0xf0, + 0x36, 0xed, 0xfd, 0x01, 0x10, 0x39, 0x86, 0x06, 0x43, 0xca, 0xe3, 0xc8, 0xd0, 0x66, 0x2c, 0xa7, + 0x07, 0x47, 0x69, 0xb0, 0x2e, 0xe1, 0x7c, 0x0a, 0xdd, 0xb9, 0xa0, 0x22, 0xe1, 0x25, 0xe9, 0x9c, + 0xbf, 0x2c, 0x38, 0x4a, 0x51, 0xa6, 0xb5, 0x63, 0x68, 0x70, 0xe5, 0x31, 0x40, 0x63, 0x91, 0x8f, + 0x01, 0x6e, 0x29, 0x17, 0x1e, 0x32, 0x16, 0x33, 0x53, 0xbd, 0x25, 0x3d, 0x33, 0xe9, 0x20, 0x2e, + 0x34, 0x43, 0x14, 0xd4, 0xa7, 0x82, 0xda, 0xd5, 0x41, 0x75, 0xd8, 0x3e, 0x7d, 0x35, 0x2a, 0x12, + 0xeb, 0x68, 0xbf, 0xe4, 0xe8, 0xc2, 0x04, 0xce, 0x22, 0xc1, 0xb6, 0x6e, 0x96, 0xa7, 0x7f, 0x06, + 0xdd, 0xbd, 0x25, 0xd2, 0x83, 0xea, 0x0d, 0x6e, 0x4d, 0x63, 0xf2, 0x53, 0x9e, 0xef, 0x7b, 0x7a, + 0x9b, 0xa4, 0x2a, 0xd2, 0xc6, 0xd7, 0x95, 0xaf, 0x2c, 0xe7, 0x1f, 0xa9, 0x84, 0x80, 0x8b, 0x72, + 0x25, 0x38, 0xd0, 0xe5, 0x82, 0x32, 0xe1, 0x49, 0x79, 0x79, 0x21, 0x57, 0x79, 0xaa, 0x6e, 0x5b, + 0x39, 0x17, 0x41, 0x88, 0x17, 0x9c, 0x7c, 0x02, 0x6d, 0x8c, 0xfc, 0x0c, 0x61, 0xf4, 0x88, 0x91, + 0x6f, 0xd6, 0xfb, 0x52, 0x1b, 0x72, 0x43, 0xc8, 0xed, 0xda, 0xa0, 0x2a, 0xf5, 0x98, 0xda, 0xe4, + 0x23, 0x68, 0x6d, 0xe8, 0x1a, 0x3d, 0x1e, 0xdc, 0xa3, 0x5d, 0x1f, 0x58, 0xc3, 0xba, 0xdb, 0x94, + 0x8e, 0x79, 0x70, 0x8f, 0x92, 0x52, 0xb5, 0x28, 0xe2, 0x1b, 0x8c, 0xec, 0x86, 0xa6, 0x54, 0x7a, + 0x16, 0xd2, 0x41, 0x5e, 0x41, 0x8d, 0xc7, 0x4c, 0xd8, 0x87, 0x4a, 0x6f, 0x4e, 0x31, 0x9d, 0x72, + 0x9b, 0xf3, 0x98, 0x09, 0x57, 0xe1, 0x9d, 0x3f, 0xaa, 0x70, 0x64, 0x76, 0x3d, 0x4f, 0xc2, 0x90, + 0xb2, 0xed, 0x83, 0x52, 0xca, 0x08, 0xa9, 0xe4, 0x09, 0x79, 0x06, 0x6d, 0x2d, 0x79, 0x2f, 0x61, + 0x01, 0x57, 0x47, 0xd9, 0x72, 0x41, 0xbb, 0xae, 0x58, 0xc0, 0x73, 0xfa, 0xa8, 0x95, 0xe8, 0xa3, + 0x5e, 0xa6, 0x8f, 0xc6, 0x63, 0xfa, 0xd8, 0xef, 0xbe, 0x48, 0x1f, 0xd9, 0xe1, 0xa1, 0xef, 0x51, + 0x21, 0x8f, 0xe6, 0x30, 0x77, 0x78, 0xe8, 0x4f, 0xc4, 0x05, 0x97, 0x98, 0x64, 0xe3, 0xd3, 0x1d, + 0xa6, 0xa9, 0x31, 0xc6, 0xa9, 0x30, 0xcf, 0xe1, 0x7f, 0xab, 0x38, 0xdc, 0xdc, 0xe2, 0x0e, 0xd5, + 0x52, 0xa8, 0x6e, 0xe6, 0x96, 0xb8, 0x0f, 0xd3, 0xe3, 0x6f, 0xd0, 0xd1, 0x72, 0x34, 0xf7, 0x6c, + 0x0a, 0x4d, 0xa6, 0xb7, 0x29, 0x6f, 0x9a, 0x24, 0x64, 0xf8, 0x5f, 0x09, 0x71, 0xb3, 0x48, 0xd9, + 0x7a, 0x84, 0xbf, 0x0a, 0x2f, 0xa7, 0x23, 0x5d, 0xb9, 0x2b, 0xdd, 0x97, 0xa9, 0x96, 0x9c, 0x13, + 0xa8, 0xeb, 0x73, 0x28, 0x7e, 0xc2, 0x11, 0x8e, 0xaf, 0x22, 0x86, 0xab, 0x78, 0x1d, 0x05, 0xf7, + 0xe8, 0xff, 0x20, 0x1b, 0xd0, 0x31, 0x5f, 0x42, 0x5d, 0x9f, 0xaa, 0xa5, 0x5e, 0xcb, 0x67, 0xc5, + 0x7d, 0x2a, 0xbc, 0xab, 0xd1, 0x0f, 0x0b, 0xcc, 0xa1, 0xf0, 0xc4, 0xec, 0xe6, 0x5d, 0x2c, 0xde, + 0xc4, 0x49, 0xe4, 0x7f, 0x50, 0x91, 0x54, 0xd9, 0x95, 0x9d, 0xb2, 0x3f, 0x9f, 0x40, 0x33, 0xbd, + 0x12, 0xa4, 0x07, 0x9d, 0xf9, 0xf7, 0xee, 0xc2, 0x9b, 0xce, 0xde, 0x4c, 0xae, 0xde, 0x2e, 0x7a, + 0x07, 0xd2, 0x33, 0x99, 0x5e, 0x7c, 0xb7, 0x58, 0xcc, 0xa6, 0xde, 0x64, 0x7e, 0xde, 0xb3, 0xc8, + 0xff, 0xa1, 0x9b, 0x79, 0xa6, 0xb3, 0xf9, 0x79, 0xaf, 0x72, 0xfa, 0x77, 0x15, 0xc8, 0x5c, 0xd5, + 0x56, 0x3c, 0x7c, 0xab, 0x4b, 0x93, 0x1f, 0xa1, 0x26, 0xa7, 0x1c, 0xf9, 0xac, 0xb8, 0xbb, 0xdc, + 0xbc, 0xec, 0x3f, 0x7f, 0x0c, 0x66, 0xde, 0xea, 0x03, 0x99, 0x58, 0x0e, 0x88, 0xb2, 0xc4, 0xb9, + 0xb1, 0x56, 0x96, 0x38, 0x3f, 0x67, 0x9c, 0x03, 0xf2, 0x33, 0x34, 0xf4, 0x60, 0x20, 0x2f, 0x8a, + 0x63, 0xf6, 0xe6, 0x4e, 0x7f, 0xf8, 0x38, 0x30, 0x9f, 0x5e, 0x3f, 0xe6, 0x65, 0xe9, 0xf7, 0xe6, + 0x50, 0x59, 0xfa, 0xfd, 0xb9, 0x60, 0x68, 0x09, 0xb8, 0x28, 0xa5, 0x65, 0xf7, 0xc6, 0x97, 0xd2, + 0x92, 0xbb, 0x7b, 0xce, 0xc1, 0xeb, 0x1b, 0x78, 0xba, 0x8a, 0xc3, 0x42, 0xf8, 0xeb, 0x8e, 0x39, + 0xf1, 0x4b, 0xf9, 0x37, 0xe8, 0xd2, 0xfa, 0xe9, 0x6c, 0x1d, 0x88, 0xeb, 0x64, 0x39, 0x5a, 0xc5, + 0xe1, 0x58, 0x06, 0x8d, 0x73, 0x41, 0x63, 0xf9, 0xb7, 0x2a, 0x6f, 0x9b, 0x24, 0xfa, 0x7f, 0xd5, + 0x66, 0xb9, 0x6c, 0xa8, 0x8f, 0x2f, 0xfe, 0x0d, 0x00, 0x00, 0xff, 0xff, 0xc4, 0xa9, 0x92, 0x5e, + 0xcd, 0x09, 0x00, 0x00, }, // api/base/change/proto/change.proto []byte{ diff --git a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go index f271309f..a63cca5e 100644 --- a/api/submitqueue/gateway/protopb/gateway_grpc.pb.go +++ b/api/submitqueue/gateway/protopb/gateway_grpc.pb.go @@ -38,6 +38,7 @@ const ( SubmitQueueGateway_Land_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Land" SubmitQueueGateway_Cancel_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Cancel" SubmitQueueGateway_Status_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/Status" + SubmitQueueGateway_List_FullMethodName = "/uber.submitqueue.gateway.SubmitQueueGateway/List" ) // SubmitQueueGatewayClient is the client API for SubmitQueueGateway service. @@ -66,6 +67,8 @@ type SubmitQueueGatewayClient interface { // Status returns the current status of a previously submitted request, identified by its sqid. // The status is eventually consistent with the request store and reconciled from the append-only request log. Status(ctx context.Context, in *StatusRequest, opts ...grpc.CallOption) (*StatusResponse, error) + // List returns request summaries for one queue admitted during a time window. + List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) } type submitQueueGatewayClient struct { @@ -116,6 +119,16 @@ func (c *submitQueueGatewayClient) Status(ctx context.Context, in *StatusRequest return out, nil } +func (c *submitQueueGatewayClient) List(ctx context.Context, in *ListRequest, opts ...grpc.CallOption) (*ListResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(ListResponse) + err := c.cc.Invoke(ctx, SubmitQueueGateway_List_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // SubmitQueueGatewayServer is the server API for SubmitQueueGateway service. // All implementations must embed UnimplementedSubmitQueueGatewayServer // for forward compatibility. @@ -142,6 +155,8 @@ type SubmitQueueGatewayServer interface { // Status returns the current status of a previously submitted request, identified by its sqid. // The status is eventually consistent with the request store and reconciled from the append-only request log. Status(context.Context, *StatusRequest) (*StatusResponse, error) + // List returns request summaries for one queue admitted during a time window. + List(context.Context, *ListRequest) (*ListResponse, error) mustEmbedUnimplementedSubmitQueueGatewayServer() } @@ -164,6 +179,9 @@ func (UnimplementedSubmitQueueGatewayServer) Cancel(context.Context, *CancelRequ func (UnimplementedSubmitQueueGatewayServer) Status(context.Context, *StatusRequest) (*StatusResponse, error) { return nil, status.Errorf(codes.Unimplemented, "method Status not implemented") } +func (UnimplementedSubmitQueueGatewayServer) List(context.Context, *ListRequest) (*ListResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method List not implemented") +} func (UnimplementedSubmitQueueGatewayServer) mustEmbedUnimplementedSubmitQueueGatewayServer() {} func (UnimplementedSubmitQueueGatewayServer) testEmbeddedByValue() {} @@ -257,6 +275,24 @@ func _SubmitQueueGateway_Status_Handler(srv interface{}, ctx context.Context, de return interceptor(ctx, in, info, handler) } +func _SubmitQueueGateway_List_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ListRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(SubmitQueueGatewayServer).List(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: SubmitQueueGateway_List_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(SubmitQueueGatewayServer).List(ctx, req.(*ListRequest)) + } + return interceptor(ctx, in, info, handler) +} + // SubmitQueueGateway_ServiceDesc is the grpc.ServiceDesc for SubmitQueueGateway service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -280,6 +316,10 @@ var SubmitQueueGateway_ServiceDesc = grpc.ServiceDesc{ MethodName: "Status", Handler: _SubmitQueueGateway_Status_Handler, }, + { + MethodName: "List", + Handler: _SubmitQueueGateway_List_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "gateway.proto", diff --git a/service/submitqueue/gateway/server/main.go b/service/submitqueue/gateway/server/main.go index e3f2593d..19efc54e 100644 --- a/service/submitqueue/gateway/server/main.go +++ b/service/submitqueue/gateway/server/main.go @@ -55,6 +55,7 @@ type GatewayServer struct { landController *controller.LandController cancelController *controller.CancelController statusController *controller.StatusController + listController *controller.ListController } // Ping delegates to the controller @@ -77,6 +78,11 @@ func (s *GatewayServer) Status(ctx context.Context, req *pb.StatusRequest) (*pb. return s.statusController.Status(ctx, req) } +// List delegates to the controller +func (s *GatewayServer) List(ctx context.Context, req *pb.ListRequest) (*pb.ListResponse, error) { + return s.listController.List(ctx, req) +} + func main() { code := 0 if err := run(); err != nil { @@ -237,6 +243,7 @@ func run() error { return fmt.Errorf("failed to create storage: %w", err) } requestLogStore := store.GetRequestLogStore() + requestSummaryStore := store.GetRequestSummaryStore() // Load queue configurations from YAML. Path is required so the gateway // can reject requests for unknown queues at the edge. @@ -252,13 +259,15 @@ func run() error { // Create controllers and wrap them for gRPC pingController := controller.NewPingController(logger, scope) landController := controller.NewLandController(logger.Sugar(), scope, cnt, store, queueConfigs, registry) - cancelController := controller.NewCancelController(logger.Sugar(), scope, requestLogStore, registry) + cancelController := controller.NewCancelController(logger.Sugar(), scope, store, registry) statusController := controller.NewStatusController(logger.Sugar(), scope, requestLogStore) + listController := controller.NewListController(logger.Sugar(), scope, requestSummaryStore, queueConfigs) gatewayServer := &GatewayServer{ pingController: pingController, landController: landController, cancelController: cancelController, statusController: statusController, + listController: listController, } pb.RegisterSubmitQueueGatewayServer(grpcServer, gatewayServer) diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go index 6dee8839..66f83bc4 100644 --- a/submitqueue/entity/request_log.go +++ b/submitqueue/entity/request_log.go @@ -148,3 +148,30 @@ func RequestLogFromBytes(data []byte) (RequestLog, error) { } return log, nil } + +// IsKnownRequestStatus reports whether status is a public status emitted by SubmitQueue. +func IsKnownRequestStatus(status RequestStatus) bool { + switch status { + case RequestStatusAccepted, + RequestStatusStarted, + RequestStatusValidating, + RequestStatusValidated, + RequestStatusBatching, + RequestStatusBatched, + RequestStatusScored, + RequestStatusSpeculating, + RequestStatusSpeculated, + RequestStatusBuilding, + RequestStatusBuilt, + RequestStatusWaitingPath, + RequestStatusLanding, + RequestStatusProcessing, + RequestStatusLanded, + RequestStatusError, + RequestStatusCancelling, + RequestStatusCancelled: + return true + default: + return false + } +} diff --git a/submitqueue/gateway/controller/BUILD.bazel b/submitqueue/gateway/controller/BUILD.bazel index a2dbdbde..9605c56c 100644 --- a/submitqueue/gateway/controller/BUILD.bazel +++ b/submitqueue/gateway/controller/BUILD.bazel @@ -5,6 +5,7 @@ go_library( srcs = [ "cancel.go", "land.go", + "list.go", "ping.go", "status.go", ], @@ -35,6 +36,7 @@ go_test( srcs = [ "cancel_test.go", "land_test.go", + "list_test.go", "ping_test.go", "status_test.go", ], diff --git a/submitqueue/gateway/controller/cancel.go b/submitqueue/gateway/controller/cancel.go index 82abaaa5..424ed7cf 100644 --- a/submitqueue/gateway/controller/cancel.go +++ b/submitqueue/gateway/controller/cancel.go @@ -24,6 +24,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -36,21 +37,21 @@ import ( // and returns a response. The orchestrator-side cancel controller performs the actual // state transitions and emits the terminal RequestStatusCancelled log entry. type CancelController struct { - logger *zap.SugaredLogger - metricsScope tally.Scope - requestLogStore storage.RequestLogStore - registry consumer.TopicRegistry + logger *zap.SugaredLogger + metricsScope tally.Scope + store storage.Storage + registry consumer.TopicRegistry } // NewCancelController creates a new instance of the gateway cancel controller. // The controller writes a RequestStatusCancelling log entry through requestLogStore and // publishes cancel requests to the topic registered under topickey.TopicKeyCancel. -func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, requestLogStore storage.RequestLogStore, registry consumer.TopicRegistry) *CancelController { +func NewCancelController(logger *zap.SugaredLogger, scope tally.Scope, store storage.Storage, registry consumer.TopicRegistry) *CancelController { return &CancelController{ - logger: logger, - metricsScope: scope, - requestLogStore: requestLogStore, - registry: registry, + logger: logger, + metricsScope: scope, + store: store, + registry: registry, } } @@ -90,7 +91,7 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (* // controller writes its "accepted" log entry synchronously to the same store, so // a NotFound here reliably means "this sqid was never accepted by the gateway" // rather than "in flight" — there is no false-negative race window. - if _, err := c.requestLogStore.List(ctx, cancelRequest.ID); err != nil { + if _, err := c.store.GetRequestLogStore().List(ctx, cancelRequest.ID); err != nil { if storage.IsNotFound(err) { c.metricsScope.Counter("cancel_request_not_found").Inc(1) return nil, errs.NewUserError(&RequestNotFoundError{Sqid: cancelRequest.ID}) @@ -106,8 +107,8 @@ func (c *CancelController) Cancel(ctx context.Context, req *pb.CancelRequest) (* metadata["reason"] = cancelRequest.Reason } logEntry := entity.NewRequestLog(cancelRequest.ID, entity.RequestStatusCancelling, 0, "", metadata) - if err := c.requestLogStore.Insert(ctx, logEntry); err != nil { - return nil, fmt.Errorf("CancelController failed to insert cancelling log for sqid=%s: %w", cancelRequest.ID, err) + if err := request.PersistLog(ctx, c.store, logEntry); err != nil { + return nil, fmt.Errorf("CancelController failed to persist cancelling log for sqid=%s: %w", cancelRequest.ID, err) } if err := c.publishToQueue(ctx, cancelRequest); err != nil { diff --git a/submitqueue/gateway/controller/cancel_test.go b/submitqueue/gateway/controller/cancel_test.go index 2b874f5d..2f2eba7d 100644 --- a/submitqueue/gateway/controller/cancel_test.go +++ b/submitqueue/gateway/controller/cancel_test.go @@ -58,27 +58,49 @@ func newCancelTestRegistryWithNoopPublisher(t *testing.T, ctrl *gomock.Controlle return registry } -// newRequestLogStoreNoop returns a RequestLogStore mock whose List returns a single -// dummy entry (so existence check passes) and whose Insert silently succeeds for any input. -func newRequestLogStoreNoop(t *testing.T, ctrl *gomock.Controller) *storagemock.MockRequestLogStore { +// newCancelStorageNoop returns a storage mock whose request log existence check, +// log insert, and summary upsert all succeed. +func newCancelStorageNoop(t *testing.T, ctrl *gomock.Controller) storage.Storage { t.Helper() - store := storagemock.NewMockRequestLogStore(ctrl) - store.EXPECT().List(gomock.Any(), gomock.Any()).Return([]entity.RequestLog{{}}, nil).AnyTimes() - store.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + logStore := storagemock.NewMockRequestLogStore(ctrl) + logStore.EXPECT().List(gomock.Any(), gomock.Any()).Return([]entity.RequestLog{{}}, nil).AnyTimes() + logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + contextStore := storagemock.NewMockRequestContextStore(ctrl) + contextStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + contextStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestContext{Queue: "test-queue"}, nil).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + summaryStore.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.RequestSummary{Version: 1}, nil).AnyTimes() + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestContextStore().Return(contextStore).AnyTimes() + return store +} + +func newCancelStorage(t *testing.T, ctrl *gomock.Controller, logStore storage.RequestLogStore, summaryStore storage.RequestSummaryStore) storage.Storage { + t.Helper() + contextStore := storagemock.NewMockRequestContextStore(ctrl) + contextStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestContext{Queue: "my-queue"}, nil).AnyTimes() + store := storagemock.NewMockStorage(ctrl) + store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestContextStore().Return(contextStore).AnyTimes() return store } func TestNewCancelController(t *testing.T) { ctrl := gomock.NewController(t) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) require.NotNil(t, controller) } func TestCancel_HappyPath(t *testing.T) { ctrl := gomock.NewController(t) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() req := &pb.CancelRequest{Sqid: "test-queue/42", Reason: "user changed their mind"} @@ -91,7 +113,7 @@ func TestCancel_HappyPath(t *testing.T) { func TestCancel_ReturnsErrorOnEmptySqid(t *testing.T) { ctrl := gomock.NewController(t) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageNoop(t, ctrl), newCancelTestRegistryWithNoopPublisher(t, ctrl)) ctx := context.Background() req := &pb.CancelRequest{Sqid: "", Reason: "anything"} @@ -116,7 +138,7 @@ func TestCancel_PublishesToQueue(t *testing.T) { }, ) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageNoop(t, ctrl), registry) ctx := context.Background() req := &pb.CancelRequest{Sqid: "my-queue/7", Reason: "obsolete change"} @@ -148,6 +170,9 @@ func TestCancel_InsertsCancellingLog(t *testing.T) { return nil }, ).Times(1) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + summaryStore.EXPECT().Get(gomock.Any(), "my-queue", "my-queue/42").Return(entity.RequestSummary{Version: 1}, nil) + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), int64(1), int64(2)).Return(nil) registry, publisher := newCancelTestRegistry(t, ctrl) insertedBeforePublish := false @@ -158,7 +183,7 @@ func TestCancel_InsertsCancellingLog(t *testing.T) { }, ) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorage(t, ctrl, logStore, summaryStore), registry) req := &pb.CancelRequest{Sqid: "my-queue/42", Reason: "obsolete change"} _, err := controller.Cancel(context.Background(), req) @@ -178,12 +203,13 @@ func TestCancel_LogInsertFailure(t *testing.T) { logStore := storagemock.NewMockRequestLogStore(ctrl) logStore.EXPECT().List(gomock.Any(), "q/1").Return([]entity.RequestLog{{}}, nil) logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(fmt.Errorf("db unavailable")) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) registry, publisher := newCancelTestRegistry(t, ctrl) // No Publish expectation: log insert must fail before publish runs. _ = publisher - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorage(t, ctrl, logStore, summaryStore), registry) _, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "q/1"}) require.Error(t, err) } @@ -194,7 +220,7 @@ func TestCancel_ReturnsErrorOnPublishFailure(t *testing.T) { registry, publisher := newCancelTestRegistry(t, ctrl) publisher.EXPECT().Publish(gomock.Any(), gomock.Any(), gomock.Any()).Return(fmt.Errorf("queue unavailable")) - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newRequestLogStoreNoop(t, ctrl), registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorageNoop(t, ctrl), registry) ctx := context.Background() req := &pb.CancelRequest{Sqid: "test-queue/1"} @@ -212,12 +238,13 @@ func TestCancel_UnknownSqidIsUserError(t *testing.T) { logStore := storagemock.NewMockRequestLogStore(ctrl) logStore.EXPECT().List(gomock.Any(), "ghost/1").Return(nil, storage.ErrNotFound) // No Insert expectation: existence check must short-circuit before Insert. + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) registry, publisher := newCancelTestRegistry(t, ctrl) // No Publish expectation: existence check must short-circuit before Publish. _ = publisher - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorage(t, ctrl, logStore, summaryStore), registry) _, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "ghost/1"}) require.Error(t, err) assert.True(t, IsRequestNotFound(err)) @@ -237,11 +264,12 @@ func TestCancel_RequestLogLookupFailure(t *testing.T) { logStore := storagemock.NewMockRequestLogStore(ctrl) logStore.EXPECT().List(gomock.Any(), "q/1").Return(nil, fmt.Errorf("log backend down")) + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) registry, publisher := newCancelTestRegistry(t, ctrl) _ = publisher - controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, logStore, registry) + controller := NewCancelController(zap.NewNop().Sugar(), tally.NoopScope, newCancelStorage(t, ctrl, logStore, summaryStore), registry) _, err := controller.Cancel(context.Background(), &pb.CancelRequest{Sqid: "q/1"}) require.Error(t, err) assert.False(t, errs.IsUserError(err)) diff --git a/submitqueue/gateway/controller/land.go b/submitqueue/gateway/controller/land.go index 6e15c72f..2290103b 100644 --- a/submitqueue/gateway/controller/land.go +++ b/submitqueue/gateway/controller/land.go @@ -29,6 +29,7 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/extension/counter" "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/core/topickey" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/queueconfig" @@ -137,8 +138,16 @@ func (c *LandController) Land(ctx context.Context, req *pb.LandRequest) (resp *p // It is important to record the status before publishing to the queue for processing. It is important to publish straight to the database and not via a entityqueue. // Gateway has to stay consistent with the request log. logEntry := entity.NewRequestLog(landRequest.ID, entity.RequestStatusAccepted, 0, "", nil) - if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil { - return nil, fmt.Errorf("LandController failed to insert request log for sqid=%s: %w", landRequest.ID, err) + if err := request.CreateContext(ctx, c.store, entity.RequestContext{ + RequestID: landRequest.ID, + Queue: queue, + ChangeURIs: append([]string(nil), change.URIs...), + AdmittedAtMs: logEntry.TimestampMs, + }); err != nil { + return nil, fmt.Errorf("LandController failed to create request context for sqid=%s: %w", landRequest.ID, err) + } + if err := request.PersistLog(ctx, c.store, logEntry); err != nil { + return nil, fmt.Errorf("LandController failed to persist request log for sqid=%s: %w", landRequest.ID, err) } c.logger.Debugw("land request created", diff --git a/submitqueue/gateway/controller/land_test.go b/submitqueue/gateway/controller/land_test.go index 01252d61..050c714c 100644 --- a/submitqueue/gateway/controller/land_test.go +++ b/submitqueue/gateway/controller/land_test.go @@ -71,8 +71,17 @@ func newTestRegistryWithNoopPublisher(t *testing.T, ctrl *gomock.Controller) con func noopStorage(ctrl *gomock.Controller) storage.Storage { logStore := storagemock.NewMockRequestLogStore(ctrl) logStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + contextStore := storagemock.NewMockRequestContextStore(ctrl) + contextStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + contextStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestContext{Queue: "test-queue"}, nil).AnyTimes() + summaryStore.EXPECT().Create(gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + summaryStore.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.RequestSummary{Version: 1}, nil).AnyTimes() + summaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestLogStore().Return(logStore).AnyTimes() + store.EXPECT().GetRequestSummaryStore().Return(summaryStore).AnyTimes() + store.EXPECT().GetRequestContextStore().Return(contextStore).AnyTimes() return store } diff --git a/submitqueue/gateway/controller/list.go b/submitqueue/gateway/controller/list.go new file mode 100644 index 00000000..bfa24de0 --- /dev/null +++ b/submitqueue/gateway/controller/list.go @@ -0,0 +1,261 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "context" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "sort" + "time" + + "github.com/uber-go/tally" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queueconfig" + "github.com/uber/submitqueue/submitqueue/extension/storage" + "go.uber.org/zap" +) + +const ( + defaultListPageSize = 50 + maxListPageSize = 200 +) + +// ListController handles queue request listing for the gateway. +type ListController struct { + logger *zap.SugaredLogger + metricsScope tally.Scope + requestSummaryStore storage.RequestSummaryStore + queueConfigs queueconfig.Store +} + +// NewListController creates a new instance of the gateway list controller. +func NewListController(logger *zap.SugaredLogger, scope tally.Scope, requestSummaryStore storage.RequestSummaryStore, queueConfigs queueconfig.Store) *ListController { + return &ListController{ + logger: logger, + metricsScope: scope.SubScope("list_controller"), + requestSummaryStore: requestSummaryStore, + queueConfigs: queueConfigs, + } +} + +// List returns request summaries for one queue admitted during a time window. +func (c *ListController) List(ctx context.Context, req *pb.ListRequest) (*pb.ListResponse, error) { + start := time.Now() + defer func() { + c.metricsScope.Timer("list_latency").Record(time.Since(start)) + }() + c.metricsScope.Counter("list_count").Inc(1) + + if req.Queue == "" { + return nil, fmt.Errorf("ListController requires the request to have a queue name specified: %w", ErrInvalidRequest) + } + if req.StartTimeMs <= 0 || req.EndTimeMs <= 0 || req.StartTimeMs >= req.EndTimeMs { + return nil, fmt.Errorf("ListController requires a valid non-empty time window: %w", ErrInvalidRequest) + } + if req.PageSize < 0 { + return nil, fmt.Errorf("ListController page_size must be non-negative: %w", ErrInvalidRequest) + } + + statuses, err := canonicalStatuses(req.Statuses) + if err != nil { + return nil, fmt.Errorf("ListController invalid status filter: %w", err) + } + sortMode, err := requestSummarySort(req.Sort) + if err != nil { + return nil, fmt.Errorf("ListController invalid sort: %w", err) + } + + if _, err := c.queueConfigs.Get(ctx, req.Queue); err != nil { + if errors.Is(err, queueconfig.ErrNotFound) { + return nil, errs.NewUserError(&UnrecognizedQueueError{Queue: req.Queue}) + } + return nil, fmt.Errorf("ListController failed to look up queue %q: %w", req.Queue, err) + } + + limit := int(req.PageSize) + if limit == 0 { + limit = defaultListPageSize + } + if limit > maxListPageSize { + limit = maxListPageSize + } + + cursor, err := decodeListPageToken(req.PageToken) + if err != nil { + return nil, fmt.Errorf("ListController invalid page token: %w", err) + } + if cursor != nil && !cursor.matches(req.Queue, req.StartTimeMs, req.EndTimeMs, statuses, sortMode) { + return nil, fmt.Errorf("ListController page token does not match query: %w", ErrInvalidRequest) + } + + result, err := c.requestSummaryStore.List(ctx, storage.RequestSummaryListOptions{ + Queue: req.Queue, + StartTimeMs: req.StartTimeMs, + EndTimeMs: req.EndTimeMs, + Statuses: statuses, + Sort: sortMode, + Cursor: cursorStorage(cursor), + Limit: limit, + }) + if err != nil { + return nil, fmt.Errorf("ListController failed to list request summaries for queue=%s: %w", req.Queue, err) + } + + resp := &pb.ListResponse{ + Requests: make([]*pb.RequestSummary, 0, len(result.Requests)), + } + for _, summary := range result.Requests { + resp.Requests = append(resp.Requests, protoRequestSummary(summary)) + } + if result.NextCursor != nil { + token, err := encodeListPageToken(listPageToken{ + Queue: req.Queue, + StartTimeMs: req.StartTimeMs, + EndTimeMs: req.EndTimeMs, + Statuses: statusesToStrings(statuses), + Sort: sortMode, + StartedAtMs: result.NextCursor.StartedAtMs, + RequestID: result.NextCursor.RequestID, + }) + if err != nil { + return nil, fmt.Errorf("ListController failed to encode next page token: %w", err) + } + resp.NextPageToken = token + } + + c.logger.Debugw("request summaries listed", + "queue", req.Queue, + "count", len(resp.Requests), + "has_next_page", resp.NextPageToken != "", + ) + + return resp, nil +} + +type listPageToken struct { + Queue string `json:"queue"` + StartTimeMs int64 `json:"start_time_ms"` + EndTimeMs int64 `json:"end_time_ms"` + Statuses []string `json:"statuses"` + Sort storage.RequestSummarySort `json:"sort"` + StartedAtMs int64 `json:"started_at_ms"` + RequestID string `json:"request_id"` +} + +func (t listPageToken) matches(queue string, startTimeMs, endTimeMs int64, statuses []entity.RequestStatus, sortMode storage.RequestSummarySort) bool { + return t.Queue == queue && + t.StartTimeMs == startTimeMs && + t.EndTimeMs == endTimeMs && + t.Sort == sortMode && + equalStrings(t.Statuses, statusesToStrings(statuses)) +} + +func encodeListPageToken(token listPageToken) (string, error) { + data, err := json.Marshal(token) + if err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(data), nil +} + +func decodeListPageToken(raw string) (*listPageToken, error) { + if raw == "" { + return nil, nil + } + data, err := base64.RawURLEncoding.DecodeString(raw) + if err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalidRequest, err) + } + var token listPageToken + if err := json.Unmarshal(data, &token); err != nil { + return nil, fmt.Errorf("%w: %w", ErrInvalidRequest, err) + } + return &token, nil +} + +func cursorStorage(token *listPageToken) *storage.RequestSummaryCursor { + if token == nil { + return nil + } + return &storage.RequestSummaryCursor{StartedAtMs: token.StartedAtMs, RequestID: token.RequestID} +} + +func canonicalStatuses(raw []string) ([]entity.RequestStatus, error) { + seen := make(map[entity.RequestStatus]struct{}, len(raw)) + var statuses []entity.RequestStatus + for _, statusRaw := range raw { + status := entity.RequestStatus(statusRaw) + if !entity.IsKnownRequestStatus(status) { + return nil, fmt.Errorf("unknown status %q: %w", statusRaw, ErrInvalidRequest) + } + if _, ok := seen[status]; ok { + continue + } + seen[status] = struct{}{} + statuses = append(statuses, status) + } + sort.Slice(statuses, func(i, j int) bool { return statuses[i] < statuses[j] }) + return statuses, nil +} + +func requestSummarySort(sort pb.ListSort) (storage.RequestSummarySort, error) { + switch sort { + case pb.ListSort_SORT_DEFAULT, pb.ListSort_ADMITTED_ASC: + return storage.RequestSummarySortAdmittedAsc, nil + case pb.ListSort_ADMITTED_DESC: + return storage.RequestSummarySortAdmittedDesc, nil + default: + return "", fmt.Errorf("unknown sort %v: %w", sort, ErrInvalidRequest) + } +} + +func statusesToStrings(statuses []entity.RequestStatus) []string { + out := make([]string, len(statuses)) + for i, status := range statuses { + out[i] = string(status) + } + return out +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func protoRequestSummary(summary entity.RequestSummary) *pb.RequestSummary { + return &pb.RequestSummary{ + Sqid: summary.RequestID, + Queue: summary.Queue, + ChangeUris: summary.ChangeURIs, + Status: string(summary.Status), + LastError: summary.LastError, + Metadata: summary.Metadata, + StartedAtMs: summary.StartedAtMs, + UpdatedAtMs: summary.UpdatedAtMs, + CompletedAtMs: summary.CompletedAtMs, + } +} diff --git a/submitqueue/gateway/controller/list_test.go b/submitqueue/gateway/controller/list_test.go new file mode 100644 index 00000000..175d86a9 --- /dev/null +++ b/submitqueue/gateway/controller/list_test.go @@ -0,0 +1,286 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package controller + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + pb "github.com/uber/submitqueue/api/submitqueue/gateway/protopb" + "github.com/uber/submitqueue/platform/errs" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/queueconfig" + qcmock "github.com/uber/submitqueue/submitqueue/extension/queueconfig/mock" + "github.com/uber/submitqueue/submitqueue/extension/storage" + storagemock "github.com/uber/submitqueue/submitqueue/extension/storage/mock" + "go.uber.org/mock/gomock" + "go.uber.org/zap" +) + +func TestList_ReturnsSummaries(t *testing.T) { + ctrl := gomock.NewController(t) + + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + summaryStore.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, opts storage.RequestSummaryListOptions) (storage.RequestSummaryListResult, error) { + assert.Equal(t, "q", opts.Queue) + assert.Equal(t, int64(100), opts.StartTimeMs) + assert.Equal(t, int64(200), opts.EndTimeMs) + assert.Equal(t, []entity.RequestStatus{entity.RequestStatusBuilding, entity.RequestStatusLanded}, opts.Statuses) + assert.Equal(t, storage.RequestSummarySortAdmittedAsc, opts.Sort) + assert.Equal(t, defaultListPageSize, opts.Limit) + require.Nil(t, opts.Cursor) + return storage.RequestSummaryListResult{ + Requests: []entity.RequestSummary{ + { + RequestID: "q/2", + Queue: "q", + ChangeURIs: []string{"github://uber/repo/pull/2/abcdef"}, + Status: entity.RequestStatusBuilding, + LastError: "last", + Metadata: map[string]string{"k": "v"}, + StartedAtMs: 150, + UpdatedAtMs: 175, + CompletedAtMs: 0, + }, + }, + NextCursor: &storage.RequestSummaryCursor{StartedAtMs: 150, RequestID: "q/2"}, + }, nil + }, + ) + + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + resp, err := controller.List(context.Background(), &pb.ListRequest{ + Queue: "q", + StartTimeMs: 100, + EndTimeMs: 200, + Statuses: []string{"landed", "building", "landed"}, + }) + + require.NoError(t, err) + require.Len(t, resp.Requests, 1) + assert.Equal(t, "q/2", resp.Requests[0].Sqid) + assert.Equal(t, []string{"github://uber/repo/pull/2/abcdef"}, resp.Requests[0].ChangeUris) + assert.Equal(t, "building", resp.Requests[0].Status) + assert.Equal(t, "last", resp.Requests[0].LastError) + assert.Equal(t, map[string]string{"k": "v"}, resp.Requests[0].Metadata) + assert.Equal(t, int64(150), resp.Requests[0].StartedAtMs) + assert.Equal(t, int64(175), resp.Requests[0].UpdatedAtMs) + assert.NotEmpty(t, resp.NextPageToken) +} + +func TestList_UsesRequestedSort(t *testing.T) { + ctrl := gomock.NewController(t) + + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + summaryStore.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, opts storage.RequestSummaryListOptions) (storage.RequestSummaryListResult, error) { + assert.Equal(t, storage.RequestSummarySortAdmittedDesc, opts.Sort) + return storage.RequestSummaryListResult{}, nil + }, + ) + + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + _, err := controller.List(context.Background(), &pb.ListRequest{ + Queue: "q", + StartTimeMs: 100, + EndTimeMs: 200, + Sort: pb.ListSort_ADMITTED_DESC, + }) + + require.NoError(t, err) +} + +func TestList_UsesPageTokenCursor(t *testing.T) { + ctrl := gomock.NewController(t) + + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + + token, err := encodeListPageToken(listPageToken{ + Queue: "q", + StartTimeMs: 100, + EndTimeMs: 200, + Statuses: []string{"building"}, + Sort: storage.RequestSummarySortAdmittedAsc, + StartedAtMs: 150, + RequestID: "q/2", + }) + require.NoError(t, err) + + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + summaryStore.EXPECT().List(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, opts storage.RequestSummaryListOptions) (storage.RequestSummaryListResult, error) { + require.NotNil(t, opts.Cursor) + assert.Equal(t, int64(150), opts.Cursor.StartedAtMs) + assert.Equal(t, "q/2", opts.Cursor.RequestID) + assert.Equal(t, storage.RequestSummarySortAdmittedAsc, opts.Sort) + assert.Equal(t, 10, opts.Limit) + return storage.RequestSummaryListResult{}, nil + }, + ) + + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + _, err = controller.List(context.Background(), &pb.ListRequest{ + Queue: "q", + StartTimeMs: 100, + EndTimeMs: 200, + Statuses: []string{"building"}, + PageSize: 10, + PageToken: token, + }) + require.NoError(t, err) +} + +func TestList_Errors(t *testing.T) { + tests := []struct { + name string + req *pb.ListRequest + setupQueueConfig func(*qcmock.MockStore) + wantInvalid bool + wantUnrecognized bool + }{ + { + name: "empty queue", + req: &pb.ListRequest{StartTimeMs: 1, EndTimeMs: 2}, + wantInvalid: true, + }, + { + name: "invalid window", + req: &pb.ListRequest{Queue: "q", StartTimeMs: 2, EndTimeMs: 1}, + wantInvalid: true, + }, + { + name: "invalid page size", + req: &pb.ListRequest{Queue: "q", StartTimeMs: 1, EndTimeMs: 2, PageSize: -1}, + wantInvalid: true, + }, + { + name: "unknown status", + req: &pb.ListRequest{Queue: "q", StartTimeMs: 1, EndTimeMs: 2, Statuses: []string{"wat"}}, + wantInvalid: true, + }, + { + name: "unknown sort", + req: &pb.ListRequest{Queue: "q", StartTimeMs: 1, EndTimeMs: 2, Sort: pb.ListSort(99)}, + wantInvalid: true, + }, + { + name: "unrecognized queue", + req: &pb.ListRequest{Queue: "missing", StartTimeMs: 1, EndTimeMs: 2}, + setupQueueConfig: func(qcs *qcmock.MockStore) { + qcs.EXPECT().Get(gomock.Any(), "missing").Return(entity.QueueConfig{}, queueconfig.ErrNotFound) + }, + wantUnrecognized: true, + }, + { + name: "queue store failure", + req: &pb.ListRequest{Queue: "q", StartTimeMs: 1, EndTimeMs: 2}, + setupQueueConfig: func(qcs *qcmock.MockStore) { + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, fmt.Errorf("backend down")) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + qcs := qcmock.NewMockStore(ctrl) + if tt.setupQueueConfig != nil { + tt.setupQueueConfig(qcs) + } + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + + _, err := controller.List(context.Background(), tt.req) + + require.Error(t, err) + assert.Equal(t, tt.wantUnrecognized, IsUnrecognizedQueue(err)) + if tt.wantInvalid { + assert.True(t, IsInvalidRequest(err)) + } + if tt.wantUnrecognized { + assert.True(t, errs.IsUserError(err)) + } + }) + } +} + +func TestList_RejectsMismatchedPageToken(t *testing.T) { + ctrl := gomock.NewController(t) + + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + + token, err := encodeListPageToken(listPageToken{ + Queue: "other", + StartTimeMs: 1, + EndTimeMs: 2, + Sort: storage.RequestSummarySortAdmittedAsc, + }) + require.NoError(t, err) + + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + + _, err = controller.List(context.Background(), &pb.ListRequest{ + Queue: "q", + StartTimeMs: 1, + EndTimeMs: 2, + PageToken: token, + }) + + require.Error(t, err) + assert.True(t, IsInvalidRequest(err)) +} + +func TestList_RejectsMismatchedPageTokenSort(t *testing.T) { + ctrl := gomock.NewController(t) + + qcs := qcmock.NewMockStore(ctrl) + qcs.EXPECT().Get(gomock.Any(), "q").Return(entity.QueueConfig{}, nil) + + token, err := encodeListPageToken(listPageToken{ + Queue: "q", + StartTimeMs: 1, + EndTimeMs: 2, + Sort: storage.RequestSummarySortAdmittedDesc, + }) + require.NoError(t, err) + + summaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + controller := NewListController(zap.NewNop().Sugar(), tally.NoopScope, summaryStore, qcs) + + _, err = controller.List(context.Background(), &pb.ListRequest{ + Queue: "q", + StartTimeMs: 1, + EndTimeMs: 2, + Sort: pb.ListSort_ADMITTED_ASC, + PageToken: token, + }) + + require.Error(t, err) + assert.True(t, IsInvalidRequest(err)) +} diff --git a/submitqueue/gateway/controller/log/BUILD.bazel b/submitqueue/gateway/controller/log/BUILD.bazel index b323d30d..a7b4708c 100644 --- a/submitqueue/gateway/controller/log/BUILD.bazel +++ b/submitqueue/gateway/controller/log/BUILD.bazel @@ -8,6 +8,7 @@ go_library( deps = [ "//platform/consumer", "//platform/metrics", + "//submitqueue/core/request", "//submitqueue/entity", "//submitqueue/extension/storage", "@com_github_uber_go_tally//:tally", diff --git a/submitqueue/gateway/controller/log/log.go b/submitqueue/gateway/controller/log/log.go index d28811c7..c69c7717 100644 --- a/submitqueue/gateway/controller/log/log.go +++ b/submitqueue/gateway/controller/log/log.go @@ -21,6 +21,7 @@ import ( "github.com/uber-go/tally" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" + corerequest "github.com/uber/submitqueue/submitqueue/core/request" "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/storage" "go.uber.org/zap" @@ -87,10 +88,10 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r "attempt", delivery.Attempt(), ) - // Persist request log to storage - if err := c.store.GetRequestLogStore().Insert(ctx, logEntry); err != nil { + // Persist request log and summary to storage. + if err := corerequest.PersistLog(ctx, c.store, logEntry); err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - return fmt.Errorf("failed to insert request log: %w", err) + return fmt.Errorf("failed to persist request log: %w", err) } return nil // Success - message will be acked diff --git a/submitqueue/gateway/controller/log/log_test.go b/submitqueue/gateway/controller/log/log_test.go index 166352bb..de2099c9 100644 --- a/submitqueue/gateway/controller/log/log_test.go +++ b/submitqueue/gateway/controller/log/log_test.go @@ -54,8 +54,15 @@ func TestController_Process(t *testing.T) { setupStore: func(ctrl *gomock.Controller) *storagemock.MockStorage { mockLogStore := storagemock.NewMockRequestLogStore(ctrl) mockLogStore.EXPECT().Insert(gomock.Any(), gomock.Any()).Return(nil) + mockSummaryStore := storagemock.NewMockRequestSummaryStore(ctrl) + mockContextStore := storagemock.NewMockRequestContextStore(ctrl) + mockContextStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(entity.RequestContext{Queue: "test-queue"}, nil) + mockSummaryStore.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(entity.RequestSummary{Version: 1}, nil) + mockSummaryStore.EXPECT().Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil) store := storagemock.NewMockStorage(ctrl) store.EXPECT().GetRequestLogStore().Return(mockLogStore).AnyTimes() + store.EXPECT().GetRequestSummaryStore().Return(mockSummaryStore).AnyTimes() + store.EXPECT().GetRequestContextStore().Return(mockContextStore).AnyTimes() return store }, wantErr: false,