go.temporal.io/server/tools/elasticsearch/tasks.go
210 LOC · 19 covered · 191 uncovered · 4 ranges · 1 concepts · 1 introducers · 1 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
"encoding/json"
6
"errors"
7
"fmt"
8
9
"github.com/olivere/elastic/v7"
10
"go.temporal.io/server/common/log"
11
"go.temporal.io/server/common/log/tag"
12
"go.temporal.io/server/common/persistence/visibility/store/elasticsearch/client"
13
)
14
15
const templateName = "temporal_visibility_v1_template"
16
17
type SetupConfig struct {
18
TemplateContent string
19
SettingsContent string
20
VisibilityIndex string
21
FailSilently bool
22
}
23
24
type SetupTask struct {
25
esClient client.CLIClient
26
config *SetupConfig
27
logger log.Logger
28
}
29
30
// setupClusterSettings handles cluster settings configuration
32
>
config := task.config
33
>
if len(config.SettingsContent) == 0 {
34
task.logger.Info("Skipping cluster settings update")
35
return nil
36
}
37
38
>
success, err := task.esClient.ClusterPutSettings(context.TODO(), config.SettingsContent)
handler.go ×4
39
>
if err != nil {
40
>
return task.handleOperationFailure("cluster settings update failed", err)
41
>
} else if !success {
42
return task.handleOperationFailure("cluster settings update failed without error", errors.New("acknowledged=false"))
43
}
44
45
task.logger.Info("Cluster settings updated successfully")
46
return nil
47
}
48
49
// setupTemplate handles template configuration
50
func (task *SetupTask) setupTemplate() error {
51
config := task.config
52
if len(config.TemplateContent) == 0 {
53
task.logger.Info("Skipping template creation, no embedded template content")
54
return nil
55
}
56
57
success, err := task.esClient.IndexPutTemplate(context.TODO(), templateName, config.TemplateContent)
58
if err != nil {
59
return task.handleOperationFailure("template creation failed", err)
60
} else if !success {
61
return task.handleOperationFailure("template creation failed without error", errors.New("acknowledged=false"))
62
}
63
64
task.logger.Info("Template created successfully", tag.String("templateName", templateName))
65
return nil
66
}
67
68
// setupIndex handles index creation. It checks if the index exists and skips creation if it does.
69
func (task *SetupTask) setupIndex(ctx context.Context) error {
70
config := task.config
71
if len(config.VisibilityIndex) == 0 {
72
task.logger.Info("Skipping index creation, missing index name")
73
return nil
74
}
75
76
success, err := task.esClient.CreateIndex(ctx, config.VisibilityIndex, nil)
77
if err != nil {
78
// Check if the error is an Elasticsearch error and if so check if the index already exists.
79
var esErr *elastic.Error
80
if errors.As(err, &esErr) {
81
if esErr.Status == 400 && esErr.Details != nil && esErr.Details.Type == "resource_already_exists_exception" {
82
task.logger.Info("Index already exists, skipping creation", tag.String("indexName", config.VisibilityIndex))
83
return nil
84
}
85
}
86
return task.handleOperationFailure("index creation failed", err)
87
} else if !success {
88
return task.handleOperationFailure("index creation failed without error", errors.New("acknowledged=false"))
89
}
90
91
task.logger.Info("Index created successfully", tag.String("indexName", config.VisibilityIndex))
92
return nil
93
}
94
95
// RunSchemaSetup runs only cluster settings and template setup (no index creation)
97
>
task.logger.Info("Starting schema setup (cluster settings and template)", tag.Any("config", task.config))
98
>
99
>
if err := task.setupClusterSettings(); err != nil {
100
>
task.logger.Error("Failed to setup cluster settings.", tag.Error(err))
101
>
return err
102
>
}
103
104
if err := task.setupTemplate(); err != nil {
105
task.logger.Error("Failed to setup template.", tag.Error(err))
106
return err
107
}
108
109
task.logger.Info("Schema setup complete (cluster settings and template)")
110
return nil
111
}
112
113
// RunTemplateUpgrade runs only template upgrade
114
func (task *SetupTask) RunTemplateUpgrade() error {
115
task.logger.Info("Starting template upgrade", tag.Any("config", task.config))
116
117
if err := task.setupTemplate(); err != nil {
118
task.logger.Error("Failed to upgrade template.", tag.Error(err))
119
return err
120
}
121
122
task.logger.Info("Template upgrade complete")
123
return nil
124
}
125
126
// RunIndexCreation runs only index creation
127
func (task *SetupTask) RunIndexCreation(ctx context.Context) error {
128
task.logger.Info("Starting index creation", tag.Any("config", task.config))
129
130
if err := task.setupIndex(ctx); err != nil {
131
task.logger.Error("Failed to create index.", tag.Error(err))
132
return err
133
}
134
135
task.logger.Info("Index creation complete")
136
return nil
137
}
138
139
// RunIndexUpdate updates the mappings of an existing index
140
func (task *SetupTask) RunIndexUpdate() error {
141
task.logger.Info("Starting index mapping update", tag.Any("config", task.config))
142
143
if err := task.updateIndexMappings(); err != nil {
144
task.logger.Error("Failed to update index mappings.", tag.Error(err))
145
return err
146
}
147
148
task.logger.Info("Index mapping update complete")
149
return nil
150
}
151
152
// updateIndexMappings updates the mappings of an existing index using raw HTTP request
153
func (task *SetupTask) updateIndexMappings() error {
154
config := task.config
155
if len(config.VisibilityIndex) == 0 {
156
task.logger.Info("Skipping index mapping update, missing index name")
157
return nil
158
}
159
160
if len(config.TemplateContent) == 0 {
161
task.logger.Info("Skipping index mapping update, no embedded template content")
162
return nil
163
}
164
165
// Parse the template to extract mappings
166
var template map[string]any
167
if err := json.Unmarshal([]byte(config.TemplateContent), &template); err != nil {
168
return fmt.Errorf("failed to parse template content: %w", err)
169
}
170
171
mappings, ok := template["mappings"]
172
if !ok {
173
return errors.New("no mappings found in template")
174
}
175
176
mappingsBytes, err := json.Marshal(mappings)
177
if err != nil {
178
return fmt.Errorf("failed to marshal mappings: %w", err)
179
}
180
181
// Check if the index exists first
182
indexName := config.VisibilityIndex
183
exists, err := task.esClient.IndexExists(context.TODO(), indexName)
184
if err != nil {
185
return task.handleOperationFailure("failed to check if index exists", err)
186
}
187
if !exists {
188
return task.handleOperationFailure("index does not exist", fmt.Errorf("index %s does not exist", indexName))
189
}
190
191
success, err := task.esClient.IndexPutMapping(context.TODO(), indexName, string(mappingsBytes))
192
if err != nil {
193
return task.handleOperationFailure("index mapping update failed", err)
194
} else if !success {
195
return task.handleOperationFailure("index mapping update failed without error", errors.New("acknowledged=false"))
196
}
197
198
task.logger.Info("Index mappings updated successfully", tag.String("indexName", indexName))
199
return nil
200
}
201
202
// handleOperationFailure handles operation failures, optionally failing silently
204
>
if !task.config.FailSilently {
205
>
task.logger.Error(msg, tag.Error(err))
206
>
return err
207
>
}
208
task.logger.Warn(msg, tag.Error(err))
209
return nil
210
}