Merge 8d480db into 037875b

pkg/memory/config/config.go

@@ -0,0 +1,20 @@
+// Copyright (c) 2018 The Jaeger Authors.
+//
+// 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 config
+
+// Configuration describes the configuration properties needed to connect to a Cassandra cluster
+type Configuration struct {
+	MaxTraces int `yaml:"max-traces"`
+}

plugin/storage/memory/factory.go

@@ -15,6 +15,9 @@
 package memory
 
 import (
+	"flag"
+
+	"github.com/spf13/viper"
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
 
@@ -24,6 +27,7 @@ import (
 
 // Factory implements storage.Factory and creates storage components backed by memory store.
 type Factory struct {
+	options        Options
 	metricsFactory metrics.Factory
 	logger         *zap.Logger
 	store          *Store
@@ -34,10 +38,20 @@ func NewFactory() *Factory {
 	return &Factory{}
 }
 
+// AddFlags implements plugin.Configurable
+func (f *Factory) AddFlags(flagSet *flag.FlagSet) {
+	f.options.AddFlags(flagSet)
+}
+
+// InitFromViper implements plugin.Configurable
+func (f *Factory) InitFromViper(v *viper.Viper) {
+	f.options.InitFromViper(v)
+}
+
 // Initialize implements storage.Factory
 func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error {
 	f.metricsFactory, f.logger = metricsFactory, logger
-	f.store = NewStore()
+	f.store = WithConfiguration(&f.options.Configuration)
 	return nil
 }
 

plugin/storage/memory/factory_test.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
+	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/storage"
 )
 
@@ -38,3 +39,11 @@ func TestMemoryStorageFactory(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, f.store, depReader)
 }
+
+func TestWithConfiguration(t *testing.T) {
+	f := NewFactory()
+	v, command := config.Viperize(f.AddFlags)
+	command.ParseFlags([]string{"--memory.max-traces=100"})
+	f.InitFromViper(v)
+	assert.Equal(t, f.options.Configuration.MaxTraces, 100)
+}

plugin/storage/memory/memory.go

@@ -21,28 +21,38 @@ import (
 
 	"github.com/jaegertracing/jaeger/model"
 	"github.com/jaegertracing/jaeger/model/adjuster"
+	"github.com/jaegertracing/jaeger/pkg/memory/config"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
 var errTraceNotFound = errors.New("Trace was not found")
 
-// Store is an unbounded in-memory store of traces
+// Store is an in-memory store of traces
 type Store struct {
-	// TODO: make this a bounded memory store
 	sync.RWMutex
+	ids        []*model.TraceID
 	traces     map[model.TraceID]*model.Trace
 	services   map[string]struct{}
 	operations map[string]map[string]struct{}
 	deduper    adjuster.Adjuster
+	config     *config.Configuration
+	index      int
 }
 
-// NewStore creates an in-memory store
+// NewStore creates an unbounded in-memory store
 func NewStore() *Store {
+	return WithConfiguration(&config.Configuration{MaxTraces: 0})
+}
+
+// WithConfiguration creates a new in memory storage based on the given configuration
+func WithConfiguration(configuration *config.Configuration) *Store {
 	return &Store{
+		ids:        make([]*model.TraceID, configuration.MaxTraces),
 		traces:     map[model.TraceID]*model.Trace{},
 		services:   map[string]struct{}{},
 		operations: map[string]map[string]struct{}{},
 		deduper:    adjuster.SpanIDDeduper(),
+		config:     configuration,
 	}
 }
 
@@ -113,6 +123,22 @@ func (m *Store) WriteSpan(span *model.Span) error {
 	m.services[span.Process.ServiceName] = struct{}{}
 	if _, ok := m.traces[span.TraceID]; !ok {
 		m.traces[span.TraceID] = &model.Trace{}
+
+		// if we have a limit, let's cleanup the oldest traces
+		if m.config.MaxTraces > 0 {
+			// we only have to deal with this slice if we have a limit
+			m.index = (m.index + 1) % m.config.MaxTraces
+
+			// do we have an item already on this position? if so, we are overriding it,
+			// and we need to remove from the map
+			if m.ids[m.index] != nil {
+				delete(m.traces, *m.ids[m.index])
+			}
+
+			// update the ring with the trace id
+			m.ids[m.index] = &span.TraceID
+		}
+
 	}
 	m.traces[span.TraceID].Spans = append(m.traces[span.TraceID].Spans, span)
 

plugin/storage/memory/memory_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/jaegertracing/jaeger/model"
+	"github.com/jaegertracing/jaeger/pkg/memory/config"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
@@ -169,6 +170,35 @@ func TestStoreWriteSpan(t *testing.T) {
 	})
 }
 
+func TestStoreWithLimit(t *testing.T) {
+	maxTraces := 100
+	store := WithConfiguration(&config.Configuration{MaxTraces: maxTraces})
+
+	for i := 0; i < maxTraces*2; i++ {
+		id := &model.TraceID{High: 1, Low: uint64(i)}
+		err := store.WriteSpan(&model.Span{
+			TraceID: *id,
+			Process: &model.Process{
+				ServiceName: "TestStoreWithLimit",
+			},
+		})
+		assert.NoError(t, err)
+
+		err = store.WriteSpan(&model.Span{
+			TraceID: *id,
+			SpanID:  model.SpanID(i),
+			Process: &model.Process{
+				ServiceName: "TestStoreWithLimit",
+			},
+			OperationName: "childOperationName",
+		})
+		assert.NoError(t, err)
+	}
+
+	assert.Equal(t, maxTraces, len(store.traces))
+	assert.Equal(t, maxTraces, len(store.ids))
+}
+
 func TestStoreGetTraceSuccess(t *testing.T) {
 	withPopulatedMemoryStore(func(store *Store) {
 		trace, err := store.GetTrace(testingSpan.TraceID)

plugin/storage/memory/options.go

@@ -0,0 +1,40 @@
+// Copyright (c) 2018 The Jaeger Authors.
+//
+// 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 memory
+
+import (
+	"flag"
+
+	"github.com/spf13/viper"
+
+	"github.com/jaegertracing/jaeger/pkg/memory/config"
+)
+
+const limit = "memory.max-traces"
+
+// Options stores the configuration entries for this storage
+type Options struct {
+	Configuration config.Configuration
+}
+
+// AddFlags from this storage to the CLI
+func (opt *Options) AddFlags(flagSet *flag.FlagSet) {
+	flagSet.Int(limit, opt.Configuration.MaxTraces, "The maximum amount of traces to store in memory")
+}
+
+// InitFromViper initializes the options struct with values from Viper
+func (opt *Options) InitFromViper(v *viper.Viper) {
+	opt.Configuration.MaxTraces = v.GetInt(limit)
+}

plugin/storage/memory/options_test.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2018 The Jaeger Authors.
+//
+// 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 memory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/jaegertracing/jaeger/pkg/config"
+)
+
+func TestOptionsWithFlags(t *testing.T) {
+	opts := &Options{}
+	v, command := config.Viperize(opts.AddFlags)
+	command.ParseFlags([]string{"--memory.max-traces=100"})
+	opts.InitFromViper(v)
+
+	assert.Equal(t, 100, opts.Configuration.MaxTraces)
+}