Skip to content

Commit 1c16980

Browse files
siddharthlatestsiddharthlatest
committed
feat: add client-side rollover for Elasticsearch Serverless
Serverless rejects rollover conditions in the API body, so evaluate max_age/max_docs/max_size client-side and call rollover without conditions. Keeps the existing @midnight/@hourly cron and post-rollover index cleanup unchanged.
1 parent f4cd08f commit 1c16980

8 files changed

Lines changed: 422 additions & 72 deletions

File tree

‎docs/env-vars.md‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,17 @@ Plugins might require certain environment variables to be in order initialize th
66

77
**Meta index naming:** ReactiveSearch stores its metadata in dot-prefixed indices (e.g. `.pipelines`, `.users`). Elasticsearch Serverless does not allow creating dot-prefixed indices, so when a Serverless cluster is detected (via `build_flavor` from `GET /`), meta indices are automatically created with the `rs_` prefix instead (e.g. `rs_pipelines`). Set `RS_META_INDEX_PREFIX` to force an alternate prefix on any cluster (the prefix must not start with `_`, `-` or `+`). On Serverless, platform-managed index settings (`index.hidden`, shard/replica counts) are also stripped from index creation requests.
88

9+
**Serverless rollover:** Elasticsearch Serverless rejects conditional index rollover (`max_age`, `max_docs`, `max_size`) with `rollover with conditions is not supported in serverless mode`. On Serverless clusters, ReactiveSearch evaluates those thresholds **client-side** before calling the rollover API without conditions. Rollover runs when any threshold is met (OR semantics), same as on self-managed clusters. After rollover, old backing indices beyond the latest two are deleted as usual.
10+
11+
| Meta index (Serverless default) | `max_age` (non-production) | `max_age` (production plan) |
12+
|---------------------------------|----------------------------|-----------------------------|
13+
| `rs_analytics` | 30 days | 30 days |
14+
| `rs_logs` | 7 days | 30 days |
15+
| `rs_pipeline_logs` | 7 days | 30 days |
16+
| `rs_pipeline_invocations` | 3 days | 7 days |
17+
18+
Rollover cron jobs run at `@midnight` and `@hourly`. On Serverless, expect `serverless rollover skipped, conditions not met` when thresholds are not yet satisfied; when rollover does run, logs include `rollover res oldIndex`, `rollover res newIndex`, and `rollover res isRolledover`. On non-Serverless clusters, conditional rollover is sent in the API request and post-rollover index cleanup (keeping the latest two backing indices) is unchanged.
19+
920
List of specific env vars required by respective plugins are listed below:
1021

1122
##### 1. Users

‎plugins/analytics/dao.go‎

Lines changed: 33 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -3043,25 +3043,38 @@ func (es *elasticsearch) rolloverIndexJob(alias string) {
30433043
}
30443044

