app: enable configuring logs queue size

app/log.go

@@ -16,6 +16,7 @@ import (
 	"github.com/globalsign/mgo/bson"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tsuru/config"
 	"github.com/tsuru/tsuru/api/shutdown"
 	"github.com/tsuru/tsuru/db"
 	"github.com/tsuru/tsuru/log"
@@ -334,12 +335,16 @@ type bulkProcessor struct {
 }
 
 func initBulkProcessor(maxWait time.Duration, bulkSize int, appName string) *bulkProcessor {
+	queueSize, err := config.GetInt("logs:queue-size")
+	if err != nil || queueSize == 0 {
+		queueSize = bulkQueueMaxSize
+	}
 	return &bulkProcessor{
 		appName:     appName,
 		maxWaitTime: maxWait,
 		bulkSize:    bulkSize,
 		finished:    make(chan struct{}),
-		ch:          make(chan *msgWithTS, bulkQueueMaxSize),
+		ch:          make(chan *msgWithTS, queueSize),
 		nextNotify:  time.NewTimer(0),
 	}
 }

app/log_test.go

@@ -397,3 +397,15 @@ loop:
 	}
 	dispatcher.Shutdown(context.Background())
 }
+
+func (s *S) TestBulkProcessorQueueSizeDefault(c *check.C) {
+	processor := initBulkProcessor(time.Second, 100, "")
+	c.Assert(cap(processor.ch), check.Equals, bulkQueueMaxSize)
+}
+
+func (s *S) TestBulkProcessorCustomQueueSize(c *check.C) {
+	config.Set("logs:queue-size", 10)
+	defer config.Unset("logs:queue-size")
+	processor := initBulkProcessor(time.Second, 100, "")
+	c.Assert(cap(processor.ch), check.Equals, 10)
+}