Move ingester options to flags

Signed-off-by: Davit Yeghshatyan <davo@uber.com>

Author: Davit Yeghshatyan

cmd/ingester/app/builder/builder.go

@@ -16,9 +16,6 @@ package builder
 
 import (
 	"fmt"
-	"strings"
-
-	"github.com/spf13/viper"
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
 
@@ -27,56 +24,19 @@ import (
 	kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/plugin/storage/kafka"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
+	"github.com/jaegertracing/jaeger/cmd/ingester/app"
 )
 
-const (
-	// EncodingJSON indicates spans are encoded as a json byte array
-	EncodingJSON = "json"
-	// EncodingProto indicates spans are encoded as a protobuf byte array
-	EncodingProto = "protobuf"
-
-	// ConfigPrefix is a prefix fro the ingester flags
-	ConfigPrefix = "ingester"
-	// SuffixBrokers is a suffix for the brokers flag
-	SuffixBrokers = ".brokers"
-	// SuffixTopic is a suffix for the topic flag
-	SuffixTopic = ".topic"
-	// SuffixGroupID is a suffix for the group-id flag
-	SuffixGroupID = ".group-id"
-	// SuffixParallelism is a suffix for the parallelism flag
-	SuffixParallelism = ".parallelism"
-	// SuffixEncoding is a suffix for the encoding flag
-	SuffixEncoding = ".encoding"
-
-	// DefaultBroker is the default kafka broker
-	DefaultBroker = "127.0.0.1:9092"
-	// DefaultTopic is the default kafka topic
-	DefaultTopic = "jaeger-spans"
-	// DefaultGroupID is the default consumer Group ID
-	DefaultGroupID = "jaeger-ingester"
-	// DefaultParallelism is the default parallelism for the span processor
-	DefaultParallelism = 1000
-	// DefaultEncoding is the default span encoding
-	DefaultEncoding = EncodingProto
-)
-
-// Builder stores the configuration options for the Ingester
-type Builder struct {
-	kafkaConsumer.Configuration
-	Parallelism int
-	Encoding    string
-}
-
 // CreateConsumer creates a new span consumer for the ingester
-func (b *Builder) CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWriter spanstore.Writer) (*consumer.Consumer, error) {
+func CreateConsumer(logger *zap.Logger, metricsFactory metrics.Factory, spanWriter spanstore.Writer, options app.Options) (*consumer.Consumer, error) {
 	var unmarshaller kafka.Unmarshaller
-	if b.Encoding == EncodingJSON {
+	if options.Encoding == app.EncodingJSON {
 		unmarshaller = kafka.NewJSONUnmarshaller()
-	} else if b.Encoding == EncodingProto {
+	} else if options.Encoding == app.EncodingProto {
 		unmarshaller = kafka.NewProtobufUnmarshaller()
 	} else {
 		return nil, fmt.Errorf(`encoding '%s' not recognised, use one of ("%s" or "%s")`,
-			b.Encoding, EncodingProto, EncodingJSON)
+			options.Encoding, app.EncodingProto, app.EncodingJSON)
 	}
 
 	spParams := processor.SpanProcessorParams{
@@ -86,18 +46,18 @@ func (b *Builder) CreateConsumer(logger *zap.Logger, metricsFactory metrics.Fact
 	spanProcessor := processor.NewSpanProcessor(spParams)
 
 	consumerConfig := kafkaConsumer.Configuration{
-		Brokers: b.Brokers,
-		Topic:   b.Topic,
-		GroupID: b.GroupID,
+		Brokers: options.Brokers,
+		Topic:   options.Topic,
+		GroupID: options.GroupID,
 	}
 	saramaConsumer, err := consumerConfig.NewConsumer()
 	if err != nil {
 		return nil, err
 	}
 
 	factoryParams := consumer.ProcessorFactoryParams{
-		Topic:          b.Topic,
-		Parallelism:    b.Parallelism,
+		Topic:          options.Topic,
+		Parallelism:    options.Parallelism,
 		SaramaConsumer: saramaConsumer,
 		BaseProcessor:  spanProcessor,
 		Logger:         logger,
@@ -116,12 +76,3 @@ func (b *Builder) CreateConsumer(logger *zap.Logger, metricsFactory metrics.Fact
 	}
 	return consumer.New(consumerParams)
 }
-
-// InitFromViper initializes Builder with properties from viper
-func (b *Builder) InitFromViper(v *viper.Viper) {
-	b.Brokers = strings.Split(v.GetString(ConfigPrefix+SuffixBrokers), ",")
-	b.Topic = v.GetString(ConfigPrefix + SuffixTopic)
-	b.GroupID = v.GetString(ConfigPrefix + SuffixGroupID)
-	b.Parallelism = v.GetInt(ConfigPrefix + SuffixParallelism)
-	b.Encoding = v.GetString(ConfigPrefix + SuffixEncoding)
-}

cmd/ingester/app/flags.go

@@ -19,29 +19,78 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
+	kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
+	"strings"
+	"github.com/spf13/viper"
 )
 
+const (
+	// EncodingJSON indicates spans are encoded as a json byte array
+	EncodingJSON = "json"
+	// EncodingProto indicates spans are encoded as a protobuf byte array
+	EncodingProto = "protobuf"
+
+	// ConfigPrefix is a prefix fro the ingester flags
+	ConfigPrefix = "ingester"
+	// SuffixBrokers is a suffix for the brokers flag
+	SuffixBrokers = ".brokers"
+	// SuffixTopic is a suffix for the topic flag
+	SuffixTopic = ".topic"
+	// SuffixGroupID is a suffix for the group-id flag
+	SuffixGroupID = ".group-id"
+	// SuffixParallelism is a suffix for the parallelism flag
+	SuffixParallelism = ".parallelism"
+	// SuffixEncoding is a suffix for the encoding flag
+	SuffixEncoding = ".encoding"
+
+	// DefaultBroker is the default kafka broker
+	DefaultBroker = "127.0.0.1:9092"
+	// DefaultTopic is the default kafka topic
+	DefaultTopic = "jaeger-spans"
+	// DefaultGroupID is the default consumer Group ID
+	DefaultGroupID = "jaeger-ingester"
+	// DefaultParallelism is the default parallelism for the span processor
+	DefaultParallelism = 1000
+	// DefaultEncoding is the default span encoding
+	DefaultEncoding = EncodingProto
+)
+
+// Options stores the configuration options for the Ingester
+type Options struct {
+	kafkaConsumer.Configuration
+	Parallelism int
+	Encoding    string
+}
+
 // AddFlags adds flags for Builder
 func AddFlags(flagSet *flag.FlagSet) {
 	flagSet.String(
-		builder.ConfigPrefix+builder.SuffixBrokers,
-		builder.DefaultBroker,
+		ConfigPrefix+SuffixBrokers,
+		DefaultBroker,
 		"The comma-separated list of kafka brokers. i.e. '127.0.0.1:9092,0.0.0:1234'")
 	flagSet.String(
-		builder.ConfigPrefix+builder.SuffixTopic,
-		builder.DefaultTopic,
+		ConfigPrefix+SuffixTopic,
+		DefaultTopic,
 		"The name of the kafka topic to consume from")
 	flagSet.String(
-		builder.ConfigPrefix+builder.SuffixGroupID,
-		builder.DefaultGroupID,
+		ConfigPrefix+SuffixGroupID,
+		DefaultGroupID,
 		"The Consumer Group that ingester will be consuming on behalf of")
 	flagSet.String(
-		builder.ConfigPrefix+builder.SuffixParallelism,
-		strconv.Itoa(builder.DefaultParallelism),
+		ConfigPrefix+SuffixParallelism,
+		strconv.Itoa(DefaultParallelism),
 		"The number of messages to process in parallel")
 	flagSet.String(
-		builder.ConfigPrefix+builder.SuffixEncoding,
-		builder.DefaultEncoding,
-		fmt.Sprintf(`The encoding of spans ("%s" or "%s") consumed from kafka`, builder.EncodingProto, builder.EncodingJSON))
+		ConfigPrefix+SuffixEncoding,
+		DefaultEncoding,
+		fmt.Sprintf(`The encoding of spans ("%s" or "%s") consumed from kafka`, EncodingProto, EncodingJSON))
+}
+
+// InitFromViper initializes Builder with properties from viper
+func (o *Options) InitFromViper(v *viper.Viper) {
+	o.Brokers = strings.Split(v.GetString(ConfigPrefix+SuffixBrokers), ",")
+	o.Topic = v.GetString(ConfigPrefix + SuffixTopic)
+	o.GroupID = v.GetString(ConfigPrefix + SuffixGroupID)
+	o.Parallelism = v.GetInt(ConfigPrefix + SuffixParallelism)
+	o.Encoding = v.GetString(ConfigPrefix + SuffixEncoding)
 }

cmd/ingester/app/flags_test.go

@@ -19,37 +19,36 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/jaegertracing/jaeger/cmd/ingester/app/builder"
 	"github.com/jaegertracing/jaeger/pkg/config"
 )
 
 func TestOptionsWithFlags(t *testing.T) {
-	b := &builder.Builder{}
+	o := &Options{}
 	v, command := config.Viperize(AddFlags)
 	command.ParseFlags([]string{
 		"--ingester.topic=topic1",
 		"--ingester.brokers=127.0.0.1:9092,0.0.0:1234",
 		"--ingester.group-id=group1",
 		"--ingester.parallelism=5",
 		"--ingester.encoding=json"})
-	b.InitFromViper(v)
+	o.InitFromViper(v)
 
-	assert.Equal(t, "topic1", b.Topic)
-	assert.Equal(t, []string{"127.0.0.1:9092", "0.0.0:1234"}, b.Brokers)
-	assert.Equal(t, "group1", b.GroupID)
-	assert.Equal(t, 5, b.Parallelism)
-	assert.Equal(t, builder.EncodingJSON, b.Encoding)
+	assert.Equal(t, "topic1", o.Topic)
+	assert.Equal(t, []string{"127.0.0.1:9092", "0.0.0:1234"}, o.Brokers)
+	assert.Equal(t, "group1", o.GroupID)
+	assert.Equal(t, 5, o.Parallelism)
+	assert.Equal(t, EncodingJSON, o.Encoding)
 }
 
 func TestFlagDefaults(t *testing.T) {
-	b := &builder.Builder{}
+	o := &Options{}
 	v, command := config.Viperize(AddFlags)
 	command.ParseFlags([]string{})
-	b.InitFromViper(v)
+	o.InitFromViper(v)
 
-	assert.Equal(t, builder.DefaultTopic, b.Topic)
-	assert.Equal(t, []string{builder.DefaultBroker}, b.Brokers)
-	assert.Equal(t, builder.DefaultGroupID, b.GroupID)
-	assert.Equal(t, builder.DefaultParallelism, b.Parallelism)
-	assert.Equal(t, builder.DefaultEncoding, b.Encoding)
+	assert.Equal(t, DefaultTopic, o.Topic)
+	assert.Equal(t, []string{DefaultBroker}, o.Brokers)
+	assert.Equal(t, DefaultGroupID, o.GroupID)
+	assert.Equal(t, DefaultParallelism, o.Parallelism)
+	assert.Equal(t, DefaultEncoding, o.Encoding)
 }

cmd/ingester/main.go

@@ -82,9 +82,9 @@ func main() {
 				logger.Fatal("Failed to create span writer", zap.Error(err))
 			}
 
-			b := builder.Builder{}
-			b.InitFromViper(v)
-			consumer, err := b.CreateConsumer(logger, metricsFactory, spanWriter)
+			options := app.Options{}
+			options.InitFromViper(v)
+			consumer, err := builder.CreateConsumer(logger, metricsFactory, spanWriter, options)
 			if err != nil {
 				logger.Fatal("Unable to create consumer", zap.Error(err))
 			}