Agent v2

beater/cloudbeat.go

@@ -71,7 +71,7 @@ func New(b *beat.Beat, cfg *agentconfig.C) (beat.Beater, error) {
 		return nil, err
 	}
 
-	reload.Register.MustRegisterList("inputs", reloader)
+	reload.RegisterV2.MustRegisterInput(reloader)
 	return s, nil
 }
 

beater/validator.go

@@ -27,14 +27,9 @@ import (
 	agentconfig "github.com/elastic/elastic-agent-libs/config"
 )
 
-type validator struct {
-}
+type validator struct{}
 
 func (v *validator) Validate(cfg *agentconfig.C) error {
-	if !cfg.HasField("streams") {
-		return fmt.Errorf("no streams in config")
-	}
-
 	c, err := config.New(cfg)
 	if err != nil {
 		return fmt.Errorf("could not parse reconfiguration %v, skipping with error: %v", cfg.FlattenedKeys(), err)

beater/validator_test.go

@@ -45,39 +45,15 @@ func TestValidatorTestSuite(t *testing.T) {
 }
 
 func (s *ValidatorTestSuite) TestConfig() {
-	configNoStreams := config.MustNewConfigFrom(`
-not_streams:
-  - runtime_cfg:
-      activated_rules:
-        cis_k8s:
-          - a
-          - b
-          - c
-          - d
-          - e
-`)
-
-	configNoRuntimeCfg := config.MustNewConfigFrom(`
-streams:
-  - not_runtime_cfg:
-      activated_rules:
-        cis_k8s:
-          - a
-          - b
-          - c
-          - d
-          - e
-`)
 	configWithRuntimeCfg := config.MustNewConfigFrom(`
-streams:
-  - runtime_cfg:
-      activated_rules:
-        cis_k8s:
-          - a
-          - b
-          - c
-          - d
-          - e
+runtime_cfg:
+  activated_rules:
+    cis_k8s:
+      - a
+      - b
+      - c
+      - d
+      - e
 `)
 
 	testcases := []struct {
@@ -87,13 +63,8 @@ streams:
 		{
 			true,
 			config.NewConfig(),
-		}, {
-			true,
-			configNoRuntimeCfg,
-		}, {
-			true,
-			configNoStreams,
-		}, {
+		},
+		{
 			false,
 			configWithRuntimeCfg,
 		},

cmd/root.go

@@ -18,16 +18,41 @@
 package cmd
 
 import (
+	"fmt"
+
 	"github.com/elastic/cloudbeat/beater"
+	"github.com/elastic/elastic-agent-client/v7/pkg/client"
+	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
 
 	cmd "github.com/elastic/beats/v7/libbeat/cmd"
 	"github.com/elastic/beats/v7/libbeat/cmd/instance"
+	"github.com/elastic/beats/v7/libbeat/common/reload"
 
 	_ "github.com/elastic/beats/v7/x-pack/libbeat/include"
+	"github.com/elastic/beats/v7/x-pack/libbeat/management"
 )
 
 // Name of this beat
 var Name = "cloudbeat"
 
 // RootCmd to handle beats cli
 var RootCmd = cmd.GenRootCmdWithSettings(beater.New, instance.Settings{Name: Name, Version: defaultBeatVersion})
+
+func cloudbeatCfg(rawIn *proto.UnitExpectedConfig, agentInfo *client.AgentInfo) ([]*reload.ConfigWithMeta, error) {
+	modules, err := management.CreateInputsFromStreams(rawIn, "logs", agentInfo)
+	if err != nil {
+		return nil, fmt.Errorf("error creating input list from raw expected config: %w", err)
+	}
+
+	// format for the reloadable list needed bythe cm.Reload() method
+	configList, err := management.CreateReloadConfigFromInputs(modules)
+	if err != nil {
+		return nil, fmt.Errorf("error creating reloader config: %w", err)
+	}
+
+	return configList, nil
+}
+
+func init() {
+	management.ConfigTransform.SetTransform(cloudbeatCfg)
+}

config/config.go

@@ -44,11 +44,6 @@ type Fetcher struct {
 	Name string `config:"name"` // Name of the fetcher
 }
 
-type AgentInput struct {
-	Streams []Stream `config:"streams"`
-	Type    string   `config:"type"`
-}
-
 type Stream struct {
 	AWSConfig  aws.ConfigAWS           `config:",inline"`
 	RuntimeCfg *RuntimeConfig          `config:"runtime_cfg"`
@@ -77,23 +72,26 @@ type Benchmarks struct {
 	CisEks []string `config:"cis_eks,omitempty" yaml:"cis_eks,omitempty" json:"cis_eks,omitempty"`
 }
 
-var DefaultConfig = AgentInput{
-	Type: InputTypeVanillaK8s,
-	Streams: []Stream{{
-		Period: 4 * time.Hour,
-	}},
+var DefaultConfig = Stream{
+	Period: 4 * time.Hour,
 }
 
 func New(cfg *config.C) (Config, error) {
+	// work with v1 cloudbeat.yml in dev mod
+	if cfg.HasField("streams") {
+		return newStandaloneConfig(cfg)
+	}
 	c := DefaultConfig
-
 	if err := cfg.Unpack(&c); err != nil {
 		return Config{}, err
 	}
-
+	inputType := InputTypeVanillaK8s
+	if c.RuntimeCfg != nil && c.RuntimeCfg.ActivatedRules != nil && len(c.RuntimeCfg.ActivatedRules.CisEks) > 0 {
+		inputType = InputTypeEks
+	}
 	return Config{
-		Stream: c.Streams[0],
-		Type:   c.Type,
+		Stream: c,
+		Type:   inputType,
 	}, nil
 }
 
@@ -105,6 +103,22 @@ func Datastream(namespace string, indexPrefix string) string {
 	return indexPrefix + "-" + namespace
 }
 
+// stanalone config is used for development flows
+// see an example deploy/kustomize/overlays/cloudbeat-vanilla/cloudbeat.yml
+func newStandaloneConfig(cfg *config.C) (Config, error) {
+	c := struct {
+		Period  time.Duration
+		Streams []Stream
+	}{4 * time.Hour, []Stream{}}
+	if err := cfg.Unpack(&c); err != nil {
+		return Config{}, err
+	}
+	return Config{
+		Type:   InputTypeVanillaK8s,
+		Stream: c.Streams[0],
+	}, nil
+}
+
 type AwsConfigProvider interface {
 	InitializeAWSConfig(ctx context.Context, cfg aws.ConfigAWS) (awssdk.Config, error)
 }