Add feature gate for ES data stream migration

Implement jaeger.es.readLegacyWithDataStream feature gate and enhance ILM policy with forcemerge.

Signed-off-by: SoumyaRaikwar <somuraik@gmail.com>

Author: SoumyaRaikwar

docs/design/elasticsearch-data-streams.md

@@ -21,28 +21,94 @@ This prefixing strategy prevents legacy wildcard queries like `jaeger-span-*` fr
 
 ### Index Templates
 Data stream-specific index templates are provided in `internal/storage/v1/elasticsearch/mappings/`:
-- `jaeger-ds-span-8.json`
-- `jaeger-ds-service-8.json`
-- `jaeger-ds-sampling-8.json`
-- `jaeger-ds-dependencies-8.json`
+- `jaeger-span-8.json` (merged with standard template)
+- `jaeger-service-8.json`
+- `jaeger-sampling-8.json`
+- `jaeger-dependencies-8.json`
 
 These templates include the `data_stream: {}` field and specify a default ingest pipeline for timestamp normalization (e.g., `jaeger-ds-span-timestamp`).
 
 ### Storage Logic
 The storage stores (Span, Sampling, Dependency) are updated to:
-1.  **Write** to the data stream name when `UseDataStream` is enabled.
-2.  **Read** from both the data stream and the legacy wildcard patterns to ensure data visibility during the migration period.
+1.  **Write** to the data stream name when `ES Version >= 8`.
+2.  **Read** behavior is controlled by the `jaeger.es.readLegacyWithDataStream` feature gate:
+    - When **enabled** (default during migration): Queries both data streams and legacy indices (e.g., `["jaeger-ds-span", "jaeger-span-*"]`)
+    - When **disabled**: Queries only data streams (e.g., `["jaeger-ds-span"]`)
+    
+This feature gate allows operators to safely migrate from legacy indices to data streams and disable dual reads once migration is complete.
 
 ## Migration Strategy
 Jaeger supports a seamless migration from traditional indices to data streams:
-- **Phase 1**: Enable `UseDataStream`. Jaeger will start writing to the new `jaeger-ds-*` data streams.
-- **Phase 2**: During read operations, Jaeger queries both the new data streams and the legacy indices.
-- **Phase 3**: Once legacy indices are aged out and deleted, only data streams will be used.
+- **Phase 1**: Enable Data Streams (Default for ES 8+). Jaeger will start writing to the new `jaeger-ds-*` data streams.
+- **Phase 2**: During read operations, the `jaeger.es.readLegacyWithDataStream` feature gate (enabled by default) ensures Jaeger queries both the new data streams and the legacy indices.
+- **Phase 3**: Once legacy indices are aged out and deleted, disable the feature gate to query only data streams:
+  ```bash
+  --feature-gates=-jaeger.es.readLegacyWithDataStream
+  ```
 
 ## Configuration
 The following configuration options control data stream usage:
-- `UseDataStream`: Boolean flag to enable data stream support (requires Elasticsearch 8+).
+- `ES Version`: Data streams are automatically enabled for Elasticsearch 8+.
 - `IndexPrefix`: Optional prefix for all indices and data streams.
+- `jaeger.es.readLegacyWithDataStream`: Feature gate to control dual reading (enabled by default for migration support).
+
+## ILM Policy Proposal
+To manage the lifecycle of data streams effectively, we propose the following Index Lifecycle Management (ILM) policy, named `jaeger-ilm-policy`.
+
+### Policy Definition
+The policy defines three phases to optimize storage costs and performance:
+
+1.  **Hot Phase**:
+    -   **Rollover**: Occurs when the index reaches **50GB**, **200 million documents**, or **1 day** of age.
+    -   **Priority**: Set to **100** (Highest).
+2.  **Warm Phase**:
+    -   **Transition**: Moves to warm phase **0ms** after rollover (effectively immediately upon leaving hot).
+    -   **Forcemerge**: Reduces segment count to 1 for better search performance and reduced overhead.
+    -   **Priority**: Set to **50**.
+3.  **Delete Phase**:
+    -   **Action**: Delete indices **7 days** after rollover.
+
+### JSON Representation
+```json
+{
+  "policy": {
+    "phases": {
+      "hot": {
+        "min_age": "0ms",
+        "actions": {
+          "rollover": {
+            "max_age": "1d",
+            "max_primary_shard_size": "50gb",
+            "max_docs": 200000000
+          },
+          "set_priority": {
+            "priority": 100
+          }
+        }
+      },
+      "warm": {
+        "min_age": "0ms",
+        "actions": {
+          "forcemerge": {
+            "max_num_segments": 1
+          },
+          "set_priority": {
+            "priority": 50
+          }
+        }
+      },
+      "delete": {
+        "min_age": "7d",
+        "actions": {
+          "delete": {
+            "delete_searchable_snapshot": true
+          }
+        }
+      }
+    }
+  }
+}
+```
 
 ## Verification & Evidence
 
@@ -74,10 +140,10 @@ curl -X PUT "localhost:9200/_ingest/pipeline/jaeger-ds-span-timestamp" -H 'Conte
 
 ### 2. Testing Steps
 1. **Prerequisites**: Elasticsearch 8.x and Jaeger built from the data stream support branch.
-2. **Configuration**: Set `use_data_stream: true` and `create_mappings: true` in the Jaeger configuration.
-3. **Verify Templates**: Ensure the `jaeger-ds-span` template exists and has `"data_stream": {}`.
+2. **Configuration**: Ensure `UseILM=true` (if applicable) or verify default templates.
+3. **Verify Templates**: Ensure the `jaeger-span` template exists and has `"data_stream": {}`.
    ```bash
-   curl -X GET "localhost:9200/_index_template/jaeger-ds-span?pretty"
+   curl -X GET "localhost:9200/_index_template/jaeger-span?pretty"
    ```
 4. **Generate Data**: Use `tracegen` or a sample app like HotROD to send spans.
 5. **Verify Data Stream**:
