Check license for CCR before attempting to fetch stats

metricbeat/module/elasticsearch/ccr/ccr.go

@@ -19,6 +19,7 @@ package ccr
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/pkg/errors"
 
@@ -41,6 +42,7 @@ const (
 // MetricSet type defines all fields of the MetricSet
 type MetricSet struct {
 	*elasticsearch.MetricSet
+	lastCCRLicenseMessageTimestamp time.Time
 }
 
 // New create a new instance of the MetricSet
@@ -56,7 +58,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 
 // Fetch gathers stats for each follower shard from the _ccr/stats API
 func (m *MetricSet) Fetch(r mb.ReporterV2) {
-	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.HostData().SanitizedURI+ccrStatsPath)
+	isMaster, err := elasticsearch.IsMaster(m.HTTP, m.getServiceURI())
 	if err != nil {
 		err = errors.Wrap(err, "error determining if connected Elasticsearch node is master")
 		elastic.ReportAndLogError(err, r, m.Log)
@@ -69,24 +71,29 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 		return
 	}
 
-	info, err := elasticsearch.GetInfo(m.HTTP, m.HostData().SanitizedURI+ccrStatsPath)
+	info, err := elasticsearch.GetInfo(m.HTTP, m.getServiceURI())
 	if err != nil {
 		elastic.ReportAndLogError(err, r, m.Log)
 		return
 	}
 
 	elasticsearchVersion := info.Version.Number
-	isCCRStatsAPIAvailable, err := elasticsearch.IsCCRStatsAPIAvailable(elasticsearchVersion)
+
+	// CCR is only available in Trial or Platinum license of Elasticsearch. So we check
+	// the license first.
+	ccrUnavailableMessage, err := m.checkCCRAvailability(elasticsearchVersion)
 	if err != nil {
+		err = errors.Wrap(err, "error determining if CCR is available")
 		elastic.ReportAndLogError(err, r, m.Log)
 		return
 	}
 
-	if !isCCRStatsAPIAvailable {
-		const errorMsg = "the %v metricset is only supported with Elasticsearch >= %v. " +
-			"You are currently running Elasticsearch %v"
-		err = fmt.Errorf(errorMsg, m.FullyQualifiedName(), elasticsearch.CCRStatsAPIAvailableVersion, elasticsearchVersion)
-		elastic.ReportAndLogError(err, r, m.Log)
+	if ccrUnavailableMessage != "" {
+		if time.Since(m.lastCCRLicenseMessageTimestamp) > 1*time.Minute {
+			err := fmt.Errorf(ccrUnavailableMessage)
+			elastic.ReportAndLogError(err, r, m.Log)
+			m.lastCCRLicenseMessageTimestamp = time.Now()
+		}
 		return
 	}
 
@@ -107,3 +114,37 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) {
 		return
 	}
 }
+
+func (m *MetricSet) checkCCRAvailability(currentElasticsearchVersion string) (message string, err error) {
+	license, err := elasticsearch.GetLicense(m.HTTP, m.getServiceURI())
+	if err != nil {
+		return "", errors.Wrap(err, "error determining Elasticsearch license")
+	}
+
+	if !license.IsOneOf("trial", "platinum") {
+		message = "the CCR feature is available with a platinum Elasticsearch license. " +
+			"You currently have a " + license.Type + " license. " +
+			"Either upgrade your license or remove the ccr metricset from your Elasticsearch module configuration."
+		return
+	}
+
+	isAvailable, err := elastic.IsFeatureAvailable(currentElasticsearchVersion, elasticsearch.CCRStatsAPIAvailableVersion)
+	if err != nil {
+		return "", errors.Wrap(err, "error determining if CCR is available in current Elasticsearch version")
+	}
+
+	if !isAvailable {
+		metricsetName := m.FullyQualifiedName()
+		message = "the " + metricsetName + " is only supported with Elasticsearch >= " +
+			elasticsearch.CCRStatsAPIAvailableVersion + ". " +
+			"You are currently running Elasticsearch " + currentElasticsearchVersion + "."
+		return
+	}
+
+	return "", nil
+}
+
+func (m *MetricSet) getServiceURI() string {
+	return m.HostData().SanitizedURI + ccrStatsPath
+
+}

