esreindexer

README

esreindexer

⚠️ Under implementation.

Elasticsearch reindex manager optimizing index parameters for reindex.

Usecase

esreindexer publish reindex tasks after optimizing the index setting to reindex. Start monitor to check the progress of the task, and if the task is finished, restore the index setting. The information of the task to be monitored is saved in the esreindexer store. The esreindexer store provides Store interface and the required store implementation is available.

Supports

Mode

• Agent mode
• API mode

Store

• Momory Store
• MySQL Store
• etcd Store

Others

• WebHook
• Before Plugin
• After Plugin

Using esreindexer as API server

API mode provides reindex API using esreindexer

// setup Elasticsearch
// example: setting with eskeeper
$ docker-compose up -d
$ eskeeper < testdata/test.eskeeper.ym 
$ ./example/load_testdata.sh

// checks data
// example-v1 has 2 docs
// example-v2 has no docs
$ curl localhost:9200/_cat/indices/example-*
yellow open example-v1 LnXp-WjXQh2iDjyCBd9fxg 2 2 2 0 4.2kb 4.2kb
yellow open example-v2 4pxR7cpvSU-p0mg1vol6-A 2 2 0 0  416b  416b

$ go run ./agent/main.go
$ curl -X POST -H "Content-Type: application/json" -d '{"source": {"index": "example-v1"}, "dest": {"index": "example-v2"}}' localhost:8888/api/v1/reindex 

$ curl localhost:9200/_cat/indices/example-*
yellow open example-v1 LnXp-WjXQh2iDjyCBd9fxg 2 2 2 0 4.2kb 4.2kb
yellow open example-v2 4pxR7cpvSU-p0mg1vol6-A 2 2 2 0 4.1kb 4.1kb

Environment Values

name	description
ESREINDEXER_LOGGING_LEVEL	reindexer logging level
ESREINDEXER_API_PORT	reindexer API port
ESREINDEXER_ES_ADDRESS	Elasticsearch address
ESREINDEXER_ES_USER	Elasticsearch user name
ESREINDEXER_ES_PASS	Elasticsearch password
ESREINDEXER_EXPIRE_DURAION	This specifies how long a task with a status of "done" will be retained

Using esreindexer components

Implementation example using esreindexer components is in the example directory.

// setup Elasticsearch
// example: setting with eskeeper
$ docker-compose up -d
$ eskeeper < testdata/test.eskeeper.ym 
$ ./example/load_testdata.sh

// checks data
// example-v1 has 2 docs
// example-v2 has no docs
$ curl localhost:9200/_cat/indices/example-*
yellow open example-v1 LnXp-WjXQh2iDjyCBd9fxg 2 2 2 0 4.2kb 4.2kb
yellow open example-v2 4pxR7cpvSU-p0mg1vol6-A 2 2 0 0  416b  416b

$ go run ./example/main.go

$ curl localhost:9200/_cat/indices/example-*
yellow open example-v1 LnXp-WjXQh2iDjyCBd9fxg 2 2 2 0 4.2kb 4.2kb
yellow open example-v2 4pxR7cpvSU-p0mg1vol6-A 2 2 2 0 4.1kb 4.1kb

Documentation

Index

• type AfterCompletionPlugin
• type ESClient
  • func NewESClient(es *elasticsearch.Client) *ESClient
  • func (c *ESClient) CompletedTask(ctx context.Context, taskID string) (bool, error)
  • func (c *ESClient) GetIndexSetting(ctx context.Context, index string) (numberOfReplicas int, refreshInterval int, err error)
  • func (c *ESClient) Reindex(ctx context.Context, src string, dest string) (string, error)
  • func (c *ESClient) UpdateIndexSetting(ctx context.Context, index string, numberOfReplicas int, refreshInterval int) error
• type GetTaskRes
• type IndesSettingRes
• type ReindexManager
  • func NewReindexManager(client *ESClient, store Store) *ReindexManager
  • func (m *ReindexManager) Monitor(ctx context.Context) error
  • func (m *ReindexManager) NotifyCompletionPlugin(p AfterCompletionPlugin)
  • func (m *ReindexManager) PublishReindexTask(ctx context.Context, src, dest string) (string, error)
• type ReindexRes
• type Setting
• type Store

Types

type AfterCompletionPlugin

type AfterCompletionPlugin interface {
	Run(ctx context.Context, taskID string) error
}

type ESClient

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

func NewESClient

func NewESClient(es *elasticsearch.Client) *ESClient

func (*ESClient) CompletedTask

func (c *ESClient) CompletedTask(ctx context.Context, taskID string) (bool, error)

func (*ESClient) GetIndexSetting

func (c *ESClient) GetIndexSetting(ctx context.Context, index string) (numberOfReplicas int, refreshInterval int, err error)

func (*ESClient) Reindex

func (c *ESClient) Reindex(ctx context.Context, src string, dest string) (string, error)

func (*ESClient) UpdateIndexSetting

func (c *ESClient) UpdateIndexSetting(ctx context.Context, index string, numberOfReplicas int, refreshInterval int) error

type GetTaskRes

type GetTaskRes struct {
	Completed bool `json:"completed"`
}

type IndesSettingRes

type IndesSettingRes map[string]map[string]Setting

type ReindexManager

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

func NewReindexManager

func NewReindexManager(client *ESClient, store Store) *ReindexManager

func (*ReindexManager) Monitor

func (m *ReindexManager) Monitor(ctx context.Context) error

func (*ReindexManager) NotifyCompletionPlugin

func (m *ReindexManager) NotifyCompletionPlugin(p AfterCompletionPlugin)

func (*ReindexManager) PublishReindexTask

func (m *ReindexManager) PublishReindexTask(ctx context.Context, src, dest string) (string, error)

type ReindexRes

type ReindexRes struct {
	Task string `json:"task"`
}

type Setting

type Setting struct {
	RefreshInterval  string `json:"index.refresh_interval"`
	NumberOfReplicas string `json:"index.number_of_replicas"`
}

type Store

type Store interface {
	PutTaskInfo(index string, taskID string, NumberOfReplicas int, RefreshInterval int) error
	TaskInfo(taskID string) (numberOfReplicas int, refreshInterval int, err error)
	DeleteTask(taskID string) error
	DoneTask(taskID string) error
	AllTask() map[string]entity.Task
}