Add support for updating server parameters

Author: Vighneswar Rao Bojja

client/nginx.go

@@ -7,15 +7,30 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"reflect"
+	"strings"
 )
 
-// APIVersion is a version of NGINX Plus API.
-const APIVersion = 5
+const (
+	// APIVersion is a version of NGINX Plus API.
+	APIVersion = 5
 
-const pathNotFoundCode = "PathNotFound"
+	pathNotFoundCode  = "PathNotFound"
+	streamContext     = true
+	httpContext       = false
+	defaultServerPort = "80"
+)
 
-const streamContext = true
-const httpContext = false
+// Default values for servers in Upstreams.
+var (
+	defaultMaxConns    = 0
+	defaultMaxFails    = 1
+	defaultFailTimeout = "10s"
+	defaultSlowStart   = "0s"
+	defaultBackup      = false
+	defaultDown        = false
+	defaultWeight      = 1
+)
 
 // NginxClient lets you access NGINX Plus API.
 type NginxClient struct {
@@ -29,13 +44,13 @@ type versions []int
 type UpstreamServer struct {
 	ID          int    `json:"id,omitempty"`
 	Server      string `json:"server"`
-	MaxConns    int    `json:"max_conns"`
+	MaxConns    *int   `json:"max_conns,omitempty"`
 	MaxFails    *int   `json:"max_fails,omitempty"`
 	FailTimeout string `json:"fail_timeout,omitempty"`
 	SlowStart   string `json:"slow_start,omitempty"`
-	Route       string `json:"route"`
-	Backup      bool   `json:"backup"`
-	Down        bool   `json:"down"`
+	Route       string `json:"route,omitempty"`
+	Backup      *bool  `json:"backup,omitempty"`
+	Down        *bool  `json:"down,omitempty"`
 	Drain       bool   `json:"drain,omitempty"`
 	Weight      *int   `json:"weight,omitempty"`
 	Service     string `json:"service,omitempty"`
@@ -45,12 +60,12 @@ type UpstreamServer struct {
 type StreamUpstreamServer struct {
 	ID          int    `json:"id,omitempty"`
 	Server      string `json:"server"`
-	MaxConns    int    `json:"max_conns"`
+	MaxConns    *int   `json:"max_conns,omitempty"`
 	MaxFails    *int   `json:"max_fails,omitempty"`
 	FailTimeout string `json:"fail_timeout,omitempty"`
 	SlowStart   string `json:"slow_start,omitempty"`
-	Backup      bool   `json:"backup"`
-	Down        bool   `json:"down"`
+	Backup      *bool  `json:"backup,omitempty"`
+	Down        *bool  `json:"down,omitempty"`
 	Weight      *int   `json:"weight,omitempty"`
 	Service     string `json:"service,omitempty"`
 }
@@ -469,32 +484,93 @@ func (client *NginxClient) DeleteHTTPServer(upstream string, server string) erro
 // UpdateHTTPServers updates the servers of the upstream.
 // Servers that are in the slice, but don't exist in NGINX will be added to NGINX.
 // Servers that aren't in the slice, but exist in NGINX, will be removed from NGINX.
-func (client *NginxClient) UpdateHTTPServers(upstream string, servers []UpstreamServer) ([]UpstreamServer, []UpstreamServer, error) {
+// Servers that are in the slice and exist in NGINX, but have different parameters, will be updated.
+func (client *NginxClient) UpdateHTTPServers(upstream string, servers []UpstreamServer) (added []UpstreamServer, deleted []UpstreamServer, updated []UpstreamServer, err error) {
 	serversInNginx, err := client.GetHTTPServers(upstream)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
+		return nil, nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
+	}
+
+	// We assume port 80 if no port is set for servers.
+	for _, server := range servers {
+		server.Server = addPortToServer(server.Server)
 	}
 
-	toAdd, toDelete := determineUpdates(servers, serversInNginx)
+	toAdd, toDelete, toUpdate := determineUpdates(servers, serversInNginx)
 
 	for _, server := range toAdd {
 		err := client.AddHTTPServer(upstream, server)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
+			return nil, nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
 		}
 	}
 
 	for _, server := range toDelete {
 		err := client.DeleteHTTPServer(upstream, server.Server)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
+			return nil, nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
 		}
 	}
 
-	return toAdd, toDelete, nil
+	for _, server := range toUpdate {
+		err := client.UpdateHTTPServer(upstream, server)
+		if err != nil {
+			return nil, nil, nil, fmt.Errorf("failed to update servers of %v upstream: %v", upstream, err)
+		}
+	}
+
+	return toAdd, toDelete, toUpdate, nil
 }
 
-func determineUpdates(updatedServers []UpstreamServer, nginxServers []UpstreamServer) (toAdd []UpstreamServer, toRemove []UpstreamServer) {
+// haveSameParameters checks if a given server has the same parameters an NGINX server already present. Order matters
+func haveSameParameters(newServer UpstreamServer, serverNGX UpstreamServer) bool {
+	newServer.ID = serverNGX.ID
+
+	if serverNGX.MaxConns != nil && newServer.MaxConns == nil {
+		newServer.MaxConns = &defaultMaxConns
+	}
+
+	if serverNGX.MaxFails != nil && newServer.MaxFails == nil {
+		newServer.MaxFails = &defaultMaxFails
+	}
+
+	if serverNGX.FailTimeout != "" && newServer.FailTimeout == "" {
+		newServer.FailTimeout = defaultFailTimeout
+	}
+
+	if serverNGX.SlowStart != "" && newServer.SlowStart == "" {
+		newServer.SlowStart = defaultSlowStart
+	}
+
+	if serverNGX.Backup != nil && newServer.Backup == nil {
+		newServer.Backup = &defaultBackup
+	}
+
+	if serverNGX.Down != nil && newServer.Down == nil {
+		newServer.Down = &defaultDown
+	}
+
+	if serverNGX.Weight != nil && newServer.Weight == nil {
+		newServer.Weight = &defaultWeight
+	}
+
+	return reflect.DeepEqual(newServer, serverNGX)
+}
+
+func determineUpdates(updatedServers []UpstreamServer, nginxServers []UpstreamServer) (toAdd []UpstreamServer, toRemove []UpstreamServer, toUpdate []UpstreamServer) {
+	for _, server := range updatedServers {
+		updateFound := false
+		for _, serverNGX := range nginxServers {
+			if server.Server == serverNGX.Server && !haveSameParameters(server, serverNGX) {
+				updateFound = true
+				break
+			}
+		}
+		if updateFound {
+			toUpdate = append(toUpdate, server)
+		}
+	}
+
 	for _, server := range updatedServers {
 		found := false
 		for _, serverNGX := range nginxServers {
@@ -608,7 +684,7 @@ func (client *NginxClient) delete(path string, expectedStatusCode int) error {
 	return nil
 }
 
-func (client *NginxClient) patch(path string, input interface{}) error {
+func (client *NginxClient) patch(path string, input interface{}, expectedStatusCode int) error {
 	path = fmt.Sprintf("%v/%v/%v/", client.apiEndpoint, APIVersion, path)
 
 	jsonInput, err := json.Marshal(input)
@@ -627,10 +703,10 @@ func (client *NginxClient) patch(path string, input interface{}) error {
 	}
 	defer resp.Body.Close()
 
-	if resp.StatusCode != http.StatusNoContent {
+	if resp.StatusCode != expectedStatusCode {
 		return createResponseMismatchError(resp.Body).Wrap(fmt.Sprintf(
 			"failed to complete patch request: expected %v response, got %v",
-			http.StatusNoContent, resp.StatusCode))
+			expectedStatusCode, resp.StatusCode))
 	}
 	return nil
 }
@@ -655,6 +731,7 @@ func (client *NginxClient) GetStreamServers(upstream string) ([]StreamUpstreamSe
 
 // AddStreamServer adds the stream server to the upstream.
 func (client *NginxClient) AddStreamServer(upstream string, server StreamUpstreamServer) error {
+	server.Server = addPortToServer(server.Server)
 	id, err := client.getIDOfStreamServer(upstream, server.Server)
 	if err != nil {
 		return fmt.Errorf("failed to add %v stream server to %v upstream: %v", server.Server, upstream, err)
@@ -673,6 +750,7 @@ func (client *NginxClient) AddStreamServer(upstream string, server StreamUpstrea
 
 // DeleteStreamServer the server from the upstream.
 func (client *NginxClient) DeleteStreamServer(upstream string, server string) error {
+	server = addPortToServer(server)
 	id, err := client.getIDOfStreamServer(upstream, server)
 	if err != nil {
 		return fmt.Errorf("failed to remove %v stream server from  %v upstream: %v", server, upstream, err)
@@ -692,29 +770,42 @@ func (client *NginxClient) DeleteStreamServer(upstream string, server string) er
 // UpdateStreamServers updates the servers of the upstream.
 // Servers that are in the slice, but don't exist in NGINX will be added to NGINX.
 // Servers that aren't in the slice, but exist in NGINX, will be removed from NGINX.
-func (client *NginxClient) UpdateStreamServers(upstream string, servers []StreamUpstreamServer) ([]StreamUpstreamServer, []StreamUpstreamServer, error) {
+// Servers that are in the slice and exist in NGINX, but have different parameters, will be updated.
+func (client *NginxClient) UpdateStreamServers(upstream string, servers []StreamUpstreamServer) (added []StreamUpstreamServer, deleted []StreamUpstreamServer, updated []StreamUpstreamServer, err error) {
 	serversInNginx, err := client.GetStreamServers(upstream)
 	if err != nil {
-		return nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
+		return nil, nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
 	}
 
-	toAdd, toDelete := determineStreamUpdates(servers, serversInNginx)
+	// We assume port 80 if no port is set for servers.
+	for _, server := range servers {
+		server.Server = addPortToServer(server.Server)
+	}
+
+	toAdd, toDelete, toUpdate := determineStreamUpdates(servers, serversInNginx)
 
 	for _, server := range toAdd {
 		err := client.AddStreamServer(upstream, server)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
+			return nil, nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
 		}
 	}
 
 	for _, server := range toDelete {
 		err := client.DeleteStreamServer(upstream, server.Server)
 		if err != nil {
-			return nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
+			return nil, nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
+		}
+	}
+
+	for _, server := range toUpdate {
+		err := client.UpdateStreamServer(upstream, server)
+		if err != nil {
+			return nil, nil, nil, fmt.Errorf("failed to update stream servers of %v upstream: %v", upstream, err)
 		}
 	}
 
-	return toAdd, toDelete, nil
+	return toAdd, toDelete, toUpdate, nil
 }
 
 func (client *NginxClient) getIDOfStreamServer(upstream string, name string) (int, error) {
@@ -732,7 +823,54 @@ func (client *NginxClient) getIDOfStreamServer(upstream string, name string) (in
 	return -1, nil
 }
 
-func determineStreamUpdates(updatedServers []StreamUpstreamServer, nginxServers []StreamUpstreamServer) (toAdd []StreamUpstreamServer, toRemove []StreamUpstreamServer) {
+// haveStreamSameParameters checks if a given stream server has the same parameters an NGINX server already present. Order matters
+func haveStreamSameParameters(newServer StreamUpstreamServer, serverNGX StreamUpstreamServer) bool {
+	newServer.ID = serverNGX.ID
+	if serverNGX.MaxConns != nil && newServer.MaxConns == nil {
+		newServer.MaxConns = &defaultMaxConns
+	}
+
+	if serverNGX.MaxFails != nil && newServer.MaxFails == nil {
+		newServer.MaxFails = &defaultMaxFails
+	}
+
+	if serverNGX.FailTimeout != "" && newServer.FailTimeout == "" {
+		newServer.FailTimeout = defaultFailTimeout
+	}
+
+	if serverNGX.SlowStart != "" && newServer.SlowStart == "" {
+		newServer.SlowStart = defaultSlowStart
+	}
+
+	if serverNGX.Backup != nil && newServer.Backup == nil {
+		newServer.Backup = &defaultBackup
+	}
+
+	if serverNGX.Down != nil && newServer.Down == nil {
+		newServer.Down = &defaultDown
+	}
+
+	if serverNGX.Weight != nil && newServer.Weight == nil {
+		newServer.Weight = &defaultWeight
+	}
+
+	return reflect.DeepEqual(newServer, serverNGX)
+}
+
+func determineStreamUpdates(updatedServers []StreamUpstreamServer, nginxServers []StreamUpstreamServer) (toAdd []StreamUpstreamServer, toRemove []StreamUpstreamServer, toUpdate []StreamUpstreamServer) {
+	for _, server := range updatedServers {
+		updateFound := false
+		for _, serverNGX := range nginxServers {
+			if server.Server == serverNGX.Server && !haveStreamSameParameters(server, serverNGX) {
+				updateFound = true
+				break
+			}
+		}
+		if updateFound {
+			toUpdate = append(toUpdate, server)
+		}
+	}
+
 	for _, server := range updatedServers {
 		found := false
 		for _, serverNGX := range nginxServers {
@@ -1059,7 +1197,7 @@ func (client *NginxClient) modifyKeyValPair(zone string, key string, val string,
 
 	path := fmt.Sprintf("%v/keyvals/%v", base, zone)
 	input := KeyValPairs{key: val}
-	err := client.patch(path, &input)
+	err := client.patch(path, &input, http.StatusNoContent)
 	if err != nil {
 		return fmt.Errorf("failed to update key value pair for %v/%v zone: %v", base, zone, err)
 	}
@@ -1092,7 +1230,7 @@ func (client *NginxClient) deleteKeyValuePair(zone string, key string, stream bo
 	keyval[key] = nil
 
 	path := fmt.Sprintf("%v/keyvals/%v", base, zone)
-	err := client.patch(path, &keyval)
+	err := client.patch(path, &keyval, http.StatusNoContent)
 	if err != nil {
 		return fmt.Errorf("failed to remove key values pair for %v/%v zone: %v", base, zone, err)
 	}
@@ -1125,3 +1263,43 @@ func (client *NginxClient) deleteKeyValPairs(zone string, stream bool) error {
 	}
 	return nil
 }
+
+// UpdateHTTPServer updates the server of the upstream.
+func (client *NginxClient) UpdateHTTPServer(upstream string, server UpstreamServer) error {
+	path := fmt.Sprintf("http/upstreams/%v/servers/%v", upstream, server.ID)
+	server.ID = 0
+	err := client.patch(path, &server, http.StatusOK)
+	if err != nil {
+		return fmt.Errorf("failed to update %v server to %v upstream: %v", server.Server, upstream, err)
+	}
+
+	return nil
+}
+
+// UpdateStreamServer updates the stream server of the upstream.
+func (client *NginxClient) UpdateStreamServer(upstream string, server StreamUpstreamServer) error {
+	path := fmt.Sprintf("stream/upstreams/%v/servers/%v", upstream, server.ID)
+	server.ID = 0
+	err := client.patch(path, &server, http.StatusOK)
+	if err != nil {
+		return fmt.Errorf("failed to update %v stream server to %v upstream: %v", server.Server, upstream, err)
+	}
+
+	return nil
+}
+
+func addPortToServer(server string) string {
+	if len(strings.Split(server, ":")) == 2 {
+		return server
+	}
+
+	if len(strings.Split(server, "]:")) == 2 {
+		return server
+	}
+
+	if strings.HasPrefix(server, "unix:") {
+		return server
+	}
+
+	return fmt.Sprintf("%v:%v", server, defaultServerPort)
+}