Query a remote Elasticsearch for License and features (#7946)

* License manager

Implements a License manager inside beats, as we development more
features that depends on the licensing and the capabilities of a remote
cluster we need a unique way to access that information. This commit
implements the following:

Add a License manager that can be started at the beginning of the beats
instance initialization. The manager takes a fetcher, currently we only
support Elasticsearch as the license backend but we could add support
for an Logstash endpoint that could proxy the license.

Notes:

- By default when the manager is started, no license is available,
calling `Get()` on the manager will return a license not found.

- The manager will periodically retrieve the license from the fetcher.

- When an error occurs on the periodic check, the license wont be
invalidated right away but will enter a grace period, after this period
the license will be invalidated and will replaced by the OSS license.

- License and capabilities and be retrieved by calling `Get()` or
registering a type implementing the `Watcher` interface.

libbeat/common/cli/cli.go

@@ -45,3 +45,14 @@ func RunWith(
 		}
 	}
 }
+
+// GetEnvOr return the value of the environment variable if the value is set, if its not set it will
+// return the default value.
+//
+// Note: if the value is set but it is an empty string we will return the empty string.
+func GetEnvOr(name, def string) string {
+	if env, ok := os.LookupEnv(name); ok {
+		return env
+	}
+	return def
+}

x-pack/beatless/data/meta.json

@@ -0,0 +1 @@
+{"uuid":"fd4356a8-16c9-4f5b-9261-e370617be071"}

x-pack/beatless/licenser/1

@@ -0,0 +1,43 @@
+{
+  "build": {
+    "hash": "595516e",
+    "date": "2018-08-17T23:22:27.102119Z"
+  },
+  "license": {
+    "uid": "936183d8-f48c-4a3f-959a-a52aa2563279",
+    "type": "trial",
+    "mode": "trial",
+    "status": "active",
+    "expiry_date_in_millis": 1538060781728
+  },
+  "features": {
+    "graph": {
+      "available": false,
+      "enabled": true
+    },
+    "logstash": {
+      "available": false,
+      "enabled": true
+    },
+    "ml": {
+      "available": false,
+      "enabled": true
+    },
+    "monitoring": {
+      "available": true,
+      "enabled": true
+    },
+    "rollup": {
+      "available": true,
+      "enabled": true
+    },
+    "security": {
+      "available": false,
+      "enabled": true
+    },
+    "watcher": {
+      "available": false,
+      "enabled": true
+    }
+  }
+}

x-pack/beatless/licenser/callback_watcher.go

@@ -0,0 +1,29 @@
+// 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
+
+// CallbackWatcher defines an addhoc listener for events generated by the manager.
+type CallbackWatcher struct {
+	New     func(License)
+	Stopped func()
+}
+
+// OnNewLicense is called when a new license is set in the manager.
+func (cb *CallbackWatcher) OnNewLicense(license License) {
+	if cb.New == nil {
+		return
+	}
+	cb.New(license)
+}
+
+// OnManagerStopped is called when the manager is stopped, watcher are expected to terminates any
+// features that depends on a specific license.
+func (cb *CallbackWatcher) OnManagerStopped() {
+	if cb.Stopped == nil {
+		return
+	}
+
+	cb.Stopped()
+}

x-pack/beatless/licenser/callback_watcher_test.go

@@ -0,0 +1,30 @@
+// 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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCallbackWatcher(t *testing.T) {
+	t.Run("when no callback is set do not execute anything", func(t *testing.T) {
+		w := &CallbackWatcher{}
+		w.OnNewLicense(License{})
+		w.OnManagerStopped()
+	})
+
+	t.Run("proxy call to callback function", func(t *testing.T) {
+		c := 0
+		w := &CallbackWatcher{
+			New:     func(license License) { c++ },
+			Stopped: func() { c++ },
+		}
+		w.OnNewLicense(License{})
+		w.OnManagerStopped()
+		assert.Equal(t, 2, c)
+	})
+}

x-pack/beatless/licenser/data/x-pack-trial-6.4.0.json

@@ -0,0 +1,43 @@
+{
+  "build": {
+    "hash": "595516e",
+    "date": "2018-08-17T23:22:27.102119Z"
+  },
+  "license": {
+    "uid": "936183d8-f48c-4a3f-959a-a52aa2563279",
+    "type": "trial",
+    "mode": "trial",
+    "status": "active",
+    "expiry_date_in_millis": 1538060781728
+  },
+  "features": {
+    "graph": {
+      "available": false,
+      "enabled": true
+    },
+    "logstash": {
+      "available": false,
+      "enabled": true
+    },
+    "ml": {
+      "available": false,
+      "enabled": true
+    },
+    "monitoring": {
+      "available": true,
+      "enabled": true
+    },
+    "rollup": {
+      "available": true,
+      "enabled": true
+    },
+    "security": {
+      "available": false,
+      "enabled": true
+    },
+    "watcher": {
+      "available": false,
+      "enabled": true
+    }
+  }
+}

