[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"
 )

x-pack/libbeat/management/config.go

@@ -68,11 +68,23 @@ const ManagedConfigTemplate = `
 #monitoring.elasticsearch:
 `
 
-// Config for central management
+const (
+	// ModeCentralManagement is a default CM mode, using existing processes.
+	ModeCentralManagement = "x-pack-cm"
+
+	// ModeFleet is a management mode where fleet is used to retrieve configurations.
+	ModeFleet = "x-pack-fleet"
+)
+
+// Config for central management.
 type Config struct {
 	// true when enrolled
 	Enabled bool `config:"enabled" yaml:"enabled"`
 
+	// Mode specifies whether beat uses Central Management or Fleet.
+	// Options: [cm, fleet]
+	Mode string `config:"mode" yaml:"mode"`
+
 	// Poll configs period
 	Period time.Duration `config:"period" yaml:"period"`
 
@@ -93,6 +105,7 @@ type EventReporterConfig struct {
 
 func defaultConfig() *Config {
 	return &Config{
+		Mode:   ModeCentralManagement,
 		Period: 60 * time.Second,
 		EventsReporter: EventReporterConfig{
 			Period:       30 * time.Second,
@@ -111,7 +124,7 @@ type templateParams struct {
 	BeatName                  string
 }
 
-// OverwriteConfigFile will overwrite beat settings file with the enrolled template
+// OverwriteConfigFile will overwrite beat settings file with the enrolled template.
 func (c *Config) OverwriteConfigFile(wr io.Writer, beatName string) error {
 	t := template.Must(template.New("beat.management.yml").Parse(ManagedConfigTemplate))
 

x-pack/libbeat/management/error.go

@@ -103,6 +103,7 @@ func (er *Errors) IsEmpty() bool {
 	return len(*er) == 0
 }
 
-func newConfigError(err error) *Error {
+// NewConfigError wraps an error to be a management error of a specific ConfigError Type
+func NewConfigError(err error) *Error {
 	return &Error{Type: ConfigError, Err: err}
 }

x-pack/libbeat/management/error_test.go

@@ -54,14 +54,14 @@ func TestErrorSerialization(t *testing.T) {
 
 func TestErrors(t *testing.T) {
 	t.Run("single error", func(t *testing.T) {
-		errors := Errors{newConfigError(errors.New("error1"))}
+		errors := Errors{NewConfigError(errors.New("error1"))}
 		assert.Equal(t, "1 error: error1", errors.Error())
 	})
 
 	t.Run("multiple errors", func(t *testing.T) {
 		errors := Errors{
-			newConfigError(errors.New("error1")),
-			newConfigError(errors.New("error2")),
+			NewConfigError(errors.New("error1")),
+			NewConfigError(errors.New("error2")),
 		}
 		assert.Equal(t, "2 errors: error1; error2", errors.Error())
 	})

x-pack/libbeat/management/fleet/config.go

@@ -0,0 +1,27 @@
+// 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 fleet
+
+import (
+	xmanagement "github.com/elastic/beats/x-pack/libbeat/management"
+)
+
+// Config for central management
+type Config struct {
+	Enabled   bool                                `config:"enabled" yaml:"enabled"`
+	Mode      string                              `config:"mode" yaml:"mode"`
+	Blacklist xmanagement.ConfigBlacklistSettings `config:"blacklist" yaml:"blacklist"`
+}
+
+func defaultConfig() *Config {
+	return &Config{
+		Mode: xmanagement.ModeCentralManagement,
+		Blacklist: xmanagement.ConfigBlacklistSettings{
+			Patterns: map[string]string{
+				"output": "console|file",
+			},
+		},
+	}
+}

x-pack/libbeat/management/fleet/config_server.go

@@ -0,0 +1,56 @@
+// 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 fleet
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/x-pack/agent/pkg/core/remoteconfig/grpc"
+)
+
+const (
+	defaultTimeout = 15 * time.Second
+)
+
+// Server is a server for handling communication between
+// beat and Elastic Agent.
+type Server struct {
+	configChan chan<- map[string]interface{}
+}
+
+// NewConfigServer creates a new grpc configuration server for receiving
+// configurations from Elastic Agent.
+func NewConfigServer(configChan chan<- map[string]interface{}) *Server {
+	return &Server{
+		configChan: configChan,
+	}
+}
+
+// Config is a handler of a call made by agent pushing latest configuration.
+func (s *Server) Config(ctx context.Context, req *grpc.ConfigRequest) (*grpc.ConfigResponse, error) {
+	cfgString := req.GetConfig()
+
+	var configMap common.MapStr
+	uconfig, err := common.NewConfigFrom(cfgString)
+	if err != nil {
+		return &grpc.ConfigResponse{}, fmt.Errorf("config blocks unsuccessfully generated: %+v", err)
+	}
+
+	err = uconfig.Unpack(&configMap)
+	if err != nil {
+		return &grpc.ConfigResponse{}, fmt.Errorf("config blocks unsuccessfully generated: %+v", err)
+	}
+
+	select {
+	case s.configChan <- configMap:
+	case <-time.After(defaultTimeout):
+		return &grpc.ConfigResponse{}, errors.New("failed to push configuration: Timeout")
+	}
+	return &grpc.ConfigResponse{}, nil
+}