Disable kubeadm for filebeat (both filbeat.input and filbeat.autodiscover)

dev-tools/kubernetes/filebeat/Dockerfile.debug

@@ -4,7 +4,7 @@ ENV PATH=/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/go/bin:/usr/local/go/bin
 
 ENV CGO_ENABLED=0
 
-RUN go install github.com/go-delve/delve/cmd/dlv@v1.21.2
+RUN go install github.com/go-delve/delve/cmd/dlv@v1.23.0
 
 COPY build/filebeat-debugger /usr/share/filebeat/filebeat-debugger
 

dev-tools/kubernetes/heartbeat/Dockerfile.debug

@@ -4,7 +4,7 @@ ENV PATH=/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/go/bin:/usr/local/go/bin
 
 ENV CGO_ENABLED=0
 
-RUN go install github.com/go-delve/delve/cmd/dlv@v1.21.2
+RUN go install github.com/go-delve/delve/cmd/dlv@v1.23.0
 
 COPY build/heartbeat-debugger /usr/share/heartbeat/heartbeat-debugger
 

dev-tools/kubernetes/metricbeat/Dockerfile.debug

@@ -4,7 +4,7 @@ ENV PATH=/usr/bin:/bin:/usr/sbin:/sbin:/usr/local/bin:/go/bin:/usr/local/go/bin
 
 ENV CGO_ENABLED=0
 
-RUN go install github.com/go-delve/delve/cmd/dlv@v1.21.2
+RUN go install github.com/go-delve/delve/cmd/dlv@v1.23.0
 
 COPY build/metricbeat-debugger /usr/share/metricbeat/metricbeat-debugger
 

filebeat/_meta/test/docs/01_playground/filebeat.yaml

@@ -20,15 +20,24 @@ data:
                 logs_path: "/var/log/containers/"
 
     # To enable hints based autodiscover, remove `filebeat.inputs` configuration and uncomment this:
-    #filebeat.autodiscover:
+    # filebeat.autodiscover:
     #  providers:
     #    - type: kubernetes
     #      node: ${NODE_NAME}
     #      hints.enabled: true
+    #      #use_kubeadm: true
     #      hints.default_config:
-    #        type: container
+    #        type: filestream
+    #        id: kubernetes-container-logs-${data.kubernetes.pod.name}-${data.kubernetes.container.id}
     #        paths:
-    #          - /var/log/containers/*${data.kubernetes.container.id}.log
+    #        - /var/log/containers/*-${data.kubernetes.container.id}.log
+    #        parsers:
+    #        - container: ~
+    #        prospector:
+    #         scanner:
+    #           fingerprint.enabled: true
+    #           symlinks: true
+    #        file_identity.fingerprint: ~
 
     processors:
       - add_cloud_metadata:
@@ -64,12 +73,12 @@ spec:
       dnsPolicy: ClusterFirstWithHostNet
       containers:
       - name: filebeat
-        image: docker.elastic.co/beats/filebeat:8.0.0-SNAPSHOT
+        image: docker.elastic.co/beats/filebeat:8.16.0-SNAPSHOT
         command: [ "sleep" ]
         args: [ "infinity" ]
         env:
         - name: ELASTICSEARCH_HOST
-          value: elasticsearch.default.svc.cluster.local
+          value: elasticsearch
         - name: ELASTICSEARCH_PORT
           value: "9200"
         - name: ELASTICSEARCH_USERNAME

libbeat/autodiscover/providers/kubernetes/config.go