x-pack/beatless/licenser/data/xpack-6.4.0.json

@@ -0,0 +1,42 @@
+{
+  "build": {
+    "hash": "053779d",
+    "date": "2018-07-20T05:25:16.206115Z"
+  },
+  "license": {
+    "uid": "936183d8-f48c-4a3f-959a-a52aa2563279",
+    "type": "platinum",
+    "mode": "platinum",
+    "status": "active"
+  },
+  "features": {
+    "graph": {
+      "available": false,
+      "enabled": true
+    },
+    "logstash": {
+      "available": false,
+      "enabled": true
+    },
+    "ml": {
+      "available": false,
+      "enabled": true
+    },
+    "monitoring": {
+      "available": true,
+      "enabled": true
+    },
+    "rollup": {
+      "available": true,
+      "enabled": true
+    },
+    "security": {
+      "available": false,
+      "enabled": true
+    },
+    "watcher": {
+      "available": false,
+      "enabled": true
+    }
+  }
+}

x-pack/beatless/licenser/elastic_fetcher.go

@@ -0,0 +1,147 @@
+// 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 (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/outputs/elasticsearch"
+)
+
+const xPackURL = "/_xpack"
+
+// params defaults query parameters to send to the '_xpack' endpoint by default we only need
+// machine parseable data.
+var params = map[string]string{
+	"human": "false",
+}
+
+var stateLookup = map[string]State{
+	"inactive": Inactive,
+	"active":   Active,
+}
+
+var licenseLookup = map[string]LicenseType{
+	"oss":      OSS,
+	"trial":    Trial,
+	"basic":    Basic,
+	"gold":     Gold,
+	"platinum": Platinum,
+}
+
+// UnmarshalJSON takes a bytes array and convert it to the appropriate license type.
+func (t *LicenseType) UnmarshalJSON(b []byte) error {
+	if len(b) <= 2 {
+		return fmt.Errorf("invalid string for license type, received: '%s'", string(b))
+	}
+	s := string(b[1 : len(b)-1])
+	if license, ok := licenseLookup[s]; ok {
+		*t = license
+		return nil
+	}
+
+	return fmt.Errorf("unknown license type, received: '%s'", s)
+}
+
+// UnmarshalJSON takes a bytes array and convert it to the appropriate state.
+func (st *State) UnmarshalJSON(b []byte) error {
+	// we are only interested in the content between the quotes.
+	if len(b) <= 2 {
+		return fmt.Errorf("invalid string for state, received: '%s'", string(b))
+	}
+
+	s := string(b[1 : len(b)-1])
+	if state, ok := stateLookup[s]; ok {
+		*st = state
+		return nil
+	}
+	return fmt.Errorf("unknown state, received: '%s'", s)
+}
+
+// UnmarshalJSON takes a bytes array and transform the int64 to a golang time.
+func (et *expiryTime) UnmarshalJSON(b []byte) error {
+	if len(b) < 0 {
+		return fmt.Errorf("invalid value for expiry time, received: '%s'", string(b))
+	}
+
+	ts, err := strconv.Atoi(string(b))
+	if err != nil {
+		return errors.Wrap(err, "could not parse value for expiry time")
+	}
+
+	*et = expiryTime(time.Unix(0, int64(time.Millisecond)*int64(ts)).UTC())
+	return nil
+}
+
+// ElasticFetcher wraps an elasticsearch clients to retrieve licensing information
+// on a specific cluster.
+type ElasticFetcher struct {
+	client *elasticsearch.Client
+	log    *logp.Logger
+}
+
+// NewElasticFetcher creates a new Elastic Fetcher
+func NewElasticFetcher(client *elasticsearch.Client) *ElasticFetcher {
+	return &ElasticFetcher{client: client, log: logp.NewLogger("elasticfetcher")}
+}
+
+// Fetch retrieves the license information from an Elasticsearch Client, it will call the `_xpack`
+// end point and will return a parsed license. If the `_xpack` endpoint is unreacheable we will
+// return the OSS License otherwise we return an error.
+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.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")
+	}
+
+	if status != http.StatusOK {
+		return nil, fmt.Errorf("could not retrieve license information, response code: %d", status)
+	}
+
+	if err != nil {
+		return nil, errors.Wrap(err, "could not retrieve the license information from the cluster")
+	}
+
+	license, err := f.parseJSON(body)
+	if err != nil {
+		f.log.Debugw("invalid response from server", "body", string(body))
+		return nil, errors.Wrap(err, "could not extract license information from the server response")
+	}
+
+	return license, nil
+}
+
+// Xpack Response, temporary struct to merge the features into the license struct.
+type xpackResponse struct {
+	License  License  `json:"license"`
+	Features features `json:"features"`
+}
+
+func (f *ElasticFetcher) parseJSON(b []byte) (*License, error) {
+	info := &xpackResponse{}
+
+	if err := json.Unmarshal(b, info); err != nil {
+		return nil, err
+	}
+
+	license := info.License
+	license.Features = info.Features
+
+	return &license, nil
+}