Check license x-pack (elastic#11296)

* Check License for basic or better

CHANGELOG.next.asciidoc

@@ -35,6 +35,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
   info from the cloud.machine.type and cloud.availability_zone. {issue}10968[10968]
 - Empty `meta.json` file will be treated as a missing meta file. {issue}8558[8558]
 - Rename `migration.enabled` config to `migration.6_to_7.enabled`. {pull}11284[11284]
+- Beats Xpack now checks for Basic license on connect. {pull}11296[11296]
 
 *Auditbeat*
 

libbeat/outputs/elasticsearch/client.go

@@ -225,6 +225,16 @@ func NewClient(
 	}
 
 	client.Connection.onConnectCallback = func() error {
+		globalCallbackRegistry.mutex.Lock()
+		defer globalCallbackRegistry.mutex.Unlock()
+
+		for _, callback := range globalCallbackRegistry.callbacks {
+			err := callback(client)
+			if err != nil {
+				return err
+			}
+		}
+
 		if onConnect != nil {
 			onConnect.mutex.Lock()
 			defer onConnect.mutex.Unlock()
@@ -721,7 +731,7 @@ func (conn *Connection) Ping() (string, error) {
 	}
 
 	debugf("Ping status code: %v", status)
-	logp.Info("Connected to Elasticsearch version %s", response.Version.Number)
+	logp.Info("Attempting to connect to Elasticsearch version %s", response.Version.Number)
 	return response.Version.Number, nil
 }
 

libbeat/outputs/elasticsearch/elasticsearch.go

@@ -61,6 +61,31 @@ type callbacksRegistry struct {
 // XXX: it would be fantastic to do this without a package global
 var connectCallbackRegistry = newCallbacksRegistry()
 
+// NOTE(ph): We need to refactor this, right now this is the only way to ensure that every calls
+// to an ES cluster executes a callback.
+var globalCallbackRegistry = newCallbacksRegistry()
+
+// RegisterGlobalCallback register a global callbacks.
+func RegisterGlobalCallback(callback connectCallback) (uuid.UUID, error) {
+	globalCallbackRegistry.mutex.Lock()
+	defer globalCallbackRegistry.mutex.Unlock()
+
+	// find the next unique key
+	var key uuid.UUID
+	var err error
+	exists := true
+	for exists {
+		key, err = uuid.NewV4()
+		if err != nil {
+			return uuid.Nil, err
+		}
+		_, exists = globalCallbackRegistry.callbacks[key]
+	}
+
+	globalCallbackRegistry.callbacks[key] = callback
+	return key, nil
+}
+
 func newCallbacksRegistry() callbacksRegistry {
 	return callbacksRegistry{
 		callbacks: make(map[uuid.UUID]connectCallback),
@@ -99,6 +124,15 @@ func DeregisterConnectCallback(key uuid.UUID) {
 	delete(connectCallbackRegistry.callbacks, key)
 }
 
+// DeregisterGlobalCallback deregisters a callback for the elasticsearch output
+// specified by its key. If a callback does not exist, nothing happens.
+func DeregisterGlobalCallback(key uuid.UUID) {
+	globalCallbackRegistry.mutex.Lock()
+	defer globalCallbackRegistry.mutex.Unlock()
+
+	delete(globalCallbackRegistry.callbacks, key)
+}
+
 func makeES(
 	im outputs.IndexManager,
 	beat beat.Info,

libbeat/outputs/elasticsearch/elasticsearch_test.go

@@ -46,3 +46,28 @@ func TestConnectCallbacksManagement(t *testing.T) {
 		t.Fatalf("third callback cannot be retrieved")
 	}
 }
+
+func TestGlobalConnectCallbacksManagement(t *testing.T) {
+	f0 := func(client *Client) error { fmt.Println("i am function #0"); return nil }
+	f1 := func(client *Client) error { fmt.Println("i am function #1"); return nil }
+	f2 := func(client *Client) error { fmt.Println("i am function #2"); return nil }
+
+	_, err := RegisterGlobalCallback(f0)
+	if err != nil {
+		t.Fatalf("error while registering callback: %v", err)
+	}
+	id1, err := RegisterGlobalCallback(f1)
+	if err != nil {
+		t.Fatalf("error while registering callback: %v", err)
+	}
+	id2, err := RegisterGlobalCallback(f2)
+	if err != nil {
+		t.Fatalf("error while registering callback: %v", err)
+	}
+
+	t.Logf("removing second callback")
+	DeregisterGlobalCallback(id1)
+	if _, ok := globalCallbackRegistry.callbacks[id2]; !ok {
+		t.Fatalf("third callback cannot be retrieved")
+	}
+}

x-pack/functionbeat/beater/functionbeat.go

@@ -20,8 +20,8 @@ import (
 	"github.com/elastic/beats/x-pack/functionbeat/config"
 	"github.com/elastic/beats/x-pack/functionbeat/core"
 	_ "github.com/elastic/beats/x-pack/functionbeat/include" // imports features
-	"github.com/elastic/beats/x-pack/functionbeat/licenser"
 	"github.com/elastic/beats/x-pack/functionbeat/provider"
+	"github.com/elastic/beats/x-pack/libbeat/licenser"
 )
 
 var (
@@ -82,7 +82,7 @@ func (bt *Functionbeat) Run(b *beat.Beat) error {
 	defer manager.Stop()
 
 	// Wait until we receive the initial license.
-	if err := licenser.WaitForLicense(bt.ctx, bt.log, manager, checkLicense); err != nil {
+	if err := licenser.WaitForLicense(bt.ctx, bt.log, manager, licenser.BasicAndAboveOrTrial); err != nil {
 		return err
 	}
 
@@ -156,7 +156,7 @@ func makeClientFactory(log *logp.Logger, manager *licenser.Manager, pipeline bea
 
 		// Make the client aware of the current license, the client will accept sending events to the
 		// pipeline until the client is closed or if the license change and is not valid.
-		licenseAware := core.NewLicenseAwareClient(client, checkLicense)
+		licenseAware := core.NewLicenseAwareClient(client, licenser.BasicAndAboveOrTrial)
 		if err := manager.AddWatcher(licenseAware); err != nil {
 			return nil, err
 		}

x-pack/functionbeat/core/license_client.go

@@ -10,7 +10,7 @@ import (
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/common/atomic"
 	"github.com/elastic/beats/libbeat/logp"
-	"github.com/elastic/beats/x-pack/functionbeat/licenser"
+	"github.com/elastic/beats/x-pack/libbeat/licenser"
 )
 
 var errInvalidLicense = errors.New("invalid license detected, cannot publish events")

x-pack/functionbeat/core/license_client_test.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/logp"
-	"github.com/elastic/beats/x-pack/functionbeat/licenser"
+	"github.com/elastic/beats/x-pack/libbeat/licenser"
 )
 
 type dummySyncClient struct{ EventCount int }

x-pack/libbeat/cmd/inject.go

@@ -6,12 +6,17 @@ package cmd
 
 import (
 	"github.com/elastic/beats/libbeat/cmd"
+	"github.com/elastic/beats/libbeat/logp"
 
 	// register central management
+	"github.com/elastic/beats/x-pack/libbeat/licenser"
 	_ "github.com/elastic/beats/x-pack/libbeat/management"
 )
 
+const licenseDebugK = "license"
+
 // AddXPack extends the given root folder with XPack features
 func AddXPack(root *cmd.BeatsRootCmd, name string) {
+	licenser.Enforce(logp.NewLogger(licenseDebugK), licenser.BasicAndAboveOrTrial)
 	root.AddCommand(genEnrollCmd(name, ""))
 }

x-pack/functionbeat/licenser/check.go → x-pack/libbeat/licenser/check.go

@@ -19,6 +19,7 @@ func CheckTrial(log *logp.Logger, license License) bool {
 			log.Error("Trial license is expired")
 			return false
 		}
+		log.Info("Trial license active")
 		return true
 	}
 	return false
@@ -27,10 +28,11 @@ func CheckTrial(log *logp.Logger, license License) bool {
 // CheckLicenseCover check that the current license cover the requested license.
 func CheckLicenseCover(licenseType LicenseType) func(*logp.Logger, License) bool {
 	return func(log *logp.Logger, license License) bool {
-		log.Debugf("Checking that license cover %s", licenseType)
+		log.Debug("Checking that license covers %s", licenseType)
 		if license.Cover(licenseType) && license.IsActive() {
 			return true
 		}
+		log.Infof("License is active for %s", licenseType)
 		return false
 	}
 }
@@ -48,3 +50,8 @@ func Validate(log *logp.Logger, license License, checks ...CheckFunc) bool {
 	}
 	return false
 }
+
+// BasicAndAboveOrTrial return true if the license is basic or if the license is trial and active.
+func BasicAndAboveOrTrial(log *logp.Logger, license License) bool {
+	return CheckBasic(log, license) || CheckTrial(log, license)
+}

x-pack/functionbeat/licenser/elastic_fetcher.go → x-pack/libbeat/licenser/elastic_fetcher.go

@@ -114,13 +114,18 @@ func (f *ElasticFetcher) Fetch() (*License, error) {
 	status, body, err := f.client.Request("GET", xPackURL, "", params, nil)
 	// When we are running an OSS release of elasticsearch the _xpack endpoint will return a 405,
 	// "Method Not Allowed", so we return the default OSS license.
+	if status == http.StatusBadRequest {
+		f.log.Debug("Received 'Bad request' (400) response from server, fallback to OSS license")
+		return OSSLicense, nil
+	}
+
 	if status == http.StatusMethodNotAllowed {
 		f.log.Debug("Received 'Method Not allowed' (405) response from server, fallback to OSS license")
 		return OSSLicense, nil
 	}
 
 	if status == http.StatusUnauthorized {
-		return nil, errors.New("Unauthorized access, could not connect to the xpack endpoint, verify your credentials")
+		return nil, errors.New("unauthorized access, could not connect to the xpack endpoint, verify your credentials")
 	}
 
 	if status != http.StatusOK {

x-pack/functionbeat/licenser/elastic_fetcher_integration_test.go → x-pack/libbeat/licenser/elastic_fetcher_integration_test.go

@@ -27,8 +27,8 @@ func getTestClient() *elasticsearch.Client {
 	client, err := elasticsearch.NewClient(elasticsearch.ClientSettings{
 		URL:              host,
 		Index:            outil.MakeSelector(),
-		Username:         cli.GetEnvOr("ES_USER", ""),
-		Password:         cli.GetEnvOr("ES_PASS", ""),
+		Username:         "myelastic", // NOTE: I will refactor this in a followup PR
+		Password:         "changeme",
 		Timeout:          60 * time.Second,
 		CompressionLevel: 3,
 	}, nil)

x-pack/functionbeat/licenser/elastic_fetcher_test.go → x-pack/libbeat/licenser/elastic_fetcher_test.go

@@ -35,7 +35,7 @@ func newServerClientPair(t *testing.T, handler http.HandlerFunc) (*httptest.Serv
 }
 
 func TestParseJSON(t *testing.T) {
-	t.Run("OSS release of Elasticsearch", func(t *testing.T) {
+	t.Run("OSS release of Elasticsearch (Code: 405)", func(t *testing.T) {
 		h := func(w http.ResponseWriter, r *http.Request) {
 			http.Error(w, "Method Not Allowed", 405)
 		}
@@ -52,6 +52,23 @@ func TestParseJSON(t *testing.T) {
 		assert.Equal(t, OSSLicense, oss)
 	})
 
+	t.Run("OSS release of Elasticsearch (Code: 400)", func(t *testing.T) {
+		h := func(w http.ResponseWriter, r *http.Request) {
+			http.Error(w, "Bad Request", 400)
+		}
+		s, c := newServerClientPair(t, h)
+		defer s.Close()
+		defer c.Close()
+
+		fetcher := NewElasticFetcher(c)
+		oss, err := fetcher.Fetch()
+		if assert.NoError(t, err) {
+			return
+		}
+
+		assert.Equal(t, OSSLicense, oss)
+	})
+
 	t.Run("malformed JSON", func(t *testing.T) {
 		h := func(w http.ResponseWriter, r *http.Request) {
 			w.Write([]byte("hello bad JSON"))
@@ -75,7 +92,7 @@ func TestParseJSON(t *testing.T) {
 
 		fetcher := NewElasticFetcher(c)
 		_, err := fetcher.Fetch()
-		assert.Equal(t, err.Error(), "Unauthorized access, could not connect to the xpack endpoint, verify your credentials")
+		assert.Equal(t, err.Error(), "unauthorized access, could not connect to the xpack endpoint, verify your credentials")
 	})
 
 	t.Run("any error from the server", func(t *testing.T) {

x-pack/libbeat/licenser/es_callback.go

@@ -0,0 +1,44 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package licenser
+
+import (
+	"fmt"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/outputs/elasticsearch"
+)
+
+// Enforce setups the corresponding callbacks in libbeat to verify the license on the
+// remote elasticsearch cluster.
+func Enforce(log *logp.Logger, checks ...CheckFunc) {
+	cb := func(client *elasticsearch.Client) error {
+		fetcher := NewElasticFetcher(client)
+		license, err := fetcher.Fetch()
+
+		if err != nil {
+			return errors.Wrapf(err, "cannot retrieve the elasticsearch license")
+		}
+
+		if license == OSSLicense {
+			return errors.New("This Beat requires the default distribution of Elasticsearch. Please " +
+				"upgrade to the default distribution of Elasticsearch from elastic.co, or downgrade to " +
+				"the oss-only distribution of beats")
+		}
+
+		if !Validate(log, *license, checks...) {
+			return fmt.Errorf(
+				"invalid license found, requires a basic or a valid trial license and received %s",
+				license.Get(),
+			)
+		}
+
+		return nil
+	}
+
+	elasticsearch.RegisterGlobalCallback(cb)
+}