Allow upstream parameters update

Gopkg.toml

@@ -39,7 +39,7 @@
 
 [[constraint]]
   name = "github.com/nginxinc/nginx-plus-go-client"
-  version = "0.4.0"
+  version = "0.6.0"
 
 # https://github.com/Azure/go-autorest/issues/439#issuecomment-521732075
 [[override]]

cmd/sync/aws.go

@@ -67,6 +67,10 @@ func (client *AWSClient) GetUpstreams() []Upstream {
 			Port:         awsU.Port,
 			Kind:         awsU.Kind,
 			ScalingGroup: awsU.AutoscalingGroup,
+			MaxConns:     &awsU.MaxConns,
+			MaxFails:     &awsU.MaxFails,
+			FailTimeout:  awsU.FailTimeout,
+			SlowStart:    awsU.SlowStart,
 		}
 		upstreams = append(upstreams, u)
 	}
@@ -108,7 +112,7 @@ func parseAWSConfig(data []byte) (*awsConfig, error) {
 func (client *AWSClient) CheckIfScalingGroupExists(name string) (bool, error) {
 	params := &ec2.DescribeInstancesInput{
 		Filters: []*ec2.Filter{
-			&ec2.Filter{
+			{
 				Name: aws.String("tag:aws:autoscaling:groupName"),
 				Values: []*string{
 					aws.String(name),
@@ -129,7 +133,7 @@ func (client *AWSClient) CheckIfScalingGroupExists(name string) (bool, error) {
 func (client *AWSClient) GetPrivateIPsForScalingGroup(name string) ([]string, error) {
 	params := &ec2.DescribeInstancesInput{
 		Filters: []*ec2.Filter{
-			&ec2.Filter{
+			{
 				Name: aws.String("tag:aws:autoscaling:groupName"),
 				Values: []*string{
 					aws.String(name),
@@ -171,6 +175,10 @@ type awsUpstream struct {
 	AutoscalingGroup string `yaml:"autoscaling_group"`
 	Port             int
 	Kind             string
+	MaxConns         int    `yaml:"max_conns"`
+	MaxFails         int    `yaml:"max_fails"`
+	FailTimeout      string `yaml:"fail_timeout"`
+	SlowStart        string `yaml:"slow_start"`
 }
 
 func validateAWSConfig(cfg *awsConfig) error {
@@ -195,6 +203,18 @@ func validateAWSConfig(cfg *awsConfig) error {
 		if ups.Kind == "" || !(ups.Kind == "http" || ups.Kind == "stream") {
 			return fmt.Errorf(upstreamKindErrorMsgFormat, ups.Name)
 		}
+		if ups.MaxConns < 0 {
+			return fmt.Errorf(upstreamMaxConnsErrorMsgFmt, ups.MaxConns)
+		}
+		if ups.MaxFails < 0 {
+			return fmt.Errorf(upstreamMaxFailsErrorMsgFmt, ups.MaxFails)
+		}
+		if !isValidTime(ups.FailTimeout) {
+			return fmt.Errorf(upstreamFailTimeoutErrorMsgFmt, ups.FailTimeout)
+		}
+		if !isValidTime(ups.SlowStart) {
+			return fmt.Errorf(upstreamSlowStartErrorMsgFmt, ups.SlowStart)
+		}
 	}
 
 	return nil

cmd/sync/aws_test.go

@@ -51,6 +51,22 @@ func getInvalidAWSConfigInput() []*testInputAWS {
 	invalidUpstreamKindCfg.Upstreams[0].Kind = ""
 	input = append(input, &testInputAWS{invalidUpstreamKindCfg, "invalid kind of the upstream"})
 
+	invalidUpstreamMaxConnsCfg := getValidAWSConfig()
+	invalidUpstreamMaxConnsCfg.Upstreams[0].MaxConns = -10
+	input = append(input, &testInputAWS{invalidUpstreamMaxConnsCfg, "invalid max_conns of the upstream"})
+
+	invalidUpstreamMaxFailsCfg := getValidAWSConfig()
+	invalidUpstreamMaxFailsCfg.Upstreams[0].MaxFails = -10
+	input = append(input, &testInputAWS{invalidUpstreamMaxFailsCfg, "invalid max_fails of the upstream"})
+
+	invalidUpstreamFailTimeoutCfg := getValidAWSConfig()
+	invalidUpstreamFailTimeoutCfg.Upstreams[0].FailTimeout = "-10s"
+	input = append(input, &testInputAWS{invalidUpstreamFailTimeoutCfg, "invalid fail_timeout of the upstream"})
+
+	invalidUpstreamSlowStartCfg := getValidAWSConfig()
+	invalidUpstreamSlowStartCfg.Upstreams[0].SlowStart = "-10s"
+	input = append(input, &testInputAWS{invalidUpstreamSlowStartCfg, "invalid slow_start of the upstream"})
+
 	return input
 }
 

cmd/sync/azure.go

@@ -136,6 +136,10 @@ func (client *AzureClient) GetUpstreams() []Upstream {
 			Port:         azureU.Port,
 			Kind:         azureU.Kind,
 			ScalingGroup: azureU.VMScaleSet,
+			MaxConns:     &azureU.MaxConns,
+			MaxFails:     &azureU.MaxFails,
+			FailTimeout:  azureU.FailTimeout,
+			SlowStart:    azureU.SlowStart,
 		}
 		upstreams = append(upstreams, u)
 	}
@@ -149,10 +153,14 @@ type azureConfig struct {
 }
 
 type azureUpstream struct {
-	Name       string
-	VMScaleSet string `yaml:"virtual_machine_scale_set"`
-	Port       int
-	Kind       string
+	Name        string
+	VMScaleSet  string `yaml:"virtual_machine_scale_set"`
+	Port        int
+	Kind        string
+	MaxConns    int    `yaml:"max_conns"`
+	MaxFails    int    `yaml:"max_fails"`
+	FailTimeout string `yaml:"fail_timeout"`
+	SlowStart   string `yaml:"slow_start"`
 }
 
 func validateAzureConfig(cfg *azureConfig) error {
@@ -165,7 +173,7 @@ func validateAzureConfig(cfg *azureConfig) error {
 	}
 
 	if len(cfg.Upstreams) == 0 {
-		return fmt.Errorf("There is no upstreams found in the config file")
+		return fmt.Errorf("There are no upstreams found in the config file")
 	}
 
 	for _, ups := range cfg.Upstreams {
@@ -181,6 +189,18 @@ func validateAzureConfig(cfg *azureConfig) error {
 		if ups.Kind == "" || !(ups.Kind == "http" || ups.Kind == "stream") {
 			return fmt.Errorf(upstreamKindErrorMsgFormat, ups.Name)
 		}
+		if ups.MaxConns < 0 {
+			return fmt.Errorf(upstreamMaxConnsErrorMsgFmt, ups.MaxConns)
+		}
+		if ups.MaxFails < 0 {
+			return fmt.Errorf(upstreamMaxFailsErrorMsgFmt, ups.MaxFails)
+		}
+		if !isValidTime(ups.FailTimeout) {
+			return fmt.Errorf(upstreamFailTimeoutErrorMsgFmt, ups.FailTimeout)
+		}
+		if !isValidTime(ups.SlowStart) {
+			return fmt.Errorf(upstreamSlowStartErrorMsgFmt, ups.SlowStart)
+		}
 	}
 	return nil
 }

cmd/sync/azure_test.go

@@ -60,6 +60,22 @@ func getInvalidAzureConfigInput() []*testInputAzure {
 	invalidUpstreamKindCfg.Upstreams[0].Kind = ""
 	input = append(input, &testInputAzure{invalidUpstreamKindCfg, "invalid kind of the upstream"})
 
+	invalidUpstreamMaxConnsCfg := getValidAzureConfig()
+	invalidUpstreamMaxConnsCfg.Upstreams[0].MaxConns = -10
+	input = append(input, &testInputAzure{invalidUpstreamMaxConnsCfg, "invalid max_conns of the upstream"})
+
+	invalidUpstreamMaxFailsCfg := getValidAzureConfig()
+	invalidUpstreamMaxFailsCfg.Upstreams[0].MaxFails = -10
+	input = append(input, &testInputAzure{invalidUpstreamMaxFailsCfg, "invalid max_fails of the upstream"})
+
+	invalidUpstreamFailTimeoutCfg := getValidAzureConfig()
+	invalidUpstreamFailTimeoutCfg.Upstreams[0].FailTimeout = "-10s"
+	input = append(input, &testInputAzure{invalidUpstreamFailTimeoutCfg, "invalid fail_timeout of the upstream"})
+
+	invalidUpstreamSlowStartCfg := getValidAzureConfig()
+	invalidUpstreamSlowStartCfg.Upstreams[0].SlowStart = "-10s"
+	input = append(input, &testInputAzure{invalidUpstreamSlowStartCfg, "invalid slow_start of the upstream"})
+
 	return input
 }
 

cmd/sync/config.go

@@ -55,4 +55,8 @@ type Upstream struct {
 	Port         int
 	ScalingGroup string
 	Kind         string
+	MaxConns     *int
+	MaxFails     *int
+	FailTimeout  string
+	SlowStart    string
 }

cmd/sync/errormessages.go

@@ -8,3 +8,7 @@ const upstreamNameErrorMsg = "The mandatory field name is either empty or missin
 const upstreamErrorMsgFormat = "The mandatory field %v is either empty or missing for the upstream %v in the config file"
 const upstreamPortErrorMsgFormat = "The mandatory field port is either zero or missing for the upstream %v in the config file"
 const upstreamKindErrorMsgFormat = "The mandatory field kind is either not equal to http or tcp or missing for the upstream %v in the config file"
+const upstreamMaxConnsErrorMsgFmt = "The field max_conns has invalid value %v in the config file"
+const upstreamMaxFailsErrorMsgFmt = "The field max_fails has invalid value %v in the config file"
+const upstreamFailTimeoutErrorMsgFmt = "The field fail_timeout has invalid value %v in the config file"
+const upstreamSlowStartErrorMsgFmt = "The field slow_start has invalid value %v in the config file"

cmd/sync/main.go

@@ -113,38 +113,44 @@ func main() {
 				for _, ip := range ips {
 					backend := fmt.Sprintf("%v:%v", ip, upstream.Port)
 					upsServers = append(upsServers, nginx.UpstreamServer{
-						Server:   backend,
-						MaxFails: 1,
+						Server:      backend,
+						MaxConns:    upstream.MaxConns,
+						MaxFails:    upstream.MaxFails,
+						FailTimeout: upstream.FailTimeout,
+						SlowStart:   upstream.SlowStart,
 					})
 				}
 
-				added, removed, err := nginxClient.UpdateHTTPServers(upstream.Name, upsServers)
+				added, removed, updated, err := nginxClient.UpdateHTTPServers(upstream.Name, upsServers)
 				if err != nil {
 					log.Printf("Couldn't update HTTP servers in NGINX: %v", err)
 					continue
 				}
 
-				if len(added) > 0 || len(removed) > 0 {
-					log.Printf("Updated HTTP servers of %v; Added: %v, Removed: %v", upstream, added, removed)
+				if len(added) > 0 || len(removed) > 0 || len(updated) > 0 {
+					log.Printf("Updated HTTP servers of %v; Added: %+v, Removed: %+v, Updated: %+v", upstream, added, removed, updated)
 				}
 			} else {
 				var upsServers []nginx.StreamUpstreamServer
 				for _, ip := range ips {
 					backend := fmt.Sprintf("%v:%v", ip, upstream.Port)
 					upsServers = append(upsServers, nginx.StreamUpstreamServer{
-						Server:   backend,
-						MaxFails: 1,
+						Server:      backend,
+						MaxConns:    upstream.MaxConns,
+						MaxFails:    upstream.MaxFails,
+						FailTimeout: upstream.FailTimeout,
+						SlowStart:   upstream.SlowStart,
 					})
 				}
 
-				added, removed, err := nginxClient.UpdateStreamServers(upstream.Name, upsServers)
+				added, removed, updated, err := nginxClient.UpdateStreamServers(upstream.Name, upsServers)
 				if err != nil {
 					log.Printf("Couldn't update Steam servers in NGINX: %v", err)
 					continue
 				}
 
-				if len(added) > 0 || len(removed) > 0 {
-					log.Printf("Updated Stream servers of %v; Added: %v, Removed: %v", upstream, added, removed)
+				if len(added) > 0 || len(removed) > 0 || len(updated) > 0 {
+					log.Printf("Updated Stream servers of %v; Added: %+v, Removed: %+v, Updated: %+v", upstream, added, removed, updated)
 				}
 			}
 

cmd/sync/validation.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"regexp"
+	"strings"
+)
+
+// http://nginx.org/en/docs/syntax.html
+var validTimeSuffixes = []string{
+	"ms",
+	"s",
+	"m",
+	"h",
+	"d",
+	"w",
+	"M",
+	"y",
+}
+
+var durationEscaped = strings.Join(validTimeSuffixes, "|")
+var validNginxTime = regexp.MustCompile(`^([0-9]+([` + durationEscaped + `]?){0,1} *)+$`)
+
+func isValidTime(time string) bool {
+	if time == "" {
+		return true
+	}
+
+	time = strings.TrimSpace(time)
+
+	return validNginxTime.MatchString(time)
+}

cmd/sync/validation_test.go

@@ -0,0 +1,21 @@
+package main
+
+import "testing"
+
+func TestIsValidTime(t *testing.T) {
+	var testsWithValidInput = []string{"1", "1m10s", "11 11", "5m 30s", "1s", "100m", "5w", "15m", "11M", "3h", "100y", "600"}
+	var invalidInput = []string{"ss", "rM", "m0m", "s1s", "-5s", "1L"}
+
+	for _, test := range testsWithValidInput {
+		valid := isValidTime(test)
+		if !valid {
+			t.Errorf("isValidTime(%q) returned false for valid input.", test)
+		}
+	}
+	for _, test := range invalidInput {
+		valid := isValidTime(test)
+		if valid {
+			t.Errorf("isValidTime(%q) returned true for invalid input.", test)
+		}
+	}
+}