Support for track events

cmd/splitcli/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/splitio/go-toolkit/v5/logging"
@@ -67,7 +68,20 @@ func executeCall(c types.ClientInterface, a *conf.CliArgs) (string, error) {
 	case "treatment":
 		res, err := c.Treatment(a.Key, a.BucketingKey, a.Feature, a.Attributes)
 		return res.Treatment, err
-	case "treatments", "treatmentWithConfig", "treatmentsWithConfig", "track":
+	case "treatments":
+		res, err := c.Treatments(a.Key, a.BucketingKey, a.Features, a.Attributes)
+		var sb strings.Builder
+		for _, result := range res {
+			if sb.Len() == 0 { // first item doesn't require a leading ','
+				sb.WriteString(result.Treatment)
+			} else {
+				sb.WriteString("," + result.Treatment)
+			}
+		}
+		return sb.String(), err
+	case "track":
+        return "", c.Track(a.Key, a.TrafficType, a.EventType, a.EventVal, nil)
+	case "treatmentWithConfig", "treatmentsWithConfig":
 		return "", fmt.Errorf("method '%s' is not yet implemented", a.Method)
 	default:
 		return "", fmt.Errorf("unknwon method '%s'", a.Method)

cmd/splitd/main.go

@@ -15,25 +15,27 @@ import (
 
 func main() {
 
-    printHeader()
+	printHeader()
 
 	cfg, err := conf.ReadConfig()
 	if err != nil {
 		fmt.Println("error reading config: ", err.Error())
 		os.Exit(1)
 	}
-    handleFlags(cfg)
+	handleFlags(cfg)
 
-	logger := logging.NewLogger(cfg.Logger.ToLoggerOptions())
+	loggerCfg, err := cfg.Logger.ToLoggerOptions()
+	exitOnErr("logging setup", err)
+	logger := logging.NewLogger(loggerCfg)
 
 	splitSDK, err := sdk.New(logger, cfg.SDK.Apikey, cfg.SDK.ToSDKConf())
-    exitOnErr("sdk initialization", err)
+	exitOnErr("sdk initialization", err)
 
-    linkCFG, err := cfg.Link.ToListenerOpts()
-    exitOnErr("link config", err)
+	linkCFG, err := cfg.Link.ToListenerOpts()
+	exitOnErr("link config", err)
 
 	errc, lShutdown, err := link.Listen(logger, splitSDK, linkCFG)
-    exitOnErr("rpc listener setup", err)
+	exitOnErr("rpc listener setup", err)
 
 	shutdown := util.NewShutdownHandler()
 	shutdown.RegisterHook(func() {
@@ -46,26 +48,26 @@ func main() {
 
 	// Wait for connection to end (either gracefully of because of an error)
 	err = <-errc
-    exitOnErr("shutdown: ", err)
+	exitOnErr("shutdown: ", err)
 }
 
 func printHeader() {
-    fmt.Println(splitio.ASCILogo)
-    fmt.Printf("Splitd Agent - Version %s. (2023)\n\n", splitio.Version)
+	fmt.Println(splitio.ASCILogo)
+	fmt.Printf("Splitd Agent - Version %s. (2023)\n\n", splitio.Version)
 }
 
 func handleFlags(cfg *conf.Config) {
-    printConf := flag.Bool("outputConfig", false, "print config (with partially obfuscated apikey)")
-    flag.Parse()
-    if *printConf {
-        fmt.Printf("\nConfig: %s\n", cfg)
-        os.Exit(0)
-    }
+	printConf := flag.Bool("outputConfig", false, "print config (with partially obfuscated apikey)")
+	flag.Parse()
+	if *printConf {
+		fmt.Printf("\nConfig: %s\n", cfg)
+		os.Exit(0)
+	}
 }
 
 func exitOnErr(ctxStr string, err error) {
 	if err != nil {
-        fmt.Printf("%s: startup error: %s\n", ctxStr, err.Error())
+		fmt.Printf("%s: startup error: %s\n", ctxStr, err.Error())
 		os.Exit(1)
 	}
 

splitio/conf/splitcli.go

@@ -14,6 +14,7 @@ import (
 )
 
 type CliArgs struct {
+	ID             string
 	LogLevel       string
 	Protocol       string
 	Serialization  string
@@ -31,27 +32,27 @@ type CliArgs struct {
 	Features     []string
 	TrafficType  string
 	EventType    string
-	EventVal     float64
+	EventVal     *float64
 	Attributes   map[string]interface{}
 }
 
 func (a *CliArgs) LinkOpts() (*link.ConsumerOptions, error) {
 
 	opts := link.DefaultConsumerOptions()
+	cc.SetIfNotEmpty(&opts.Consumer.ID, &a.ID)
+
 
 	var err error
 	if a.Protocol != "" {
 		if opts.Consumer.Protocol, err = cc.ParseProtocolVersion(a.Protocol); err != nil {
 			return nil, fmt.Errorf("invalid protocol version %s", a.Protocol)
 		}
 	}
-
 	if a.ConnType != "" {
 		if opts.Transfer.ConnType, err = cc.ParseConnType(a.ConnType); err != nil {
 			return nil, fmt.Errorf("invalid connection type %s", a.ConnType)
 		}
 	}
-
 	if a.Serialization != "" {
 		if opts.Serialization, err = cc.ParseSerializer(a.Serialization); err != nil {
 			return nil, fmt.Errorf("invalid serialization %s", a.Serialization)
@@ -69,6 +70,7 @@ func (a *CliArgs) LinkOpts() (*link.ConsumerOptions, error) {
 func ParseCliArgs() (*CliArgs, error) {
 
 	cliFlags := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
+	id := cliFlags.String("id", "", "ID to use for internal event queuing separation. Defaults to app's pid")
 	p := cliFlags.String("protocol", "", "Protocol version [v1]")
 	s := cliFlags.String("serialization", "", "Client-Daemon communication serialization mechanism [msgpack]")
 	ll := cliFlags.String("log-level", "INFO", "log level [ERROR,WARNING,INFO,DEBUG]")
@@ -89,9 +91,13 @@ func ParseCliArgs() (*CliArgs, error) {
 		return nil, fmt.Errorf("error parsing arguments: %w", err)
 	}
 
-	val, err := strconv.ParseFloat(*ev, 64)
-	if *ev != "" && err != nil {
-		return nil, fmt.Errorf("error parsing event value")
+	var eventVal *float64
+	if *ev != "" {
+		val, err := strconv.ParseFloat(*ev, 64)
+		if err != nil {
+			return nil, fmt.Errorf("error parsing event value")
+		}
+		eventVal = &val
 	}
 
 	if *at == "" {
@@ -103,20 +109,21 @@ func ParseCliArgs() (*CliArgs, error) {
 	}
 
 	return &CliArgs{
-        Serialization: *s,
-		Protocol:     *p,
-		LogLevel:     *ll,
-		ConnType:     *ct,
-		ConnAddr:     *ca,
-		BufSize:      *bs,
-		Method:       *m,
-		Key:          *k,
-		BucketingKey: *bk,
-		Feature:      *f,
-		Features:     strings.Split(*fs, ","),
-		TrafficType:  *tt,
-		EventType:    *et,
-		EventVal:     val,
-		Attributes:   attrs,
+		ID:            *id,
+		Serialization: *s,
+		Protocol:      *p,
+		LogLevel:      *ll,
+		ConnType:      *ct,
+		ConnAddr:      *ca,
+		BufSize:       *bs,
+		Method:        *m,
+		Key:           *k,
+		BucketingKey:  *bk,
+		Feature:       *f,
+		Features:      strings.Split(*fs, ","),
+		TrafficType:   *tt,
+		EventType:     *et,
+		EventVal:      eventVal,
+		Attributes:    attrs,
 	}, nil
 }

splitio/conf/splitcli_test.go

@@ -39,7 +39,7 @@ func TestCliConfig(t *testing.T) {
 	assert.Equal(t, []string{"someFeature1", "someFeature2"}, parsed.Features)
 	assert.Equal(t, "someTrafficType", parsed.TrafficType)
 	assert.Equal(t, "someEventType", parsed.EventType)
-	assert.Equal(t, 0.123, parsed.EventVal)
+	assert.Equal(t, ref(float64(0.123)), parsed.EventVal)
 	assert.Equal(t, map[string]interface{}{"some": "attribute"}, parsed.Attributes)
 
 	// test bad buffer size
@@ -63,32 +63,32 @@ func TestCliConfig(t *testing.T) {
 }
 
 func TestLinkOptions(t *testing.T) {
-    // test defaults
-    os.Args = os.Args[:1]
+	// test defaults
+	os.Args = os.Args[:1]
 	parsed, err := ParseCliArgs()
 	assert.Nil(t, err)
 	lo, err := parsed.LinkOpts()
 	assert.Nil(t, err)
 	assert.Equal(t, link.DefaultConsumerOptions(), *lo)
 
-    // test bad protocol
-    os.Args = []string{os.Args[0], "-protocol=sarasa"}
+	// test bad protocol
+	os.Args = []string{os.Args[0], "-protocol=sarasa"}
 	parsed, err = ParseCliArgs()
 	assert.Nil(t, err)
 	lo, err = parsed.LinkOpts()
 	assert.NotNil(t, err)
 	assert.ErrorContains(t, err, "protocol")
 
-    // test bad conn type
-    os.Args = []string{os.Args[0], "-conn-type=sarasa"}
+	// test bad conn type
+	os.Args = []string{os.Args[0], "-conn-type=sarasa"}
 	parsed, err = ParseCliArgs()
 	assert.Nil(t, err)
 	lo, err = parsed.LinkOpts()
 	assert.NotNil(t, err)
 	assert.ErrorContains(t, err, "connection type")
 
-    // test bad serialization
-    os.Args = []string{os.Args[0], "-serialization=pinpin"}
+	// test bad serialization
+	os.Args = []string{os.Args[0], "-serialization=pinpin"}
 	parsed, err = ParseCliArgs()
 	assert.Nil(t, err)
 	lo, err = parsed.LinkOpts()

splitio/conf/splitd.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/splitio/go-toolkit/v5/logging"
 	"github.com/splitio/splitd/splitio/link"
+	sdlogging "github.com/splitio/splitd/splitio/logging"
 	sdkConf "github.com/splitio/splitd/splitio/sdk/conf"
 	cc "github.com/splitio/splitd/splitio/util/conf"
 	"gopkg.in/yaml.v3"
@@ -94,20 +95,50 @@ func (l *Link) ToListenerOpts() (*link.ListenerOptions, error) {
 }
 
 type SDK struct {
-	Apikey           string `yaml:"apikey"`
-	LabelsEnabled    *bool  `yaml:"labelsEnabled"`
-	StreamingEnabled *bool  `yaml:"streamingEnabled"`
-	URLs             URLs   `yaml:"urls"`
+	Apikey           string       `yaml:"apikey"`
+	LabelsEnabled    *bool        `yaml:"labelsEnabled"`
+	StreamingEnabled *bool        `yaml:"streamingEnabled"`
+	URLs             URLs         `yaml:"urls"`
+	FeatureFlags     FeatureFlags `yaml:"featureFlags"`
+	Impressions      Impressions  `yaml:"impressions"`
 }
 
-func (s *SDK) ToSDKConf() *sdkConf.Config {
+type FeatureFlags struct {
+	SplitNotificationQueueSize   *int `yaml:"splitNotificationQueueSize"`
+	SplitRefreshRateSeconds      *int `yaml:"splitRefreshSeconds"`
+	SegmentNotificationQueueSize *int `yaml:"segmentNotificationQueueSize"`
+	SegmentRefreshRateSeconds    *int `yaml:"segmentRefreshSeconds"`
+	SegmentUpdateWorkers         *int `yaml:"segmentUpdateWorkers"`
+	SegmentUpdateQueueSize       *int `yaml:"segmentUpdateQueueSize"`
+}
 
+type Impressions struct {
+	Mode                    *string `yaml:"mode"`
+	RefreshRateSeconds      *int    `yaml:"refreshRateSeconds"`
+	CountRefreshRateSeconds *int    `yaml:"countRefreshRateSeconds"`
+	QueueSize               *int    `yaml:"queueSize"`
+	ObserverSize            *int    `yaml:"observerSize"`
+	Watermark               *int    `yaml:"watermark"`
+}
+
+func (s *SDK) ToSDKConf() *sdkConf.Config {
 	cfg := sdkConf.DefaultConfig()
+	durationFromSeconds := func(seconds int) time.Duration { return time.Duration(seconds) * time.Second }
 	cc.SetIfNotNil(&cfg.LabelsEnabled, s.LabelsEnabled)
 	cc.SetIfNotNil(&cfg.StreamingEnabled, s.StreamingEnabled)
+	cc.SetIfNotEmpty(&cfg.Splits.UpdateBufferSize, s.FeatureFlags.SplitNotificationQueueSize)
+	cc.MapIfNotNil(&cfg.Splits.SyncPeriod, s.FeatureFlags.SplitRefreshRateSeconds, durationFromSeconds)
+	cc.SetIfNotEmpty(&cfg.Segments.UpdateBufferSize, s.FeatureFlags.SegmentNotificationQueueSize)
+	cc.SetIfNotEmpty(&cfg.Segments.QueueSize, s.FeatureFlags.SegmentUpdateQueueSize)
+	cc.SetIfNotEmpty(&cfg.Segments.WorkerCount, s.FeatureFlags.SegmentUpdateWorkers)
+	cc.MapIfNotNil(&cfg.Segments.SyncPeriod, s.FeatureFlags.SegmentRefreshRateSeconds, durationFromSeconds)
+	cc.SetIfNotEmpty(&cfg.Impressions.Mode, s.Impressions.Mode)
+	cc.SetIfNotEmpty(&cfg.Impressions.ObserverSize, s.Impressions.ObserverSize)
+	cc.SetIfNotEmpty(&cfg.Impressions.QueueSize, s.Impressions.QueueSize)
+	cc.MapIfNotNil(&cfg.Impressions.SyncPeriod, s.Impressions.RefreshRateSeconds, durationFromSeconds)
+	cc.MapIfNotNil(&cfg.Impressions.CountSyncPeriod, s.Impressions.CountRefreshRateSeconds, durationFromSeconds)
 	s.URLs.updateSDKConfURLs(&cfg.URLs)
 	return cfg
-
 }
 
 type URLs struct {
@@ -127,21 +158,34 @@ func (u *URLs) updateSDKConfURLs(dst *sdkConf.URLs) {
 }
 
 type Logger struct {
-	Level *string `yaml:"level"`
+	Level                   *string `yaml:"level"`
+	Output                  *string `yaml:"file"`
+	RotationMaxFiles        *int    `yaml:"rotationMaxFiles"`
+	RotationMaxBytesPerFile *int    `yaml:"rotationMaxBytesPerFile"`
 }
 
-func (l *Logger) ToLoggerOptions() *logging.LoggerOptions {
+func (l *Logger) ToLoggerOptions() (*logging.LoggerOptions, error) {
+
+	writer, err := sdlogging.GetWriter(l.Output, l.RotationMaxFiles, l.RotationMaxBytesPerFile)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing logger options: %w", err)
+	}
 
 	opts := &logging.LoggerOptions{
 		LogLevel:            logging.LevelError,
 		StandardLoggerFlags: log.Ltime | log.Lshortfile,
+		ErrorWriter:         writer,
+		WarningWriter:       writer,
+		InfoWriter:          writer,
+		DebugWriter:         writer,
+		VerboseWriter:       writer,
 	}
 
 	if l.Level != nil {
 		opts.LogLevel = logging.Level(strings.ToUpper(*l.Level))
 	}
 
-	return opts
+	return opts, nil
 }
 
 func ReadConfig() (*Config, error) {

splitio/conf/splitd_test.go

@@ -117,6 +117,22 @@ func TestSDK(t *testing.T) {
 			Streaming: ref("streamingURL"),
 			Telemetry: ref("telemetryURL"),
 		},
+		FeatureFlags: FeatureFlags{
+			SplitNotificationQueueSize:   ref(1),
+			SplitRefreshRateSeconds:      ref(2),
+			SegmentNotificationQueueSize: ref(3),
+			SegmentRefreshRateSeconds:    ref(4),
+			SegmentUpdateWorkers:         ref(5),
+			SegmentUpdateQueueSize:       ref(6),
+		},
+		Impressions: Impressions{
+			Mode:                    ref("optimized"),
+			RefreshRateSeconds:      ref(1),
+			CountRefreshRateSeconds: ref(2),
+			QueueSize:               ref(3),
+			ObserverSize:            ref(4),
+			Watermark:               ref(5),
+		},
 	}
 
 	expected := conf.DefaultConfig()
@@ -127,7 +143,17 @@ func TestSDK(t *testing.T) {
 	expected.URLs.Events = "eventsURL"
 	expected.URLs.Streaming = "streamingURL"
 	expected.URLs.Telemetry = "telemetryURL"
-
+	expected.Splits.UpdateBufferSize = 1
+	expected.Splits.SyncPeriod = 2 * time.Second
+    expected.Segments.UpdateBufferSize = 3
+    expected.Segments.SyncPeriod = 4 * time.Second
+    expected.Segments.WorkerCount = 5
+    expected.Segments.QueueSize = 6
+    expected.Impressions.Mode = "optimized"
+    expected.Impressions.SyncPeriod = 1 * time.Second
+    expected.Impressions.CountSyncPeriod = 2 * time.Second
+    expected.Impressions.QueueSize = 3
+    expected.Impressions.ObserverSize = 4
 	assert.Equal(t, expected, sdkCFG.ToSDKConf())
 }