all the things

Signed-off-by: Won Jun Jang <wjang@uber.com>

Author: black-adder

cmd/collector/app/builder/span_handler_builder.go

@@ -29,7 +29,6 @@ import (
 	cascfg "github.com/uber/jaeger/pkg/cassandra/config"
 	escfg "github.com/uber/jaeger/pkg/es/config"
 	casSpanstore "github.com/uber/jaeger/plugin/storage/cassandra/spanstore"
-	casSpanstoreModel "github.com/uber/jaeger/plugin/storage/cassandra/spanstore/dbmodel"
 	esSpanstore "github.com/uber/jaeger/plugin/storage/es/spanstore"
 	"github.com/uber/jaeger/storage/spanstore"
 )
@@ -96,7 +95,6 @@ func (spanHb *SpanHandlerBuilder) initCassStore(builder cascfg.SessionBuilder) (
 		spanHb.collectorOpts.WriteCacheTTL,
 		spanHb.metricsFactory,
 		spanHb.logger,
-		casSpanstoreModel.DefaultTagFilter(),
 	), nil
 }
 

…dra/spanstore/dbmodel/log_tags_filter.go …a/spanstore/dbmodel/log_fields_filter.goplugin/storage/cassandra/spanstore/dbmodel/log_tags_filter.go renamed to plugin/storage/cassandra/spanstore/dbmodel/log_fields_filter.go plugin/storage/cassandra/spanstore/dbmodel/log_tags_filter.go renamed to plugin/storage/cassandra/spanstore/dbmodel/log_fields_filter.go

@@ -16,11 +16,17 @@ package dbmodel
 
 import "github.com/uber/jaeger/model"
 
