Introduce Beat CRD

pkg/controller/beat/metricbeat/config.go

@@ -14,7 +14,7 @@ var (
   autodiscover:
     providers:
     - hints:
-        default_config: null
+        default_config: {}
         enabled: "true"
       node: ${NODE_NAME}
       type: kubernetes
@@ -37,36 +37,7 @@ var (
   - module: system
     period: 1m
     metricsets:
-    - filesystem
     - fsstat
-    processors:
-    - drop_event:
-        when:
-          regexp:
-            system:
-              filesystem:
-                mount_point: ^/(sys|cgroup|proc|dev|etc|host|lib)($|/)
-  - module: kubernetes
-    period: 10s
-    host: ${NODE_NAME}
-    hosts:
-    - https://${HOSTNAME}:10250
-    bearer_token_file: /var/run/secrets/kubernetes.io/serviceaccount/token
-    ssl:
-      verification_mode: none
-    metricsets:
-    - node
-    - system
-    - pod
-    - container
-    - volume
-  - module: kubernetes
-    period: 10s
-    host: ${NODE_NAME}
-    hosts:
-    - https://${HOSTNAME}:10249
-    metricsets:
-    - proxy
 processors:
 - add_cloud_metadata: {}
 `))

test/e2e/beat/config_test.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"testing"
 
+	v1 "github.com/elastic/cloud-on-k8s/pkg/apis/elasticsearch/v1"
 	"github.com/stretchr/testify/require"
 
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
@@ -29,7 +30,10 @@ func TestFilebeatDefaultConfig(t *testing.T) {
 
 	fbBuilder := beat.NewBuilder(name, filebeat.Type).
 		WithElasticsearchRef(esBuilder.Ref()).
-		WithESValidations(beat.HasEventFromPod(testPodBuilder.Pod.Name))
+		WithESValidations(
+			beat.HasEventFromBeat(filebeat.Type),
+			beat.HasEventFromPod(testPodBuilder.Pod.Name),
+			beat.HasMessageContaining(testPodBuilder.Logged))
 
 	test.Sequence(nil, test.EmptySteps, esBuilder, fbBuilder, testPodBuilder).RunSequential(t)
 }
@@ -44,7 +48,16 @@ func TestMetricbeatDefaultConfig(t *testing.T) {
 
 	mbBuilder := beat.NewBuilder(name, metricbeat.Type).
 		WithElasticsearchRef(esBuilder.Ref()).
-		WithESValidations(beat.HasEventFromBeat(metricbeat.Type))
+		WithESValidations(
+			beat.HasEventFromBeat(metricbeat.Type),
+			beat.HasEvent("event.dataset:system.cpu"),
+			beat.HasEvent("event.dataset:system.load"),
+			beat.HasEvent("event.dataset:system.memory"),
+			beat.HasEvent("event.dataset:system.network"),
+			beat.HasEvent("event.dataset:system.process"),
+			beat.HasEvent("event.dataset:system.process.summary"),
+			beat.HasEvent("event.dataset:system.fsstat"),
+		)
 
 	test.Sequence(nil, test.EmptySteps, esBuilder, mbBuilder, testPodBuilder).RunSequential(t)
 }
@@ -58,14 +71,16 @@ func TestHeartbeatConfig(t *testing.T) {
 	hbBuilder := beat.NewBuilder(name, "heartbeat").
 		WithElasticsearchRef(esBuilder.Ref()).
 		WithImage("docker.elastic.co/beats/heartbeat:7.7.0").
-		WithESValidations(beat.HasEventFromBeat("heartbeat"))
+		WithESValidations(
+			beat.HasEventFromBeat("heartbeat"),
+			beat.HasEvent("monitor.status:up"))
 
 	yaml := fmt.Sprintf(`
 heartbeat.monitors:
 - type: tcp
   schedule: '@every 5s'
   hosts: ["%s.%s.svc:9200"]
-`, esBuilder.Elasticsearch.Name, esBuilder.Elasticsearch.Namespace)
+`, v1.HTTPService(esBuilder.Elasticsearch.Name), esBuilder.Elasticsearch.Namespace)
 	hbBuilder = applyConfigYaml(t, hbBuilder, yaml)
 
 	test.Sequence(nil, test.EmptySteps, esBuilder, hbBuilder).RunSequential(t)

test/e2e/test/beat/builder.go

@@ -8,6 +8,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/rand"
 
 	beatv1beta1 "github.com/elastic/cloud-on-k8s/pkg/apis/beat/v1beta1"
 	commonv1 "github.com/elastic/cloud-on-k8s/pkg/apis/common/v1"
@@ -23,7 +24,15 @@ type Builder struct {
 	Validations []ValidationFunc
 }
 
+func NewBuilderWithoutSuffix(name string, typ beatcommon.Type) Builder {
+	return newBuilder(name, typ, "")
+}
+
 func NewBuilder(name string, typ beatcommon.Type) Builder {
+	return newBuilder(name, typ, rand.String(4))
+}
+
+func newBuilder(name string, typ beatcommon.Type, suffix string) Builder {
 	meta := metav1.ObjectMeta{
 		Name:      name,
 		Namespace: test.Ctx().ManagedNamespace(0),
@@ -38,7 +47,9 @@ func NewBuilder(name string, typ beatcommon.Type) Builder {
 				Version: test.Ctx().ElasticStackVersion,
 			},
 		},
-	}
+	}.
+		WithSuffix(suffix).
+		WithLabel(run.TestNameLabel, name)
 }
 
 type ValidationFunc func(client.Client) error

