Skip to content

[AutoOps] Serialize custom struct before sending as event #45015

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

Merged
pickypg merged 2 commits into from
Jun 25, 2025
Merged

[AutoOps] Serialize custom struct before sending as event #45015

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c3510cc
[AutoOps] Serialize custom struct before sending as event
pickypg Jun 24, 2025
b07e45f
Serialize NodeShardCounts directly without JSON marshalling
pickypg Jun 24, 2025
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
8 changes: 3 additions & 5 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 @@ -11,8 +11,6 @@ import (

"github.com/elastic/beats/v7/x-pack/metricbeat/module/autoops_es/events"

"golang.org/x/exp/maps"

"github.com/elastic/elastic-agent-libs/mapstr"

"github.com/elastic/beats/v7/metricbeat/mb"
Expand Down Expand Up @@ -82,7 +80,7 @@ func eventsMapping(m *elasticsearch.MetricSet, r mb.ReporterV2, info *utils.Clus

transactionID := utils.NewUUIDV4()

sendNodeShardsEvent(r, info, maps.Values(nodeShards), transactionID)
sendNodeShardsEvent(r, info, nodeShardCountToMapArray(nodeShards), transactionID)

indexMetadata, err := getResolvedIndices(m)

Expand All @@ -97,7 +95,7 @@ func eventsMapping(m *elasticsearch.MetricSet, r mb.ReporterV2, info *utils.Clus
return err
}

func sendNodeShardsEvent(r mb.ReporterV2, info *utils.ClusterInfo, nodeToShards []NodeShardCount, transactionId string) {
func sendNodeShardsEvent(r mb.ReporterV2, info *utils.ClusterInfo, nodeToShards []map[string]any, transactionId string) {
r.Event(events.CreateEvent(info, mapstr.M{"node_shards_count": nodeToShards}, transactionId))
}

Expand All @@ -111,7 +109,7 @@ func sendNodeIndexShardsEvent(r mb.ReporterV2, info *utils.ClusterInfo, nodeInde
for i := 0; i < size; i += nodeIndexShardsPerEvent {
group := nodeIndexShards[i:min(i+nodeIndexShardsPerEvent, size)]

groups = append(groups, mapstr.M{"node_index_shards": group})
groups = append(groups, mapstr.M{"node_index_shards": convertObjectArrayToMapArray(group)})
}

events.CreateAndReportEvents(r, info, groups, transactionId)
Expand Down
45 changes: 23 additions & 22 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 @@ -29,8 +29,8 @@ var (
func TestSendNodeShardsEvent(t *testing.T) {
reporter := &mbtest.CapturingReporterV2{}
info := auto_ops_testing.CreateClusterInfo("8.15.3")
nodeToShards := []NodeShardCount{
{
nodeToShards := map[string]NodeShardCount{
"node1": {
NodeId: "node1",
NodeName: "name1",
Shards: 100,
Expand All @@ -43,7 +43,7 @@ func TestSendNodeShardsEvent(t *testing.T) {
RelocatingPrimaryShards: 1,
RelocatingReplicaShards: 2,
},
{
"node2": {
NodeId: "node2",
NodeName: "name2",
Shards: 99,
Expand All @@ -59,7 +59,7 @@ func TestSendNodeShardsEvent(t *testing.T) {
}
transactionId := "xyz"

sendNodeShardsEvent(reporter, &info, nodeToShards, transactionId)
sendNodeShardsEvent(reporter, &info, nodeShardCountToMapArray(nodeToShards), transactionId)

require.Equal(t, 0, len(reporter.GetErrors()))
require.Equal(t, 1, len(reporter.GetEvents()))
Expand All @@ -68,7 +68,7 @@ func TestSendNodeShardsEvent(t *testing.T) {

auto_ops_testing.CheckEventWithTransactionId(t, event, info, transactionId)

require.ElementsMatch(t, nodeToShards, auto_ops_testing.GetObjectValue(event.MetricSetFields, "node_shards_count"))
require.ElementsMatch(t, maps.Values(nodeToShards), mapArrayToType[NodeShardCount](auto_ops_testing.GetObjectValue(event.MetricSetFields, "node_shards_count").([]map[string]any)))
}

func TestSendNodeIndexShardsEventInBatch(t *testing.T) {
Expand All @@ -86,7 +86,7 @@ func TestSendNodeIndexShardsEventInBatch(t *testing.T) {

auto_ops_testing.CheckEventWithTransactionId(t, event, info, transactionId)

require.ElementsMatch(t, nodeIndexShards, auto_ops_testing.GetObjectValue(event.MetricSetFields, "node_index_shards"))
require.ElementsMatch(t, convertObjectArrayToMapArray(nodeIndexShards), auto_ops_testing.GetObjectValue(event.MetricSetFields, "node_index_shards"))
}

func TestSendNodeIndexShardsEvent(t *testing.T) {
Expand All @@ -106,16 +106,16 @@ func TestSendNodeIndexShardsEvent(t *testing.T) {

auto_ops_testing.CheckAllEventsUseSameTransactionId(t, events)

eventData := []NodeIndexShards{}
eventData := []map[string]any{}

for _, event := range events {
auto_ops_testing.CheckEventWithTransactionId(t, event, info, transactionId)

array := auto_ops_testing.GetObjectValue(event.MetricSetFields, "node_index_shards")
eventData = append(eventData, array.([]NodeIndexShards)...)
eventData = append(eventData, array.([]map[string]any)...)
}

require.ElementsMatch(t, nodeIndexShards, eventData)
require.ElementsMatch(t, convertObjectArrayToMapArray(nodeIndexShards), eventData)
}

func expectValidParsedData(t *testing.T, data metricset.FetcherData[[]JSONShard]) {
Expand All @@ -131,12 +131,12 @@ func expectValidParsedData(t *testing.T, data metricset.FetcherData[[]JSONShard]
nodeShardsCountEvents := auto_ops_testing.GetEventsWithField(t, events, "node_shards_count")

require.Equal(t, 1, len(nodeShardsCountEvents))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]NodeShardCount)))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]map[string]any)))

nodeIndexShardsEvents := auto_ops_testing.GetEventsWithField(t, events, "node_index_shards")

require.Equal(t, 1, len(nodeIndexShardsEvents))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any)))
}

func expectValidParsedDetailedShards(t *testing.T, data metricset.FetcherData[[]JSONShard]) {
Expand All @@ -147,7 +147,7 @@ func expectValidParsedDetailedShards(t *testing.T, data metricset.FetcherData[[]
require.Equal(t, 2, len(events))

nodeShardCountsEvents := auto_ops_testing.GetEventsWithField(t, events, "node_shards_count")
nodeShardCounts := auto_ops_testing.GetObjectValue(nodeShardCountsEvents[0].MetricSetFields, "node_shards_count").([]NodeShardCount)
nodeShardCounts := mapArrayToType[NodeShardCount](auto_ops_testing.GetObjectValue(nodeShardCountsEvents[0].MetricSetFields, "node_shards_count").([]map[string]any))

require.Equal(t, 1, len(nodeShardCountsEvents))
require.Equal(t, 2, len(nodeShardCounts))
Expand All @@ -167,17 +167,18 @@ func expectValidParsedDetailedShards(t *testing.T, data metricset.FetcherData[[]
require.EqualValues(t, 0, node2.RelocatingReplicaShards)

nodeIndexShardsEvents := auto_ops_testing.GetEventsWithField(t, events, "node_index_shards")
nodeIndexShards := auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)
nodeIndexShards := mapArrayToType[NodeIndexShards](auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any))

require.Equal(t, 1, len(nodeIndexShardsEvents))
require.LessOrEqual(t, 14, len(nodeIndexShards))

myIndexNode2 := nodeIndexShards[slices.IndexFunc(nodeIndexShards, func(node NodeIndexShards) bool { return node.IndexNode == "my-index-node_id-node2" })]

if data.Version == "7.17.0" {
switch data.Version {
case "7.17.0":
require.Equal(t, 14, len(nodeIndexShards))
require.EqualValues(t, 14, myIndexNode2.TotalFractions)
} else if data.Version == "8.15.3" {
case "8.15.3":
require.Equal(t, 35, len(nodeIndexShards))
require.EqualValues(t, 35, myIndexNode2.TotalFractions)
}
Expand Down Expand Up @@ -235,7 +236,7 @@ func expectValidParsedDetailedShardsWithCache(t *testing.T, data metricset.Fetch
expectValidParsedDetailedShards(t, data)

nodeIndexShardsEvents := auto_ops_testing.GetEventsWithField(t, data.Reporter.GetEvents(), "node_index_shards")
nodeIndexShards := auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)
nodeIndexShards := mapArrayToType[NodeIndexShards](auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any))
myIndexNode2 := nodeIndexShards[slices.IndexFunc(nodeIndexShards, func(node NodeIndexShards) bool { return node.IndexNode == "my-index-node_id-node2" })]

require.NotNil(t, myIndexNode2.TimestampDiff)
Expand Down Expand Up @@ -263,14 +264,14 @@ func expectValidParsedWithoutResolvedIndexDataWithoutElasticSearchError(t *testi
nodeShardsCountEvents := auto_ops_testing.GetEventsWithField(t, events, "node_shards_count")

require.Equal(t, 1, len(nodeShardsCountEvents))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]NodeShardCount)))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]map[string]any)))

