Optimize the effective logic of num_most_recent_indicesconfiguration in the es plugin #1396
Conversation
… in elasticsearch.toml and better interact with indices_include configuration
|
|
||
| //add indices filter | ||
| if len(i.indicesIncluded) == 0 { | ||
| if len(i.indicesIncluded) == 0 || len(i.indicesIncluded) > 80 { |
There was a problem hiding this comment.
Pull request overview
This PR optimizes the num_most_recent_indices configuration logic in the Elasticsearch plugin to better interact with indices_include configuration. The changes introduce dynamic index pattern matching using regex, centralize index filtering logic, and add batch processing to handle large numbers of indices.
Changes:
- Introduced dynamic index matching with configurable regex patterns to group and limit indices by recency
- Refactored collector functions to remove distributed filtering logic and accept pre-filtered index lists
- Added batch processing (80 indices per request) to prevent URL length issues when querying many indices
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 29 comments.
Show a summary per file
| File | Description |
|---|---|
| inputs/elasticsearch/elasticsearch.go | Added dynamic index classification logic, new queryURL method, and centralized index filtering before passing to collectors |
| inputs/elasticsearch/collector/indices.go | Removed signature parameters for filtering, added batch processing for large index lists, and added merge function for batched results |
| inputs/elasticsearch/collector/indices_settings.go | Simplified to accept pre-filtered indices, added filterMapByKeys for post-query filtering when >80 indices |
| inputs/elasticsearch/collector/indices_mappings.go | Simplified to accept pre-filtered indices, added filterMapByKeys for post-query filtering when >80 indices |
| inputs/elasticsearch/collector/ilm_indices.go | Simplified to accept pre-filtered indices, added filterMapByKeys for post-query filtering when >80 indices |
| inputs/elasticsearch/collector/indices_test.go | Updated test to match new collector signature without filtering parameters |
| inputs/elasticsearch/collector/indices_settings_test.go | Updated test to match new collector signature without filtering parameters |
| inputs/elasticsearch/collector/indices_mappings_test.go | Updated test to match new collector signature without filtering parameters |
| inputs/elasticsearch/collector/ilm_indices_test.go | Updated test to match new collector signature without filtering parameters |
| conf/input.elasticsearch/elasticsearch.toml | Added configuration documentation for num_most_recent_indices and dynamic_index_matcher_regexp |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| func (im *IndicesMappings) fetchAndDecodeIndicesMappings() (*IndicesMappingsResponse, error) { | ||
| u := *im.url | ||
| //add indices filter | ||
| if len(im.indicesIncluded) == 0 || len(im.indicesIncluded) > 80 { |
There was a problem hiding this comment.
The magic number 80 is used as a threshold without explanation. This should be extracted as a named constant with documentation explaining why this specific value is used.
| uu := *i.url | ||
| end := k + batchSize | ||
| if end > len(i.indicesIncluded) { | ||
| end = len(i.indicesIncluded) // Ensure not to cross the line |
There was a problem hiding this comment.
The comment "Ensure not to cross the line" is unclear. It should be more specific, such as "Ensure we don't exceed the slice bounds" or "Prevent index out of bounds".
| end = len(i.indicesIncluded) // Ensure not to cross the line | |
| end = len(i.indicesIncluded) // Ensure end does not exceed indicesIncluded slice bounds |
| "encoding/json" | ||
| "flashcat.cloud/categraf/pkg/filter" | ||
| "fmt" | ||
| "golang.org/x/exp/slices" |
There was a problem hiding this comment.
The package golang.org/x/exp/slices is from the experimental packages and was later standardized in Go 1.21 as the standard library slices package. If this project uses Go 1.21 or later, consider using the standard library version instead for better long-term stability.
| "golang.org/x/exp/slices" | |
| "slices" |
| func (im *IndicesMappings) filterMapByKeys(response IndicesMappingsResponse, included []string) *IndicesMappingsResponse { | ||
| resultMap := make(map[string]IndexMapping) | ||
| for key, value := range response { | ||
| if slices.Contains(included, key) { | ||
| resultMap[key] = value | ||
| } | ||
| } |
There was a problem hiding this comment.
The filterMapByKeys function uses slices.Contains inside a loop which results in O(n*m) time complexity where n is the size of the response map and m is the size of the included slice. For better performance when filtering large datasets, consider converting the included slice to a map for O(1) lookup time, reducing overall complexity to O(n).
| if ins.NumMostRecentIndices != 0 { | ||
| //match Dynamic indexing | ||
| //query all indices | ||
| uu := *EsUrl | ||
| //url: /_cat/indices?format=json&h=index | ||
| if len(ins.IndicesInclude) > 0 { | ||
| uu.Path = path.Join(uu.Path, "/_cat/indices/"+strings.Join(ins.IndicesInclude, ",")) | ||
| } else { | ||
| uu.Path = path.Join(uu.Path, "/_cat/indices") | ||
| } | ||
| uu.RawQuery = "format=json&s=index:desc&h=index" | ||
| indices_bts, err := ins.queryURL(&uu) | ||
| if err != nil { | ||
| log.Println("E! failed to query all indices:", err) | ||
| } | ||
| var indices []IndicesInfo | ||
| if err := json.Unmarshal(indices_bts, &indices); err != nil { | ||
| log.Println("E! json unmarshal to query all indices:", err) | ||
| } | ||
|
|
||
| var indexList []string | ||
| //match Dynamic indexing，exchange index name | ||
| indexList = ins.classifyDynamicIndexes(indices) | ||
| //must use NewIndicesInclude,cannot recover IndicesInclude | ||
| ins.NewIndicesInclude = indexList | ||
|
|
||
| } else { | ||
| ins.NewIndicesInclude = ins.IndicesInclude | ||
| } |
There was a problem hiding this comment.
There is a potential race condition. The field ins.NewIndicesInclude is being written to in line 261 and 264 from within a goroutine (created at line 214), but there's no synchronization mechanism protecting these writes. If multiple servers are configured, multiple goroutines could be writing to this field concurrently, leading to a data race. Consider using a mutex to protect access to this field or making it local to each goroutine.
There was a problem hiding this comment.
@copilot open a new pull request to apply changes based on this feedback
| } else { | ||
| u.RawQuery = "ignore_unavailable=true" | ||
| } | ||
| //Prevent GET requests from failing to fully query words when there are too many indicesIncluded |
There was a problem hiding this comment.
The comment "Prevent GET requests from failing to fully query words when there are too many indicesIncluded" is unclear and has grammatical issues. It should be reworded to something like "Split into batches to prevent GET request failures when there are too many indices" for clarity.
| //Prevent GET requests from failing to fully query words when there are too many indicesIncluded | |
| // Split indices into batches to prevent GET request failures when too many indices are requested at once. |
| func (ins *Instance) queryURL(u *url.URL) ([]byte, error) { | ||
| res, err := ins.Client.Get(u.String()) | ||
| if err != nil { | ||
| return []byte{}, fmt.Errorf("failed to get resource from %s://%s:%s%s: %s", | ||
| u.Scheme, u.Hostname(), u.Port(), u.Path, err) | ||
| } | ||
| defer func() { | ||
| err := res.Body.Close() | ||
| if err != nil { | ||
| log.Println("E! failed to close response body:", err) | ||
| } | ||
| }() | ||
|
|
||
| if res.StatusCode != http.StatusOK { | ||
| return []byte{}, fmt.Errorf("HTTP Request failed with status code %d", res.StatusCode) | ||
| } | ||
|
|
||
| bts, err := io.ReadAll(res.Body) | ||
| if err != nil { | ||
| return []byte{}, err | ||
| } | ||
| return bts, nil | ||
| } | ||
|
|
||
| // match Dynamic Indexes | ||
| func (ins *Instance) classifyDynamicIndexes(indicesInfo []IndicesInfo) []string { | ||
|
|
||
| if len(ins.DynamicIndexMatcherRegexp) == 0 { | ||
| //default matcher | ||
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$`) | ||
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `[\\.-._]\\d+(\\.\\d+){0,2}$`) | ||
| } | ||
|
|
||
| var patterns []*regexp.Regexp | ||
|
|
||
| for _, patternStr := range ins.DynamicIndexMatcherRegexp { | ||
| re := regexp.MustCompile(patternStr) | ||
| patterns = append(patterns, re) | ||
| } | ||
|
|
||
| groups := make(map[string][]string) | ||
|
|
||
| for _, index := range indicesInfo { | ||
| matched := false | ||
|
|
||
| // Attempt to match known dynamic patterns | ||
| for _, pattern := range patterns { | ||
| if loc := pattern.FindStringIndex(index.Index); loc != nil { | ||
| // Construct group patterns (replace dynamic parts with *) | ||
| groupKey := index.Index[:loc[0]] + "*" + index.Index[loc[1]:] | ||
| groups[groupKey] = append(groups[groupKey], index.Index) | ||
| matched = true | ||
| break | ||
| } | ||
| } | ||
|
|
||
| // Indexes not matching known patterns are grouped separately | ||
| if !matched { | ||
| groups[index.Index] = []string{index.Index} | ||
|
|
||
| } | ||
| } | ||
|
|
||
| if ins.DebugMod { | ||
| for pattern, indexes := range groups { | ||
| fmt.Printf("[%s] (%d index total \n)", pattern, len(indexes)) | ||
| if len(indexes) > 5 { | ||
| fmt.Printf(" result: %v ... \n", indexes[:5]) | ||
| } else { | ||
| fmt.Printf("result: %v \n", indexes) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| var new_indices []string | ||
|
|
||
| //Retrieve the first n indexes | ||
| for pattern, indexes := range groups { | ||
| if ins.DebugMod { | ||
| fmt.Printf("[%s] (%d index total) \n", pattern, len(indexes)) | ||
| } | ||
| if len(indexes) > ins.NumMostRecentIndices { | ||
| if ins.DebugMod { | ||
| fmt.Printf(" result: %v \n", indexes[:ins.NumMostRecentIndices]) | ||
| } | ||
| new_indices = append(new_indices, indexes[:ins.NumMostRecentIndices]...) | ||
| } else { | ||
| if ins.DebugMod { | ||
| fmt.Printf("result: %v \n", indexes) | ||
| } | ||
| new_indices = append(new_indices, indexes[:]...) | ||
| } | ||
| } | ||
|
|
||
| return new_indices | ||
| } |
There was a problem hiding this comment.
The new methods queryURL and classifyDynamicIndexes added to the Instance struct lack test coverage. These methods implement core functionality for dynamic index matching and should have unit tests to verify correct behavior with various index naming patterns.
| ## match //YYYY.MM.DD 或 YYYY-MM-DD 或 YYYYMMDD 或 YYYY-MM-DD-HH | ||
| ## //YYYY.MM 或 YYYY-MM 或 YYYYMM 或YYYYMMDDHH | ||
| ## //YY.MM.DD 或 YY-MM-DD 或 YYMMDD 或YYYY.MM.DD.HH |
There was a problem hiding this comment.
The comment uses "oder" which appears to be a typo from German. It should be "or" in English to match the rest of the codebase.
| ## match //YYYY.MM.DD 或 YYYY-MM-DD 或 YYYYMMDD 或 YYYY-MM-DD-HH | |
| ## //YYYY.MM 或 YYYY-MM 或 YYYYMM 或YYYYMMDDHH | |
| ## //YY.MM.DD 或 YY-MM-DD 或 YYMMDD 或YYYY.MM.DD.HH | |
| ## match //YYYY.MM.DD or YYYY-MM-DD or YYYYMMDD or YYYY-MM-DD-HH | |
| ## //YYYY.MM or YYYY-MM or YYYYMM or YYYYMMDDHH | |
| ## //YY.MM.DD or YY-MM-DD or YYMMDD or YYYY.MM.DD.HH |
| //dst.Shards = src.Shards | ||
| //dst.All = src.All | ||
|
|
||
| // 2. MergeIndices map |
There was a problem hiding this comment.
The comment "2. MergeIndices map" is missing a space. It should be "2. Merge Indices map" for consistency with the comment style.
| // 2. MergeIndices map | |
| // 2. Merge Indices map |
| for pattern, indexes := range groups { | ||
| if ins.DebugMod { | ||
| fmt.Printf("[%s] (%d index total) \n", pattern, len(indexes)) | ||
| } | ||
| if len(indexes) > ins.NumMostRecentIndices { | ||
| if ins.DebugMod { | ||
| fmt.Printf(" result: %v \n", indexes[:ins.NumMostRecentIndices]) | ||
| } | ||
| new_indices = append(new_indices, indexes[:ins.NumMostRecentIndices]...) | ||
| } else { | ||
| if ins.DebugMod { | ||
| fmt.Printf("result: %v \n", indexes) | ||
| } | ||
| new_indices = append(new_indices, indexes[:]...) | ||
| } |
There was a problem hiding this comment.
The indices returned from the groups map iteration (line 523) are not sorted before being sliced. Since Go's map iteration order is random, the "most recent" indices may not actually be the most recent ones. The indices should be sorted before taking the first N elements to ensure the correct indices are selected.
| indicesMappingsResponse = im.gatherIndividualIndicesStats(indicesMappingsResponse) | ||
|
|
||
| //过滤 include | ||
| if len(im.indicesIncluded) > 80 { |
|
配置也辛苦补充到README
|
2、Configuring the judgment parameters for batch processing of the maximum number of indices_include configurations，set default 80
已补充 |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated 23 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| func (cs *IndicesSettings) filterMapByKeys(response IndicesSettingsResponse, included []string) IndicesSettingsResponse { | ||
| resultMap := make(map[string]Index) | ||
| for key, value := range response { | ||
| if slices.Contains(included, key) { | ||
| resultMap[key] = value | ||
| } | ||
| } |
There was a problem hiding this comment.
Using slices.Contains in a loop over potentially large maps creates O(n*m) complexity, where n is the number of indices in the response and m is the number of included indices. For large datasets, this could be inefficient. Consider converting the included slice to a map for O(1) lookup performance.
| func (i *IlmIndiciesCollector) filterMapByKeys(originalMap map[string]IlmIndexResponse, allowedKeys []string) map[string]IlmIndexResponse { | ||
|
|
||
| // Gather only the number of indexes that have been configured, in descending order (most recent, if date-stamped). | ||
| for i := indicesCount - 1; i >= indicesCount-indicesToTrackCount; i-- { | ||
| indexName := matchingIndices[i] | ||
| newIndicesMappings[indexName] = resp.Indices[indexName] | ||
| resultMap := make(map[string]IlmIndexResponse) | ||
| for key, value := range originalMap { | ||
| if slices.Contains(allowedKeys, key) { | ||
| resultMap[key] = value | ||
| } | ||
| } |
There was a problem hiding this comment.
Using slices.Contains in a loop over potentially large maps creates O(n*m) complexity, where n is the number of indices in the response and m is the number of included indices. For large datasets, this could be inefficient. Consider converting the allowedKeys slice to a map for O(1) lookup performance.
|
|
||
| if ins.DebugMod { | ||
| for pattern, indexes := range groups { | ||
| fmt.Printf("[%s] (%d index total \n)", pattern, len(indexes)) |
There was a problem hiding this comment.
The comment has a formatting issue with an extra space in "(%d index total" - there should be only one space before "index". It should be "(%d index total".
| fmt.Printf("[%s] (%d index total \n)", pattern, len(indexes)) | |
| fmt.Printf("[%s] (%d index total \n)", pattern, len(indexes)) |
| url *url.URL | ||
| indicesIncluded []string | ||
| numMostRecentIndices int | ||
| indexMatchers map[string]filter.Filter |
There was a problem hiding this comment.
The field indexMatchers is no longer being initialized in the constructor but is still declared in the struct. Since the old filtering logic that used this field has been removed, this unused field should be removed from the struct definition to keep the code clean.
| |Set the indicator data for the latest index of the first "num_mast_decent_indice" in the date class dynamic index | ||
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | ||
| |Can be used together with the 'indices_inclub' configuration | ||
|
|
||
|
|
||
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | ||
| |Used in conjunction with num_mast_decent_indice to specify the matching logic for dynamic indexes, default value: |
There was a problem hiding this comment.
There are multiple spelling errors in the documentation:
- "num_mast_decent_indice" should be "num_most_recent_indices"
- "indices_inclub" should be "indices_include"
| |Set the indicator data for the latest index of the first "num_mast_decent_indice" in the date class dynamic index | |
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | |
| |Can be used together with the 'indices_inclub' configuration | |
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | |
| |Used in conjunction with num_mast_decent_indice to specify the matching logic for dynamic indexes, default value: | |
| |Set the indicator data for the latest index of the first "num_most_recent_indices" in the date class dynamic index | |
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | |
| |Can be used together with the 'indices_include' configuration | |
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | |
| |Used in conjunction with num_most_recent_indices to specify the matching logic for dynamic indexes, default value: |
| |Set the indicator data for the latest index of the first "num_mast_decent_indice" in the date class dynamic index | ||
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | ||
| |Can be used together with the 'indices_inclub' configuration | ||
|
|
||
|
|
||
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | ||
| |Used in conjunction with num_mast_decent_indice to specify the matching logic for dynamic indexes, default value: |
There was a problem hiding this comment.
The spelling "num_mast_decent_indice" should be "num_most_recent_indices" for consistency with the actual configuration parameter name.
| |Set the indicator data for the latest index of the first "num_mast_decent_indice" in the date class dynamic index | |
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | |
| |Can be used together with the 'indices_inclub' configuration | |
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | |
| |Used in conjunction with num_mast_decent_indice to specify the matching logic for dynamic indexes, default value: | |
| |Set the indicator data for the latest index of the first "num_most_recent_indices" in the date class dynamic index | |
| |It can greatly reduce the large scale of indicators caused by historical dynamic indexing | |
| |Can be used together with the 'indices_inclub' configuration | |
| #### `dynamic_index_matcher_regexp` = ["(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$","[\\.-._]\\d+(\\.\\d+){0,2}$"] | |
| |Used in conjunction with num_most_recent_indices to specify the matching logic for dynamic indexes, default value: |
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$`) | ||
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `[\\.-._]\\d+(\\.\\d+){0,2}$`) |
There was a problem hiding this comment.
The regular expressions contain excessive escaping (double backslashes) in the raw string literals. Since these are raw strings enclosed in backticks, the backslashes don't need to be escaped. For example, (?P<date>(?:\\d{4}|\\d{2}) should be (?P<date>(?:\d{4}|\d{2}). The double backslashes will be interpreted literally, which may cause the regex patterns to not match as intended.
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `(?P<date>(?:\\d{4}|\\d{2})[.-]?(?:\\d{2})[.-]?(?:\\d{2})?[.-]?(?:\\d{2})?)$`) | |
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `[\\.-._]\\d+(\\.\\d+){0,2}$`) | |
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `(?P<date>(?:\d{4}|\d{2})[.-]?(?:\d{2})[.-]?(?:\d{2})?[.-]?(?:\d{2})?)$`) | |
| ins.DynamicIndexMatcherRegexp = append(ins.DynamicIndexMatcherRegexp, `[._-]\d+(\.\d+){0,2}$`) |
| for pattern, indexes := range groups { | ||
| if ins.DebugMod { | ||
| fmt.Printf("[%s] (%d index total) \n", pattern, len(indexes)) | ||
| } | ||
| if len(indexes) > ins.NumMostRecentIndices { | ||
| if ins.DebugMod { | ||
| fmt.Printf(" result: %v \n", indexes[:ins.NumMostRecentIndices]) | ||
| } | ||
| new_indices = append(new_indices, indexes[:ins.NumMostRecentIndices]...) | ||
| } else { | ||
| if ins.DebugMod { | ||
| fmt.Printf("result: %v \n", indexes) | ||
| } | ||
| new_indices = append(new_indices, indexes[:]...) | ||
| } |
There was a problem hiding this comment.
The indices are queried with sorting by index name in descending order (s=index:desc), but the classifyDynamicIndexes function doesn't preserve this order when grouping. The function returns indices from the groups map using a range loop, which has non-deterministic iteration order. Since num_most_recent_indices is meant to get the most recent indices, the lack of sorting could result in the wrong indices being selected. The grouped indices should be sorted before taking the first N elements.
| //add indices filter | ||
| if len(cs.indicesIncluded) == 0 || len(cs.indicesIncluded) > cs.maxIndicesIncludeCount { | ||
| u.Path = path.Join(u.Path, "/_all/_settings") | ||
| } else if len(cs.indicesIncluded) <= cs.maxIndicesIncludeCount { |
There was a problem hiding this comment.
The conditional logic is redundant. The first condition checks len(cs.indicesIncluded) == 0 || len(cs.indicesIncluded) > cs.maxIndicesIncludeCount, and the else-if checks len(cs.indicesIncluded) <= cs.maxIndicesIncludeCount. The else-if is unnecessary because it's the only remaining case. This should be simplified to use just an if-else structure.
| } else if len(cs.indicesIncluded) <= cs.maxIndicesIncludeCount { | |
| } else { |
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
|
Thank you @longjun91 |
2 participants
Optimize the effective logic of num_most_recent_indices configuration in elasticsearch.toml and better interact with indices_include configuration