refact pkg/acquisition: split victorialogs.go (#4037)

Author: mmetc

pkg/acquisition/modules/victorialogs/config.go

@@ -0,0 +1,215 @@
+package victorialogs
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	yaml "github.com/goccy/go-yaml"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/victorialogs/internal/vlclient"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+)
+
+const (
+	defaultLimit int = 100
+)
+
+type AuthConfiguration struct {
+	Username string `yaml:"username"`
+	Password string `yaml:"password"`
+}
+
+type Configuration struct {
+	URL                               string              `yaml:"url"`    // VictoriaLogs url
+	Prefix                            string              `yaml:"prefix"` // VictoriaLogs prefix
+	Query                             string              `yaml:"query"`  // LogsQL query
+	Limit                             int                 `yaml:"limit"`  // Limit of logs to read
+	Since                             time.Duration       `yaml:"since"`
+	Headers                           map[string]string   `yaml:"headers"`        // HTTP headers for talking to VictoriaLogs
+	WaitForReady                      time.Duration       `yaml:"wait_for_ready"` // Retry interval, default is 10 seconds
+	Auth                              AuthConfiguration `yaml:"auth"`
+	MaxFailureDuration                time.Duration       `yaml:"max_failure_duration"` // Max duration of failure before stopping the source
+	configuration.DataSourceCommonCfg `yaml:",inline"`
+}
+
+func (s *Source) UnmarshalConfig(yamlConfig []byte) error {
+	err := yaml.UnmarshalWithOptions(yamlConfig, &s.Config, yaml.Strict())
+	if err != nil {
+		return fmt.Errorf("cannot parse VictoriaLogs acquisition configuration: %s", yaml.FormatError(err, false, false))
+	}
+
+	if s.Config.URL == "" {
+		return errors.New("VictoriaLogs url is mandatory")
+	}
+
+	if s.Config.Query == "" {
+		return errors.New("VictoriaLogs query is mandatory")
+	}
+
+	if s.Config.WaitForReady == 0 {
+		s.Config.WaitForReady = 10 * time.Second
+	}
+
+	if s.Config.Mode == "" {
+		s.Config.Mode = configuration.TAIL_MODE
+	}
+
+	if s.Config.Prefix == "" {
+		s.Config.Prefix = "/"
+	}
+
+	if !strings.HasSuffix(s.Config.Prefix, "/") {
+		s.Config.Prefix += "/"
+	}
+
+	if s.Config.Limit == 0 {
+		s.Config.Limit = defaultLimit
+	}
+
+	if s.Config.Mode == configuration.TAIL_MODE {
+		s.logger.Infof("Resetting since")
+		s.Config.Since = 0
+	}
+
+	if s.Config.MaxFailureDuration == 0 {
+		s.Config.MaxFailureDuration = 30 * time.Second
+	}
+
+	return nil
+}
+
+func (s *Source) Configure(_ context.Context, config []byte, logger *log.Entry, metricsLevel metrics.AcquisitionMetricsLevel) error {
+	s.Config = Configuration{}
+	s.logger = logger
+	s.metricsLevel = metricsLevel
+
+	err := s.UnmarshalConfig(config)
+	if err != nil {
+		return err
+	}
+
+	s.logger.Infof("Since value: %s", s.Config.Since.String())
+
+	clientConfig := vlclient.Config{
+		URL:             s.Config.URL,
+		Headers:         s.Config.Headers,
+		Limit:           s.Config.Limit,
+		Query:           s.Config.Query,
+		Since:           s.Config.Since,
+		Username:        s.Config.Auth.Username,
+		Password:        s.Config.Auth.Password,
+		FailMaxDuration: s.Config.MaxFailureDuration,
+	}
+
+	s.Client = vlclient.NewVLClient(clientConfig)
+	s.Client.Logger = logger.WithFields(log.Fields{"component": "victorialogs-client", "source": s.Config.URL})
+
+	return nil
+}
+
+func (s *Source) ConfigureByDSN(_ context.Context, dsn string, labels map[string]string, logger *log.Entry, uuid string) error {
+	s.logger = logger
+	s.Config = Configuration{}
+	s.Config.Mode = configuration.CAT_MODE
+	s.Config.Labels = labels
+	s.Config.UniqueId = uuid
+
+	u, err := url.Parse(dsn)
+	if err != nil {
+		return fmt.Errorf("while parsing dsn '%s': %w", dsn, err)
+	}
+
+	if u.Scheme != "victorialogs" {
+		return fmt.Errorf("invalid DSN %s for VictoriaLogs source, must start with victorialogs://", dsn)
+	}
+
+	if u.Host == "" {
+		return errors.New("empty host")
+	}
+
+	scheme := "http"
+
+	params := u.Query()
+
+	if q := params.Get("ssl"); q != "" {
+		scheme = "https"
+	}
+
+	if q := params.Get("query"); q != "" {
+		s.Config.Query = q
+	}
+
+	if w := params.Get("wait_for_ready"); w != "" {
+		s.Config.WaitForReady, err = time.ParseDuration(w)
+		if err != nil {
+			return err
+		}
+	} else {
+		s.Config.WaitForReady = 10 * time.Second
+	}
+
+	if since := params.Get("since"); since != "" {
+		s.Config.Since, err = time.ParseDuration(since)
+		if err != nil {
+			return fmt.Errorf("invalid since in dsn: %w", err)
+		}
+	}
+
+	if maxFailureDuration := params.Get("max_failure_duration"); maxFailureDuration != "" {
+		duration, err := time.ParseDuration(maxFailureDuration)
+		if err != nil {
+			return fmt.Errorf("invalid max_failure_duration in dsn: %w", err)
+		}
+
+		s.Config.MaxFailureDuration = duration
+	} else {
+		s.Config.MaxFailureDuration = 5 * time.Second // for OneShot mode it doesn't make sense to have longer duration
+	}
+
+	if limit := params.Get("limit"); limit != "" {
+		limit, err := strconv.Atoi(limit)
+		if err != nil {
+			return fmt.Errorf("invalid limit in dsn: %w", err)
+		}
+
+		s.Config.Limit = limit
+	}
+
+	if logLevel := params.Get("log_level"); logLevel != "" {
+		level, err := log.ParseLevel(logLevel)
+		if err != nil {
+			return fmt.Errorf("invalid log_level in dsn: %w", err)
+		}
+
+		s.Config.LogLevel = level
+		s.logger.Logger.SetLevel(level)
+	}
+
+	s.Config.URL = fmt.Sprintf("%s://%s", scheme, u.Host)
+	if u.User != nil {
+		s.Config.Auth.Username = u.User.Username()
+		s.Config.Auth.Password, _ = u.User.Password()
+	}
+
+	clientConfig := vlclient.Config{
+		URL:      s.Config.URL,
+		Headers:  s.Config.Headers,
+		Limit:    s.Config.Limit,
+		Query:    s.Config.Query,
+		Since:    s.Config.Since,
+		Username: s.Config.Auth.Username,
+		Password: s.Config.Auth.Password,
+	}
+
+	s.Client = vlclient.NewVLClient(clientConfig)
+	s.Client.Logger = logger.WithFields(log.Fields{"component": "victorialogs-client", "source": s.Config.URL})
+
+	return nil
+}

pkg/acquisition/modules/victorialogs/metrics.go

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

pkg/acquisition/modules/victorialogs/run.go

@@ -0,0 +1,147 @@
+package victorialogs
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"gopkg.in/tomb.v2"
+
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/configuration"
+	"github.com/crowdsecurity/crowdsec/pkg/acquisition/modules/victorialogs/internal/vlclient"
+	"github.com/crowdsecurity/crowdsec/pkg/metrics"
+	"github.com/crowdsecurity/crowdsec/pkg/pipeline"
+)
+
+// OneShotAcquisition reads a set of file and returns when done
+func (s *Source) OneShotAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
+	s.logger.Debug("VictoriaLogs one shot acquisition")
+	s.Client.SetTomb(t)
+
+	readyCtx, cancel := context.WithTimeout(ctx, s.Config.WaitForReady)
+	defer cancel()
+
+	err := s.Client.Ready(readyCtx)
+	if err != nil {
+		return fmt.Errorf("VictoriaLogs is not ready: %w", err)
+	}
+
+	ctx, cancel = context.WithCancel(ctx)
+	defer cancel()
+
+	respChan, err := s.getResponseChan(ctx, false)
+	if err != nil {
+		return fmt.Errorf("error when starting acquisition: %w", err)
+	}
+
+	for {
+		select {
+		case <-t.Dying():
+			s.logger.Debug("VictoriaLogs one shot acquisition stopped")
+			return nil
+		case resp, ok := <-respChan:
+			if !ok {
+				s.logger.Info("VictoriaLogs acquisition completed")
+				return nil
+			}
+
+			s.readOneEntry(resp, s.Config.Labels, out)
+		}
+	}
+}
+
+func (s *Source) readOneEntry(entry *vlclient.Log, labels map[string]string, out chan pipeline.Event) {
+	ll := pipeline.Line{}
+	ll.Raw = entry.Message
+	ll.Time = entry.Time
+	ll.Src = s.Config.URL
+	ll.Labels = labels
+	ll.Process = true
+	ll.Module = s.GetName()
+
+	if s.metricsLevel != metrics.AcquisitionMetricsLevelNone {
+		metrics.VictorialogsDataSourceLinesRead.With(prometheus.Labels{"source": s.Config.URL, "datasource_type": "victorialogs", "acquis_type": s.Config.Labels["type"]}).Inc()
+	}
+
+	expectMode := pipeline.LIVE
+	if s.Config.UseTimeMachine {
+		expectMode = pipeline.TIMEMACHINE
+	}
+
+	out <- pipeline.Event{
+		Line:       ll,
+		Process:    true,
+		Type:       pipeline.LOG,
+		ExpectMode: expectMode,
+	}
+}
+
+func (s *Source) StreamingAcquisition(ctx context.Context, out chan pipeline.Event, t *tomb.Tomb) error {
+	s.Client.SetTomb(t)
+
+	readyCtx, cancel := context.WithTimeout(ctx, s.Config.WaitForReady)
+	defer cancel()
+
+	err := s.Client.Ready(readyCtx)
+	if err != nil {
+		return fmt.Errorf("VictoriaLogs is not ready: %w", err)
+	}
+
+	lctx, clientCancel := context.WithCancel(ctx)
+	// Don't defer clientCancel(), the client outlives this function call
+
+	t.Go(func() error {
+		<-t.Dying()
+		clientCancel()
+
+		return nil
+	})
+
+	t.Go(func() error {
+		respChan, err := s.getResponseChan(lctx, true)
+		if err != nil {
+			clientCancel()
+			s.logger.Errorf("could not start VictoriaLogs tail: %s", err)
+
+			return fmt.Errorf("while starting VictoriaLogs tail: %w", err)
+		}
+
+		for {
+			select {
+			case resp, ok := <-respChan:
+				if !ok {
+					s.logger.Warnf("VictoriaLogs channel closed")
+					clientCancel()
+
+					return err
+				}
+
+				s.readOneEntry(resp, s.Config.Labels, out)
+			case <-t.Dying():
+				clientCancel()
+				return nil
+			}
+		}
+	})
+
+	return nil
+}
+
+func (s *Source) getResponseChan(ctx context.Context, infinite bool) (chan *vlclient.Log, error) {
+	var (
+		respChan chan *vlclient.Log
+		err      error
+	)
+
+	if s.Config.Mode == configuration.TAIL_MODE {
+		respChan, err = s.Client.Tail(ctx)
+		if err != nil {
+			s.logger.Errorf("could not start VictoriaLogs tail: %s", err)
+			return respChan, fmt.Errorf("while starting VictoriaLogs tail: %w", err)
+		}
+	} else {
+		respChan = s.Client.QueryRange(ctx, infinite)
+	}
+
+	return respChan, err
+}