elasticsearch

Documentation

Index

• Constants
• Variables
• func SerializePageToken(token *ElasticVisibilityPageToken) ([]byte, error)
• func ShouldSearchAfter(token *ElasticVisibilityPageToken) bool
• type BulkProcessorParameters
• type ElasticVisibilityPageToken
  • func DeserializePageToken(data []byte) (*ElasticVisibilityPageToken, error)
  • func GetNextPageToken(token []byte) (*ElasticVisibilityPageToken, error)
• type GenericBackoff
  • func NewExponentialBackoff(initialTimeout, maxTimeout time.Duration) GenericBackoff
• type GenericBulkAfterFunc
• type GenericBulkBeforeFunc
• type GenericBulkProcessor
• type GenericBulkResponse
• type GenericBulkResponseItem
• type GenericBulkableAddRequest
• type GenericBulkableRequest
• type GenericClient
  • func NewGenericClient(connectConfig *config.ElasticSearchConfig, logger log.Logger) (GenericClient, error)
  • func NewV6Client(connectConfig *config.ElasticSearchConfig, logger log.Logger, ...) (GenericClient, error)
  • func NewV7Client(connectConfig *config.ElasticSearchConfig, logger log.Logger, ...) (GenericClient, error)
• type GenericError
• type GenericMatch
• type IsRecordValidFilter
• type ScanByQueryRequest
• type SearchByQueryRequest
• type SearchForOneClosedExecutionRequest
• type SearchForOneClosedExecutionResponse
• type SearchRequest
• type SearchResponse
• type VisibilityRecord

Constants

const (
	DomainID      = "DomainID"
	WorkflowID    = "WorkflowID"
	RunID         = "RunID"
	WorkflowType  = "WorkflowType"
	StartTime     = "StartTime"
	ExecutionTime = "ExecutionTime"
	CloseTime     = "CloseTime"
	CloseStatus   = "CloseStatus"
	HistoryLength = "HistoryLength"
	Memo          = "Memo"
	Encoding      = "Encoding"
	TaskList      = "TaskList"

	KafkaKey = "KafkaKey"
)

All legal fields allowed in elastic search index

Variables

var (
	FieldTypeString = indexer.FieldTypeString
	FieldTypeInt    = indexer.FieldTypeInt
	FieldTypeBool   = indexer.FieldTypeBool
	FieldTypeBinary = indexer.FieldTypeBinary
)

Supported field types

Functions

func SerializePageToken

func SerializePageToken(token *ElasticVisibilityPageToken) ([]byte, error)

SerializePageToken return the token blob

func ShouldSearchAfter

func ShouldSearchAfter(token *ElasticVisibilityPageToken) bool

ShouldSearchAfter decides if should search after

Types

type BulkProcessorParameters

type BulkProcessorParameters struct {
	Name          string
	NumOfWorkers  int
	BulkActions   int
	BulkSize      int
	FlushInterval time.Duration
	Backoff       GenericBackoff
	BeforeFunc    GenericBulkBeforeFunc
	AfterFunc     GenericBulkAfterFunc
}

BulkProcessorParameters holds all required and optional parameters for executing bulk service

type ElasticVisibilityPageToken

type ElasticVisibilityPageToken struct {
	// for ES API From+Size
	From int
	// for ES API searchAfter
	SortValue  interface{}
	TieBreaker string // runID
	// for ES scroll API
	ScrollID string
}

ElasticVisibilityPageToken holds the paging token for ElasticSearch

func DeserializePageToken

func DeserializePageToken(data []byte) (*ElasticVisibilityPageToken, error)

DeserializePageToken return the structural token

func GetNextPageToken

func GetNextPageToken(token []byte) (*ElasticVisibilityPageToken, error)

GetNextPageToken returns the structural token with nil handling

type GenericBackoff

type GenericBackoff interface {
	// Next implements a BackoffFunc.
	Next(retry int) (time.Duration, bool)
}

GenericBackoff allows callers to implement their own Backoff strategy.

func NewExponentialBackoff

func NewExponentialBackoff(initialTimeout, maxTimeout time.Duration) GenericBackoff

NewExponentialBackoff returns a exponentialBackoff backoff policy. Use initialTimeout to set the first/minimal interval and maxTimeout to set the maximum wait interval.

type GenericBulkAfterFunc

type GenericBulkAfterFunc func(executionId int64, requests []GenericBulkableRequest, response *GenericBulkResponse, err *GenericError)

GenericBulkAfterFunc defines the signature of callbacks that are executed after a commit to Elasticsearch. The err parameter signals an error.

type GenericBulkBeforeFunc

type GenericBulkBeforeFunc func(executionId int64, requests []GenericBulkableRequest)

GenericBulkBeforeFunc defines the signature of callbacks that are executed before a commit to Elasticsearch.

type GenericBulkProcessor

type GenericBulkProcessor interface {
	Start(ctx context.Context) error
	Stop() error
	Close() error
	Add(request *GenericBulkableAddRequest)
	Flush() error
	RetrieveKafkaKey(request GenericBulkableRequest, logger log.Logger, client metrics.Client) string
}

GenericBulkProcessor is a bulk processor

type GenericBulkResponse

type GenericBulkResponse struct {
	Took   int                                   `json:"took,omitempty"`
	Errors bool                                  `json:"errors,omitempty"`
	Items  []map[string]*GenericBulkResponseItem `json:"items,omitempty"`
}

GenericBulkResponse is generic struct of bulk response

type GenericBulkResponseItem

