k8s log acquisition datasource (#4221)

* first commit for kubernetes acquisition
* add functional tests

---------

Co-authored-by: mmetc <[email protected]>
Co-authored-by: marco <[email protected]>
Co-authored-by: copilot-swe-agent[bot] <[email protected]>
Co-authored-by: sabban <[email protected]>

Makefile

@@ -171,6 +171,7 @@ COMPONENTS := \
 	datasource_kafka \
 	datasource_journalctl \
 	datasource_kinesis \
+	datasource_kubernetes \
 	datasource_loki \
 	datasource_victorialogs \
 	datasource_s3 \

go.mod

@@ -98,7 +98,10 @@ require (
 	gopkg.in/tomb.v2 v2.0.0-20161208151619-d5d1b5820637
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
+	k8s.io/api v0.36.2
+	k8s.io/apimachinery v0.36.2
 	k8s.io/apiserver v0.36.2
+	k8s.io/client-go v0.36.2
 	modernc.org/sqlite v1.54.0
 )
 
@@ -139,6 +142,7 @@ require (
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/ebitengine/purego v0.10.0 // indirect
+	github.com/emicklei/go-restful/v3 v3.13.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.1 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.13 // indirect
@@ -171,6 +175,7 @@ require (
 	github.com/goccy/go-json v0.10.6 // indirect
 	github.com/golang/glog v1.2.5 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/hashicorp/hcl/v2 v2.24.0 // indirect
 	github.com/hashicorp/yamux v0.1.2 // indirect
@@ -251,14 +256,14 @@ require (
 	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/arch v0.25.0 // indirect
+	golang.org/x/oauth2 v0.36.0 // indirect
 	golang.org/x/term v0.45.0 // indirect
 	golang.org/x/tools v0.47.0 // indirect
 	google.golang.org/appengine v1.6.8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect
+	gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 // indirect
-	k8s.io/api v0.36.2 // indirect
-	k8s.io/apimachinery v0.36.2 // indirect
 	k8s.io/klog/v2 v2.140.0 // indirect
 	k8s.io/kube-openapi v0.0.0-20260330154417-16be699c7b31 // indirect
 	k8s.io/utils v0.0.0-20260319190234-28399d86e0b5 // indirect
@@ -269,6 +274,7 @@ require (
 	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.2 // indirect
+	sigs.k8s.io/yaml v1.6.0 // indirect
 )
 
 replace golang.org/x/time => github.com/crowdsecurity/time v0.13.0-crowdsec.20250912

go.sum

@@ -157,6 +157,8 @@ github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkp
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
 github.com/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU=
 github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ=
+github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes=
+github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanw/esbuild v0.28.1 h1:ds+yuRyUaZGx++GR56CrCeuXh8PVhVM4xq8v7PNELFc=
 github.com/evanw/esbuild v0.28.1/go.mod h1:D2vIQZqV/vIf/VRHtViaUtViZmG7o+kKmlBfVQuRi48=
 github.com/expr-lang/expr v1.17.8 h1:W1loDTT+0PQf5YteHSTpju2qfUfNoBt4yw9+wOEU9VM=
@@ -277,6 +279,8 @@ github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaS
 github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
 github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
+github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
@@ -705,6 +709,8 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
 golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -795,6 +801,8 @@ gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
+gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo=
+gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M=
 gopkg.in/inconshreveable/log15.v2 v2.0.0-20180818164646-67afb5ed74ec/go.mod h1:aPpfJ7XW+gOuirDoZ8gHhLh3kZ1B08FtV2bbmy7Jv3s=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
@@ -819,6 +827,8 @@ k8s.io/apimachinery v0.36.2 h1:0PE/W/WNy1UX61NLbXY5TMbJ6UwLL6E6lAPkYrKFxbQ=
 k8s.io/apimachinery v0.36.2/go.mod h1:fvf/HOLXq9RId0rnDIbN1OEBvHXdQbLMM8nu0LcBUf4=
 k8s.io/apiserver v0.36.2 h1:6vMnkmHZPeBloNkHUhmZYq7Ylv8WIB8xjyEl+eSt26E=
 k8s.io/apiserver v0.36.2/go.mod h1:9PoQ2ikCytrZyZg11mGhLEF5m8Rgsb5FJmYJ4Wvnl1k=
+k8s.io/client-go v0.36.2 h1:bfgxmFKc9CgqsgX4xKLAAdmTQlWee7Ob/HlDOrJ5TBI=
+k8s.io/client-go v0.36.2/go.mod h1:1vgO4OAlfPnoLcb+Rze2GF5rAr14w8qjrYMoyXJzQj0=
 k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc=
 k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0=
 k8s.io/kube-openapi v0.0.0-20260330154417-16be699c7b31 h1:V+sn9a/1fEYDGwnllCmqXBk8x7obZ+hl869Q3Abumkg=

pkg/acquisition/modules/journalctl/config.go

@@ -18,7 +18,7 @@ type Configuration struct {
 	configuration.DataSourceCommonCfg `yaml:",inline"`
 
 	Filters []string `yaml:"journalctl_filter"`
-	since   string // set only by DSN
+	since   string   // set only by DSN
 }
 
 func ConfigurationFromYAML(y []byte) (Configuration, error) {
@@ -70,9 +70,7 @@ func (s *Source) Configure(_ context.Context, yamlConfig []byte, logger *log.Ent
 	}
 
 	s.setLogger(logger, 0, s.src)
-
 	s.metricsLevel = metricsLevel
-
 	return nil
 }
 

pkg/acquisition/modules/kubernetes.go

@@ -0,0 +1,5 @@
+//go:build !no_datasource_kubernetes
+
+package modules
+
+import _ "github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/kubernetes" // register the datasource

pkg/acquisition/modules/kubernetes/config.go

@@ -0,0 +1,99 @@
+package kubernetes
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	yaml "github.com/goccy/go-yaml"
+	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/labels"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+type Configuration struct {
+	configuration.DataSourceCommonCfg `yaml:",inline"`
+
+	Selector       string `yaml:"selector"`
+	Namespace      string `yaml:"namespace"`
+	KubeConfigFile string `yaml:"kube_config,omitempty"`
+	KubeContext    string `yaml:"kube_context,omitempty"`
+}
+
+func ConfigurationFromYAML(yamlConfig []byte) (Configuration, error) {
+	var cfg Configuration
+
+	if err := yaml.UnmarshalWithOptions(yamlConfig, &cfg, yaml.Strict()); err != nil {
+		return cfg, fmt.Errorf("cannot parse: %s", yaml.FormatError(err, false, false))
+	}
+
+	cfg.SetDefaults()
+
+	if err := cfg.Validate(); err != nil {
+		return cfg, err
+	}
+
+	return cfg, nil
+}
+
+func (c *Configuration) SetDefaults() {
+	if c.Namespace == "" {
+		c.Namespace = "default"
+	}
+
+	if c.Mode == "" {
+		c.Mode = configuration.TAIL_MODE
+	}
+	if c.KubeConfigFile == "" {
+		if home, err := os.UserHomeDir(); err == nil {
+			c.KubeConfigFile = filepath.Join(home, ".kube", "config")
+		}
+	}
+}
+
+func (c *Configuration) Validate() error {
+	if c.Selector == "" {
+		return errors.New("selector must be set in kubernetes acquisition")
+	}
+	if _, err := labels.Parse(c.Selector); err != nil {
+		return fmt.Errorf("invalid selector %q in kubernetes acquisition: %w", c.Selector, err)
+	}
+	if c.Mode != configuration.TAIL_MODE {
+		return fmt.Errorf("unsupported mode %q in kubernetes acquisition, only %q is supported", c.Mode, configuration.TAIL_MODE)
+	}
+	return nil
+}
+
+func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
+	cfg, err := ConfigurationFromYAML(yamlConfig)
+	if err != nil {
+		return err
+	}
+
+	if s.logger != nil {
+		s.logger.Tracef("Kubernetes configuration: %+v", cfg)
+	}
+
+	s.config = cfg
+	s.client = nil
+	s.cancels = nil
+
+	return nil
+}
+
+func (s *Source) Configure(_ context.Context, yamlConfig []byte, logger *log.Entry, metricsLevel metrics.AcquisitionMetricsLevel) error {
+	s.logger = logger
+
+	err := s.UnmarshalConfig(yamlConfig)
+	if err != nil {
+		return err
+	}
+
+	s.metricsLevel = metricsLevel
+
+	return nil
+}

pkg/acquisition/modules/kubernetes/config_test.go

@@ -0,0 +1,169 @@
+package kubernetes
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/crowdsecurity/go-cs-lib/cstest"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+func TestSetDefaults(t *testing.T) {
+	t.Run("namespace defaults to 'default'", func(t *testing.T) {
+		c := Configuration{}
+		c.SetDefaults()
+		assert.Equal(t, "default", c.Namespace)
+	})
+
+	t.Run("namespace is preserved when set", func(t *testing.T) {
+		c := Configuration{DataSourceCommonCfg: configuration.DataSourceCommonCfg{}}
+		c.Namespace = "kube-system"
+		c.SetDefaults()
+		assert.Equal(t, "kube-system", c.Namespace)
+	})
+
+	t.Run("mode defaults to tail", func(t *testing.T) {
+		c := Configuration{}
+		c.SetDefaults()
+		assert.Equal(t, configuration.TAIL_MODE, c.Mode)
+	})
+
+	t.Run("mode is preserved when set", func(t *testing.T) {
+		c := Configuration{}
+		c.Mode = configuration.TAIL_MODE
+		c.SetDefaults()
+		assert.Equal(t, configuration.TAIL_MODE, c.Mode)
+	})
+
+	t.Run("kube_config defaults to ~/.kube/config when HOME is set", func(t *testing.T) {
+		home, err := os.UserHomeDir()
+		if err != nil {
+			t.Skip("no home directory available")
+		}
+		c := Configuration{}
+		c.SetDefaults()
+		assert.Equal(t, filepath.Join(home, ".kube", "config"), c.KubeConfigFile)
+	})
+
+	t.Run("kube_config is preserved when set", func(t *testing.T) {
+		c := Configuration{}
+		c.KubeConfigFile = "/custom/kube/config"
+		c.SetDefaults()
+		assert.Equal(t, "/custom/kube/config", c.KubeConfigFile)
+	})
+}
+
+func TestValidate(t *testing.T) {
+	tests := []struct {
+		name        string
+		cfg         Configuration
+		expectedErr string
+	}{
+		{
+			name:        "missing selector",
+			cfg:         Configuration{DataSourceCommonCfg: configuration.DataSourceCommonCfg{Mode: configuration.TAIL_MODE}},
+			expectedErr: "selector must be set",
+		},
+		{
+			name: "invalid selector",
+			cfg: Configuration{
+				DataSourceCommonCfg: configuration.DataSourceCommonCfg{Mode: configuration.TAIL_MODE},
+				Selector:            "app in (",
+			},
+			expectedErr: "invalid selector",
+		},
+		{
+			name: "unsupported mode",
+			cfg: Configuration{
+				DataSourceCommonCfg: configuration.DataSourceCommonCfg{Mode: configuration.CAT_MODE},
+				Selector:            "app=nginx",
+			},
+			expectedErr: "unsupported mode",
+		},
+		{
+			name: "valid config",
+			cfg: Configuration{
+				DataSourceCommonCfg: configuration.DataSourceCommonCfg{Mode: configuration.TAIL_MODE},
+				Selector:            "app=nginx",
+			},
+			expectedErr: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.cfg.Validate()
+			cstest.AssertErrorContains(t, err, tc.expectedErr)
+		})
+	}
+}
+
+func TestConfigure(t *testing.T) {
+	logger := log.WithField("type", ModuleName)
+
+	tests := []struct {
+		name        string
+		yaml        string
+		expectedErr string
+	}{
+		{
+			name:        "invalid YAML",
+			yaml:        "}{not yaml",
+			expectedErr: "cannot parse",
+		},
+		{
+			name:        "unknown field",
+			yaml:        "source: kubernetes\nselector: app=nginx\nunknown_field: true\n",
+			expectedErr: "cannot parse",
+		},
+		{
+			name:        "missing selector",
+			yaml:        "source: kubernetes\n",
+			expectedErr: "selector must be set",
+		},
+		{
+			name:        "unsupported mode",
+			yaml:        "source: kubernetes\nselector: app=nginx\nmode: cat\n",
+			expectedErr: "unsupported mode",
+		},
+		{
+			name:        "valid minimal config",
+			yaml:        "source: kubernetes\nselector: app=nginx\n",
+			expectedErr: "",
+		},
+		{
+			name:        "valid config with namespace",
+			yaml:        "source: kubernetes\nselector: app=nginx\nnamespace: production\n",
+			expectedErr: "",
+		},
+		{
+			name:        "valid config with kube_config",
+			yaml:        "source: kubernetes\nselector: app=nginx\nkube_config: /path/to/kubeconfig\n",
+			expectedErr: "",
+		},
+		{
+			name:        "valid config with kube_context",
+			yaml:        "source: kubernetes\nselector: app=nginx\nkube_context: my-context\n",
+			expectedErr: "",
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := &Source{}
+			err := s.Configure(t.Context(), []byte(tc.yaml), logger, metrics.AcquisitionMetricsLevelNone)
+			cstest.AssertErrorContains(t, err, tc.expectedErr)
+			if tc.expectedErr == "" {
+				require.NotNil(t, s.logger)
+				assert.Equal(t, configuration.TAIL_MODE, s.config.Mode)
+			}
+		})
+	}
+}

