Skip to content

Commit

Permalink
ENH: Adds support for multiple listeners (#35)
Browse files Browse the repository at this point in the history
  • Loading branch information
thomasjpfan authored Jul 3, 2018
1 parent 8437e05 commit e977d5b
Show file tree
Hide file tree
Showing 11 changed files with 265 additions and 57 deletions.
1 change: 1 addition & 0 deletions Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ ENV CERTS="" \
LISTENER_ADDRESS="" \
MODE="default" \
PROXY_INSTANCE_NAME="docker-flow" \
RELOAD_ATTEMPTS="5" \
RELOAD_INTERVAL="5000" REPEAT_RELOAD=false \
RECONFIGURE_ATTEMPTS="20" \
SEPARATOR="," \
Expand Down
1 change: 1 addition & 0 deletions Dockerfile.linux-arm
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ ENV CERTS="" \
LISTENER_ADDRESS="" \
MODE="default" \
PROXY_INSTANCE_NAME="docker-flow" \
RELOAD_ATTEMPTS="5" \
RELOAD_INTERVAL="5000" REPEAT_RELOAD=false \
RECONFIGURE_ATTEMPTS="20" \
SEPARATOR="," \
Expand Down
4 changes: 3 additions & 1 deletion actions/fetch.go
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
package actions

import (
"../proxy"
"encoding/json"
"fmt"
"net/http"
"strings"

"../proxy"
)

// Fetchable defines interface that fetches information from other sources
Expand Down Expand Up @@ -46,6 +47,7 @@ func (m *fetch) ReloadConfig(baseData BaseReconfigure, listenerAddr string) erro
if err = json.NewDecoder(resp.Body).Decode(&services); err != nil {
return err
}

needsReload := false
for _, s := range services {
proxyService := proxy.GetServiceFromMap(&s)
Expand Down
3 changes: 2 additions & 1 deletion args.go
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
package main

import (
"github.com/jessevdk/go-flags"
"os"

"github.com/jessevdk/go-flags"
)

type args struct{}
Expand Down
72 changes: 69 additions & 3 deletions args_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,11 @@ package main

import (
"fmt"
"github.com/stretchr/testify/suite"
"net/http"
"os"
"testing"

"github.com/stretchr/testify/suite"
)

type ArgsTestSuite struct {
Expand Down Expand Up @@ -49,6 +50,7 @@ func (s ArgsTestSuite) Test_Parse_ParsesServerLongArgs() {
for _, d := range data {
s.Equal(d.expected, *d.value)
}
s.Len(serverImpl.ListenerAddresses, 1)
}

func (s ArgsTestSuite) Test_Parse_ParsesServerShortArgs() {
Expand All @@ -69,12 +71,11 @@ func (s ArgsTestSuite) Test_Parse_ParsesServerShortArgs() {
for _, d := range data {
s.Equal(d.expected, *d.value)
}
s.Len(serverImpl.ListenerAddresses, 1)
}

func (s ArgsTestSuite) Test_Parse_ServerHasDefaultValues() {
os.Args = []string{"myProgram", "server"}
os.Unsetenv("IP")
os.Unsetenv("PORT")
data := []struct {
expected string
value *string
Expand All @@ -87,6 +88,7 @@ func (s ArgsTestSuite) Test_Parse_ServerHasDefaultValues() {
for _, d := range data {
s.Equal(d.expected, *d.value)
}
s.Len(serverImpl.ListenerAddresses, 1)
}

func (s ArgsTestSuite) Test_Parse_ServerDefaultsToEnvVars() {
Expand All @@ -103,10 +105,74 @@ func (s ArgsTestSuite) Test_Parse_ServerDefaultsToEnvVars() {
for _, d := range data {
os.Setenv(d.key, d.expected)
}
defer func() {
for _, d := range data {
os.Unsetenv(d.key)
}
}()

args{}.parse()
for _, d := range data {
s.Equal(d.expected, *d.value)
}

s.Len(serverImpl.ListenerAddresses, 1)
}

func (s ArgsTestSuite) Test_Parse_ParsesListnerAddressShortArgs() {
dataTable := []struct {
value []string
expected []string
}{
{[]string{"-l", "dfsl1", "-l", "dfsl2"}, []string{"dfsl1", "dfsl2"}},
{[]string{"-l", "dfsl1"}, []string{"dfsl1"}},
}

rootArgs := []string{"myProgram", "server"}
for _, data := range dataTable {
os.Args = append(rootArgs, data.value...)
args{}.parse()
s.Require().Equal(data.expected, serverImpl.ListenerAddresses)
}
}

func (s ArgsTestSuite) Test_Parse_ParsesListnerAddressLongArgs() {
dataTable := []struct {
value []string
expected []string
}{
{[]string{"--listener-address", "dfsl1", "--listener-address", "dfsl2"}, []string{"dfsl1", "dfsl2"}},
{[]string{"--listener-address", "dfsl1"}, []string{"dfsl1"}},
}

rootArgs := []string{"myProgram", "server"}
for _, data := range dataTable {
os.Args = append(rootArgs, data.value...)
args{}.parse()
s.Require().Equal(data.expected, serverImpl.ListenerAddresses)
}
}

func (s ArgsTestSuite) Test_Parse_ParsesListnerAddressEnvVars() {
os.Args = []string{"myProgram", "server"}
dataTable := []struct {
value string
expected []string
}{
{"dfsl1,dfsl2", []string{"dfsl1", "dfsl2"}},
{"dfsl1", []string{"dfsl1"}},
}

defer func() {
os.Unsetenv("LISTENER_ADDRESS")
}()

for _, data := range dataTable {
os.Setenv("LISTENER_ADDRESS", data.value)
args{}.parse()
s.Require().Equal(data.expected, serverImpl.ListenerAddresses)
}

}

// Suite
Expand Down
3 changes: 2 additions & 1 deletion docs/config.md
Original file line number Diff line number Diff line change
Expand Up @@ -31,9 +31,10 @@ The following environment variables can be used to configure the *Docker Flow Pr
|EXTRA_FRONTEND |Value will be added to the default `frontend` configuration. Multiple lines should be separated with comma (*,*).|
|EXTRA_GLOBAL |Value will be added to the default `global` configuration. Multiple lines should be separated with comma (*,*).|
|HTTPS_ONLY |If set to true, all requests to all services will be redirected to HTTPS.<br>**Example:** `true`<br>**Default Value:** `false`|
|LISTENER_ADDRESS |The address of the [Docker Flow: Swarm Listener](https://github.com/docker-flow/docker-flow-swarm-listener) used for automatic proxy configuration.<br>**Example:** `swarm-listener:8080`|
|LISTENER_ADDRESS |The address of the [Docker Flow: Swarm Listener](https://github.com/docker-flow/docker-flow-swarm-listener) used for automatic proxy configuration. Multiple values can be separated with comma (`,`). When set to multiple values, the proxy will query each address in order.<br>**Example:** `swarm-listener`|
PROXY_INSTANCE_NAME|The name of the proxy instance. Useful if multiple proxies are running inside a cluster.<br>**Default value:** `docker-flow`|
|RECONFIGURE_ATTEMPTS|The number of attempts the proxy will try to reconfigure itself before giving up and removing the offending service. The period between reconfigure attempts is 1 second.<br>**Example:** `15`<br>**Default value:** `20`|
|RELOAD_ATTEMPTS |The number of attempts the proxy will query a listener addresss during startup. Only used when LISTENER_ADDRESS is a comma seperated list of addresses.<br>**Default value:** `5`|
|RELOAD_INTERVAL |Defines the frequency (in milliseconds) between automatic config reloads from Swarm Listener.<br>**Default value:** `5000`|
|REPEAT_RELOAD |If set to `true`, the proxy will periodically reload the config, using `RELOAD_INTERVAL` as pause between iterations.<br>**Example:** `true`<br>**Default value:** `false`|
|RESOLVERS |The list of resolvers separated with comma (`,`). The `CHECK_RESOLVERS` environment variable must be set to `true`.<br>**Example:** `nameserver dns-0 4.4.2.1:53,nameserver dns-1 8.8.8.8:53`|
Expand Down
57 changes: 46 additions & 11 deletions server.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,16 +23,18 @@ type Server interface {
}

type serve struct {
IP string `short:"i" long:"ip" default:"0.0.0.0" env:"IP" description:"IP the server listens to."`
ListenerAddress string `short:"l" long:"listener-address" env:"LISTENER_ADDRESS" description:"The address of the Docker Flow: Swarm Listener. The address matches the name of the Swarm service (e.g. swarm-listener)"`
Port string `short:"p" long:"port" default:"8080" env:"PORT" description:"Port the server listens to."`
ServiceName string `short:"n" long:"service-name" default:"proxy" env:"SERVICE_NAME" description:"The name of the proxy service. It is used only when running in 'swarm' mode and must match the '--name' parameter used to launch the service."`
IP string `short:"i" long:"ip" default:"0.0.0.0" env:"IP" description:"IP the server listens to."`
ListenerAddresses []string `short:"l" long:"listener-address" env:"LISTENER_ADDRESS" env-delim:"," description:"The address of the Docker Flow: Swarm Listener. The address matches the name of the Swarm service (e.g. swarm-listener)" default:""`
Port string `short:"p" long:"port" default:"8080" env:"PORT" description:"Port the server listens to."`
ServiceName string `short:"n" long:"service-name" default:"proxy" env:"SERVICE_NAME" description:"The name of the proxy service. It is used only when running in 'swarm' mode and must match the '--name' parameter used to launch the service."`
SuccessfulInitReload bool
// TODO: Remove
actions.BaseReconfigure
}

var serverImpl = serve{}
var serverImpl = serve{
ListenerAddresses: []string{},
}
var cert server.Certer = server.NewCert("/certs")

// Execute runs the Web server.
Expand All @@ -46,7 +48,7 @@ func (m *serve) Execute(args []string) error {
address := fmt.Sprintf("%s:%s", m.IP, m.Port)
cert.Init()
var server2 = server.NewServer(
m.ListenerAddress,
m.ListenerAddresses,
m.Port,
m.ServiceName,
m.ConfigsPath,
Expand Down Expand Up @@ -77,12 +79,10 @@ func (m *serve) Execute(args []string) error {
}

func (m *serve) reconfigure(server server.Server) error {
lAddr := ""
if len(m.ListenerAddress) > 0 {
lAddr = fmt.Sprintf("http://%s:8080", m.ListenerAddress)
}
fetch := actions.NewFetch(m.BaseReconfigure)
if len(lAddr) > 0 {

if len(m.ListenerAddresses) == 1 && len(m.ListenerAddresses[0]) > 0 {
lAddr := fmt.Sprintf("http://%s:8080", m.ListenerAddresses[0])
go func() {
retryInterval := os.Getenv("RELOAD_INTERVAL")
interval, _ := time.ParseDuration(retryInterval + "ms")
Expand All @@ -105,6 +105,41 @@ func (m *serve) reconfigure(server server.Server) error {
}()
}

// Handlers Listener Addresses
if len(m.ListenerAddresses) > 1 {
reloadAttemptsStr := os.Getenv("RELOAD_ATTEMPTS")
retryInterval := os.Getenv("RELOAD_INTERVAL")
interval, _ := time.ParseDuration(retryInterval + "ms")
for _, addr := range m.ListenerAddresses {
if len(addr) == 0 {
continue
}
lAddr := fmt.Sprintf("http://%s:8080", addr)
go func(lAddr string) {
reloadAttempts, err := strconv.ParseInt(reloadAttemptsStr, 10, 64)
if err != nil {
reloadAttempts = 5
}
for range time.Tick(interval) {
if err := fetch.ReloadConfig(m.BaseReconfigure, lAddr); err != nil {
logPrintf(
"Error: Fetching config from swarm listener failed: %s. Will retry in %d seconds.",
err.Error(),
interval/time.Second,
)
} else {
m.SuccessfulInitReload = true
break
}
reloadAttempts = reloadAttempts - 1
if reloadAttempts <= 0 {
break
}
}
}(lAddr)
}
}

services := server.GetServicesFromEnvVars()

for _, service := range *services {
Expand Down
7 changes: 4 additions & 3 deletions server/cert_test.go
Original file line number Diff line number Diff line change
@@ -1,11 +1,8 @@
package server

import (
"../proxy"
"encoding/json"
"fmt"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/suite"
"io/ioutil"
"net"
"net/http"
Expand All @@ -14,6 +11,10 @@ import (
"path/filepath"
"strings"
"testing"

"../proxy"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/suite"
)

type CertTestSuite struct {
Expand Down
51 changes: 29 additions & 22 deletions server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -30,23 +30,23 @@ const (
)

type serve struct {
listenerAddress string
port string
serviceName string
configsPath string
templatesPath string
cert Certer
listenerAddresses []string
port string
serviceName string
configsPath string
templatesPath string
cert Certer
}

// NewServer returns instance of the Server with populated data
var NewServer = func(listenerAddr, port, serviceName, configsPath, templatesPath string, cert Certer) Server {
var NewServer = func(listenerAddr []string, port, serviceName, configsPath, templatesPath string, cert Certer) Server {
return &serve{
listenerAddress: listenerAddr,
port: port,
serviceName: serviceName,
configsPath: configsPath,
templatesPath: templatesPath,
cert: cert,
listenerAddresses: listenerAddr,
port: port,
serviceName: serviceName,
configsPath: configsPath,
templatesPath: templatesPath,
cert: cert,
}
}

Expand Down Expand Up @@ -143,22 +143,29 @@ func (m *serve) ReloadHandler(w http.ResponseWriter, req *http.Request) {
req.ParseForm()
params := new(reloadParams)
decoder.Decode(params, req.Form)
listenerAddr := ""
response := Response{
Status: "OK",
}
if params.FromListener {
listenerAddr = m.listenerAddress
}

//MW: I've reconstructed original behavior. BUT.
//shouldn't reload call ReloadServicesFromRegistry not just
//reload in else, if so ReloadClusterConfig & ReloadServicesFromRegistry
//could be enclosed in one method
if len(listenerAddr) > 0 {
fetch := actions.NewFetch(m.getBaseReconfigure())
if err := fetch.ReloadClusterConfig(listenerAddr); err != nil {
logPrintf("Error: ReloadClusterConfig failed: %s", err.Error())
m.writeInternalServerError(w, &Response{}, err.Error())
if params.FromListener {
errs := []string{}
for _, listenerAddr := range m.listenerAddresses {
if len(listenerAddr) == 0 {
continue
}
fetch := actions.NewFetch(m.getBaseReconfigure())
if err := fetch.ReloadClusterConfig(listenerAddr); err != nil {
errs = append(errs, err.Error())
logPrintf("Error: ReloadClusterConfig failed: %s", err.Error())
}
}
if len(errs) != 0 {
errMsg := strings.Join(errs, " ,")
m.writeInternalServerError(w, &Response{}, errMsg)
} else {
w.WriteHeader(http.StatusOK)
}
Expand Down
Loading

0 comments on commit e977d5b

Please sign in to comment.