nodeIndexShardsEvents := auto_ops_testing.GetEventsWithField(t, events, "node_index_shards")

require.Equal(t, 1, len(nodeIndexShardsEvents))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any)))

nodeIndexShards := auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)
nodeIndexShards := mapArrayToType[NodeIndexShards](auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any))
myIndexNode2 := nodeIndexShards[slices.IndexFunc(nodeIndexShards, func(node NodeIndexShards) bool { return node.IndexNode == "my-index-node_id-node2" })]

require.Nil(t, myIndexNode2.IndexType)
Expand All @@ -293,14 +294,14 @@ func expectValidParsedWithoutResolvedIndexDataWithElasticSearchError(t *testing.
nodeShardsCountEvents := auto_ops_testing.GetEventsWithField(t, events, "node_shards_count")

require.Equal(t, 1, len(nodeShardsCountEvents))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]NodeShardCount)))
require.Equal(t, 2, len(auto_ops_testing.GetObjectValue(nodeShardsCountEvents[0].MetricSetFields, "node_shards_count").([]map[string]any)))

nodeIndexShardsEvents := auto_ops_testing.GetEventsWithField(t, events, "node_index_shards")

require.Equal(t, 1, len(nodeIndexShardsEvents))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)))
require.LessOrEqual(t, 2, len(auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any)))

