Add batching to queued sender implementations (#361)

Adds batching (as a feature of queuing) to the current queued sender implementations.

Testing Done: unit tests

Author: Steven Karis

cmd/occollector/app/builder/processor_builder.go

@@ -32,6 +32,10 @@ const (
 	InvalidSenderType = "invalid"
 )
 
+const (
+	queuedExportersConfigKey = "queued-exporters"
+)
+
 // JaegerThriftTChannelSenderCfg holds configuration for Jaeger Thrift Tchannel sender
 type JaegerThriftTChannelSenderCfg struct {
 	CollectorHostPorts        []string      `mapstructure:"collector-host-ports"`
@@ -63,6 +67,30 @@ func NewJaegerThriftHTTPSenderCfg() *JaegerThriftHTTPSenderCfg {
 	return opts
 }
 
+// BatchingConfig contains configuration around the queueing batching.
+// It contains some advanced configurations, which should not be used
+// by a typical user, but are provided as advanced features to increase
+// scalability.
+type BatchingConfig struct {
+	// Enable marks batching as enabled or not
+	Enable bool `mapstructure:"enable"`
+	// Timeout sets the time after which a batch will be sent regardless of size
+	Timeout *time.Duration `mapstructure:"timeout,omitempty"`
+	// SendBatchSize is the size of a batch which after hit, will trigger it to be sent.
+	SendBatchSize *int `mapstructure:"send-batch-size,omitempty"`
+
+	// NumTickers sets the number of tickers to use to divide the work of looping
+	// over batch buckets. This is an advanced configuration option.
+	NumTickers int `mapstructure:"num-tickers,omitempty"`
+	// TickTime sets time interval at which the tickers tick. This is an advanced
+	// configuration option.
+	TickTime *time.Duration `mapstructure:"tick-time,omitempty"`
+	// RemoveAfterTicks is the number of ticks that must pass without a span arriving
+	// from a node after which the batcher for that node will be deleted. This is an
+	// advanved configuration option.
+	RemoveAfterTicks *int `mapstructure:"remove-after-ticks,omitempty"`
+}
+
 // QueuedSpanProcessorCfg holds configuration for the queued span processor
 type QueuedSpanProcessorCfg struct {
 	// Name is the friendly name of the processor
@@ -78,6 +106,8 @@ type QueuedSpanProcessorCfg struct {
 	// SenderType indicates the type of sender to instantiate
 	SenderType   SenderType `mapstructure:"sender-type"`
 	SenderConfig interface{}
+	// BatchingConfig sets config parameters related to batching
+	BatchingConfig BatchingConfig `mapstructure:"batching"`
 }
 
 // AttributesCfg holds configuration for attributes that can be added to all spans
@@ -144,10 +174,9 @@ func NewDefaultMultiSpanProcessorCfg() *MultiSpanProcessorCfg {
 
 // InitFromViper initializes MultiSpanProcessorCfg with properties from viper
 func (mOpts *MultiSpanProcessorCfg) InitFromViper(v *viper.Viper) *MultiSpanProcessorCfg {
-	const baseKey = "queued-exporters"
-	procsv := v.Sub(baseKey)
+	procsv := v.Sub(queuedExportersConfigKey)
 	if procsv != nil {
-		for procName := range v.GetStringMap(baseKey) {
+		for procName := range v.GetStringMap(queuedExportersConfigKey) {
 			procv := procsv.Sub(procName)
 			procOpts := NewDefaultQueuedSpanProcessorCfg()
 			procOpts.Name = procName

cmd/occollector/app/collector/processors.go

@@ -27,12 +27,13 @@ import (
 	"github.com/census-instrumentation/opencensus-service/cmd/occollector/app/sender"
 	"github.com/census-instrumentation/opencensus-service/exporter"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
+	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/nodebatcher"
+	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/queued"
 	"github.com/census-instrumentation/opencensus-service/internal/config"
 )
 
 func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []exporter.TraceExporter, []exporter.MetricsExporter) {
 	// TODO: (@pjanotti) this is slightly modified from agent but in the end duplication, need to consolidate style and visibility.
-
 	cfg := builder.GetConfigFile(v)
 	if cfg == "" {
 		logger.Info("No config file, exporters can be only configured via the file.")
@@ -63,7 +64,9 @@ func createExporters(v *viper.Viper, logger *zap.Logger) ([]func(), []exporter.T
 	return wrappedDoneFns, traceExporters, metricsExporters
 }
 
-func buildQueuedSpanProcessor(logger *zap.Logger, opts *builder.QueuedSpanProcessorCfg) (processor.SpanProcessor, error) {
+func buildQueuedSpanProcessor(
+	logger *zap.Logger, opts *builder.QueuedSpanProcessorCfg,
+) (closeFns []func(), queuedSpanProcessor processor.SpanProcessor, err error) {
 	logger.Info("Constructing queue processor with name", zap.String("name", opts.Name))
 
 	// build span batch sender from configured options
@@ -80,7 +83,7 @@ func buildQueuedSpanProcessor(logger *zap.Logger, opts *builder.QueuedSpanProces
 		tchreporter, err := tchrepbuilder.CreateReporter(metrics.NullFactory, logger)
 		if err != nil {
 			logger.Fatal("Cannot create tchannel reporter.", zap.Error(err))
-			return nil, err
+			return nil, nil, err
 		}
 		spanSender = sender.NewJaegerThriftTChannelSender(tchreporter, logger)
 	case builder.ThriftHTTPSenderType:
@@ -93,21 +96,53 @@ func buildQueuedSpanProcessor(logger *zap.Logger, opts *builder.QueuedSpanProces
 			logger,
 			sender.HTTPTimeout(thriftHTTPSenderOpts.Timeout),
 		)
-	default:
-		logger.Fatal("Unrecognized sender type configured")
+	}
+
+	if spanSender == nil {
+		logger.Fatal("Unrecognized sender type or no exporters configured", zap.String("SenderType", string(opts.SenderType)))
+	}
+
+	var batchingOptions []nodebatcher.Option
+	if opts.BatchingConfig.Enable {
+		cfg := opts.BatchingConfig
+		if cfg.Timeout != nil {
+			batchingOptions = append(batchingOptions, nodebatcher.WithTimeout(*cfg.Timeout))
+		}
+		if cfg.NumTickers > 0 {
+			batchingOptions = append(
+				batchingOptions, nodebatcher.WithNumTickers(cfg.NumTickers),
+			)
+		}
+		if cfg.TickTime != nil {
+			batchingOptions = append(
+				batchingOptions, nodebatcher.WithTickTime(*cfg.TickTime),
+			)
+		}
+		if cfg.SendBatchSize != nil {
+			batchingOptions = append(
+				batchingOptions, nodebatcher.WithSendBatchSize(*cfg.SendBatchSize),
+			)
+		}
+		if cfg.RemoveAfterTicks != nil {
+			batchingOptions = append(
+				batchingOptions, nodebatcher.WithRemoveAfterTicks(*cfg.RemoveAfterTicks),
+			)
+		}
 	}
 
 	// build queued span processor with underlying sender
-	queuedSpanProcessor := processor.NewQueuedSpanProcessor(
+	queuedSpanProcessor = queued.NewQueuedSpanProcessor(
 		spanSender,
-		processor.Options.WithLogger(logger),
-		processor.Options.WithName(opts.Name),
-		processor.Options.WithNumWorkers(opts.NumWorkers),
-		processor.Options.WithQueueSize(opts.QueueSize),
-		processor.Options.WithRetryOnProcessingFailures(opts.RetryOnFailure),
-		processor.Options.WithBackoffDelay(opts.BackoffDelay),
+		queued.Options.WithLogger(logger),
+		queued.Options.WithName(opts.Name),
+		queued.Options.WithNumWorkers(opts.NumWorkers),
+		queued.Options.WithQueueSize(opts.QueueSize),
+		queued.Options.WithRetryOnProcessingFailures(opts.RetryOnFailure),
+		queued.Options.WithBackoffDelay(opts.BackoffDelay),
+		queued.Options.WithBatching(opts.BatchingConfig.Enable),
+		queued.Options.WithBatchingOptions(batchingOptions...),
 	)
-	return queuedSpanProcessor, nil
+	return nil, queuedSpanProcessor, nil
 }
 
 func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor, []func()) {
@@ -136,12 +171,13 @@ func startProcessor(v *viper.Viper, logger *zap.Logger) (processor.SpanProcessor
 	multiProcessorCfg := builder.NewDefaultMultiSpanProcessorCfg().InitFromViper(v)
 	for _, queuedJaegerProcessorCfg := range multiProcessorCfg.Processors {
 		logger.Info("Queued Jaeger Sender Enabled")
-		queuedJaegerProcessor, err := buildQueuedSpanProcessor(logger, queuedJaegerProcessorCfg)
+		doneFns, queuedJaegerProcessor, err := buildQueuedSpanProcessor(logger, queuedJaegerProcessorCfg)
 		if err != nil {
 			logger.Error("Failed to build the queued span processor", zap.Error(err))
 			os.Exit(1)
 		}
 		spanProcessors = append(spanProcessors, queuedJaegerProcessor)
+		closeFns = append(closeFns, doneFns...)
 	}
 
 	if len(spanProcessors) == 0 {

cmd/occollector/app/collector/telemetry.go

@@ -27,6 +27,8 @@ import (
 
 	"github.com/census-instrumentation/opencensus-service/internal"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
+	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/nodebatcher"
+	"github.com/census-instrumentation/opencensus-service/internal/collector/processor/queued"
 	"github.com/census-instrumentation/opencensus-service/internal/collector/telemetry"
 )
 
@@ -54,7 +56,8 @@ func initTelemetry(asyncErrorChannel chan<- error, v *viper.Viper, logger *zap.L
 	port := v.GetInt(metricsPortCfg)
 
 	views := processor.MetricViews(level)
-	views = append(views, processor.QueuedProcessorMetricViews(level)...)
+	views = append(views, queued.MetricViews(level)...)
+	views = append(views, nodebatcher.MetricViews(level)...)
 	views = append(views, internal.AllViews...)
 	processMetricsViews := telemetry.NewProcessMetricsViews()
 	views = append(views, processMetricsViews.Views()...)

internal/collector/processor/idbatcher/id_batcher_test.go

@@ -68,10 +68,12 @@ func BenchmarkConcurrentEnqueue(b *testing.B) {
 	ticker := time.NewTicker(100 * time.Millisecond)
 	defer ticker.Stop()
 	var ticked int32
+	var received int32
 	go func() {
 		for range ticker.C {
-			batcher.CloseCurrentAndTakeFirstBatch()
+			batch, _ := batcher.CloseCurrentAndTakeFirstBatch()
 			atomic.AddInt32(&ticked, 1)
+			atomic.AddInt32(&received, int32(len(batch)))
 		}
 	}()
 
@@ -82,9 +84,6 @@ func BenchmarkConcurrentEnqueue(b *testing.B) {
 			batcher.AddToCurrentBatch(ids[0])
 		}
 	})
-
-	closedBatches := atomic.LoadInt32(&ticked)
-	b.Logf("Closed %d batches", closedBatches)
 }
 
 func concurrencyTest(t *testing.T, numBatches, newBatchesInitialCapacity, batchChannelSize uint64) {

internal/collector/processor/nodebatcher/metrics.go

@@ -0,0 +1,107 @@
+// Copyright 2019, OpenCensus 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 nodebatcher
+
+import (
+	"go.opencensus.io/stats"
+	"go.opencensus.io/stats/view"
+	"go.opencensus.io/tag"
+
+	"github.com/census-instrumentation/opencensus-service/internal/collector/processor"
+	"github.com/census-instrumentation/opencensus-service/internal/collector/telemetry"
+)
+
+var (
+	statBatchSize               = stats.Int64("batch_size", "Size of batches sent from the batcher (in span)", stats.UnitDimensionless)
+	statNodesAddedToBatches     = stats.Int64("nodes_added_to_batches", "Count of nodes that are being batched.", stats.UnitDimensionless)
+	statNodesRemovedFromBatches = stats.Int64("nodes_removed_from_batches", "Number of nodes that have been removed from batching.", stats.UnitDimensionless)
+
+	statBatchSizeTriggerSend = stats.Int64("batch_size_trigger_send", "Number of times the batch was sent due to a size trigger", stats.UnitDimensionless)
+	statTimeoutTriggerSend   = stats.Int64("timeout_trigger_send", "Number of times the batch was sent due to a timeout trigger", stats.UnitDimensionless)
+	statBatchOnDeadNode      = stats.Int64("removed_node_send", "Number of times the batch was sent due to spans being added for a no longer active node", stats.UnitDimensionless)
+)
+
+// MetricViews returns the metrics views related to batching
+func MetricViews(level telemetry.Level) []*view.View {
+	if level == telemetry.None {
+		return nil
+	}
+
+	tagKeys := processor.MetricTagKeys(level)
+	if tagKeys == nil {
+		return nil
+	}
+
+	exporterTagKeys := []tag.Key{processor.TagExporterNameKey}
+
+	batchSizeAggregation := view.Distribution(10, 25, 50, 75, 100, 250, 500, 750, 1000, 2000, 3000, 4000, 5000, 10000, 20000, 30000, 50000, 100000)
+
+	batchSizeView := &view.View{
+		Name:        statBatchSize.Name(),
+		Measure:     statBatchSize,
+		Description: statBatchSize.Description(),
+		TagKeys:     exporterTagKeys,
+		Aggregation: batchSizeAggregation,
+	}
+
+	nodesAddedToBatchesView := &view.View{
+		Name:        statNodesAddedToBatches.Name(),
+		Measure:     statNodesAddedToBatches,
+		Description: statNodesAddedToBatches.Description(),
+		TagKeys:     exporterTagKeys,
+		Aggregation: view.Count(),
+	}
+
+	nodesRemovedFromBatchesView := &view.View{
+		Name:        statNodesRemovedFromBatches.Name(),
+		Measure:     statNodesRemovedFromBatches,
+		Description: statNodesRemovedFromBatches.Description(),
+		TagKeys:     exporterTagKeys,
+		Aggregation: view.Count(),
+	}
+
+	countBatchSizeTriggerSendView := &view.View{
+		Name:        statBatchSizeTriggerSend.Name(),
+		Measure:     statBatchSizeTriggerSend,
+		Description: statBatchSizeTriggerSend.Description(),
+		TagKeys:     tagKeys,
+		Aggregation: view.Count(),
+	}
+
+	countTimeoutTriggerSendView := &view.View{
+		Name:        statTimeoutTriggerSend.Name(),
+		Measure:     statTimeoutTriggerSend,
+		Description: statTimeoutTriggerSend.Description(),
+		TagKeys:     tagKeys,
+		Aggregation: view.Count(),
+	}
+
+	countBatchOnDeadNode := &view.View{
+		Name:        statBatchOnDeadNode.Name(),
+		Measure:     statBatchOnDeadNode,
+		Description: statBatchOnDeadNode.Description(),
+		TagKeys:     tagKeys,
+		Aggregation: view.Count(),
+	}
+
+	return []*view.View{
+		batchSizeView,
+		nodesAddedToBatchesView,
+		nodesRemovedFromBatchesView,
+		countBatchSizeTriggerSendView,
+		countTimeoutTriggerSendView,
+		countBatchOnDeadNode,
+	}
+}