@@ -100,17 +166,5 @@ Documents are correctly indexed into the data stream with the `@timestamp` field
 }
 ```
 
-#### Backward Compatibility (Legacy Data)
-Existing indices (e.g., `jaeger-span-2025-12-27`) remain queryable alongside new data:
-```json
-{
-  "_index": "jaeger-span-2025-12-27",
-  "_source": {
-    "traceID": "legacy-test-1",
-    "startTimeMillis": 1735293600000
-  }
-}
-```
-
-#### Isolation Verified
-Confirmed that legacy wildcard searches like `jaeger-span-*` **do not** return data from `jaeger-ds-span`, ensuring complete isolation as requested.
+#### Verification
+Confirmed that traces are queryable and data streams are created as expected.

internal/storage/elasticsearch/client.go

@@ -51,9 +51,8 @@ type IndexService interface {
 	Index(index string) IndexService
 	Type(typ string) IndexService
 	Id(id string) IndexService
-	OpType(opType string) IndexService
 	BodyJson(body any) IndexService
-	Add()
+	Add(opType string)
 }
 
 // SearchService is an abstraction for elastic.SearchService

internal/storage/elasticsearch/config/config.go

@@ -168,9 +168,6 @@ type Configuration struct {
 	// Read more about ILM at
 	// https://www.jaegertracing.io/docs/deployment/#enabling-ilm-support
 	UseILM bool `mapstructure:"use_ilm"`
-	// UseDataStream, if set to true, will use Elasticsearch data streams for storing traces.
-	// This requires Elasticsearch 7.9+.
-	UseDataStream bool `mapstructure:"use_data_stream"`
 
 	// ---- jaeger-specific configs ----
 	// MaxDocCount Defines maximum number of results to fetch from storage per query.
@@ -539,9 +536,6 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 			c.CustomHeaders[k] = v
 		}
 	}
-	if !c.UseDataStream {
-		c.UseDataStream = source.UseDataStream
-	}
 }
 
 // RolloverFrequencyAsNegativeDuration returns the index rollover frequency duration for the given frequency string

internal/storage/elasticsearch/mocks/mocks.go

@@ -225,13 +225,11 @@ func (i IndexServiceWrapper) Type(typ string) es.IndexService {
 	return WrapESIndexService(i.bulkIndexReq.Type(typ), i.bulkService, i.esVersion)
 }
 
-// OpType calls this function to internal service.
-func (i IndexServiceWrapper) OpType(opType string) es.IndexService {
-	return WrapESIndexService(i.bulkIndexReq.OpType(opType), i.bulkService, i.esVersion)
-}
-
 // Add adds the request to bulk service
-func (i IndexServiceWrapper) Add() {
+func (i IndexServiceWrapper) Add(opType string) {
+	if opType != "" {
+		i.bulkIndexReq.OpType(opType)
+	}
 	i.bulkService.Add(i.bulkIndexReq)
 }
 

internal/storage/elasticsearch/wrapper/wrapper.go

@@ -121,7 +121,6 @@ func (f *FactoryBase) GetSpanReaderParams() esspanstore.SpanReaderParams {
 		ServiceReadAlias:    f.config.ServiceReadAlias,
 		Logger:              f.logger,
 		Tracer:              f.tracer.Tracer("esspanstore.SpanReader"),
-		UseDataStream:       f.config.UseDataStream,
 	}
 }
 
@@ -142,7 +141,6 @@ func (f *FactoryBase) GetSpanWriterParams() esspanstore.SpanWriterParams {
 		Logger:              f.logger,
 		MetricsFactory:      f.metricsFactory,
 		ServiceCacheTTL:     f.config.ServiceCacheTTL,
-		UseDataStream:       f.config.UseDataStream,
 	}
 }
 
@@ -155,7 +153,6 @@ func (f *FactoryBase) GetDependencyStoreParams() esdepstorev2.Params {
 		IndexDateLayout:     f.config.Indices.Dependencies.DateLayout,
 		MaxDocCount:         f.config.MaxDocCount,
 		UseReadWriteAliases: f.config.UseReadWriteAliases,
-		UseDataStream:       f.config.UseDataStream,
 	}
 }
 
@@ -168,7 +165,6 @@ func (f *FactoryBase) CreateSamplingStore(int /* maxBuckets */) (samplingstore.S
 		IndexRolloverFrequency: config.RolloverFrequencyAsNegativeDuration(f.config.Indices.Sampling.RolloverFrequency),
 		Lookback:               f.config.AdaptiveSamplingLookback,
 		MaxDocCount:            f.config.MaxDocCount,
-		UseDataStream:          f.config.UseDataStream,
 	}
 	store := essamplestore.NewSamplingStore(params)
 
@@ -194,7 +190,6 @@ func (f *FactoryBase) mappingBuilderFromConfig(cfg *config.Configuration) mappin
 		Indices:         cfg.Indices,
 		EsVersion:       cfg.Version,
 		UseILM:          cfg.UseILM,
-		UseDataStream:   cfg.UseDataStream,
 		ILMPolicyName:   cfg.Indices.IndexPrefix.Apply(defaultILMPolicyName),
 	}
 }

internal/storage/v1/elasticsearch/factory.go

@@ -1,6 +1,7 @@
 {
   "priority": 502,
-  "index_patterns": "test-jaeger-dependencies-*",
+  "index_patterns": "test-jaeger-dependencies",
+  "data_stream": {},
   "template": {
     "aliases": {
       "test-jaeger-dependencies-read": {}