Add cassandra tag filter (#442)

* Add cassandra tag filter

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

Author: black-adder

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

@@ -0,0 +1,32 @@
+// 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 dbmodel
+
+import "github.com/uber/jaeger/model"
+
+// LogFieldsFilter filters all span.Logs.Fields.
+type LogFieldsFilter struct {
+	tagFilterImpl
+}
+
+// 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{}
+}

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

@@ -0,0 +1,40 @@
+// 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 dbmodel
+
+import (
+	"testing"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+)
+
+var filter TagFilter = &LogFieldsFilter{} // Check API compliance
+
+func TestFilterLogTags(t *testing.T) {
+	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)...)
+	}
+	if !assert.EqualValues(t, expectedTags, filteredTags) {
+		for _, diff := range pretty.Diff(expectedTags, filteredTags) {
+			t.Log(diff)
+		}
+	}
+}

plugin/storage/cassandra/spanstore/dbmodel/tag_filter.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 dbmodel
+
+import (
+	"github.com/uber/jaeger/model"
+)
+
+// 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.
+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 (f tagFilterImpl) FilterLogFields(logFields model.KeyValues) model.KeyValues {
+	return logFields
+}

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

@@ -0,0 +1,37 @@
+// 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 dbmodel
+
+import (
+	"testing"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDefaultTagFilter(t *testing.T) {
+	span := getTestJaegerSpan()
+	expectedTags := append(append(someTags, someTags...), someTags...)
+	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

@@ -16,13 +16,12 @@ package dbmodel
 
 import "github.com/uber/jaeger/model"
 
-// GetAllUniqueTags creates a list of all unique tags found in a span and process
-func GetAllUniqueTags(span *model.Span) []TagInsertion {
-	process := span.Process
-	allTags := span.Tags
-	allTags = append(allTags, process.Tags...)
+// GetAllUniqueTags creates a list of all unique tags from a set of filtered tags.
+func GetAllUniqueTags(span *model.Span, tagFilter TagFilter) []TagInsertion {
+	allTags := append(model.KeyValues{}, tagFilter.FilterProcessTags(span.Process.Tags)...)
+	allTags = append(allTags, tagFilter.FilterTags(span.Tags)...)
 	for _, log := range span.Logs {
-		allTags = append(allTags, log.Fields...)
+		allTags = append(allTags, tagFilter.FilterLogFields(log.Fields)...)
 	}
 	allTags.Sort()
 	uniqueTags := make([]TagInsertion, 0, len(allTags))
@@ -34,7 +33,7 @@ func GetAllUniqueTags(span *model.Span) []TagInsertion {
 			continue // skip identical tags
 		}
 		uniqueTags = append(uniqueTags, TagInsertion{
-			ServiceName: process.ServiceName,
+			ServiceName: span.Process.ServiceName,
 			TagKey:      allTags[i].Key,
 			TagValue:    allTags[i].AsString(),
 		})

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

@@ -23,7 +23,7 @@ import (
 
 func TestGetUniqueTags(t *testing.T) {
 	expectedTags := getTestUniqueTags()
-	uniqueTags := GetAllUniqueTags(getTestJaegerSpan())
+	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,6 +86,7 @@ type SpanWriter struct {
 	logger               *zap.Logger
 	tagIndexSkipped      metrics.Counter
 	bucketCounter        uint32
+	tagFilter            dbmodel.TagFilter
 }
 
 // NewSpanWriter returns a SpanWriter
@@ -94,10 +95,12 @@ func NewSpanWriter(
 	writeCacheTTL time.Duration,
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
+	options ...Option,
 ) *SpanWriter {
 	serviceNamesStorage := NewServiceNamesStorage(session, writeCacheTTL, metricsFactory, logger)
 	operationNamesStorage := NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger)
 	tagIndexSkipped := metricsFactory.Counter("tagIndexSkipped", nil)
+	opts := applyOptions(options...)
 	return &SpanWriter{
 		session:              session,
 		serviceNamesWriter:   serviceNamesStorage.Write,
@@ -111,6 +114,7 @@ func NewSpanWriter(
 		},
 		logger:          logger,
 		tagIndexSkipped: tagIndexSkipped,
+		tagFilter:       opts.tagFilter,
 	}
 }
 
@@ -160,7 +164,7 @@ func (s *SpanWriter) WriteSpan(span *model.Span) error {
 }
 
 func (s *SpanWriter) indexByTags(span *model.Span, ds *dbmodel.Span) error {
-	for _, v := range dbmodel.GetAllUniqueTags(span) {
+	for _, v := range dbmodel.GetAllUniqueTags(span, s.tagFilter) {
 		// we should introduce retries or just ignore failures imo, retrying each individual tag insertion might be better
 		// we should consider bucketing.
 		if s.shouldIndexTag(v) {

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_options_test.go

@@ -0,0 +1,28 @@
+// 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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/uber/jaeger/plugin/storage/cassandra/spanstore/dbmodel"
+)
+
+func TestWriterOpetions(t *testing.T) {
+	opts := applyOptions(TagFilter(dbmodel.DefaultTagFilter))
+	assert.Equal(t, dbmodel.DefaultTagFilter, opts.tagFilter)
+}