nodeIndexShards := auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]NodeIndexShards)
nodeIndexShards := mapArrayToType[NodeIndexShards](auto_ops_testing.GetObjectValue(nodeIndexShardsEvents[0].MetricSetFields, "node_index_shards").([]map[string]any))
myIndexNode2 := nodeIndexShards[slices.IndexFunc(nodeIndexShards, func(node NodeIndexShards) bool { return node.IndexNode == "my-index-node_id-node2" })]

require.Nil(t, myIndexNode2.IndexType)
Expand Down
22 changes: 22 additions & 0 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 @@ -210,6 +210,28 @@ func appendNodeShards(nodeShards map[string]NodeShardCount, shard *Shard) {
nodeShards[shard.node_id] = node
}

func nodeShardCountToMapArray(nodeShards map[string]NodeShardCount) []map[string]any {
mapArray := make([]map[string]any, 0, len(nodeShards))

for _, nodeShard := range nodeShards {
mapArray = append(mapArray, map[string]any{
"node_id": nodeShard.NodeId,
"node_name": nodeShard.NodeName,
"shards_count": nodeShard.Shards,
"primary_shards": nodeShard.PrimaryShards,
"replica_shards": nodeShard.ReplicaShards,
"initializing_shards": nodeShard.InitializingShards,
"initializing_primary_shards": nodeShard.InitializingPrimaryShards,
"initializing_replica_shards": nodeShard.InitializingReplicaShards,
"relocating_shards": nodeShard.RelocatingShards,
"relocating_primary_shards": nodeShard.RelocatingPrimaryShards,
"relocating_replica_shards": nodeShard.RelocatingReplicaShards,
})
}

return mapArray
}