metricbeat/module/elasticsearch/cluster_stats/data_xpack.go

@@ -33,24 +33,19 @@ import (
 	"github.com/elastic/beats/metricbeat/module/elasticsearch"
 )
 
-func clusterNeedsTLSEnabled(license, stackStats common.MapStr) (bool, error) {
-	// TLS does not need to be enabled if license type is something other than trial
-	value, err := license.GetValue("type")
-	if err != nil {
-		return false, elastic.MakeErrorForMissingField("type", elastic.Elasticsearch)
-	}
-
-	licenseType, ok := value.(string)
-	if !ok {
-		return false, fmt.Errorf("license type is not a string")
-	}
+type clusterStatsLicense struct {
+	*elasticsearch.License
+	ClusterNeedsTLS bool `json:"cluster_needs_tls"`
+}
 
-	if licenseType != "trial" {
+func clusterNeedsTLSEnabled(license *elasticsearch.License, stackStats common.MapStr) (bool, error) {
+	// TLS does not need to be enabled if license type is something other than trial
+	if !license.IsOneOf("trial") {
 		return false, nil
 	}
 
 	// TLS does not need to be enabled if security is not enabled
-	value, err = stackStats.GetValue("security.enabled")
+	value, err := stackStats.GetValue("security.enabled")
 	if err != nil {
 		return false, elastic.MakeErrorForMissingField("security.enabled", elastic.Elasticsearch)
 	}
@@ -206,7 +201,8 @@ func eventMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, c
 	if err != nil {
 		return errors.Wrap(err, "failed to determine if cluster needs TLS enabled")
 	}
-	license.Put("cluster_needs_tls", clusterNeedsTLS) // This powers a cluster alert for enabling TLS on the ES transport protocol
+
+	l := clusterStatsLicense{license, clusterNeedsTLS}
 
 	isAPMFound, err := apmIndicesExist(clusterState)
 	if err != nil {
@@ -228,7 +224,7 @@ func eventMappingXPack(r mb.ReporterV2, m *MetricSet, info elasticsearch.Info, c
 		"timestamp":     common.Time(time.Now()),
 		"interval_ms":   m.Module().Config().Period / time.Millisecond,
 		"type":          "cluster_stats",
-		"license":       license,
+		"license":       l,
 		"version":       info.Version.Number,
 		"cluster_stats": clusterStats,
 		"cluster_state": clusterState,

metricbeat/module/elasticsearch/elasticsearch.go

@@ -59,6 +59,23 @@ type NodeInfo struct {
 	ID               string
 }
 
+// License contains data about the Elasticsearch license
+type License struct {
+	Status            string    `json:"status"`
+	ID                string    `json:"uid"`
+	Type              string    `json:"type"`
+	IssueDate         time.Time `json:"issue_date"`
+	IssueDateInMillis int       `json:"issue_date_in_millis"`
+	MaxNodes          int       `json:"max_nodes"`
+	IssuedTo          string    `json:"issued_to"`
+	Issuer            string    `json:"issuer"`
+	StartDateInMillis int       `json:"start_date_in_millis"`
+}
+
+type licenseWrapper struct {
+	License License `json:"license"`
+}
+
 // GetClusterID fetches cluster id for given nodeID.
 func GetClusterID(http *helper.HTTP, uri string, nodeID string) (string, error) {
 	// Check if cluster id already cached. If yes, return it.
@@ -187,7 +204,7 @@ func GetNodeInfo(http *helper.HTTP, uri string, nodeID string) (*NodeInfo, error
 // GetLicense returns license information. Since we don't expect license information
 // to change frequently, the information is cached for 1 minute to avoid
 // hitting Elasticsearch frequently.
-func GetLicense(http *helper.HTTP, resetURI string) (common.MapStr, error) {
+func GetLicense(http *helper.HTTP, resetURI string) (*License, error) {
 	// First, check the cache
 	license := licenseCache.get()
 
@@ -198,23 +215,14 @@ func GetLicense(http *helper.HTTP, resetURI string) (common.MapStr, error) {
 			return nil, err
 		}
 
-		var data common.MapStr
+		var data licenseWrapper
 		err = json.Unmarshal(content, &data)
 		if err != nil {
 			return nil, err
 		}
 
-		l, err := data.GetValue("license")
-		if err != nil {
-			return nil, err
-		}
-		license, ok := l.(map[string]interface{})
-		if !ok {
-			return nil, elastic.MakeErrorForMissingField("license", elastic.Elasticsearch)
-		}
-
 		// Cache license for a minute
-		licenseCache.set(license, time.Minute)
+		licenseCache.set(&data.License, time.Minute)
 	}
 
 	return licenseCache.get(), nil
@@ -332,23 +340,17 @@ func MergeClusterSettings(clusterSettings common.MapStr) (common.MapStr, error)
 	return settings, nil
 }
 
-// IsCCRStatsAPIAvailable returns whether the CCR stats API is available in the given version
-// of Elasticsearch.
-func IsCCRStatsAPIAvailable(currentElasticsearchVersion string) (bool, error) {
-	return elastic.IsFeatureAvailable(currentElasticsearchVersion, CCRStatsAPIAvailableVersion)
-}
-
 // Global cache for license information. Assumption is that license information changes infrequently.
 var licenseCache = &_licenseCache{}
 
 type _licenseCache struct {
 	sync.RWMutex
-	license  common.MapStr
+	license  *License
 	cachedOn time.Time
 	ttl      time.Duration
 }
 
-func (c *_licenseCache) get() common.MapStr {
+func (c *_licenseCache) get() *License {
 	c.Lock()
 	defer c.Unlock()
 
@@ -360,7 +362,7 @@ func (c *_licenseCache) get() common.MapStr {
 	return c.license
 }
 
-func (c *_licenseCache) set(license common.MapStr, ttl time.Duration) {
+func (c *_licenseCache) set(license *License, ttl time.Duration) {
 	c.Lock()
 	defer c.Unlock()
 
@@ -369,6 +371,18 @@ func (c *_licenseCache) set(license common.MapStr, ttl time.Duration) {
 	c.cachedOn = time.Now()
 }
 
+func (l *License) IsOneOf(candidateLicenses ...string) bool {
+	t := l.Type
+
+	for _, candidateLicense := range candidateLicenses {
+		if candidateLicense == t {
+			return true
+		}
+	}
+
+	return false
+}
+
 func getSettingGroup(allSettings common.MapStr, groupKey string) (common.MapStr, error) {
 	hasSettingGroup, err := allSettings.HasKey(groupKey)
 	if err != nil {

metricbeat/module/elasticsearch/elasticsearch_integration_test.go

@@ -33,6 +33,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/tests/compose"
+	"github.com/elastic/beats/metricbeat/helper/elastic"
 	mbtest "github.com/elastic/beats/metricbeat/mb/testing"
 	"github.com/elastic/beats/metricbeat/module/elasticsearch"
 	_ "github.com/elastic/beats/metricbeat/module/elasticsearch/ccr"
@@ -242,7 +243,7 @@ func checkSkip(t *testing.T, metricset string, host string) {
 		t.Fatal("getting elasticsearch version", err)
 	}
 
-	isCCRStatsAPIAvailable, err := elasticsearch.IsCCRStatsAPIAvailable(version)
+	isCCRStatsAPIAvailable, err := elastic.IsFeatureAvailable(version, elasticsearch.CCRStatsAPIAvailableVersion)
 	if err != nil {
 		t.Fatal("checking if elasticsearch CCR stats API is available", err)
 	}