feature cscli|crowdsec add additional labels on crowdsec dsn run (#1053)

* feature cscli|crowdsec add additional labels on crowdsec dsn run

cmd/crowdsec-cli/hubtest.go

@@ -52,6 +52,7 @@ func NewHubTestCmd() *cobra.Command {
 	postoverflows := []string{}
 	scenarios := []string{}
 	var ignoreParsers bool
+	var labels map[string]string
 
 	var cmdHubTestCreate = &cobra.Command{
 		Use:   "create",
@@ -119,6 +120,7 @@ cscli hubtest create my-scenario-test --parsers crowdsecurity/nginx --scenarios
 				LogFile:       logFileName,
 				LogType:       logType,
 				IgnoreParsers: ignoreParsers,
+				Labels: 	   labels,
 			}
 
 			configFilePath := filepath.Join(testPath, "config.yaml")

cmd/crowdsec/main.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"sort"
+	"strings"
 
 	_ "net/http/pprof"
 	"time"
@@ -56,12 +57,15 @@ type Flags struct {
 	InfoLevel      bool
 	PrintVersion   bool
 	SingleFileType string
+	Labels         map[string]string
 	OneShotDSN     string
 	TestMode       bool
 	DisableAgent   bool
 	DisableAPI     bool
 }
 
+type labelsMap map[string]string
+
 type parsers struct {
 	ctx             *parser.UnixParserCtx
 	povfwctx        *parser.UnixParserCtx
@@ -146,8 +150,10 @@ func LoadAcquisition(cConfig *csconfig.Config) error {
 		if flags.OneShotDSN == "" || flags.SingleFileType == "" {
 			return fmt.Errorf("-type requires a -dsn argument")
 		}
+		flags.Labels = labels
+		flags.Labels["type"] = flags.SingleFileType
 
-		dataSources, err = acquisition.LoadAcquisitionFromDSN(flags.OneShotDSN, flags.SingleFileType)
+		dataSources, err = acquisition.LoadAcquisitionFromDSN(flags.OneShotDSN, flags.Labels)
 		if err != nil {
 			return errors.Wrapf(err, "failed to configure datasource for %s", flags.OneShotDSN)
 		}
@@ -163,6 +169,20 @@ func LoadAcquisition(cConfig *csconfig.Config) error {
 
 var dumpFolder string
 var dumpStates bool
+var labels = make(labelsMap)
+
+func (l *labelsMap) String() string {
+	return "labels"
+}
+
+func (l labelsMap) Set(label string) error {
+	split := strings.Split(label, ":")
+	if len(split) != 2 {
+		return errors.Wrapf(errors.New("Bad Format"), "for Label '%s'", label)
+	}
+	l[split[0]] = split[1]
+	return nil
+}
 
 func (f *Flags) Parse() {
 
@@ -173,6 +193,7 @@ func (f *Flags) Parse() {
 	flag.BoolVar(&f.PrintVersion, "version", false, "display version")
 	flag.StringVar(&f.OneShotDSN, "dsn", "", "Process a single data source in time-machine")
 	flag.StringVar(&f.SingleFileType, "type", "", "Labels.type for file in time-machine")
+	flag.Var(&labels, "label", "Additional Labels for file in time-machine")
 	flag.BoolVar(&f.TestMode, "t", false, "only test configs")
 	flag.BoolVar(&f.DisableAgent, "no-cs", false, "disable crowdsec agent")
 	flag.BoolVar(&f.DisableAPI, "no-api", false, "disable local API")

pkg/acquisition/acquisition.go

@@ -27,7 +27,7 @@ type DataSource interface {
 	GetMetrics() []prometheus.Collector                      // Returns pointers to metrics that are managed by the module
 	GetAggregMetrics() []prometheus.Collector                // Returns pointers to metrics that are managed by the module (aggregated mode, limits cardinality)
 	Configure([]byte, *log.Entry) error                      // Configure the datasource
-	ConfigureByDSN(string, string, *log.Entry) error         // Configure the datasource
+	ConfigureByDSN(string, map[string]string, *log.Entry) error	// Configure the datasource
 	GetMode() string                                         // Get the mode (TAIL, CAT or SERVER)
 	GetName() string                                         // Get the name of the module
 	OneShotAcquisition(chan types.Event, *tomb.Tomb) error   // Start one shot acquisition(eg, cat a file)
@@ -120,7 +120,7 @@ func detectBackwardCompatAcquis(sub configuration.DataSourceCommonCfg) string {
 	return ""
 }
 
-func LoadAcquisitionFromDSN(dsn string, label string) ([]DataSource, error) {
+func LoadAcquisitionFromDSN(dsn string, labels map[string]string) ([]DataSource, error) {
 	var sources []DataSource
 
 	frags := strings.Split(dsn, ":")
@@ -139,7 +139,7 @@ func LoadAcquisitionFromDSN(dsn string, label string) ([]DataSource, error) {
 	subLogger := clog.WithFields(log.Fields{
 		"type": dsn,
 	})
-	err := dataSrc.ConfigureByDSN(dsn, label, subLogger)
+	err := dataSrc.ConfigureByDSN(dsn, labels, subLogger)
 	if err != nil {
 		return nil, errors.Wrapf(err, "while configuration datasource for %s", dsn)
 	}

pkg/acquisition/acquisition_test.go

@@ -47,7 +47,7 @@ func (f *MockSource) GetMetrics() []prometheus.Collector                      {
 func (f *MockSource) GetAggregMetrics() []prometheus.Collector                { return nil }
 func (f *MockSource) Dump() interface{}                                       { return f }
 func (f *MockSource) GetName() string                                         { return "mock" }
-func (f *MockSource) ConfigureByDSN(string, string, *log.Entry) error {
+func (f *MockSource) ConfigureByDSN(string, map[string]string, *log.Entry) error {
 	return fmt.Errorf("not supported")
 }
 
@@ -342,7 +342,7 @@ func (f *MockCat) CanRun() error                                   { return nil
 func (f *MockCat) GetMetrics() []prometheus.Collector              { return nil }
 func (f *MockCat) GetAggregMetrics() []prometheus.Collector        { return nil }
 func (f *MockCat) Dump() interface{}                               { return f }
-func (f *MockCat) ConfigureByDSN(string, string, *log.Entry) error { return fmt.Errorf("not supported") }
+func (f *MockCat) ConfigureByDSN(string, map[string]string, *log.Entry) error { return fmt.Errorf("not supported") }
 
 //----
 
@@ -381,7 +381,7 @@ func (f *MockTail) CanRun() error                            { return nil }
 func (f *MockTail) GetMetrics() []prometheus.Collector       { return nil }
 func (f *MockTail) GetAggregMetrics() []prometheus.Collector { return nil }
 func (f *MockTail) Dump() interface{}                        { return f }
-func (f *MockTail) ConfigureByDSN(string, string, *log.Entry) error {
+func (f *MockTail) ConfigureByDSN(string, map[string]string, *log.Entry) error {
 	return fmt.Errorf("not supported")
 }
 
@@ -511,7 +511,7 @@ func (f *MockSourceByDSN) GetMetrics() []prometheus.Collector
 func (f *MockSourceByDSN) GetAggregMetrics() []prometheus.Collector                { return nil }
 func (f *MockSourceByDSN) Dump() interface{}                                       { return f }
 func (f *MockSourceByDSN) GetName() string                                         { return "mockdsn" }
-func (f *MockSourceByDSN) ConfigureByDSN(dsn string, logType string, logger *log.Entry) error {
+func (f *MockSourceByDSN) ConfigureByDSN(dsn string, labels map[string]string, logger *log.Entry) error {
 	dsn = strings.TrimPrefix(dsn, "mockdsn://")
 	if dsn != "test_expect" {
 		return fmt.Errorf("unexpected value")
@@ -555,7 +555,7 @@ func TestConfigureByDSN(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		srcs, err := LoadAcquisitionFromDSN(test.dsn, "test_label")
+		srcs, err := LoadAcquisitionFromDSN(test.dsn, map[string]string{"type": "test_label"})
 		if err != nil && test.ExpectedError != "" {
 			if !strings.Contains(err.Error(), test.ExpectedError) {
 				t.Fatalf("expected '%s', got '%s'", test.ExpectedError, err.Error())

pkg/acquisition/modules/cloudwatch/cloudwatch.go

@@ -484,7 +484,7 @@ func (cw *CloudwatchSource) TailLogStream(cfg *LogStreamTailConfig, outChan chan
 	}
 }
 
-func (cw *CloudwatchSource) ConfigureByDSN(dsn string, logtype string, logger *log.Entry) error {
+func (cw *CloudwatchSource) ConfigureByDSN(dsn string, labels map[string]string, logger *log.Entry) error {
 	cw.logger = logger
 
 	dsn = strings.TrimPrefix(dsn, cw.GetName()+"://")
@@ -498,9 +498,7 @@ func (cw *CloudwatchSource) ConfigureByDSN(dsn string, logtype string, logger *l
 	}
 	cw.Config.GroupName = frags[0]
 	cw.Config.StreamName = &frags[1]
-	cw.Config.Labels = make(map[string]string)
-	cw.Config.Labels["type"] = logtype
-
+	cw.Config.Labels = labels
 	u, err := url.ParseQuery(args[1])
 	if err != nil {
 		return errors.Wrapf(err, "while parsing %s", dsn)

pkg/acquisition/modules/cloudwatch/cloudwatch_test.go

@@ -609,7 +609,8 @@ func TestConfigureByDSN(t *testing.T) {
 	var err error
 	log.SetLevel(log.DebugLevel)
 	tests := []struct {
-		dsn, logtype   string
+		dsn            string
+		labels         map[string]string
 		expectedCfgErr string
 		name           string
 	}{
@@ -640,7 +641,7 @@ func TestConfigureByDSN(t *testing.T) {
 		dbgLogger.Logger.SetLevel(log.DebugLevel)
 		log.Printf("%d/%d", idx, len(tests))
 		cw := CloudwatchSource{}
-		err = cw.ConfigureByDSN(test.dsn, test.logtype, dbgLogger)
+		err = cw.ConfigureByDSN(test.dsn, test.labels, dbgLogger)
 		if err != nil && test.expectedCfgErr != "" {
 			if !strings.Contains(err.Error(), test.expectedCfgErr) {
 				t.Fatalf("%s expected error '%s' got error '%s'", test.name, test.expectedCfgErr, err.Error())
@@ -769,7 +770,7 @@ func TestOneShotAcquisition(t *testing.T) {
 		dbgLogger.Logger.SetLevel(log.DebugLevel)
 		dbgLogger.Infof("starting test")
 		cw := CloudwatchSource{}
-		err = cw.ConfigureByDSN(test.dsn, "test", dbgLogger)
+		err = cw.ConfigureByDSN(test.dsn, map[string]string{"type": "test"}, dbgLogger)
 		if err != nil && test.expectedCfgErr != "" {
 			if !strings.Contains(err.Error(), test.expectedCfgErr) {
 				t.Fatalf("%s expected error '%s' got error '%s'", test.name, test.expectedCfgErr, err.Error())

pkg/acquisition/modules/file/file.go

@@ -117,7 +117,7 @@ func (f *FileSource) Configure(Config []byte, logger *log.Entry) error {
 	return nil
 }
 
-func (f *FileSource) ConfigureByDSN(dsn string, labelType string, logger *log.Entry) error {
+func (f *FileSource) ConfigureByDSN(dsn string, labels map[string]string, logger *log.Entry) error {
 	if !strings.HasPrefix(dsn, "file://") {
 		return fmt.Errorf("invalid DSN %s for file source, must start with file://", dsn)
 	}
@@ -153,7 +153,7 @@ func (f *FileSource) ConfigureByDSN(dsn string, labelType string, logger *log.En
 	}
 
 	f.config = FileConfiguration{}
-	f.config.Labels = map[string]string{"type": labelType}
+	f.config.Labels = labels
 	f.config.Mode = configuration.CAT_MODE
 
 	f.logger.Debugf("Will try pattern %s", args[0])

pkg/acquisition/modules/file/file_test.go

@@ -69,7 +69,7 @@ func TestConfigureDSN(t *testing.T) {
 	})
 	for _, test := range tests {
 		f := FileSource{}
-		err := f.ConfigureByDSN(test.dsn, "testtype", subLogger)
+		err := f.ConfigureByDSN(test.dsn, map[string]string{"type": "testtype"}, subLogger)
 		if test.expectedErr != "" {
 			assert.Contains(t, err.Error(), test.expectedErr)
 		} else {

pkg/acquisition/modules/journalctl/journalctl.go

@@ -182,11 +182,11 @@ func (j *JournalCtlSource) Configure(yamlConfig []byte, logger *log.Entry) error
 	return nil
 }
 
-func (j *JournalCtlSource) ConfigureByDSN(dsn string, labelType string, logger *log.Entry) error {
+func (j *JournalCtlSource) ConfigureByDSN(dsn string, labels map[string]string, logger *log.Entry) error {
 	j.logger = logger
 	j.config = JournalCtlConfiguration{}
 	j.config.Mode = configuration.CAT_MODE
-	j.config.Labels = map[string]string{"type": labelType}
+	j.config.Labels = labels
 
 	//format for the DSN is : journalctl://filters=FILTER1&filters=FILTER2
 	if !strings.HasPrefix(dsn, "journalctl://") {

pkg/acquisition/modules/journalctl/journalctl_test.go

@@ -92,7 +92,7 @@ func TestConfigureDSN(t *testing.T) {
 	})
 	for _, test := range tests {
 		f := JournalCtlSource{}
-		err := f.ConfigureByDSN(test.dsn, "testtype", subLogger)
+		err := f.ConfigureByDSN(test.dsn, map[string]string{"type": "testtype"}, subLogger)
 		if test.expectedErr != "" {
 			assert.Contains(t, err.Error(), test.expectedErr)
 		} else {

pkg/acquisition/modules/syslog/syslog.go

@@ -73,7 +73,7 @@ func (s *SyslogSource) GetAggregMetrics() []prometheus.Collector {
 	return []prometheus.Collector{linesReceived, linesParsed}
 }
 
-func (s *SyslogSource) ConfigureByDSN(dsn string, labelType string, logger *log.Entry) error {
+func (s *SyslogSource) ConfigureByDSN(dsn string, labels map[string]string, logger *log.Entry) error {
 	return fmt.Errorf("syslog datasource does not support one shot acquisition")
 }
 

pkg/cstest/hubtest_item.go

@@ -15,12 +15,13 @@ import (
 )
 
 type HubTestItemConfig struct {
-	Parsers       []string `yaml:"parsers"`
-	Scenarios     []string `yaml:"scenarios"`
-	PostOVerflows []string `yaml:"postoverflows"`
-	LogFile       string   `yaml:"log_file"`
-	LogType       string   `yaml:"log_type"`
-	IgnoreParsers bool     `yaml:"ignore_parsers"` // if we test a scenario, we don't want to assert on Parser
+	Parsers       []string          `yaml:"parsers"`
+	Scenarios     []string          `yaml:"scenarios"`
+	PostOVerflows []string          `yaml:"postoverflows"`
+	LogFile       string            `yaml:"log_file"`
+	LogType       string            `yaml:"log_type"`
+	Labels        map[string]string `yaml:"labels"`
+	IgnoreParsers bool              `yaml:"ignore_parsers"` // if we test a scenario, we don't want to assert on Parser
 }
 
 type HubIndex struct {
@@ -513,6 +514,10 @@ func (t *HubTestItem) Run() error {
 	}
 
 	cmdArgs = []string{"-c", t.RuntimeConfigFilePath, "-type", logType, "-dsn", dsn, "-dump-data", t.ResultsPath}
+	for labelKey, labelValue := range t.Config.Labels {
+		arg := fmt.Sprintf("%s:%s", labelKey, labelValue)
+		cmdArgs = append(cmdArgs, "-label", arg)
+	}
 	crowdsecCmd := exec.Command(t.CrowdSecPath, cmdArgs...)
 	log.Debugf("%s", crowdsecCmd.String())
 	output, err = crowdsecCmd.CombinedOutput()