Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add Elasticsearch CLI tool (#8296)

Browse Source 
## What changed?

- Revive and update original PR #2977 for Elasticsearch CLI tool

Replaces manual curl invocations with a proper CLI tool that leverages
Temporal's built-in Elasticsearch auth providers and provides better
error handling and logging.

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

## Potential risks

None, net new.

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Introduces `temporal-elasticsearch-tool` for ES schema/index
management and ping, extends ES client APIs, embeds ES schema, and
updates Makefile to use the tool.
> 
> - **Tools**:
> - New `temporal-elasticsearch-tool` CLI with commands: `setup-schema`,
`update-schema`, `create-index`, `drop-index`, `ping`; supports AWS auth
and uses embedded schema files.
> - Adds entrypoint `cmd/tools/elasticsearch`, README, and basic tests.
> - **Elasticsearch Client**:
> - Extends `CLIClient` with `ClusterPutSettings`, `IndexPutTemplate`,
`IndexPutMapping`, `Ping` and implements them (v7) using raw requests
where needed; allows custom HTTP client from config (e.g., AWS-signed).
> - **Schema**:
> - Embeds ES v7 cluster settings and index template
(`schema.Embedded...` accessors).
> - **Build/Makefile**:
> - Adds build target and binary cleanup for
`temporal-elasticsearch-tool`; updates `install-schema-es` and
`install-schema-xdc` to use the CLI instead of curl.
> - Includes binary in `.goreleaser.yml`; excludes it in
`.dockerignore`.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
b37864809a. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
This commit is contained in:
Rob Holland
2025-10-02 10:52:21 +01:00
committed by GitHub
parent 30d33ce050
commit 1dc6264f4f
15 changed files with 1098 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.dockerignore
View File

@@ -14,3 +14,4 @@ tdbg
temporal-server
temporal-cassandra-tool
temporal-sql-tool
temporal-elasticsearch-tool

13
.goreleaser.yml
View File

@@ -8,6 +8,7 @@ archives:
- temporal-server
- temporal-cassandra-tool
- temporal-sql-tool
- temporal-elasticsearch-tool
name_template: "{{ .ProjectName }}_{{ .Version }}_{{ .Os }}_{{ .Arch }}"
format_overrides:
- goos: windows
@@ -52,6 +53,18 @@ builds:
goarch:
- amd64
- arm64
- id: temporal-elasticsearch-tool
dir: cmd/tools/elasticsearch
binary: temporal-elasticsearch-tool
env:
- CGO_ENABLED=0
goos:
- linux
- darwin
- windows
goarch:
- amd64
- arm64
- id: tdbg
dir: cmd/tools/tdbg
binary: tdbg

44
Makefile
View File

@@ -3,7 +3,7 @@
install: bins
# Rebuild binaries (used by Dockerfile).
bins: temporal-server temporal-cassandra-tool temporal-sql-tool tdbg
bins: temporal-server temporal-cassandra-tool temporal-sql-tool temporal-elasticsearch-tool tdbg
# Install all tools, recompile proto files, run all possible checks and tests (long but comprehensive).
all: clean proto bins check test
@@ -334,6 +334,7 @@ clean-bins:
@rm -f temporal-cassandra-tool
@rm -f tdbg
@rm -f temporal-sql-tool
@rm -f temporal-elasticsearch-tool
temporal-server: $(ALL_SRC)
@printf $(COLOR) "Build temporal-server with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
@@ -351,6 +352,10 @@ temporal-sql-tool: $(ALL_SRC)
@printf $(COLOR) "Build temporal-sql-tool with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
CGO_ENABLED=$(CGO_ENABLED) go build $(BUILD_TAG_FLAG) -o temporal-sql-tool ./cmd/tools/sql
temporal-elasticsearch-tool: $(ALL_SRC)
@printf $(COLOR) "Build temporal-elasticsearch-tool with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
CGO_ENABLED=$(CGO_ENABLED) go build $(BUILD_TAG_FLAG) -o temporal-elasticsearch-tool ./cmd/tools/elasticsearch
temporal-server-debug: $(ALL_SRC)
@printf $(COLOR) "Build temporal-server-debug with CGO_ENABLED=$(CGO_ENABLED) for $(GOOS)/$(GOARCH)..."
CGO_ENABLED=$(CGO_ENABLED) go build $(BUILD_TAG_FLAG),TEMPORAL_DEBUG -o temporal-server-debug ./cmd/server
@@ -519,22 +524,17 @@ install-schema-postgresql12: temporal-sql-tool
./temporal-sql-tool -u $(SQL_USER) --pw $(SQL_PASSWORD) -p 5432 --pl postgres12 --db $(VISIBILITY_DB) setup-schema -v 0.0
./temporal-sql-tool -u $(SQL_USER) --pw $(SQL_PASSWORD) -p 5432 --pl postgres12 --db $(VISIBILITY_DB) update-schema -d ./schema/postgresql/v12/visibility/versioned
install-schema-es:
install-schema-es: temporal-elasticsearch-tool
@printf $(COLOR) "Install Elasticsearch schema..."
curl --fail -X PUT "http://127.0.0.1:9200/_cluster/settings" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/cluster_settings_v7.json --write-out "\n"
curl --fail -X PUT "http://127.0.0.1:9200/_template/temporal_visibility_v1_template" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/index_template_v7.json --write-out "\n"
# No --fail here because create index is not idempotent operation.
curl -X PUT "http://127.0.0.1:9200/temporal_visibility_v1_dev" --write-out "\n"
# curl -X PUT "http://127.0.0.1:9200/temporal_visibility_v1_secondary" --write-out "\n"
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 setup-schema
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 create-index --index temporal_visibility_v1_dev
install-schema-es-secondary:
install-schema-es-secondary: temporal-elasticsearch-tool
@printf $(COLOR) "Install Elasticsearch schema..."
curl --fail -X PUT "http://127.0.0.1:8200/_cluster/settings" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/cluster_settings_v7.json --write-out "\n"
curl --fail -X PUT "http://127.0.0.1:8200/_template/temporal_visibility_v1_template" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/index_template_v7.json --write-out "\n"
# No --fail here because create index is not idempotent operation.
curl -X PUT "http://127.0.0.1:8200/temporal_visibility_v1_secondary" --write-out "\n"
./temporal-elasticsearch-tool -e http://127.0.0.1:8200 setup-schema
./temporal-elasticsearch-tool -e http://127.0.0.1:8200 create-index --index temporal_visibility_v1_secondary
install-schema-xdc: temporal-cassandra-tool
install-schema-xdc: temporal-cassandra-tool temporal-elasticsearch-tool
@printf $(COLOR) "Install Cassandra schema (active)..."
./temporal-cassandra-tool drop -k temporal_cluster_a -f
./temporal-cassandra-tool create -k temporal_cluster_a --rf 1
@@ -554,15 +554,15 @@ install-schema-xdc: temporal-cassandra-tool
./temporal-cassandra-tool -k temporal_cluster_c update-schema -d ./schema/cassandra/temporal/versioned
@printf $(COLOR) "Install Elasticsearch schemas..."
curl --fail -X PUT "http://127.0.0.1:9200/_cluster/settings" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/cluster_settings_v7.json --write-out "\n"
curl --fail -X PUT "http://127.0.0.1:9200/_template/temporal_visibility_v1_template" -H "Content-Type: application/json" --data-binary @./schema/elasticsearch/visibility/index_template_v7.json --write-out "\n"
# No --fail here because create index is not idempotent operation.
curl -X DELETE http://localhost:9200/temporal_visibility_v1_dev_cluster_a
curl -X DELETE http://localhost:9200/temporal_visibility_v1_dev_cluster_b
curl -X DELETE http://localhost:9200/temporal_visibility_v1_dev_cluster_c
curl -X PUT "http://127.0.0.1:9200/temporal_visibility_v1_dev_cluster_a" --write-out "\n"
curl -X PUT "http://127.0.0.1:9200/temporal_visibility_v1_dev_cluster_b" --write-out "\n"
curl -X PUT "http://127.0.0.1:9200/temporal_visibility_v1_dev_cluster_c" --write-out "\n"
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 setup-schema
# Delete indices if they exist (drop-index fails silently if index doesn't exist)
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 drop-index --index temporal_visibility_v1_dev_cluster_a --fail
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 drop-index --index temporal_visibility_v1_dev_cluster_b --fail
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 drop-index --index temporal_visibility_v1_dev_cluster_c --fail
# Create indices
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 create-index --index temporal_visibility_v1_dev_cluster_a
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 create-index --index temporal_visibility_v1_dev_cluster_b
./temporal-elasticsearch-tool -e http://127.0.0.1:9200 create-index --index temporal_visibility_v1_dev_cluster_c
##### Run server #####
DOCKER_COMPOSE_FILES := -f ./develop/docker-compose/docker-compose.yml -f ./develop/docker-compose/docker-compose.$(GOOS).yml

13
cmd/tools/elasticsearch/main.go Normal file
View File

@@ -0,0 +1,13 @@
package main
import (
"os"
"go.temporal.io/server/tools/elasticsearch"
)
func main() {
if err := elasticsearch.RunTool(os.Args); err != nil {
os.Exit(1)
}
}

4
common/persistence/visibility/store/elasticsearch/client/client.go
View File

@@ -37,6 +37,10 @@ type (
CLIClient interface {
Client
Delete(ctx context.Context, indexName string, docID string, version int64) error
IndexPutTemplate(ctx context.Context, templateName string, bodyString string) (bool, error)
IndexPutMapping(ctx context.Context, indexName string, bodyString string) (bool, error)
ClusterPutSettings(ctx context.Context, bodyString string) (bool, error)
Ping(ctx context.Context) error
}
IntegrationTestsClient interface {

59
common/persistence/visibility/store/elasticsearch/client/client_mock.go
View File

@@ -261,6 +261,21 @@ func (mr *MockCLIClientMockRecorder) CatIndices(ctx, target any) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CatIndices", reflect.TypeOf((*MockCLIClient)(nil).CatIndices), ctx, target)
}
// ClusterPutSettings mocks base method.
func (m *MockCLIClient) ClusterPutSettings(ctx context.Context, bodyString string) (bool, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ClusterPutSettings", ctx, bodyString)
ret0, _ := ret[0].(bool)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// ClusterPutSettings indicates an expected call of ClusterPutSettings.
func (mr *MockCLIClientMockRecorder) ClusterPutSettings(ctx, bodyString any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClusterPutSettings", reflect.TypeOf((*MockCLIClient)(nil).ClusterPutSettings), ctx, bodyString)
}
// Count mocks base method.
func (m *MockCLIClient) Count(ctx context.Context, index string, query elastic.Query) (int64, error) {
m.ctrl.T.Helper()
@@ -380,6 +395,50 @@ func (mr *MockCLIClientMockRecorder) IndexExists(ctx, indexName any) *gomock.Cal
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexExists", reflect.TypeOf((*MockCLIClient)(nil).IndexExists), ctx, indexName)
}
// IndexPutMapping mocks base method.
func (m *MockCLIClient) IndexPutMapping(ctx context.Context, indexName, bodyString string) (bool, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "IndexPutMapping", ctx, indexName, bodyString)
ret0, _ := ret[0].(bool)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// IndexPutMapping indicates an expected call of IndexPutMapping.
func (mr *MockCLIClientMockRecorder) IndexPutMapping(ctx, indexName, bodyString any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexPutMapping", reflect.TypeOf((*MockCLIClient)(nil).IndexPutMapping), ctx, indexName, bodyString)
}
// IndexPutTemplate mocks base method.
func (m *MockCLIClient) IndexPutTemplate(ctx context.Context, templateName, bodyString string) (bool, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "IndexPutTemplate", ctx, templateName, bodyString)
ret0, _ := ret[0].(bool)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// IndexPutTemplate indicates an expected call of IndexPutTemplate.
func (mr *MockCLIClientMockRecorder) IndexPutTemplate(ctx, templateName, bodyString any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "IndexPutTemplate", reflect.TypeOf((*MockCLIClient)(nil).IndexPutTemplate), ctx, templateName, bodyString)
}
// Ping mocks base method.
func (m *MockCLIClient) Ping(ctx context.Context) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Ping", ctx)
ret0, _ := ret[0].(error)
return ret0
}
// Ping indicates an expected call of Ping.
func (mr *MockCLIClientMockRecorder) Ping(ctx any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ping", reflect.TypeOf((*MockCLIClient)(nil).Ping), ctx)
}
// PutMapping mocks base method.
func (m *MockCLIClient) PutMapping(ctx context.Context, index string, mapping map[string]enums.IndexedValueType) (bool, error) {
m.ctrl.T.Helper()

52
common/persistence/visibility/store/elasticsearch/client/client_v7.go
View File

@@ -52,7 +52,10 @@ func newClient(cfg *Config, httpClient *http.Client, logger log.Logger) (*client
options = append(options, getLoggerOptions(cfg.LogLevel, logger)...)
if httpClient == nil {
if cfg.TLS != nil && cfg.TLS.Enabled {
// Check if httpClient is set in config (e.g., AWS HTTP client)
if configHTTPClient := cfg.GetHttpClient(); configHTTPClient != nil {
httpClient = configHTTPClient
} else if cfg.TLS != nil && cfg.TLS.Enabled {
tlsHttpClient, err := buildTLSHTTPClient(cfg.TLS)
if err != nil {
return nil, fmt.Errorf("unable to create TLS HTTP client: %w", err)
@@ -253,6 +256,53 @@ func (c *clientImpl) IndexPutTemplate(ctx context.Context, templateName string,
return resp.Acknowledged, nil
}
func (c *clientImpl) IndexPutMapping(ctx context.Context, indexName string, bodyString string) (bool, error) {
// Use raw HTTP request to update index mappings
path := fmt.Sprintf("/%s/_mapping", indexName)
resp, err := c.esClient.PerformRequest(ctx, elastic.PerformRequestOptions{
Method: "PUT",
Path: path,
Body: bodyString,
ContentType: "application/json",
})
if err != nil {
return false, err
}
// Parse the response to check if it was acknowledged
var result struct {
Acknowledged bool `json:"acknowledged"`
}
if err := json.Unmarshal(resp.Body, &result); err != nil {
return false, err
}
return result.Acknowledged, nil
}
func (c *clientImpl) ClusterPutSettings(ctx context.Context, bodyString string) (bool, error) {
// Use raw HTTP request since ClusterPutSettings is not available in olivere/elastic v7
resp, err := c.esClient.PerformRequest(ctx, elastic.PerformRequestOptions{
Method: "PUT",
Path: "/_cluster/settings",
Body: bodyString,
ContentType: "application/json",
})
if err != nil {
return false, err
}
// Parse the response to check if it was acknowledged
var result struct {
Acknowledged bool `json:"acknowledged"`
}
if err := json.Unmarshal(resp.Body, &result); err != nil {
return false, err
}
return result.Acknowledged, nil
}
func (c *clientImpl) IndexExists(ctx context.Context, indexName string) (bool, error) {
return c.esClient.IndexExists(indexName).Do(ctx)
}

18
schema/embed.go
View File

@@ -46,3 +46,21 @@ func PathsByDB(dbName string) []string {
}
return PathsByDir(dbName)
}
// ElasticsearchClusterSettings returns the embedded cluster settings for Elasticsearch v7
func ElasticsearchClusterSettings() (string, error) {
data, err := assets.ReadFile("elasticsearch/visibility/cluster_settings_v7.json")
if err != nil {
return "", err
}
return string(data), nil
}
// ElasticsearchIndexTemplate returns the embedded index template for Elasticsearch v7 (latest version)
func ElasticsearchIndexTemplate() (string, error) {
data, err := assets.ReadFile("elasticsearch/visibility/index_template_v7.json")
if err != nil {
return "", err
}
return string(data), nil
}

191
tools/elasticsearch/README.md Normal file
View File

@@ -0,0 +1,191 @@
## Using the Elasticsearch schema tool
⚠️ **EXPERIMENTAL**: This tool is experimental and may change in future versions.
This package contains the tooling for temporal elasticsearch operations.
### Usage
```
NAME:
temporal-elasticsearch-tool - Command line tool for temporal elasticsearch operations (EXPERIMENTAL)
USAGE:
temporal-elasticsearch-tool [global options] command [command options] [arguments...]
VERSION:
0.0.1
COMMANDS:
setup-schema setup elasticsearch cluster settings and index template
update-schema update elasticsearch index template, or both template and index mappings if --index is specified
create-index create elasticsearch visibility index
drop-index delete elasticsearch visibility index
ping pings the elasticsearch host
help, h Shows a list of commands or help for one command
GLOBAL OPTIONS:
--endpoint value hostname or ip address of elasticsearch server (default: "http://127.0.0.1:9200") [$ES_SERVER]
--user value username for elasticsearch or aws_access_key_id if using static aws credentials [$ES_USER]
--password value password for elasticsearch or aws_secret_access_key if using static aws credentials [$ES_PWD]
--aws-credentials value AWS credentials provider (supported ['static', 'environment', 'aws-sdk-default']) [$AWS_CREDENTIALS]
--aws-session-token value AWS sessiontoken for use with 'static' AWS credentials provider [$AWS_SESSION_TOKEN]
--index value name of the visibility index [$ES_VISIBILITY_INDEX]
--quiet don't log errors to stderr (default: false)
--help, -h show help
--version, -v print the version
```
## For localhost development
```
make install-schema-es
```
## For production
### Create the binaries
- Run `make temporal-elasticsearch-tool`
- You should see an executable `temporal-elasticsearch-tool`
### Schema setup
```
NAME:
temporal-elasticsearch-tool setup-schema - setup elasticsearch cluster settings and index template
USAGE:
temporal-elasticsearch-tool setup-schema [command options]
OPTIONS:
--fail fail silently on HTTP errors (default: false)
--help, -h show help
```
```
NAME:
temporal-elasticsearch-tool create-index - create elasticsearch visibility index
USAGE:
temporal-elasticsearch-tool create-index [command options]
OPTIONS:
--index value name of the visibility index to create
--fail fail silently on HTTP errors (default: false)
--help, -h show help
```
The tool now uses embedded schema files (cluster settings and index template) and provides separate commands for different operations. You can set up the schema and create indexes separately:
```
export ES_SERVER=http://127.0.0.1:9200
export ES_USER=$USER
export ES_PWD=$PWD
export ES_VISIBILITY_INDEX=temporal_visibility_v1
# Setup cluster settings and index template
temporal-elasticsearch-tool setup-schema
# Create the visibility index (uses ES_VISIBILITY_INDEX environment variable)
temporal-elasticsearch-tool create-index
# Or combine both operations
temporal-elasticsearch-tool setup-schema && \
temporal-elasticsearch-tool create-index
```
### Ping
Ping the ES server to ensure connectivity and successful authentication
```
export ES_SERVER=http://127.0.0.1:9200
export ES_VERSION=v7
export AWS_REGION=us-east-1
temporal-elasticsearch-tool --aws environment ping
```
### AWS Authentication
The CLI supports 3 AWS authentication mechanisms: `aws-sdk-default`, `environment`, `static`.
```
# aws-go-sdk defaults
export AWS_REGION=us-east-1
export ES_SERVER=http://127.0.0.1:9200
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool --aws aws-sdk-default setup-schema
temporal-elasticsearch-tool --aws aws-sdk-default create-index
```
```
# Environment Credentials
export AWS_REGION=us-east-1
export ES_SERVER=http://127.0.0.1:9200
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool --aws environment setup-schema
temporal-elasticsearch-tool --aws environment create-index
```
```
# Static Credentials
export AWS_REGION=us-east-1
export ES_SERVER=http://127.0.0.1:9200
export ES_USER=$AWS_ACCESS_KEY_ID
export ES_PWD=$AWS_SECRET_ACCESS_KEY
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool --aws static setup-schema
temporal-elasticsearch-tool --aws static create-index
```
```
# Static w/ Session Token
export AWS_REGION=us-east-1
export ES_SERVER=http://127.0.0.1:9200
export ES_USER=$AWS_ACCESS_KEY_ID
export ES_PWD=$AWS_SECRET_ACCESS_KEY
export AWS_SESSION_TOKEN
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool --aws static setup-schema
temporal-elasticsearch-tool --aws static create-index
```
### Additional Commands
#### Update Schema
Updates the index template to the latest version, or updates both the template and index mappings if `--index` is specified:
Update template only:
```bash
temporal-elasticsearch-tool update-schema
```
Update both template and index mappings:
```bash
temporal-elasticsearch-tool update-schema --index temporal_visibility_v1
```
Update both template and index mappings using environment variable:
```bash
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool update-schema
```
#### Drop Index
Deletes a visibility index:
```bash
temporal-elasticsearch-tool drop-index --index temporal_visibility_v1
```
Or using environment variable:
```bash
export ES_VISIBILITY_INDEX=temporal_visibility_v1
temporal-elasticsearch-tool drop-index
```
#### Command Summary
- `setup-schema`: Sets up cluster settings and index template (no index creation)
- `update-schema`: Updates index template, or both template and index mappings if --index is specified
- `create-index`: Creates a new visibility index (requires --index flag)
- `drop-index`: Deletes a visibility index (requires --index flag)
- `ping`: Tests connectivity to Elasticsearch server

227
tools/elasticsearch/handler.go Normal file
View File

@@ -0,0 +1,227 @@
package elasticsearch
import (
"context"
"errors"
"fmt"
"net/url"
"github.com/urfave/cli/v2"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
esclient "go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
"go.temporal.io/server/schema"
commonschema "go.temporal.io/server/tools/common/schema"
)
func createClient(cli *cli.Context, logger log.Logger) (esclient.CLIClient, error) {
cfg, err := parseElasticConfig(cli)
if err != nil {
logger.Error("Unable to parse elasticsearch config.", tag.Error(err))
return nil, err
}
if cfg.AWSRequestSigning.Enabled {
awsHTTPClient, err := esclient.NewAwsHttpClient(cfg.AWSRequestSigning)
if err != nil {
logger.Error("Unable to create AWS HTTP client.", tag.Error(err))
return nil, err
}
cfg.SetHttpClient(awsHTTPClient)
}
esClient, err := esclient.NewCLIClient(cfg, logger)
if err != nil {
logger.Error("Unable to create elasticsearch client.", tag.Error(err))
return nil, err
}
return esClient, nil
}
// setupSchema creates cluster settings and index template, but not the index
func setupSchema(cli *cli.Context, logger log.Logger) error {
client, err := createClient(cli, logger)
if err != nil {
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
return err
}
settingsContent, err := schema.ElasticsearchClusterSettings()
if err != nil {
logger.Error("Unable to load embedded cluster settings.", tag.Error(err))
return err
}
templateContent, err := schema.ElasticsearchIndexTemplate()
if err != nil {
logger.Error("Unable to load embedded index template.", tag.Error(err))
return err
}
task := SetupTask{
esClient: client,
logger: logger,
config: &SetupConfig{
SettingsContent: settingsContent,
TemplateContent: templateContent,
VisibilityIndex: "", // Don't create index in setup-schema
FailSilently: cli.Bool(CLIOptFailSilently),
},
}
return task.RunSchemaSetup()
}
// ping the elasticsearch host and return the json response
func ping(cli *cli.Context, logger log.Logger) error {
client, err := createClient(cli, logger)
if err != nil {
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
return err
}
err = client.Ping(context.TODO())
if err != nil {
logger.Error("Ping failed", tag.Error(err))
return err
}
logger.Info("Pong - Elasticsearch is reachable")
return nil
}
func parseElasticConfig(cli *cli.Context) (*esclient.Config, error) {
cfg := new(esclient.Config)
u, err := url.Parse(cli.String(CLIOptESURL))
if err != nil {
return nil, fmt.Errorf("invalid elasticsearch URL %q: %w", cli.String(CLIOptESURL), err)
}
cfg.URL = *u
cfg.Username = cli.String(CLIOptESUsername)
cfg.Password = cli.String(CLIOptESPassword)
cfg.Version = "v7" // Fixed schema version 7
cfg.Indices = map[string]string{}
if cli.String(CLIOptVisibilityIndex) != "" {
cfg.Indices[esclient.VisibilityAppName] = cli.String(CLIOptVisibilityIndex)
}
if cli.String(CLIOptAWSCredentials) != "" {
cfg.AWSRequestSigning.CredentialProvider = cli.String(CLIOptAWSCredentials)
cfg.AWSRequestSigning.Enabled = true
if cfg.AWSRequestSigning.CredentialProvider == "static" {
cfg.AWSRequestSigning.Static.AccessKeyID = cfg.Username
cfg.AWSRequestSigning.Static.SecretAccessKey = cfg.Password
cfg.AWSRequestSigning.Static.Token = cli.String(CLIOptAWSToken)
}
}
return cfg, nil
}
// updateSchema updates the index template to the latest version, or index mappings if --index is specified
func updateSchema(cli *cli.Context, logger log.Logger) error {
client, err := createClient(cli, logger)
if err != nil {
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
return err
}
templateContent, err := schema.ElasticsearchIndexTemplate()
if err != nil {
logger.Error("Unable to load embedded index template.", tag.Error(err))
return err
}
indexName := cli.String(CLIOptVisibilityIndex)
task := SetupTask{
esClient: client,
logger: logger,
config: &SetupConfig{
SettingsContent: "", // Don't update cluster settings
TemplateContent: templateContent,
VisibilityIndex: indexName,
FailSilently: cli.Bool(CLIOptFailSilently),
},
}
err = task.RunTemplateUpgrade()
if err != nil {
return err
}
if indexName != "" {
return task.RunIndexUpdate()
}
return nil
}
// createIndex creates a new visibility index
func createIndex(cli *cli.Context, logger log.Logger) error {
client, err := createClient(cli, logger)
if err != nil {
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
return err
}
task := SetupTask{
esClient: client,
logger: logger,
config: &SetupConfig{
SettingsContent: "", // Don't update cluster settings
TemplateContent: "", // Don't update template
VisibilityIndex: cli.String(CLIOptVisibilityIndex),
FailSilently: cli.Bool(CLIOptFailSilently),
},
}
return task.RunIndexCreation()
}
// dropIndex deletes a visibility index
func dropIndex(cli *cli.Context, logger log.Logger) error {
client, err := createClient(cli, logger)
if err != nil {
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
return err
}
indexName := cli.String(CLIOptVisibilityIndex)
if indexName == "" {
err := errors.New("index name is required")
logger.Error("Missing index name.", tag.Error(err))
return err
}
failSilently := cli.Bool(CLIOptFailSilently)
success, err := client.DeleteIndex(context.TODO(), indexName)
if err != nil {
if !failSilently {
logger.Error("Index deletion failed", tag.Error(err), tag.NewStringTag("indexName", indexName))
return err
}
logger.Warn("Index deletion failed", tag.Error(err), tag.NewStringTag("indexName", indexName))
return nil
} else if !success {
err := errors.New("acknowledged=false")
if !failSilently {
logger.Error("Index deletion failed without error", tag.Error(err), tag.NewStringTag("indexName", indexName))
return err
}
logger.Warn("Index deletion failed without error", tag.Error(err), tag.NewStringTag("indexName", indexName))
return nil
}
logger.Info("Index deleted successfully", tag.NewStringTag("indexName", indexName))
return nil
}
func flag(opt string) string {
return fmt.Sprintf("(--%s)", opt)
}

28
tools/elasticsearch/handler_test.go Normal file
View File

@@ -0,0 +1,28 @@
package elasticsearch
import (
"testing"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
)
type (
HandlerTestSuite struct {
*require.Assertions
suite.Suite
}
)
func TestHandlerTestSuite(t *testing.T) {
suite.Run(t, new(HandlerTestSuite))
}
func (s *HandlerTestSuite) SetupTest() {
s.Assertions = require.New(s.T())
}
func (s *HandlerTestSuite) TestFlag() {
result := flag("test-option")
s.Equal("(--test-option)", result)
}

179
tools/elasticsearch/main.go Normal file
View File

@@ -0,0 +1,179 @@
package elasticsearch
import (
"os"
"github.com/urfave/cli/v2"
"go.temporal.io/server/common/log"
"go.temporal.io/server/tools/common/schema"
)
const (
CLIOptESURL = "endpoint"
CLIOptESUsername = "user"
CLIOptESPassword = "password"
CLIOptVisibilityIndex = "index"
CLIOptAWSCredentials = "aws-credentials"
CLIOptAWSToken = "aws-session-token"
CLIOptFailSilently = "fail"
CLIFlagESURL = CLIOptESURL + ", e"
CLIFlagESUsername = CLIOptESUsername + ", u"
CLIFlagESPassword = CLIOptESPassword + ", p"
CLIFlagAWSToken = CLIOptAWSToken
CLIFlagVisibilityIndex = CLIOptVisibilityIndex + ", i"
CLIFlagAWSCredentials = CLIOptAWSCredentials + ", aws"
CLIFlagFailSilently = CLIOptFailSilently
)
// RunTool runs the temporal-elasticsearch-tool command line tool
func RunTool(args []string) error {
app := BuildCLIOptions()
return app.Run(args)
}
var osExit = os.Exit
// root handler for all cli commands
func cliHandler(c *cli.Context, handler func(c *cli.Context, logger log.Logger) error, logger log.Logger) {
quiet := c.Bool(schema.CLIOptQuiet)
err := handler(c, logger)
if err != nil && !quiet {
osExit(1)
}
}
// BuildCLIOptions builds the options for cli
func BuildCLIOptions() *cli.App {
app := cli.NewApp()
app.Name = "temporal-elasticsearch-tool"
app.Usage = "Command line tool for temporal elasticsearch operations (EXPERIMENTAL)"
app.Version = "0.0.1"
logger := log.NewCLILogger()
app.Flags = []cli.Flag{
&cli.StringFlag{
Name: CLIFlagESURL,
Value: "http://127.0.0.1:9200",
Usage: "hostname or ip address of elasticsearch server",
EnvVars: []string{"ES_SERVER"},
},
&cli.StringFlag{
Name: CLIFlagESUsername,
Value: "",
Usage: "username for elasticsearch or aws_access_key_id if using static aws credentials",
EnvVars: []string{"ES_USER"},
},
&cli.StringFlag{
Name: CLIFlagESPassword,
Value: "",
Usage: "password for elasticsearch or aws_secret_access_key if using static aws credentials",
EnvVars: []string{"ES_PWD"},
},
&cli.StringFlag{
Name: CLIFlagAWSCredentials,
Value: "",
Usage: "AWS credentials provider (supported ['static', 'environment', 'aws-sdk-default'])",
EnvVars: []string{"AWS_CREDENTIALS"},
},
&cli.StringFlag{
Name: CLIFlagAWSToken,
Value: "",
Usage: "AWS sessiontoken for use with 'static' AWS credentials provider",
EnvVars: []string{"AWS_SESSION_TOKEN"},
},
&cli.BoolFlag{
Name: schema.CLIOptQuiet,
Usage: "don't log errors to stderr",
},
}
app.Commands = []*cli.Command{
{
Name: "setup-schema",
Usage: "setup elasticsearch cluster settings and index template",
Flags: []cli.Flag{
&cli.BoolFlag{
Name: CLIFlagFailSilently,
Usage: "fail silently on HTTP errors",
},
},
Action: func(c *cli.Context) error {
cliHandler(c, setupSchema, logger)
return nil
},
},
{
Name: "update-schema",
Usage: "update elasticsearch index template, and index mappings if --index is specified",
Flags: []cli.Flag{
&cli.StringFlag{
Name: CLIFlagVisibilityIndex,
Usage: "name of the visibility index to update mappings for (optional)",
EnvVars: []string{"ES_VISIBILITY_INDEX"},
},
&cli.BoolFlag{
Name: CLIFlagFailSilently,
Usage: "fail silently on HTTP errors",
},
},
Action: func(c *cli.Context) error {
cliHandler(c, updateSchema, logger)
return nil
},
},
{
Name: "create-index",
Usage: "create elasticsearch visibility index",
Flags: []cli.Flag{
&cli.StringFlag{
Name: CLIFlagVisibilityIndex,
Usage: "name of the visibility index to create",
Required: true,
EnvVars: []string{"ES_VISIBILITY_INDEX"},
},
&cli.BoolFlag{
Name: CLIFlagFailSilently,
Usage: "fail silently on HTTP errors",
},
},
Action: func(c *cli.Context) error {
cliHandler(c, createIndex, logger)
return nil
},
},
{
Name: "drop-index",
Usage: "delete elasticsearch visibility index",
Flags: []cli.Flag{
&cli.StringFlag{
Name: CLIFlagVisibilityIndex,
Usage: "name of the visibility index to delete",
Required: true,
EnvVars: []string{"ES_VISIBILITY_INDEX"},
},
&cli.BoolFlag{
Name: CLIFlagFailSilently,
Usage: "fail silently on HTTP errors",
},
},
Action: func(c *cli.Context) error {
cliHandler(c, dropIndex, logger)
return nil
},
},
{
Name: "ping",
Usage: "pings the elasticsearch host",
Flags: []cli.Flag{},
Action: func(c *cli.Context) error {
cliHandler(c, ping, logger)
return nil
},
},
}
return app
}

50
tools/elasticsearch/main_test.go Normal file
View File

@@ -0,0 +1,50 @@
package elasticsearch
import (
"testing"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
)
type (
MainTestSuite struct {
*require.Assertions
suite.Suite
}
)
func TestMainTestSuite(t *testing.T) {
suite.Run(t, new(MainTestSuite))
}
func (s *MainTestSuite) SetupTest() {
s.Assertions = require.New(s.T())
}
// Test CLI error scenarios similar to cassandra tests
func (s *MainTestSuite) TestSetupSchemaError() {
// fake exit function to avoid exiting the application
back := osExit
defer func() { osExit = back }()
osExit = func(i int) {
s.Equal(1, i)
}
args := []string{"./tool", "setup-schema"}
app := BuildCLIOptions()
err := app.Run(args)
s.Nil(err)
}
func (s *MainTestSuite) TestPingError() {
// fake exit function to avoid exiting the application
back := osExit
defer func() { osExit = back }()
osExit = func(i int) {
s.Equal(1, i)
}
args := []string{"./tool", "--endpoint", "http://nonexistent:9200", "ping"}
app := BuildCLIOptions()
err := app.Run(args)
s.Nil(err)
}

224
tools/elasticsearch/tasks.go Normal file
View File

@@ -0,0 +1,224 @@
package elasticsearch
import (
"context"
"encoding/json"
"errors"
"fmt"
"go.temporal.io/server/common/log"
"go.temporal.io/server/common/log/tag"
"go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
)
const templateName = "temporal_visibility_v1_template"
type SetupConfig struct {
TemplateContent string
SettingsContent string
VisibilityIndex string
FailSilently bool
}
type SetupTask struct {
esClient client.CLIClient
config *SetupConfig
logger log.Logger
}
// Run executes the task
func (task *SetupTask) Run() error {
task.logger.Info("Starting schema setup", tag.NewAnyTag("config", task.config))
if err := task.setupClusterSettings(); err != nil {
task.logger.Error("Failed to setup cluster settings.", tag.Error(err))
return err
}
if err := task.setupTemplate(); err != nil {
task.logger.Error("Failed to setup template.", tag.Error(err))
return err
}
if err := task.setupIndex(); err != nil {
task.logger.Error("Failed to setup index.", tag.Error(err))
return err
}
task.logger.Info("Schema setup complete")
return nil
}
// setupClusterSettings handles cluster settings configuration
func (task *SetupTask) setupClusterSettings() error {
config := task.config
if len(config.SettingsContent) == 0 {
task.logger.Info("Skipping cluster settings update, no embedded settings content")
return nil
}
success, err := task.esClient.ClusterPutSettings(context.TODO(), config.SettingsContent)
if err != nil {
return task.handleOperationFailure("cluster settings update failed", err)
} else if !success {
return task.handleOperationFailure("cluster settings update failed without error", errors.New("acknowledged=false"))
}
task.logger.Info("Cluster settings updated successfully")
return nil
}
// setupTemplate handles template configuration
func (task *SetupTask) setupTemplate() error {
config := task.config
if len(config.TemplateContent) == 0 {
task.logger.Info("Skipping template creation, no embedded template content")
return nil
}
success, err := task.esClient.IndexPutTemplate(context.TODO(), templateName, config.TemplateContent)
if err != nil {
return task.handleOperationFailure("template creation failed", err)
} else if !success {
return task.handleOperationFailure("template creation failed without error", errors.New("acknowledged=false"))
}
task.logger.Info("Template created successfully", tag.NewStringTag("templateName", templateName))
return nil
}
// setupIndex handles index creation
func (task *SetupTask) setupIndex() error {
config := task.config
if len(config.VisibilityIndex) == 0 {
task.logger.Info("Skipping index creation, missing index name")
return nil
}
success, err := task.esClient.CreateIndex(context.TODO(), config.VisibilityIndex, nil)
if err != nil {
return task.handleOperationFailure("index creation failed", err)
} else if !success {
return task.handleOperationFailure("index creation failed without error", errors.New("acknowledged=false"))
}
task.logger.Info("Index created successfully", tag.NewStringTag("indexName", config.VisibilityIndex))
return nil
}
// RunSchemaSetup runs only cluster settings and template setup (no index creation)
func (task *SetupTask) RunSchemaSetup() error {
task.logger.Info("Starting schema setup (cluster settings and template)", tag.NewAnyTag("config", task.config))
if err := task.setupClusterSettings(); err != nil {
task.logger.Error("Failed to setup cluster settings.", tag.Error(err))
return err
}
if err := task.setupTemplate(); err != nil {
task.logger.Error("Failed to setup template.", tag.Error(err))
return err
}
task.logger.Info("Schema setup complete (cluster settings and template)")
return nil
}
// RunTemplateUpgrade runs only template upgrade
func (task *SetupTask) RunTemplateUpgrade() error {
task.logger.Info("Starting template upgrade", tag.NewAnyTag("config", task.config))
if err := task.setupTemplate(); err != nil {
task.logger.Error("Failed to upgrade template.", tag.Error(err))
return err
}
task.logger.Info("Template upgrade complete")
return nil
}
// RunIndexCreation runs only index creation
func (task *SetupTask) RunIndexCreation() error {
task.logger.Info("Starting index creation", tag.NewAnyTag("config", task.config))
if err := task.setupIndex(); err != nil {
task.logger.Error("Failed to create index.", tag.Error(err))
return err
}
task.logger.Info("Index creation complete")
return nil
}
// RunIndexUpdate updates the mappings of an existing index
func (task *SetupTask) RunIndexUpdate() error {
task.logger.Info("Starting index mapping update", tag.NewAnyTag("config", task.config))
if err := task.updateIndexMappings(); err != nil {
task.logger.Error("Failed to update index mappings.", tag.Error(err))
return err
}
task.logger.Info("Index mapping update complete")
return nil
}
// updateIndexMappings updates the mappings of an existing index using raw HTTP request
func (task *SetupTask) updateIndexMappings() error {
config := task.config
if len(config.VisibilityIndex) == 0 {
task.logger.Info("Skipping index mapping update, missing index name")
return nil
}
if len(config.TemplateContent) == 0 {
task.logger.Info("Skipping index mapping update, no embedded template content")
return nil
}
// Parse the template to extract mappings
var template map[string]interface{}
if err := json.Unmarshal([]byte(config.TemplateContent), &template); err != nil {
return fmt.Errorf("failed to parse template content: %w", err)
}
mappings, ok := template["mappings"]
if !ok {
return errors.New("no mappings found in template")
}
mappingsBytes, err := json.Marshal(mappings)
if err != nil {
return fmt.Errorf("failed to marshal mappings: %w", err)
}
// Check if the index exists first
indexName := config.VisibilityIndex
exists, err := task.esClient.IndexExists(context.TODO(), indexName)
if err != nil {
return task.handleOperationFailure("failed to check if index exists", err)
}
if !exists {
return task.handleOperationFailure("index does not exist", fmt.Errorf("index %s does not exist", indexName))
}
success, err := task.esClient.IndexPutMapping(context.TODO(), indexName, string(mappingsBytes))
if err != nil {
return task.handleOperationFailure("index mapping update failed", err)
} else if !success {
return task.handleOperationFailure("index mapping update failed without error", errors.New("acknowledged=false"))
}
task.logger.Info("Index mappings updated successfully", tag.NewStringTag("indexName", indexName))
return nil
}
// handleOperationFailure handles operation failures, optionally failing silently
func (task *SetupTask) handleOperationFailure(msg string, err error) error {
if !task.config.FailSilently {
task.logger.Error(msg, tag.Error(err))
return err
}
task.logger.Warn(msg, tag.Error(err))
return nil
}

18
tools/elasticsearch/tasks_test.go Normal file
View File

@@ -0,0 +1,18 @@
package elasticsearch
import (
"go.temporal.io/server/tools/common/schema/test"
)
type (
SetupSchemaTestSuite struct {
test.SetupSchemaTestBase
}
)
func (s *SetupSchemaTestSuite) TestSetupSchema() {
// Integration test similar to cassandra - this would require a real ES instance
// For now, just test that the CLI tool can be created
app := BuildCLIOptions()
s.NotNil(app)
}