From 04036dc6a6b154570959116799d0aa14d27eb8f4 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 15 Jul 2026 18:34:33 +0200 Subject: [PATCH 01/18] fix: migrate plugin to official ES clients --- go.mod | 2 +- go.sum | 11 +- plugins/inputs/elasticsearch_query/client.go | 521 ++++++++++++++++++ .../inputs/elasticsearch_query/client_v5.go | 349 ++---------- .../inputs/elasticsearch_query/client_v6.go | 54 ++ .../elasticsearch_query.go | 23 +- .../elasticsearch_query_test.go | 75 ++- 7 files changed, 679 insertions(+), 356 deletions(-) create mode 100644 plugins/inputs/elasticsearch_query/client.go create mode 100644 plugins/inputs/elasticsearch_query/client_v6.go diff --git a/go.mod b/go.mod index 6bee0fcf19277..0135159da8978 100644 --- a/go.mod +++ b/go.mod @@ -88,6 +88,7 @@ require ( github.com/eclipse/paho.golang v0.23.0 github.com/eclipse/paho.mqtt.golang v1.5.1 github.com/elastic/go-elasticsearch/v5 v5.6.1 + github.com/elastic/go-elasticsearch/v6 v6.8.10 github.com/emiago/sipgo v1.4.3 github.com/facebook/time v0.0.0-20250903103710-a5911c32cdb9 github.com/fatih/color v1.19.0 @@ -252,7 +253,6 @@ require ( google.golang.org/grpc v1.82.0 google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af gopkg.in/gorethink/gorethink.v3 v3.0.5 - gopkg.in/olivere/elastic.v5 v5.0.86 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 k8s.io/api v0.36.2 k8s.io/apimachinery v0.36.2 diff --git a/go.sum b/go.sum index cc3ed568a10cc..0fd000a9fce02 100644 --- a/go.sum +++ b/go.sum @@ -900,7 +900,6 @@ github.com/awnumar/memguard v0.23.0/go.mod h1:olVofBrsPdITtJ2HgxQKrEYEMyIBAIciVG github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4 h1:2jAwFwA0Xgcx94dUId+K24yFabsKYDtAhCgyMit6OqE= github.com/aws/aws-msk-iam-sasl-signer-go v1.0.4/go.mod h1:MVYeeOhILFFemC/XlYTClvBjYZrg/EPd3ts885KrNTI= github.com/aws/aws-sdk-go v1.20.6/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo= -github.com/aws/aws-sdk-go v1.29.11/go.mod h1:1KvfttTE3SPKMpo8g2c6jL3ZKfXtFvKscTgahTma5Xg= github.com/aws/aws-sdk-go v1.44.263/go.mod h1:aVsgQcEevwlmQ7qHE9I3h+dtQgpqhFB+i8Phjh7fkwI= github.com/aws/aws-sdk-go-v2 v1.18.0/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw= github.com/aws/aws-sdk-go-v2 v1.42.1 h1:9eOTgu1z/dVtYpNZ3/8/XbbaX0x/BqE3HUzAzs6K0ek= @@ -1188,6 +1187,8 @@ github.com/eclipse/paho.mqtt.golang v1.5.1 h1:/VSOv3oDLlpqR2Epjn1Q7b2bSTplJIeV2I github.com/eclipse/paho.mqtt.golang v1.5.1/go.mod h1:1/yJCneuyOoCOzKSsOTUc0AJfpsItBGWvYpBLimhArU= github.com/elastic/go-elasticsearch/v5 v5.6.1 h1:RnL2wcXepOT5SdoKMMO1j1OBX0vxHYbBtkQNL2E3xs4= github.com/elastic/go-elasticsearch/v5 v5.6.1/go.mod h1:r7uV7HidpfkYh7D8SB4lkS13TNlNy3oa5GNmTZvuVqY= +github.com/elastic/go-elasticsearch/v6 v6.8.10 h1:2lN0gJ93gMBXvkhwih5xquldszpm8FlUwqG5sPzr6a8= +github.com/elastic/go-elasticsearch/v6 v6.8.10/go.mod h1:UwaDJsD3rWLM5rKNFzv9hgox93HoX8utj1kxD9aFUcI= github.com/elastic/go-sysinfo v1.8.1 h1:4Yhj+HdV6WjbCRgGdZpPJ8lZQlXZLKDAeIkmQ/VRvi4= github.com/elastic/go-sysinfo v1.8.1/go.mod h1:JfllUnzoQV/JRYymbH3dO1yggI3mV2oTKSXsDHM+uIM= github.com/elastic/go-windows v1.0.0 h1:qLURgZFkkrYyTTkvYpsZIgf83AUsdIHfvlJaqaZ7aSY= @@ -1355,7 +1356,6 @@ github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo= github.com/go-resty/resty/v2 v2.17.2 h1:FQW5oHYcIlkCNrMD2lloGScxcHJ0gkjshV3qcQAyHQk= github.com/go-resty/resty/v2 v2.17.2/go.mod h1:kCKZ3wWmwJaNc7S29BRtUhJwy7iqmn+2mLtQrOyQlVA= -github.com/go-sql-driver/mysql v1.5.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LBy8hT2VhHyBg= github.com/go-sql-driver/mysql v1.10.0 h1:Q+1LV8DkHJvSYAdR83XzuhDaTykuDx0l6fkXxoWCWfw= github.com/go-sql-driver/mysql v1.10.0/go.mod h1:M+cqaI7+xxXGG9swrdeUIoPG3Y3KCkF0pZej+SK+nWk= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= @@ -1876,7 +1876,6 @@ github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czP github.com/magiconair/properties v1.8.4/go.mod h1:y3VJvCyxH9uVvJTWEGAELF3aiYNyPKd5NZ3oSwXrF60= github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= -github.com/mailru/easyjson v0.7.1/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs= github.com/mailru/easyjson v0.9.2 h1:dX8U45hQsZpxd80nLvDGihsQ/OxlvTkVUXH2r/8cb2M= github.com/mailru/easyjson v0.9.2/go.mod h1:1+xMtQp2MRNVL/V1bOzuP3aP8VNwRW55fQUto+XFtTU= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= @@ -2056,7 +2055,6 @@ github.com/oklog/ulid/v2 v2.1.1/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNs github.com/oliveagle/jsonpath v0.0.0-20180606110733-2e52cf6e6852/go.mod h1:eqOVx5Vwu4gd2mmMZvVZsgIqNSaW3xxRThUJ0k/TPk4= github.com/olivere/elastic v6.2.37+incompatible h1:UfSGJem5czY+x/LqxgeCBgjDn6St+z8OnsCuxwD3L0U= github.com/olivere/elastic v6.2.37+incompatible/go.mod h1:J+q1zQJTgAz9woqsbVRqGeB5G1iqDKVBWLNSYW8yfJ8= -github.com/olivere/elastic/v7 v7.0.12/go.mod h1:14rWX28Pnh3qCKYRVnSGXWLf9MbLonYS/4FDCY3LAPo= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= @@ -2103,7 +2101,6 @@ github.com/opensearch-project/opensearch-go/v2 v2.3.0 h1:nQIEMr+A92CkhHrZgUhcfsr github.com/opensearch-project/opensearch-go/v2 v2.3.0/go.mod h1:8LDr9FCgUTVoT+5ESjc2+iaZuldqE+23Iq0r1XeNue8= github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492 h1:lM6RxxfUMrYL/f8bWEUqdXrANWtrL7Nndbm9iFN0DlU= github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492/go.mod h1:Ngi6UdF0k5OKD5t5wlmGhe/EDKPoUM3BXZSSfIuJbis= -github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b h1:FfH+VrHHk6Lxt9HdVS0PXzSXFyS2NbZKXv33FYPol0A= github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b/go.mod h1:AC62GU6hc0BrNm+9RK9VSiwa/EUe1bkIeFORAMcHvJU= github.com/openzipkin-contrib/zipkin-go-opentracing v0.5.0 h1:uhcF5Jd7rP9DVEL10Siffyepr6SvlKbUsjH5JpNCRi8= @@ -2329,14 +2326,12 @@ github.com/smarty/assertions v1.16.0 h1:EvHNkdRA4QHMrn75NZSoUQ/mAUXAYWfatfB01yTC github.com/smarty/assertions v1.16.0/go.mod h1:duaaFdCS0K9dnoM50iyek/eYINOZ64gbh1Xlf6LG7AI= github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= github.com/smartystreets/assertions v1.0.0/go.mod h1:kHHU4qYBaI3q23Pp3VPrmWhuIUrLW/7eUrw0BU5VaoM= -github.com/smartystreets/assertions v1.0.1/go.mod h1:kHHU4qYBaI3q23Pp3VPrmWhuIUrLW/7eUrw0BU5VaoM= github.com/smartystreets/assertions v1.2.0/go.mod h1:tcbTF8ujkAEcZ8TElKY+i30BzYlVhC/LOxJk7iOWnoo= github.com/smartystreets/go-aws-auth v0.0.0-20180515143844-0c1422d1fdb9/go.mod h1:SnhjPscd9TpLiy1LpzGSKh3bXCfxxXuqd9xmQJy3slM= github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= github.com/smartystreets/goconvey v1.8.1 h1:qGjIddxOk4grTu9JPOU31tVfq3cNdBlNa5sSznIX1xY= github.com/smartystreets/goconvey v1.8.1/go.mod h1:+/u4qLyY6x1jReYOp7GOM2FSt8aP9CzCZL03bI28W60= github.com/smartystreets/gunit v1.0.0/go.mod h1:qwPWnhz6pn0NnRBP++URONOVyNkPyr4SauJk4cUOwJs= -github.com/smartystreets/gunit v1.1.3/go.mod h1:EH5qMBab2UclzXUcpR8b93eHsIlp9u+pDQIRp5DZNzQ= github.com/snowflakedb/gosnowflake v1.19.1 h1:NZMErtdZMu6kooehbONNQmu/W5BPsaX8hYdlBBEHgxs= github.com/snowflakedb/gosnowflake v1.19.1/go.mod h1:9vGW6LYbUD1UqfjpuNN5a5vtha+u4n1AlsR1BqhHwPA= github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM= @@ -3468,8 +3463,6 @@ gopkg.in/ini.v1 v1.67.2 h1:JtOSMb9OuaCZKr7h5D/h6iii14sK0hLbplTc6frx4Ss= gopkg.in/ini.v1 v1.67.2/go.mod h1:x/cyOwCgZqOkJoDIJ3c1KNHMo10+nLGAhh+kn3Zizss= gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= -gopkg.in/olivere/elastic.v5 v5.0.86 h1:xFy6qRCGAmo5Wjx96srho9BitLhZl2fcnpuidPwduXM= -gopkg.in/olivere/elastic.v5 v5.0.86/go.mod h1:M3WNlsF+WhYn7api4D87NIflwTV/c0iVs8cqfWhK+68= gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo= gopkg.in/sourcemap.v1 v1.0.5 h1:inv58fC9f9J3TK2Y2R1NPntXEn3/wjWHkonhIUODNTI= gopkg.in/sourcemap.v1 v1.0.5/go.mod h1:2RlvNNSMglmRrcvhfuzp4hQHwOtjxlbjX7UPY/GXb78= diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go new file mode 100644 index 0000000000000..e4fc875c7673a --- /dev/null +++ b/plugins/inputs/elasticsearch_query/client.go @@ -0,0 +1,521 @@ +package elasticsearch_query + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strconv" + "strings" + "sync" + "time" + + elasticsearch5 "github.com/elastic/go-elasticsearch/v5" + + "github.com/influxdata/telegraf" +) + +type clientConfig struct { + urls []string + username string + password string + enableSniffer bool + discoveryInterval time.Duration + httpClient *http.Client + log telegraf.Logger +} + +func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { + // Use the v5 client only for the version-agnostic GET / probe. + probe, err := elasticsearch5.NewClient(elasticsearch5.Config{ + Addresses: cfg.urls, + Username: cfg.username, + Password: cfg.password, + Transport: roundTripper{client: cfg.httpClient}, + }) + if err != nil { + return "", 0, fmt.Errorf("creating Elasticsearch client failed: %w", err) + } + + res, err := probe.Info(probe.Info.WithContext(ctx)) + if err != nil { + return "", 0, err + } + response := &esResponse{statusCode: res.StatusCode, body: res.Body} + + var info struct { + Version struct { + Number string `json:"number"` + } `json:"version"` + } + if err := response.handle(nil, &info); err != nil { + return "", 0, err + } + + majorText, _, _ := strings.Cut(info.Version.Number, ".") + major, err := strconv.Atoi(majorText) + if err != nil { + return "", 0, fmt.Errorf("parsing server version %q failed: %w", info.Version.Number, err) + } + + return info.Version.Number, major, nil +} + +type esResponse struct { + statusCode int + body io.ReadCloser +} + +func (r *esResponse) handle(err error, out interface{}) error { + if err != nil { + return err + } + if r == nil { + return nil + } + defer r.body.Close() + + if r.statusCode > 299 { + data, err := io.ReadAll(r.body) + if err != nil { + return err + } + + var result struct { + Error struct { + Type string `json:"type"` + Reason string `json:"reason"` + } `json:"error"` + } + if err := json.Unmarshal(data, &result); err == nil { + if result.Error.Type != "" && result.Error.Reason != "" { + return fmt.Errorf("%s - %s", result.Error.Type, result.Error.Reason) + } + if result.Error.Reason != "" { + return fmt.Errorf("error %d (%s): %s", r.statusCode, http.StatusText(r.statusCode), result.Error.Reason) + } + } + + return fmt.Errorf("error %d (%s): %s", r.statusCode, http.StatusText(r.statusCode), strings.TrimSpace(string(data))) + } + if out != nil { + return json.NewDecoder(r.body).Decode(out) + } + + return nil +} + +type roundTripper struct { + client *http.Client +} + +// RoundTrip implements http.RoundTripper. +func (t roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + return t.client.Do(req) +} + +// queryClient provides shared query behavior. Per-major adapters wrap their +// distinct response types, which expose status and body as fields. +type queryClient struct { + httpClient *http.Client + log telegraf.Logger + closed bool + stopDiscovery func() + + getFieldMappingResponse func(context.Context, string, string) (*esResponse, error) + search func(context.Context, string, io.Reader) (*esResponse, error) +} + +func (e *ElasticsearchQuery) newClient() (client, error) { + // Make sure the HTTP client exists + httpClient, err := e.HTTPClientConfig.CreateClient(context.Background(), e.Log) + if err != nil { + return nil, fmt.Errorf("creating HTTP client failed: %w", err) + } + + cfg := clientConfig{ + urls: e.URLs, + username: e.Username, + password: e.Password, + enableSniffer: e.EnableSniffer, + discoveryInterval: time.Duration(e.HealthCheckInterval), + httpClient: httpClient, + log: e.Log, + } + + version, major, err := cfg.probeVersion(context.Background()) + if err != nil { + httpClient.CloseIdleConnections() + return nil, fmt.Errorf("getting server version failed: %w", err) + } + + switch major { + case 5: + return newClientV5(cfg) + case 6: + return newClientV6(cfg) + default: + httpClient.CloseIdleConnections() + return nil, fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) + } +} + +func (c *queryClient) close() { + c.closed = true + if c.stopDiscovery != nil { + c.stopDiscovery() + c.stopDiscovery = nil + } + if c.httpClient != nil { + c.httpClient.CloseIdleConnections() + c.httpClient = nil + } +} + +func (c *queryClient) startDiscovery(interval time.Duration, discover func(context.Context) error) { + ctx, cancel := context.WithCancel(context.Background()) + + var wg sync.WaitGroup + wg.Go(func() { + if err := discover(ctx); err != nil && ctx.Err() == nil { + c.log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + } + if interval <= 0 { + return + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if err := discover(ctx); err != nil && ctx.Err() == nil { + c.log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + } + } + } + }) + + c.stopDiscovery = func() { + cancel() + wg.Wait() + } +} + +func (c *queryClient) isRunning() bool { + return !c.closed +} + +func (c *queryClient) getFieldMapping(ctx context.Context, index, field string) (map[string]interface{}, error) { + res, err := c.getFieldMappingResponse(ctx, index, field) + var result map[string]interface{} + if err := res.handle(err, &result); err != nil { + return nil, err + } + return result, nil +} + +type queryData struct { + measurement string + name string + function string + isParent bool + aggregation map[string]interface{} +} + +func (q *queryData) addSubAggregation(name string, subAggregation map[string]interface{}) { + aggs, ok := q.aggregation["aggs"].(map[string]interface{}) + if !ok { + aggs = make(map[string]interface{}) + q.aggregation["aggs"] = aggs + } + aggs[name] = subAggregation +} + +func (*queryClient) buildQueries(aggregation *aggregation) error { + // Create one aggregation per metric field found or function defined for + // numeric fields + queries := make([]queryData, 0, len(aggregation.mapMetricFields)+len(aggregation.Tags)) + for k, v := range aggregation.mapMetricFields { + switch v { + case "long", "float", "integer", "short", "double", "scaled_float": + default: + continue + } + + var agg map[string]interface{} + switch aggregation.MetricFunction { + case "avg", "sum", "min", "max": + agg = map[string]interface{}{ + aggregation.MetricFunction: map[string]interface{}{ + "field": k, + }, + } + default: + return fmt.Errorf("aggregation function %q not supported", aggregation.MetricFunction) + } + + query := queryData{ + measurement: aggregation.MeasurementName, + function: aggregation.MetricFunction, + name: strings.ReplaceAll(k, ".", "_") + "_" + aggregation.MetricFunction, + isParent: true, + aggregation: agg, + } + queries = append(queries, query) + } + + // Create a terms aggregation per tag + for _, term := range aggregation.Tags { + terms := map[string]interface{}{ + "field": term, + "size": 1000, + } + if aggregation.IncludeMissingTag && aggregation.MissingTagValue != "" { + terms["missing"] = aggregation.MissingTagValue + } + query := queryData{ + measurement: aggregation.MeasurementName, + function: "terms", + name: strings.ReplaceAll(term, ".", "_"), + isParent: true, + aggregation: map[string]interface{}{"terms": terms}, + } + + // add each previous parent aggregations as subaggregations of this terms aggregation + for key, q := range queries { + if !q.isParent { + continue + } + + query.addSubAggregation(q.name, q.aggregation) + + // Update subaggregation map with parent information + queries[key].isParent = false + } + + queries = append(queries, query) + } + aggregation.queries = queries + + // Prepare measurement mapping to organize the aggregation query data + // by measurement + measurements := make(map[string]map[string]string, len(queries)) + for _, query := range queries { + nameFunctions, ok := measurements[query.measurement] + if !ok { + nameFunctions = make(map[string]string) + measurements[query.measurement] = nameFunctions + } + nameFunctions[query.name] = query.function + } + aggregation.measurements = measurements + + return nil +} + +func (a *aggregation) buildRangeQuery(from, to time.Time) map[string]interface{} { + rangeQuery := map[string]interface{}{ + "gte": from, + "lte": to, + } + if a.DateFieldFormat != "" { + rangeQuery["format"] = a.DateFieldFormat + } + return rangeQuery +} + +type searchResponse struct { + Hits struct { + Total json.RawMessage `json:"total"` + } `json:"hits"` + Aggregations map[string]json.RawMessage `json:"aggregations"` +} + +func (r *searchResponse) totalHits() int64 { + var total int64 + if err := json.Unmarshal(r.Hits.Total, &total); err == nil { + return total + } + + // Elasticsearch 7 and later return hits.total as an object. + var totalObject struct { + Value int64 `json:"value"` + } + if err := json.Unmarshal(r.Hits.Total, &totalObject); err == nil { + return totalObject.Value + } + + return 0 +} + +func (c *queryClient) query(ctx context.Context, aggregation *aggregation) (interface{}, int64, error) { + // buildQueries stores []queryData in this field before query execution. + // If the assertion fails, it indicates a programming error in this package. + queries := aggregation.queries.([]queryData) + + now := time.Now().UTC() + from := now.Add(-time.Duration(aggregation.QueryPeriod)) + + query := map[string]interface{}{ + "bool": map[string]interface{}{ + "filter": []interface{}{ + map[string]interface{}{ + "query_string": map[string]interface{}{ + "query": aggregation.FilterQuery, + }, + }, + map[string]interface{}{ + "range": map[string]interface{}{ + aggregation.DateField: aggregation.buildRangeQuery(from, now), + }, + }, + }, + }, + } + + data, err := json.Marshal(query) + if err != nil { + return nil, 0, fmt.Errorf("marshal request failed: %w", err) + } + c.log.Debugf("{\"query\": %s}", string(data)) + + body := map[string]interface{}{ + "query": query, + "size": 0, + } + + aggs := make(map[string]interface{}) + for _, v := range queries { + if v.isParent && v.aggregation != nil { + aggs[v.name] = v.aggregation + } + } + if len(aggs) > 0 { + body["aggs"] = aggs + } + + data, err = json.Marshal(body) + if err != nil { + return nil, 0, fmt.Errorf("marshal request failed: %w", err) + } + + res, err := c.search(ctx, aggregation.Index, bytes.NewReader(data)) + var result searchResponse + if err := res.handle(err, &result); err != nil { + return nil, 0, err + } + + if len(result.Aggregations) == 0 { + return nil, result.totalHits(), nil + } + + return result.Aggregations, result.totalHits(), nil +} + +type aggregationIterator struct { + name string + fields map[string]interface{} + tags map[string]string +} + +func (m *aggregationIterator) iterate(acc telegraf.Accumulator, nameFunction map[string]string, response map[string]json.RawMessage) error { + names := make([]string, 0, len(response)) + for k := range response { + if k != "key" && k != "doc_count" { + names = append(names, k) + } + } + if len(names) == 0 { + // We've reached a single bucket or response without aggregation, i.e. + // we've reached a leaf node. Add the accumulated metric and reset it + if len(m.fields) > 0 { + acc.AddFields(m.name, m.fields, m.tags) + m.fields = make(map[string]interface{}) + } + return nil + } + + // Metrics aggregations response can contain multiple field values, so we + // iterate over them + for _, name := range names { + function, found := nameFunction[name] + if !found { + return fmt.Errorf("child aggregation function %q not found %v", name, nameFunction) + } + + // Execute the aggregation function + switch function { + case "avg", "sum", "min", "max": + var result struct { + Value *float64 `json:"value"` + } + if err := json.Unmarshal(response[name], &result); err != nil { + return err + } + if result.Value != nil { + m.fields[name] = *result.Value + } else { + m.fields[name] = float64(0) + } + case "terms": + var result struct { + Buckets []map[string]json.RawMessage `json:"buckets"` + } + if err := json.Unmarshal(response[name], &result); err != nil { + return err + } + + // We've found a terms aggregation, iterate over the buckets and try + // to retrieve the inner aggregation values + for _, bucket := range result.Buckets { + var key string + if err := json.Unmarshal(bucket["key"], &key); err != nil { + return fmt.Errorf("bucket key is not a string (%s, %s)", name, function) + } + m.tags[name] = key + + var docCount int64 + if err := json.Unmarshal(bucket["doc_count"], &docCount); err != nil { + return err + } + m.fields["doc_count"] = docCount + + // We need to recurse down through the buckets, as it may + // contain another terms aggregation + if err := m.iterate(acc, nameFunction, bucket); err != nil { + return err + } + delete(m.tags, name) + } + default: + return fmt.Errorf("aggregation %q not supported", function) + } + } + + // If there are fields here it comes from a metrics aggregation without a + // parent terms aggregation + if len(m.fields) > 0 { + acc.AddFields(m.name, m.fields, m.tags) + m.fields = make(map[string]interface{}) + } + + return nil +} + +func (*queryClient) aggregate(acc telegraf.Accumulator, measurement string, nameFunction map[string]string, response interface{}) error { + // The query method returns map[string]json.RawMessage for aggregation responses. + r := response.(map[string]json.RawMessage) + + m := &aggregationIterator{ + name: measurement, + fields: make(map[string]interface{}), + tags: make(map[string]string), + } + + return m.iterate(acc, nameFunction, r) +} diff --git a/plugins/inputs/elasticsearch_query/client_v5.go b/plugins/inputs/elasticsearch_query/client_v5.go index 81b4f058ef2fc..fc302b765f335 100644 --- a/plugins/inputs/elasticsearch_query/client_v5.go +++ b/plugins/inputs/elasticsearch_query/client_v5.go @@ -2,321 +2,54 @@ package elasticsearch_query import ( "context" - "encoding/json" + "errors" "fmt" - "net/http" - "strings" - "time" + "io" - elastic5 "gopkg.in/olivere/elastic.v5" - - "github.com/influxdata/telegraf" + elasticsearch5 "github.com/elastic/go-elasticsearch/v5" ) -type clientV5 struct { - url string - httpClient *http.Client - client *elastic5.Client - log telegraf.Logger -} - -func (e *ElasticsearchQuery) newClientV5() (client, error) { - // Make sure the HTTP client exists - httpClient, err := e.HTTPClientConfig.CreateClient(context.Background(), e.Log) - if err != nil { - return nil, fmt.Errorf("creating HTTP client failed: %w", err) - } - - // Create a new ElasticSearch client - clientOptions := []elastic5.ClientOptionFunc{ - elastic5.SetHttpClient(httpClient), - elastic5.SetSniff(e.EnableSniffer), - elastic5.SetURL(e.URLs...), - elastic5.SetHealthcheckInterval(time.Duration(e.HealthCheckInterval)), - } - if e.Username != "" { - clientOptions = append(clientOptions, elastic5.SetBasicAuth(e.Username, e.Password)) - } - if time.Duration(e.HealthCheckInterval) == 0 { - clientOptions = append(clientOptions, elastic5.SetHealthcheck(false)) - } - - c, err := elastic5.NewClient(clientOptions...) - if err != nil { - return nil, fmt.Errorf("creating ElasticSearch client failed: %w", err) - } - - return &clientV5{ - url: e.URLs[0], - httpClient: httpClient, - client: c, - log: e.Log, - }, nil -} - -func (c *clientV5) close() { - if c.client != nil { - c.client.Stop() - c.client = nil - } - if c.httpClient != nil { - c.httpClient.CloseIdleConnections() - c.httpClient = nil - } -} - -func (c *clientV5) version() (string, error) { - return c.client.ElasticsearchVersion(c.url) -} - -func (c *clientV5) isRunning() bool { - return c.client.IsRunning() -} - -func (c *clientV5) getFieldMapping(ctx context.Context, index, field string) (map[string]interface{}, error) { - return c.client.GetFieldMapping().Index(index).Field(field).Do(ctx) -} - -type queryDataV5 struct { - measurement string - name string - function string - field string - isParent bool - aggregation elastic5.Aggregation -} - -func (*clientV5) buildQueries(aggregation *aggregation) error { - // Create one aggregation per metric field found or function defined for - // numeric fields - queries := make([]queryDataV5, 0, len(aggregation.mapMetricFields)+len(aggregation.Tags)) - for k, v := range aggregation.mapMetricFields { - switch v { - case "long", "float", "integer", "short", "double", "scaled_float": - default: - continue - } - - var agg elastic5.Aggregation - switch aggregation.MetricFunction { - case "avg": - agg = elastic5.NewAvgAggregation().Field(k) - case "sum": - agg = elastic5.NewSumAggregation().Field(k) - case "min": - agg = elastic5.NewMinAggregation().Field(k) - case "max": - agg = elastic5.NewMaxAggregation().Field(k) - default: - return fmt.Errorf("aggregation function %q not supported", aggregation.MetricFunction) - } - - query := queryDataV5{ - measurement: aggregation.MeasurementName, - function: aggregation.MetricFunction, - field: k, - name: strings.ReplaceAll(k, ".", "_") + "_" + aggregation.MetricFunction, - isParent: true, - aggregation: agg, - } - queries = append(queries, query) - } - - // Create a terms aggregation per tag - for _, term := range aggregation.Tags { - agg := elastic5.NewTermsAggregation() - if aggregation.IncludeMissingTag && aggregation.MissingTagValue != "" { - agg.Missing(aggregation.MissingTagValue) - } - agg.Field(term).Size(1000) - - // add each previous parent aggregations as subaggregations of this terms aggregation - for key, aggMap := range queries { - if !aggMap.isParent { - continue - } - - agg.Field(term).SubAggregation(aggMap.name, aggMap.aggregation).Size(1000) - - // Update subaggregation map with parent information - queries[key].isParent = false - } - - query := queryDataV5{ - measurement: aggregation.MeasurementName, - function: "terms", - field: term, - name: strings.ReplaceAll(term, ".", "_"), - isParent: true, - aggregation: agg, - } - queries = append(queries, query) - } - aggregation.queries = queries - - // Prepare measurement mapping to organize the aggregation query data - // by measurement - measurements := make(map[string]map[string]string, len(queries)) - for _, query := range queries { - if measurements[query.measurement] == nil { - measurements[query.measurement] = map[string]string{ - query.name: query.function, - } - } else { - measurements[query.measurement][query.name] = query.function - } +func newClientV5(cfg clientConfig) (client, error) { + if cfg.enableSniffer { + cfg.httpClient.CloseIdleConnections() + return nil, errors.New("enable_sniffer is not supported by the official Elasticsearch v5 client") } - aggregation.measurements = measurements - - return nil -} - -func (c *clientV5) query(ctx context.Context, aggregation *aggregation) (interface{}, int64, error) { - // Type assert client-specific data - // This can only be the expected type as it is set by this code with exactly - // the expected type. If the assertion fails code will panic here but that's - // okay as it can only happen on programming errors e.g. by mixing different - // client versions! - queries := aggregation.queries.([]queryDataV5) - now := time.Now().UTC() - from := now.Add(time.Duration(-aggregation.QueryPeriod)) - - query := elastic5.NewBoolQuery() - query = query.Filter(elastic5.NewQueryStringQuery(aggregation.FilterQuery)) - query = query.Filter(elastic5.NewRangeQuery(aggregation.DateField).From(from).To(now).Format(aggregation.DateFieldFormat)) - - src, err := query.Source() + c, err := elasticsearch5.NewClient(elasticsearch5.Config{ + Addresses: cfg.urls, + Username: cfg.username, + Password: cfg.password, + Transport: roundTripper{client: cfg.httpClient}, + }) if err != nil { - return nil, 0, fmt.Errorf("getting query source failed: %w", err) - } - data, err := json.Marshal(src) - if err != nil { - return nil, 0, fmt.Errorf("unmarshal response failed: %w", err) - } - c.log.Debugf("{\"query\": %s}", string(data)) - - // Add only parent elastic.Aggregations to the search request, all the rest - // are subaggregations of these - search := c.client.Search().Index(aggregation.Index).Query(query).Size(0) - for _, v := range queries { - if v.isParent && v.aggregation != nil { - search.Aggregation(v.name, v.aggregation) - } - } - - result, err := search.Do(ctx) - if err != nil { - if result != nil { - return result.Aggregations, result.Hits.TotalHits, fmt.Errorf("%s - %s", result.Error.Type, result.Error.Reason) - } - return nil, 0, err - } - - if len(result.Aggregations) == 0 { - return nil, result.Hits.TotalHits, nil - } - - return result.Aggregations, result.Hits.TotalHits, nil -} - -func (*clientV5) aggregate(acc telegraf.Accumulator, measurement string, nameFunction map[string]string, response interface{}) error { - // Type assert client-specific data - r := response.(elastic5.Aggregations) - - m := &iteratorV5{ - name: measurement, - fields: make(map[string]interface{}), - tags: make(map[string]string), - } - - return m.iterate(acc, nameFunction, r) -} - -type iteratorV5 struct { - name string - fields map[string]interface{} - tags map[string]string -} - -func (m *iteratorV5) iterate(acc telegraf.Accumulator, nameFunction map[string]string, response elastic5.Aggregations) error { - names := make([]string, 0, len(response)) - for k := range response { - if k != "key" && k != "doc_count" { - names = append(names, k) - } - } - if len(names) == 0 { - // We've reached a single bucket or response without aggregation, i.e. - // we've reached a leaf node. Add the accumulated metric and reset it - if len(m.fields) > 0 { - acc.AddFields(m.name, m.fields, m.tags) - m.fields = make(map[string]interface{}) - } - return nil - } - - // Metrics aggregations response can contain multiple field values, so we - // iterate over them - for _, name := range names { - function, found := nameFunction[name] - if !found { - return fmt.Errorf("child aggregation function %q not found %v", name, nameFunction) - } - - // Execute the aggregation function - var result interface{} - switch function { - case "avg": - result, _ = response.Avg(name) - case "sum": - result, _ = response.Sum(name) - case "min": - result, _ = response.Min(name) - case "max": - result, _ = response.Max(name) - case "terms": - result, _ = response.Terms(name) - default: - return fmt.Errorf("aggregation %q not supported", function) - } - - switch r := result.(type) { - case *elastic5.AggregationBucketKeyItems: - // We've found a terms aggregation, iterate over the buckets and try - // to retrieve the inner aggregation values - for _, bucket := range r.Buckets { - s, ok := bucket.Key.(string) - if !ok { - return fmt.Errorf("bucket key is not a string (%s, %s)", name, function) - } - m.tags[name] = s - m.fields["doc_count"] = bucket.DocCount - - // We need to recurse down through the buckets, as it may - // contain another terms aggregation - if err := m.iterate(acc, nameFunction, bucket.Aggregations); err != nil { - return err - } - delete(m.tags, name) + cfg.httpClient.CloseIdleConnections() + return nil, fmt.Errorf("creating Elasticsearch client failed: %w", err) + } + + return &queryClient{ + httpClient: cfg.httpClient, + log: cfg.log, + getFieldMappingResponse: func(ctx context.Context, index, field string) (*esResponse, error) { + res, err := c.Indices.GetFieldMapping( + []string{field}, + c.Indices.GetFieldMapping.WithContext(ctx), + c.Indices.GetFieldMapping.WithIndex(index), + ) + if res == nil { + return nil, err } - case *elastic5.AggregationValueMetric: - if r.Value != nil { - m.fields[name] = *r.Value - } else { - m.fields[name] = float64(0) + return &esResponse{statusCode: res.StatusCode, body: res.Body}, err + }, + search: func(ctx context.Context, index string, body io.Reader) (*esResponse, error) { + res, err := c.Search( + c.Search.WithContext(ctx), + c.Search.WithIndex(index), + c.Search.WithBody(body), + ) + if res == nil { + return nil, err } - default: - return fmt.Errorf("aggregation type %T not supported", r) - } - } - - // If there are fields here it comes from a metrics aggregation without a - // parent terms aggregation - if len(m.fields) > 0 { - acc.AddFields(m.name, m.fields, m.tags) - m.fields = make(map[string]interface{}) - } - - return nil + return &esResponse{statusCode: res.StatusCode, body: res.Body}, err + }, + }, nil } diff --git a/plugins/inputs/elasticsearch_query/client_v6.go b/plugins/inputs/elasticsearch_query/client_v6.go new file mode 100644 index 0000000000000..dbdc4018c3b06 --- /dev/null +++ b/plugins/inputs/elasticsearch_query/client_v6.go @@ -0,0 +1,54 @@ +package elasticsearch_query + +import ( + "context" + "fmt" + "io" + + elasticsearch6 "github.com/elastic/go-elasticsearch/v6" +) + +func newClientV6(cfg clientConfig) (client, error) { + c, err := elasticsearch6.NewClient(elasticsearch6.Config{ + Addresses: cfg.urls, + Username: cfg.username, + Password: cfg.password, + Transport: roundTripper{client: cfg.httpClient}, + }) + if err != nil { + cfg.httpClient.CloseIdleConnections() + return nil, fmt.Errorf("creating Elasticsearch client failed: %w", err) + } + + client := &queryClient{ + httpClient: cfg.httpClient, + log: cfg.log, + getFieldMappingResponse: func(ctx context.Context, index, field string) (*esResponse, error) { + res, err := c.Indices.GetFieldMapping( + []string{field}, + c.Indices.GetFieldMapping.WithContext(ctx), + c.Indices.GetFieldMapping.WithIndex(index), + ) + if res == nil { + return nil, err + } + return &esResponse{statusCode: res.StatusCode, body: res.Body}, err + }, + search: func(ctx context.Context, index string, body io.Reader) (*esResponse, error) { + res, err := c.Search( + c.Search.WithContext(ctx), + c.Search.WithIndex(index), + c.Search.WithBody(body), + ) + if res == nil { + return nil, err + } + return &esResponse{statusCode: res.StatusCode, body: res.Body}, err + }, + } + if cfg.enableSniffer { + // The v6 client exposes only DiscoverNodes(), so in-flight calls cannot be canceled. + client.startDiscovery(cfg.discoveryInterval, func(context.Context) error { return c.DiscoverNodes() }) + } + return client, nil +} diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index 1a3ee9eb8abfd..0148756b12c71 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -10,8 +10,6 @@ import ( "sync" "time" - "github.com/Masterminds/semver/v3" - "github.com/influxdata/telegraf" "github.com/influxdata/telegraf/config" common_http "github.com/influxdata/telegraf/plugins/common/http" @@ -19,7 +17,6 @@ import ( ) type client interface { - version() (string, error) isRunning() bool close() buildQueries(aggregation *aggregation) error @@ -59,7 +56,7 @@ type aggregation struct { mapMetricFields map[string]string measurements map[string]map[string]string - queries interface{} // client specific data to execute the query + queries interface{} // query data built during initialization } func (*ElasticsearchQuery) SampleConfig() string { @@ -90,28 +87,12 @@ func (e *ElasticsearchQuery) Init() error { func (e *ElasticsearchQuery) Start(telegraf.Accumulator) error { // Create a new ElasticSearch client - client, err := e.newClientV5() + client, err := e.newClient() if err != nil { return err } e.client = client - // Get the ElasticSearch version on first node and check if it's supported - version, err := e.client.version() - if err != nil { - e.Stop() - return fmt.Errorf("getting server version failed: %w", err) - } - ver, err := semver.NewVersion(version) - if err != nil { - e.Stop() - return fmt.Errorf("parsing server version %q failed: %w", version, err) - } - if ver.Major() < 5 || ver.Major() > 6 { - e.Stop() - return fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) - } - // Setup the aggregations, this needs to be done in Start as it will require // API calls to the ElasticSearch endpoint and can thus not happen in Init ctx, cancel := context.WithTimeout(context.Background(), time.Duration(e.Timeout)) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index 268111e2ab9c9..ec4c958935880 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -187,17 +187,48 @@ func TestGatherIntegration(t *testing.T) { ), } + tests := []struct { + name string + image string + env map[string]string + }{ + { + name: "v5", + image: "elasticsearch:5.6.16", + env: map[string]string{"ES_JAVA_OPTS": "-Xms512m -Xmx512m"}, + }, + { + name: "v6", + image: "elasticsearch:6.8.23", + env: map[string]string{ + "discovery.type": "single-node", + "ES_JAVA_OPTS": "-Xms512m -Xmx512m", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + runGatherIntegration(t, tt.image, tt.env, expectedFields, expectedMetrics) + }) + } +} + +func runGatherIntegration( + t *testing.T, + image string, + env map[string]string, + expectedFields []map[string]string, + expectedMetrics []telegraf.Metric, +) { + t.Helper() + // Setup the container container := &testutil.Container{ - Image: "elasticsearch:6.8.23", + Image: image, ExposedPorts: []string{servicePort}, - Env: map[string]string{ - "discovery.type": "single-node", - }, - WaitingFor: wait.ForAll( - wait.ForLog("] mode [basic] - valid"), - wait.ForListeningPort(servicePort), - ), + Env: env, + WaitingFor: wait.ForHTTP("/").WithPort(servicePort).WithStartupTimeout(5 * time.Minute), } require.NoError(t, container.Start(), "failed to start container") defer container.Terminate() @@ -492,27 +523,37 @@ func TestGatherFailGatherIntegration(t *testing.T) { } func TestStartupFailureReleasesClient(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") - if _, err := w.Write([]byte(`{"version": {"number": "8.1.2"}}`)); err != nil { + if r.URL.Path != "/" { + http.Error(w, `{"error":{"type":"test_error","reason":"mapping failed"}}`, http.StatusInternalServerError) + return + } + if _, err := w.Write([]byte(`{"version":{"number":"6.8.23"}}`)); err != nil { t.Error(err) } })) defer server.Close() plugin := &ElasticsearchQuery{ - URLs: []string{server.URL}, - HealthCheckInterval: config.Duration(10 * time.Second), - Log: testutil.Logger{}, + URLs: []string{server.URL}, + Aggregations: []aggregation{{ + Index: "test", + MeasurementName: "test", + DateField: "@timestamp", + MetricFields: []string{"value"}, + MetricFunction: "avg", + }}, + Log: testutil.Logger{}, } require.NoError(t, plugin.Init()) var acc testutil.Accumulator - require.ErrorContains(t, plugin.Start(&acc), "not supported") + require.ErrorContains(t, plugin.Start(&acc), "initializing aggregation") - // The failed start must release the client to not leak the - // health-check goroutine - require.Nil(t, plugin.client.(*clientV5).client) + // The failed start must release the client resources. + require.False(t, plugin.client.isRunning()) + require.Nil(t, plugin.client.(*queryClient).httpClient) } func sendData(ctx context.Context, url string) error { From 19d43c9ad266645fbb9815031287265192ad7b8c Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Thu, 16 Jul 2026 11:39:21 +0200 Subject: [PATCH 02/18] chore: update dependencies licenses file --- docs/LICENSE_OF_DEPENDENCIES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/LICENSE_OF_DEPENDENCIES.md b/docs/LICENSE_OF_DEPENDENCIES.md index fff320227cc48..8591e9f48e648 100644 --- a/docs/LICENSE_OF_DEPENDENCIES.md +++ b/docs/LICENSE_OF_DEPENDENCIES.md @@ -149,6 +149,7 @@ following works: - github.com/ebitengine/purego [Apache License 2.0](https://github.com/ebitengine/purego/blob/main/LICENSE) - github.com/eclipse/paho.golang [Eclipse Public License - v 2.0](https://github.com/eclipse/paho.golang/blob/master/LICENSE) - github.com/eclipse/paho.mqtt.golang [Eclipse Public License - v 2.0](https://github.com/eclipse/paho.mqtt.golang/blob/master/LICENSE) +- github.com/elastic/go-elasticsearch [Apache License 2.0](https://github.com/elastic/go-elasticsearch/blob/main/LICENSE) - github.com/elastic/go-sysinfo [Apache License 2.0](https://github.com/elastic/go-sysinfo/blob/main/LICENSE.txt) - github.com/elastic/go-windows [Apache License 2.0](https://github.com/elastic/go-windows/blob/main/LICENSE.txt) - github.com/emiago/sipgo [BSD 2-Clause "Simplified" License](https://github.com/emiago/sipgo/blob/main/LICENSE) @@ -515,7 +516,6 @@ following works: - gopkg.in/inf.v0 [BSD 3-Clause "New" or "Revised" License](https://github.com/go-inf/inf/blob/v0.9.1/LICENSE) - gopkg.in/ini.v1 [Apache License 2.0](https://github.com/go-ini/ini/blob/master/LICENSE) - gopkg.in/natefinch/lumberjack.v2 [MIT License](https://github.com/natefinch/lumberjack/blob/v2.0/LICENSE) -- gopkg.in/olivere/elastic.v5 [MIT License](https://github.com/olivere/elastic/blob/v5.0.76/LICENSE) - gopkg.in/tomb.v1 [BSD 3-Clause Clear License](https://github.com/go-tomb/tomb/blob/v1/LICENSE) - gopkg.in/tomb.v2 [BSD 3-Clause Clear License](https://github.com/go-tomb/tomb/blob/v2/LICENSE) - gopkg.in/yaml.v2 [Apache License 2.0](https://github.com/go-yaml/yaml/blob/v2.2.2/LICENSE) From 7c1dd275b7c3b1df585bae354990dbc223c3ab00 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Thu, 16 Jul 2026 11:50:08 +0200 Subject: [PATCH 03/18] fix: preserve error details --- plugins/inputs/elasticsearch_query/client.go | 36 +++++++---- .../elasticsearch_query_test.go | 62 +++++++++++++++++++ 2 files changed, 87 insertions(+), 11 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index e4fc875c7673a..f01f6f0e4932e 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -69,13 +70,13 @@ type esResponse struct { } func (r *esResponse) handle(err error, out interface{}) error { - if err != nil { - return err - } if r == nil { - return nil + return err } defer r.body.Close() + if err != nil { + return err + } if r.statusCode > 299 { data, err := io.ReadAll(r.body) @@ -89,16 +90,29 @@ func (r *esResponse) handle(err error, out interface{}) error { Reason string `json:"reason"` } `json:"error"` } - if err := json.Unmarshal(data, &result); err == nil { - if result.Error.Type != "" && result.Error.Reason != "" { - return fmt.Errorf("%s - %s", result.Error.Type, result.Error.Reason) - } - if result.Error.Reason != "" { - return fmt.Errorf("error %d (%s): %s", r.statusCode, http.StatusText(r.statusCode), result.Error.Reason) + if err := json.Unmarshal(data, &result); err == nil && result.Error.Reason != "" { + msg := fmt.Sprintf( + "elastic: Error %d (%s): %s", + r.statusCode, + http.StatusText(r.statusCode), + result.Error.Reason, + ) + if result.Error.Type != "" { + msg += " [type=" + result.Error.Type + "]" } + return errors.New(msg) + } + + if detail := strings.TrimSpace(string(data)); detail != "" { + return fmt.Errorf( + "elastic: Error %d (%s): %s", + r.statusCode, + http.StatusText(r.statusCode), + detail, + ) } - return fmt.Errorf("error %d (%s): %s", r.statusCode, http.StatusText(r.statusCode), strings.TrimSpace(string(data))) + return fmt.Errorf("elastic: Error %d (%s)", r.statusCode, http.StatusText(r.statusCode)) } if out != nil { return json.NewDecoder(r.body).Decode(out) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index ec4c958935880..ff499d20ec24b 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -34,6 +34,49 @@ const ( testindex = "test-elasticsearch" ) +func TestResponseHandleError(t *testing.T) { + tests := []struct { + name string + statusCode int + body string + expected string + }{ + { + name: "type and reason", + statusCode: http.StatusNotFound, + body: `{"error":{"type":"index_not_found_exception","reason":"no such index"}}`, + expected: "elastic: Error 404 (Not Found): no such index [type=index_not_found_exception]", + }, + { + name: "reason without type", + statusCode: http.StatusBadGateway, + body: `{"error":{"reason":"upstream failed"}}`, + expected: "elastic: Error 502 (Bad Gateway): upstream failed", + }, + { + name: "unstructured response", + statusCode: http.StatusBadGateway, + body: " proxy failed\n", + expected: "elastic: Error 502 (Bad Gateway): proxy failed", + }, + { + name: "empty response", + statusCode: http.StatusBadGateway, + expected: "elastic: Error 502 (Bad Gateway)", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + response := &esResponse{ + statusCode: tt.statusCode, + body: io.NopCloser(strings.NewReader(tt.body)), + } + require.EqualError(t, response.handle(nil, nil), tt.expected) + }) + } +} + type nginxlog struct { IPaddress string `json:"IP"` Timestamp time.Time `json:"@timestamp"` @@ -522,6 +565,25 @@ func TestGatherFailGatherIntegration(t *testing.T) { } } +func TestUnsupportedVersion(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if _, err := w.Write([]byte(`{"version":{"number":"8.1.2"}}`)); err != nil { + t.Error(err) + } + })) + defer server.Close() + + plugin := &ElasticsearchQuery{ + URLs: []string{server.URL}, + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Init()) + + var acc testutil.Accumulator + require.ErrorContains(t, plugin.Start(&acc), `server version "8.1.2" not supported`) +} + func TestStartupFailureReleasesClient(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") From 8f763fe9c83e524b636fdec4258625d075801373 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Thu, 16 Jul 2026 12:45:07 +0200 Subject: [PATCH 04/18] fix: improve code comment --- plugins/inputs/elasticsearch_query/elasticsearch_query.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index 0148756b12c71..638775fd35d11 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -56,7 +56,7 @@ type aggregation struct { mapMetricFields map[string]string measurements map[string]map[string]string - queries interface{} // query data built during initialization + queries interface{} // prepared once and reused across collections } func (*ElasticsearchQuery) SampleConfig() string { From 059f78e5713abdbb82fa57eaaf8cffa17124e893 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Thu, 16 Jul 2026 13:15:41 +0200 Subject: [PATCH 05/18] chore: trigger CI workflow From e8f7e3ad20b91ed7c686eb4ff2117a8fcfa9f3e5 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Fri, 17 Jul 2026 15:30:06 +0200 Subject: [PATCH 06/18] refactor: simplify official client integration --- plugins/inputs/elasticsearch_query/client.go | 257 +++++------- .../inputs/elasticsearch_query/client_v5.go | 105 +++-- .../inputs/elasticsearch_query/client_v6.go | 109 +++-- .../elasticsearch_query.go | 49 ++- .../elasticsearch_query_test.go | 376 ++++++++++++++---- 5 files changed, 578 insertions(+), 318 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index f01f6f0e4932e..4959518e45d9d 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -1,18 +1,16 @@ package elasticsearch_query import ( - "bytes" "context" "encoding/json" - "errors" "fmt" "io" "net/http" - "strconv" "strings" "sync" "time" + "github.com/Masterminds/semver/v3" elasticsearch5 "github.com/elastic/go-elasticsearch/v5" "github.com/influxdata/telegraf" @@ -28,6 +26,14 @@ type clientConfig struct { log telegraf.Logger } +type serverVersion struct { + Number string `json:"number"` +} + +type serverInfo struct { + Version serverVersion `json:"version"` +} + func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { // Use the v5 client only for the version-agnostic GET / probe. probe, err := elasticsearch5.NewClient(elasticsearch5.Config{ @@ -37,164 +43,101 @@ func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { Transport: roundTripper{client: cfg.httpClient}, }) if err != nil { - return "", 0, fmt.Errorf("creating Elasticsearch client failed: %w", err) + return "", 0, fmt.Errorf("creating ElasticSearch client failed: %w", err) } res, err := probe.Info(probe.Info.WithContext(ctx)) if err != nil { return "", 0, err } - response := &esResponse{statusCode: res.StatusCode, body: res.Body} + defer res.Body.Close() - var info struct { - Version struct { - Number string `json:"number"` - } `json:"version"` + if err := checkForError(res.StatusCode, res.Body); err != nil { + return "", 0, err } - if err := response.handle(nil, &info); err != nil { + + var info serverInfo + if err := json.NewDecoder(res.Body).Decode(&info); err != nil { return "", 0, err } - majorText, _, _ := strings.Cut(info.Version.Number, ".") - major, err := strconv.Atoi(majorText) + version, err := semver.NewVersion(info.Version.Number) if err != nil { return "", 0, fmt.Errorf("parsing server version %q failed: %w", info.Version.Number, err) } - return info.Version.Number, major, nil + return info.Version.Number, int(version.Major()), nil +} + +type apiErrorDetails struct { + Type string `json:"type"` + Reason string `json:"reason"` } -type esResponse struct { +type apiErrorResponse struct { + Error apiErrorDetails `json:"error"` +} + +type apiError struct { statusCode int - body io.ReadCloser + errorType string + reason string } -func (r *esResponse) handle(err error, out interface{}) error { - if r == nil { - return err +func (e *apiError) Error() string { + msg := fmt.Sprintf("received error %d (%s)", e.statusCode, http.StatusText(e.statusCode)) + if e.reason != "" { + msg += ": " + e.reason } - defer r.body.Close() - if err != nil { - return err + if e.errorType != "" { + msg += " [type=" + e.errorType + "]" } + return msg +} - if r.statusCode > 299 { - data, err := io.ReadAll(r.body) - if err != nil { - return err - } +func checkForError(statusCode int, body io.Reader) error { + if statusCode <= 299 { + return nil + } - var result struct { - Error struct { - Type string `json:"type"` - Reason string `json:"reason"` - } `json:"error"` - } - if err := json.Unmarshal(data, &result); err == nil && result.Error.Reason != "" { - msg := fmt.Sprintf( - "elastic: Error %d (%s): %s", - r.statusCode, - http.StatusText(r.statusCode), - result.Error.Reason, - ) - if result.Error.Type != "" { - msg += " [type=" + result.Error.Type + "]" - } - return errors.New(msg) - } + data, err := io.ReadAll(body) + if err != nil { + return err + } - if detail := strings.TrimSpace(string(data)); detail != "" { - return fmt.Errorf( - "elastic: Error %d (%s): %s", - r.statusCode, - http.StatusText(r.statusCode), - detail, - ) + var response apiErrorResponse + if err := json.Unmarshal(data, &response); err == nil && response.Error.Reason != "" { + return &apiError{ + statusCode: statusCode, + errorType: response.Error.Type, + reason: response.Error.Reason, } - - return fmt.Errorf("elastic: Error %d (%s)", r.statusCode, http.StatusText(r.statusCode)) - } - if out != nil { - return json.NewDecoder(r.body).Decode(out) } - return nil + return &apiError{statusCode: statusCode, reason: strings.TrimSpace(string(data))} } type roundTripper struct { client *http.Client } -// RoundTrip implements http.RoundTripper. +// RoundTrip delegates to the configured HTTP client to preserve its overall +// timeout, including response-body reads, and cookie and redirect handling +// that the underlying transport does not provide. func (t roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { return t.client.Do(req) } -// queryClient provides shared query behavior. Per-major adapters wrap their -// distinct response types, which expose status and body as fields. -type queryClient struct { - httpClient *http.Client - log telegraf.Logger - closed bool - stopDiscovery func() - - getFieldMappingResponse func(context.Context, string, string) (*esResponse, error) - search func(context.Context, string, io.Reader) (*esResponse, error) -} - -func (e *ElasticsearchQuery) newClient() (client, error) { - // Make sure the HTTP client exists - httpClient, err := e.HTTPClientConfig.CreateClient(context.Background(), e.Log) - if err != nil { - return nil, fmt.Errorf("creating HTTP client failed: %w", err) - } - - cfg := clientConfig{ - urls: e.URLs, - username: e.Username, - password: e.Password, - enableSniffer: e.EnableSniffer, - discoveryInterval: time.Duration(e.HealthCheckInterval), - httpClient: httpClient, - log: e.Log, - } - - version, major, err := cfg.probeVersion(context.Background()) - if err != nil { - httpClient.CloseIdleConnections() - return nil, fmt.Errorf("getting server version failed: %w", err) - } - - switch major { - case 5: - return newClientV5(cfg) - case 6: - return newClientV6(cfg) - default: - httpClient.CloseIdleConnections() - return nil, fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) - } -} - -func (c *queryClient) close() { - c.closed = true - if c.stopDiscovery != nil { - c.stopDiscovery() - c.stopDiscovery = nil - } - if c.httpClient != nil { - c.httpClient.CloseIdleConnections() - c.httpClient = nil - } -} - -func (c *queryClient) startDiscovery(interval time.Duration, discover func(context.Context) error) { +func startDiscovery(log telegraf.Logger, interval time.Duration, discover func(context.Context) error) func() { ctx, cancel := context.WithCancel(context.Background()) var wg sync.WaitGroup - wg.Go(func() { + wg.Add(1) + go func() { + defer wg.Done() + if err := discover(ctx); err != nil && ctx.Err() == nil { - c.log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + log.Errorf("Discovering Elasticsearch nodes failed: %v", err) } if interval <= 0 { return @@ -208,31 +151,18 @@ func (c *queryClient) startDiscovery(interval time.Duration, discover func(conte return case <-ticker.C: if err := discover(ctx); err != nil && ctx.Err() == nil { - c.log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + log.Errorf("Discovering Elasticsearch nodes failed: %v", err) } } } - }) + }() - c.stopDiscovery = func() { + return func() { cancel() wg.Wait() } } -func (c *queryClient) isRunning() bool { - return !c.closed -} - -func (c *queryClient) getFieldMapping(ctx context.Context, index, field string) (map[string]interface{}, error) { - res, err := c.getFieldMappingResponse(ctx, index, field) - var result map[string]interface{} - if err := res.handle(err, &result); err != nil { - return nil, err - } - return result, nil -} - type queryData struct { measurement string name string @@ -250,7 +180,7 @@ func (q *queryData) addSubAggregation(name string, subAggregation map[string]int aggs[name] = subAggregation } -func (*queryClient) buildQueries(aggregation *aggregation) error { +func buildQueries(aggregation *aggregation) error { // Create one aggregation per metric field found or function defined for // numeric fields queries := make([]queryData, 0, len(aggregation.mapMetricFields)+len(aggregation.Tags)) @@ -343,13 +273,19 @@ func (a *aggregation) buildRangeQuery(from, to time.Time) map[string]interface{} return rangeQuery } +type searchHits struct { + Total json.RawMessage `json:"total"` +} + type searchResponse struct { - Hits struct { - Total json.RawMessage `json:"total"` - } `json:"hits"` + Hits searchHits `json:"hits"` Aggregations map[string]json.RawMessage `json:"aggregations"` } +type totalHits struct { + Value int64 `json:"value"` +} + func (r *searchResponse) totalHits() int64 { var total int64 if err := json.Unmarshal(r.Hits.Total, &total); err == nil { @@ -357,17 +293,15 @@ func (r *searchResponse) totalHits() int64 { } // Elasticsearch 7 and later return hits.total as an object. - var totalObject struct { - Value int64 `json:"value"` - } - if err := json.Unmarshal(r.Hits.Total, &totalObject); err == nil { - return totalObject.Value + var result totalHits + if err := json.Unmarshal(r.Hits.Total, &result); err == nil { + return result.Value } return 0 } -func (c *queryClient) query(ctx context.Context, aggregation *aggregation) (interface{}, int64, error) { +func buildSearchBody(aggregation *aggregation, log telegraf.Logger) ([]byte, error) { // buildQueries stores []queryData in this field before query execution. // If the assertion fails, it indicates a programming error in this package. queries := aggregation.queries.([]queryData) @@ -394,9 +328,9 @@ func (c *queryClient) query(ctx context.Context, aggregation *aggregation) (inte data, err := json.Marshal(query) if err != nil { - return nil, 0, fmt.Errorf("marshal request failed: %w", err) + return nil, fmt.Errorf("marshal request failed: %w", err) } - c.log.Debugf("{\"query\": %s}", string(data)) + log.Debugf("{\"query\": %s}", string(data)) body := map[string]interface{}{ "query": query, @@ -415,20 +349,17 @@ func (c *queryClient) query(ctx context.Context, aggregation *aggregation) (inte data, err = json.Marshal(body) if err != nil { - return nil, 0, fmt.Errorf("marshal request failed: %w", err) - } - - res, err := c.search(ctx, aggregation.Index, bytes.NewReader(data)) - var result searchResponse - if err := res.handle(err, &result); err != nil { - return nil, 0, err + return nil, fmt.Errorf("marshal request failed: %w", err) } + return data, nil +} - if len(result.Aggregations) == 0 { - return nil, result.totalHits(), nil - } +type aggregationValue struct { + Value *float64 `json:"value"` +} - return result.Aggregations, result.totalHits(), nil +type aggregationBuckets struct { + Buckets []map[string]json.RawMessage `json:"buckets"` } type aggregationIterator struct { @@ -465,9 +396,7 @@ func (m *aggregationIterator) iterate(acc telegraf.Accumulator, nameFunction map // Execute the aggregation function switch function { case "avg", "sum", "min", "max": - var result struct { - Value *float64 `json:"value"` - } + var result aggregationValue if err := json.Unmarshal(response[name], &result); err != nil { return err } @@ -477,9 +406,7 @@ func (m *aggregationIterator) iterate(acc telegraf.Accumulator, nameFunction map m.fields[name] = float64(0) } case "terms": - var result struct { - Buckets []map[string]json.RawMessage `json:"buckets"` - } + var result aggregationBuckets if err := json.Unmarshal(response[name], &result); err != nil { return err } @@ -521,7 +448,7 @@ func (m *aggregationIterator) iterate(acc telegraf.Accumulator, nameFunction map return nil } -func (*queryClient) aggregate(acc telegraf.Accumulator, measurement string, nameFunction map[string]string, response interface{}) error { +func aggregate(acc telegraf.Accumulator, measurement string, nameFunction map[string]string, response interface{}) error { // The query method returns map[string]json.RawMessage for aggregation responses. r := response.(map[string]json.RawMessage) diff --git a/plugins/inputs/elasticsearch_query/client_v5.go b/plugins/inputs/elasticsearch_query/client_v5.go index fc302b765f335..b757673be1be0 100644 --- a/plugins/inputs/elasticsearch_query/client_v5.go +++ b/plugins/inputs/elasticsearch_query/client_v5.go @@ -1,18 +1,28 @@ package elasticsearch_query import ( + "bytes" "context" + "encoding/json" "errors" "fmt" - "io" + "net/http" elasticsearch5 "github.com/elastic/go-elasticsearch/v5" + + "github.com/influxdata/telegraf" ) +type clientV5 struct { + client *elasticsearch5.Client + httpClient *http.Client + log telegraf.Logger +} + func newClientV5(cfg clientConfig) (client, error) { if cfg.enableSniffer { cfg.httpClient.CloseIdleConnections() - return nil, errors.New("enable_sniffer is not supported by the official Elasticsearch v5 client") + return nil, errors.New("enable_sniffer is not supported by the official ElasticSearch v5 client") } c, err := elasticsearch5.NewClient(elasticsearch5.Config{ @@ -23,33 +33,66 @@ func newClientV5(cfg clientConfig) (client, error) { }) if err != nil { cfg.httpClient.CloseIdleConnections() - return nil, fmt.Errorf("creating Elasticsearch client failed: %w", err) - } - - return &queryClient{ - httpClient: cfg.httpClient, - log: cfg.log, - getFieldMappingResponse: func(ctx context.Context, index, field string) (*esResponse, error) { - res, err := c.Indices.GetFieldMapping( - []string{field}, - c.Indices.GetFieldMapping.WithContext(ctx), - c.Indices.GetFieldMapping.WithIndex(index), - ) - if res == nil { - return nil, err - } - return &esResponse{statusCode: res.StatusCode, body: res.Body}, err - }, - search: func(ctx context.Context, index string, body io.Reader) (*esResponse, error) { - res, err := c.Search( - c.Search.WithContext(ctx), - c.Search.WithIndex(index), - c.Search.WithBody(body), - ) - if res == nil { - return nil, err - } - return &esResponse{statusCode: res.StatusCode, body: res.Body}, err - }, - }, nil + return nil, fmt.Errorf("creating ElasticSearch client failed: %w", err) + } + + return &clientV5{client: c, httpClient: cfg.httpClient, log: cfg.log}, nil +} + +func (c *clientV5) close() { + if c.httpClient != nil { + c.httpClient.CloseIdleConnections() + } +} + +func (c *clientV5) getFieldMapping(ctx context.Context, index, field string) (map[string]interface{}, error) { + res, err := c.client.Indices.GetFieldMapping( + []string{field}, + c.client.Indices.GetFieldMapping.WithContext(ctx), + c.client.Indices.GetFieldMapping.WithIndex(index), + ) + if err != nil { + return nil, err + } + defer res.Body.Close() + + if err := checkForError(res.StatusCode, res.Body); err != nil { + return nil, err + } + + var result map[string]interface{} + if err := json.NewDecoder(res.Body).Decode(&result); err != nil { + return nil, err + } + return result, nil +} + +func (c *clientV5) query(ctx context.Context, aggregation *aggregation) (interface{}, int64, error) { + data, err := buildSearchBody(aggregation, c.log) + if err != nil { + return nil, 0, err + } + + res, err := c.client.Search( + c.client.Search.WithContext(ctx), + c.client.Search.WithIndex(aggregation.Index), + c.client.Search.WithBody(bytes.NewReader(data)), + ) + if err != nil { + return nil, 0, err + } + defer res.Body.Close() + + if err := checkForError(res.StatusCode, res.Body); err != nil { + return nil, 0, err + } + + var result searchResponse + if err := json.NewDecoder(res.Body).Decode(&result); err != nil { + return nil, 0, err + } + if len(result.Aggregations) == 0 { + return nil, result.totalHits(), nil + } + return result.Aggregations, result.totalHits(), nil } diff --git a/plugins/inputs/elasticsearch_query/client_v6.go b/plugins/inputs/elasticsearch_query/client_v6.go index dbdc4018c3b06..6bbc4bf88f251 100644 --- a/plugins/inputs/elasticsearch_query/client_v6.go +++ b/plugins/inputs/elasticsearch_query/client_v6.go @@ -1,13 +1,24 @@ package elasticsearch_query import ( + "bytes" "context" + "encoding/json" "fmt" - "io" + "net/http" elasticsearch6 "github.com/elastic/go-elasticsearch/v6" + + "github.com/influxdata/telegraf" ) +type clientV6 struct { + client *elasticsearch6.Client + httpClient *http.Client + log telegraf.Logger + stopDiscovery func() +} + func newClientV6(cfg clientConfig) (client, error) { c, err := elasticsearch6.NewClient(elasticsearch6.Config{ Addresses: cfg.urls, @@ -17,38 +28,76 @@ func newClientV6(cfg clientConfig) (client, error) { }) if err != nil { cfg.httpClient.CloseIdleConnections() - return nil, fmt.Errorf("creating Elasticsearch client failed: %w", err) - } - - client := &queryClient{ - httpClient: cfg.httpClient, - log: cfg.log, - getFieldMappingResponse: func(ctx context.Context, index, field string) (*esResponse, error) { - res, err := c.Indices.GetFieldMapping( - []string{field}, - c.Indices.GetFieldMapping.WithContext(ctx), - c.Indices.GetFieldMapping.WithIndex(index), - ) - if res == nil { - return nil, err - } - return &esResponse{statusCode: res.StatusCode, body: res.Body}, err - }, - search: func(ctx context.Context, index string, body io.Reader) (*esResponse, error) { - res, err := c.Search( - c.Search.WithContext(ctx), - c.Search.WithIndex(index), - c.Search.WithBody(body), - ) - if res == nil { - return nil, err - } - return &esResponse{statusCode: res.StatusCode, body: res.Body}, err - }, + return nil, fmt.Errorf("creating ElasticSearch client failed: %w", err) } + + client := &clientV6{client: c, httpClient: cfg.httpClient, log: cfg.log} if cfg.enableSniffer { // The v6 client exposes only DiscoverNodes(), so in-flight calls cannot be canceled. - client.startDiscovery(cfg.discoveryInterval, func(context.Context) error { return c.DiscoverNodes() }) + client.stopDiscovery = startDiscovery(cfg.log, cfg.discoveryInterval, func(context.Context) error { + return c.DiscoverNodes() + }) } return client, nil } + +func (c *clientV6) close() { + if c.stopDiscovery != nil { + c.stopDiscovery() + } + if c.httpClient != nil { + c.httpClient.CloseIdleConnections() + } +} + +func (c *clientV6) getFieldMapping(ctx context.Context, index, field string) (map[string]interface{}, error) { + res, err := c.client.Indices.GetFieldMapping( + []string{field}, + c.client.Indices.GetFieldMapping.WithContext(ctx), + c.client.Indices.GetFieldMapping.WithIndex(index), + ) + if err != nil { + return nil, err + } + defer res.Body.Close() + + if err := checkForError(res.StatusCode, res.Body); err != nil { + return nil, err + } + + var result map[string]interface{} + if err := json.NewDecoder(res.Body).Decode(&result); err != nil { + return nil, err + } + return result, nil +} + +func (c *clientV6) query(ctx context.Context, aggregation *aggregation) (interface{}, int64, error) { + data, err := buildSearchBody(aggregation, c.log) + if err != nil { + return nil, 0, err + } + + res, err := c.client.Search( + c.client.Search.WithContext(ctx), + c.client.Search.WithIndex(aggregation.Index), + c.client.Search.WithBody(bytes.NewReader(data)), + ) + if err != nil { + return nil, 0, err + } + defer res.Body.Close() + + if err := checkForError(res.StatusCode, res.Body); err != nil { + return nil, 0, err + } + + var result searchResponse + if err := json.NewDecoder(res.Body).Decode(&result); err != nil { + return nil, 0, err + } + if len(result.Aggregations) == 0 { + return nil, result.totalHits(), nil + } + return result.Aggregations, result.totalHits(), nil +} diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index 638775fd35d11..db9ad25f0f35e 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -17,12 +17,9 @@ import ( ) type client interface { - isRunning() bool close() - buildQueries(aggregation *aggregation) error getFieldMapping(context.Context, string, string) (map[string]interface{}, error) query(context.Context, *aggregation) (interface{}, int64, error) - aggregate(telegraf.Accumulator, string, map[string]string, interface{}) error } //go:embed sample.conf @@ -85,6 +82,40 @@ func (e *ElasticsearchQuery) Init() error { return nil } +func (e *ElasticsearchQuery) newClient() (client, error) { + // Make sure the HTTP client exists + httpClient, err := e.HTTPClientConfig.CreateClient(context.Background(), e.Log) + if err != nil { + return nil, fmt.Errorf("creating HTTP client failed: %w", err) + } + + cfg := clientConfig{ + urls: e.URLs, + username: e.Username, + password: e.Password, + enableSniffer: e.EnableSniffer, + discoveryInterval: time.Duration(e.HealthCheckInterval), + httpClient: httpClient, + log: e.Log, + } + + version, major, err := cfg.probeVersion(context.Background()) + if err != nil { + httpClient.CloseIdleConnections() + return nil, fmt.Errorf("getting server version failed: %w", err) + } + + switch major { + case 5: + return newClientV5(cfg) + case 6: + return newClientV6(cfg) + default: + httpClient.CloseIdleConnections() + return nil, fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) + } +} + func (e *ElasticsearchQuery) Start(telegraf.Accumulator) error { // Create a new ElasticSearch client client, err := e.newClient() @@ -117,14 +148,6 @@ func (e *ElasticsearchQuery) Stop() { // Gather writes the results of the queries from Elasticsearch to the Accumulator. func (e *ElasticsearchQuery) Gather(acc telegraf.Accumulator) error { - // Make sure we are connected - if !e.client.isRunning() { - e.Stop() - if err := e.Start(acc); err != nil { - return err - } - } - var wg sync.WaitGroup for i := range e.Aggregations { wg.Add(1) @@ -162,7 +185,7 @@ func (e *ElasticsearchQuery) initAggregation(ctx context.Context, agg *aggregati } } - if err := e.client.buildQueries(agg); err != nil { + if err := buildQueries(agg); err != nil { return fmt.Errorf("building aggregation query failed: %w", err) } @@ -190,7 +213,7 @@ func (e *ElasticsearchQuery) gatherAggregation(acc telegraf.Accumulator, aggrega // Aggregate results that support aggregation for measurement, aggNameFunction := range aggregation.measurements { - if err := e.client.aggregate(acc, measurement, aggNameFunction, result); err != nil { + if err := aggregate(acc, measurement, aggNameFunction, result); err != nil { return fmt.Errorf("recursing response failed: %w", err) } } diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index ff499d20ec24b..5726dcb03e510 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -34,7 +34,7 @@ const ( testindex = "test-elasticsearch" ) -func TestResponseHandleError(t *testing.T) { +func TestCheckForError(t *testing.T) { tests := []struct { name string statusCode int @@ -45,34 +45,30 @@ func TestResponseHandleError(t *testing.T) { name: "type and reason", statusCode: http.StatusNotFound, body: `{"error":{"type":"index_not_found_exception","reason":"no such index"}}`, - expected: "elastic: Error 404 (Not Found): no such index [type=index_not_found_exception]", + expected: "received error 404 (Not Found): no such index [type=index_not_found_exception]", }, { name: "reason without type", statusCode: http.StatusBadGateway, body: `{"error":{"reason":"upstream failed"}}`, - expected: "elastic: Error 502 (Bad Gateway): upstream failed", + expected: "received error 502 (Bad Gateway): upstream failed", }, { name: "unstructured response", statusCode: http.StatusBadGateway, body: " proxy failed\n", - expected: "elastic: Error 502 (Bad Gateway): proxy failed", + expected: "received error 502 (Bad Gateway): proxy failed", }, { name: "empty response", statusCode: http.StatusBadGateway, - expected: "elastic: Error 502 (Bad Gateway)", + expected: "received error 502 (Bad Gateway)", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - response := &esResponse{ - statusCode: tt.statusCode, - body: io.NopCloser(strings.NewReader(tt.body)), - } - require.EqualError(t, response.handle(nil, nil), tt.expected) + require.EqualError(t, checkForError(tt.statusCode, strings.NewReader(tt.body)), tt.expected) }) } } @@ -230,47 +226,299 @@ func TestGatherIntegration(t *testing.T) { ), } - tests := []struct { - name string - image string - env map[string]string - }{ - { - name: "v5", - image: "elasticsearch:5.6.16", - env: map[string]string{"ES_JAVA_OPTS": "-Xms512m -Xmx512m"}, + // Setup the container + container := &testutil.Container{ + Image: "elasticsearch:6.8.23", + ExposedPorts: []string{servicePort}, + Env: map[string]string{ + "discovery.type": "single-node", }, - { - name: "v6", - image: "elasticsearch:6.8.23", - env: map[string]string{ - "discovery.type": "single-node", - "ES_JAVA_OPTS": "-Xms512m -Xmx512m", + WaitingFor: wait.ForAll( + wait.ForLog("] mode [basic] - valid"), + wait.ForListeningPort(servicePort), + ), + } + require.NoError(t, container.Start(), "failed to start container") + defer container.Terminate() + + addr := "http://" + container.Address + ":" + container.Ports[servicePort] + + // Fill the database + require.NoError(t, sendData(t.Context(), addr)) + + // Setup the plugin + plugin := &ElasticsearchQuery{ + URLs: []string{addr}, + Aggregations: []aggregation{ + { + Index: testindex, + MeasurementName: "measurement1", + MetricFields: []string{"size"}, + FilterQuery: "product_1", + MetricFunction: "avg", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + Tags: []string{"URI.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement2", + MetricFields: []string{"size"}, + FilterQuery: "downloads", + MetricFunction: "max", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + Tags: []string{"URI.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement3", + MetricFields: []string{"size"}, + FilterQuery: "downloads", + MetricFunction: "sum", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + Tags: []string{"response.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement4", + MetricFields: []string{"size", "response_time"}, + FilterQuery: "downloads", + MetricFunction: "min", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + IncludeMissingTag: true, + MissingTagValue: "missing", + Tags: []string{"response.keyword", "URI.keyword", "method.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement5", + FilterQuery: "product_2", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + Tags: []string{"URI.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement6", + FilterQuery: "response: 200", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + Tags: []string{"URI.keyword", "response.keyword"}, + }, + { + Index: testindex, + MeasurementName: "measurement7", + FilterQuery: "response: 200", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + }, + { + Index: testindex, + MeasurementName: "measurement8", + MetricFields: []string{"size"}, + FilterQuery: "downloads", + MetricFunction: "max", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + }, + { + Index: testindex, + MeasurementName: "measurement12", + MetricFields: []string{"size"}, + MetricFunction: "avg", + DateField: "@notatimestamp", + QueryPeriod: config.Duration(time.Second * 600), + }, + { + Index: testindex, + MeasurementName: "measurement13", + MetricFields: []string{"size"}, + MetricFunction: "avg", + DateField: "@timestamp", + QueryPeriod: config.Duration(time.Second * 600), + IncludeMissingTag: false, + Tags: []string{"nothere"}, }, }, + HTTPClientConfig: common_http.HTTPClientConfig{ + Timeout: config.Duration(30 * time.Second), + TransportConfig: common_http.TransportConfig{ + ResponseHeaderTimeout: config.Duration(30 * time.Second), + }, + }, + Log: testutil.Logger{}, } + require.NoError(t, plugin.Init()) - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - runGatherIntegration(t, tt.image, tt.env, expectedFields, expectedMetrics) - }) + var acc testutil.Accumulator + require.NoError(t, plugin.Start(&acc)) + defer plugin.Stop() + + // Check the ES field mapping + for i, agg := range plugin.Aggregations { + actual := agg.mapMetricFields + expected := expectedFields[i] + require.Equalf(t, expected, actual, "mismatch in aggregation %d", i) } + + // Collect metrics and check + require.NoError(t, acc.GatherError(plugin.Gather)) + require.Empty(t, acc.Errors) + + // Check the metrics + testutil.RequireMetricsEqual(t, expectedMetrics, acc.GetTelegrafMetrics(), testutil.SortMetrics(), testutil.IgnoreTime()) } -func runGatherIntegration( - t *testing.T, - image string, - env map[string]string, - expectedFields []map[string]string, - expectedMetrics []telegraf.Metric, -) { - t.Helper() +func TestGatherV5Integration(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + // Define expectations + expectedFields := []map[string]string{ + {"size": "long"}, + {"size": "long"}, + {"size": "long"}, + {"size": "long", "response_time": "long"}, + {}, + {}, + {}, + {"size": "long"}, + {"size": "long"}, + {"size": "long"}, + } + + expectedMetrics := []telegraf.Metric{ + metric.New( + "measurement1", + map[string]string{"URI_keyword": "/downloads/product_1"}, + map[string]interface{}{"size_avg": float64(202.30038022813687), "doc_count": int64(263)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement2", + map[string]string{"URI_keyword": "/downloads/product_1"}, + map[string]interface{}{"size_max": float64(3301), "doc_count": int64(263)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement2", + map[string]string{"URI_keyword": "/downloads/product_2"}, + map[string]interface{}{"size_max": float64(3318), "doc_count": int64(237)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement3", + map[string]string{"response_keyword": "200"}, + map[string]interface{}{"size_sum": float64(22790), "doc_count": int64(22)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement3", + map[string]string{"response_keyword": "304"}, + map[string]interface{}{"size_sum": float64(0), "doc_count": int64(219)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement3", + map[string]string{"response_keyword": "404"}, + map[string]interface{}{"size_sum": float64(86932), "doc_count": int64(259)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "404", "URI_keyword": "/downloads/product_1", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(318), "response_time_min": float64(126), "doc_count": int64(146)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "304", "URI_keyword": "/downloads/product_1", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(0), "response_time_min": float64(71), "doc_count": int64(113)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_1", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(490), "response_time_min": float64(1514), "doc_count": int64(3)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "404", "URI_keyword": "/downloads/product_2", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(318), "response_time_min": float64(237), "doc_count": int64(113)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "304", "URI_keyword": "/downloads/product_2", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(0), "response_time_min": float64(134), "doc_count": int64(106)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_2", "method_keyword": "GET"}, + map[string]interface{}{"size_min": float64(490), "response_time_min": float64(2), "doc_count": int64(13)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_1", "method_keyword": "HEAD"}, + map[string]interface{}{"size_min": float64(0), "response_time_min": float64(8479), "doc_count": int64(1)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement4", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_2", "method_keyword": "HEAD"}, + map[string]interface{}{"size_min": float64(0), "response_time_min": float64(1059), "doc_count": int64(5)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement5", + map[string]string{"URI_keyword": "/downloads/product_2"}, + map[string]interface{}{"doc_count": int64(237)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement6", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_1"}, + map[string]interface{}{"doc_count": int64(4)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement6", + map[string]string{"response_keyword": "200", "URI_keyword": "/downloads/product_2"}, + map[string]interface{}{"doc_count": int64(18)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement7", + map[string]string{}, + map[string]interface{}{"doc_count": int64(22)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement8", + map[string]string{}, + map[string]interface{}{"size_max": float64(3318)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + metric.New( + "measurement12", + map[string]string{}, + map[string]interface{}{"size_avg": float64(0)}, + time.Date(2018, 6, 14, 5, 51, 53, 266176036, time.UTC), + ), + } // Setup the container container := &testutil.Container{ - Image: image, + Image: "elasticsearch:5.6.16", ExposedPorts: []string{servicePort}, - Env: env, + Env: map[string]string{"ES_JAVA_OPTS": "-Xms512m -Xmx512m"}, WaitingFor: wait.ForHTTP("/").WithPort(servicePort).WithStartupTimeout(5 * time.Minute), } require.NoError(t, container.Start(), "failed to start container") @@ -524,7 +772,7 @@ func TestGatherFailGatherIntegration(t *testing.T) { DateField: "@timestamp", QueryPeriod: config.Duration(time.Second * 600), }, - expected: "Error 404 (Not Found): no such index", + expected: "received error 404 (Not Found): no such index", }, { name: "invalid time format", @@ -535,7 +783,7 @@ func TestGatherFailGatherIntegration(t *testing.T) { DateFieldFormat: "yyyy", QueryPeriod: config.Duration(time.Second * 600), }, - expected: "Error 400 (Bad Request): all shards failed", + expected: "received error 400 (Bad Request): all shards failed", }, } @@ -565,57 +813,27 @@ func TestGatherFailGatherIntegration(t *testing.T) { } } -func TestUnsupportedVersion(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - if _, err := w.Write([]byte(`{"version":{"number":"8.1.2"}}`)); err != nil { - t.Error(err) - } - })) - defer server.Close() - - plugin := &ElasticsearchQuery{ - URLs: []string{server.URL}, - Log: testutil.Logger{}, - } - require.NoError(t, plugin.Init()) - - var acc testutil.Accumulator - require.ErrorContains(t, plugin.Start(&acc), `server version "8.1.2" not supported`) -} - func TestStartupFailureReleasesClient(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") - if r.URL.Path != "/" { - http.Error(w, `{"error":{"type":"test_error","reason":"mapping failed"}}`, http.StatusInternalServerError) - return - } - if _, err := w.Write([]byte(`{"version":{"number":"6.8.23"}}`)); err != nil { + if _, err := w.Write([]byte(`{"version": {"number": "8.1.2"}}`)); err != nil { t.Error(err) } })) defer server.Close() plugin := &ElasticsearchQuery{ - URLs: []string{server.URL}, - Aggregations: []aggregation{{ - Index: "test", - MeasurementName: "test", - DateField: "@timestamp", - MetricFields: []string{"value"}, - MetricFunction: "avg", - }}, - Log: testutil.Logger{}, + URLs: []string{server.URL}, + HealthCheckInterval: config.Duration(10 * time.Second), + Log: testutil.Logger{}, } require.NoError(t, plugin.Init()) var acc testutil.Accumulator - require.ErrorContains(t, plugin.Start(&acc), "initializing aggregation") + require.ErrorContains(t, plugin.Start(&acc), "not supported") - // The failed start must release the client resources. - require.False(t, plugin.client.isRunning()) - require.Nil(t, plugin.client.(*queryClient).httpClient) + // The failed start must not retain a client. + require.Nil(t, plugin.client) } func sendData(ctx context.Context, url string) error { From ac9d16e3feb01248391e8e9b24ec8b90d76b9918 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Fri, 17 Jul 2026 18:26:58 +0200 Subject: [PATCH 07/18] fix: discovery --- plugins/inputs/elasticsearch_query/client.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 4959518e45d9d..5361957550ca9 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -128,6 +128,8 @@ func (t roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { return t.client.Do(req) } +// startDiscovery runs node discovery immediately and repeats it when the interval is positive. +// The returned function stops the loop and waits for any active call to return. func startDiscovery(log telegraf.Logger, interval time.Duration, discover func(context.Context) error) func() { ctx, cancel := context.WithCancel(context.Background()) @@ -137,7 +139,7 @@ func startDiscovery(log telegraf.Logger, interval time.Duration, discover func(c defer wg.Done() if err := discover(ctx); err != nil && ctx.Err() == nil { - log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + log.Errorf("Discovering ElasticSearch nodes failed: %v", err) } if interval <= 0 { return @@ -151,7 +153,7 @@ func startDiscovery(log telegraf.Logger, interval time.Duration, discover func(c return case <-ticker.C: if err := discover(ctx); err != nil && ctx.Err() == nil { - log.Errorf("Discovering Elasticsearch nodes failed: %v", err) + log.Errorf("Discovering ElasticSearch nodes failed: %v", err) } } } From da3774246215e4a9e4249ac32dc2b13ca26fa967 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Fri, 17 Jul 2026 18:27:16 +0200 Subject: [PATCH 08/18] test: add dicsovery test --- .../elasticsearch_query_test.go | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index 5726dcb03e510..65cb9df432ec9 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -73,6 +73,73 @@ func TestCheckForError(t *testing.T) { } } +func TestClientV5Sniffer(t *testing.T) { + c, err := newClientV5(clientConfig{ + enableSniffer: true, + httpClient: &http.Client{}, + }) + + require.EqualError(t, err, "enable_sniffer is not supported by the official ElasticSearch v5 client") + require.Nil(t, c) +} + +func TestClientV6Sniffer(t *testing.T) { + discovered := make(chan struct{}, 1) + + var server *httptest.Server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/_nodes/http" { + http.NotFound(w, r) + return + } + + response := map[string]interface{}{ + "nodes": map[string]interface{}{ + "node": map[string]interface{}{ + "name": "node", + "roles": []string{"data", "ingest"}, + "http": map[string]string{ + "publish_address": strings.TrimPrefix(server.URL, "http://"), + }, + }, + }, + } + if err := json.NewEncoder(w).Encode(response); err != nil { + t.Error(err) + return + } + discovered <- struct{}{} + })) + defer server.Close() + + c, err := newClientV6(clientConfig{ + urls: []string{server.URL}, + enableSniffer: true, + discoveryInterval: time.Hour, + httpClient: server.Client(), + log: testutil.Logger{}, + }) + require.NoError(t, err) + + select { + case <-discovered: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for node discovery") + } + + stopped := make(chan struct{}) + go func() { + c.close() + close(stopped) + }() + + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("timed out stopping node discovery") + } +} + type nginxlog struct { IPaddress string `json:"IP"` Timestamp time.Time `json:"@timestamp"` From b318ea4b6f83a702601d12046f35affe56f4174f Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Mon, 20 Jul 2026 10:25:13 +0200 Subject: [PATCH 09/18] fix: preserve server version error messages --- plugins/inputs/elasticsearch_query/client.go | 6 ++--- .../elasticsearch_query.go | 2 +- .../elasticsearch_query_test.go | 22 +++++++++++++++++++ 3 files changed, 26 insertions(+), 4 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 5361957550ca9..2f9edbe2948f9 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -48,17 +48,17 @@ func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { res, err := probe.Info(probe.Info.WithContext(ctx)) if err != nil { - return "", 0, err + return "", 0, fmt.Errorf("getting server version failed: %w", err) } defer res.Body.Close() if err := checkForError(res.StatusCode, res.Body); err != nil { - return "", 0, err + return "", 0, fmt.Errorf("getting server version failed: %w", err) } var info serverInfo if err := json.NewDecoder(res.Body).Decode(&info); err != nil { - return "", 0, err + return "", 0, fmt.Errorf("getting server version failed: %w", err) } version, err := semver.NewVersion(info.Version.Number) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index db9ad25f0f35e..8ac810fcee292 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -102,7 +102,7 @@ func (e *ElasticsearchQuery) newClient() (client, error) { version, major, err := cfg.probeVersion(context.Background()) if err != nil { httpClient.CloseIdleConnections() - return nil, fmt.Errorf("getting server version failed: %w", err) + return nil, err } switch major { diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index 65cb9df432ec9..3ed68bbb8b9c9 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -880,6 +880,28 @@ func TestGatherFailGatherIntegration(t *testing.T) { } } +func TestInvalidServerVersion(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if _, err := w.Write([]byte(`{"version":{"number":"invalid"}}`)); err != nil { + t.Error(err) + } + })) + defer server.Close() + + plugin := &ElasticsearchQuery{ + URLs: []string{server.URL}, + Log: testutil.Logger{}, + } + require.NoError(t, plugin.Init()) + + var acc testutil.Accumulator + err := plugin.Start(&acc) + require.ErrorContains(t, err, `parsing server version "invalid" failed`) + require.NotContains(t, err.Error(), "getting server version failed") + require.Nil(t, plugin.client) +} + func TestStartupFailureReleasesClient(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") From f74f7aef582417fe83d9e3c0037e6f96a538a2d1 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 09:32:36 +0200 Subject: [PATCH 10/18] fix: remove unnecessary cast --- plugins/inputs/elasticsearch_query/client.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 2f9edbe2948f9..7abeabe338fe4 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -34,7 +34,7 @@ type serverInfo struct { Version serverVersion `json:"version"` } -func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { +func (cfg clientConfig) probeVersion(ctx context.Context) (string, uint64, error) { // Use the v5 client only for the version-agnostic GET / probe. probe, err := elasticsearch5.NewClient(elasticsearch5.Config{ Addresses: cfg.urls, @@ -66,7 +66,7 @@ func (cfg clientConfig) probeVersion(ctx context.Context) (string, int, error) { return "", 0, fmt.Errorf("parsing server version %q failed: %w", info.Version.Number, err) } - return info.Version.Number, int(version.Major()), nil + return info.Version.Number, version.Major(), nil } type apiErrorDetails struct { From 2b9a46e8078a7b37be30ea307a21d82cd064f3a8 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 09:38:37 +0200 Subject: [PATCH 11/18] fix: accept only 2xx --- plugins/inputs/elasticsearch_query/client.go | 2 +- .../inputs/elasticsearch_query/elasticsearch_query_test.go | 5 +++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 7abeabe338fe4..8d2c4e8f33ce0 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -96,7 +96,7 @@ func (e *apiError) Error() string { } func checkForError(statusCode int, body io.Reader) error { - if statusCode <= 299 { + if statusCode >= 200 && statusCode < 300 { return nil } diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index 3ed68bbb8b9c9..eeff7a2b210cb 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -41,6 +41,11 @@ func TestCheckForError(t *testing.T) { body string expected string }{ + { + name: "informational response", + statusCode: http.StatusContinue, + expected: "received error 100 (Continue)", + }, { name: "type and reason", statusCode: http.StatusNotFound, From 6b817e4d42057f4a9762136b9e8c23ff93962977 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:11:58 +0200 Subject: [PATCH 12/18] refactor: discovery lifecycle --- plugins/inputs/elasticsearch_query/client.go | 44 ++++++------------- .../inputs/elasticsearch_query/client_v6.go | 29 +++++++----- 2 files changed, 32 insertions(+), 41 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 8d2c4e8f33ce0..82e531686b272 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -7,7 +7,6 @@ import ( "io" "net/http" "strings" - "sync" "time" "github.com/Masterminds/semver/v3" @@ -128,40 +127,23 @@ func (t roundTripper) RoundTrip(req *http.Request) (*http.Response, error) { return t.client.Do(req) } -// startDiscovery runs node discovery immediately and repeats it when the interval is positive. -// The returned function stops the loop and waits for any active call to return. -func startDiscovery(log telegraf.Logger, interval time.Duration, discover func(context.Context) error) func() { - ctx, cancel := context.WithCancel(context.Background()) - - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() +// startDiscovery runs node discovery immediately and then at the configured interval. +func startDiscovery(ctx context.Context, interval time.Duration, discover func(context.Context) error, log telegraf.Logger) { + if err := discover(ctx); err != nil && ctx.Err() == nil { + log.Errorf("Discovering ElasticSearch nodes failed: %v", err) + } - if err := discover(ctx); err != nil && ctx.Err() == nil { - log.Errorf("Discovering ElasticSearch nodes failed: %v", err) - } - if interval <= 0 { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): return - } - - ticker := time.NewTicker(interval) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return - case <-ticker.C: - if err := discover(ctx); err != nil && ctx.Err() == nil { - log.Errorf("Discovering ElasticSearch nodes failed: %v", err) - } + case <-ticker.C: + if err := discover(ctx); err != nil && ctx.Err() == nil { + log.Errorf("Discovering ElasticSearch nodes failed: %v", err) } } - }() - - return func() { - cancel() - wg.Wait() } } diff --git a/plugins/inputs/elasticsearch_query/client_v6.go b/plugins/inputs/elasticsearch_query/client_v6.go index 6bbc4bf88f251..a891f88b76d77 100644 --- a/plugins/inputs/elasticsearch_query/client_v6.go +++ b/plugins/inputs/elasticsearch_query/client_v6.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "net/http" + "sync" elasticsearch6 "github.com/elastic/go-elasticsearch/v6" @@ -13,10 +14,11 @@ import ( ) type clientV6 struct { - client *elasticsearch6.Client - httpClient *http.Client - log telegraf.Logger - stopDiscovery func() + client *elasticsearch6.Client + httpClient *http.Client + log telegraf.Logger + cancelDiscovery context.CancelFunc + discoveryWG sync.WaitGroup } func newClientV6(cfg clientConfig) (client, error) { @@ -32,18 +34,25 @@ func newClientV6(cfg clientConfig) (client, error) { } client := &clientV6{client: c, httpClient: cfg.httpClient, log: cfg.log} - if cfg.enableSniffer { + if cfg.enableSniffer && cfg.discoveryInterval > 0 { // The v6 client exposes only DiscoverNodes(), so in-flight calls cannot be canceled. - client.stopDiscovery = startDiscovery(cfg.log, cfg.discoveryInterval, func(context.Context) error { - return c.DiscoverNodes() - }) + ctx, cancel := context.WithCancel(context.Background()) + client.cancelDiscovery = cancel + client.discoveryWG.Add(1) + go func() { + defer client.discoveryWG.Done() + startDiscovery(ctx, cfg.discoveryInterval, func(context.Context) error { + return c.DiscoverNodes() + }, cfg.log) + }() } return client, nil } func (c *clientV6) close() { - if c.stopDiscovery != nil { - c.stopDiscovery() + if c.cancelDiscovery != nil { + c.cancelDiscovery() + c.discoveryWG.Wait() } if c.httpClient != nil { c.httpClient.CloseIdleConnections() From 75595085639da4a93faf5a8c22cc6fe9fdfa38e9 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:18:34 +0200 Subject: [PATCH 13/18] refactor: extract API types --- plugins/inputs/elasticsearch_query/client.go | 62 ++++---------------- plugins/inputs/elasticsearch_query/types.go | 41 +++++++++++++ 2 files changed, 53 insertions(+), 50 deletions(-) create mode 100644 plugins/inputs/elasticsearch_query/types.go diff --git a/plugins/inputs/elasticsearch_query/client.go b/plugins/inputs/elasticsearch_query/client.go index 82e531686b272..b8da96e45c779 100644 --- a/plugins/inputs/elasticsearch_query/client.go +++ b/plugins/inputs/elasticsearch_query/client.go @@ -25,14 +25,6 @@ type clientConfig struct { log telegraf.Logger } -type serverVersion struct { - Number string `json:"number"` -} - -type serverInfo struct { - Version serverVersion `json:"version"` -} - func (cfg clientConfig) probeVersion(ctx context.Context) (string, uint64, error) { // Use the v5 client only for the version-agnostic GET / probe. probe, err := elasticsearch5.NewClient(elasticsearch5.Config{ @@ -68,15 +60,6 @@ func (cfg clientConfig) probeVersion(ctx context.Context) (string, uint64, error return info.Version.Number, version.Major(), nil } -type apiErrorDetails struct { - Type string `json:"type"` - Reason string `json:"reason"` -} - -type apiErrorResponse struct { - Error apiErrorDetails `json:"error"` -} - type apiError struct { statusCode int errorType string @@ -257,34 +240,6 @@ func (a *aggregation) buildRangeQuery(from, to time.Time) map[string]interface{} return rangeQuery } -type searchHits struct { - Total json.RawMessage `json:"total"` -} - -type searchResponse struct { - Hits searchHits `json:"hits"` - Aggregations map[string]json.RawMessage `json:"aggregations"` -} - -type totalHits struct { - Value int64 `json:"value"` -} - -func (r *searchResponse) totalHits() int64 { - var total int64 - if err := json.Unmarshal(r.Hits.Total, &total); err == nil { - return total - } - - // Elasticsearch 7 and later return hits.total as an object. - var result totalHits - if err := json.Unmarshal(r.Hits.Total, &result); err == nil { - return result.Value - } - - return 0 -} - func buildSearchBody(aggregation *aggregation, log telegraf.Logger) ([]byte, error) { // buildQueries stores []queryData in this field before query execution. // If the assertion fails, it indicates a programming error in this package. @@ -338,12 +293,19 @@ func buildSearchBody(aggregation *aggregation, log telegraf.Logger) ([]byte, err return data, nil } -type aggregationValue struct { - Value *float64 `json:"value"` -} +func (r *searchResponse) totalHits() int64 { + var total int64 + if err := json.Unmarshal(r.Hits.Total, &total); err == nil { + return total + } + + // Elasticsearch 7 and later return hits.total as an object. + var result totalHits + if err := json.Unmarshal(r.Hits.Total, &result); err == nil { + return result.Value + } -type aggregationBuckets struct { - Buckets []map[string]json.RawMessage `json:"buckets"` + return 0 } type aggregationIterator struct { diff --git a/plugins/inputs/elasticsearch_query/types.go b/plugins/inputs/elasticsearch_query/types.go new file mode 100644 index 0000000000000..95b31d937d3e2 --- /dev/null +++ b/plugins/inputs/elasticsearch_query/types.go @@ -0,0 +1,41 @@ +package elasticsearch_query + +import "encoding/json" + +type serverVersion struct { + Number string `json:"number"` +} + +type serverInfo struct { + Version serverVersion `json:"version"` +} + +type apiErrorDetails struct { + Type string `json:"type"` + Reason string `json:"reason"` +} + +type apiErrorResponse struct { + Error apiErrorDetails `json:"error"` +} + +type searchHits struct { + Total json.RawMessage `json:"total"` +} + +type searchResponse struct { + Hits searchHits `json:"hits"` + Aggregations map[string]json.RawMessage `json:"aggregations"` +} + +type totalHits struct { + Value int64 `json:"value"` +} + +type aggregationValue struct { + Value *float64 `json:"value"` +} + +type aggregationBuckets struct { + Buckets []map[string]json.RawMessage `json:"buckets"` +} From 89f4a0d55ee0b9da02a7690dc18e9973ddf0c56b Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:22:32 +0200 Subject: [PATCH 14/18] fix: ignore unsupported ES5 discovery --- plugins/inputs/elasticsearch_query/README.md | 1 + .../inputs/elasticsearch_query/client_v5.go | 6 ----- .../elasticsearch_query.go | 3 +++ .../elasticsearch_query_test.go | 27 ++++++++++++++----- 4 files changed, 25 insertions(+), 12 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/README.md b/plugins/inputs/elasticsearch_query/README.md index b1133535bbb66..51b51a3a5f5e5 100644 --- a/plugins/inputs/elasticsearch_query/README.md +++ b/plugins/inputs/elasticsearch_query/README.md @@ -9,6 +9,7 @@ particular field. > [!IMPORTANT] > This plugins supports Elasticsearch 5.x and 6.x but is known to break on 7.x > or higher. +> Node discovery is not supported with Elasticsearch 5.x. ⭐ Telegraf v1.20.0 🏷️ datastore diff --git a/plugins/inputs/elasticsearch_query/client_v5.go b/plugins/inputs/elasticsearch_query/client_v5.go index b757673be1be0..4aebac8604376 100644 --- a/plugins/inputs/elasticsearch_query/client_v5.go +++ b/plugins/inputs/elasticsearch_query/client_v5.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "encoding/json" - "errors" "fmt" "net/http" @@ -20,11 +19,6 @@ type clientV5 struct { } func newClientV5(cfg clientConfig) (client, error) { - if cfg.enableSniffer { - cfg.httpClient.CloseIdleConnections() - return nil, errors.New("enable_sniffer is not supported by the official ElasticSearch v5 client") - } - c, err := elasticsearch5.NewClient(elasticsearch5.Config{ Addresses: cfg.urls, Username: cfg.username, diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index 8ac810fcee292..8b109fc88c57c 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -107,6 +107,9 @@ func (e *ElasticsearchQuery) newClient() (client, error) { switch major { case 5: + if cfg.enableSniffer { + e.Log.Warn("'enable_sniffer' is not supported for ElasticSearch 5.x and will be ignored") + } return newClientV5(cfg) case 6: return newClientV6(cfg) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index eeff7a2b210cb..527e50d3f3d6a 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -79,13 +79,28 @@ func TestCheckForError(t *testing.T) { } func TestClientV5Sniffer(t *testing.T) { - c, err := newClientV5(clientConfig{ - enableSniffer: true, - httpClient: &http.Client{}, - }) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + if _, err := w.Write([]byte(`{"version":{"number":"5.6.16"}}`)); err != nil { + t.Error(err) + } + })) + defer server.Close() + + logger := &testutil.CaptureLogger{} + plugin := &ElasticsearchQuery{ + URLs: []string{server.URL}, + EnableSniffer: true, + Log: logger, + } + + c, err := plugin.newClient() + require.NoError(t, err) + defer c.close() - require.EqualError(t, err, "enable_sniffer is not supported by the official ElasticSearch v5 client") - require.Nil(t, c) + warnings := logger.Warnings() + require.Len(t, warnings, 1) + require.Contains(t, warnings[0], "'enable_sniffer' is not supported") } func TestClientV6Sniffer(t *testing.T) { From 5e336efc8dccb1e5dc162e0222f24a475e334d30 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:25:28 +0200 Subject: [PATCH 15/18] fix: add context to response decoding errors --- plugins/inputs/elasticsearch_query/client_v5.go | 4 ++-- plugins/inputs/elasticsearch_query/client_v6.go | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/client_v5.go b/plugins/inputs/elasticsearch_query/client_v5.go index 4aebac8604376..571a1431d09b1 100644 --- a/plugins/inputs/elasticsearch_query/client_v5.go +++ b/plugins/inputs/elasticsearch_query/client_v5.go @@ -56,7 +56,7 @@ func (c *clientV5) getFieldMapping(ctx context.Context, index, field string) (ma var result map[string]interface{} if err := json.NewDecoder(res.Body).Decode(&result); err != nil { - return nil, err + return nil, fmt.Errorf("decoding message body failed: %w", err) } return result, nil } @@ -83,7 +83,7 @@ func (c *clientV5) query(ctx context.Context, aggregation *aggregation) (interfa var result searchResponse if err := json.NewDecoder(res.Body).Decode(&result); err != nil { - return nil, 0, err + return nil, 0, fmt.Errorf("decoding message body failed: %w", err) } if len(result.Aggregations) == 0 { return nil, result.totalHits(), nil diff --git a/plugins/inputs/elasticsearch_query/client_v6.go b/plugins/inputs/elasticsearch_query/client_v6.go index a891f88b76d77..c4ac1ea4cfa7d 100644 --- a/plugins/inputs/elasticsearch_query/client_v6.go +++ b/plugins/inputs/elasticsearch_query/client_v6.go @@ -76,7 +76,7 @@ func (c *clientV6) getFieldMapping(ctx context.Context, index, field string) (ma var result map[string]interface{} if err := json.NewDecoder(res.Body).Decode(&result); err != nil { - return nil, err + return nil, fmt.Errorf("decoding message body failed: %w", err) } return result, nil } @@ -103,7 +103,7 @@ func (c *clientV6) query(ctx context.Context, aggregation *aggregation) (interfa var result searchResponse if err := json.NewDecoder(res.Body).Decode(&result); err != nil { - return nil, 0, err + return nil, 0, fmt.Errorf("decoding message body failed: %w", err) } if len(result.Aggregations) == 0 { return nil, result.totalHits(), nil From dc8ae3d3f93760ac45f9bca8cb0ece4d2140e30e Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:28:46 +0200 Subject: [PATCH 16/18] refactor: merge client setup into Start --- .../elasticsearch_query.go | 20 ++++++++----------- .../elasticsearch_query_test.go | 8 +++++--- 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query.go b/plugins/inputs/elasticsearch_query/elasticsearch_query.go index 8b109fc88c57c..5a177fe7076ea 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query.go @@ -82,11 +82,11 @@ func (e *ElasticsearchQuery) Init() error { return nil } -func (e *ElasticsearchQuery) newClient() (client, error) { +func (e *ElasticsearchQuery) Start(telegraf.Accumulator) error { // Make sure the HTTP client exists httpClient, err := e.HTTPClientConfig.CreateClient(context.Background(), e.Log) if err != nil { - return nil, fmt.Errorf("creating HTTP client failed: %w", err) + return fmt.Errorf("creating HTTP client failed: %w", err) } cfg := clientConfig{ @@ -102,30 +102,26 @@ func (e *ElasticsearchQuery) newClient() (client, error) { version, major, err := cfg.probeVersion(context.Background()) if err != nil { httpClient.CloseIdleConnections() - return nil, err + return err } + var c client switch major { case 5: if cfg.enableSniffer { e.Log.Warn("'enable_sniffer' is not supported for ElasticSearch 5.x and will be ignored") } - return newClientV5(cfg) + c, err = newClientV5(cfg) case 6: - return newClientV6(cfg) + c, err = newClientV6(cfg) default: httpClient.CloseIdleConnections() - return nil, fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) + return fmt.Errorf("server version %q not supported (currently supported versions are 5.x and 6.x)", version) } -} - -func (e *ElasticsearchQuery) Start(telegraf.Accumulator) error { - // Create a new ElasticSearch client - client, err := e.newClient() if err != nil { return err } - e.client = client + e.client = c // Setup the aggregations, this needs to be done in Start as it will require // API calls to the ElasticSearch endpoint and can thus not happen in Init diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index 527e50d3f3d6a..ca71bbe6be533 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -94,9 +94,11 @@ func TestClientV5Sniffer(t *testing.T) { Log: logger, } - c, err := plugin.newClient() - require.NoError(t, err) - defer c.close() + require.NoError(t, plugin.Init()) + + var acc testutil.Accumulator + require.NoError(t, plugin.Start(&acc)) + defer plugin.Stop() warnings := logger.Warnings() require.Len(t, warnings, 1) From 5c5e7c973ffb347baf304d302abdfac76b69e050 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:31:49 +0200 Subject: [PATCH 17/18] test: ignore error message prefixes --- .../inputs/elasticsearch_query/elasticsearch_query_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index ca71bbe6be533..fa218f28c2987 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -861,7 +861,7 @@ func TestGatherFailGatherIntegration(t *testing.T) { DateField: "@timestamp", QueryPeriod: config.Duration(time.Second * 600), }, - expected: "received error 404 (Not Found): no such index", + expected: "404 (Not Found): no such index", }, { name: "invalid time format", @@ -872,7 +872,7 @@ func TestGatherFailGatherIntegration(t *testing.T) { DateFieldFormat: "yyyy", QueryPeriod: config.Duration(time.Second * 600), }, - expected: "received error 400 (Bad Request): all shards failed", + expected: "400 (Bad Request): all shards failed", }, } From 873ca771912a2ea625da80b79697972a27dce7c0 Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Wed, 22 Jul 2026 10:33:01 +0200 Subject: [PATCH 18/18] test: remove redundant error assertion --- plugins/inputs/elasticsearch_query/elasticsearch_query_test.go | 1 - 1 file changed, 1 deletion(-) diff --git a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go index fa218f28c2987..c8f66e9249469 100644 --- a/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go +++ b/plugins/inputs/elasticsearch_query/elasticsearch_query_test.go @@ -920,7 +920,6 @@ func TestInvalidServerVersion(t *testing.T) { var acc testutil.Accumulator err := plugin.Start(&acc) require.ErrorContains(t, err, `parsing server version "invalid" failed`) - require.NotContains(t, err.Error(), "getting server version failed") require.Nil(t, plugin.client) }