[Agent] Grpc support for agent to beat communication

filebeat/beater/filebeat.go

@@ -22,15 +22,14 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/elastic/beats/libbeat/common/reload"
-
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/libbeat/autodiscover"
 	"github.com/elastic/beats/libbeat/beat"
 	"github.com/elastic/beats/libbeat/cfgfile"
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/cfgwarn"
+	"github.com/elastic/beats/libbeat/common/reload"
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/management"
 	"github.com/elastic/beats/libbeat/monitoring"

filebeat/filebeat.spec

@@ -0,0 +1 @@
+{"BinaryPath":"filebeat","Args":["-e"],"Configurable":"grpc"}

libbeat/cmd/instance/beat.go

@@ -618,7 +618,7 @@ func (b *Beat) configure(settings Settings) error {
 	logp.Info("Beat ID: %v", b.Info.ID)
 
 	// initialize config manager
-	b.ConfigManager, err = management.Factory()(b.Config.Management, reload.Register, b.Beat.Info.ID)
+	b.ConfigManager, err = management.Factory(b.Config.Management)(b.Config.Management, reload.Register, b.Beat.Info.ID)
 	if err != nil {
 		return err
 	}

libbeat/management/management.go

@@ -31,6 +31,8 @@ var Namespace = "libbeat.management"
 // DebugK used as key for all things central management
 var DebugK = "centralmgmt"
 
+var centralMgmtKey = "x-pack-cm"
+
 // ConfigManager interacts with the beat to update configurations
 // from an external source
 type ConfigManager interface {
@@ -47,32 +49,47 @@ type ConfigManager interface {
 	CheckRawConfig(cfg *common.Config) error
 }
 
+// PluginFunc for creating FactoryFunc if it matches a config
+type PluginFunc func(*common.Config) FactoryFunc
+
 // FactoryFunc for creating a config manager
 type FactoryFunc func(*common.Config, *reload.Registry, uuid.UUID) (ConfigManager, error)
 
 // Register a config manager
-func Register(name string, fn FactoryFunc, stability feature.Stability) {
+func Register(name string, fn PluginFunc, stability feature.Stability) {
 	f := feature.New(Namespace, name, fn, feature.NewDetails(name, "", stability))
 	feature.MustRegister(f)
 }
 
 // Factory retrieves config manager constructor. If no one is registered
 // it will create a nil manager
-func Factory() FactoryFunc {
+func Factory(cfg *common.Config) FactoryFunc {
 	factories, err := feature.GlobalRegistry().LookupAll(Namespace)
 	if err != nil {
 		return nilFactory
 	}
 
 	for _, f := range factories {
-		if factory, ok := f.Factory().(FactoryFunc); ok {
-			return factory
+		if plugin, ok := f.Factory().(PluginFunc); ok {
+			if factory := plugin(cfg); factory != nil {
+				return factory
+			}
 		}
 	}
 
 	return nilFactory
 }
 
+type modeConfig struct {
+	Mode string `config:"mode" yaml:"mode"`
+}
+
+func defaultModeConfig() *modeConfig {
+	return &modeConfig{
+		Mode: centralMgmtKey,
+	}
+}
+
 // nilManager, fallback when no manager is present
 type nilManager struct{}
 

x-pack/agent/pkg/agent/configrequest/step.go

@@ -9,10 +9,6 @@ const (
 	StepRun = "sc-run"
 	// StepRemove is a name of Remove program event causing beat in version to be uninstalled
 	StepRemove = "sc-remove"
-	// StepStartSidecar is a name of start program monitoring event
-	StepStartSidecar = "sc-sidecar-start"
-	// StepStopSidecar is a name of stop program monitoring event
-	StepStopSidecar = "sc-sidecar-stop"
 
 	// MetaConfigKey is key used to store configuration in metadata
 	MetaConfigKey = "config"

x-pack/agent/pkg/agent/operation/operator_handlers.go

@@ -20,8 +20,6 @@ func (o *Operator) initHandlerMap() {
 
 	hm[configrequest.StepRun] = o.handleRun
 	hm[configrequest.StepRemove] = o.handleRemove
-	hm[configrequest.StepStartSidecar] = o.handleStartSidecar
-	hm[configrequest.StepStopSidecar] = o.handleStopSidecar
 
 	o.handlers = hm
 }
@@ -46,7 +44,7 @@ func (o *Operator) handleRemove(step configrequest.Step) error {
 
 	p, _, err := getProgramFromStep(step)
 	if err != nil {
-		return errors.Wrap(err, "operator.handleStart failed to create program")
+		return errors.Wrap(err, "operator.handleRemove failed to stop program")
 	}
 
 	return o.stop(p)
@@ -67,15 +65,24 @@ func getProgramFromStepWithTags(step configrequest.Step, tags map[app.Tag]string
 }
 
 func getConfigFromStep(step configrequest.Step) (map[string]interface{}, error) {
-	metConfig, ok := step.Meta[configrequest.MetaConfigKey]
-	if !ok {
+	metConfig, hasConfig := step.Meta[configrequest.MetaConfigKey]
+
+	if !hasConfig && needsMetaConfig(step) {
 		return nil, fmt.Errorf("step: %s, no config in metadata", step.ID)
 	}
 
-	config, ok := metConfig.(map[string]interface{})
-	if !ok {
-		return nil, fmt.Errorf("step: %s, program config is in invalid format", step.ID)
+	var config map[string]interface{}
+	if hasConfig {
+		var ok bool
+		config, ok = metConfig.(map[string]interface{})
+		if !ok {
+			return nil, fmt.Errorf("step: %s, program config is in invalid format", step.ID)
+		}
 	}
 
 	return config, nil
 }
+
+func needsMetaConfig(step configrequest.Step) bool {
+	return step.ID == configrequest.StepRun
+}

x-pack/filebeat/filebeat.spec

@@ -0,0 +1 @@
+{"BinaryPath":"filebeat","Args":["-e"],"Configurable":"grpc"}

x-pack/libbeat/cmd/inject.go

@@ -10,8 +10,13 @@ import (
 
 	// register central management
 	"github.com/elastic/beats/x-pack/libbeat/licenser"
+
+	// Register Central Management
 	_ "github.com/elastic/beats/x-pack/libbeat/management"
 
+	// Register fleet
+	_ "github.com/elastic/beats/x-pack/libbeat/management/fleet"
+
 	// register autodiscover providers
 	_ "github.com/elastic/beats/x-pack/libbeat/autodiscover/providers/aws/elb"
 )