[Chasm] Introduce proto ChasmExecutionInfo for Visibility (#9652)

## What changed?
Introduce proto types for Visibility:
- `ChasmExecutionInfo`
- `ListChasmExecutionsRequest`
- `ListChasmExecutionsResponse`

Changed VisibilityManager API to use these protos.

## Why?
Make VisibilityManager API more accessible for potential external usage.

## How did you test it?
- [x] built
- [ ] run locally and tested manually
- [x] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
This commit is contained in:
Rodrigo Zhou
2026-03-25 14:18:30 -07:00
committed by GitHub
parent 85d9fd6360
commit 9c8ea108b3
24 changed files with 1330 additions and 298 deletions
+43
View File
@@ -0,0 +1,43 @@
// Code generated by protoc-gen-go-helpers. DO NOT EDIT.
package chasm
import (
"google.golang.org/protobuf/proto"
)
// Marshal an object of type VisibilityExecutionInfo to the protobuf v3 wire format
func (val *VisibilityExecutionInfo) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type VisibilityExecutionInfo from the protobuf v3 wire format
func (val *VisibilityExecutionInfo) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *VisibilityExecutionInfo) Size() int {
return proto.Size(val)
}
// Equal returns whether two VisibilityExecutionInfo values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *VisibilityExecutionInfo) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *VisibilityExecutionInfo
switch t := that.(type) {
case *VisibilityExecutionInfo:
that1 = t
case VisibilityExecutionInfo:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
+230
View File
@@ -0,0 +1,230 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// plugins:
// protoc-gen-go
// protoc
// source: temporal/server/api/chasm/v1/message.proto
package chasm
import (
reflect "reflect"
sync "sync"
unsafe "unsafe"
v1 "go.temporal.io/api/common/v1"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
timestamppb "google.golang.org/protobuf/types/known/timestamppb"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type VisibilityExecutionInfo struct {
state protoimpl.MessageState `protogen:"open.v1"`
BusinessId string `protobuf:"bytes,1,opt,name=business_id,json=businessId,proto3" json:"business_id,omitempty"`
RunId string `protobuf:"bytes,2,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"`
StartTime *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=start_time,json=startTime,proto3" json:"start_time,omitempty"`
CloseTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=close_time,json=closeTime,proto3" json:"close_time,omitempty"`
HistoryLength int64 `protobuf:"varint,5,opt,name=history_length,json=historyLength,proto3" json:"history_length,omitempty"`
HistorySizeBytes int64 `protobuf:"varint,6,opt,name=history_size_bytes,json=historySizeBytes,proto3" json:"history_size_bytes,omitempty"`
StateTransitionCount int64 `protobuf:"varint,7,opt,name=state_transition_count,json=stateTransitionCount,proto3" json:"state_transition_count,omitempty"`
ChasmSearchAttributes *v1.SearchAttributes `protobuf:"bytes,8,opt,name=chasm_search_attributes,json=chasmSearchAttributes,proto3" json:"chasm_search_attributes,omitempty"`
CustomSearchAttributes *v1.SearchAttributes `protobuf:"bytes,9,opt,name=custom_search_attributes,json=customSearchAttributes,proto3" json:"custom_search_attributes,omitempty"`
Memo *v1.Memo `protobuf:"bytes,10,opt,name=memo,proto3" json:"memo,omitempty"`
ChasmMemo *v1.Payload `protobuf:"bytes,11,opt,name=chasm_memo,json=chasmMemo,proto3" json:"chasm_memo,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *VisibilityExecutionInfo) Reset() {
*x = VisibilityExecutionInfo{}
mi := &file_temporal_server_api_chasm_v1_message_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *VisibilityExecutionInfo) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*VisibilityExecutionInfo) ProtoMessage() {}
func (x *VisibilityExecutionInfo) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_chasm_v1_message_proto_msgTypes[0]
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 VisibilityExecutionInfo.ProtoReflect.Descriptor instead.
func (*VisibilityExecutionInfo) Descriptor() ([]byte, []int) {
return file_temporal_server_api_chasm_v1_message_proto_rawDescGZIP(), []int{0}
}
func (x *VisibilityExecutionInfo) GetBusinessId() string {
if x != nil {
return x.BusinessId
}
return ""
}
func (x *VisibilityExecutionInfo) GetRunId() string {
if x != nil {
return x.RunId
}
return ""
}
func (x *VisibilityExecutionInfo) GetStartTime() *timestamppb.Timestamp {
if x != nil {
return x.StartTime
}
return nil
}
func (x *VisibilityExecutionInfo) GetCloseTime() *timestamppb.Timestamp {
if x != nil {
return x.CloseTime
}
return nil
}
func (x *VisibilityExecutionInfo) GetHistoryLength() int64 {
if x != nil {
return x.HistoryLength
}
return 0
}
func (x *VisibilityExecutionInfo) GetHistorySizeBytes() int64 {
if x != nil {
return x.HistorySizeBytes
}
return 0
}
func (x *VisibilityExecutionInfo) GetStateTransitionCount() int64 {
if x != nil {
return x.StateTransitionCount
}
return 0
}
func (x *VisibilityExecutionInfo) GetChasmSearchAttributes() *v1.SearchAttributes {
if x != nil {
return x.ChasmSearchAttributes
}
return nil
}
func (x *VisibilityExecutionInfo) GetCustomSearchAttributes() *v1.SearchAttributes {
if x != nil {
return x.CustomSearchAttributes
}
return nil
}
func (x *VisibilityExecutionInfo) GetMemo() *v1.Memo {
if x != nil {
return x.Memo
}
return nil
}
func (x *VisibilityExecutionInfo) GetChasmMemo() *v1.Payload {
if x != nil {
return x.ChasmMemo
}
return nil
}
var File_temporal_server_api_chasm_v1_message_proto protoreflect.FileDescriptor
const file_temporal_server_api_chasm_v1_message_proto_rawDesc = "" +
"\n" +
"*temporal/server/api/chasm/v1/message.proto\x12\x1ctemporal.server.api.chasm.v1\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\"\x8a\x05\n" +
"\x17VisibilityExecutionInfo\x12\x1f\n" +
"\vbusiness_id\x18\x01 \x01(\tR\n" +
"businessId\x12\x15\n" +
"\x06run_id\x18\x02 \x01(\tR\x05runId\x129\n" +
"\n" +
"start_time\x18\x03 \x01(\v2\x1a.google.protobuf.TimestampR\tstartTime\x129\n" +
"\n" +
"close_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\tcloseTime\x12%\n" +
"\x0ehistory_length\x18\x05 \x01(\x03R\rhistoryLength\x12,\n" +
"\x12history_size_bytes\x18\x06 \x01(\x03R\x10historySizeBytes\x124\n" +
"\x16state_transition_count\x18\a \x01(\x03R\x14stateTransitionCount\x12`\n" +
"\x17chasm_search_attributes\x18\b \x01(\v2(.temporal.api.common.v1.SearchAttributesR\x15chasmSearchAttributes\x12b\n" +
"\x18custom_search_attributes\x18\t \x01(\v2(.temporal.api.common.v1.SearchAttributesR\x16customSearchAttributes\x120\n" +
"\x04memo\x18\n" +
" \x01(\v2\x1c.temporal.api.common.v1.MemoR\x04memo\x12>\n" +
"\n" +
"chasm_memo\x18\v \x01(\v2\x1f.temporal.api.common.v1.PayloadR\tchasmMemoB*Z(go.temporal.io/server/api/chasm/v1;chasmb\x06proto3"
var (
file_temporal_server_api_chasm_v1_message_proto_rawDescOnce sync.Once
file_temporal_server_api_chasm_v1_message_proto_rawDescData []byte
)
func file_temporal_server_api_chasm_v1_message_proto_rawDescGZIP() []byte {
file_temporal_server_api_chasm_v1_message_proto_rawDescOnce.Do(func() {
file_temporal_server_api_chasm_v1_message_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_temporal_server_api_chasm_v1_message_proto_rawDesc), len(file_temporal_server_api_chasm_v1_message_proto_rawDesc)))
})
return file_temporal_server_api_chasm_v1_message_proto_rawDescData
}
var file_temporal_server_api_chasm_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_temporal_server_api_chasm_v1_message_proto_goTypes = []any{
(*VisibilityExecutionInfo)(nil), // 0: temporal.server.api.chasm.v1.VisibilityExecutionInfo
(*timestamppb.Timestamp)(nil), // 1: google.protobuf.Timestamp
(*v1.SearchAttributes)(nil), // 2: temporal.api.common.v1.SearchAttributes
(*v1.Memo)(nil), // 3: temporal.api.common.v1.Memo
(*v1.Payload)(nil), // 4: temporal.api.common.v1.Payload
}
var file_temporal_server_api_chasm_v1_message_proto_depIdxs = []int32{
1, // 0: temporal.server.api.chasm.v1.VisibilityExecutionInfo.start_time:type_name -> google.protobuf.Timestamp
1, // 1: temporal.server.api.chasm.v1.VisibilityExecutionInfo.close_time:type_name -> google.protobuf.Timestamp
2, // 2: temporal.server.api.chasm.v1.VisibilityExecutionInfo.chasm_search_attributes:type_name -> temporal.api.common.v1.SearchAttributes
2, // 3: temporal.server.api.chasm.v1.VisibilityExecutionInfo.custom_search_attributes:type_name -> temporal.api.common.v1.SearchAttributes
3, // 4: temporal.server.api.chasm.v1.VisibilityExecutionInfo.memo:type_name -> temporal.api.common.v1.Memo
4, // 5: temporal.server.api.chasm.v1.VisibilityExecutionInfo.chasm_memo:type_name -> temporal.api.common.v1.Payload
6, // [6:6] is the sub-list for method output_type
6, // [6:6] is the sub-list for method input_type
6, // [6:6] is the sub-list for extension type_name
6, // [6:6] is the sub-list for extension extendee
0, // [0:6] is the sub-list for field type_name
}
func init() { file_temporal_server_api_chasm_v1_message_proto_init() }
func file_temporal_server_api_chasm_v1_message_proto_init() {
if File_temporal_server_api_chasm_v1_message_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_chasm_v1_message_proto_rawDesc), len(file_temporal_server_api_chasm_v1_message_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_chasm_v1_message_proto_goTypes,
DependencyIndexes: file_temporal_server_api_chasm_v1_message_proto_depIdxs,
MessageInfos: file_temporal_server_api_chasm_v1_message_proto_msgTypes,
}.Build()
File_temporal_server_api_chasm_v1_message_proto = out.File
file_temporal_server_api_chasm_v1_message_proto_goTypes = nil
file_temporal_server_api_chasm_v1_message_proto_depIdxs = nil
}
@@ -0,0 +1,154 @@
// Code generated by protoc-gen-go-helpers. DO NOT EDIT.
package visibilityservice
import (
"google.golang.org/protobuf/proto"
)
// Marshal an object of type ListChasmExecutionsRequest to the protobuf v3 wire format
func (val *ListChasmExecutionsRequest) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type ListChasmExecutionsRequest from the protobuf v3 wire format
func (val *ListChasmExecutionsRequest) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *ListChasmExecutionsRequest) Size() int {
return proto.Size(val)
}
// Equal returns whether two ListChasmExecutionsRequest values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *ListChasmExecutionsRequest) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *ListChasmExecutionsRequest
switch t := that.(type) {
case *ListChasmExecutionsRequest:
that1 = t
case ListChasmExecutionsRequest:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type ListChasmExecutionsResponse to the protobuf v3 wire format
func (val *ListChasmExecutionsResponse) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type ListChasmExecutionsResponse from the protobuf v3 wire format
func (val *ListChasmExecutionsResponse) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *ListChasmExecutionsResponse) Size() int {
return proto.Size(val)
}
// Equal returns whether two ListChasmExecutionsResponse values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *ListChasmExecutionsResponse) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *ListChasmExecutionsResponse
switch t := that.(type) {
case *ListChasmExecutionsResponse:
that1 = t
case ListChasmExecutionsResponse:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type CountChasmExecutionsRequest to the protobuf v3 wire format
func (val *CountChasmExecutionsRequest) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type CountChasmExecutionsRequest from the protobuf v3 wire format
func (val *CountChasmExecutionsRequest) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *CountChasmExecutionsRequest) Size() int {
return proto.Size(val)
}
// Equal returns whether two CountChasmExecutionsRequest values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *CountChasmExecutionsRequest) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *CountChasmExecutionsRequest
switch t := that.(type) {
case *CountChasmExecutionsRequest:
that1 = t
case CountChasmExecutionsRequest:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
// Marshal an object of type CountChasmExecutionsResponse to the protobuf v3 wire format
func (val *CountChasmExecutionsResponse) Marshal() ([]byte, error) {
return proto.Marshal(val)
}
// Unmarshal an object of type CountChasmExecutionsResponse from the protobuf v3 wire format
func (val *CountChasmExecutionsResponse) Unmarshal(buf []byte) error {
return proto.Unmarshal(buf, val)
}
// Size returns the size of the object, in bytes, once serialized
func (val *CountChasmExecutionsResponse) Size() int {
return proto.Size(val)
}
// Equal returns whether two CountChasmExecutionsResponse values are equivalent by recursively
// comparing the message's fields.
// For more information see the documentation for
// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
func (this *CountChasmExecutionsResponse) Equal(that interface{}) bool {
if that == nil {
return this == nil
}
var that1 *CountChasmExecutionsResponse
switch t := that.(type) {
case *CountChasmExecutionsResponse:
that1 = t
case CountChasmExecutionsResponse:
that1 = &t
default:
return false
}
return proto.Equal(this, that1)
}
@@ -0,0 +1,426 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// plugins:
// protoc-gen-go
// protoc
// source: temporal/server/api/visibilityservice/v1/request_response.proto
package visibilityservice
import (
reflect "reflect"
sync "sync"
unsafe "unsafe"
v11 "go.temporal.io/api/common/v1"
v1 "go.temporal.io/server/api/chasm/v1"
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type ListChasmExecutionsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
// (-- api-linter: core::0141::forbidden-types=disabled --)
ArchetypeId uint32 `protobuf:"varint,1,opt,name=archetype_id,json=archetypeId,proto3" json:"archetype_id,omitempty"`
NamespaceId string `protobuf:"bytes,2,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
Namespace string `protobuf:"bytes,3,opt,name=namespace,proto3" json:"namespace,omitempty"`
Query string `protobuf:"bytes,4,opt,name=query,proto3" json:"query,omitempty"`
// Maximum number of executions per page
PageSize int32 `protobuf:"varint,5,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"`
// Token to continue reading next page of executions.
// Pass in empty slice for first page.
NextPageToken []byte `protobuf:"bytes,6,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ListChasmExecutionsRequest) Reset() {
*x = ListChasmExecutionsRequest{}
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ListChasmExecutionsRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ListChasmExecutionsRequest) ProtoMessage() {}
func (x *ListChasmExecutionsRequest) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[0]
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 ListChasmExecutionsRequest.ProtoReflect.Descriptor instead.
func (*ListChasmExecutionsRequest) Descriptor() ([]byte, []int) {
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP(), []int{0}
}
func (x *ListChasmExecutionsRequest) GetArchetypeId() uint32 {
if x != nil {
return x.ArchetypeId
}
return 0
}
func (x *ListChasmExecutionsRequest) GetNamespaceId() string {
if x != nil {
return x.NamespaceId
}
return ""
}
func (x *ListChasmExecutionsRequest) GetNamespace() string {
if x != nil {
return x.Namespace
}
return ""
}
func (x *ListChasmExecutionsRequest) GetQuery() string {
if x != nil {
return x.Query
}
return ""
}
func (x *ListChasmExecutionsRequest) GetPageSize() int32 {
if x != nil {
return x.PageSize
}
return 0
}
func (x *ListChasmExecutionsRequest) GetNextPageToken() []byte {
if x != nil {
return x.NextPageToken
}
return nil
}
type ListChasmExecutionsResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Executions []*v1.VisibilityExecutionInfo `protobuf:"bytes,1,rep,name=executions,proto3" json:"executions,omitempty"`
NextPageToken []byte `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ListChasmExecutionsResponse) Reset() {
*x = ListChasmExecutionsResponse{}
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[1]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ListChasmExecutionsResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ListChasmExecutionsResponse) ProtoMessage() {}
func (x *ListChasmExecutionsResponse) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[1]
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 ListChasmExecutionsResponse.ProtoReflect.Descriptor instead.
func (*ListChasmExecutionsResponse) Descriptor() ([]byte, []int) {
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP(), []int{1}
}
func (x *ListChasmExecutionsResponse) GetExecutions() []*v1.VisibilityExecutionInfo {
if x != nil {
return x.Executions
}
return nil
}
func (x *ListChasmExecutionsResponse) GetNextPageToken() []byte {
if x != nil {
return x.NextPageToken
}
return nil
}
type CountChasmExecutionsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
// (-- api-linter: core::0141::forbidden-types=disabled --)
ArchetypeId uint32 `protobuf:"varint,1,opt,name=archetype_id,json=archetypeId,proto3" json:"archetype_id,omitempty"`
NamespaceId string `protobuf:"bytes,2,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
Namespace string `protobuf:"bytes,3,opt,name=namespace,proto3" json:"namespace,omitempty"`
Query string `protobuf:"bytes,4,opt,name=query,proto3" json:"query,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CountChasmExecutionsRequest) Reset() {
*x = CountChasmExecutionsRequest{}
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[2]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CountChasmExecutionsRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CountChasmExecutionsRequest) ProtoMessage() {}
func (x *CountChasmExecutionsRequest) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[2]
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 CountChasmExecutionsRequest.ProtoReflect.Descriptor instead.
func (*CountChasmExecutionsRequest) Descriptor() ([]byte, []int) {
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP(), []int{2}
}
func (x *CountChasmExecutionsRequest) GetArchetypeId() uint32 {
if x != nil {
return x.ArchetypeId
}
return 0
}
func (x *CountChasmExecutionsRequest) GetNamespaceId() string {
if x != nil {
return x.NamespaceId
}
return ""
}
func (x *CountChasmExecutionsRequest) GetNamespace() string {
if x != nil {
return x.Namespace
}
return ""
}
func (x *CountChasmExecutionsRequest) GetQuery() string {
if x != nil {
return x.Query
}
return ""
}
type CountChasmExecutionsResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Count int64 `protobuf:"varint,1,opt,name=count,proto3" json:"count,omitempty"`
// `groups` contains the groups if the request is grouping by a field.
// The list might not be complete, and the counts of each group are approximations.
Groups []*CountChasmExecutionsResponse_AggregationGroup `protobuf:"bytes,2,rep,name=groups,proto3" json:"groups,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CountChasmExecutionsResponse) Reset() {
*x = CountChasmExecutionsResponse{}
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[3]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CountChasmExecutionsResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CountChasmExecutionsResponse) ProtoMessage() {}
func (x *CountChasmExecutionsResponse) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[3]
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 CountChasmExecutionsResponse.ProtoReflect.Descriptor instead.
func (*CountChasmExecutionsResponse) Descriptor() ([]byte, []int) {
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP(), []int{3}
}
func (x *CountChasmExecutionsResponse) GetCount() int64 {
if x != nil {
return x.Count
}
return 0
}
func (x *CountChasmExecutionsResponse) GetGroups() []*CountChasmExecutionsResponse_AggregationGroup {
if x != nil {
return x.Groups
}
return nil
}
type CountChasmExecutionsResponse_AggregationGroup struct {
state protoimpl.MessageState `protogen:"open.v1"`
GroupValues []*v11.Payload `protobuf:"bytes,1,rep,name=group_values,json=groupValues,proto3" json:"group_values,omitempty"`
Count int64 `protobuf:"varint,2,opt,name=count,proto3" json:"count,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CountChasmExecutionsResponse_AggregationGroup) Reset() {
*x = CountChasmExecutionsResponse_AggregationGroup{}
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[4]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CountChasmExecutionsResponse_AggregationGroup) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CountChasmExecutionsResponse_AggregationGroup) ProtoMessage() {}
func (x *CountChasmExecutionsResponse_AggregationGroup) ProtoReflect() protoreflect.Message {
mi := &file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes[4]
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 CountChasmExecutionsResponse_AggregationGroup.ProtoReflect.Descriptor instead.
func (*CountChasmExecutionsResponse_AggregationGroup) Descriptor() ([]byte, []int) {
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP(), []int{3, 0}
}
func (x *CountChasmExecutionsResponse_AggregationGroup) GetGroupValues() []*v11.Payload {
if x != nil {
return x.GroupValues
}
return nil
}
func (x *CountChasmExecutionsResponse_AggregationGroup) GetCount() int64 {
if x != nil {
return x.Count
}
return 0
}
var File_temporal_server_api_visibilityservice_v1_request_response_proto protoreflect.FileDescriptor
const file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc = "" +
"\n" +
"?temporal/server/api/visibilityservice/v1/request_response.proto\x12(temporal.server.api.visibilityservice.v1\x1a$temporal/api/common/v1/message.proto\x1a*temporal/server/api/chasm/v1/message.proto\"\xdb\x01\n" +
"\x1aListChasmExecutionsRequest\x12!\n" +
"\farchetype_id\x18\x01 \x01(\rR\varchetypeId\x12!\n" +
"\fnamespace_id\x18\x02 \x01(\tR\vnamespaceId\x12\x1c\n" +
"\tnamespace\x18\x03 \x01(\tR\tnamespace\x12\x14\n" +
"\x05query\x18\x04 \x01(\tR\x05query\x12\x1b\n" +
"\tpage_size\x18\x05 \x01(\x05R\bpageSize\x12&\n" +
"\x0fnext_page_token\x18\x06 \x01(\fR\rnextPageToken\"\x9c\x01\n" +
"\x1bListChasmExecutionsResponse\x12U\n" +
"\n" +
"executions\x18\x01 \x03(\v25.temporal.server.api.chasm.v1.VisibilityExecutionInfoR\n" +
"executions\x12&\n" +
"\x0fnext_page_token\x18\x02 \x01(\fR\rnextPageToken\"\x97\x01\n" +
"\x1bCountChasmExecutionsRequest\x12!\n" +
"\farchetype_id\x18\x01 \x01(\rR\varchetypeId\x12!\n" +
"\fnamespace_id\x18\x02 \x01(\tR\vnamespaceId\x12\x1c\n" +
"\tnamespace\x18\x03 \x01(\tR\tnamespace\x12\x14\n" +
"\x05query\x18\x04 \x01(\tR\x05query\"\x93\x02\n" +
"\x1cCountChasmExecutionsResponse\x12\x14\n" +
"\x05count\x18\x01 \x01(\x03R\x05count\x12o\n" +
"\x06groups\x18\x02 \x03(\v2W.temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse.AggregationGroupR\x06groups\x1al\n" +
"\x10AggregationGroup\x12B\n" +
"\fgroup_values\x18\x01 \x03(\v2\x1f.temporal.api.common.v1.PayloadR\vgroupValues\x12\x14\n" +
"\x05count\x18\x02 \x01(\x03R\[email protected]/server/api/visibilityservice/v1;visibilityserviceb\x06proto3"
var (
file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescOnce sync.Once
file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescData []byte
)
func file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescGZIP() []byte {
file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescOnce.Do(func() {
file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc)))
})
return file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDescData
}
var file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes = make([]protoimpl.MessageInfo, 5)
var file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes = []any{
(*ListChasmExecutionsRequest)(nil), // 0: temporal.server.api.visibilityservice.v1.ListChasmExecutionsRequest
(*ListChasmExecutionsResponse)(nil), // 1: temporal.server.api.visibilityservice.v1.ListChasmExecutionsResponse
(*CountChasmExecutionsRequest)(nil), // 2: temporal.server.api.visibilityservice.v1.CountChasmExecutionsRequest
(*CountChasmExecutionsResponse)(nil), // 3: temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse
(*CountChasmExecutionsResponse_AggregationGroup)(nil), // 4: temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse.AggregationGroup
(*v1.VisibilityExecutionInfo)(nil), // 5: temporal.server.api.chasm.v1.VisibilityExecutionInfo
(*v11.Payload)(nil), // 6: temporal.api.common.v1.Payload
}
var file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs = []int32{
5, // 0: temporal.server.api.visibilityservice.v1.ListChasmExecutionsResponse.executions:type_name -> temporal.server.api.chasm.v1.VisibilityExecutionInfo
4, // 1: temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse.groups:type_name -> temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse.AggregationGroup
6, // 2: temporal.server.api.visibilityservice.v1.CountChasmExecutionsResponse.AggregationGroup.group_values:type_name -> temporal.api.common.v1.Payload
3, // [3:3] is the sub-list for method output_type
3, // [3:3] is the sub-list for method input_type
3, // [3:3] is the sub-list for extension type_name
3, // [3:3] is the sub-list for extension extendee
0, // [0:3] is the sub-list for field type_name
}
func init() { file_temporal_server_api_visibilityservice_v1_request_response_proto_init() }
func file_temporal_server_api_visibilityservice_v1_request_response_proto_init() {
if File_temporal_server_api_visibilityservice_v1_request_response_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc)),
NumEnums: 0,
NumMessages: 5,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes,
DependencyIndexes: file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs,
MessageInfos: file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes,
}.Build()
File_temporal_server_api_visibilityservice_v1_request_response_proto = out.File
file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes = nil
file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs = nil
}
+15 -9
View File
@@ -425,18 +425,24 @@ func NewSearchAttributesMap(values map[string]VisibilityValue) SearchAttributesM
return SearchAttributesMap{values: values}
}
// ToProto encodes the search attributes as proto object.
func (m *SearchAttributesMap) ToProto() *commonpb.SearchAttributes {
if m == nil || m.values == nil {
return nil
// newSearchAttributesMapFromProto creates a new SearchAttributesMap from commonpb.SearchAttributes.
func newSearchAttributesMapFromProto(
searchAttributes *commonpb.SearchAttributes,
) (SearchAttributesMap, error) {
if len(searchAttributes.GetIndexedFields()) == 0 {
return SearchAttributesMap{}, nil
}
res := &commonpb.SearchAttributes{
IndexedFields: make(map[string]*commonpb.Payload, len(m.values)),
result := SearchAttributesMap{
values: make(map[string]VisibilityValue),
}
for k, v := range m.values {
res.IndexedFields[k] = v.MustEncode()
for saName, saPayload := range searchAttributes.IndexedFields {
value, err := visibilityValueFromPayload(saPayload)
if err != nil {
return SearchAttributesMap{}, nil
}
result.values[saName] = value
}
return res
return result, nil
}
// SearchAttributeValue returns the value for a given SearchAttribute with compile-time type safety.
+74 -45
View File
@@ -5,6 +5,7 @@ import (
"time"
"github.com/stretchr/testify/require"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/server/common/searchattribute/sadefs"
)
@@ -82,63 +83,91 @@ func TestSearchAttributesMap_Get(t *testing.T) {
})
}
func TestSearchAttributesMap_ToProto(t *testing.T) {
t.Run("NilReceiver", func(t *testing.T) {
var m *SearchAttributesMap
require.Nil(t, m.ToProto())
func TestNewSearchAttributesMapFromProto(t *testing.T) {
t.Run("NilSearchAttributes", func(t *testing.T) {
m, err := newSearchAttributesMapFromProto(nil)
require.NoError(t, err)
require.Empty(t, m.values)
})
t.Run("NilValues", func(t *testing.T) {
m := &SearchAttributesMap{values: nil}
require.Nil(t, m.ToProto())
t.Run("EmptyIndexedFields", func(t *testing.T) {
m, err := newSearchAttributesMapFromProto(&commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{},
})
require.NoError(t, err)
require.Empty(t, m.values)
})
t.Run("EmptyMap", func(t *testing.T) {
m := NewSearchAttributesMap(map[string]VisibilityValue{})
proto := m.ToProto()
require.NotNil(t, proto)
require.Empty(t, proto.IndexedFields)
})
t.Run("AllTypes", func(t *testing.T) {
now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
values := map[string]VisibilityValue{
"bool_key": VisibilityValueBool(true),
"int_key": VisibilityValueInt64(42),
"double_key": VisibilityValueFloat64(3.14),
"keyword_key": VisibilityValueKeyword("active"),
"datetime_key": VisibilityValueTime(now),
"keyword_list_key": VisibilityValueStringSlice([]string{"a", "b"}),
t.Run("SingleBoolValue", func(t *testing.T) {
sa := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
"completed": sadefs.MustEncodeValue(true, enumspb.INDEXED_VALUE_TYPE_BOOL),
},
}
m := NewSearchAttributesMap(values)
proto := m.ToProto()
require.NotNil(t, proto)
require.Len(t, proto.IndexedFields, 6)
// Verify each field was encoded correctly by decoding and comparing.
decoded, err := sadefs.DecodeValue(proto.IndexedFields["bool_key"], enumspb.INDEXED_VALUE_TYPE_BOOL, false)
m, err := newSearchAttributesMapFromProto(sa)
require.NoError(t, err)
require.Equal(t, true, decoded)
decoded, err = sadefs.DecodeValue(proto.IndexedFields["int_key"], enumspb.INDEXED_VALUE_TYPE_INT, false)
require.NoError(t, err)
require.Equal(t, int64(42), decoded)
boolAttr := NewSearchAttributeBool("completed", SearchAttributeFieldBool01)
val, ok := SearchAttributeValue(m, boolAttr)
require.True(t, ok)
require.True(t, val)
})
decoded, err = sadefs.DecodeValue(proto.IndexedFields["double_key"], enumspb.INDEXED_VALUE_TYPE_DOUBLE, false)
t.Run("MultipleValueTypes", func(t *testing.T) {
now := time.Now().UTC().Truncate(time.Millisecond)
sa := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
"completed": sadefs.MustEncodeValue(true, enumspb.INDEXED_VALUE_TYPE_BOOL),
"count": sadefs.MustEncodeValue(int64(42), enumspb.INDEXED_VALUE_TYPE_INT),
"score": sadefs.MustEncodeValue(3.14, enumspb.INDEXED_VALUE_TYPE_DOUBLE),
"status": sadefs.MustEncodeValue("active", enumspb.INDEXED_VALUE_TYPE_KEYWORD),
"timestamp": sadefs.MustEncodeValue(now, enumspb.INDEXED_VALUE_TYPE_DATETIME),
"tags": sadefs.MustEncodeValue([]string{"tag1", "tag2"}, enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST),
},
}
m, err := newSearchAttributesMapFromProto(sa)
require.NoError(t, err)
require.InDelta(t, 3.14, decoded, 0.0001)
decoded, err = sadefs.DecodeValue(proto.IndexedFields["keyword_key"], enumspb.INDEXED_VALUE_TYPE_KEYWORD, false)
require.NoError(t, err)
require.Equal(t, "active", decoded)
boolAttr := NewSearchAttributeBool("completed", SearchAttributeFieldBool01)
boolVal, ok := SearchAttributeValue(m, boolAttr)
require.True(t, ok)
require.True(t, boolVal)
decoded, err = sadefs.DecodeValue(proto.IndexedFields["datetime_key"], enumspb.INDEXED_VALUE_TYPE_DATETIME, false)
require.NoError(t, err)
require.True(t, now.Equal(decoded.(time.Time)))
intAttr := NewSearchAttributeInt("count", SearchAttributeFieldInt01)
intVal, ok := SearchAttributeValue(m, intAttr)
require.True(t, ok)
require.Equal(t, int64(42), intVal)
decoded, err = sadefs.DecodeValue(proto.IndexedFields["keyword_list_key"], enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, false)
doubleAttr := NewSearchAttributeDouble("score", SearchAttributeFieldDouble01)
doubleVal, ok := SearchAttributeValue(m, doubleAttr)
require.True(t, ok)
require.InDelta(t, 3.14, doubleVal, 0.0001)
keywordAttr := NewSearchAttributeKeyword("status", SearchAttributeFieldKeyword01)
keywordVal, ok := SearchAttributeValue(m, keywordAttr)
require.True(t, ok)
require.Equal(t, "active", keywordVal)
datetimeAttr := NewSearchAttributeDateTime("timestamp", SearchAttributeFieldDateTime01)
timeVal, ok := SearchAttributeValue(m, datetimeAttr)
require.True(t, ok)
require.True(t, now.Equal(timeVal))
keywordListAttr := NewSearchAttributeKeywordList("tags", SearchAttributeFieldKeywordList01)
listVal, ok := SearchAttributeValue(m, keywordListAttr)
require.True(t, ok)
require.Equal(t, []string{"tag1", "tag2"}, listVal)
})
t.Run("InvalidPayload", func(t *testing.T) {
sa := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
"bad": {Data: []byte("not valid")},
},
}
m, err := newSearchAttributesMapFromProto(sa)
// Current implementation returns nil error on decode failure
require.NoError(t, err)
require.Equal(t, []string{"a", "b"}, decoded)
require.Empty(t, m.values)
})
}
+32 -12
View File
@@ -9,6 +9,7 @@ import (
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/payload"
"google.golang.org/protobuf/proto"
)
@@ -18,13 +19,13 @@ type VisibilityManager interface {
context.Context,
reflect.Type,
*ListExecutionsRequest,
) (*ListExecutionsResponse[*commonpb.Payload], error)
) (*visibilityservice.ListChasmExecutionsResponse, error)
CountExecutions(
context.Context,
reflect.Type,
*CountExecutionsRequest,
) (*CountExecutionsResponse, error)
) (*visibilityservice.CountChasmExecutionsResponse, error)
}
type ExecutionInfo[M proto.Message] struct {
@@ -89,28 +90,32 @@ func ListExecutions[C Component, M proto.Message](
return nil, err
}
// Convert response, unmarshaling ChasmMemo to type M
// Convert response: decode ChasmSearchAttributes and ChasmMemo to type M
executions := make([]*ExecutionInfo[M], len(response.Executions))
for i, execution := range response.Executions {
chasmSAs, err := newSearchAttributesMapFromProto(execution.ChasmSearchAttributes)
if err != nil {
return nil, err
}
chasmMemoInterface := reflect.New(reflect.TypeFor[M]().Elem()).Interface()
chasmMemo, ok := chasmMemoInterface.(M)
if !ok {
return nil, serviceerror.NewInternalf("failed to cast chasm memo to type %s", reflect.TypeFor[M]().String())
}
err := payload.Decode(execution.ChasmMemo, chasmMemo)
if err != nil {
if err := payload.Decode(execution.ChasmMemo, chasmMemo); err != nil {
return nil, serviceerror.NewInternalf("failed to decode chasm memo: %v", err)
}
executions[i] = &ExecutionInfo[M]{
BusinessID: execution.BusinessID,
RunID: execution.RunID,
StartTime: execution.StartTime,
CloseTime: execution.CloseTime,
BusinessID: execution.BusinessId,
RunID: execution.RunId,
StartTime: execution.StartTime.AsTime(),
CloseTime: execution.CloseTime.AsTime(),
HistoryLength: execution.HistoryLength,
HistorySizeBytes: execution.HistorySizeBytes,
StateTransitionCount: execution.StateTransitionCount,
ChasmSearchAttributes: execution.ChasmSearchAttributes,
CustomSearchAttributes: execution.CustomSearchAttributes,
ChasmSearchAttributes: chasmSAs,
CustomSearchAttributes: execution.CustomSearchAttributes.GetIndexedFields(),
Memo: execution.Memo,
ChasmMemo: chasmMemo,
}
@@ -135,7 +140,22 @@ func CountExecutions[C Component](
request *CountExecutionsRequest,
) (*CountExecutionsResponse, error) {
archetypeType := reflect.TypeFor[C]()
return visibilityManagerFromContext(ctx).CountExecutions(ctx, archetypeType, request)
visResponse, err := visibilityManagerFromContext(ctx).CountExecutions(ctx, archetypeType, request)
if err != nil {
return nil, err
}
response := &CountExecutionsResponse{
Count: visResponse.Count,
Groups: make([]Group, len(visResponse.Groups)),
}
for k, group := range visResponse.Groups {
response.Groups[k] = Group{
Values: group.GroupValues,
Count: group.Count,
}
}
return response, nil
}
type visibilityManagerCtxKeyType string
+5 -5
View File
@@ -13,7 +13,7 @@ import (
context "context"
reflect "reflect"
common "go.temporal.io/api/common/v1"
visibilityservice "go.temporal.io/server/api/visibilityservice/v1"
gomock "go.uber.org/mock/gomock"
)
@@ -42,10 +42,10 @@ func (m *MockVisibilityManager) EXPECT() *MockVisibilityManagerMockRecorder {
}
// CountExecutions mocks base method.
func (m *MockVisibilityManager) CountExecutions(arg0 context.Context, arg1 reflect.Type, arg2 *CountExecutionsRequest) (*CountExecutionsResponse, error) {
func (m *MockVisibilityManager) CountExecutions(arg0 context.Context, arg1 reflect.Type, arg2 *CountExecutionsRequest) (*visibilityservice.CountChasmExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CountExecutions", arg0, arg1, arg2)
ret0, _ := ret[0].(*CountExecutionsResponse)
ret0, _ := ret[0].(*visibilityservice.CountChasmExecutionsResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
@@ -57,10 +57,10 @@ func (mr *MockVisibilityManagerMockRecorder) CountExecutions(arg0, arg1, arg2 an
}
// ListExecutions mocks base method.
func (m *MockVisibilityManager) ListExecutions(arg0 context.Context, arg1 reflect.Type, arg2 *ListExecutionsRequest) (*ListExecutionsResponse[*common.Payload], error) {
func (m *MockVisibilityManager) ListExecutions(arg0 context.Context, arg1 reflect.Type, arg2 *ListExecutionsRequest) (*visibilityservice.ListChasmExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ListExecutions", arg0, arg1, arg2)
ret0, _ := ret[0].(*ListExecutionsResponse[*common.Payload])
ret0, _ := ret[0].(*visibilityservice.ListChasmExecutionsResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
+31
View File
@@ -1,6 +1,7 @@
package chasm
import (
"fmt"
"slices"
"time"
@@ -132,3 +133,33 @@ func isVisibilityValueEqual(v1, v2 VisibilityValue) bool {
}
return v1.Equal(v2)
}
// visibilityValueFromPayload decoded payload based on type set in its metadata.
func visibilityValueFromPayload(payload *commonpb.Payload) (VisibilityValue, error) {
value, err := sadefs.DecodeValue(payload, enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, false)
if err != nil {
return nil, err
}
switch val := value.(type) {
case int64:
return VisibilityValueInt64(val), nil
case float64:
return VisibilityValueFloat64(val), nil
case bool:
return VisibilityValueBool(val), nil
case time.Time:
return VisibilityValueTime(val), nil
case string:
// Try to parse as datetime first
if parsedTime, err := time.Parse(time.RFC3339, val); err == nil {
return VisibilityValueTime(parsedTime), nil
}
return VisibilityValueKeyword(val), nil
case []string:
return VisibilityValueStringSlice(val), nil
default:
// this should never happen given that DecodeValue did not return an error
return nil, fmt.Errorf("unexpected search attribute value type %T", value)
}
}
@@ -4,8 +4,8 @@ import (
"context"
"reflect"
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence/visibility/manager"
@@ -44,7 +44,7 @@ func (e *ChasmVisibilityManager) ListExecutions(
ctx context.Context,
archetypeType reflect.Type,
request *chasm.ListExecutionsRequest,
) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
) (*visibilityservice.ListChasmExecutionsResponse, error) {
archetypeID, ok := e.registry.ArchetypeIDOf(archetypeType)
if !ok {
return nil, serviceerror.NewInternal("unknown chasm component type: " + archetypeType.String())
@@ -55,11 +55,11 @@ func (e *ChasmVisibilityManager) ListExecutions(
return nil, err
}
visReq := &manager.ListChasmExecutionsRequest{
ArchetypeID: archetypeID,
NamespaceID: namespaceID,
Namespace: namespace.Name(request.NamespaceName),
PageSize: request.PageSize,
visReq := &visibilityservice.ListChasmExecutionsRequest{
ArchetypeId: archetypeID,
NamespaceId: namespaceID.String(),
Namespace: request.NamespaceName,
PageSize: int32(request.PageSize),
NextPageToken: request.NextPageToken,
Query: request.Query,
}
@@ -72,7 +72,7 @@ func (e *ChasmVisibilityManager) CountExecutions(
ctx context.Context,
archetypeType reflect.Type,
request *chasm.CountExecutionsRequest,
) (*chasm.CountExecutionsResponse, error) {
) (*visibilityservice.CountChasmExecutionsResponse, error) {
archetypeID, ok := e.registry.ArchetypeIDOf(archetypeType)
if !ok {
return nil, serviceerror.NewInternal("unknown chasm component type: " + archetypeType.String())
@@ -83,10 +83,10 @@ func (e *ChasmVisibilityManager) CountExecutions(
return nil, err
}
visReq := &manager.CountChasmExecutionsRequest{
ArchetypeID: archetypeID,
NamespaceID: namespaceID,
Namespace: namespace.Name(request.NamespaceName),
visReq := &visibilityservice.CountChasmExecutionsRequest{
ArchetypeId: archetypeID,
NamespaceId: namespaceID.String(),
Namespace: request.NamespaceName,
Query: request.Query,
}
@@ -9,14 +9,19 @@ import (
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
chasmspb "go.temporal.io/server/api/chasm/v1"
persistencespb "go.temporal.io/server/api/persistence/v1"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/payload"
"go.temporal.io/server/common/persistence/visibility/manager"
"go.temporal.io/server/common/searchattribute/sadefs"
"go.uber.org/mock/gomock"
"google.golang.org/protobuf/proto"
"google.golang.org/protobuf/types/known/timestamppb"
)
type (
@@ -143,33 +148,37 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_Success() {
s.True(ok)
// Create chasm search attributes map
chasmSearchAttributes := chasm.NewSearchAttributesMap(map[string]chasm.VisibilityValue{
testChasmSA1Name: chasm.VisibilityValueInt64(123),
testChasmSA2Name: chasm.VisibilityValueKeyword("test-value"),
})
chasmSearchAttributes := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
testChasmSA1Name: sadefs.MustEncodeValue(int64(123), enumspb.INDEXED_VALUE_TYPE_INT),
testChasmSA2Name: sadefs.MustEncodeValue("test-value", enumspb.INDEXED_VALUE_TYPE_KEYWORD),
},
}
// Create custom search attributes map
customSearchAttributes := map[string]*commonpb.Payload{
testCustomSAName: customSAPayload,
customSearchAttributes := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
testCustomSAName: customSAPayload,
},
}
// Setup visibility manager mock
expectedRequest := &manager.ListChasmExecutionsRequest{
ArchetypeID: archetypeID,
NamespaceID: testChasmNamespaceID,
Namespace: testChasmNamespace,
PageSize: pageSize,
expectedRequest := &visibilityservice.ListChasmExecutionsRequest{
ArchetypeId: archetypeID,
NamespaceId: testChasmNamespaceID.String(),
Namespace: testChasmNamespace.String(),
PageSize: int32(pageSize),
NextPageToken: pageToken,
Query: query,
}
expectedResponse := &chasm.ListExecutionsResponse[*commonpb.Payload]{
Executions: []*chasm.ExecutionInfo[*commonpb.Payload]{
expectedResponse := &visibilityservice.ListChasmExecutionsResponse{
Executions: []*chasmspb.VisibilityExecutionInfo{
{
BusinessID: testBusinessID,
RunID: testRunID,
StartTime: testComponentStartTime,
CloseTime: testComponentCloseTime,
BusinessId: testBusinessID,
RunId: testRunID,
StartTime: timestamppb.New(testComponentStartTime),
CloseTime: timestamppb.New(testComponentCloseTime),
HistoryLength: 100,
HistorySizeBytes: 5000,
StateTransitionCount: 42,
@@ -184,9 +193,12 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_Success() {
s.visibilityManager.EXPECT().
ListChasmExecutions(ctx, gomock.Any()).
DoAndReturn(func(_ context.Context, req *manager.ListChasmExecutionsRequest) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
s.Equal(expectedRequest.ArchetypeID, req.ArchetypeID)
s.Equal(expectedRequest.NamespaceID, req.NamespaceID)
DoAndReturn(func(
_ context.Context,
req *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
s.Equal(expectedRequest.ArchetypeId, req.ArchetypeId)
s.Equal(expectedRequest.NamespaceId, req.NamespaceId)
s.Equal(expectedRequest.Namespace, req.Namespace)
s.Equal(expectedRequest.PageSize, req.PageSize)
s.Equal(expectedRequest.NextPageToken, req.NextPageToken)
@@ -215,10 +227,10 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_Success() {
s.Equal([]byte("next-token"), response.NextPageToken)
execution := response.Executions[0]
s.Equal(testBusinessID, execution.BusinessID)
s.Equal(testRunID, execution.RunID)
s.Equal(testComponentStartTime, execution.StartTime)
s.Equal(testComponentCloseTime, execution.CloseTime)
s.Equal(testBusinessID, execution.BusinessId)
s.Equal(testRunID, execution.RunId)
s.Equal(testComponentStartTime, execution.StartTime.AsTime())
s.Equal(testComponentCloseTime, execution.CloseTime.AsTime())
s.Equal(int64(100), execution.HistoryLength)
s.Equal(int64(5000), execution.HistorySizeBytes)
s.Equal(int64(42), execution.StateTransitionCount)
@@ -250,22 +262,25 @@ func (s *ChasmVisibilityManagerSuite) TestCountExecutions_Success() {
expectedCount := int64(42)
// Setup visibility manager mock
expectedRequest := &manager.CountChasmExecutionsRequest{
ArchetypeID: archetypeID,
NamespaceID: testChasmNamespaceID,
Namespace: testChasmNamespace,
expectedRequest := &visibilityservice.CountChasmExecutionsRequest{
ArchetypeId: archetypeID,
NamespaceId: testChasmNamespaceID.String(),
Namespace: testChasmNamespace.String(),
Query: query,
}
expectedResponse := &chasm.CountExecutionsResponse{
expectedResponse := &visibilityservice.CountChasmExecutionsResponse{
Count: expectedCount,
}
s.visibilityManager.EXPECT().
CountChasmExecutions(ctx, gomock.Any()).
DoAndReturn(func(_ context.Context, req *manager.CountChasmExecutionsRequest) (*chasm.CountExecutionsResponse, error) {
s.Equal(expectedRequest.ArchetypeID, req.ArchetypeID)
s.Equal(expectedRequest.NamespaceID, req.NamespaceID)
DoAndReturn(func(
_ context.Context,
req *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
s.Equal(expectedRequest.ArchetypeId, req.ArchetypeId)
s.Equal(expectedRequest.NamespaceId, req.NamespaceId)
s.Equal(expectedRequest.Namespace, req.Namespace)
s.Equal(expectedRequest.Query, req.Query)
return expectedResponse, nil
@@ -343,8 +358,11 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_VisibilityManagerError(
// Setup visibility manager mock to return an error
s.visibilityManager.EXPECT().
ListChasmExecutions(ctx, gomock.Any()).
DoAndReturn(func(_ context.Context, req *manager.ListChasmExecutionsRequest) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
s.Equal(archetypeID, req.ArchetypeID)
DoAndReturn(func(
_ context.Context,
req *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
s.Equal(archetypeID, req.ArchetypeId)
return nil, errTestVisibilityError
})
@@ -376,8 +394,11 @@ func (s *ChasmVisibilityManagerSuite) TestCountExecutions_VisibilityManagerError
// Setup visibility manager mock to return an error
s.visibilityManager.EXPECT().
CountChasmExecutions(ctx, gomock.Any()).
DoAndReturn(func(_ context.Context, req *manager.CountChasmExecutionsRequest) (*chasm.CountExecutionsResponse, error) {
s.Equal(archetypeID, req.ArchetypeID)
DoAndReturn(func(
_ context.Context,
req *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
s.Equal(archetypeID, req.ArchetypeId)
return nil, errTestVisibilityError
})
@@ -442,17 +463,19 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_WithTaskQueueSearchAttr
s.True(ok)
expectedTaskQueue := "my-task-queue"
chasmSearchAttributes := chasm.NewSearchAttributesMap(map[string]chasm.VisibilityValue{
"TaskQueue": chasm.VisibilityValueKeyword(expectedTaskQueue),
})
chasmSearchAttributes := &commonpb.SearchAttributes{
IndexedFields: map[string]*commonpb.Payload{
sadefs.TaskQueue: sadefs.MustEncodeValue(expectedTaskQueue, enumspb.INDEXED_VALUE_TYPE_KEYWORD),
},
}
expectedResponse := &chasm.ListExecutionsResponse[*commonpb.Payload]{
Executions: []*chasm.ExecutionInfo[*commonpb.Payload]{
expectedResponse := &visibilityservice.ListChasmExecutionsResponse{
Executions: []*chasmspb.VisibilityExecutionInfo{
{
BusinessID: testBusinessID,
RunID: testRunID,
StartTime: testComponentStartTime,
CloseTime: testComponentCloseTime,
BusinessId: testBusinessID,
RunId: testRunID,
StartTime: timestamppb.New(testComponentStartTime),
CloseTime: timestamppb.New(testComponentCloseTime),
ChasmSearchAttributes: chasmSearchAttributes,
},
},
@@ -461,8 +484,11 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_WithTaskQueueSearchAttr
visibilityMgr.EXPECT().
ListChasmExecutions(ctx, gomock.Any()).
DoAndReturn(func(_ context.Context, req *manager.ListChasmExecutionsRequest) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
s.Equal(archetypeID, req.ArchetypeID)
DoAndReturn(func(
_ context.Context,
req *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
s.Equal(archetypeID, req.ArchetypeId)
return expectedResponse, nil
})
@@ -485,8 +511,13 @@ func (s *ChasmVisibilityManagerSuite) TestListExecutions_WithTaskQueueSearchAttr
execution := response.Executions[0]
s.NotNil(execution.ChasmSearchAttributes)
// Verify TaskQueue is in the CHASM search attributes using GetValue with the preallocated search attribute
taskQueueVal, ok := chasm.SearchAttributeValue(execution.ChasmSearchAttributes, chasm.SearchAttributeTaskQueue)
s.True(ok)
// Verify TaskQueue is in the CHASM search attributes
s.Contains(execution.ChasmSearchAttributes.GetIndexedFields(), sadefs.TaskQueue)
taskQueueVal, err := sadefs.DecodeValue(
execution.ChasmSearchAttributes.IndexedFields[sadefs.TaskQueue],
enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED,
false,
)
s.NoError(err)
s.Equal(expectedTaskQueue, taskQueueVal)
}
@@ -11,7 +11,7 @@ import (
enumspb "go.temporal.io/api/enums/v1"
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/api/workflowservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence"
)
@@ -34,11 +34,18 @@ type (
// Read APIs.
ListWorkflowExecutions(ctx context.Context, request *ListWorkflowExecutionsRequestV2) (*ListWorkflowExecutionsResponse, error)
ListChasmExecutions(ctx context.Context, request *ListChasmExecutionsRequest) (*chasm.ListExecutionsResponse[*commonpb.Payload], error)
CountWorkflowExecutions(ctx context.Context, request *CountWorkflowExecutionsRequest) (*CountWorkflowExecutionsResponse, error)
CountChasmExecutions(ctx context.Context, request *CountChasmExecutionsRequest) (*chasm.CountExecutionsResponse, error)
GetWorkflowExecution(ctx context.Context, request *GetWorkflowExecutionRequest) (*GetWorkflowExecutionResponse, error)
ListChasmExecutions(
ctx context.Context,
request *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error)
CountChasmExecutions(
ctx context.Context,
request *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error)
// Admin APIs
AddSearchAttributes(ctx context.Context, request *AddSearchAttributesRequest) error
}
@@ -113,24 +120,6 @@ type (
NextPageToken []byte
}
ListChasmExecutionsRequest struct {
ArchetypeID chasm.ArchetypeID
NamespaceID namespace.ID
Namespace namespace.Name
PageSize int // Maximum number of workflow executions per page
Query string
// Token to continue reading next page of workflow executions.
// Pass in empty slice for first page.
NextPageToken []byte
}
CountChasmExecutionsRequest struct {
ArchetypeID chasm.ArchetypeID
NamespaceID namespace.ID
Namespace namespace.Name
Query string
}
// CountWorkflowExecutionsRequest is request from CountWorkflowExecutions
CountWorkflowExecutionsRequest struct {
NamespaceID namespace.ID
@@ -13,8 +13,7 @@ import (
context "context"
reflect "reflect"
common "go.temporal.io/api/common/v1"
chasm "go.temporal.io/server/chasm"
visibilityservice "go.temporal.io/server/api/visibilityservice/v1"
namespace "go.temporal.io/server/common/namespace"
gomock "go.uber.org/mock/gomock"
)
@@ -70,10 +69,10 @@ func (mr *MockVisibilityManagerMockRecorder) Close() *gomock.Call {
}
// CountChasmExecutions mocks base method.
func (m *MockVisibilityManager) CountChasmExecutions(ctx context.Context, request *CountChasmExecutionsRequest) (*chasm.CountExecutionsResponse, error) {
func (m *MockVisibilityManager) CountChasmExecutions(ctx context.Context, request *visibilityservice.CountChasmExecutionsRequest) (*visibilityservice.CountChasmExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CountChasmExecutions", ctx, request)
ret0, _ := ret[0].(*chasm.CountExecutionsResponse)
ret0, _ := ret[0].(*visibilityservice.CountChasmExecutionsResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
@@ -185,10 +184,10 @@ func (mr *MockVisibilityManagerMockRecorder) HasStoreName(stName any) *gomock.Ca
}
// ListChasmExecutions mocks base method.
func (m *MockVisibilityManager) ListChasmExecutions(ctx context.Context, request *ListChasmExecutionsRequest) (*chasm.ListExecutionsResponse[*common.Payload], error) {
func (m *MockVisibilityManager) ListChasmExecutions(ctx context.Context, request *visibilityservice.ListChasmExecutionsRequest) (*visibilityservice.ListChasmExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ListChasmExecutions", ctx, request)
ret0, _ := ret[0].(*chasm.ListExecutionsResponse[*common.Payload])
ret0, _ := ret[0].(*visibilityservice.ListChasmExecutionsResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
}
@@ -17,6 +17,7 @@ import (
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log"
@@ -390,11 +391,11 @@ func (s *VisibilityStore) ListWorkflowExecutions(
func (s *VisibilityStore) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
request *visibilityservice.ListChasmExecutionsRequest,
) (*store.InternalListExecutionsResponse, error) {
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeID)
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeId)
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeID))
return nil, serviceerror.NewInvalidArgumentf("unknown archetype ID: %d", request.ArchetypeId)
}
chasmMapper := rc.SearchAttributesMapper()
@@ -408,28 +409,45 @@ func (s *VisibilityStore) ListChasmExecutions(
return nil, ConvertElasticsearchClientError("ListChasmExecutions failed", err)
}
return s.GetListWorkflowExecutionsResponse(searchResult, request.Namespace, request.PageSize, chasmMapper)
return s.GetListWorkflowExecutionsResponse(
searchResult,
namespace.Name(request.Namespace),
int(request.PageSize),
chasmMapper,
)
}
func (s *VisibilityStore) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
request *visibilityservice.CountChasmExecutionsRequest,
) (*store.InternalCountExecutionsResponse, error) {
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeID)
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeId)
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeID))
return nil, serviceerror.NewInvalidArgumentf("unknown archetype ID: %d", request.ArchetypeId)
}
mapper := rc.SearchAttributesMapper()
var queryParams *esQueryParams
var err error
if s.enableUnifiedQueryConverter() {
queryParams, err = s.convertQuery(request.Namespace, request.NamespaceID, request.Query, mapper, request.ArchetypeID)
queryParams, err = s.convertQuery(
namespace.Name(request.Namespace),
namespace.ID(request.NamespaceId),
request.Query,
mapper,
request.ArchetypeId,
)
if err != nil {
return nil, err
}
} else {
queryParamsLegacy, err := s.convertQueryLegacy(request.Namespace, request.NamespaceID, request.Query, mapper, request.ArchetypeID)
queryParamsLegacy, err := s.convertQueryLegacy(
namespace.Name(request.Namespace),
namespace.ID(request.NamespaceId),
request.Query,
mapper,
request.ArchetypeId,
)
if err != nil {
return nil, err
}
@@ -572,18 +590,18 @@ func (s *VisibilityStore) BuildSearchParametersV2(
}
func (s *VisibilityStore) BuildChasmSearchParameters(
request *manager.ListChasmExecutionsRequest,
request *visibilityservice.ListChasmExecutionsRequest,
getFieldSorter func([]elastic.Sorter) ([]elastic.Sorter, error),
chasmMapper *chasm.VisibilitySearchAttributesMapper,
) (*client.SearchParameters, error) {
return s.buildSearchParametersInternal(&searchParametersInternal{
NamespaceName: request.Namespace,
NamespaceID: request.NamespaceID,
NamespaceName: namespace.Name(request.Namespace),
NamespaceID: namespace.ID(request.NamespaceId),
Query: request.Query,
PageSize: request.PageSize,
PageSize: int(request.PageSize),
NextPageToken: request.NextPageToken,
ChasmMapper: chasmMapper,
ArchetypeID: request.ArchetypeID,
ArchetypeID: request.ArchetypeId,
})
}
@@ -17,6 +17,7 @@ import (
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/api/temporalproto"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/debug"
"go.temporal.io/server/common/dynamicconfig"
@@ -2119,10 +2120,10 @@ func (s *ESVisibilitySuite) TestBuildSearchParametersV2_ChasmMapper() {
newBoolQuery().Filter(filterQuery).MustNot(namespaceDivisionExists),
)
p, err := s.visibilityStore.BuildChasmSearchParameters(
&manager.ListChasmExecutionsRequest{
NamespaceID: testNamespaceID,
Namespace: testNamespace,
PageSize: testPageSize,
&visibilityservice.ListChasmExecutionsRequest{
NamespaceId: testNamespaceID.String(),
Namespace: testNamespace.String(),
PageSize: int32(testPageSize),
Query: request.Query,
},
s.visibilityStore.GetListFieldSorter,
@@ -2146,10 +2147,10 @@ func (s *ESVisibilitySuite) TestBuildSearchParametersV2_ChasmMapper() {
s.mockMetricsHandler.EXPECT().WithTags(metrics.NamespaceTag(request.Namespace.String())).Return(s.mockMetricsHandler).AnyTimes()
s.mockMetricsHandler.EXPECT().Counter(metrics.ElasticsearchCustomOrderByClauseCount.Name()).Return(metrics.NoopCounterMetricFunc)
p, err = s.visibilityStore.BuildChasmSearchParameters(
&manager.ListChasmExecutionsRequest{
NamespaceID: testNamespaceID,
Namespace: testNamespace,
PageSize: testPageSize,
&visibilityservice.ListChasmExecutionsRequest{
NamespaceId: testNamespaceID.String(),
Namespace: testNamespace.String(),
PageSize: int32(testPageSize),
Query: request.Query,
},
s.visibilityStore.GetListFieldSorter,
@@ -10,6 +10,7 @@ import (
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/config"
"go.temporal.io/server/common/dynamicconfig"
@@ -204,22 +205,22 @@ func (s *VisibilityStore) ListWorkflowExecutions(
func (s *VisibilityStore) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
request *visibilityservice.ListChasmExecutionsRequest,
) (*store.InternalListExecutionsResponse, error) {
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeID)
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeId)
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeID))
return nil, serviceerror.NewInvalidArgumentf("unknown archetype ID: %d", request.ArchetypeId)
}
mapper := rc.SearchAttributesMapper()
requestInternal := &listExecutionsRequestInternal{
NamespaceID: request.NamespaceID,
Namespace: request.Namespace,
NamespaceID: namespace.ID(request.NamespaceId),
Namespace: namespace.Name(request.Namespace),
Query: request.Query,
PageSize: request.PageSize,
PageSize: int(request.PageSize),
NextPageToken: request.NextPageToken,
ChasmMapper: mapper,
ArchetypeID: request.ArchetypeID,
ArchetypeID: request.ArchetypeId,
}
if s.enableUnifiedQueryConverter() {
@@ -231,11 +232,11 @@ func (s *VisibilityStore) ListChasmExecutions(
func (s *VisibilityStore) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
request *visibilityservice.CountChasmExecutionsRequest,
) (*store.InternalCountExecutionsResponse, error) {
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeID)
rc, ok := s.chasmRegistry.ComponentByID(request.ArchetypeId)
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeID))
return nil, serviceerror.NewInvalidArgumentf("unknown archetype ID: %d", request.ArchetypeId)
}
mapper := rc.SearchAttributesMapper()
@@ -247,7 +248,7 @@ func (s *VisibilityStore) CountChasmExecutions(
func (s *VisibilityStore) countChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
request *visibilityservice.CountChasmExecutionsRequest,
mapper *chasm.VisibilitySearchAttributesMapper,
) (*store.InternalCountExecutionsResponse, error) {
sqlQC, err := NewSQLQueryConverter(s.GetName())
@@ -260,20 +261,20 @@ func (s *VisibilityStore) countChasmExecutions(
return nil, err
}
saMapper, err := s.searchAttributesMapperProvider.GetMapper(request.Namespace)
saMapper, err := s.searchAttributesMapperProvider.GetMapper(namespace.Name(request.Namespace))
if err != nil {
return nil, err
}
queryParams, err := buildQueryParams(
request.NamespaceID,
request.Namespace,
namespace.ID(request.NamespaceId),
namespace.Name(request.Namespace),
request.Query,
sqlQC,
saTypeMap,
saMapper,
mapper,
request.ArchetypeID,
request.ArchetypeId,
)
if err != nil {
var converterErr *query.ConverterError
@@ -300,7 +301,7 @@ func (s *VisibilityStore) countChasmExecutions(
func (s *VisibilityStore) countChasmExecutionsLegacy(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
request *visibilityservice.CountChasmExecutionsRequest,
mapper *chasm.VisibilitySearchAttributesMapper,
) (*store.InternalCountExecutionsResponse, error) {
saTypeMap, err := s.searchAttributesProvider.GetSearchAttributes(s.GetIndexName(), false)
@@ -308,20 +309,20 @@ func (s *VisibilityStore) countChasmExecutionsLegacy(
return nil, err
}
saMapper, err := s.searchAttributesMapperProvider.GetMapper(request.Namespace)
saMapper, err := s.searchAttributesMapperProvider.GetMapper(namespace.Name(request.Namespace))
if err != nil {
return nil, err
}
converter := NewQueryConverterLegacy(
s.GetName(),
request.Namespace,
request.NamespaceID,
namespace.Name(request.Namespace),
namespace.ID(request.NamespaceId),
saTypeMap,
saMapper,
request.Query,
mapper,
request.ArchetypeID,
request.ArchetypeId,
)
selectFilter, err := converter.BuildCountStmt()
if err != nil {
@@ -9,6 +9,7 @@ import (
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/persistence"
"go.temporal.io/server/common/persistence/visibility/manager"
)
@@ -33,11 +34,12 @@ type (
// Read APIs.
ListWorkflowExecutions(ctx context.Context, request *manager.ListWorkflowExecutionsRequestV2) (*InternalListExecutionsResponse, error)
ListChasmExecutions(ctx context.Context, request *manager.ListChasmExecutionsRequest) (*InternalListExecutionsResponse, error)
CountWorkflowExecutions(ctx context.Context, request *manager.CountWorkflowExecutionsRequest) (*InternalCountExecutionsResponse, error)
CountChasmExecutions(ctx context.Context, request *manager.CountChasmExecutionsRequest) (*InternalCountExecutionsResponse, error)
GetWorkflowExecution(ctx context.Context, request *manager.GetWorkflowExecutionRequest) (*InternalGetWorkflowExecutionResponse, error)
ListChasmExecutions(ctx context.Context, request *visibilityservice.ListChasmExecutionsRequest) (*InternalListExecutionsResponse, error)
CountChasmExecutions(ctx context.Context, request *visibilityservice.CountChasmExecutionsRequest) (*InternalCountExecutionsResponse, error)
// Admin APIs
// AddSearchAttributes makes schema changes to add the search attributes. This function must be
@@ -13,6 +13,7 @@ import (
context "context"
reflect "reflect"
visibilityservice "go.temporal.io/server/api/visibilityservice/v1"
manager "go.temporal.io/server/common/persistence/visibility/manager"
gomock "go.uber.org/mock/gomock"
)
@@ -68,7 +69,7 @@ func (mr *MockVisibilityStoreMockRecorder) Close() *gomock.Call {
}
// CountChasmExecutions mocks base method.
func (m *MockVisibilityStore) CountChasmExecutions(ctx context.Context, request *manager.CountChasmExecutionsRequest) (*InternalCountExecutionsResponse, error) {
func (m *MockVisibilityStore) CountChasmExecutions(ctx context.Context, request *visibilityservice.CountChasmExecutionsRequest) (*InternalCountExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "CountChasmExecutions", ctx, request)
ret0, _ := ret[0].(*InternalCountExecutionsResponse)
@@ -155,7 +156,7 @@ func (mr *MockVisibilityStoreMockRecorder) GetWorkflowExecution(ctx, request any
}
// ListChasmExecutions mocks base method.
func (m *MockVisibilityStore) ListChasmExecutions(ctx context.Context, request *manager.ListChasmExecutionsRequest) (*InternalListExecutionsResponse, error) {
func (m *MockVisibilityStore) ListChasmExecutions(ctx context.Context, request *visibilityservice.ListChasmExecutionsRequest) (*InternalListExecutionsResponse, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ListChasmExecutions", ctx, request)
ret0, _ := ret[0].(*InternalListExecutionsResponse)
@@ -5,8 +5,7 @@ import (
"errors"
"sync"
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/namespace"
"go.temporal.io/server/common/persistence/visibility/manager"
@@ -163,13 +162,13 @@ func (v *VisibilityManagerDual) ListWorkflowExecutions(
func (v *VisibilityManagerDual) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
request *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
return dualReadWrapper(
ctx,
v,
request,
request.Namespace,
namespace.Name(request.Namespace),
manager.VisibilityManager.ListChasmExecutions,
)
}
@@ -189,13 +188,13 @@ func (v *VisibilityManagerDual) CountWorkflowExecutions(
func (v *VisibilityManagerDual) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
) (*chasm.CountExecutionsResponse, error) {
request *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
return dualReadWrapper(
ctx,
v,
request,
request.Namespace,
namespace.Name(request.Namespace),
manager.VisibilityManager.CountChasmExecutions,
)
}
@@ -12,6 +12,8 @@ import (
"go.temporal.io/api/serviceerror"
workflowpb "go.temporal.io/api/workflow/v1"
"go.temporal.io/api/workflowservice/v1"
chasmspb "go.temporal.io/server/api/chasm/v1"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/namespace"
@@ -154,57 +156,61 @@ func (p *visibilityManagerImpl) ListWorkflowExecutions(
func (p *visibilityManagerImpl) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
request *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
response, err := p.store.ListChasmExecutions(ctx, request)
if err != nil {
return nil, err
}
rc, ok := p.chasmRegistry.ComponentByID(request.ArchetypeID)
rc, ok := p.chasmRegistry.ComponentByID(request.ArchetypeId)
if !ok {
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeID))
return nil, serviceerror.NewInvalidArgument(fmt.Sprintf("unknown archetype ID: %d", request.ArchetypeId))
}
mapper := rc.SearchAttributesMapper()
executions := make([]*chasm.ExecutionInfo[*commonpb.Payload], 0, len(response.Executions))
executions := make([]*chasmspb.VisibilityExecutionInfo, 0, len(response.Executions))
for _, exec := range response.Executions {
executionInfo, err := p.convertToChasmExecutionInfo(exec, mapper, request.Namespace)
executionInfo, err := p.convertToChasmExecutionInfo(
exec,
mapper,
namespace.Name(request.Namespace),
)
if err != nil {
return nil, err
}
executions = append(executions, executionInfo)
}
return &chasm.ListExecutionsResponse[*commonpb.Payload]{
return &visibilityservice.ListChasmExecutionsResponse{
Executions: executions,
NextPageToken: response.NextPageToken,
}, nil
}
// convertInternalExecutionToChasmInfo converts a store.InternalExecutionInfo to chasm.ExecutionInfo.
// convertInternalExecutionToChasmInfo converts a store.InternalExecutionInfo to ChasmExecutionInfo.
func (p *visibilityManagerImpl) convertToChasmExecutionInfo(
exec *store.InternalExecutionInfo,
mapper *chasm.VisibilitySearchAttributesMapper,
namespaceName namespace.Name,
) (*chasm.ExecutionInfo[*commonpb.Payload], error) {
) (*chasmspb.VisibilityExecutionInfo, error) {
customSAs, chasmSAs := splitSearchAttributes(exec.SearchAttributes)
chasmTypeMap := searchattribute.NewNameTypeMap(mapper.SATypeMap())
decodedChasmSAs, err := searchattribute.Decode(chasmSAs, &chasmTypeMap, false)
if err != nil {
return nil, err
}
chasmAliasedSAs, err := aliasChasmSearchAttributes(decodedChasmSAs, mapper)
chasmAliasedSAs, err := aliasChasmSearchAttributes(chasmSAs, mapper)
if err != nil {
return nil, err
}
if chasmAliasedSAs == nil {
chasmAliasedSAs = make(map[string]chasm.VisibilityValue)
if len(chasmAliasedSAs.GetIndexedFields()) == 0 {
chasmAliasedSAs = &commonpb.SearchAttributes{
IndexedFields: make(map[string]*commonpb.Payload),
}
}
if exec.TaskQueue != "" {
chasmAliasedSAs[sadefs.TaskQueue] = chasm.VisibilityValueKeyword(exec.TaskQueue)
chasmAliasedSAs.IndexedFields[sadefs.TaskQueue] = sadefs.MustEncodeValue(
exec.TaskQueue,
enumspb.INDEXED_VALUE_TYPE_KEYWORD,
)
}
customAliasedSAs, err := searchattribute.AliasFields(
@@ -221,16 +227,16 @@ func (p *visibilityManagerImpl) convertToChasmExecutionInfo(
return nil, err
}
return &chasm.ExecutionInfo[*commonpb.Payload]{
BusinessID: exec.WorkflowID,
RunID: exec.RunID,
StartTime: exec.StartTime,
CloseTime: exec.CloseTime,
return &chasmspb.VisibilityExecutionInfo{
BusinessId: exec.WorkflowID,
RunId: exec.RunID,
StartTime: timestamppb.New(exec.StartTime),
CloseTime: timestamppb.New(exec.CloseTime),
HistoryLength: exec.HistoryLength,
HistorySizeBytes: exec.HistorySizeBytes,
StateTransitionCount: exec.StateTransitionCount,
ChasmSearchAttributes: chasm.NewSearchAttributesMap(chasmAliasedSAs),
CustomSearchAttributes: customAliasedSAs.GetIndexedFields(),
ChasmSearchAttributes: chasmAliasedSAs,
CustomSearchAttributes: customAliasedSAs,
Memo: userMemo,
ChasmMemo: chasmMemoPayload,
}, nil
@@ -274,8 +280,8 @@ func splitUserAndChasmMemo(exec *store.InternalExecutionInfo) (userMemo *commonp
func (p *visibilityManagerImpl) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
) (*chasm.CountExecutionsResponse, error) {
request *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
internalResp, err := p.store.CountChasmExecutions(ctx, request)
if err != nil {
return nil, err
@@ -340,17 +346,20 @@ func (p *visibilityManagerImpl) convertToCountWorkflowExecutionsResponse(
func (p *visibilityManagerImpl) convertToCountChasmExecutionsResponse(
internal *store.InternalCountExecutionsResponse,
) (*chasm.CountExecutionsResponse, error) {
response := &chasm.CountExecutionsResponse{
Count: internal.Count,
Groups: make([]chasm.Group, 0, len(internal.Groups)),
) (*visibilityservice.CountChasmExecutionsResponse, error) {
response := &visibilityservice.CountChasmExecutionsResponse{
Count: internal.Count,
Groups: make(
[]*visibilityservice.CountChasmExecutionsResponse_AggregationGroup,
len(internal.Groups),
),
}
for _, group := range internal.Groups {
response.Groups = append(response.Groups, chasm.Group{
Values: group.GroupValues,
Count: group.Count,
})
for k, group := range internal.Groups {
response.Groups[k] = &visibilityservice.CountChasmExecutionsResponse_AggregationGroup{
GroupValues: group.GroupValues,
Count: group.Count,
}
}
return response, nil
@@ -546,18 +555,20 @@ func isChasmExecution(searchAttributes *commonpb.SearchAttributes) bool {
return false
}
// aliasChasmSearchAttributes aliases CHASM search attribute field names and converts them to VisibilityValue.
// aliasChasmSearchAttributes aliases CHASM search attribute field names.
// This function mirrors the pattern of searchattribute.AliasFields for custom search attributes.
func aliasChasmSearchAttributes(
decodedSearchAttributes map[string]any,
searchAttributes *commonpb.SearchAttributes,
mapper *chasm.VisibilitySearchAttributesMapper,
) (map[string]chasm.VisibilityValue, error) {
if mapper == nil || len(decodedSearchAttributes) == 0 {
) (*commonpb.SearchAttributes, error) {
if mapper == nil || len(searchAttributes.GetIndexedFields()) == 0 {
return nil, nil
}
result := make(map[string]chasm.VisibilityValue)
for fieldName, value := range decodedSearchAttributes {
result := &commonpb.SearchAttributes{
IndexedFields: make(map[string]*commonpb.Payload),
}
for fieldName, saPayload := range searchAttributes.GetIndexedFields() {
if !sadefs.IsChasmSearchAttribute(fieldName) {
continue
}
@@ -573,36 +584,8 @@ func aliasChasmSearchAttributes(
return nil, err
}
// Convert value to VisibilityValue (types should be correct after Decode)
visValue := convertToVisibilityValue(value)
result[aliasName] = visValue
result.IndexedFields[aliasName] = saPayload
}
return result, nil
}
// convertToVisibilityValue converts a value to VisibilityValue based on its runtime type.
// After Decode, the types should be correct, so simple type detection is sufficient.
func convertToVisibilityValue(value any) chasm.VisibilityValue {
switch val := value.(type) {
case int64:
return chasm.VisibilityValueInt64(val)
case float64:
return chasm.VisibilityValueFloat64(val)
case bool:
return chasm.VisibilityValueBool(val)
case time.Time:
return chasm.VisibilityValueTime(val)
case string:
// Try to parse as datetime first
if parsedTime, err := time.Parse(time.RFC3339, val); err == nil {
return chasm.VisibilityValueTime(parsedTime)
}
return chasm.VisibilityValueKeyword(val)
case []string:
return chasm.VisibilityValueStringSlice(val)
default:
// Return as string if type is unknown
return chasm.VisibilityValueKeyword(fmt.Sprintf("%v", val))
}
}
@@ -4,8 +4,7 @@ import (
"context"
"time"
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/server/chasm"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/headers"
"go.temporal.io/server/common/namespace"
@@ -120,8 +119,8 @@ func (m *visibilityManagerRateLimited) ListWorkflowExecutions(
func (m *visibilityManagerRateLimited) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
request *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
if ok := allow(ctx, "ListChasmExecutions", m.readRateLimiter); !ok {
return nil, persistence.ErrPersistenceSystemLimitExceeded
}
@@ -140,8 +139,8 @@ func (m *visibilityManagerRateLimited) CountWorkflowExecutions(
func (m *visibilityManagerRateLimited) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
) (*chasm.CountExecutionsResponse, error) {
request *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
if ok := allow(ctx, "CountChasmExecutions", m.readRateLimiter); !ok {
return nil, persistence.ErrPersistenceSystemLimitExceeded
}
@@ -4,9 +4,8 @@ import (
"context"
"time"
commonpb "go.temporal.io/api/common/v1"
"go.temporal.io/api/serviceerror"
"go.temporal.io/server/chasm"
"go.temporal.io/server/api/visibilityservice/v1"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
@@ -142,8 +141,8 @@ func (m *visibilityManagerMetrics) ListWorkflowExecutions(
func (m *visibilityManagerMetrics) ListChasmExecutions(
ctx context.Context,
request *manager.ListChasmExecutionsRequest,
) (*chasm.ListExecutionsResponse[*commonpb.Payload], error) {
request *visibilityservice.ListChasmExecutionsRequest,
) (*visibilityservice.ListChasmExecutionsResponse, error) {
handler, startTime := m.tagScope(metrics.VisibilityPersistenceListChasmExecutionsScope)
response, err := m.delegate.ListChasmExecutions(ctx, request)
elapsed := time.Since(startTime)
@@ -151,7 +150,7 @@ func (m *visibilityManagerMetrics) ListChasmExecutions(
m.logger.Warn("List query exceeded threshold",
tag.Duration("duration", elapsed),
tag.String("visibility-query", request.Query),
tag.Stringer("namespace", request.Namespace),
tag.String("namespace", request.Namespace),
)
}
metrics.VisibilityPersistenceLatency.With(handler).Record(elapsed)
@@ -173,8 +172,8 @@ func (m *visibilityManagerMetrics) CountWorkflowExecutions(
func (m *visibilityManagerMetrics) CountChasmExecutions(
ctx context.Context,
request *manager.CountChasmExecutionsRequest,
) (*chasm.CountExecutionsResponse, error) {
request *visibilityservice.CountChasmExecutionsRequest,
) (*visibilityservice.CountChasmExecutionsResponse, error) {
handler, startTime := m.tagScope(metrics.VisibilityPersistenceCountChasmExecutionsScope)
response, err := m.delegate.CountChasmExecutions(ctx, request)
elapsed := time.Since(startTime)
@@ -0,0 +1,22 @@
syntax = "proto3";
package temporal.server.api.chasm.v1;
option go_package = "go.temporal.io/server/api/chasm/v1;chasm";
import "google/protobuf/timestamp.proto";
import "temporal/api/common/v1/message.proto";
message VisibilityExecutionInfo {
string business_id = 1;
string run_id = 2;
google.protobuf.Timestamp start_time = 3;
google.protobuf.Timestamp close_time = 4;
int64 history_length = 5;
int64 history_size_bytes = 6;
int64 state_transition_count = 7;
temporal.api.common.v1.SearchAttributes chasm_search_attributes = 8;
temporal.api.common.v1.SearchAttributes custom_search_attributes = 9;
temporal.api.common.v1.Memo memo = 10;
temporal.api.common.v1.Payload chasm_memo = 11;
}
@@ -0,0 +1,49 @@
syntax = "proto3";
package temporal.server.api.visibilityservice.v1;
option go_package = "go.temporal.io/server/api/visibilityservice/v1;visibilityservice";
// visibilityservice does not have a corresponding gRPC service.
// The following request/response types are used by VisibilityManager.
import "temporal/api/common/v1/message.proto";
import "temporal/server/api/chasm/v1/message.proto";
message ListChasmExecutionsRequest {
// (-- api-linter: core::0141::forbidden-types=disabled --)
uint32 archetype_id = 1;
string namespace_id = 2;
string namespace = 3;
string query = 4;
// Maximum number of executions per page
int32 page_size = 5;
// Token to continue reading next page of executions.
// Pass in empty slice for first page.
bytes next_page_token = 6;
}
message ListChasmExecutionsResponse {
repeated temporal.server.api.chasm.v1.VisibilityExecutionInfo executions = 1;
bytes next_page_token = 2;
}
message CountChasmExecutionsRequest {
// (-- api-linter: core::0141::forbidden-types=disabled --)
uint32 archetype_id = 1;
string namespace_id = 2;
string namespace = 3;
string query = 4;
}
message CountChasmExecutionsResponse {
int64 count = 1;
// `groups` contains the groups if the request is grouping by a field.
// The list might not be complete, and the counts of each group are approximations.
repeated AggregationGroup groups = 2;
message AggregationGroup {
repeated temporal.api.common.v1.Payload group_values = 1;
int64 count = 2;
}
}