indexer

Documentation

Index

• type Config
• type ESProcessor
• type ESProcessorImpl
  • func (p *ESProcessorImpl) Add(request *bulk.GenericBulkableAddRequest, key string, ...)
  • func (p *ESProcessorImpl) Start()
  • func (p *ESProcessorImpl) Stop()
• type Indexer
  • func NewIndexer(config *Config, client messaging.Client, esClient es.GenericClient, ...) *Indexer
  • func (i *Indexer) Start() error
  • func (i *Indexer) Stop()

Types

type Config

type Config struct {
	IndexerConcurrency             dynamicconfig.IntPropertyFn
	ESProcessorNumOfWorkers        dynamicconfig.IntPropertyFn
	ESProcessorBulkActions         dynamicconfig.IntPropertyFn // max number of requests in bulk
	ESProcessorBulkSize            dynamicconfig.IntPropertyFn // max total size of bytes in bulk
	ESProcessorFlushInterval       dynamicconfig.DurationPropertyFn
	ValidSearchAttributes          dynamicconfig.MapPropertyFn
	EnableQueryAttributeValidation dynamicconfig.BoolPropertyFn
}

Config contains all configs for indexer

type ESProcessor

type ESProcessor interface {
	common.Daemon
	Add(request *bulk.GenericBulkableAddRequest, key string, kafkaMsg messaging.Message)
}

type ESProcessorImpl

type ESProcessorImpl struct {
	// contains filtered or unexported fields
}

ESProcessorImpl implements ESProcessor, it's an agent of GenericBulkProcessor

func (*ESProcessorImpl) Add

func (p *ESProcessorImpl) Add(request *bulk.GenericBulkableAddRequest, key string, kafkaMsg messaging.Message)

Add an ES request, and an map item for kafka message

func (*ESProcessorImpl) Start

func (p *ESProcessorImpl) Start()

func (*ESProcessorImpl) Stop

func (p *ESProcessorImpl) Stop()

type Indexer

type Indexer struct {
	// contains filtered or unexported fields
}

Indexer used to consumer data from kafka then send to ElasticSearch

func NewIndexer

func NewIndexer(
	config *Config,
	client messaging.Client,
	esClient es.GenericClient,
	visibilityName string,
	logger log.Logger,
	metricsClient metrics.Client,
) *Indexer

NewIndexer create a new Indexer

func (*Indexer) Start

func (i *Indexer) Start() error

func (*Indexer) Stop

func (i *Indexer) Stop()