feat: validate benchmark type

beater/cloudbeat.go

@@ -20,9 +20,10 @@ package beater
 import (
 	"context"
 	"fmt"
-	"github.com/elastic/cloudbeat/resources/providers"
 	"time"
 
+	"github.com/elastic/cloudbeat/resources/providers"
+
 	"github.com/elastic/cloudbeat/config"
 	"github.com/elastic/cloudbeat/dataprovider"
 	"github.com/elastic/cloudbeat/evaluator"

beater/validator.go

@@ -32,7 +32,7 @@ type validator struct{}
 func (v *validator) Validate(cfg *agentconfig.C) error {
 	c, err := config.New(cfg)
 	if err != nil {
-		return fmt.Errorf("could not parse reconfiguration %v, skipping with error: %v", cfg.FlattenedKeys(), err)
+		return fmt.Errorf("could not parse reconfiguration %v, skipping with error: %w", cfg.FlattenedKeys(), err)
 	}
 
 	if c.RuntimeCfg == nil {

config/benchmark.go

@@ -0,0 +1,30 @@
+// 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
+
+// https://github.com/elastic/integrations/tree/main/packages/cloud_security_posture/data_stream/findings/agent/stream
+const (
+	CIS_K8S = "cis_k8s"
+	CIS_EKS = "cis_eks"
+	CIS_AWS = "cis_aws"
+)
+
+var SupportedCIS = []string{CIS_AWS, CIS_K8S, CIS_EKS}

config/config.go

@@ -22,11 +22,14 @@ package config
 
 import (
 	"context"
-	"github.com/elastic/elastic-agent-libs/logp"
 	"os"
 	"path/filepath"
 	"time"
 
+	cb_errors "github.com/elastic/cloudbeat/errors"
+
+	"github.com/elastic/elastic-agent-libs/logp"
+
 	"github.com/elastic/beats/v7/libbeat/processors"
 	"github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
 	"github.com/elastic/elastic-agent-libs/config"
@@ -38,25 +41,21 @@ const DefaultNamespace = "default"
 
 const ResultsDatastreamIndexPrefix = "logs-cloud_security_posture.findings"
 
-const (
-	InputTypeVanillaK8s = "cloudbeat/cis_k8s"
-	InputTypeEks        = "cloudbeat/cis_eks"
-	InputTypeAws        = "cloudbeat/cis_aws"
-)
+var ErrBenchmarkNotSupported = cb_errors.NewUnhealthyError("benchmark is not supported")
 
 type Fetcher struct {
 	Name string `config:"name"` // Name of the fetcher
 }
 
 type Config struct {
-	Type       string                  `config:"type"`
 	AWSConfig  aws.ConfigAWS           `config:",inline"`
 	RuntimeCfg *RuntimeConfig          `config:"runtime_cfg"`
 	Fetchers   []*config.C             `config:"fetchers"`
 	KubeConfig string                  `config:"kube_config"`
 	Period     time.Duration           `config:"period"`
 	Processors processors.PluginConfig `config:"processors"`
 	BundlePath string                  `config:"bundle_path"`
+	Benchmark  string                  `config:"config.v1.benchmark"`
 }
 
 type RuntimeConfig struct {
@@ -79,16 +78,18 @@ func New(cfg *config.C) (*Config, error) {
 		return nil, err
 	}
 
-	if c.RuntimeCfg != nil && c.RuntimeCfg.ActivatedRules != nil && len(c.RuntimeCfg.ActivatedRules.CisEks) > 0 {
-		c.Type = InputTypeEks
+	if c.Benchmark != "" {
+		if !isSupportedBenchmark(c.Benchmark) {
+			return c, ErrBenchmarkNotSupported
+		}
 	}
 	return c, nil
 }
 
 func defaultConfig() (*Config, error) {
 	ret := &Config{
-		Period: 4 * time.Hour,
-		Type:   InputTypeVanillaK8s,
+		Period:    4 * time.Hour,
+		Benchmark: CIS_K8S,
 	}
 
 	bundle, err := getBundlePath()
@@ -120,3 +121,12 @@ func Datastream(namespace string, indexPrefix string) string {
 type AwsConfigProvider interface {
 	InitializeAWSConfig(ctx context.Context, cfg aws.ConfigAWS, log *logp.Logger) (awssdk.Config, error)
 }
+
+func isSupportedBenchmark(benchmark string) bool {
+	for _, s := range SupportedCIS {
+		if benchmark == s {
+			return true
+		}
+	}
+	return false
+}

config/config_test.go

@@ -72,7 +72,7 @@ fetchers:
     directory: b
 `,
 			&Benchmarks{CisK8s: []string{"a", "b", "c", "d", "e"}},
-			"cloudbeat/cis_k8s",
+			"cis_k8s",
 			aws.ConfigAWS{},
 			2,
 		},
@@ -86,6 +86,9 @@ runtime_cfg:
       - c
       - d
       - e
+config:
+  v1:
+    benchmark: cis_eks
 access_key_id: key
 secret_access_key: secret
 session_token: session
@@ -101,7 +104,7 @@ fetchers:
     directory: c
 `,
 			&Benchmarks{CisEks: []string{"a", "b", "c", "d", "e"}},
-			"cloudbeat/cis_eks",
+			"cis_eks",
 			aws.ConfigAWS{
 				AccessKeyID:          "key",
 				SecretAccessKey:      "secret",
@@ -122,7 +125,7 @@ fetchers:
 			c, err := New(cfg)
 			s.NoError(err)
 
-			s.Equal(test.expectedType, c.Type)
+			s.Equal(test.expectedType, c.Benchmark)
 			s.EqualValues(test.expectedActivatedRules, c.RuntimeCfg.ActivatedRules)
 			s.Equal(test.expectedAWSConfig, c.AWSConfig)
 			s.Equal(test.expectedFetchers, len(c.Fetchers))
@@ -170,6 +173,48 @@ not_runtime_cfg:
 	}
 }
 
+func (s *ConfigTestSuite) TestBenchmarkType() {
+	tests := []struct {
+		config    string
+		expected  string
+		wantError bool
+	}{
+		{
+			`
+config:
+  v1:
+    benchmark: cis_eks
+`,
+			"cis_eks",
+			false,
+		},
+		{
+			`
+config:
+  v1:
+    benchmark: cis_gcp
+`,
+			"",
+			true,
+		},
+	}
+
+	for i, test := range tests {
+		s.Run(fmt.Sprint(i), func() {
+			cfg, err := config.NewConfigFrom(test.config)
+			s.NoError(err)
+
+			c, err := New(cfg)
+			if test.wantError {
+				s.Error(err)
+				return
+			}
+			s.NoError(err)
+			s.Equal(test.expected, c.Benchmark)
+		})
+	}
+}
+
 func (s *ConfigTestSuite) TestRuntimeConfig() {
 	tests := []struct {
 		config   string
@@ -256,7 +301,7 @@ runtime_cfg:
 `,
 			[]string{"a", "b"},
 			nil,
-			"cloudbeat/cis_k8s",
+			"cis_k8s",
 		},
 		{
 			`
@@ -265,10 +310,13 @@ runtime_cfg:
     cis_eks:
       - a
       - b
+config:
+  v1:
+    benchmark: cis_eks
 `,
 			nil,
 			[]string{"a", "b"},
-			"cloudbeat/cis_eks",
+			"cis_eks",
 		},
 	}
 
@@ -280,7 +328,7 @@ runtime_cfg:
 			c, err := New(cfg)
 			s.NoError(err)
 
-			s.Equal(test.expectedType, c.Type)
+			s.Equal(test.expectedType, c.Benchmark)
 			s.Equal(test.expectedActivatedRules, c.RuntimeCfg.ActivatedRules.CisK8s)
 			s.Equal(test.expectedEksActivatedRules, c.RuntimeCfg.ActivatedRules.CisEks)
 		})

errors/errors.go

@@ -0,0 +1,37 @@
+// 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 errors
+
+// BeaterUnhealthyError error is an error that is desgined to have an information that
+// can help to end user to operate cloudbeat health issues.
+// For example, when a cloudbeat configuration is invalid, the error will include
+// more information about what is missing/expected and might have links to external sources as well
+type BeaterUnhealthyError struct {
+	msg string
+}
+
+func NewUnhealthyError(msg string) BeaterUnhealthyError {
+	return BeaterUnhealthyError{msg}
+}
+
+func (c BeaterUnhealthyError) Error() string {
+	return c.msg
+}

errors/errors_test.go

@@ -0,0 +1,38 @@
+// 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 errors
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestUnwrapError(t *testing.T) {
+	e1 := NewUnhealthyError("error_1")
+	e2 := fmt.Errorf("error 2 = %w", e1)
+	healthErr := &BeaterUnhealthyError{}
+	assert.False(t, errors.Is(e1, healthErr))
+	assert.True(t, errors.As(e2, healthErr))
+	assert.Equal(t, "error_1", healthErr.Error())
+}

launcher/launcher.go

@@ -21,11 +21,14 @@
 package launcher
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/beat"
+	"github.com/elastic/beats/v7/libbeat/management"
+	cb_errors "github.com/elastic/cloudbeat/errors"
 	"github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/go-ucfg"
@@ -239,6 +242,10 @@ func (l *launcher) reconfigureWait(timeout time.Duration) (*config.C, error) {
 				err := l.validator.Validate(update)
 				if err != nil {
 					l.log.Errorf("Config update validation failed: %v", err)
+					heatlhErr := &cb_errors.BeaterUnhealthyError{}
+					if errors.As(err, heatlhErr) {
+						l.beat.Manager.UpdateStatus(management.Degraded, heatlhErr.Error())
+					}
 					continue
 				}
 			}

resources/fetchersManager/factory.go

@@ -94,7 +94,7 @@ func (fa *factories) parseConfigFetcher(log *logp.Logger, fcfg *agentconfig.C, c
 // This function takes the configuration file provided by the integration the `cfg` file
 // and depending on the input type, extract the relevant credentials and add them to the fetcher config
 func addCredentialsToFetcherConfiguration(log *logp.Logger, cfg *config.Config, fcfg *agentconfig.C) {
-	if cfg.Type == config.InputTypeEks || cfg.Type == config.InputTypeAws {
+	if cfg.Benchmark == config.CIS_EKS || cfg.Benchmark == config.CIS_AWS {
 		err := fcfg.Merge(cfg.AWSConfig)
 		if err != nil {
 			log.Errorf("Failed to merge aws configuration to fetcher configuration: %v", err)

resources/fetchersManager/factory_aws_test.go

@@ -169,7 +169,7 @@ func (s *FactoriesTestSuite) TestRegisterFetchersWithAwsCredentials() {
 
 func createEksAgentConfig(s *FactoriesTestSuite, awsConfig aws.ConfigAWS, fetcherName string) *config.Config {
 	conf := &config.Config{
-		Type:       config.InputTypeEks,
+		Benchmark:  config.CIS_EKS,
 		AWSConfig:  awsConfig,
 		RuntimeCfg: nil,
 		Fetchers:   []*agentconfig.C{agentconfig.MustNewConfigFrom(fmt.Sprint("name: ", fetcherName))},

resources/fetchersManager/factory_test.go

@@ -172,7 +172,7 @@ func (s *FactoriesTestSuite) TestRegisterFetchers() {
 		err := numCfg.SetString("name", -1, test.key)
 		s.NoError(err, "Could not set name: %v", err)
 
-		conf := &config.Config{Type: test.integrationType}
+		conf := &config.Config{Benchmark: test.integrationType}
 		conf.Fetchers = []*agentconfig.C{numCfg}
 
 		parsedList, err := s.F.ParseConfigFetchers(s.log, conf, s.resourceCh)