Merge branch 'master' into jaegertracing#1921

CONTRIBUTING.md

@@ -142,3 +142,13 @@ a comment explaining why it is there, for example:
 $ cat ./pkg/cassandra/config/.nocover
 requires connection to Cassandra
 ```
+
+## Merging PRs
+Before merging a PR make sure:
+* the title is descriptive and follows [a good commit message](./CONTRIBUTING_GUIDELINES.md)
+* pull request is assigned to the current release milestone
+* add `changelog:*` and other labels
+
+Merge the PR by using "Squash and merge" option on Github. Avoid creating merge commits.
+After the merge make sure referenced issues were closed.
+

RELEASE.md

@@ -11,12 +11,13 @@
         git fetch
         git checkout {new_version} //e.g. v1.5.0
         ```
-2. After the PR is merged, create a release on Github:
+2. Add all merged pull requests to the milestone for the release and create a new milestone for a next release e.g. `Release 1.16`.
+3. After the PR is merged, create a release on Github:
     * Title "Release X.Y.Z" 
     * Tag `vX.Y.Z` (note the `v` prefix) and choose appropriate branch
     * Copy the new CHANGELOG.md section into the release notes
-3. The release tag will trigger a build of the docker images
-4. Once the images are available on [Docker Hub](https://hub.docker.com/r/jaegertracing/), announce the release on the mailing list, gitter, and twitter.
-5. Publish documentation for the new version in [jaegertracing.io](https://github.com/jaegertracing/documentation).
+4. The release tag will trigger a build of the docker images
+5. Once the images are available on [Docker Hub](https://hub.docker.com/r/jaegertracing/), announce the release on the mailing list, gitter, and twitter.
+6. Publish documentation for the new version in [jaegertracing.io](https://github.com/jaegertracing/documentation).
 
 Maintenance branches should follow naming convention: `release-major.minor` (e.g.`release-1.8`).

cmd/all-in-one/main.go

@@ -345,14 +345,19 @@ func archiveOptions(storageFactory istorage.Factory, logger *zap.Logger) *querys
 }
 
 func initTracer(metricsFactory metrics.Factory, logger *zap.Logger) io.Closer {
-	tracer, closer, err := jaegerClientConfig.Configuration{
+	traceCfg := &jaegerClientConfig.Configuration{
 		ServiceName: "jaeger-query",
 		Sampler: &jaegerClientConfig.SamplerConfig{
 			Type:  "const",
 			Param: 1.0,
 		},
 		RPCMetrics: true,
-	}.NewTracer(
+	}
+	traceCfg, err := traceCfg.FromEnv()
+	if err != nil {
+		logger.Fatal("Failed to read tracer configuration", zap.Error(err))
+	}
+	tracer, closer, err := traceCfg.NewTracer(
 		jaegerClientConfig.Metrics(metricsFactory),
 		jaegerClientConfig.Logger(jaegerClientZapLog.NewLogger(logger)),
 	)

cmd/query/main.go

@@ -64,14 +64,19 @@ func main() {
 			baseFactory := svc.MetricsFactory.Namespace(metrics.NSOptions{Name: "jaeger"})
 			metricsFactory := baseFactory.Namespace(metrics.NSOptions{Name: "query"})
 
-			tracer, closer, err := jaegerClientConfig.Configuration{
+			traceCfg := &jaegerClientConfig.Configuration{
 				ServiceName: "jaeger-query",
 				Sampler: &jaegerClientConfig.SamplerConfig{
-					Type:  "probabilistic",
+					Type:  "const",
 					Param: 1.0,
 				},
 				RPCMetrics: true,
-			}.NewTracer(
+			}
+			traceCfg, err = traceCfg.FromEnv()
+			if err != nil {
+				logger.Fatal("Failed to read tracer configuration", zap.Error(err))
+			}
+			tracer, closer, err := traceCfg.NewTracer(
 				jaegerClientConfig.Metrics(svc.MetricsFactory),
 				jaegerClientConfig.Logger(jaegerClientZapLog.NewLogger(logger)),
 			)

pkg/es/config/config.go

@@ -300,7 +300,9 @@ func (c *Configuration) getConfigOptions(logger *zap.Logger) ([]elastic.ClientOp
 			TLSClientConfig: ctlsConfig,
 		}
 	} else {
-		httpTransport := &http.Transport{}
+		httpTransport := &http.Transport{
+			Proxy: http.ProxyFromEnvironment,
+		}
 		if c.TLS.CaPath != "" {
 			ctls := &TLSConfig{CaPath: c.TLS.CaPath}
 			ca, err := ctls.loadCertificate()

plugin/storage/cassandra/factory.go

@@ -16,6 +16,7 @@
 package cassandra
 
 import (
+	"errors"
 	"flag"
 
 	"github.com/spf13/viper"
@@ -26,6 +27,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/cassandra/config"
 	cDepStore "github.com/jaegertracing/jaeger/plugin/storage/cassandra/dependencystore"
 	cSpanStore "github.com/jaegertracing/jaeger/plugin/storage/cassandra/spanstore"
+	"github.com/jaegertracing/jaeger/plugin/storage/cassandra/spanstore/dbmodel"
 	"github.com/jaegertracing/jaeger/storage"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
@@ -102,7 +104,11 @@ func (f *Factory) CreateSpanReader() (spanstore.Reader, error) {
 
 // CreateSpanWriter implements storage.Factory
 func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) {
-	return cSpanStore.NewSpanWriter(f.primarySession, f.Options.SpanStoreWriteCacheTTL, f.primaryMetricsFactory, f.logger), nil
+	options, err := writerOptions(f.Options)
+	if err != nil {
+		return nil, err
+	}
+	return cSpanStore.NewSpanWriter(f.primarySession, f.Options.SpanStoreWriteCacheTTL, f.primaryMetricsFactory, f.logger, options...), nil
 }
 
 // CreateDependencyReader implements storage.Factory
@@ -124,5 +130,38 @@ func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 	if f.archiveSession == nil {
 		return nil, storage.ErrArchiveStorageNotConfigured
 	}
-	return cSpanStore.NewSpanWriter(f.archiveSession, f.Options.SpanStoreWriteCacheTTL, f.archiveMetricsFactory, f.logger), nil
+	options, err := writerOptions(f.Options)
+	if err != nil {
+		return nil, err
+	}
+	return cSpanStore.NewSpanWriter(f.archiveSession, f.Options.SpanStoreWriteCacheTTL, f.archiveMetricsFactory, f.logger, options...), nil
+}
+
+func writerOptions(opts *Options) ([]cSpanStore.Option, error) {
+	var tagFilters []dbmodel.TagFilter
+
+	// drop all tag filters
+	if opts.DisableTagsIndex || opts.DisableProcessTagsIndex || opts.DisableLogsIndex {
+		tagFilters = append(tagFilters, dbmodel.NewTagFilterDropAll(opts.DisableTagsIndex, opts.DisableProcessTagsIndex, opts.DisableLogsIndex))
+	}
+
+	// black/white list tag filters
+	tagIndexBlacklist := opts.TagIndexBlacklist()
+	tagIndexWhitelist := opts.TagIndexWhitelist()
+	if len(tagIndexBlacklist) > 0 && len(tagIndexWhitelist) > 0 {
+		return nil, errors.New("only one of TagIndexBlacklist and TagIndexWhitelist can be specified")
+	}
+	if len(tagIndexBlacklist) > 0 {
+		tagFilters = append(tagFilters, dbmodel.NewBlacklistFilter(tagIndexBlacklist))
+	} else if len(tagIndexWhitelist) > 0 {
+		tagFilters = append(tagFilters, dbmodel.NewWhitelistFilter(tagIndexWhitelist))
+	}
+
+	if len(tagFilters) == 0 {
+		return nil, nil
+	} else if len(tagFilters) == 1 {
+		return []cSpanStore.Option{cSpanStore.TagFilter(tagFilters[0])}, nil
+	}
+
+	return []cSpanStore.Option{cSpanStore.TagFilter(dbmodel.NewChainedTagFilter(tagFilters...))}, nil
 }

plugin/storage/cassandra/factory_test.go

@@ -100,3 +100,84 @@ func TestCassandraFactory(t *testing.T) {
 	_, err = f.CreateArchiveSpanWriter()
 	assert.NoError(t, err)
 }
+
+func TestExclusiveWhitelistBlacklist(t *testing.T) {
+	logger, logBuf := testutils.NewLogger()
+	f := NewFactory()
+	v, command := config.Viperize(f.AddFlags)
+	command.ParseFlags([]string{"--cassandra-archive.enabled=true",
+		"--cassandra.enable-dependencies-v2=true",
+		"--cassandra.index.tag-whitelist=a,b,c",
+		"--cassandra.index.tag-blacklist=a,b,c"})
+	f.InitFromViper(v)
+
+	// after InitFromViper, f.primaryConfig points to a real session builder that will fail in unit tests,
+	// so we override it with a mock.
+	f.primaryConfig = newMockSessionBuilder(nil, errors.New("made-up error"))
+	assert.EqualError(t, f.Initialize(metrics.NullFactory, zap.NewNop()), "made-up error")
+
+	var (
+		session = &mocks.Session{}
+		query   = &mocks.Query{}
+	)
+	session.On("Query", mock.AnythingOfType("string"), mock.Anything).Return(query)
+	query.On("Exec").Return(nil)
+	f.primaryConfig = newMockSessionBuilder(session, nil)
+	f.archiveConfig = newMockSessionBuilder(nil, errors.New("made-up error"))
+	assert.EqualError(t, f.Initialize(metrics.NullFactory, zap.NewNop()), "made-up error")
+
+	f.archiveConfig = nil
+	assert.NoError(t, f.Initialize(metrics.NullFactory, logger))
+	assert.Contains(t, logBuf.String(), "Cassandra archive storage configuration is empty, skipping")
+
+	_, err := f.CreateSpanWriter()
+	assert.EqualError(t, err, "only one of TagIndexBlacklist and TagIndexWhitelist can be specified")
+
+	f.archiveConfig = &mockSessionBuilder{}
+	assert.NoError(t, f.Initialize(metrics.NullFactory, zap.NewNop()))
+
+	_, err = f.CreateArchiveSpanWriter()
+	assert.EqualError(t, err, "only one of TagIndexBlacklist and TagIndexWhitelist can be specified")
+}
+
+func TestWriterOptions(t *testing.T) {
+	opts := NewOptions("cassandra")
+	v, command := config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{"--cassandra.index.tag-whitelist=a,b,c"})
+	opts.InitFromViper(v)
+
+	options, _ := writerOptions(opts)
+	assert.Len(t, options, 1)
+
+	opts = NewOptions("cassandra")
+	v, command = config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{"--cassandra.index.tag-blacklist=a,b,c"})
+	opts.InitFromViper(v)
+
+	options, _ = writerOptions(opts)
+	assert.Len(t, options, 1)
+
+	opts = NewOptions("cassandra")
+	v, command = config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{"--cassandra.index.tags=false"})
+	opts.InitFromViper(v)
+
+	options, _ = writerOptions(opts)
+	assert.Len(t, options, 1)
+
+	opts = NewOptions("cassandra")
+	v, command = config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{"--cassandra.index.tags=false", "--cassandra.index.tag-blacklist=a,b,c"})
+	opts.InitFromViper(v)
+
+	options, _ = writerOptions(opts)
+	assert.Len(t, options, 1)
+
+	opts = NewOptions("cassandra")
+	v, command = config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{""})
+	opts.InitFromViper(v)
+
+	options, _ = writerOptions(opts)
+	assert.Len(t, options, 0)
+}

plugin/storage/cassandra/options.go

@@ -53,15 +53,25 @@ const (
 
 	// common storage settings
 	suffixSpanStoreWriteCacheTTL = ".span-store-write-cache-ttl"
+	suffixIndexTagsBlacklist     = ".index.tag-blacklist"
+	suffixIndexTagsWhitelist     = ".index.tag-whitelist"
+	suffixIndexLogs              = ".index.logs"
+	suffixIndexTags              = ".index.tags"
+	suffixIndexProcessTags       = ".index.process-tags"
 )
 
 // Options contains various type of Cassandra configs and provides the ability
 // to bind them to command line flag and apply overlays, so that some configurations
 // (e.g. archive) may be underspecified and infer the rest of its parameters from primary.
 type Options struct {
-	primary                *namespaceConfig
-	others                 map[string]*namespaceConfig
-	SpanStoreWriteCacheTTL time.Duration
+	primary                 *namespaceConfig
+	others                  map[string]*namespaceConfig
+	SpanStoreWriteCacheTTL  time.Duration
+	tagIndexBlacklist       string
+	tagIndexWhitelist       string
+	DisableLogsIndex        bool
+	DisableTagsIndex        bool
+	DisableProcessTagsIndex bool
 }
 
 // the Servers field in config.Configuration is a list, which we cannot represent with flags.
@@ -116,6 +126,26 @@ func (opt *Options) AddFlags(flagSet *flag.FlagSet) {
 	flagSet.Duration(opt.primary.namespace+suffixSpanStoreWriteCacheTTL,
 		opt.SpanStoreWriteCacheTTL,
 		"The duration to wait before rewriting an existing service or operation name")
+	flagSet.String(
+		opt.primary.namespace+suffixIndexTagsBlacklist,
+		opt.tagIndexBlacklist,
+		"The comma-separated list of span tags to blacklist from being indexed. All other tags will be indexed. Mutually exclusive with the whitelist option.")
+	flagSet.String(
+		opt.primary.namespace+suffixIndexTagsWhitelist,
+		opt.tagIndexWhitelist,
+		"The comma-separated list of span tags to whitelist for being indexed. All other tags will not be indexed. Mutually exclusive with the blacklist option.")
+	flagSet.Bool(
+		opt.primary.namespace+suffixIndexLogs,
+		!opt.DisableLogsIndex,
+		"Controls log field indexing. Set to false to disable.")
+	flagSet.Bool(
+		opt.primary.namespace+suffixIndexTags,
+		!opt.DisableTagsIndex,
+		"Controls tag indexing. Set to false to disable.")
+	flagSet.Bool(
+		opt.primary.namespace+suffixIndexProcessTags,
+		!opt.DisableProcessTagsIndex,
+		"Controls process tag indexing. Set to false to disable.")
 }
 
 func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
@@ -222,6 +252,11 @@ func (opt *Options) InitFromViper(v *viper.Viper) {
 		cfg.initFromViper(v)
 	}
 	opt.SpanStoreWriteCacheTTL = v.GetDuration(opt.primary.namespace + suffixSpanStoreWriteCacheTTL)
+	opt.tagIndexBlacklist = stripWhiteSpace(v.GetString(opt.primary.namespace + suffixIndexTagsBlacklist))
+	opt.tagIndexWhitelist = stripWhiteSpace(v.GetString(opt.primary.namespace + suffixIndexTagsWhitelist))
+	opt.DisableTagsIndex = !v.GetBool(opt.primary.namespace + suffixIndexTags)
+	opt.DisableLogsIndex = !v.GetBool(opt.primary.namespace + suffixIndexLogs)
+	opt.DisableProcessTagsIndex = !v.GetBool(opt.primary.namespace + suffixIndexProcessTags)
 }
 
 func (cfg *namespaceConfig) initFromViper(v *viper.Viper) {
@@ -276,6 +311,24 @@ func (opt *Options) Get(namespace string) *config.Configuration {
 	return &nsCfg.Configuration
 }
 
+// TagIndexBlacklist returns the list of blacklisted tags
+func (opt *Options) TagIndexBlacklist() []string {
+	if len(opt.tagIndexBlacklist) > 0 {
+		return strings.Split(opt.tagIndexBlacklist, ",")
+	}
+
+	return nil
+}
+
+// TagIndexWhitelist returns the list of whitelisted tags
+func (opt *Options) TagIndexWhitelist() []string {
+	if len(opt.tagIndexWhitelist) > 0 {
+		return strings.Split(opt.tagIndexWhitelist, ",")
+	}
+
+	return nil
+}
+
 // stripWhiteSpace removes all whitespace characters from a string
 func stripWhiteSpace(str string) string {
 	return strings.Replace(str, " ", "", -1)

plugin/storage/cassandra/options_test.go

@@ -60,6 +60,10 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--cas.consistency=ONE",
 		"--cas.proto-version=3",
 		"--cas.socket-keep-alive=42s",
+		"--cas.index.tag-blacklist=blerg, blarg,blorg ",
+		"--cas.index.tag-whitelist=flerg, flarg,florg ",
+		"--cas.index.tags=true",
+		"--cas.index.process-tags=false",
 		// enable aux with a couple overrides
 		"--cas-aux.enabled=true",
 		"--cas-aux.keyspace=jaeger-archive",
@@ -74,6 +78,11 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, []string{"1.1.1.1", "2.2.2.2"}, primary.Servers)
 	assert.Equal(t, "ONE", primary.Consistency)
 	assert.Equal(t, false, primary.EnableDependenciesV2)
+	assert.Equal(t, []string{"blerg", "blarg", "blorg"}, opts.TagIndexBlacklist())
+	assert.Equal(t, []string{"flerg", "flarg", "florg"}, opts.TagIndexWhitelist())
+	assert.Equal(t, false, opts.DisableTagsIndex)
+	assert.Equal(t, true, opts.DisableProcessTagsIndex)
+	assert.Equal(t, false, opts.DisableLogsIndex)
 
 	aux := opts.Get("cas-aux")
 	require.NotNil(t, aux)
@@ -89,3 +98,13 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, 42*time.Second, aux.SocketKeepAlive)
 	assert.Equal(t, true, aux.EnableDependenciesV2)
 }
+
+func TestEmptyBlackWhiteLists(t *testing.T) {
+	opts := NewOptions("cas")
+	v, command := config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{})
+	opts.InitFromViper(v)
+
+	assert.Len(t, opts.TagIndexBlacklist(), 0)
+	assert.Len(t, opts.TagIndexWhitelist(), 0)
+}