[Elasticsearch-xpack] [Metricbeat] Fix elasticsearch-xpack error when…

… Machine Learning is explicitly disabled (#39990)

* Fix metricbeat monitoring module elasticsearch-xpack error when Machine Learning is explicitly disabled

* CR changes

* formatting changes

* Fix test

---------

Co-authored-by: Kevin Lacabane <[email protected]>

metricbeat/module/elasticsearch/elasticsearch.go

@@ -351,6 +351,9 @@ type XPack struct {
  CCR struct {
  Enabled bool `json:"enabled"`
  } `json:"CCR"`
+ ML struct {
+ Enabled bool `json:"enabled"`
+ } `json:"ml"`
  } `json:"features"`
 }
 

metricbeat/module/elasticsearch/ml_job/data_test.go

@@ -82,11 +82,3 @@ func TestData(t *testing.T) {
  t.Fatal("error trying to write event:", err)
  }
 }
-
-func getConfig(host string) map[string]interface{} {
- return map[string]interface{}{
- "module": elasticsearch.ModuleName,
- "metricsets": []string{"ml_job"},
- "hosts": []string{host},
- }
-}

metricbeat/module/elasticsearch/ml_job/ml_job.go

@@ -18,6 +18,8 @@
 package ml_job
 
 import (
+ "fmt"
+
  "github.com/elastic/beats/v7/metricbeat/mb"
  "github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
 )
@@ -59,6 +61,13 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
  if shouldSkip {
  return nil
  }
+ mLAvailability, mlErr := m.checkMLAvailability()
+ if mlErr != nil {
+ return mlErr
+ }
+ if !mLAvailability {
+ return nil
+ }
 
  info, err := elasticsearch.GetInfo(m.HTTP, m.GetServiceURI())
  if err != nil {
@@ -72,3 +81,16 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
 
  return eventsMapping(r, info, content, m.XPackEnabled)
 }
+
+func (m *MetricSet) checkMLAvailability() (isAvailable bool, err error) {
+ xpack, err := elasticsearch.GetXPack(m.HTTP, m.GetServiceURI())
+ if err != nil {
+ return false, fmt.Errorf("error determining xpack features: %w", err)
+ }
+
+ if !xpack.Features.ML.Enabled {
+ return false, nil
+ }
+
+ return true, nil
+}

metricbeat/module/elasticsearch/ml_job/ml_job_test.go

@@ -0,0 +1,115 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package ml_job
+
+import (
+ "io/ioutil"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "strings"
+ "testing"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
+
+ mbtest "github.com/elastic/beats/v7/metricbeat/mb/testing"
+)
+
+func createEsMuxer(mlEnabled bool) *http.ServeMux {
+ license := "platinum"
+ nodesLocalHandler := func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(`{"nodes": { "foobar": {}}}`))
+ }
+ clusterStateMasterHandler := func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(`{"master_node": "foobar"}`))
+ }
+ rootHandler := func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/" {
+ http.NotFound(w, r)
+ }
+
+ input, _ := ioutil.ReadFile("./_meta/test/root.710.json")
+ input = []byte(strings.Replace(string(input), "7.10.0", "7.10.0", -1))
+ w.Write(input)
+ }
+ licenseHandler := func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(`{ "license": { "type": "` + license + `" } }`))
+ }
+ xpackHandler := func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(`{ "features": { "ml": { "enabled": ` + strconv.FormatBool(mlEnabled) + `}}, "ccr": { "enabled": ` + strconv.FormatBool(true) + `}}`))
+ }
+
+ mux := http.NewServeMux()
+ mux.Handle("/_nodes/_local/nodes", http.HandlerFunc(nodesLocalHandler))
+ mux.Handle("/_cluster/state/master_node", http.HandlerFunc(clusterStateMasterHandler))
+ mux.Handle("/", http.HandlerFunc(rootHandler))
+ mux.Handle("/_license", http.HandlerFunc(licenseHandler)) // for 7.0 and above
+ mux.Handle("/_xpack/license", http.HandlerFunc(licenseHandler)) // for before 7.0
+ mux.Handle("/_xpack", http.HandlerFunc(xpackHandler))
+ if mlEnabled {
+ // Should call the API
+ mux.Handle("/_ml/anomaly_detectors/_all/_stats", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(`{}`))
+ }))
+ }
+
+ return mux
+}
+
+func TestMLNotAvailable(t *testing.T) {
+ tests := map[string]struct {
+ mlEnabled bool
+ }{
+ "feature_available": {
+ true,
+ },
+ "feature_unavailable": {
+ false,
+ },
+ }
+
+ for name, test := range tests {
+ t.Run(name, func(t *testing.T) {
+ mux := createEsMuxer(test.mlEnabled)
+ if !test.mlEnabled {
+ mux.Handle("/_ml/anomaly_detectors/_all/_stats", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ http.Error(w, "this should never have been called", http.StatusTeapot)
+ }))
+ }
+
+ server := httptest.NewServer(mux)
+ defer server.Close()
+
+ ms := mbtest.NewReportingMetricSetV2Error(t, getConfig(server.URL))
+ events, errs := mbtest.ReportingFetchV2Error(ms)
+
+ require.Empty(t, errs)
+ require.Empty(t, events)
+ })
+ }
+}
+
+func getConfig(host string) map[string]interface{} {
+ return map[string]interface{}{
+ "module": elasticsearch.ModuleName,
+ "metricsets": []string{"ml_job"},
+ "hosts": []string{host},
+ }
+}