Allow ECK deployed Elastic Filebeats to output to a Logstash Ref with metricbeat monitoring

[dadez73]

Hi,
we are trying to configure filebeat output to a ECK deployed logstash.

If we set elasticsearchRef with a logstash output is not a valid config and throw the following error.
Exiting: error unpacking config data: more than one namespace configured accessing 'output' (source:'/etc/beat.yml')

If we set only the monitoring.metrics.elasticsearchRef and logstash output like:

spec:
  type: filebeat
  monitoring:
    metrics:
      elasticsearchRefs:
        - name: elastic-monitoring
  config:
    filebeat:
      autodiscover:
        providers:
          - node: ${NODE_NAME}
            type: kubernetes
            stream: stdout
            hints:
              enabled: true
              default_config:
                type: container
                paths:
                - /var/log/containers/*${data.kubernetes.container.id}.log
      shutdown_timeout: 10s
    output.logstash:
      loadbalance: true
      hosts: [ "logstashv2-eck-logstash-ls-beatsinput.elk-system:10004" ]
      worker: 2
      pipelining: 2
      timeout: "60s"
      bulk_max_size: 700

Operator error logs logs:

{"log.level":"error","@timestamp":"2024-11-12T14:57:32.950Z","log.logger":"manager.eck-operator","message":"Reconciler error","service.version":"2.14.0+e4e2391c","service.type":"eck","ecs.version":"1.4.0","controller":"beat-controller","object":{"name":"filebeatv2-eck-beats","namespace":"elk-system"},"namespace":"elk-system","name":"filebeatv2-eck-beats","reconcileID":"89344895-c1aa-4d04-ac56-dd7037ee6873","error":"Elasticsearch.elasticsearch.k8s.elastic.co "" not found","errorCauses":[{"error":"Elasticsearch.elasticsearch.k8s.elastic.co "" not found"}],"error.stack_trace":"sigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).reconcileHandler\n\t/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.18.4/pkg/internal/controller/controller.go:324\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).processNextWorkItem\n\t/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.18.4/pkg/internal/controller/controller.go:261\nsigs.k8s.io/controller-runtime/pkg/internal/controller.(*Controller).Start.func2.2\n\t/go/pkg/mod/sigs.k8s.io/controller-runtime@v0.18.4/pkg/internal/controller/controller.go:222"}

[pebrc]

I cannot reproduce your issue. I used the following manifest (a combination of the Filebeat sample and the Logstash + service sample from the ECK GitHub repo):

apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: elasticsearch-sample
spec:
  version: 8.16.0
  nodeSets:
    - name: default
      count: 3
      config:
        node.store.allow_mmap: false
---
apiVersion: elasticsearch.k8s.elastic.co/v1
kind: Elasticsearch
metadata:
  name: elastic-monitoring
spec:
  version: 8.16.0
  nodeSets:
  - name: default
    count: 3
    config:
      node.store.allow_mmap: false
---
apiVersion: logstash.k8s.elastic.co/v1alpha1
kind: Logstash
metadata:
  name: logstash-sample
spec:
  count: 2
  version: 8.16.0
  config:
    log.level: info
    api.http.host: "0.0.0.0"
    api.http.port: 9601
    queue.type: memory
  pipelines:
    - pipeline.id: main
      pipeline.workers: 2
      config.string: "input { beats { port => 5044 }} output { stdout {}}"
  services:
    - name: api
      service:
        spec:
          type: ClusterIP
          ports:
            - port: 9601
              name: "api"
              protocol: TCP
              targetPort: 9601
    - name: beats
      service:
        spec:
          type: ClusterIP
          ports:
            - port: 5044
              name: "filebeat"
              protocol: TCP
              targetPort: 5044
            - port: 5045
              name: "winlogbeat"
              protocol: TCP
              targetPort: 5045
---
apiVersion: beat.k8s.elastic.co/v1beta1
kind: Beat
metadata:
  name: filebeat
spec:
  type: filebeat
  monitoring:
    metrics:
      elasticsearchRefs:
        - name: elastic-monitoring
  version: 8.16.0
  config:
    filebeat:
      autodiscover:
        providers:
        - type: kubernetes
          node: ${NODE_NAME}
          hints:
            enabled: true
            default_config:
              type: container
              paths:
              - /var/log/containers/*${data.kubernetes.container.id}.log
    output.logstash:
      loadbalance: true
      hosts: [ "logstash-sample-ls-beats.default:5044" ]
      worker: 2
      pipelining: 2
      timeout: "60s"
      bulk_max_size: 700
    processors:
    - add_cloud_metadata: {}
    - add_host_metadata: {}
  daemonSet:
    podTemplate:
      spec:
        serviceAccountName: filebeat
        automountServiceAccountToken: true
        terminationGracePeriodSeconds: 30
        dnsPolicy: ClusterFirstWithHostNet
        hostNetwork: true # Allows to provide richer host metadata
        containers:
        - name: filebeat
          securityContext:
            runAsUser: 0
            # If using Red Hat OpenShift uncomment this:
            #privileged: true
          volumeMounts:
          - name: varlogcontainers
            mountPath: /var/log/containers
          - name: varlogpods
            mountPath: /var/log/pods
          - name: varlibdockercontainers
            mountPath: /var/lib/docker/containers
          env:
            - name: NODE_NAME
              valueFrom:
                fieldRef:
                  fieldPath: spec.nodeName
        volumes:
        - name: varlogcontainers
          hostPath:
            path: /var/log/containers
        - name: varlogpods
          hostPath:
            path: /var/log/pods
        - name: varlibdockercontainers
          hostPath:
            path: /var/lib/docker/containers
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: filebeat
rules:
- apiGroups: [""] # "" indicates the core API group
  resources:
  - namespaces
  - pods
  - nodes
  verbs:
  - get
  - watch
  - list
- apiGroups: ["apps"]
  resources:
  - replicasets
  verbs:
  - get
  - list
  - watch
- apiGroups: ["batch"]
  resources:
  - jobs
  verbs:
  - get
  - list
  - watch
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: filebeat
  namespace: default
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: filebeat
subjects:
- kind: ServiceAccount
  name: filebeat
  namespace: default
roleRef:
  kind: ClusterRole
  name: filebeat
  apiGroup: rbac.authorization.k8s.io

Please use our community discussion forum at https://discuss.elastic.co/c/eck for any further questions.

[dadez73]

Hi, The filebeat monitoring is missing in your config. Try to set a monitoring cluster and filebeat will not deploy.

[pebrc]

My apologies. You are right. I missed this. The underlying problem is that when you specify a monitoring cluster ECK is trying to work out the cluster UUID for the cluster that your filebeat data is going into. By going through Logstash we don't have a direct association with an Elasticsearch cluster, but the code here has two bugs imo:

cloud-on-k8s/pkg/controller/beat/common/stackmon/stackmon.go

Lines 128 to 152 in 54ec0e3

	func associatedESUUID(ctx context.Context, client k8s.Client, beat *v1beta1.Beat) (string, error) {
	esAssociation := beat.EsAssociation()
	esRef := esAssociation.AssociationRef()
	if esRef.IsExternal() {
	remoteES, err := association.GetUnmanagedAssociationConnectionInfoFromSecret(client, esAssociation)
	if err != nil {
	return "", fmt.Errorf("while retrieving external ES connection info: %w", err)
	}
	clusterUUIDResponse := &clusterUUIDResponse{}
	if err := remoteES.Request("/", clusterUUIDResponse); err != nil {
	return "", fmt.Errorf("while retrieving remote cluster UUID %w", err)
	}
	return clusterUUIDResponse.ClusterUUID, nil
	}
	var es esv1.Elasticsearch
	if err := client.Get(ctx, esRef.NamespacedName(), &es); err != nil {
	return "", err
	}
	uuid, ok := es.Annotations[bootstrap.ClusterUUIDAnnotationName]
	if !ok {
	// returning specific error here so this operation can be retried.
	return "", ErrMonitoringClusterUUIDUnavailable
	}
	return uuid, nil
	}

1. it does not check if the association is actually defined
2. the surrounding code does not cater for the case where we do not have a cluster UUID.

cc @thbkrkr