pkg/acquisition/modules/kubernetes/init.go

@@ -0,0 +1,20 @@
+package kubernetes
+
+import (
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/registry"
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/types"
+)
+
+var (
+	// verify interface compliance
+	_ types.DataSource          = (*Source)(nil)
+	_ types.RestartableStreamer = (*Source)(nil)
+	_ types.MetricsProvider     = (*Source)(nil)
+)
+
+const ModuleName = "kubernetes"
+
+//nolint:gochecknoinits
+func init() {
+	registry.RegisterFactory(ModuleName, func() types.DataSource { return &Source{} })
+}

pkg/acquisition/modules/kubernetes/metrics.go

@@ -0,0 +1,19 @@
+package kubernetes
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+func (*Source) GetMetrics() []prometheus.Collector {
+	return []prometheus.Collector{
+		metrics.KubernetesDataSourceLinesRead,
+	}
+}
+
+func (*Source) GetAggregMetrics() []prometheus.Collector {
+	return []prometheus.Collector{
+		metrics.KubernetesDataSourceLinesRead,
+	}
+}

pkg/acquisition/modules/kubernetes/run.go

@@ -0,0 +1,338 @@
+package kubernetes
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	log "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
+)
+
+// errProcessLine marks onLineFunc failures as fatal, as opposed to stream/scan
+// errors which followPodLogs retries.
+var errProcessLine = errors.New("failed to process line")
+
+func podRef(p *corev1.Pod) string {
+	if p == nil {
+		return "<nil pod>"
+	}
+	return fmt.Sprintf("%s/%s uid=%s phase=%s rv=%s node=%s",
+		p.Namespace,
+		p.Name,
+		p.UID,
+		p.Status.Phase,
+		p.ResourceVersion,
+		p.Spec.NodeName,
+	)
+}
+
+func (s *Source) initClient() error {
+	cfg, err := s.config.buildClientConfig(s.logger)
+	if err != nil {
+		return fmt.Errorf("building kubernetes client config for namespace=%q selector=%q: %w", s.config.Namespace, s.config.Selector, err)
+	}
+
+	client, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return fmt.Errorf("can't create a kubernetes client for namespace=%q selector=%q: %w", s.config.Namespace, s.config.Selector, err)
+	}
+
+	s.client = client
+
+	return nil
+}
+
+func (s *Source) initCancels() {
+	s.mu.Lock()
+	s.cancels = map[types.UID]context.CancelFunc{}
+	s.mu.Unlock()
+}
+
+func (s *Source) Stream(ctx context.Context, out chan pipeline.Event) error {
+	var wg sync.WaitGroup
+
+	s.logger.WithFields(log.Fields{
+		"namespace": s.config.Namespace,
+		"selector":  s.config.Selector,
+	}).Info("starting kubernetes acquisition")
+
+	err := s.initClient()
+	if err != nil {
+		return err
+	}
+
+	informerCtx, cancelInformer := context.WithCancel(ctx)
+	defer cancelInformer()
+
+	s.initCancels()
+	watchErrCh := make(chan error, 1)
+
+	f := informers.NewSharedInformerFactoryWithOptions(s.client, 0,
+		informers.WithNamespace(s.config.Namespace),
+		informers.WithTweakListOptions(func(o *metav1.ListOptions) {
+			// We set the LabelSelector on the ListOptions to filter pods at the
+			// API level, so we only get events for pods that match our
+			// selector. This is more efficient than getting all pod events and
+			// filtering them in our event handlers.
+			o.LabelSelector = s.config.Selector
+		}),
+	)
+	inf := f.Core().V1().Pods().Informer()
+	if err := inf.SetWatchErrorHandler(func(_ *cache.Reflector, watchErr error) {
+		fields := log.Fields{
+			"namespace": s.config.Namespace,
+			"selector":  s.config.Selector,
+			"error":     watchErr,
+		}
+		if apierrors.IsUnauthorized(watchErr) {
+			s.logger.WithFields(fields).Error("kubernetes informer received Unauthorized, forcing datasource restart")
+			select {
+			case watchErrCh <- fmt.Errorf("kubernetes informer unauthorized for namespace=%q selector=%q unique_id=%q: %w", s.config.Namespace, s.config.Selector, s.config.UniqueId, watchErr):
+			default:
+			}
+			cancelInformer()
+			return
+		}
+		s.logger.WithFields(fields).Warn("kubernetes informer watch error")
+	}); err != nil {
+		return fmt.Errorf("while setting watch error handler for namespace=%q selector=%q: %w", s.config.Namespace, s.config.Selector, err)
+	}
+
+	// We ignore the ResourceEventHandlerRegistration returned by
+	// AddEventHandler since we don't need to remove the handlers until shutdown,
+	// and we will stop the entire informer at that time.
+	s.logger.WithFields(log.Fields{
+		"namespace": s.config.Namespace,
+		"selector":  s.config.Selector,
+	}).Info("adding kubernetes event handler")
+	_, err = inf.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj any) {
+			p := obj.(*corev1.Pod)
+			s.logger.Debugf("ADD %s labels=%v", podRef(p), p.Labels)
+			s.tailPod(informerCtx, p, out, &wg)
+		},
+		UpdateFunc: func(oldObj, newObj any) {
+			oldP := oldObj.(*corev1.Pod)
+			newP := newObj.(*corev1.Pod)
+
+			if oldP.Status.Phase != newP.Status.Phase {
+				s.logger.Debugf("UPDATE phase %s -> %s", podRef(oldP), podRef(newP))
+				if newP.Status.Phase == corev1.PodRunning {
+					s.tailPod(informerCtx, newP, out, &wg)
+				}
+			} else {
+				s.logger.Tracef("UPDATE %s", podRef(newP))
+			}
+		},
+		DeleteFunc: func(obj any) {
+			pod, ok := obj.(*corev1.Pod)
+			if !ok {
+				t, _ := obj.(cache.DeletedFinalStateUnknown)
+				pod, _ = t.Obj.(*corev1.Pod)
+				s.logger.Debugf("DELETE(tombstone) %s", podRef(pod))
+			} else {
+				s.logger.Debugf("DELETE %s", podRef(pod))
+			}
+
+			if pod != nil {
+				s.stopPod(pod)
+			}
+		},
+	})
+
+	if err != nil {
+		return fmt.Errorf("while adding event handler for namespace=%q selector=%q: %w", s.config.Namespace, s.config.Selector, err)
+	}
+	f.Start(informerCtx.Done())
+	if !cache.WaitForCacheSync(informerCtx.Done(), inf.HasSynced) {
+		select {
+		case watchErr := <-watchErrCh:
+			return watchErr
+		default:
+		}
+		return fmt.Errorf("cache sync failed for namespace=%q selector=%q", s.config.Namespace, s.config.Selector)
+	}
+
+	select {
+	case <-ctx.Done():
+	case watchErr := <-watchErrCh:
+		s.stopPods()
+		wg.Wait()
+		return watchErr
+	}
+	s.stopPods()
+	wg.Wait()
+
+	return nil
+}
+
+func (s *Source) Dump() any {
+	return s
+}
+
+func (s *Source) followPodLogs(ctx context.Context, ns string, pod string, container string, out chan pipeline.Event,
+	onLineFunc func(string, string, chan pipeline.Event) error) error {
+	client := s.client
+	if client == nil {
+		return errors.New("kubernetes client is not initialized")
+	}
+
+	// TailLines: 0 means "no historical catch-up, only new lines from now on".
+	// Unlike SinceTime, this stays correct across reconnects: a fixed SinceTime
+	// would cause every retry to re-stream everything since the original call.
+	tailLines := int64(0)
+	req := client.CoreV1().Pods(ns).GetLogs(pod, &corev1.PodLogOptions{
+		Container:  container,
+		Follow:     true,
+		Timestamps: false,
+		TailLines:  &tailLines,
+	})
+	fn := func() error {
+		if err := ctx.Err(); err != nil {
+			return nil
+		}
+		stream, err := req.Stream(ctx)
+		if err != nil {
+			return err
+		}
+		defer stream.Close()
+
+		sc := bufio.NewScanner(stream)
+		for sc.Scan() {
+			if err := ctx.Err(); err != nil {
+				return nil
+			}
+			if err := onLineFunc(sc.Text(), ns+"/"+pod+"/"+container, out); err != nil {
+				return fmt.Errorf("%w: %w", errProcessLine, err)
+			}
+		}
+		if ctx.Err() != nil {
+			return nil
+		}
+		return sc.Err()
+	}
+	for {
+		err := fn()
+		if err != nil {
+			if errors.Is(err, errProcessLine) {
+				return err
+			}
+			// Stream open failure or a scanner/connection error: log and retry
+			// below instead of giving up, to survive transient API server
+			// disconnects.
+			s.logger.Warnf("error following logs for %s/%s/%s, retrying: %s", ns, pod, container, err)
+		}
+		// Clean EOF or transient error: check if the pod has terminated so we
+		// don't keep retrying against a dead pod / re-stream old logs.
+		// If the Get fails or the pod is still Running, fall through and retry.
+		if p, getErr := client.CoreV1().Pods(ns).Get(ctx, pod, metav1.GetOptions{}); getErr == nil && p.Status.Phase != corev1.PodRunning {
+			s.logger.Debugf("stopped following logs for %s/%s/%s: pod is in phase %s", ns, pod, container, p.Status.Phase)
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(time.Second):
+		}
+	}
+}
+
+func (s *Source) processLine(line string, source string, out chan pipeline.Event) error {
+	l := pipeline.Line{
+		Raw:     line,
+		Labels:  s.config.Labels,
+		Time:    time.Now().UTC(),
+		Src:     source,
+		Process: true,
+		Module:  s.GetName(),
+	}
+	if s.metricsLevel != metrics.AcquisitionMetricsLevelNone {
+		metrics.KubernetesDataSourceLinesRead.With(prometheus.Labels{"source": source, "acquis_type": l.Labels["type"], "datasource_type": ModuleName}).Inc()
+	}
+	evt := pipeline.MakeEvent(s.config.UseTimeMachine, pipeline.LOG, true)
+	evt.Line = l
+	out <- evt
+	s.logger.Tracef("got one line from %s: %s", source, line)
+	return nil
+}
+
+func (s *Source) podWorker(parentCtx context.Context,
+	pod *corev1.Pod,
+	out chan pipeline.Event,
+	wg *sync.WaitGroup) context.CancelFunc {
+	podCtx, cancel := context.WithCancel(parentCtx)
+	wg.Go(func() {
+		defer func() {
+			s.mu.Lock()
+			delete(s.cancels, pod.UID)
+			s.mu.Unlock()
+		}()
+		var cw sync.WaitGroup
+		for _, cont := range pod.Spec.Containers {
+			cw.Go(func() {
+				err := s.followPodLogs(podCtx, pod.Namespace, pod.Name, cont.Name, out, s.processLine)
+				if err != nil {
+					s.logger.Errorf("error following logs for %s/%s/%s: %s", pod.Namespace, pod.Name, cont.Name, err)
+				} else {
+					s.logger.Debugf("stopped following logs for %s/%s/%s", pod.Namespace, pod.Name, cont.Name)
+				}
+			})
+		}
+		cw.Wait()
+	})
+	return cancel
+}
+
+func (s *Source) tailPod(ctx context.Context, p *corev1.Pod, out chan pipeline.Event, wg *sync.WaitGroup) {
+	if p.Status.Phase != corev1.PodRunning {
+		s.logger.Debugf("SKIP tailPod(non-running) %s", podRef(p))
+		return
+	}
+
+	key := p.UID
+	s.mu.Lock()
+	if _, ok := s.cancels[key]; ok {
+		s.logger.Tracef("tail already running %s", podRef(p))
+		s.mu.Unlock()
+		return
+	}
+
+	s.logger.Debugf("START tail %s", podRef(p))
+	s.cancels[key] = s.podWorker(ctx, p, out, wg)
+	s.mu.Unlock()
+}
+
+func (s *Source) stopPod(p *corev1.Pod) {
+	key := p.UID
+	s.mu.Lock()
+	cancel, ok := s.cancels[key]
+	if ok {
+		delete(s.cancels, key)
+	}
+	s.mu.Unlock()
+	if ok {
+		cancel()
+	}
+}
+
+func (s *Source) stopPods() {
+	s.mu.Lock()
+	for _, c := range s.cancels {
+		c()
+	}
+	s.mu.Unlock()
+}