func indexShardsToNodeIndexShards(nodeIndexShardsMap map[string]NodeIndexShards, index string, shards []Shard) {
status := GREEN
indexStatus := &status
Expand Down
80 changes: 80 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 @@ -615,6 +615,86 @@ func TestIndexShardsToNodeIndexShardsGreenIndex(t *testing.T) {
require.Equal(t, STARTED, assignedShard4.State)
}

func TestNodeShardCountToMapArray(t *testing.T) {
// Create a map with multiple node shard counts
nodeShards := map[string]NodeShardCount{
"node1": {
NodeId: "node1",
NodeName: "node-1",
Shards: 10,
PrimaryShards: 5,
ReplicaShards: 5,
InitializingShards: 2,
InitializingPrimaryShards: 1,
InitializingReplicaShards: 1,
RelocatingShards: 3,
RelocatingPrimaryShards: 1,
RelocatingReplicaShards: 2,
},
"node2": {
NodeId: "node2",
NodeName: "node-2",
Shards: 8,
PrimaryShards: 4,
ReplicaShards: 4,
InitializingShards: 1,
InitializingPrimaryShards: 1,
InitializingReplicaShards: 0,
RelocatingShards: 0,
RelocatingPrimaryShards: 0,
RelocatingReplicaShards: 0,
},
}

// Call the function
mapArray := nodeShardCountToMapArray(nodeShards)

// Verify the results
require.Equal(t, 2, len(mapArray))

// Check that we have entries for both nodes
foundNode1 := false
foundNode2 := false

for _, nodeMap := range mapArray {
if nodeId, ok := nodeMap["node_id"].(string); ok {
switch nodeId {
case "node1":
foundNode1 = true
require.Equal(t, "node-1", nodeMap["node_name"])
require.EqualValues(t, 10, nodeMap["shards_count"])
require.EqualValues(t, 5, nodeMap["primary_shards"])
require.EqualValues(t, 5, nodeMap["replica_shards"])
require.EqualValues(t, 2, nodeMap["initializing_shards"])
require.EqualValues(t, 1, nodeMap["initializing_primary_shards"])
require.EqualValues(t, 1, nodeMap["initializing_replica_shards"])
require.EqualValues(t, 3, nodeMap["relocating_shards"])
require.EqualValues(t, 1, nodeMap["relocating_primary_shards"])
require.EqualValues(t, 2, nodeMap["relocating_replica_shards"])
case "node2":
foundNode2 = true
require.Equal(t, "node-2", nodeMap["node_name"])
require.EqualValues(t, 8, nodeMap["shards_count"])
require.EqualValues(t, 4, nodeMap["primary_shards"])
require.EqualValues(t, 4, nodeMap["replica_shards"])
require.EqualValues(t, 1, nodeMap["initializing_shards"])
require.EqualValues(t, 1, nodeMap["initializing_primary_shards"])
require.EqualValues(t, 0, nodeMap["initializing_replica_shards"])
require.EqualValues(t, 0, nodeMap["relocating_shards"])
require.EqualValues(t, 0, nodeMap["relocating_primary_shards"])
require.EqualValues(t, 0, nodeMap["relocating_replica_shards"])
default:
require.Fail(t, "Unexpected node in map array", nodeId)
}
} else {
require.Fail(t, "Unexpected node in map array", nodeId)
}
}

require.True(t, foundNode1, "Node1 should be in the map array")
require.True(t, foundNode2, "Node2 should be in the map array")
}

func TestIndexShardsToNodeIndexShardsRelocating(t *testing.T) {
docs := int64(100001)
segments := int64(10)
Expand Down
27 changes: 27 additions & 0 deletions x-pack/metricbeat/module/autoops_es/cat_shards/struct_to_map.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
// or more contributor license agreements. Licensed under the Elastic License;
// you may not use this file except in compliance with the Elastic License.

package cat_shards

import (
"encoding/json"
)

// convertObjectArrayToMapArray converts an array of structs to an array of maps one by one.
func convertObjectArrayToMapArray[T any](objects []T) []map[string]any {
mapArray := make([]map[string]any, 0, len(objects))

for _, object := range objects {
if data, err := json.Marshal(object); err == nil {
// Unmarshal the JSON into a map
var result map[string]any

if err := json.Unmarshal(data, &result); err == nil {
mapArray = append(mapArray, result)
}
}
}

return mapArray
}
Loading