type GenericBulkResponseItem struct {
	Index         string `json:"_index,omitempty"`
	Type          string `json:"_type,omitempty"`
	ID            string `json:"_id,omitempty"`
	Version       int64  `json:"_version,omitempty"`
	Result        string `json:"result,omitempty"`
	SeqNo         int64  `json:"_seq_no,omitempty"`
	PrimaryTerm   int64  `json:"_primary_term,omitempty"`
	Status        int    `json:"status,omitempty"`
	ForcedRefresh bool   `json:"forced_refresh,omitempty"`
	// the error details
	Error interface{}
}

GenericBulkResponseItem is the result of a single bulk request.

type GenericBulkableAddRequest

type GenericBulkableAddRequest struct {
	Index       string
	Type        string
	ID          string
	VersionType string
	Version     int64
	// true means it's delete, otherwise it's a index request
	IsDelete bool
	// should be nil if IsDelete is true
	Doc interface{}
}

GenericBulkableAddRequest a struct to hold a bulk request

type GenericBulkableRequest

type GenericBulkableRequest interface {
	fmt.Stringer
	Source() ([]string, error)
}

GenericBulkableRequest is a generic interface to bulkable requests.

type GenericClient

type GenericClient interface {
	// Search API is only for supporting various List[Open/Closed]WorkflowExecutions(ByXyz).
	// Use SearchByQuery or ScanByQuery for generic purpose searching.
	Search(ctx context.Context, request *SearchRequest) (*SearchResponse, error)
	// SearchByQuery is the generic purpose searching
	SearchByQuery(ctx context.Context, request *SearchByQueryRequest) (*SearchResponse, error)
	// ScanByQuery is also generic purpose searching, but implemented with ScrollService of ElasticSearch,
	// which is more performant for pagination, but comes with some limitation of in-parallel requests.
	ScanByQuery(ctx context.Context, request *ScanByQueryRequest) (*SearchResponse, error)
	// TODO remove it in https://github.com/uber/cadence/issues/3682
	SearchForOneClosedExecution(ctx context.Context, index string, request *SearchForOneClosedExecutionRequest) (*SearchForOneClosedExecutionResponse, error)
	// CountByQuery is for returning the count of workflow executions that match the query
	CountByQuery(ctx context.Context, index, query string) (int64, error)

	// RunBulkProcessor returns a processor for adding/removing docs into ElasticSearch index
	RunBulkProcessor(ctx context.Context, p *BulkProcessorParameters) (GenericBulkProcessor, error)

	// PutMapping adds new field type to the index
	PutMapping(ctx context.Context, index, root, key, valueType string) error
	// CreateIndex creates a new index
	CreateIndex(ctx context.Context, index string) error

	IsNotFoundError(err error) bool
}

GenericClient is a generic interface for all versions of ElasticSearch clients

func NewGenericClient

func NewGenericClient(
	connectConfig *config.ElasticSearchConfig,
	logger log.Logger,
) (GenericClient, error)

NewGenericClient create a ES client

func NewV6Client

func NewV6Client(
	connectConfig *config.ElasticSearchConfig,
	logger log.Logger,
	clientOptFuncs ...elastic.ClientOptionFunc,
) (GenericClient, error)

NewV6Client returns a new implementation of GenericClient

func NewV7Client

func NewV7Client(
	connectConfig *config.ElasticSearchConfig,
	logger log.Logger,
	clientOptFuncs ...elastic.ClientOptionFunc,
) (GenericClient, error)

NewV7Client returns a new implementation of GenericClient

type GenericError

type GenericError struct {
	Status  int   `json:"status"`
	Details error `json:"error,omitempty"`
}

GenericError encapsulates error status and details returned from Elasticsearch.

type GenericMatch

type GenericMatch struct {
	Name string
	Text interface{}
}

GenericMatch is a match struct

type IsRecordValidFilter

type IsRecordValidFilter func(rec *p.InternalVisibilityWorkflowExecutionInfo) bool

IsRecordValidFilter is a function to filter visibility records

type ScanByQueryRequest

type ScanByQueryRequest struct {
	Index         string
	Query         string
	NextPageToken []byte
	PageSize      int
}

ScanByQueryRequest is request for SearchByQuery

type SearchByQueryRequest

type SearchByQueryRequest struct {
	Index           string
	Query           string
	NextPageToken   []byte
	PageSize        int
	Filter          IsRecordValidFilter
	MaxResultWindow int
}

SearchByQueryRequest is request for SearchByQuery

type SearchForOneClosedExecutionRequest

type SearchForOneClosedExecutionRequest = p.InternalGetClosedWorkflowExecutionRequest

SearchForOneClosedExecutionRequest is request for SearchForOneClosedExecution

type SearchForOneClosedExecutionResponse

type SearchForOneClosedExecutionResponse = p.InternalGetClosedWorkflowExecutionResponse

SearchForOneClosedExecutionResponse is response for SearchForOneClosedExecution

type SearchRequest

type SearchRequest struct {
	Index           string
	ListRequest     *p.InternalListWorkflowExecutionsRequest
	IsOpen          bool
	Filter          IsRecordValidFilter
	MatchQuery      *GenericMatch
	MaxResultWindow int
}

SearchRequest is request for Search

type SearchResponse

type SearchResponse = p.InternalListWorkflowExecutionsResponse

SearchResponse is a response to Search, SearchByQuery and ScanByQuery

type VisibilityRecord

type VisibilityRecord struct {
	WorkflowID    string
	RunID         string
	WorkflowType  string
	StartTime     int64
	ExecutionTime int64
	CloseTime     int64
	CloseStatus   workflow.WorkflowExecutionCloseStatus
	HistoryLength int64
	Memo          []byte
	Encoding      string
	TaskList      string
	Attr          map[string]interface{}
}

VisibilityRecord is a struct of doc for deserialization