test/e2e/test/beat/checks.go

@@ -16,16 +16,24 @@ import (
 )
 
 func HasEventFromBeat(name beatcommon.Type) ValidationFunc {
-	return HasEvent(fmt.Sprintf("/*beat*/_search?q=agent.type:%s", name))
+	return HasEvent(fmt.Sprintf("agent.type:%s", name))
 }
 
 func HasEventFromPod(name string) ValidationFunc {
-	return HasEvent(fmt.Sprintf("/*beat*/_search?q=kubernetes.pod.name:%s", name))
+	return HasEvent(fmt.Sprintf("kubernetes.pod.name:%s", name))
+}
+
+func HasMessageContaining(message string) ValidationFunc {
+	return HasEvent(fmt.Sprintf("message:%s", message))
 }
 
 func HasEvent(query string) ValidationFunc {
+	return hasEvent(fmt.Sprintf("/*beat*/_search?q=%s", query))
+}
+
+func hasEvent(url string) ValidationFunc {
 	return func(esClient client.Client) error {
-		req, err := http.NewRequest(http.MethodGet, query, nil)
+		req, err := http.NewRequest(http.MethodGet, url, nil)
 		if err != nil {
 			return err
 		}
@@ -45,7 +53,7 @@ func HasEvent(query string) ValidationFunc {
 			return err
 		}
 		if len(results.Hits.Hits) == 0 {
-			return fmt.Errorf("hit count should be more than 0 for %s", query)
+			return fmt.Errorf("hit count should be more than 0 for %s", url)
 		}
 
 		return nil

test/e2e/test/beat/pod_builder.go

@@ -16,6 +16,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/rand"
 
 	"github.com/elastic/cloud-on-k8s/pkg/utils/k8s"
 	"github.com/elastic/cloud-on-k8s/test/e2e/cmd/run"
@@ -24,16 +25,24 @@ import (
 
 // Builder to create Beats
 type PodBuilder struct {
-	Pod corev1.Pod
+	Pod    corev1.Pod
+	Logged string
 }
 
 func NewPodBuilder(name string) PodBuilder {
+	return newPodBuilder(name, rand.String(4))
+}
+
+func newPodBuilder(name, suffix string) PodBuilder {
 	meta := metav1.ObjectMeta{
 		Name:      name,
 		Namespace: test.Ctx().ManagedNamespace(0),
 		Labels:    map[string]string{run.TestNameLabel: name},
 	}
 
+	// inject random string into the logs to allow validating whether they end up in ES easily
+	loggedString := fmt.Sprintf("_%s_", rand.String(6))
+
 	return PodBuilder{
 		Pod: corev1.Pod{
 			ObjectMeta: meta,
@@ -45,14 +54,33 @@ func NewPodBuilder(name string) PodBuilder {
 						Command: []string{
 							"bash",
 							"-c",
-							"while [ true ]; do echo \"$(date)\"; sleep 5; done",
+							fmt.Sprintf("while [ true ]; do echo \"$(date) - %s\"; sleep 5; done", loggedString),
 						},
 					},
 				},
 				SecurityContext: test.DefaultSecurityContext(),
 			},
 		},
+		Logged: loggedString,
+	}.
+		WithSuffix(suffix).
+		WithLabel(run.TestNameLabel, name)
+}
+
+func (pb PodBuilder) WithSuffix(suffix string) PodBuilder {
+	if suffix != "" {
+		pb.Pod.ObjectMeta.Name = pb.Pod.ObjectMeta.Name + "-" + suffix
 	}
+	return pb
+}
+
+func (pb PodBuilder) WithLabel(key, value string) PodBuilder {
+	if pb.Pod.Labels == nil {
+		pb.Pod.Labels = make(map[string]string)
+	}
+	pb.Pod.Labels[key] = value
+
+	return pb
 }
 
 func (pb PodBuilder) RuntimeObjects() []runtime.Object {

test/e2e/test/helper/yaml.go

@@ -75,7 +75,7 @@ func (yd *YAMLDecoder) ToBuilders(reader *bufio.Reader, transform BuilderTransfo
 			b.ApmServer = *decodedObj
 			builder = transform(b)
 		case *beatv1beta1.Beat:
-			b := beat.NewBuilder(decodedObj.Name, beatcommon.Type(decodedObj.Spec.Type))
+			b := beat.NewBuilderWithoutSuffix(decodedObj.Name, beatcommon.Type(decodedObj.Spec.Type))
 			b.Beat = *decodedObj
 			builder = transform(b)
 		default: