WIP AddFlags and InitFromViper

Author: mhoffmann

cmd/ingester/app/flags.go

@@ -32,47 +32,14 @@ const (
 	ConfigPrefix = "ingester"
 	// KafkaConsumerConfigPrefix is a prefix for the Kafka flags
 	KafkaConsumerConfigPrefix = "kafka.consumer"
-	// 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"
-	// SuffixClientID is a suffix for the client-id flag
-	SuffixClientID = ".client-id"
+
 	// SuffixEncoding is a suffix for the encoding flag
 	SuffixEncoding = ".encoding"
-	// SuffixTLS is a suffix for the tls flag
-	SuffixTLS = ".tls"
-	// SuffixCert is a suffix for the tls certificate path flag
-	SuffixCert = ".tls.cert"
-	// SuffixKey is a suffix for the tls key path flag
-	SuffixKey = ".tls.key"
-	// SuffixCA is a suffix for the tls ca path flag
-	SuffixCA = ".tls.ca"
 	// SuffixDeadlockInterval is a suffix for deadlock detecor flag
 	SuffixDeadlockInterval = ".deadlockInterval"
 	// SuffixParallelism is a suffix for the parallelism flag
 	SuffixParallelism = ".parallelism"
-	// SuffixHTTPPort is a suffix for the HTTP port
-	SuffixHTTPPort = ".http-port"
 
-	// 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"
-	// DefaultClientID is the default consumer Client ID
-	DefaultClientID = "jaeger-ingester"
-	// DefaultTLS is the default for TLS enabled
-	DefaultTLS = false
-	// DefaultCAPath is the default for the TLS CA path
-	DefaultCAPath = ""
-	// DefaultCertPath is the default for the TLS Cert path
-	DefaultCertPath = ""
-	// DefaultKeyPath is the default for the TLS key path
-	DefaultKeyPath = ""
 	// DefaultParallelism is the default parallelism for the span processor
 	DefaultParallelism = 1000
 	// DefaultEncoding is the default span encoding
@@ -84,49 +51,21 @@ const (
 // Options stores the configuration options for the Ingester
 type Options struct {
 	kafkaConsumer.Configuration
+
 	Parallelism      int
 	Encoding         string
 	DeadlockInterval time.Duration
 }
 
 // AddFlags adds flags for Builder
 func AddFlags(flagSet *flag.FlagSet) {
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixBrokers,
-		DefaultBroker,
-		"The comma-separated list of kafka brokers. i.e. '127.0.0.1:9092,0.0.0:1234'")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixTopic,
-		DefaultTopic,
-		"The name of the kafka topic to consume from")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixGroupID,
-		DefaultGroupID,
-		"The Consumer Group that ingester will be consuming on behalf of")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixClientID,
-		DefaultClientID,
-		"The Consumer Client ID that ingester will use")
+	kafkaConsumer.AddFlags(KafkaConsumerConfigPrefix, flagSet)
+
 	flagSet.String(
 		KafkaConsumerConfigPrefix+SuffixEncoding,
 		DefaultEncoding,
 		fmt.Sprintf(`The encoding of spans ("%s") consumed from kafka`, strings.Join(kafka.AllEncodings, "\", \"")))
-	flagSet.Bool(
-		KafkaConsumerConfigPrefix+SuffixTLS,
-		DefaultTLS,
-		"Enable TLS for the Kafka connection")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixCA,
-		DefaultCAPath,
-		"Path to the TLS CA for the Kafka connection")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixCert,
-		DefaultCertPath,
-		"Path to the TLS Certificate for the Kafka connection")
-	flagSet.String(
-		KafkaConsumerConfigPrefix+SuffixKey,
-		DefaultKeyPath,
-		"Path to the TLS Key for the Kafka connection")
+
 	flagSet.String(
 		ConfigPrefix+SuffixParallelism,
 		strconv.Itoa(DefaultParallelism),
@@ -139,17 +78,9 @@ func AddFlags(flagSet *flag.FlagSet) {
 
 // InitFromViper initializes Builder with properties from viper
 func (o *Options) InitFromViper(v *viper.Viper) {
-	o.Brokers = strings.Split(stripWhiteSpace(v.GetString(KafkaConsumerConfigPrefix+SuffixBrokers)), ",")
-	o.Topic = v.GetString(KafkaConsumerConfigPrefix + SuffixTopic)
-	o.GroupID = v.GetString(KafkaConsumerConfigPrefix + SuffixGroupID)
-	o.ClientID = v.GetString(KafkaConsumerConfigPrefix + SuffixClientID)
-	o.Encoding = v.GetString(KafkaConsumerConfigPrefix + SuffixEncoding)
+	o.Configuration.InitFromViper(KafkaConsumerConfigPrefix, v)
 
+	o.Encoding = v.GetString(KafkaConsumerConfigPrefix + SuffixEncoding)
 	o.Parallelism = v.GetInt(ConfigPrefix + SuffixParallelism)
 	o.DeadlockInterval = v.GetDuration(ConfigPrefix + SuffixDeadlockInterval)
 }
-
-// stripWhiteSpace removes all whitespace characters from a string
-func stripWhiteSpace(str string) string {
-	return strings.Replace(str, " ", "", -1)
-}

cmd/ingester/app/flags_test.go

@@ -15,6 +15,8 @@
 package app
 
 import (
+	kafkaConfig "github.com/jaegertracing/jaeger/pkg/kafka/config"
+	"github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"testing"
 	"time"
 
@@ -53,10 +55,10 @@ func TestFlagDefaults(t *testing.T) {
 	command.ParseFlags([]string{})
 	o.InitFromViper(v)
 
-	assert.Equal(t, DefaultTopic, o.Topic)
-	assert.Equal(t, []string{DefaultBroker}, o.Brokers)
-	assert.Equal(t, DefaultGroupID, o.GroupID)
-	assert.Equal(t, DefaultClientID, o.ClientID)
+	assert.Equal(t, consumer.DefaultTopic, o.Topic)
+	assert.Equal(t, []string{kafkaConfig.DefaultBroker}, o.Brokers)
+	assert.Equal(t, consumer.DefaultGroupID, o.GroupID)
+	assert.Equal(t, consumer.DefaultClientID, o.ClientID)
 	assert.Equal(t, DefaultParallelism, o.Parallelism)
 	assert.Equal(t, DefaultEncoding, o.Encoding)
 	assert.Equal(t, DefaultDeadlockInterval, o.DeadlockInterval)

pkg/es/config/.nocover

@@ -17,56 +17,114 @@ package config
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"flag"
+	"github.com/pkg/errors"
+	"github.com/spf13/viper"
 	"io/ioutil"
+	"strings"
+)
+
+const (
+	// SuffixBrokers is a suffix for the brokers flag
+	SuffixBrokers = ".brokers"
+	// SuffixTopic is a suffix for the topic flag
+	SuffixTLS = ".tls"
+	// SuffixCert is a suffix for the tls certificate path flag
+	SuffixCert = ".tls.cert"
+	// SuffixKey is a suffix for the tls key path flag
+	SuffixKey = ".tls.key"
+	// SuffixCA is a suffix for the tls ca path flag
+	SuffixCA = ".tls.ca"
+
+	// DefaultBroker is the default kafka broker
+	DefaultBroker = "127.0.0.1:9092"
+	// DefaultTLS is the default for TLS enabled
+	DefaultTLS = false
+	// DefaultCAPath is the default for the TLS CA path
+	DefaultCAPath = ""
+	// DefaultCertPath is the default for the TLS Cert path
+	DefaultCertPath = ""
+	// DefaultKeyPath is the default for the TLS key path
+	DefaultKeyPath = ""
 )
 
 // Configuration describes the shared configuration options for Producers or Consumers
 type Configuration struct {
 	Brokers []string
-	TLS     TLSConfig
+	TLS     TLSConfiguration
 }
 
-// TLSConfig describes the configuration properties for TLS Connections to the Kafka Brokers
-type TLSConfig struct {
+// TLSConfiguration describes the configuration properties for TLS Connections to the Kafka Brokers
+type TLSConfiguration struct {
 	Enabled  bool
 	CertPath string
 	KeyPath  string
 	CaPath   string
 }
 
 // GetTLSConfig creates TLS Configuration
-func (tlsConfig *TLSConfig) GetTLSConfig() (*tls.Config, error) {
-	rootCerts, err := tlsConfig.loadCertificate()
+func (tlsConfig TLSConfiguration) GetTLSConfig() (*tls.Config, error) {
+	ca, err := loadCA(tlsConfig.CaPath)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrapf(err, "error reading ca")
 	}
-	clientPrivateKey, err := tlsConfig.loadPrivateKey()
+
+	cert, err := tls.LoadX509KeyPair(tlsConfig.KeyPath, tlsConfig.CertPath)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrap(err, "error loading certificate")
 	}
+
 	return &tls.Config{
-		RootCAs:      rootCerts,
-		Certificates: []tls.Certificate{*clientPrivateKey},
+		RootCAs:      ca,
+		Certificates: []tls.Certificate{cert},
 	}, nil
-
 }
 
-// loadCertificate is used to load root certification
-func (tlsConfig *TLSConfig) loadCertificate() (*x509.CertPool, error) {
-	caCert, err := ioutil.ReadFile(tlsConfig.CaPath)
+func loadCA(caPath string) (*x509.CertPool, error) {
+	caBytes, err := ioutil.ReadFile(caPath)
 	if err != nil {
-		return nil, err
+		return nil, errors.Wrapf(err, "error reading caFile %s", caPath)
 	}
 	certificates := x509.NewCertPool()
-	certificates.AppendCertsFromPEM(caCert)
+	if ok := certificates.AppendCertsFromPEM(caBytes); !ok {
+		return nil, errors.Errorf("no ca certificates could be parsed")
+	}
 	return certificates, nil
 }
 
-// loadPrivateKey is used to load the private certificate and key for TLS
-func (tlsConfig *TLSConfig) loadPrivateKey() (*tls.Certificate, error) {
-	privateKey, err := tls.LoadX509KeyPair(tlsConfig.CertPath, tlsConfig.KeyPath)
-	if err != nil {
-		return nil, err
-	}
-	return &privateKey, nil
+func AddFlags(prefix string, flagSet *flag.FlagSet) {
+	flagSet.String(
+		prefix+SuffixBrokers,
+		DefaultBroker,
+		"The comma-separated list of kafka brokers. i.e. '127.0.0.1:9092,0.0.0:1234'")
+	flagSet.Bool(
+		prefix+SuffixTLS,
+		DefaultTLS,
+		"Enable TLS for the Kafka connection")
+	flagSet.String(
+		prefix+SuffixCA,
+		DefaultCAPath,
+		"Path to the TLS CA for the Kafka connection")
+	flagSet.String(
+		prefix+SuffixCert,
+		DefaultCertPath,
+		"Path to the TLS Certificate for the Kafka connection")
+	flagSet.String(
+		prefix+SuffixKey,
+		DefaultKeyPath,
+		"Path to the TLS Key for the Kafka connection")
+
+}
+
+func (c *Configuration) InitFromViper(prefix string, v *viper.Viper) {
+	c.Brokers = strings.Split(stripWhiteSpace(v.GetString(prefix+SuffixBrokers)), ",")
+	c.TLS.Enabled = v.GetBool(prefix + SuffixTLS)
+	c.TLS.CaPath = v.GetString(prefix + SuffixCA)
+	c.TLS.CertPath = v.GetString(prefix + SuffixCert)
+	c.TLS.KeyPath = v.GetString(prefix + SuffixKey)
+}
+
+// stripWhiteSpace removes all whitespace characters from a string
+func stripWhiteSpace(str string) string {
+	return strings.Replace(str, " ", "", -1)
 }

pkg/kafka/config/.nocover

@@ -0,0 +1,19 @@
+package config
+
+import (
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+const nonExistantFilepath = "/hope/that/this/does/not/exist"
+
+func TestTLSConfiguration(t *testing.T) {
+	_, err := TLSConfiguration{
+		Enabled:  true,
+		CertPath: nonExistantFilepath,
+		KeyPath:  nonExistantFilepath,
+		CaPath:   nonExistantFilepath,
+	}.GetTLSConfig()
+
+	assert.Error(t, err)
+}

pkg/kafka/config/config.go

@@ -15,12 +15,30 @@
 package consumer
 
 import (
+	"flag"
+	"github.com/spf13/viper"
 	"io"
 
 	"github.com/bsm/sarama-cluster"
 	"github.com/pkg/errors"
 
-	"github.com/jaegertracing/jaeger/pkg/kafka/config"
+	kafkaConfig "github.com/jaegertracing/jaeger/pkg/kafka/config"
+)
+
+const (
+	// SuffixTopic is a suffix for the topic flag
+	SuffixTopic = ".topic"
+	// SuffixGroupID is a suffix for the group-id flag
+	SuffixGroupID = ".group-id"
+	// SuffixClientID is a suffix for the client-id flag
+	SuffixClientID = ".client-id"
+
+	// DefaultTopic is the default kafka topic
+	DefaultTopic = "jaeger-spans"
+	// DefaultGroupID is the default consumer Group ID
+	DefaultGroupID = "jaeger-ingester"
+	// DefaultClientID is the default consumer Client ID
+	DefaultClientID = "jaeger-ingester"
 )
 
 // Consumer is an interface to features of Sarama that are necessary for the consumer
@@ -37,7 +55,7 @@ type Builder interface {
 
 // Configuration describes the configuration properties needed to create a Kafka consumer
 type Configuration struct {
-	config.Configuration
+	kafkaConfig.Configuration
 	Topic    string
 	GroupID  string
 	ClientID string
@@ -59,3 +77,27 @@ func (c *Configuration) NewConsumer() (Consumer, error) {
 	}
 	return cluster.NewConsumer(c.Brokers, c.GroupID, []string{c.Topic}, saramaConfig)
 }
+
+func AddFlags(prefix string, flagSet *flag.FlagSet) {
+	kafkaConfig.AddFlags(prefix, flagSet)
+	flagSet.String(
+		prefix+SuffixTopic,
+		DefaultTopic,
+		"The name of the kafka topic to consume from")
+	flagSet.String(
+		prefix+SuffixGroupID,
+		DefaultGroupID,
+		"The Consumer Group the consumer will use")
+	flagSet.String(
+		prefix+SuffixClientID,
+		DefaultClientID,
+		"The Consumer Client ID the consumer")
+}
+
+func (c *Configuration) InitFromViper(prefix string, v *viper.Viper) {
+	c.Configuration.InitFromViper(prefix, v)
+
+	c.Topic = v.GetString(prefix + SuffixTopic)
+	c.GroupID = v.GetString(prefix + SuffixGroupID)
+	c.ClientID = v.GetString(prefix + SuffixClientID)
+}