Consume dataYaml in config, add config testing (#61)

Author: yashtewari

beater/cloudbeat.go

@@ -40,12 +40,13 @@ type cloudbeat struct {
 	ctx    context.Context
 	cancel context.CancelFunc
 
-	config      config.Config
-	client      beat.Client
-	data        *manager.Data
-	evaluator   evaluator.Evaluator
-	transformer transformer.Transformer
-	log         *logp.Logger
+	config        config.Config
+	configUpdates <-chan *common.Config
+	client        beat.Client
+	data          *manager.Data
+	evaluator     evaluator.Evaluator
+	transformer   transformer.Transformer
+	log           *logp.Logger
 }
 
 // New creates an instance of cloudbeat.
@@ -54,8 +55,8 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	c := config.DefaultConfig
-	if err := cfg.Unpack(&c); err != nil {
+	c, err := config.New(cfg)
+	if err != nil {
 		cancel()
 		return nil, fmt.Errorf("error reading config file: %w", err)
 	}
@@ -102,13 +103,14 @@ func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 	t := transformer.NewTransformer(ctx, eval, commonData, resultsIndex)
 
 	bt := &cloudbeat{
-		ctx:         ctx,
-		cancel:      cancel,
-		config:      c,
-		evaluator:   eval,
-		data:        data,
-		transformer: t,
-		log:         log,
+		ctx:           ctx,
+		cancel:        cancel,
+		config:        c,
+		configUpdates: config.Updates(ctx),
+		evaluator:     eval,
+		data:          data,
+		transformer:   t,
+		log:           log,
 	}
 	return bt, nil
 }
@@ -148,6 +150,12 @@ func (bt *cloudbeat) Run(b *beat.Beat) error {
 		select {
 		case <-bt.ctx.Done():
 			return nil
+
+		case update := <-bt.configUpdates:
+			if err := bt.config.Update(update); err != nil {
+				logp.L().Errorf("could not update cloudbeat config: %v", err)
+			}
+
 		case fetchedResources := <-output:
 			cycleId, _ := uuid.NewV4()
 			bt.log.Debugf("Cycle % has started", cycleId)

config/config.go

@@ -36,12 +36,40 @@ type Config struct {
 	Period     time.Duration           `config:"period"`
 	Processors processors.PluginConfig `config:"processors"`
 	Fetchers   []*common.Config        `config:"fetchers"`
+
+	Streams []Stream `config:"streams"`
+}
+
+type Stream struct {
+	DataYaml struct {
+		ActivatedRules struct {
+			CISK8S []string `config:"cis_k8s"`
+		} `config:"activated_rules"`
+	} `config:"data_yaml"`
 }
 
 var DefaultConfig = Config{
 	Period: 10 * time.Second,
 }
 
+func New(cfg *common.Config) (Config, error) {
+	c := DefaultConfig
+
+	if err := cfg.Unpack(&c); err != nil {
+		return c, err
+	}
+
+	return c, nil
+}
+
+func (c *Config) Update(cfg *common.Config) error {
+	if err := cfg.Unpack(&c); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // Datastream function to generate the datastream value
 func Datastream(namespace string, indexPrefix string) string {
 	if namespace == "" {

config/config_test.go

@@ -19,3 +19,52 @@
 // +build !integration
 
 package config
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/stretchr/testify/suite"
+)
+
+type ConfigTestSuite struct {
+	suite.Suite
+}
+
+func TestConfigTestSuite(t *testing.T) {
+	suite.Run(t, new(ConfigTestSuite))
+}
+
+func (s *ConfigTestSuite) SetupTest() {
+}
+
+func (s *ConfigTestSuite) TestNew() {
+	var tests = []struct {
+		config           string
+		expectedPatterns []string
+	}{
+		{
+			`
+  streams:
+    - data_yaml:
+        activated_rules:
+          cis_k8s:
+            - a
+            - b
+            - c
+            - d
+`,
+			[]string{"a", "b", "c", "d"},
+		},
+	}
+
+	for _, test := range tests {
+		cfg, err := common.NewConfigFrom(test.config)
+		s.NoError(err)
+
+		c, err := New(cfg)
+		s.NoError(err)
+
+		s.Equal(test.expectedPatterns, c.Streams[0].DataYaml.ActivatedRules.CISK8S)
+	}
+}

config/updates.go

@@ -0,0 +1,64 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Config is put into a different package to prevent cyclic imports in case
+// it is needed in several locations
+
+package config
+
+import (
+	"context"
+
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/common/reload"
+	"github.com/elastic/beats/v7/libbeat/logp"
+)
+
+type reloader struct {
+	ctx context.Context
+	ch  chan<- *common.Config
+}
+
+func (r *reloader) Reload(configs []*reload.ConfigWithMeta) error {
+	if len(configs) == 0 {
+		return nil
+	}
+
+	logp.L().Infof("Received %v new configs for reload.", len(configs))
+
+	select {
+	case <-r.ctx.Done():
+	default:
+		// TODO(yashtewari): Based on limitations elsewhere, such as the CSP integration,
+		// don't think we should receive more than one Config here. Need to confirm and handle.
+		r.ch <- configs[len(configs)-1].Config
+	}
+
+	return nil
+}
+
+func Updates(ctx context.Context) <-chan *common.Config {
+	ch := make(chan *common.Config)
+	r := &reloader{
+		ctx: ctx,
+		ch:  ch,
+	}
+
+	reload.Register.MustRegisterList("inputs", r)
+
+	return ch
+}