-// FilterLogTags returns a filter that filters span.Logs.
-func FilterLogTags() FilterTags {
-	return filterLogTags
+// LogFieldsFilter filters all span.Logs.Fields.
+type LogFieldsFilter struct {
+	tagFilterImpl
 }
 
-func filterLogTags(span *model.Span) model.KeyValues {
-	return append(span.Tags, span.Process.Tags...)
+// NewLogFieldsFilter return a filter that filters all span.Logs.Fields.
+func NewLogFieldsFilter() *LogFieldsFilter {
+	return &LogFieldsFilter{}
+}
+
+// FilterLogFields implements TagFilter#FilterLogFields
+func (f *LogFieldsFilter) FilterLogFields(logFields model.KeyValues) model.KeyValues {
+	return model.KeyValues{}
 }

…panstore/dbmodel/log_tags_filter_test.go …nstore/dbmodel/log_fields_filter_test.goplugin/storage/cassandra/spanstore/dbmodel/log_tags_filter_test.go renamed to plugin/storage/cassandra/spanstore/dbmodel/log_fields_filter_test.go plugin/storage/cassandra/spanstore/dbmodel/log_tags_filter_test.go renamed to plugin/storage/cassandra/spanstore/dbmodel/log_fields_filter_test.go

@@ -17,32 +17,23 @@ package dbmodel
 import (
 	"testing"
 
-	"github.com/uber/jaeger/model"
-
 	"github.com/kr/pretty"
 	"github.com/stretchr/testify/assert"
 )
 
+var filter TagFilter = &LogFieldsFilter{} // Check API compliance
+
 func TestFilterLogTags(t *testing.T) {
-	expectedTags := model.KeyValues{
-		model.String(someStringTagKey, someStringTagValue),
-		model.Bool(someBoolTagKey, someBoolTagValue),
-		model.Int64(someLongTagKey, someLongTagValue),
-	}
-	testSpan := getTestJaegerSpan()
-	testSpan.Tags = expectedTags
-	testSpan.Process.Tags = model.KeyValues{}
-	testSpan.Logs = []model.Log{
-		{
-			Timestamp: someLogTimestamp,
-			Fields: model.KeyValues{
-				model.Float64(someDoubleTagKey, someDoubleTagValue),
-			},
-		},
+	span := getTestJaegerSpan()
+	filter := NewLogFieldsFilter()
+	expectedTags := append(someTags, someTags...)
+	filteredTags := filter.FilterProcessTags(span.Process.Tags)
+	filteredTags = append(filteredTags, filter.FilterTags(span.Tags)...)
+	for _, log := range span.Logs {
+		filteredTags = append(filteredTags, filter.FilterLogFields(log.Fields)...)
 	}
-	uniqueTags := FilterLogTags()(testSpan)
-	if !assert.EqualValues(t, expectedTags, uniqueTags) {
-		for _, diff := range pretty.Diff(expectedTags, uniqueTags) {
+	if !assert.EqualValues(t, expectedTags, filteredTags) {
+		for _, diff := range pretty.Diff(expectedTags, filteredTags) {
 			t.Log(diff)
 		}
 	}

plugin/storage/cassandra/spanstore/dbmodel/tag_filter.go

@@ -18,20 +18,28 @@ import (
 	"github.com/uber/jaeger/model"
 )
 
-// FilterTags filters out any tags that should not be persisted.
-type FilterTags func(span *model.Span) model.KeyValues
+// TODO (black-adder) add a chain filter
+
+// TagFilter filters out any tags that should not be indexed.
+type TagFilter interface {
+	FilterProcessTags(processTags model.KeyValues) model.KeyValues
+	FilterTags(tags model.KeyValues) model.KeyValues
+	FilterLogFields(logFields model.KeyValues) model.KeyValues
+}
 
 // DefaultTagFilter returns a filter that retrieves all tags from span.Tags, span.Logs, and span.Process.
-func DefaultTagFilter() FilterTags {
-	return filterNothing
+var DefaultTagFilter = tagFilterImpl{}
+
+type tagFilterImpl struct{}
+
+func (f tagFilterImpl) FilterProcessTags(processTags model.KeyValues) model.KeyValues {
+	return processTags
+}
+
+func (f tagFilterImpl) FilterTags(tags model.KeyValues) model.KeyValues {
+	return tags
 }
 
-func filterNothing(span *model.Span) model.KeyValues {
-	process := span.Process
-	allTags := span.Tags
-	allTags = append(allTags, process.Tags...)
-	for _, log := range span.Logs {
-		allTags = append(allTags, log.Fields...)
-	}
-	return allTags
+func (f tagFilterImpl) FilterLogFields(logFields model.KeyValues) model.KeyValues {
+	return logFields
 }

plugin/storage/cassandra/spanstore/dbmodel/tag_filter_test.go

@@ -22,8 +22,13 @@ import (
 )
 
 func TestDefaultTagFilter(t *testing.T) {
+	span := getTestJaegerSpan()
 	expectedTags := append(append(someTags, someTags...), someTags...)
-	filteredTags := DefaultTagFilter()(getTestJaegerSpan())
+	filteredTags := DefaultTagFilter.FilterProcessTags(span.Process.Tags)
+	filteredTags = append(filteredTags, DefaultTagFilter.FilterTags(span.Tags)...)
+	for _, log := range span.Logs {
+		filteredTags = append(filteredTags, DefaultTagFilter.FilterLogFields(log.Fields)...)
+	}
 	if !assert.EqualValues(t, expectedTags, filteredTags) {
 		for _, diff := range pretty.Diff(expectedTags, filteredTags) {
 			t.Log(diff)

plugin/storage/cassandra/spanstore/dbmodel/unique_tags.go

@@ -17,8 +17,12 @@ package dbmodel
 import "github.com/uber/jaeger/model"
 
 // GetAllUniqueTags creates a list of all unique tags from a set of filtered tags.
-func GetAllUniqueTags(span *model.Span, tagFilter FilterTags) []TagInsertion {
-	tags := tagFilter(span)
+func GetAllUniqueTags(span *model.Span, tagFilter TagFilter) []TagInsertion {
+	tags := tagFilter.FilterProcessTags(span.Process.Tags)
+	tags = append(tags, tagFilter.FilterTags(span.Tags)...)
+	for _, log := range span.Logs {
+		tags = append(tags, tagFilter.FilterLogFields(log.Fields)...)
+	}
 	tags.Sort()
 	uniqueTags := make([]TagInsertion, 0, len(tags))
 	for i := range tags {

plugin/storage/cassandra/spanstore/dbmodel/unique_tags_test.go

@@ -23,7 +23,7 @@ import (
 
 func TestGetUniqueTags(t *testing.T) {
 	expectedTags := getTestUniqueTags()
-	uniqueTags := GetAllUniqueTags(getTestJaegerSpan(), DefaultTagFilter())
+	uniqueTags := GetAllUniqueTags(getTestJaegerSpan(), DefaultTagFilter)
 	if !assert.EqualValues(t, expectedTags, uniqueTags) {
 		for _, diff := range pretty.Diff(expectedTags, uniqueTags) {
 			t.Log(diff)

plugin/storage/cassandra/spanstore/writer.go

@@ -86,7 +86,7 @@ type SpanWriter struct {
 	logger               *zap.Logger
 	tagIndexSkipped      metrics.Counter
 	bucketCounter        uint32
-	tagFilter            dbmodel.FilterTags
+	tagFilter            dbmodel.TagFilter
 }
 
 // NewSpanWriter returns a SpanWriter
@@ -95,14 +95,12 @@ func NewSpanWriter(
 	writeCacheTTL time.Duration,
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
-	tagFilter dbmodel.FilterTags,
+	options ...Option,
 ) *SpanWriter {
 	serviceNamesStorage := NewServiceNamesStorage(session, writeCacheTTL, metricsFactory, logger)
 	operationNamesStorage := NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger)
 	tagIndexSkipped := metricsFactory.Counter("tagIndexSkipped", nil)
-	if tagFilter == nil {
-		tagFilter = dbmodel.DefaultTagFilter()
-	}
+	opts := applyOptions(options...)
 	return &SpanWriter{
 		session:              session,
 		serviceNamesWriter:   serviceNamesStorage.Write,
@@ -116,7 +114,7 @@ func NewSpanWriter(
 		},
 		logger:          logger,
 		tagIndexSkipped: tagIndexSkipped,
-		tagFilter:       tagFilter,
+		tagFilter:       opts.tagFilter,
 	}
 }
 

plugin/storage/cassandra/spanstore/writer_options.go

@@ -0,0 +1,45 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package spanstore
+
+import (
+	"github.com/uber/jaeger/plugin/storage/cassandra/spanstore/dbmodel"
+)
+
+// Option is a function that sets some option on the writer.
+type Option func(c *Options)
+
+// Options control behavior of the writer.
+type Options struct {
+	tagFilter dbmodel.TagFilter
+}
+
+// TagFilter can be provided to filter any tags that should not be indexed.
+func TagFilter(tagFilter dbmodel.TagFilter) Option {
+	return func(o *Options) {
+		o.tagFilter = tagFilter
+	}
+}
+
+func applyOptions(opts ...Option) Options {
+	o := Options{}
+	for _, opt := range opts {
+		opt(&o)
+	}
+	if o.tagFilter == nil {
+		o.tagFilter = dbmodel.DefaultTagFilter
+	}
+	return o
+}

plugin/storage/cassandra/spanstore/writer_test.go

@@ -46,7 +46,7 @@ func withSpanWriter(writeCacheTTL time.Duration, fn func(w *spanWriterTest)) {
 		session:   session,
 		logger:    logger,
 		logBuffer: logBuffer,
-		writer:    NewSpanWriter(session, writeCacheTTL, metricsFactory, logger, nil),
+		writer:    NewSpanWriter(session, writeCacheTTL, metricsFactory, logger),
 	}
 	fn(w)
 }