Refactor: Introduce a Global Registry

libbeat/feature/bundle.go

@@ -0,0 +1,55 @@
+package feature
+
+import "fmt"
+
+// Bundle defines a list of features available in the current beat.
+type Bundle struct {
+	Features []Featurable
+}
+
+// NewBundle creates a new Bundle of feature to be registered.
+func NewBundle(features []Featurable) *Bundle {
+	return &Bundle{Features: features}
+}
+
+// Filter creates a new bundle with only the feature matching the requested stability.
+func (b *Bundle) Filter(stabilities ...Stability) *Bundle {
+	var filtered []Featurable
+
+	for _, feature := range b.Features {
+		for _, stability := range stabilities {
+			if feature.Stability() == stability {
+				filtered = append(filtered, feature)
+				break
+			}
+		}
+	}
+	return NewBundle(filtered)
+}
+
+// MustBundle takes existing bundle or features and create a new Bundle with all the merged Features,
+// will panic on errors.
+func MustBundle(features ...interface{}) *Bundle {
+	b, err := BundleFeature(features...)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
+// BundleFeature takes existing bundle or features and create a new Bundle with all the merged
+// Features,
+func BundleFeature(features ...interface{}) (*Bundle, error) {
+	var merged []Featurable
+	for _, feature := range features {
+		switch v := feature.(type) {
+		case Featurable:
+			merged = append(merged, v)
+		case *Bundle:
+			merged = append(merged, v.Features...)
+		default:
+			return nil, fmt.Errorf("unknown type, expecting 'Featurable' or 'Bundle' and received '%T'", v)
+		}
+	}
+	return NewBundle(merged), nil
+}

libbeat/feature/bundle_test.go

@@ -0,0 +1,54 @@
+package feature
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBundle(t *testing.T) {
+	factory := func() {}
+	features := []Featurable{
+		New("libbeat.outputs", "elasticsearch", factory, Stable),
+		New("libbeat.outputs", "edge", factory, Experimental),
+		New("libbeat.input", "tcp", factory, Beta),
+	}
+
+	t.Run("Creates a new Bundle", func(t *testing.T) {
+		b := NewBundle(features)
+		assert.Equal(t, 3, len(b.Features))
+	})
+
+	t.Run("Filters feature based on stability", func(t *testing.T) {
+		b := NewBundle(features)
+		new := b.Filter(Experimental)
+		assert.Equal(t, 1, len(new.Features))
+	})
+
+	t.Run("Filters feature based on multiple different stability", func(t *testing.T) {
+		b := NewBundle(features)
+		new := b.Filter(Experimental, Stable)
+		assert.Equal(t, 2, len(new.Features))
+	})
+
+	t.Run("Creates a new Bundle from specified feature", func(t *testing.T) {
+		f1 := New("libbeat.outputs", "elasticsearch", factory, Stable)
+		b := MustBundle(f1)
+		assert.Equal(t, 1, len(b.Features))
+	})
+
+	t.Run("Creates a new Bundle with grouped features", func(t *testing.T) {
+		f1 := New("libbeat.outputs", "elasticsearch", factory, Stable)
+		f2 := New("libbeat.outputs", "edge", factory, Experimental)
+		f3 := New("libbeat.input", "tcp", factory, Beta)
+		f4 := New("libbeat.input", "udp", factory, Beta)
+
+		b := MustBundle(
+			MustBundle(f1),
+			MustBundle(f2),
+			MustBundle(f3, f4),
+		)
+
+		assert.Equal(t, 4, len(b.Features))
+	})
+}

libbeat/feature/feature.go

@@ -0,0 +1,120 @@
+package feature
+
+import (
+	"fmt"
+	"reflect"
+)
+
+//go:generate stringer -type=Stability
+
+// Registry is the global plugin registry, this variable is mean to be temporary to move all the
+// internal factory to receive a context that include the current beat registry.
+var Registry = newRegistry()
+
+// Featurable implements the description of a feature.
+type Featurable interface {
+	// Namespace returns the namespace of the Feature.
+	Namespace() string
+
+	// Name returns the name of the feature. The name must be unique for each namespace.
+	Name() string
+
+	// Factory returns the factory func.
+	Factory() interface{}
+
+	// Stability returns the stability of the feature.
+	Stability() Stability
+
+	// Equal returns true if the two object are equal.
+	Equal(other Featurable) bool
+
+	String() string
+}
+
+// Feature contains the information for a specific feature
+type Feature struct {
+	namespace string
+	name      string
+	factory   interface{}
+	stability Stability
+}
+
+// Namespace return the namespace of the feature.
+func (f *Feature) Namespace() string {
+	return f.namespace
+}
+
+// Name returns the name of the feature.
+func (f *Feature) Name() string {
+	return f.name
+}
+
+// Factory returns the factory for the feature.
+func (f *Feature) Factory() interface{} {
+	return f.factory
+}
+
+// Stability returns the stability level of the feature, current: stable, beta, experimental.
+func (f *Feature) Stability() Stability {
+	return f.stability
+}
+
+// Equal return true if both object are equals.
+func (f *Feature) Equal(other Featurable) bool {
+	// There is no safe way to compare function in go,
+	// but since the method are global it should be stable.
+	if f.Name() == other.Name() &&
+		f.Namespace() == other.Namespace() &&
+		reflect.ValueOf(f.Factory()).Pointer() == reflect.ValueOf(other.Factory()).Pointer() {
+		return true
+	}
+
+	return false
+}
+
+// String return the debug information
+func (f *Feature) String() string {
+	return fmt.Sprintf("%s/%s (stability: %s)", f.namespace, f.name, f.stability)
+}
+
+// Stability defines the stability of the feature, this value can be used to filter a bundler.
+type Stability int
+
+// List all the available stability for a feature.
+const (
+	Stable Stability = iota
+	Beta
+	Experimental
+	Undefined
+)
+
+// New returns a new Feature.
+func New(namespace, name string, factory interface{}, stability Stability) *Feature {
+	return &Feature{
+		namespace: namespace,
+		name:      name,
+		factory:   factory,
+		stability: stability,
+	}
+}
+
+// RegisterBundle registers a bundle of features.
+func RegisterBundle(bundle Bundle) error {
+	for _, f := range bundle.Features {
+		Registry.Register(f)
+	}
+	return nil
+}
+
+// Register register a new feature on the global registry.
+func Register(feature Featurable) error {
+	return Registry.Register(feature)
+}
+
+// MustRegister register a new Feature on the global registry and panic on error.
+func MustRegister(feature Featurable) {
+	err := Register(feature)
+	if err != nil {
+		panic(err)
+	}
+}

libbeat/feature/registry.go

@@ -0,0 +1,141 @@
+package feature
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/elastic/beats/libbeat/logp"
+)
+
+type mapper map[string]map[string]Featurable
+
+// Registry implements a global registry for any kind of feature in beats.
+// feature are grouped by namespace, a namespace is a kind of plugin like outputs, inputs, or queue.
+// The feature name must be unique.
+type registry struct {
+	sync.RWMutex
+	namespaces mapper
+	log        *logp.Logger
+}
+
+// NewRegistry returns a new registry.
+func newRegistry() *registry {
+	return &registry{
+		namespaces: make(mapper),
+		log:        logp.NewLogger("registry"),
+	}
+}
+
+// Register registers a new feature into a specific namespace, namespace are lazy created.
+// Feature name must be unique.
+func (r *registry) Register(feature Featurable) error {
+	r.Lock()
+	defer r.Unlock()
+
+	// Lazy create namespaces
+	_, found := r.namespaces[feature.Namespace()]
+	if !found {
+		r.namespaces[feature.Namespace()] = make(map[string]Featurable)
+	}
+
+	f, found := r.namespaces[feature.Namespace()][feature.Name()]
+	if found {
+		if feature.Equal(f) {
+			// Allow both old style and new style of plugin to work together.
+			r.log.Debugw(
+				"ignoring, feature '%s' is already registered in the namespace '%s'",
+				feature.Name(),
+				feature.Namespace(),
+			)
+			return nil
+		}
+
+		return fmt.Errorf(
+			"could not register new feature '%s' in namespace '%s', feature name must be unique",
+			feature.Name(),
+			feature.Namespace(),
+		)
+	}
+
+	r.log.Debugw(
+		"registering new feature",
+		"namespace",
+		feature.Namespace(),
+		"name",
+		feature.Name(),
+	)
+
+	r.namespaces[feature.Namespace()][feature.Name()] = feature
+
+	return nil
+}
+
+// Unregister removes a feature from the registry.
+func (r *registry) Unregister(namespace, name string) error {
+	r.Lock()
+	defer r.Unlock()
+
+	v, found := r.namespaces[namespace]
+	if !found {
+		return fmt.Errorf("unknown namespace named '%s'", namespace)
+	}
+
+	_, found = v[name]
+	if !found {
+		return fmt.Errorf("unknown feature '%s' in namespace '%s'", name, namespace)
+	}
+
+	delete(r.namespaces[namespace], name)
+	return nil
+}
+
+// Find returns a specific Find from a namespace or an error if not found.
+func (r *registry) Find(namespace, name string) (Featurable, error) {
+	r.RLock()
+	defer r.RUnlock()
+
+	v, found := r.namespaces[namespace]
+	if !found {
+		return nil, fmt.Errorf("unknown namespace named '%s'", namespace)
+	}
+
+	m, found := v[name]
+	if !found {
+		return nil, fmt.Errorf("unknown feature '%s' in namespace '%s'", name, namespace)
+	}
+
+	return m, nil
+}
+
+// FindAll returns all the features for a specific namespace.
+func (r *registry) FindAll(namespace string) ([]Featurable, error) {
+	r.RLock()
+	defer r.RUnlock()
+
+	v, found := r.namespaces[namespace]
+	if !found {
+		return nil, fmt.Errorf("unknown namespace named '%s'", namespace)
+	}
+
+	list := make([]Featurable, len(v))
+	c := 0
+	for _, feature := range v {
+		list[c] = feature
+		c++
+	}
+
+	return list, nil
+}
+
+// Size returns the number of registered features in the registry.
+func (r *registry) Size() int {
+	r.RLock()
+	defer r.RUnlock()
+
+	c := 0
+	for _, namespace := range r.namespaces {
+		c += len(namespace)
+	}
+
+	return c
+}