Skip to content

feat(autoops_es): add ingest volume enrichment fields #52226

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ilya-shevelyov wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat(autoops_es): add ingest volume enrichment fields #52226

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5a52655
feat(autoops_es): add ingest volume enrichment fields (#51404)
ilya-shevelyov Jul 23, 2026
d82ae6e
rename JSONShard.DatasetSize to TotalDataSetSizeInBytes for consistency
ilya-shevelyov Jul 23, 2026
cdda877
add changelog fragment for ingest volume enrichment fields
ilya-shevelyov Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
212 changes: 176 additions & 36 deletions x-pack/metricbeat/module/autoops_es/cat_shards/_meta/test/cat_shards.8.15.3.json
View file
Open in desktop

Large diffs are not rendered by default.

32 changes: 32 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,38 @@ var (
IsUsable: func(obj *NodeIndexShards) bool { return obj.SearchQueryTotal != nil },
WriteValue: func(obj *NodeIndexShards, value float64) { obj.SearchRatePerSecond = &value },
},
{
CalculateValue: utils.CalculateRate,
ConvertTime: utils.MillisToSeconds,
GetTime: utils.UseTimestamp[*NodeIndexShards],
GetValue: func(obj *NodeIndexShards) int64 { return *obj.DocsCount },
IsUsable: func(obj *NodeIndexShards) bool { return obj.DocsCount != nil },
WriteValue: func(obj *NodeIndexShards, value float64) { obj.IngestDocsPerSecond = &value },
},
{
CalculateValue: utils.CalculateRate,
ConvertTime: utils.MillisToSeconds,
GetTime: utils.UseTimestamp[*NodeIndexShards],
GetValue: func(obj *NodeIndexShards) int64 { return *obj.SizeInBytes },
IsUsable: func(obj *NodeIndexShards) bool { return obj.SizeInBytes != nil },
WriteValue: func(obj *NodeIndexShards, value float64) { obj.IngestBytesPerSecond = &value },
},
{
CalculateValue: utils.CalculateRate,
ConvertTime: utils.MillisToSeconds,
GetTime: utils.UseTimestamp[*NodeIndexShards],
GetValue: func(obj *NodeIndexShards) int64 { return *obj.BulkTotalSizeInBytes },
IsUsable: func(obj *NodeIndexShards) bool { return obj.BulkTotalSizeInBytes != nil },
WriteValue: func(obj *NodeIndexShards, value float64) { obj.BulkBytesPerSecond = &value },
},
{
CalculateValue: utils.CalculateRate,
ConvertTime: utils.MillisToSeconds,
GetTime: utils.UseTimestamp[*NodeIndexShards],
GetValue: func(obj *NodeIndexShards) int64 { return *obj.BulkTotalOperations },
IsUsable: func(obj *NodeIndexShards) bool { return obj.BulkTotalOperations != nil },
WriteValue: func(obj *NodeIndexShards, value float64) { obj.BulkOperationsPerSecond = &value },
},
// LATENCIES:
{
CalculateValue: utils.CalculateLatency,
Expand Down
125 changes: 125 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,11 @@ func getShard(nodeId string, nodeName string, shardId int32, primary bool, state
get_missing_time := []int64{100, 101, 102, 103}
get_missing_total := []int64{110, 111, 112, 113}

bulk_total_size_in_bytes := []int64{200, 201, 202, 203}
bulk_total_operations := []int64{300, 301, 302, 303}
total_data_set_size_in_bytes := []int64{10, 11, 12, 13}
dense_vector_count := []int64{0, 0, 0, 0}

shard.docs = &docs[index]
shard.store = &store[index]
shard.segments_count = &segments_count[index]
Expand All @@ -70,6 +75,10 @@ func getShard(nodeId string, nodeName string, shardId int32, primary bool, state
shard.merges_total_time = &merges_total_time[index]
shard.get_missing_time = &get_missing_time[index]
shard.get_missing_total = &get_missing_total[index]
shard.bulk_total_size_in_bytes = &bulk_total_size_in_bytes[index]
shard.bulk_total_operations = &bulk_total_operations[index]
shard.total_data_set_size_in_bytes = &total_data_set_size_in_bytes[index]
shard.dense_vector_count = &dense_vector_count[index]

return shard
}
Expand Down Expand Up @@ -823,3 +832,119 @@ func TestConvertToNodeIndexShardsWithCache(t *testing.T) {
require.Nil(t, myIndexNode3.SearchRatePerSecond)
require.Nil(t, myIndexNode3.TimestampDiff)
}

func getNodeIndexShardsWithIngestBulk() map[string]NodeIndexShards {
docsCount := []int64{1, 2}
sizeInBytes := []int64{10, 11}
bulkTotalSizeInBytes := []int64{200, 201}
bulkTotalOperations := []int64{300, 301}

base := getNodeIndexShards()

n1 := base["my-index-node_id-node1"]
n1.DocsCount = &docsCount[0]
n1.SizeInBytes = &sizeInBytes[0]
n1.BulkTotalSizeInBytes = &bulkTotalSizeInBytes[0]
n1.BulkTotalOperations = &bulkTotalOperations[0]
base["my-index-node_id-node1"] = n1

n2 := base["my-index-node_id-node2"]
n2.DocsCount = &docsCount[1]
n2.SizeInBytes = &sizeInBytes[1]
n2.BulkTotalSizeInBytes = &bulkTotalSizeInBytes[1]
n2.BulkTotalOperations = &bulkTotalOperations[1]
base["my-index-node_id-node2"] = n2

return base
}

func TestEnrichNodeIndexShardsIngestAndBulkRatesWithoutCache(t *testing.T) {
clearCache()

nodeIndexShardsMap := getNodeIndexShardsWithIngestBulk()
nodeIndexShardsList := enrichNodeIndexShards(nodeIndexShardsMap, map[string]IndexMetadata{})

for _, nodeIndexShards := range nodeIndexShardsList {
require.Nil(t, nodeIndexShards.IngestDocsPerSecond)
require.Nil(t, nodeIndexShards.IngestBytesPerSecond)
require.Nil(t, nodeIndexShards.BulkBytesPerSecond)
require.Nil(t, nodeIndexShards.BulkOperationsPerSecond)
}
}

func TestEnrichNodeIndexShardsIngestAndBulkRatesWithCachedValues(t *testing.T) {
// 10s ago cache
initCache(getNodeIndexShardsWithIngestBulk(), 10)

nodeIndexShardsMap := getNodeIndexShardsWithIngestBulk()

for key, nodeIndexShards := range nodeIndexShardsMap {
*nodeIndexShards.DocsCount += 10 // delta=10 → 10/10 = 1/s
*nodeIndexShards.SizeInBytes += 20 // delta=20 → 20/10 = 2/s
*nodeIndexShards.BulkTotalSizeInBytes += 100 // delta=100 → 100/10 = 10/s
*nodeIndexShards.BulkTotalOperations += 50 // delta=50 → 50/10 = 5/s

nodeIndexShardsMap[key] = nodeIndexShards
}

nodeIndexShardsList := enrichNodeIndexShards(nodeIndexShardsMap, map[string]IndexMetadata{})

for _, nodeIndexShards := range nodeIndexShardsList {
require.EqualValues(t, 1, *nodeIndexShards.IngestDocsPerSecond)
require.EqualValues(t, 2, *nodeIndexShards.IngestBytesPerSecond)
require.EqualValues(t, 10, *nodeIndexShards.BulkBytesPerSecond)
require.EqualValues(t, 5, *nodeIndexShards.BulkOperationsPerSecond)
}
}

func TestEnrichNodeIndexShardsIngestAndBulkRatesWithNoChange(t *testing.T) {
// 10s ago cache — identical values, expect zero rates
initCache(getNodeIndexShardsWithIngestBulk(), 10)

nodeIndexShardsMap := getNodeIndexShardsWithIngestBulk()
nodeIndexShardsList := enrichNodeIndexShards(nodeIndexShardsMap, map[string]IndexMetadata{})

for _, nodeIndexShards := range nodeIndexShardsList {
require.EqualValues(t, 0, *nodeIndexShards.IngestDocsPerSecond)
require.EqualValues(t, 0, *nodeIndexShards.IngestBytesPerSecond)
require.EqualValues(t, 0, *nodeIndexShards.BulkBytesPerSecond)
require.EqualValues(t, 0, *nodeIndexShards.BulkOperationsPerSecond)
}
}

func TestEnrichNodeIndexShardsIngestAndBulkRatesWithNewNode(t *testing.T) {
// 10s ago cache with only node1 — node2 is a cache miss
initCache(getNodeIndexShardsWithIngestBulk(), 10)

nodeIndexShardsMap := getNodeIndexShardsWithIngestBulk()

for key, nodeIndexShards := range nodeIndexShardsMap {
*nodeIndexShards.DocsCount += 10
*nodeIndexShards.SizeInBytes += 20
*nodeIndexShards.BulkTotalSizeInBytes += 100
*nodeIndexShards.BulkTotalOperations += 50
nodeIndexShardsMap[key] = nodeIndexShards
}

newNode := getNodeIndexShardsWithIngestBulk()["my-index-node_id-node2"]
newNode.Index = "my-other-index"
newNode.NodeId = "node3"
newNode.IndexNode = "my-other-index-node_id-node3"
nodeIndexShardsMap["my-other-index-node_id-node3"] = newNode

nodeIndexShardsList := enrichNodeIndexShards(nodeIndexShardsMap, map[string]IndexMetadata{})

for _, nodeIndexShards := range nodeIndexShardsList {
if nodeIndexShards.NodeId != "node3" {
require.EqualValues(t, 1, *nodeIndexShards.IngestDocsPerSecond)
require.EqualValues(t, 2, *nodeIndexShards.IngestBytesPerSecond)
require.EqualValues(t, 10, *nodeIndexShards.BulkBytesPerSecond)
require.EqualValues(t, 5, *nodeIndexShards.BulkOperationsPerSecond)
} else {
require.Nil(t, nodeIndexShards.IngestDocsPerSecond)
require.Nil(t, nodeIndexShards.IngestBytesPerSecond)
require.Nil(t, nodeIndexShards.BulkBytesPerSecond)
require.Nil(t, nodeIndexShards.BulkOperationsPerSecond)
}
}
}
2 changes: 1 addition & 1 deletion x-pack/metricbeat/module/autoops_es/cat_shards/cat_shards.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import (

const (
catShardsMetricSet = "cat_shards"
catShardsPath = "/_cat/shards?s=i&h=n,i,id,s,p,st,d,sto,sc,sqto,sqti,iito,iiti,iif,mt,mtt,gmto,gmti,ur,ud&bytes=b&time=ms&format=json"
catShardsPath = "/_cat/shards?s=i&h=n,i,id,s,p,st,d,sto,sc,sqto,sqti,iito,iiti,iif,mt,mtt,gmto,gmti,ur,ud,btsi,bto,dataset,dvc&bytes=b&time=ms&format=json"
resolveIndexPath = "/_resolve/index/*?expand_wildcards=all&filter_path=indices"
)

Expand Down
5 changes: 5 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/data.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,11 @@ type JSONShard struct {
SearchQueryTime json.Number `json:"sqti"`
SearchQueryTotal json.Number `json:"sqto"`

BulkTotalSizeInBytes json.Number `json:"btsi"`
BulkTotalOperations json.Number `json:"bto"`
DatasetSize json.Number `json:"dataset"`
Comment thread
ilya-shevelyov marked this conversation as resolved.
Outdated
DenseVectorCount json.Number `json:"dvc"`

// only used for unassigned state
UnassignedReason *string `json:"ur"`
UnassignedDetails *string `json:"ud"`
Expand Down
5 changes: 5 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/data_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,11 @@ func expectValidParsedDetailedShardsWithCache(t *testing.T, data metricset.Fetch
require.Nil(t, myIndexNode2.MergeRatePerSecond)
require.Nil(t, myIndexNode2.IndexLatencyInMillis)
require.Nil(t, myIndexNode2.MergeLatencyInMillis)
// Replica-only node: primary-only fields are nil, so ingest/bulk rates are nil
require.Nil(t, myIndexNode2.IngestDocsPerSecond)
require.Nil(t, myIndexNode2.IngestBytesPerSecond)
require.Nil(t, myIndexNode2.BulkBytesPerSecond)
require.Nil(t, myIndexNode2.BulkOperationsPerSecond)
}

// Expect a valid response from Elasticsearch to create N events
Expand Down
4 changes: 4 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/deserialize.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,10 @@ func deserializeShard(jsonShard JSONShard) Shard {
shard.segments_count = toInt64(jsonShard.SegmentsCount)
shard.search_query_time = toInt64(jsonShard.SearchQueryTime)
shard.search_query_total = toInt64(jsonShard.SearchQueryTotal)
shard.bulk_total_size_in_bytes = toInt64(jsonShard.BulkTotalSizeInBytes)
shard.bulk_total_operations = toInt64(jsonShard.BulkTotalOperations)
shard.total_data_set_size_in_bytes = toInt64(jsonShard.DatasetSize)
Comment thread
ilya-shevelyov marked this conversation as resolved.
Outdated
shard.dense_vector_count = toInt64(jsonShard.DenseVectorCount)

return shard
}
48 changes: 34 additions & 14 deletions x-pack/metricbeat/module/autoops_es/cat_shards/index_shards.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,20 +16,24 @@
state string

// optional; set if state != "UNASSIGNED"
docs *int64
store *int64
segments_count *int64
search_query_total *int64
search_query_time *int64
indexing_index_total *int64
indexing_index_time *int64
indexing_index_failed *int64
merges_total *int64
merges_total_time *int64
get_missing_time *int64
get_missing_total *int64
unassigned_reason *string
unassigned_details *string
docs *int64

Check failure on line 19 in x-pack/metricbeat/module/autoops_es/cat_shards/index_shards.go

View workflow job for this annotation

GitHub Actions / lint (ubuntu-latest) 

File is not properly formatted (goimports)

Check failure on line 19 in x-pack/metricbeat/module/autoops_es/cat_shards/index_shards.go

View workflow job for this annotation

GitHub Actions / lint (windows-latest) 

File is not properly formatted (goimports)

Check failure on line 19 in x-pack/metricbeat/module/autoops_es/cat_shards/index_shards.go

View workflow job for this annotation

GitHub Actions / lint (macos-latest) 

File is not properly formatted (goimports)
store *int64
segments_count *int64
search_query_total *int64
search_query_time *int64
indexing_index_total *int64
indexing_index_time *int64
indexing_index_failed *int64
merges_total *int64
merges_total_time *int64
get_missing_time *int64
get_missing_total *int64
bulk_total_size_in_bytes *int64
bulk_total_operations *int64
total_data_set_size_in_bytes *int64
dense_vector_count *int64
unassigned_reason *string
unassigned_details *string
}

type AssignedShard struct {
Expand Down Expand Up @@ -117,6 +121,18 @@
TotalMergesTotal *int64 `json:"total_merges_total"` // includes replicas
TotalMergesTotalTime *int64 `json:"total_merges_total_time"` // includes replicas
TimestampDiff *int64 `json:"timestamp_diff"`

// primary-only snapshot fields
BulkTotalSizeInBytes *int64 `json:"bulk_total_size_in_bytes"`
BulkTotalOperations *int64 `json:"bulk_total_operations"`
TotalDataSetSizeInBytes *int64 `json:"total_data_set_size_in_bytes"`
DenseVectorCount *int64 `json:"dense_vector_count"`

// ingest / bulk rate fields
IngestDocsPerSecond *float64 `json:"ingest_docs_per_second"`
IngestBytesPerSecond *float64 `json:"ingest_bytes_per_second"`
BulkBytesPerSecond *float64 `json:"bulk_bytes_per_second"`
BulkOperationsPerSecond *float64 `json:"bulk_operations_per_second"`
}

type NodeShardCount struct {
Expand Down Expand Up @@ -294,6 +310,10 @@
nodeIndex.IndexingIndexTotalTime = utils.AddInt64OrNull(nodeIndex.IndexingIndexTotalTime, shard.indexing_index_time)
nodeIndex.MergesTotal = utils.AddInt64OrNull(nodeIndex.MergesTotal, shard.merges_total)
nodeIndex.MergesTotalTime = utils.AddInt64OrNull(nodeIndex.MergesTotalTime, shard.merges_total_time)
nodeIndex.BulkTotalSizeInBytes = utils.AddInt64OrNull(nodeIndex.BulkTotalSizeInBytes, shard.bulk_total_size_in_bytes)
nodeIndex.BulkTotalOperations = utils.AddInt64OrNull(nodeIndex.BulkTotalOperations, shard.bulk_total_operations)
nodeIndex.TotalDataSetSizeInBytes = utils.AddInt64OrNull(nodeIndex.TotalDataSetSizeInBytes, shard.total_data_set_size_in_bytes)
nodeIndex.DenseVectorCount = utils.AddInt64OrNull(nodeIndex.DenseVectorCount, shard.dense_vector_count)
}

assignedShard := toAssignedShard(shard)
Expand Down
22 changes: 22 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/index_shards_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,6 +420,10 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
indexingFailed := []int64{60, 61, 62, 63}
indexingTotals := []int64{70, 71, 72, 73}
indexingTimes := []int64{74, 75, 76, 77}
bulkSizes := []int64{200, 201, 202, 203}
bulkOps := []int64{300, 301, 302, 303}
datasetSizes := []int64{20, 21, 22, 23}
dvcCounts := []int64{0, 0, 5, 0}

shards := []Shard{
{shard: 0, primary: true, node_id: "node1", node_name: "name1"},
Expand All @@ -442,6 +446,10 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
shard.indexing_index_failed = &indexingFailed[i]
shard.indexing_index_total = &indexingTotals[i]
shard.indexing_index_time = &indexingTimes[i]
shard.bulk_total_size_in_bytes = &bulkSizes[i]
shard.bulk_total_operations = &bulkOps[i]
shard.total_data_set_size_in_bytes = &datasetSizes[i]
shard.dense_vector_count = &dvcCounts[i]

shards[i] = shard
}
Expand Down Expand Up @@ -484,6 +492,10 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
require.Equal(t, searchQueryTimes[0], *nodeIndex.SearchQueryTime)
require.Equal(t, mergeTotals[0], *nodeIndex.TotalMergesTotal)
require.Equal(t, mergeTimes[0], *nodeIndex.TotalMergesTotalTime)
require.Equal(t, bulkSizes[0], *nodeIndex.BulkTotalSizeInBytes)
require.Equal(t, bulkOps[0], *nodeIndex.BulkTotalOperations)
require.Equal(t, datasetSizes[0], *nodeIndex.TotalDataSetSizeInBytes)
require.Equal(t, dvcCounts[0], *nodeIndex.DenseVectorCount)
require.Equal(t, 1, len(nodeIndex.AssignShards))
require.Equal(t, 0, len(nodeIndex.InitializingShards))
require.EqualValues(t, 0, nodeIndex.Initializing)
Expand Down Expand Up @@ -535,6 +547,11 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
require.Equal(t, searchQueryTimes[1]+searchQueryTimes[2], *nodeIndex.SearchQueryTime)
require.Equal(t, mergeTotals[1]+mergeTotals[2], *nodeIndex.TotalMergesTotal)
require.Equal(t, mergeTimes[1]+mergeTimes[2], *nodeIndex.TotalMergesTotalTime)
// primary-only bulk/dataset/dvc fields: only shard[2] (index 2) is primary for node2
require.Equal(t, bulkSizes[2], *nodeIndex.BulkTotalSizeInBytes)
require.Equal(t, bulkOps[2], *nodeIndex.BulkTotalOperations)
require.Equal(t, datasetSizes[2], *nodeIndex.TotalDataSetSizeInBytes)
require.Equal(t, dvcCounts[2], *nodeIndex.DenseVectorCount)
require.Equal(t, 2, len(nodeIndex.AssignShards))
require.Equal(t, 0, len(nodeIndex.InitializingShards))
require.EqualValues(t, 0, nodeIndex.Initializing)
Expand Down Expand Up @@ -589,6 +606,11 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
require.Nil(t, nodeIndex.IndexingIndexTotalTime)
require.Nil(t, nodeIndex.MergesTotal)
require.Nil(t, nodeIndex.MergesTotalTime)
// replica-only node: no primary-only snapshot fields
require.Nil(t, nodeIndex.BulkTotalSizeInBytes)
require.Nil(t, nodeIndex.BulkTotalOperations)
require.Nil(t, nodeIndex.TotalDataSetSizeInBytes)
require.Nil(t, nodeIndex.DenseVectorCount)
require.Equal(t, getMissingTotals[3], *nodeIndex.GetMissingDocTotal)
require.Equal(t, getMissingTimes[3], *nodeIndex.GetMissingDocTotalTime)
require.Equal(t, searchQueryTotals[3], *nodeIndex.SearchQueryTotal)
Expand Down
4 changes: 4 additions & 0 deletions x-pack/metricbeat/module/autoops_es/node_stats/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,10 @@ var (
createRate("index_rate_per_second", "indices.indexing.index_total"),
createRate("merge_rate_per_second", "indices.merges.total"),
createRate("search_rate_per_second", "indices.search.query_total"),
createRate("ingest_docs_per_second", "indices.docs.count"),
createRate("ingest_bytes_per_second", "indices.store.size_in_bytes"),
createRate("bulk_bytes_per_second", "indices.bulk.total_size_in_bytes"),
createRate("bulk_operations_per_second", "indices.bulk.total_operations"),
// LATENCIES:
createLatency("index_latency_in_millis", "indices.indexing.index_total", "indices.indexing.index_time_in_millis"),
createLatency("merge_latency_in_millis", "indices.merges.total", "indices.merges.total_time_in_millis"),
Expand Down
Loading
Loading