go.temporal.io/server/tools/elasticsearch/handler.go
242 LOC · 54 covered · 188 uncovered · 16 ranges · 4 concepts · 4 introducers · 3 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package elasticsearch
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"net/url"
8
9
"github.com/urfave/cli"
10
"go.temporal.io/server/common/auth"
11
"go.temporal.io/server/common/log"
12
"go.temporal.io/server/common/log/tag"
13
esclient "go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
14
"go.temporal.io/server/schema"
15
commonschema "go.temporal.io/server/tools/common/schema"
16
)
17
18
>
func createClient(cli *cli.Context, logger log.Logger) (esclient.CLIClient, error) {
client_v7.go ×10
19
>
cfg, err := parseElasticConfig(cli)
20
>
if err != nil {
21
logger.Error("Unable to parse elasticsearch config.", tag.Error(err))
22
return nil, err
23
}
24
26
awsHTTPClient, err := esclient.NewAwsHttpClient(cfg.AWSRequestSigning)
27
if err != nil {
28
logger.Error("Unable to create AWS HTTP client.", tag.Error(err))
29
return nil, err
30
}
31
cfg.SetHttpClient(awsHTTPClient)
32
}
33
35
>
if err != nil {
36
logger.Error("Unable to create elasticsearch client.", tag.Error(err))
37
return nil, err
38
}
39
41
}
42
43
// setupSchema creates cluster settings and index template, but not the index
45
>
client, err := createClient(cli, logger)
46
>
if err != nil {
47
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
48
return err
49
}
50
52
>
if !cli.Bool(CLIOptSkipClusterSettings) {
53
>
settingsContent, err = schema.ElasticsearchClusterSettings()
54
>
if err != nil {
55
logger.Error("Unable to load embedded cluster settings.", tag.Error(err))
56
return err
57
}
58
}
59
61
>
if err != nil {
62
logger.Error("Unable to load embedded index template.", tag.Error(err))
63
return err
64
}
65
67
>
esClient: client,
68
>
logger: logger,
69
>
config: &SetupConfig{
70
>
SettingsContent: settingsContent,
71
>
TemplateContent: templateContent,
72
>
VisibilityIndex: "", // Don't create index in setup-schema
73
>
FailSilently: cli.Bool(CLIOptFailSilently),
74
>
},
75
>
}
76
>
77
>
return task.RunSchemaSetup()
78
}
79
80
// ping the elasticsearch host and return the json response
82
>
client, err := createClient(cli, logger)
83
>
if err != nil {
84
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
85
return err
86
}
87
89
>
if err != nil {
90
>
logger.Error("Ping failed", tag.Error(err))
91
>
return err
92
>
}
93
94
logger.Info("Pong - Elasticsearch is reachable")
95
return nil
96
}
97
99
>
cfg := new(esclient.Config)
100
>
101
>
u, err := url.Parse(cli.GlobalString(commonschema.CLIOptEndpoint))
102
>
if err != nil {
103
return nil, fmt.Errorf("invalid elasticsearch URL %q: %w", cli.GlobalString(commonschema.CLIOptEndpoint), err)
104
}
105
107
>
cfg.Username = cli.GlobalString(commonschema.CLIOptUser)
108
>
cfg.Password = cli.GlobalString(commonschema.CLIOptPassword)
109
>
cfg.Version = "v7" // Fixed schema version 7
110
>
cfg.Indices = map[string]string{}
111
>
112
>
if cli.String(CLIOptVisibilityIndex) != "" {
113
cfg.Indices[esclient.VisibilityAppName] = cli.String(CLIOptVisibilityIndex)
114
}
115
117
cfg.AWSRequestSigning.CredentialProvider = cli.GlobalString(CLIOptAWSCredentials)
118
cfg.AWSRequestSigning.Enabled = true
119
120
if cfg.AWSRequestSigning.CredentialProvider == "static" {
121
cfg.AWSRequestSigning.Static.AccessKeyID = cfg.Username
122
cfg.AWSRequestSigning.Static.SecretAccessKey = cfg.Password
123
cfg.AWSRequestSigning.Static.Token = cli.GlobalString(CLIOptAWSToken)
124
}
125
}
126
128
cfg.TLS = &auth.TLS{
129
Enabled: true,
130
CertFile: cli.GlobalString(commonschema.CLIFlagTLSCertFile),
131
KeyFile: cli.GlobalString(commonschema.CLIFlagTLSKeyFile),
132
CaFile: cli.GlobalString(commonschema.CLIFlagTLSCaFile),
133
ServerName: cli.GlobalString(commonschema.CLIFlagTLSHostName),
134
EnableHostVerification: !cli.GlobalBool(commonschema.CLIFlagTLSDisableHostVerification),
135
}
136
}
137
139
}
140
141
// updateSchema updates the index template to the latest version, or index mappings if --index is specified
142
func updateSchema(cli *cli.Context, logger log.Logger) error {
143
client, err := createClient(cli, logger)
144
if err != nil {
145
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
146
return err
147
}
148
149
templateContent, err := schema.ElasticsearchIndexTemplate()
150
if err != nil {
151
logger.Error("Unable to load embedded index template.", tag.Error(err))
152
return err
153
}
154
155
indexName := cli.String(CLIOptVisibilityIndex)
156
157
task := SetupTask{
158
esClient: client,
159
logger: logger,
160
config: &SetupConfig{
161
SettingsContent: "", // Don't update cluster settings
162
TemplateContent: templateContent,
163
VisibilityIndex: indexName,
164
FailSilently: cli.Bool(CLIOptFailSilently),
165
},
166
}
167
168
err = task.RunTemplateUpgrade()
169
if err != nil {
170
return err
171
}
172
173
if indexName != "" {
174
return task.RunIndexUpdate()
175
}
176
return nil
177
}
178
179
// createIndex creates a new visibility index
180
func createIndex(cli *cli.Context, logger log.Logger) error {
181
client, err := createClient(cli, logger)
182
if err != nil {
183
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
184
return err
185
}
186
187
task := SetupTask{
188
esClient: client,
189
logger: logger,
190
config: &SetupConfig{
191
SettingsContent: "", // Don't update cluster settings
192
TemplateContent: "", // Don't update template
193
VisibilityIndex: cli.String(CLIOptVisibilityIndex),
194
FailSilently: cli.Bool(CLIOptFailSilently),
195
},
196
}
197
198
return task.RunIndexCreation(context.Background())
199
}
200
201
// dropIndex deletes a visibility index
202
func dropIndex(cli *cli.Context, logger log.Logger) error {
203
client, err := createClient(cli, logger)
204
if err != nil {
205
logger.Error("Unable to read config.", tag.Error(commonschema.NewConfigError(err.Error())))
206
return err
207
}
208
209
indexName := cli.String(CLIOptVisibilityIndex)
210
if indexName == "" {
211
err := errors.New("index name is required")
212
logger.Error("Missing index name.", tag.Error(err))
213
return err
214
}
215
216
failSilently := cli.Bool(CLIOptFailSilently)
217
218
success, err := client.DeleteIndex(context.TODO(), indexName)
219
if err != nil {
220
if !failSilently {
221
logger.Error("Index deletion failed", tag.Error(err), tag.String("indexName", indexName))
222
return err
223
}
224
logger.Warn("Index deletion failed", tag.Error(err), tag.String("indexName", indexName))
225
return nil
226
} else if !success {
227
err := errors.New("acknowledged=false")
228
if !failSilently {
229
logger.Error("Index deletion failed without error", tag.Error(err), tag.String("indexName", indexName))
230
return err
231
}
232
logger.Warn("Index deletion failed without error", tag.Error(err), tag.String("indexName", indexName))
233
return nil
234
}
235
236
logger.Info("Index deleted successfully", tag.String("indexName", indexName))
237
return nil
238
}
239
241
>
return fmt.Sprintf("(--%s)", opt)
242
>
}