diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc index 2786f60e3a9..88154f73247 100644 --- a/CHANGELOG.next.asciidoc +++ b/CHANGELOG.next.asciidoc @@ -159,6 +159,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d - Add dashboard for `redisenterprise` module. {pull}16752[16752] - Convert increments of 100 nanoseconds/ticks to milliseconds for WriteTime and ReadTime in diskio metricset (Windows) for consistency. {issue}14233[14233] - Dynamically choose a method for the system/service metricset to support older linux distros. {pull}16902[16902] +- Reduce memory usage in `elasticsearch/index` metricset. {issue}16503[16503] {pull}16538[16538] *Packetbeat* diff --git a/metricbeat/module/elasticsearch/index/data_xpack.go b/metricbeat/module/elasticsearch/index/data_xpack.go index 8f81c930f84..160211ea61c 100644 --- a/metricbeat/module/elasticsearch/index/data_xpack.go +++ b/metricbeat/module/elasticsearch/index/data_xpack.go @@ -27,96 +27,117 @@ import ( "github.com/pkg/errors" "github.com/elastic/beats/v7/libbeat/common" - s "github.com/elastic/beats/v7/libbeat/common/schema" - c "github.com/elastic/beats/v7/libbeat/common/schema/mapstriface" "github.com/elastic/beats/v7/metricbeat/helper/elastic" "github.com/elastic/beats/v7/metricbeat/mb" "github.com/elastic/beats/v7/metricbeat/module/elasticsearch" ) var ( - // Based on https://github.com/elastic/elasticsearch/blob/master/x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/indices/IndexStatsMonitoringDoc.java#L127-L203 - xpackSchema = s.Schema{ - "uuid": c.Str("uuid"), - "primaries": c.Dict("primaries", indexStatsSchema), - "total": c.Dict("total", indexStatsSchema), - } + errParse = errors.New("failure parsing Indices Stats Elasticsearch API response") +) - indexStatsSchema = s.Schema{ - "docs": c.Dict("docs", s.Schema{ - "count": c.Int("count"), - }), - "fielddata": c.Dict("fielddata", s.Schema{ - "memory_size_in_bytes": c.Int("memory_size_in_bytes"), - "evictions": c.Int("evictions"), - }), - "indexing": c.Dict("indexing", s.Schema{ - "index_total": c.Int("index_total"), - "index_time_in_millis": c.Int("index_time_in_millis"), - "throttle_time_in_millis": c.Int("throttle_time_in_millis"), - }), - "merges": c.Dict("merges", s.Schema{ - "total_size_in_bytes": c.Int("total_size_in_bytes"), - }), - "query_cache": c.Dict("query_cache", cacheStatsSchema), - "request_cache": c.Dict("request_cache", cacheStatsSchema), - "search": c.Dict("search", s.Schema{ - "query_total": c.Int("query_total"), - "query_time_in_millis": c.Int("query_time_in_millis"), - }), - "segments": c.Dict("segments", s.Schema{ - "count": c.Int("count"), - "memory_in_bytes": c.Int("memory_in_bytes"), - "terms_memory_in_bytes": c.Int("terms_memory_in_bytes"), - "stored_fields_memory_in_bytes": c.Int("stored_fields_memory_in_bytes"), - "term_vectors_memory_in_bytes": c.Int("term_vectors_memory_in_bytes"), - "norms_memory_in_bytes": c.Int("norms_memory_in_bytes"), - "points_memory_in_bytes": c.Int("points_memory_in_bytes"), - "doc_values_memory_in_bytes": c.Int("doc_values_memory_in_bytes"), - "index_writer_memory_in_bytes": c.Int("index_writer_memory_in_bytes"), - "version_map_memory_in_bytes": c.Int("version_map_memory_in_bytes"), - "fixed_bit_set_memory_in_bytes": c.Int("fixed_bit_set_memory_in_bytes"), - }), - "store": c.Dict("store", s.Schema{ - "size_in_bytes": c.Int("size_in_bytes"), - }), - "refresh": c.Dict("refresh", s.Schema{ - "external_total_time_in_millis": c.Int("external_total_time_in_millis", s.Optional), - "total_time_in_millis": c.Int("total_time_in_millis"), - }), - } +// Based on https://github.com/elastic/elasticsearch/blob/master/x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/indices/IndexStatsMonitoringDoc.java#L127-L203 +type stats struct { + Indices map[string]index `json:"indices"` +} - cacheStatsSchema = s.Schema{ - "memory_size_in_bytes": c.Int("memory_size_in_bytes"), - "evictions": c.Int("evictions"), - "hit_count": c.Int("hit_count"), - "miss_count": c.Int("miss_count"), - } -) +type index struct { + UUID string `json:"uuid"` + Primaries indexStats `json:"primaries"` + Total indexStats `json:"total"` -func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, content []byte) error { - var indicesStruct IndicesStruct - if err := parseAPIResponse(content, &indicesStruct); err != nil { - return errors.Wrap(err, "failure parsing Indices Stats Elasticsearch API response") - } + Index string `json:"index"` + Created int64 `json:"created"` + Status string `json:"status"` + Shards shardStats `json:"shards"` +} + +type indexStats struct { + Docs struct { + Count int `json:"count"` + } `json:"docs"` + FieldData struct { + MemorySizeInBytes int `json:"memory_size_in_bytes"` + Evictions int `json:"evictions"` + } `json:"fielddata"` + Indexing struct { + IndexTotal int `json:"index_total"` + IndexTimeInMillis int `json:"index_time_in_millis"` + ThrottleTimeInMillis int `json:"throttle_time_in_millis"` + } `json:"indexing"` + Merges struct { + TotalSizeInBytes int `json:"total_size_in_bytes"` + } `json:"merges"` + QueryCache cacheStats `json:"query_stats"` + RequestCache cacheStats `json:"request_cache"` + Search struct { + QueryTotal int `json:"query_total"` + QueryTimeInMillis int `json:"query_time_in_millis"` + } `json:"search"` + Segments struct { + Count int `json:"count"` + MemoryInBytes int `json:"memory_in_bytes"` + TermsMemoryInBytes int `json:"terms_memory_in_bytes"` + StoredFieldsMemoryInBytes int `json:"stored_fields_memory_in_bytes"` + TermVectorsMemoryInBytes int `json:"term_vectors_memory_in_bytes"` + NormsMemoryInBytes int `json:"norms_memory_in_bytes"` + PointsMemoryInBytes int `json:"points_memory_in_bytes"` + DocValuesMemoryInBytes int `json:"doc_values_memory_in_bytes"` + IndexWriterMemoryInBytes int `json:"index_writer_memory_in_bytes"` + VersionMapMemoryInBytes int `json:"version_map_memory_in_bytes"` + FixedBitSetMemoryInBytes int `json:"fixed_bit_set_memory_in_bytes"` + } `json:"segments"` + Store struct { + SizeInBytes int `json:"size_in_bytes"` + } `json:"store"` + Refresh struct { + ExternalTotalTimeInMillis int `json:"external_total_time_in_millis"` + TotalTimeInMillis int `json:"total_time_in_millis"` + } `json:"refresh"` +} + +type cacheStats struct { + MemorySizeInBytes int `json:"memory_size_in_bytes"` + Evictions int `json:"evictions"` + HitCount int `json:"hit_count"` + MissCount int `json:"miss_count"` +} + +type shardStats struct { + Total int `json:"total"` + Primaries int `json:"primaries"` + Replicas int `json:"replicas"` + ActiveTotal int `json:"active_total"` + ActivePrimaries int `json:"active_primaries"` + ActiveReplicas int `json:"active_replicas"` + + UnassignedTotal int `json:"unassigned_total"` + UnassignedPrimaries int `json:"unassigned_primaries"` + UnassignedReplicas int `json:"unassigned_replicas"` + + Initializing int `json:"initializing"` + Relocating int `json:"relocationg"` +} + +func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, content []byte) error { clusterStateMetrics := []string{"metadata", "routing_table"} clusterState, err := elasticsearch.GetClusterState(m.HTTP, m.HTTP.GetURI(), clusterStateMetrics) if err != nil { return errors.Wrap(err, "failure retrieving cluster state from Elasticsearch") } + var indicesStats stats + if err := parseAPIResponse(content, &indicesStats); err != nil { + return errors.Wrap(err, "failure parsing Indices Stats Elasticsearch API response") + } + var errs multierror.Errors - for name, index := range indicesStruct.Indices { + for name, idx := range indicesStats.Indices { event := mb.Event{} - indexStats, err := xpackSchema.Apply(index) - if err != nil { - errs = append(errs, errors.Wrap(err, "failure applying index stats schema")) - continue - } - indexStats["index"] = name + idx.Index = name - err = addClusterStateFields(name, indexStats, clusterState) + err = addClusterStateFields(&idx, clusterState) if err != nil { errs = append(errs, errors.Wrap(err, "failure adding cluster state fields")) continue @@ -127,7 +148,7 @@ func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, "timestamp": common.Time(time.Now()), "interval_ms": m.Module().Config().Period / time.Millisecond, "type": "index_stats", - "index_stats": indexStats, + "index_stats": idx, } event.Index = elastic.MakeXPackMonitoringIndexName(elastic.Elasticsearch) @@ -137,19 +158,19 @@ func eventsMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, return errs.Err() } -func parseAPIResponse(content []byte, indicesStruct *IndicesStruct) error { - return json.Unmarshal(content, indicesStruct) +func parseAPIResponse(content []byte, indicesStats *stats) error { + return json.Unmarshal(content, indicesStats) } // Fields added here are based on same fields being added by internal collection in // https://github.com/elastic/elasticsearch/blob/master/x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/collector/indices/IndexStatsMonitoringDoc.java#L62-L124 -func addClusterStateFields(indexName string, indexStats, clusterState common.MapStr) error { - indexMetadata, err := getClusterStateMetricForIndex(clusterState, indexName, "metadata") +func addClusterStateFields(idx *index, clusterState common.MapStr) error { + indexMetadata, err := getClusterStateMetricForIndex(clusterState, idx.Index, "metadata") if err != nil { return errors.Wrap(err, "failed to get index metadata from cluster state") } - indexRoutingTable, err := getClusterStateMetricForIndex(clusterState, indexName, "routing_table") + indexRoutingTable, err := getClusterStateMetricForIndex(clusterState, idx.Index, "routing_table") if err != nil { return errors.Wrap(err, "failed to get index routing table from cluster state") } @@ -163,7 +184,7 @@ func addClusterStateFields(indexName string, indexStats, clusterState common.Map if err != nil { return errors.Wrap(err, "failed to get index creation time") } - indexStats.Put("created", created) + idx.Created = created // "index_stats.version.created", <--- don't think this is being used in the UI, so can we skip it? // "index_stats.version.upgraded", <--- don't think this is being used in the UI, so can we skip it? @@ -172,13 +193,13 @@ func addClusterStateFields(indexName string, indexStats, clusterState common.Map if err != nil { return errors.Wrap(err, "failed to get index status") } - indexStats.Put("status", status) + idx.Status = status shardStats, err := getIndexShardStats(shards) if err != nil { return errors.Wrap(err, "failed to get index shard stats") } - indexStats.Put("shards", shardStats) + idx.Shards = *shardStats return nil } @@ -241,7 +262,7 @@ func getIndexStatus(shards map[string]interface{}) (string, error) { return "red", nil } -func getIndexShardStats(shards common.MapStr) (common.MapStr, error) { +func getIndexShardStats(shards common.MapStr) (*shardStats, error) { primaries := 0 replicas := 0 @@ -298,21 +319,18 @@ func getIndexShardStats(shards common.MapStr) (common.MapStr, error) { } } - return common.MapStr{ - "total": primaries + replicas, - "primaries": primaries, - "replicas": replicas, - - "active_total": activePrimaries + activeReplicas, - "active_primaries": activePrimaries, - "active_replicas": activeReplicas, - - "unassigned_total": unassignedPrimaries + unassignedReplicas, - "unassigned_primaries": unassignedPrimaries, - "unassigned_replicas": unassignedReplicas, - - "initializing": initializing, - "relocating": relocating, + return &shardStats{ + Total: primaries + replicas, + Primaries: primaries, + Replicas: replicas, + ActiveTotal: activePrimaries + activeReplicas, + ActivePrimaries: activePrimaries, + ActiveReplicas: activeReplicas, + UnassignedTotal: unassignedPrimaries + unassignedReplicas, + UnassignedPrimaries: unassignedPrimaries, + UnassignedReplicas: unassignedReplicas, + Initializing: initializing, + Relocating: relocating, }, nil } diff --git a/metricbeat/module/elasticsearch/index/data_xpack_test.go b/metricbeat/module/elasticsearch/index/data_xpack_test.go index 749467ef72d..1bdc790d9ae 100644 --- a/metricbeat/module/elasticsearch/index/data_xpack_test.go +++ b/metricbeat/module/elasticsearch/index/data_xpack_test.go @@ -31,10 +31,10 @@ func BenchmarkParseAPIResponse(b *testing.B) { content, err := ioutil.ReadFile("_meta/test/stats.800.bench.json") require.NoError(b, err) - var indicesStruct IndicesStruct + var indicesStats stats for i := 0; i < b.N; i++ { - err = parseAPIResponse(content, &indicesStruct) + err = parseAPIResponse(content, &indicesStats) require.NoError(b, err) } diff --git a/metricbeat/module/elasticsearch/index/index.go b/metricbeat/module/elasticsearch/index/index.go index 6936b836b16..cd2dc3ffca0 100644 --- a/metricbeat/module/elasticsearch/index/index.go +++ b/metricbeat/module/elasticsearch/index/index.go @@ -35,7 +35,7 @@ func init() { const ( statsMetrics = "docs,fielddata,indexing,merge,search,segments,store,refresh,query_cache,request_cache" - statsPath = "/_stats/" + statsMetrics + statsPath = "/_stats/" + statsMetrics + "?filter_path=indices" ) // MetricSet type defines all fields of the MetricSet