Merge pull request #38 from mikekap/deterministic-ports

Add the ability to have slightly more deterministic ports

Author: everesio

README.md

@@ -85,6 +85,7 @@ See:
           --default-listener-ip string                     Default listener IP (default "127.0.0.1")
           --dial-address-mapping stringArray               Mapping of target broker address to new one (host:port,host:port). The mapping is performed during connection establishment
           --dynamic-listeners-disable                      Disable dynamic listeners.
+          --dynamic-sequential-min-port int                If set to non-zero, makes the dynamic listener use a sequential port starting with this value rather than a random port every time.
           --external-server-mapping stringArray            Mapping of Kafka server address to external address (host:port,host:port). A listener for the external address is not started
           --forbidden-api-keys intSlice                    Forbidden Kafka request types. The restriction should prevent some Kafka operations e.g. 20 - DeleteTopics
           --forward-proxy string                           URL of the forward proxy. Supported schemas are socks5 and http
@@ -568,4 +569,4 @@ Use localhost:19092 as bootstrap servers
 ### Embedded third-party source code 
 
 * [Cloud SQL Proxy](https://github.com/GoogleCloudPlatform/cloudsql-proxy)
-* [Sarama](https://github.com/Shopify/sarama)
+* [Sarama](https://github.com/Shopify/sarama)

cmd/kafka-proxy/server.go

@@ -87,6 +87,7 @@ func initFlags() {
 	Server.Flags().StringArrayVar(&externalServersMapping, "external-server-mapping", []string{}, "Mapping of Kafka server address to external address (host:port,host:port). A listener for the external address is not started")
 	Server.Flags().StringArrayVar(&dialAddressMapping, "dial-address-mapping", []string{}, "Mapping of target broker address to new one (host:port,host:port). The mapping is performed during connection establishment")
 	Server.Flags().BoolVar(&c.Proxy.DisableDynamicListeners, "dynamic-listeners-disable", false, "Disable dynamic listeners.")
+	Server.Flags().IntVar(&c.Proxy.DynamicSequentialMinPort, "dynamic-sequential-min-port", 0, "If set to non-zero, makes the dynamic listener use a sequential port starting with this value rather than a random port every time.")
 
 	Server.Flags().IntVar(&c.Proxy.RequestBufferSize, "proxy-request-buffer-size", 4096, "Request buffer size pro tcp connection")
 	Server.Flags().IntVar(&c.Proxy.ResponseBufferSize, "proxy-response-buffer-size", 4096, "Response buffer size pro tcp connection")

config/config.go

@@ -47,16 +47,17 @@ type Config struct {
 		LevelFieldName string
 	}
 	Proxy struct {
-		DefaultListenerIP       string
-		BootstrapServers        []ListenerConfig
-		ExternalServers         []ListenerConfig
-		DialAddressMappings     []DialAddressMapping
-		DisableDynamicListeners bool
-		RequestBufferSize       int
-		ResponseBufferSize      int
-		ListenerReadBufferSize  int // SO_RCVBUF
-		ListenerWriteBufferSize int // SO_SNDBUF
-		ListenerKeepAlive       time.Duration
+		DefaultListenerIP        string
+		BootstrapServers         []ListenerConfig
+		ExternalServers          []ListenerConfig
+		DialAddressMappings      []DialAddressMapping
+		DisableDynamicListeners  bool
+		DynamicSequentialMinPort int
+		RequestBufferSize        int
+		ResponseBufferSize       int
+		ListenerReadBufferSize   int // SO_RCVBUF
+		ListenerWriteBufferSize  int // SO_SNDBUF
+		ListenerKeepAlive        time.Duration
 
 		TLS struct {
 			Enable                   bool

proxy/proxy.go

@@ -22,7 +22,8 @@ type Listeners struct {
 
 	listenFunc ListenFunc
 
-	disableDynamicListeners bool
+	disableDynamicListeners  bool
+	dynamicSequentialMinPort int
 
 	brokerToListenerConfig map[string]config.ListenerConfig
 	lock                   sync.RWMutex
@@ -60,12 +61,13 @@ func NewListeners(cfg *config.Config) (*Listeners, error) {
 	}
 
 	return &Listeners{
-		defaultListenerIP:       defaultListenerIP,
-		connSrc:                 make(chan Conn, 1),
-		brokerToListenerConfig:  brokerToListenerConfig,
-		tcpConnOptions:          tcpConnOptions,
-		listenFunc:              listenFunc,
-		disableDynamicListeners: cfg.Proxy.DisableDynamicListeners,
+		defaultListenerIP:        defaultListenerIP,
+		connSrc:                  make(chan Conn, 1),
+		brokerToListenerConfig:   brokerToListenerConfig,
+		tcpConnOptions:           tcpConnOptions,
+		listenFunc:               listenFunc,
+		disableDynamicListeners:  cfg.Proxy.DisableDynamicListeners,
+		dynamicSequentialMinPort: cfg.Proxy.DynamicSequentialMinPort,
 	}, nil
 }
 
@@ -140,7 +142,10 @@ func (p *Listeners) ListenDynamicInstance(brokerAddress string) (string, int32,
 		return v.AdvertisedAddress, 0, nil
 	}
 
-	defaultListenerAddress := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(0))
+	defaultListenerAddress := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(p.dynamicSequentialMinPort))
+	if p.dynamicSequentialMinPort != 0 {
+		p.dynamicSequentialMinPort += 1
+	}
 
 	cfg := config.ListenerConfig{ListenerAddress: defaultListenerAddress, BrokerAddress: brokerAddress}
 	l, err := listenInstance(p.connSrc, cfg, p.tcpConnOptions, p.listenFunc)