30453045
json.Unmarshal([]byte(rolloverConfiguration), &rolloverConditions)
3046-
settingsString := util.AdaptIndexBody(fmt.Sprintf(`{%s "index.number_of_shards": 2, "index.number_of_replicas": %d}`, util.HiddenIndexSettings(), util.GetReplicas()))
3047-
settings := make(map[string]interface{})
3048-
json.Unmarshal([]byte(settingsString), &settings)
3049-
rolloverService, err := es7.NewIndicesRolloverService(util.GetClient7()).
3050-
Alias(alias).
3051-
Conditions(rolloverConditions).
3052-
Settings(settings).
3053-
Mappings(mappings).
3054-
Do(ctx)
3055-
if err != nil {
3056-
log.Printf("%s: error while creating a rollover service %s %v", logTag, alias, err)
3046+
3047+
shouldRollover := true
3048+
if util.IsServerless() {
3049+
var conditionErr error
3050+
shouldRollover, conditionErr = util.WriteIndexMeetsRolloverConditions(ctx, alias, rolloverConditions)
3051+
if conditionErr != nil {
3052+
log.Errorln(logTag, ": serverless rollover condition check error, skipping rollover", conditionErr)
3053+
shouldRollover = false
3054+
} else if !shouldRollover {
3055+
log.Println(logTag, ": serverless rollover skipped, conditions not met for alias", alias)
3056+
}
30573057
}
3058-
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
3059-
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
3060-
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
30613058

3062-
if rolloverService.RolledOver {
3063-
classify.SetIndexAlias(rolloverService.NewIndex, alias)
3064-
classify.SetAliasIndex(alias, rolloverService.NewIndex)
3059+
if shouldRollover {
3060+
rolloverSvc := util.NewIndicesRolloverService(alias, rolloverConditions).
3061+
Mappings(mappings)
3062+
if settings := util.RolloverIndexSettings(2); len(settings) > 0 {
3063+
rolloverSvc = rolloverSvc.Settings(settings)
3064+
}
3065+
rolloverService, err := rolloverSvc.Do(ctx)
3066+
if err != nil {
3067+
log.Printf("%s: error while creating a rollover service %s %v", logTag, alias, err)
3068+
return
3069+
}
3070+
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
3071+
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
3072+
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
3073+
3074+
if rolloverService.RolledOver {
3075+
classify.SetIndexAlias(rolloverService.NewIndex, alias)
3076+
classify.SetAliasIndex(alias, rolloverService.NewIndex)
3077+
}
30653078
}
30663079

30673080
// We cannot rely on rollover service response here,
@@ -3081,9 +3094,9 @@ func (es *elasticsearch) rolloverIndexJob(alias string) {
30813094
rolloverIndices = rolloverIndices[:len(rolloverIndices)-2]
30823095

30833096
log.Println(logTag, ": rollover cronjob, indices to delete", rolloverIndices)
3084-
_, err = util.GetClient7().DeleteIndex(strings.Join(rolloverIndices, ",")).Do(ctx)
3085-
if err != nil {
3086-
log.Errorln(logTag, ": rollover cronjob, error while deleting indices", err)
3097+
_, deleteErr := util.GetClient7().DeleteIndex(strings.Join(rolloverIndices, ",")).Do(ctx)
3098+
if deleteErr != nil {
3099+
log.Errorln(logTag, ": rollover cronjob, error while deleting indices", deleteErr)
30873100
}
30883101
}
30893102
}

‎plugins/logs/dao.go‎

Lines changed: 31 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -148,15 +148,13 @@ func (es *elasticsearch) getRawLog(ctx context.Context, ID string, parseDiffs bo
148148

149149
func (es *elasticsearch) rolloverIndexJob(alias string) {
150150
ctx := context.Background()
151+
151152
rolloverConditions := make(map[string]interface{})
152153
rolloverConfiguration := fmt.Sprintf(rolloverConfig, "7d", 10000, "1gb")
153154
if util.IsProductionPlan() {
154155
rolloverConfiguration = fmt.Sprintf(rolloverConfig, "30d", 1000000, "10gb")
155156
}
156157
json.Unmarshal([]byte(rolloverConfiguration), &rolloverConditions)
157-
settingsString := util.AdaptIndexBody(fmt.Sprintf(`{%s "index.number_of_shards": 2, "index.number_of_replicas": %d}`, util.HiddenIndexSettings(), util.GetReplicas()))
158-
settings := make(map[string]interface{})
159-
json.Unmarshal([]byte(settingsString), &settings)
160158

161159
mappingString := LogsMappings
162160
if util.GetVersion() == 6 {
@@ -165,22 +163,38 @@ func (es *elasticsearch) rolloverIndexJob(alias string) {
165163

166164
mappings := make(map[string]interface{})
167165
json.Unmarshal([]byte(mappingString), &mappings)
168-
rolloverService, err := es7.NewIndicesRolloverService(util.GetClient7()).
169-
Alias(alias).
170-
Conditions(rolloverConditions).
171-
Settings(settings).
172-
Mappings(mappings).
173-
Do(ctx)
174-
if err != nil {
175-
log.Println(logTag, "error while creating a rollover service", alias, err)
166+
167+
shouldRollover := true
168+
if util.IsServerless() {
169+
var conditionErr error
170+
shouldRollover, conditionErr = util.WriteIndexMeetsRolloverConditions(ctx, alias, rolloverConditions)
171+
if conditionErr != nil {
172+
log.Errorln(logTag, ": serverless rollover condition check error, skipping rollover", conditionErr)
173+
shouldRollover = false
174+
} else if !shouldRollover {
175+
log.Println(logTag, ": serverless rollover skipped, conditions not met for alias", alias)
176+
}
176177
}
177-
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
178-
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
179-
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
180178

181-
if rolloverService.RolledOver {
182-
classify.SetIndexAlias(rolloverService.NewIndex, alias)
183-
classify.SetAliasIndex(alias, rolloverService.NewIndex)
179+
if shouldRollover {
180+
rolloverSvc := util.NewIndicesRolloverService(alias, rolloverConditions).
181+
Mappings(mappings)
182+
if settings := util.RolloverIndexSettings(2); len(settings) > 0 {
183+
rolloverSvc = rolloverSvc.Settings(settings)
184+
}
185+
rolloverService, err := rolloverSvc.Do(ctx)
186+
if err != nil {
187+
log.Println(logTag, "error while creating a rollover service", alias, err)
188+
return
189+
}
190+
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
191+
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
192+
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
193+
194+
if rolloverService.RolledOver {
195+
classify.SetIndexAlias(rolloverService.NewIndex, alias)
196+
classify.SetAliasIndex(alias, rolloverService.NewIndex)
197+
}
184198
}
185199

186200
// We cannot rely on rollover service response here,

‎plugins/pipelines/analytics.go‎

Lines changed: 31 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -794,37 +794,50 @@ func withPipelineIDFallback(ctx context.Context, executor func(string) (*es7.Sea
794794
// in ElasticSearch.
795795
func (es *invocationElasticsearch) rolloverIndexJob(alias string) {
796796
ctx := context.Background()
797+
797798
rolloverConditions := make(map[string]interface{})
798799
rolloverConfiguration := fmt.Sprintf(rolloverConfig, "3d", 100000, "1gb")
799800
if util.IsProductionPlan() {
800801
rolloverConfiguration = fmt.Sprintf(rolloverConfig, "7d", 10000000, "10gb")
801802
}
802803
json.Unmarshal([]byte(rolloverConfiguration), &rolloverConditions)
803-
settingsString := util.AdaptIndexBody(fmt.Sprintf(`{%s "index.number_of_shards": 3, "index.number_of_replicas": %d}`, util.HiddenIndexSettings(), util.GetReplicas()))
804-
settings := make(map[string]interface{})
805-
json.Unmarshal([]byte(settingsString), &settings)
806804

807805
mappingString := pipelineInvocationMapping
808806

809807
mappings := make(map[string]interface{})
810808
json.Unmarshal([]byte(mappingString), &mappings)
811-
rolloverService, err := es7.NewIndicesRolloverService(util.GetClient7()).
812-
Alias(alias).
813-
Conditions(rolloverConditions).
814-
Settings(settings).
815-
Mappings(mappings).
816-
Do(ctx)
817-
if err != nil {
818-
log.Println(logTag, "error while creating a rollover service", alias, err)
819-
return
809+
810+
shouldRollover := true
811+
if util.IsServerless() {
812+
var conditionErr error
813+
shouldRollover, conditionErr = util.WriteIndexMeetsRolloverConditions(ctx, alias, rolloverConditions)
814+
if conditionErr != nil {
815+
log.Errorln(logTag, ": serverless rollover condition check error, skipping rollover", conditionErr)
816+
shouldRollover = false
817+
} else if !shouldRollover {
818+
log.Println(logTag, ": serverless rollover skipped, conditions not met for alias", alias)
819+
}
820820
}
821-
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
822-
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
823-
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
824821

825-
if rolloverService.RolledOver {
826-
classify.SetIndexAlias(rolloverService.NewIndex, alias)
827-
classify.SetAliasIndex(alias, rolloverService.NewIndex)
822+
if shouldRollover {
823+
rolloverSvc := util.NewIndicesRolloverService(alias, rolloverConditions).
824+
Mappings(mappings)
825+
if settings := util.RolloverIndexSettings(3); len(settings) > 0 {
826+
rolloverSvc = rolloverSvc.Settings(settings)
827+
}
828+
rolloverService, err := rolloverSvc.Do(ctx)
829+
if err != nil {
830+
log.Println(logTag, "error while creating a rollover service", alias, err)
831+
return
832+
}
833+
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
834+
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
835+
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
836+
837+
if rolloverService.RolledOver {
838+
classify.SetIndexAlias(rolloverService.NewIndex, alias)
839+
classify.SetAliasIndex(alias, rolloverService.NewIndex)
840+
}
828841
}
829842

830843
// We cannot rely on rollover service response here,

‎plugins/pipelines/logs.go‎

Lines changed: 31 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -239,36 +239,50 @@ func (es *logsElasticsearch) getPipelineLogById(ctx context.Context, logId strin
239239
// in ElasticSearch.
240240
func (es *logsElasticsearch) rolloverIndexJob(alias string) {
241241
ctx := context.Background()
242+
242243
rolloverConditions := make(map[string]interface{})
243244
rolloverConfiguration := fmt.Sprintf(rolloverConfig, "7d", 10000, "1gb")
244245
if util.IsProductionPlan() {
245246
rolloverConfiguration = fmt.Sprintf(rolloverConfig, "30d", 1000000, "10gb")
246247
}
247248
json.Unmarshal([]byte(rolloverConfiguration), &rolloverConditions)
248-
settingsString := util.AdaptIndexBody(fmt.Sprintf(`{%s "index.number_of_shards": 2, "index.number_of_replicas": %d}`, util.HiddenIndexSettings(), util.GetReplicas()))
249-
settings := make(map[string]interface{})
250-
json.Unmarshal([]byte(settingsString), &settings)
251249

252250
mappingString := pipelineLogsMapping
253251

254252
mappings := make(map[string]interface{})
255253
json.Unmarshal([]byte(mappingString), &mappings)
256-
rolloverService, err := es7.NewIndicesRolloverService(util.GetClient7()).
257-
Alias(alias).
258-
Conditions(rolloverConditions).
259-
Settings(settings).
260-
Mappings(mappings).
261-
Do(ctx)
262-
if err != nil {
263-
log.Println(logTag, "error while creating a rollover service", alias, err)
254+
255+
shouldRollover := true
256+
if util.IsServerless() {
257+
var conditionErr error
258+
shouldRollover, conditionErr = util.WriteIndexMeetsRolloverConditions(ctx, alias, rolloverConditions)
259+
if conditionErr != nil {
260+
log.Errorln(logTag, ": serverless rollover condition check error, skipping rollover", conditionErr)
261+
shouldRollover = false
262+
} else if !shouldRollover {
263+
log.Println(logTag, ": serverless rollover skipped, conditions not met for alias", alias)
264+
}
264265
}
265-
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
266-
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
267-
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
268266

269-
if rolloverService.RolledOver {
270-
classify.SetIndexAlias(rolloverService.NewIndex, alias)
271-
classify.SetAliasIndex(alias, rolloverService.NewIndex)
267+
if shouldRollover {
268+
rolloverSvc := util.NewIndicesRolloverService(alias, rolloverConditions).
269+
Mappings(mappings)
270+
if settings := util.RolloverIndexSettings(2); len(settings) > 0 {
271+
rolloverSvc = rolloverSvc.Settings(settings)
272+
}
273+
rolloverService, err := rolloverSvc.Do(ctx)
274+
if err != nil {
275+
log.Println(logTag, "error while creating a rollover service", alias, err)
276+
return
277+
}
278+
log.Println(logTag, ": rollover res oldIndex", rolloverService.OldIndex)
279+
log.Println(logTag, ": rollover res newIndex", rolloverService.NewIndex)
280+
log.Println(logTag, ": rollover res isRolledover", rolloverService.RolledOver)
281+
282+
if rolloverService.RolledOver {
283+
classify.SetIndexAlias(rolloverService.NewIndex, alias)
284+
classify.SetAliasIndex(alias, rolloverService.NewIndex)
285+
}
272286
}
273287

274288
// We cannot rely on rollover service response here,

‎util/meta_index.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package util
33
import (
44
"context"
55
"encoding/json"
6+
"fmt"
67
"net/http"
78
"os"
89
"strings"
@@ -153,3 +154,18 @@ func stripUnsupportedSettings(settings map[string]interface{}) {
153154
}
154155
}
155156
}
157+
158+
// RolloverIndexSettings returns index settings for a rollover new-index template,
159+
// adapted for the target cluster. On Elasticsearch Serverless, platform-managed
160+
// settings (shards, replicas, hidden) are stripped via AdaptIndexBody. Rollover
161+
// conditions (max_age, max_docs, max_size) must be evaluated client-side on
162+
// Serverless via WriteIndexMeetsRolloverConditions before calling the rollover API.
163+
func RolloverIndexSettings(numberOfShards int) map[string]interface{} {
164+
body := AdaptIndexBody(fmt.Sprintf(`{%s "index.number_of_shards": %d, "index.number_of_replicas": %d}`, HiddenIndexSettings(), numberOfShards, GetReplicas()))
165+
settings := make(map[string]interface{})
166+
if err := json.Unmarshal([]byte(body), &settings); err != nil {
167+
log.Warnln("error while parsing rollover index settings, using empty settings: ", err)
168+
return map[string]interface{}{}
169+
}
170+
return settings
171+
}

0 commit comments

Comments
 (0)