pkg/acquisition/modules/kubernetes/run_test.go

@@ -0,0 +1,229 @@
+package kubernetes
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	k8stesting "k8s.io/client-go/testing"
+
+	fakekube "k8s.io/client-go/kubernetes/fake"
+
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
+)
+
+// retryBackoffMidpoint sits halfway between an immediate return (expected to
+// take microseconds) and the 1s retry backoff in followPodLogs (run.go), so
+// tests can tell the two apart with margin on both sides instead of
+// asserting against the exact backoff duration.
+const retryBackoffMidpoint = 500 * time.Millisecond
+
+func testLogger() *log.Entry {
+	return log.WithField("type", ModuleName)
+}
+
+// reactToLogs makes the fake clientset answer Pods().GetLogs() requests with
+// the given body/error, without interfering with plain pod Get/List calls.
+func reactToLogs(client *fakekube.Clientset, body string, err error) {
+	client.PrependReactor("get", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		if action.GetSubresource() != "log" {
+			return false, nil, nil
+		}
+		if err != nil {
+			return true, nil, err
+		}
+		return true, &runtime.Unknown{Raw: []byte(body)}, nil
+	})
+}
+
+func newTestPod(name, ns string, phase corev1.PodPhase) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: ns,
+			UID:       types.UID("uid-" + name),
+		},
+		Status: corev1.PodStatus{Phase: phase},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{Name: "container"}},
+		},
+	}
+}
+
+func TestFollowPodLogs_NoClient(t *testing.T) {
+	s := &Source{logger: testLogger()}
+	// followPodLogs returns before ever writing to out, but keep it buffered
+	// for consistency with the other followPodLogs tests below.
+	out := make(chan pipeline.Event, 1)
+
+	err := s.followPodLogs(t.Context(), "ns", "pod", "container", out, s.processLine)
+	require.ErrorContains(t, err, "kubernetes client is not initialized")
+}
+
+func TestFollowPodLogs_ReadsLinesAndStopsWhenPodNotRunning(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodSucceeded)
+	client := fakekube.NewSimpleClientset(pod)
+	reactToLogs(client, "line1\nline2\n", nil)
+
+	s := &Source{logger: testLogger(), client: client}
+	// followPodLogs is called synchronously below with no concurrent reader
+	// on out, so it must be buffered or processLine's blocking send would
+	// deadlock the test.
+	out := make(chan pipeline.Event, 10)
+
+	err := s.followPodLogs(t.Context(), "ns", "pod", "container", out, s.processLine)
+	require.NoError(t, err)
+
+	close(out)
+
+	var lines []string
+	for evt := range out {
+		lines = append(lines, evt.Line.Raw)
+	}
+
+	assert.Equal(t, []string{"line1", "line2"}, lines)
+}
+
+func TestFollowPodLogs_FatalOnLineProcessingError(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodRunning)
+	client := fakekube.NewSimpleClientset(pod)
+	reactToLogs(client, "line1\n", nil)
+
+	s := &Source{logger: testLogger(), client: client}
+	// onLine below never actually writes to out, but keep it buffered for
+	// consistency with the other followPodLogs tests, which call it
+	// synchronously with no concurrent reader.
+	out := make(chan pipeline.Event, 10)
+
+	simpleErr := errors.New("fatal error in onLineFunc")
+	onLine := func(_ string, _ string, _ chan pipeline.Event) error {
+		return simpleErr
+	}
+
+	start := time.Now()
+	err := s.followPodLogs(t.Context(), "ns", "pod", "container", out, onLine)
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	require.ErrorIs(t, err, simpleErr)
+	assert.Less(t, elapsed, retryBackoffMidpoint, "a fatal onLineFunc error must not go through the retry backoff")
+}
+
+func TestFollowPodLogs_RetriesOnStreamError(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodRunning)
+	client := fakekube.NewSimpleClientset(pod)
+	reactToLogs(client, "", errors.New("connection refused"))
+
+	s := &Source{logger: testLogger(), client: client}
+	// the stream never succeeds here, so nothing is ever written to out, but
+	// keep it buffered for consistency with the other followPodLogs tests.
+	out := make(chan pipeline.Event, 10)
+
+	ctx, cancel := context.WithTimeout(t.Context(), 1500*time.Millisecond)
+	defer cancel()
+
+	start := time.Now()
+	err := s.followPodLogs(ctx, "ns", "pod", "container", out, s.processLine)
+	elapsed := time.Since(start)
+
+	require.NoError(t, err, "a persistent stream error must be retried, not returned")
+	assert.GreaterOrEqual(t, elapsed, retryBackoffMidpoint, "expected at least one retry backoff before the context timed out")
+}
+
+func TestFollowPodLogs_StopsOnContextCancel(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodRunning)
+	client := fakekube.NewSimpleClientset(pod)
+
+	s := &Source{logger: testLogger(), client: client}
+	// Unbuffered on purpose: followPodLogs must never write to out once ctx
+	// is already canceled. If it did, this blocking send would hang the
+	// test instead of silently passing.
+	out := make(chan pipeline.Event)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	err := s.followPodLogs(ctx, "ns", "pod", "container", out, s.processLine)
+	require.NoError(t, err)
+}
+
+func TestTailPod_StartsWorkerAndStopPodCancelsIt(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodRunning)
+	client := fakekube.NewSimpleClientset(pod)
+	reactToLogs(client, "hello\n", nil)
+
+	s := &Source{logger: testLogger(), client: client}
+	s.initCancels()
+
+	// unlike the followPodLogs tests above, tailPod runs the worker in its
+	// own goroutine and the test reads out concurrently via select below, so
+	// out can stay unbuffered: the worker's write and the test's read
+	// rendezvous directly.
+	out := make(chan pipeline.Event)
+	var wg sync.WaitGroup
+
+	s.tailPod(t.Context(), pod, out, &wg)
+
+	// tailPod must treat this as a no-op since pod.UID is already tracked in
+	// s.cancels. Not asserted directly, but if it weren't a no-op this would
+	// spawn a second worker racing the first one over the same single-line
+	// fake log stream, which the single <-out receive below can't account
+	// for.
+	s.tailPod(t.Context(), pod, out, &wg)
+
+	select {
+	case evt := <-out:
+		assert.Equal(t, "hello", evt.Line.Raw)
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for a log line from the tailed pod")
+	}
+
+	s.stopPod(pod)
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("stopPod did not stop the worker goroutine in time")
+	}
+
+	s.mu.Lock()
+	_, stillTracked := s.cancels[pod.UID]
+	s.mu.Unlock()
+	assert.False(t, stillTracked, "cancel func should be removed once the worker stops")
+}
+
+func TestTailPod_SkipsNonRunningPod(t *testing.T) {
+	pod := newTestPod("pod", "ns", corev1.PodPending)
+	client := fakekube.NewSimpleClientset(pod)
+
+	s := &Source{logger: testLogger(), client: client}
+	s.initCancels()
+
+	// the pod is non-running so tailPod must skip it and never write to out;
+	// unbuffered so a stray write would hang the test instead of passing
+	// silently.
+	out := make(chan pipeline.Event)
+	var wg sync.WaitGroup
+
+	s.tailPod(t.Context(), pod, out, &wg)
+
+	s.mu.Lock()
+	_, tracked := s.cancels[pod.UID]
+	s.mu.Unlock()
+	assert.False(t, tracked, "a non-running pod must not be tailed")
+}

