refact pkg/acquisition: split k8s_audit.go (#4033)

Author: mmetc

pkg/acquisition/k8s.go

@@ -8,12 +8,12 @@ import (
 
 var (
 	// verify interface compliance
-	_ DataSource      = (*k8sauditacquisition.KubernetesAuditSource)(nil)
-	_ Tailer          = (*k8sauditacquisition.KubernetesAuditSource)(nil)
-	_ MetricsProvider = (*k8sauditacquisition.KubernetesAuditSource)(nil)
+	_ DataSource      = (*k8sauditacquisition.Source)(nil)
+	_ Tailer          = (*k8sauditacquisition.Source)(nil)
+	_ MetricsProvider = (*k8sauditacquisition.Source)(nil)
 )
 
 //nolint:gochecknoinits
 func init() {
-	registerDataSource("k8s-audit", func() DataSource { return &k8sauditacquisition.KubernetesAuditSource{} })
+	registerDataSource("k8s-audit", func() DataSource { return &k8sauditacquisition.Source{} })
 }

pkg/acquisition/modules/kubernetesaudit/config.go

@@ -0,0 +1,84 @@
+package kubernetesauditacquisition
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+
+	yaml "github.com/goccy/go-yaml"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+type Configuration struct {
+	ListenAddr                        string `yaml:"listen_addr"`
+	ListenPort                        int    `yaml:"listen_port"`
+	WebhookPath                       string `yaml:"webhook_path"`
+	configuration.DataSourceCommonCfg `yaml:",inline"`
+}
+
+func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
+	k8sConfig := Configuration{}
+
+	err := yaml.UnmarshalWithOptions(yamlConfig, &k8sConfig, yaml.Strict())
+	if err != nil {
+		return fmt.Errorf("cannot parse k8s-audit configuration: %s", yaml.FormatError(err, false, false))
+	}
+
+	s.config = k8sConfig
+
+	if s.config.ListenAddr == "" {
+		return errors.New("listen_addr cannot be empty")
+	}
+
+	if s.config.ListenPort == 0 {
+		return errors.New("listen_port cannot be empty")
+	}
+
+	if s.config.WebhookPath == "" {
+		return errors.New("webhook_path cannot be empty")
+	}
+
+	if s.config.WebhookPath[0] != '/' {
+		s.config.WebhookPath = "/" + s.config.WebhookPath
+	}
+
+	if s.config.Mode == "" {
+		s.config.Mode = configuration.TAIL_MODE
+	}
+
+	return nil
+}
+
+func (s *Source) Configure(_ context.Context, config []byte, logger *log.Entry, metricsLevel metrics.AcquisitionMetricsLevel) error {
+	s.logger = logger
+	s.metricsLevel = metricsLevel
+
+	err := s.UnmarshalConfig(config)
+	if err != nil {
+		return err
+	}
+
+	s.logger.Tracef("K8SAudit configuration: %+v", s.config)
+
+	s.addr = fmt.Sprintf("%s:%d", s.config.ListenAddr, s.config.ListenPort)
+
+	s.mux = http.NewServeMux()
+
+	s.server = &http.Server{
+		Addr:      s.addr,
+		Handler:   s.mux,
+		Protocols: &http.Protocols{},
+	}
+
+	s.server.Protocols.SetHTTP1(true)
+	s.server.Protocols.SetUnencryptedHTTP2(true)
+	s.server.Protocols.SetHTTP2(true)
+
+	s.mux.HandleFunc(s.config.WebhookPath, s.webhookHandler)
+
+	return nil
+}

pkg/acquisition/modules/kubernetesaudit/k8s_audit.go

@@ -48,13 +48,13 @@ source: k8s-audit
 listen_addr: 0.0.0.0
 listen_port: true
 `,
-			expectedErr: `[4:14] cannot unmarshal bool into Go struct field KubernetesAuditConfiguration.ListenPort of type int`,
+			expectedErr: `[4:14] cannot unmarshal bool into Go struct field Configuration.ListenPort of type int`,
 		},
 	}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
-			f := KubernetesAuditSource{}
+			f := Source{}
 
 			err := f.UnmarshalConfig([]byte(test.config))
 			cstest.RequireErrorContains(t, err, test.expectedErr)
@@ -86,7 +86,7 @@ webhook_path: /k8s-audit`,
 			out := make(chan pipeline.Event)
 			tb := &tomb.Tomb{}
 
-			f := KubernetesAuditSource{}
+			f := Source{}
 
 			err := f.UnmarshalConfig([]byte(test.config))
 
@@ -244,7 +244,7 @@ func TestHandler(t *testing.T) {
 				}
 			})
 
-			f := KubernetesAuditSource{}
+			f := Source{}
 
 			port := 49234+idx
 			config := fmt.Sprintf(`source: k8s-audit

pkg/acquisition/modules/kubernetesaudit/k8s_audit_test.go

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