Add ability to set length of Signaller processing queue (#87)

Co-authored-by: mqmr <me@mqmr.com>
Co-authored-by: Martin Helmich <m.helmich@mittwald.de>

cmd/kube-httpcache/internal/flags.go

@@ -36,6 +36,7 @@ type KubeHTTPProxyFlags struct {
 		MaxRetries         int
 		RetryBackoffString string
 		RetryBackoff       time.Duration
+		QueueLength        int
 	}
 	Admin struct {
 		Address string
@@ -81,6 +82,7 @@ func (f *KubeHTTPProxyFlags) Parse() error {
 	flag.IntVar(&f.Signaller.WorkersCount, "signaller-workers", 1, "number of workers to process requests")
 	flag.IntVar(&f.Signaller.MaxRetries, "signaller-retries", 5, "maximum number of attempts for signalling request")
 	flag.StringVar(&f.Signaller.RetryBackoffString, "signaller-backoff", "30s", "backoff for signalling request attempts")
+	flag.IntVar(&f.Signaller.QueueLength, "signaller-queue-length", 0, "length of signaller's processing queue")
 
 	flag.StringVar(&f.Admin.Address, "admin-addr", "127.0.0.1", "TCP address for the Varnish admin")
 	flag.IntVar(&f.Admin.Port, "admin-port", 6082, "TCP port for the Varnish admin")

cmd/kube-httpcache/main.go

@@ -86,6 +86,7 @@ func main() {
 			opts.Signaller.WorkersCount,
 			opts.Signaller.MaxRetries,
 			opts.Signaller.RetryBackoff,
+			opts.Signaller.QueueLength,
 		)
 		varnishSignallerErrors = varnishSignaller.GetErrors()
 

pkg/signaller/types.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/golang/glog"
 	"github.com/mittwald/kube-httpcache/pkg/watcher"
 )
 
@@ -32,7 +33,13 @@ func NewSignaller(
 	workersCount int,
 	maxRetries int,
 	retryBackoff time.Duration,
+	queueLength int,
 ) *Signaller {
+	if queueLength < 0 {
+		glog.Warning("signaller processing queue cannot have negative length, fall back to default value: 0")
+		queueLength = 0
+	}
+
 	return &Signaller{
 		Address:        address,
 		Port:           port,
@@ -41,7 +48,7 @@ func NewSignaller(
 		RetryBackoff:   retryBackoff,
 		EndpointScheme: "http",
 		endpoints:      watcher.NewEndpointConfig(),
-		signalQueue:    make(chan Signal),
+		signalQueue:    make(chan Signal, queueLength),
 		errors:         make(chan error),
 	}
 }