pkg/acquisition/modules/kubernetes/source.go

@@ -0,0 +1,38 @@
+package kubernetes
+
+import (
+	"context"
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/kubernetes"
+
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+type Source struct {
+	metricsLevel metrics.AcquisitionMetricsLevel
+	config       Configuration
+
+	client  kubernetes.Interface
+	cancels map[types.UID]context.CancelFunc
+	mu      sync.Mutex
+	logger  *log.Entry
+}
+
+func (*Source) GetName() string {
+	return ModuleName
+}
+
+func (s *Source) GetMode() string {
+	return s.config.Mode
+}
+
+func (*Source) CanRun() error {
+	return nil
+}
+
+func (s *Source) GetUuid() string {
+	return s.config.UniqueId
+}

pkg/acquisition/modules/kubernetes/utils.go

@@ -0,0 +1,58 @@
+package kubernetes
+
+import (
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func (c *Configuration) buildClientConfig(logger *log.Entry) (*rest.Config, error) {
+	cfg, err := rest.InClusterConfig()
+	if err == nil {
+		if logger != nil {
+			logger.WithFields(log.Fields{
+				"auth_mode":         "in_cluster",
+				"k8s_host":          cfg.Host,
+				"bearer_token_file": cfg.BearerTokenFile,
+				"namespace":         c.Namespace,
+				"selector":          c.Selector,
+			}).Info("using in-cluster kubernetes client configuration")
+		}
+		return cfg, nil
+	}
+
+	if logger != nil {
+		logger.WithFields(log.Fields{
+			"auth_mode": "in_cluster",
+			"error":     err,
+			"namespace": c.Namespace,
+			"selector":  c.Selector,
+		}).Warn("failed to build in-cluster kubernetes client configuration, falling back to kubeconfig")
+	}
+
+	loadingRules := &clientcmd.ClientConfigLoadingRules{ExplicitPath: c.KubeConfigFile}
+	overrides := &clientcmd.ConfigOverrides{}
+	overrides.CurrentContext = c.KubeContext
+
+	cc := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, overrides)
+	cfg, err = cc.ClientConfig()
+	if err != nil {
+		return nil, fmt.Errorf("building client config for context %q and kube config file %s: %w", c.KubeContext, c.KubeConfigFile, err)
+	}
+
+	if logger != nil {
+		logger.WithFields(log.Fields{
+			"auth_mode":         "kubeconfig",
+			"k8s_host":          cfg.Host,
+			"bearer_token_file": cfg.BearerTokenFile,
+			"kube_config":       c.KubeConfigFile,
+			"kube_context":      c.KubeContext,
+			"namespace":         c.Namespace,
+			"selector":          c.Selector,
+		}).Info("using kubeconfig-based kubernetes client configuration")
+	}
+
+	return cfg, nil
+}

pkg/acquisition/schemas/datasource.yaml

@@ -6,3 +6,4 @@ description: >
   configuration DataSourceCommonCfg.
 anyOf:
   - $ref: docker.yaml
+  - $ref: kubernetes.yaml

pkg/acquisition/schemas/kubernetes.yaml

@@ -0,0 +1,73 @@
+$schema: https://json-schema.org/draft/2020-12/schema
+title: CrowdSec Kubernetes datasource
+description: >
+  Schema for Kubernetes acquisition entries consumed by CrowdSec. Every field
+  mirrors pkg/acquisition/modules/kubernetes.Configuration and the embedded
+  configuration.DataSourceCommonCfg.
+type: object
+additionalProperties: false
+properties:
+  source:
+    type: string
+    const: kubernetes
+    description: >
+      Must be kubernetes to bind this acquisition entry to the Kubernetes
+      datasource.
+  mode:
+    type: string
+    enum: [tail]
+    default: tail
+    description: >
+      Acquisition mode (only tail streaming is supported).
+  labels:
+    type: object
+    minProperties: 1
+    description: >
+      Labels attached to emitted events (for example type: kubernetes).
+    additionalProperties:
+      type: string
+    properties:
+      type:
+        type: string
+        description: Parser/collection selector; strongly recommended.
+  log_level:
+    type: string
+    enum: [panic, fatal, error, warn, warning, info, debug, trace]
+    description: >
+      Overrides the module logger level for this datasource.
+  name:
+    type: string
+    description: Friendly identifier for the datasource entry.
+  use_time_machine:
+    type: boolean
+    default: false
+    description: >
+      Replays past events when supported by the acquisition module.
+  unique_id:
+    type: string
+    description: >
+      Stable identifier injected by cscli/crowdsec auto-run (usually not user set).
+  transform:
+    type: string
+    description: >
+      expr program applied to events before they enter the pipeline.
+  selector:
+    type: string
+    minLength: 1
+    description: >
+      Kubernetes label selector applied to pods; at least one selector is required.
+  namespace:
+    type: string
+    description: >
+      Namespace whose pods should be tailed; defaults to default when omitted.
+  kube_config:
+    type: string
+    description: >
+      Path to a kubeconfig file to use when running outside the cluster.
+  kube_context:
+    type: string
+    description: >
+      Name of the kubeconfig context to use when multiple contexts are available.
+required:
+  - source
+  - selector

pkg/cwversion/component/component.go

@@ -16,6 +16,7 @@ var Built = map[string]bool{
 	"datasource_k8s-audit":    false,
 	"datasource_kafka":        false,
 	"datasource_kinesis":      false,
+	"datasource_kubernetes":   false,
 	"datasource_loki":         false,
 	"datasource_s3":           false,
 	"datasource_syslog":       false,

pkg/metrics/acquisition_kubernetes.go

@@ -0,0 +1,21 @@
+//go:build !no_datasource_kubernetes
+
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const KubernetesDataSourceLinesReadMetricName = "cs_kubernetessource_hits_total"
+
+var KubernetesDataSourceLinesRead = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: KubernetesDataSourceLinesReadMetricName,
+		Help: "Total lines that were read.",
+	},
+	[]string{"source", "datasource_type", "acquis_type"})
+
+//nolint:gochecknoinits
+func init() {
+	RegisterAcquisitionMetric(KubernetesDataSourceLinesReadMetricName)
+}