Merge pull request #56 from Chrisss93/dynamicListenerAddress

Advertised address for dynamic listeners

Author: everesio

README.md

@@ -85,6 +85,7 @@ See:
           --debug-listen-address string                    Debug listen address (default "0.0.0.0:6060")
           --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-advertised-listener                    Advertised address for dynamic listeners (default "127.0.0.1")
           --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

cmd/kafka-proxy/server.go

@@ -83,6 +83,7 @@ func init() {
 func initFlags() {
 	// proxy
 	Server.Flags().StringVar(&c.Proxy.DefaultListenerIP, "default-listener-ip", "127.0.0.1", "Default listener IP")
+	Server.Flags().StringVar(&c.Proxy.DynamicAdvertisedListener, "dynamic-advertised-listener", "127.0.0.1", "Advertised address for dynamic listeners")
 	Server.Flags().StringArrayVar(&bootstrapServersMapping, "bootstrap-server-mapping", []string{}, "Mapping of Kafka bootstrap server address to local address (host:port,host:port(,advhost:advport))")
 	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")

config/config.go

@@ -49,17 +49,18 @@ type Config struct {
 		MsgFiledName   string
 	}
 	Proxy struct {
-		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
+		DefaultListenerIP         string
+		BootstrapServers          []ListenerConfig
+		ExternalServers           []ListenerConfig
+		DialAddressMappings       []DialAddressMapping
+		DisableDynamicListeners   bool
+		DynamicAdvertisedListener string
+		DynamicSequentialMinPort  int
+		RequestBufferSize         int
+		ResponseBufferSize        int
+		ListenerReadBufferSize    int // SO_RCVBUF
+		ListenerWriteBufferSize   int // SO_SNDBUF
+		ListenerKeepAlive         time.Duration
 
 		TLS struct {
 			Enable                   bool
@@ -251,6 +252,7 @@ func NewConfig() *Config {
 	c.Http.HealthPath = "/health"
 
 	c.Proxy.DefaultListenerIP = "127.0.0.1"
+	c.Proxy.DynamicAdvertisedListener = "127.0.0.1"
 	c.Proxy.DisableDynamicListeners = false
 	c.Proxy.RequestBufferSize = 4096
 	c.Proxy.ResponseBufferSize = 4096

proxy/proxy.go

@@ -17,6 +17,8 @@ type Listeners struct {
 	connSrc chan Conn
 	// listen IP for dynamically start
 	defaultListenerIP string
+	// advertised address for dynamic listeners
+	dynamicAdvertisedListener string
 	// socket TCP options
 	tcpConnOptions TCPConnOptions
 
@@ -32,6 +34,7 @@ type Listeners struct {
 func NewListeners(cfg *config.Config) (*Listeners, error) {
 
 	defaultListenerIP := cfg.Proxy.DefaultListenerIP
+	dynamicAdvertisedListener := cfg.Proxy.DynamicAdvertisedListener
 
 	tcpConnOptions := TCPConnOptions{
 		KeepAlive:       cfg.Proxy.ListenerKeepAlive,
@@ -61,13 +64,14 @@ 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,
-		dynamicSequentialMinPort: cfg.Proxy.DynamicSequentialMinPort,
+		defaultListenerIP:         defaultListenerIP,
+		dynamicAdvertisedListener: dynamicAdvertisedListener,
+		connSrc:                   make(chan Conn, 1),
+		brokerToListenerConfig:    brokerToListenerConfig,
+		tcpConnOptions:            tcpConnOptions,
+		listenFunc:                listenFunc,
+		disableDynamicListeners:   cfg.Proxy.DisableDynamicListeners,
+		dynamicSequentialMinPort:  cfg.Proxy.DynamicSequentialMinPort,
 	}, nil
 }
 
@@ -154,7 +158,8 @@ func (p *Listeners) ListenDynamicInstance(brokerAddress string) (string, int32,
 	}
 	port := l.Addr().(*net.TCPAddr).Port
 	address := net.JoinHostPort(p.defaultListenerIP, fmt.Sprint(port))
-	p.brokerToListenerConfig[brokerAddress] = config.ListenerConfig{BrokerAddress: brokerAddress, ListenerAddress: address, AdvertisedAddress: address}
+	advertisedAddress := net.JoinHostPort(p.dynamicAdvertisedListener, fmt.Sprint(port))
+	p.brokerToListenerConfig[brokerAddress] = config.ListenerConfig{BrokerAddress: brokerAddress, ListenerAddress: address, AdvertisedAddress: advertisedAddress}
 	return p.defaultListenerIP, int32(port), nil
 }