@@ -37,6 +37,7 @@ var AllSupportedHints = []string{"enabled", "module", "metricsets", "hosts", "pe
 type Config struct {
 	KubeConfig        string                       `config:"kube_config"`
 	KubeClientOptions kubernetes.KubeClientOptions `config:"kube_client_options"`
+	KubeAdm           bool                         `config:"use_kubeadm"`
 
 	Namespace      string        `config:"namespace"`
 	SyncPeriod     time.Duration `config:"sync_period"`
@@ -71,6 +72,7 @@ var DefaultCleanupTimeout time.Duration = 0
 func defaultConfig() *Config {
 	return &Config{
 		SyncPeriod:          10 * time.Minute,
+		KubeAdm:             true,
 		Resource:            "pod",
 		CleanupTimeout:      DefaultCleanupTimeout,
 		Prefix:              "co.elastic",

libbeat/autodiscover/providers/kubernetes/kubernetes.go

@@ -241,11 +241,11 @@ func NewEventerManager(
 	em := &eventerManager{}
 	switch cfg.Resource {
 	case "pod":
-		em.eventer, err = NewPodEventer(uuid, c, client, publish)
+		em.eventer, err = NewPodEventer(uuid, c, cfg.KubeAdm, client, publish)
 	case "node":
-		em.eventer, err = NewNodeEventer(uuid, c, client, publish)
+		em.eventer, err = NewNodeEventer(uuid, c, cfg.KubeAdm, client, publish)
 	case "service":
-		em.eventer, err = NewServiceEventer(uuid, c, client, publish)
+		em.eventer, err = NewServiceEventer(uuid, c, cfg.KubeAdm, client, publish)
 	default:
 		return nil, fmt.Errorf("unsupported autodiscover resource %s", cfg.Resource)
 	}

libbeat/autodiscover/providers/kubernetes/node.go

@@ -49,7 +49,7 @@ type node struct {
 }
 
 // NewNodeEventer creates an eventer that can discover and process node objects
-func NewNodeEventer(uuid uuid.UUID, cfg *config.C, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
+func NewNodeEventer(uuid uuid.UUID, cfg *config.C, kubeadm bool, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
 	logger := logp.NewLogger("autodiscover.node")
 
 	config := defaultConfig()
@@ -76,6 +76,7 @@ func NewNodeEventer(uuid uuid.UUID, cfg *config.C, client k8s.Interface, publish
 	}
 
 	logger.Debugf("Initializing a new Kubernetes watcher using node: %v", config.Node)
+	logger.Errorf("Initializing a new Kubernetes watcher using kubeadm: %v", config.KubeAdm)
 
 	watcher, err := kubernetes.NewNamedWatcher("node", client, &kubernetes.Node{}, kubernetes.WatchOptions{
 		SyncTimeout:  config.SyncPeriod,
@@ -88,6 +89,7 @@ func NewNodeEventer(uuid uuid.UUID, cfg *config.C, client k8s.Interface, publish
 		return nil, fmt.Errorf("couldn't create watcher for %T due to error %w", &kubernetes.Node{}, err)
 	}
 
+	cfg.SetBool("use_kubeadm", -1, kubeadm)
 	p := &node{
 		config:  config,
 		uuid:    uuid,

libbeat/autodiscover/providers/kubernetes/pod.go

@@ -56,7 +56,7 @@ type pod struct {
 }
 
 // NewPodEventer creates an eventer that can discover and process pod objects
-func NewPodEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
+func NewPodEventer(uuid uuid.UUID, cfg *conf.C, kubeadm bool, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
 	logger := logp.NewLogger("autodiscover.pod")
 
 	var replicaSetWatcher, jobWatcher, namespaceWatcher, nodeWatcher kubernetes.Watcher
@@ -97,6 +97,9 @@ func NewPodEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publish fu
 	}
 
 	metaConf := config.AddResourceMetadata
+	// We initialise the use_kubeadm variable based on modules KubeAdm base configuration
+	metaConf.Namespace.SetBool("use_kubeadm", -1, config.KubeAdm)
+	metaConf.Node.SetBool("use_kubeadm", -1, config.KubeAdm)
 
 	if metaConf.Node.Enabled() || config.Hints.Enabled() {
 		options := kubernetes.WatchOptions{

libbeat/autodiscover/providers/kubernetes/pod_test.go

@@ -2203,7 +2203,7 @@ func TestPodEventer_Namespace_Node_Watcher(t *testing.T) {
 			err = config.Unpack(&c)
 			assert.NoError(t, err)
 
-			eventer, err := NewPodEventer(uuid, config, client, nil)
+			eventer, err := NewPodEventer(uuid, config, true, client, nil)
 			if err != nil {
 				t.Fatal(err)
 			}

libbeat/autodiscover/providers/kubernetes/service.go

@@ -48,7 +48,7 @@ type service struct {
 }
 
 // NewServiceEventer creates an eventer that can discover and process service objects
-func NewServiceEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
+func NewServiceEventer(uuid uuid.UUID, cfg *conf.C, kubeadm bool, client k8s.Interface, publish func(event []bus.Event)) (Eventer, error) {
 	logger := logp.NewLogger("autodiscover.service")
 
 	config := defaultConfig()
@@ -71,6 +71,9 @@ func NewServiceEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publis
 	var namespaceWatcher kubernetes.Watcher
 
 	metaConf := config.AddResourceMetadata
+	// We initialise the use_kubeadm variable based on modules KubeAdm base configuration
+	metaConf.Namespace.SetBool("use_kubeadm", -1, config.KubeAdm)
+	metaConf.Node.SetBool("use_kubeadm", -1, config.KubeAdm)
 
 	if metaConf.Namespace.Enabled() || config.Hints.Enabled() {
 		namespaceWatcher, err = kubernetes.NewNamedWatcher("namespace", client, &kubernetes.Namespace{}, kubernetes.WatchOptions{
@@ -81,6 +84,7 @@ func NewServiceEventer(uuid uuid.UUID, cfg *conf.C, client k8s.Interface, publis
 		if err != nil {
 			return nil, fmt.Errorf("couldn't create watcher for %T due to error %w", &kubernetes.Namespace{}, err)
 		}
+		metaConf.Namespace.SetBool("use_kubeadm", -1, kubeadm)
 		namespaceMeta = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), client)
 	}
 

libbeat/autodiscover/providers/kubernetes/service_test.go

@@ -515,7 +515,7 @@ func TestServiceEventer_NamespaceWatcher(t *testing.T) {
 		t.Run(test.name, func(t *testing.T) {
 			config := conf.MustNewConfigFrom(&test.cfg)
 
-			eventer, err := NewServiceEventer(uuid, config, client, nil)
+			eventer, err := NewServiceEventer(uuid, config, true, client, nil)
 			if err != nil {
 				t.Fatal(err)
 			}

libbeat/processors/add_kubernetes_metadata/config.go

@@ -29,6 +29,7 @@ import (
 type kubeAnnotatorConfig struct {
 	KubeConfig        string                       `config:"kube_config"`
 	KubeClientOptions kubernetes.KubeClientOptions `config:"kube_client_options"`
+	kubeAdm           bool                         `config:"use_kubeadm"`
 	Node              string                       `config:"node"`
 	Scope             string                       `config:"scope"`
 	Namespace         string                       `config:"namespace"`

metricbeat/module/kubernetes/util/kubernetes.go

@@ -464,7 +464,7 @@
 
 // createMetadataGen creates and returns the metadata generator for resources other than pod and service
 // metaGen is a struct of type Resource and implements Generate method for metadata generation for a given resource kind.
-func createMetadataGen(client k8sclient.Interface, commonConfig *conf.C, addResourceMetadata *metadata.AddResourceMetadataConfig,
+func createMetadataGen(client k8sclient.Interface, commonConfig *conf.C, addResourceMetadata *metadata.AddResourceMetadataConfig, kubeadm bool,
 	resourceName string, resourceWatchers *Watchers) (*metadata.Resource, error) {
 
 	resourceWatchers.lock.RLock()
@@ -492,7 +492,7 @@
 
 // createMetadataGenSpecific creates and returns the metadata generator for a specific resource - pod or service
 // A metaGen struct implements a MetaGen interface and is designed to utilize the necessary watchers to collect(Generate) metadata for a specific resource.
-func createMetadataGenSpecific(client k8sclient.Interface, commonConfig *conf.C, addResourceMetadata *metadata.AddResourceMetadataConfig,
+func createMetadataGenSpecific(client k8sclient.Interface, commonConfig *conf.C, addResourceMetadata *metadata.AddResourceMetadataConfig, kubeadm bool,
 	resourceName string, resourceWatchers *Watchers) (metadata.MetaGen, error) {
 
 	resourceWatchers.lock.RLock()
@@ -590,8 +590,8 @@
 		return &nilEnricher{}
 	}
 
 	var specificMetaGen metadata.MetaGen
 	var generalMetaGen *metadata.Resource
 	// We initialise the use_kubeadm variable based on modules KubeAdm base configuration
 	config.AddResourceMetadata.Namespace.SetBool("use_kubeadm", -1, commonMetaConfig.KubeAdm)
 	config.AddResourceMetadata.Node.SetBool("use_kubeadm", -1, commonMetaConfig.KubeAdm)
@@ -599,9 +599,9 @@
 	// Create the metadata generator to be used in the watcher's event handler.
 	// Both specificMetaGen and generalMetaGen implement Generate method for metadata collection.
 	if resourceName == ServiceResource || resourceName == PodResource {
-		specificMetaGen, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, resourceName, resourceWatchers)
+		specificMetaGen, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, config.KubeAdm, resourceName, resourceWatchers)
 	} else {
-		generalMetaGen, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, resourceName, resourceWatchers)
+		generalMetaGen, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, config.KubeAdm, resourceName, resourceWatchers)
 	}
 	if err != nil {
 		log.Errorf("Error trying to create the metadata generators: %s", err)
@@ -763,7 +763,7 @@
 	config.AddResourceMetadata.Namespace.SetBool("use_kubeadm", -1, commonMetaConfig.KubeAdm)
 	config.AddResourceMetadata.Node.SetBool("use_kubeadm", -1, commonMetaConfig.KubeAdm)
 
-	metaGen, err := createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, PodResource, resourceWatchers)
+	metaGen, err := createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, config.KubeAdm, PodResource, resourceWatchers)
 	if err != nil {
 		log.Errorf("Error trying to create the metadata generators: %s", err)
 		return &nilEnricher{}

metricbeat/module/kubernetes/util/kubernetes_test.go

@@ -243,7 +243,7 @@ func TestCreateMetaGen(t *testing.T) {
 	}
 	client := k8sfake.NewSimpleClientset()
 
-	_, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, DeploymentResource, resourceWatchers)
+	_, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, true, DeploymentResource, resourceWatchers)
 	// At this point, no watchers were created
 	require.Error(t, err)
 
@@ -253,7 +253,7 @@ func TestCreateMetaGen(t *testing.T) {
 	require.NoError(t, err)
 
 	// Create the generators, this time without error
-	_, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, DeploymentResource, resourceWatchers)
+	_, err = createMetadataGen(client, commonConfig, config.AddResourceMetadata, true, DeploymentResource, resourceWatchers)
 	require.NoError(t, err)
 }
 
@@ -286,19 +286,19 @@ func TestCreateMetaGenSpecific(t *testing.T) {
 	// For pod:
 	metricsetPod := "pod"
 
-	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, PodResource, resourceWatchers)
+	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, true, PodResource, resourceWatchers)
 	// At this point, no watchers were created
 	require.Error(t, err)
 
 	// Create the pod resource + the extras
 	err = createAllWatchers(client, metricsetPod, PodResource, false, config, log, resourceWatchers)
 	require.NoError(t, err)
 
-	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, PodResource, resourceWatchers)
+	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, true, PodResource, resourceWatchers)
 	require.NoError(t, err)
 
 	// For service:
-	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, ServiceResource, resourceWatchers)
+	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, true, ServiceResource, resourceWatchers)
 	// At this point, no watchers were created
 	require.Error(t, err)
 
@@ -307,7 +307,7 @@ func TestCreateMetaGenSpecific(t *testing.T) {
 	err = createAllWatchers(client, metricsetService, ServiceResource, false, config, log, resourceWatchers)
 	require.NoError(t, err)
 
-	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, ServiceResource, resourceWatchers)
+	_, err = createMetadataGenSpecific(client, commonConfig, config.AddResourceMetadata, true, ServiceResource, resourceWatchers)
 	require.NoError(t, err)
 }