Skip to content

Commit f625289

Browse files
dylan-waydylan-way
committed
Update to reduce unnecessary API calls and sanitize input
Update UpdateHTTPServers and UpdateStreamServers: - No longer make extra GET requests for each PUT and DELETE request. - Removes identical duplicate servers. - Returns errors for duplicate servers with different parameters.
1 parent eb3d45d commit f625289

File tree

3 files changed

+642
-184
lines changed

3 files changed

+642
-184
lines changed

‎client/nginx.go‎

Lines changed: 156 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,7 @@ var (
4848
ErrServerExists = errors.New("server already exists")
4949
ErrNotSupported = errors.New("not supported")
5050
ErrInvalidTimeout = errors.New("invalid timeout")
51+
ErrParameterMismatch = errors.New("encountered duplicate server with different parameters")
5152
ErrPlusVersionNotFound = errors.New("plus version not found in the input string")
5253
)
5354

@@ -775,9 +776,13 @@ func (client *NginxClient) AddHTTPServer(ctx context.Context, upstream string, s
775776
if id != -1 {
776777
return fmt.Errorf("failed to add %v server to %v upstream: %w", server.Server, upstream, ErrServerExists)
777778
}
779+
err = client.addHTTPServer(ctx, upstream, server)
780+
return err
781+
}
778782

783+
func (client *NginxClient) addHTTPServer(ctx context.Context, upstream string, server UpstreamServer) error {
779784
path := fmt.Sprintf("http/upstreams/%v/servers/", upstream)
780-
err = client.post(ctx, path, &server)
785+
err := client.post(ctx, path, &server)
781786
if err != nil {
782787
return fmt.Errorf("failed to add %v server to %v upstream: %w", server.Server, upstream, err)
783788
}
@@ -794,9 +799,13 @@ func (client *NginxClient) DeleteHTTPServer(ctx context.Context, upstream string
794799
if id == -1 {
795800
return fmt.Errorf("failed to remove %v server from %v upstream: %w", server, upstream, ErrServerNotFound)
796801
}
802+
err = client.deleteHTTPServer(ctx, upstream, server, id)
803+
return err
804+
}
797805

798-
path := fmt.Sprintf("http/upstreams/%v/servers/%v", upstream, id)
799-
err = client.delete(ctx, path, http.StatusOK)
806+
func (client *NginxClient) deleteHTTPServer(ctx context.Context, upstream, server string, serverID int) error {
807+
path := fmt.Sprintf("http/upstreams/%v/servers/%v", upstream, serverID)
808+
err := client.delete(ctx, path, http.StatusOK)
800809
if err != nil {
801810
return fmt.Errorf("failed to remove %v server from %v upstream: %w", server, upstream, err)
802811
}
@@ -809,6 +818,8 @@ func (client *NginxClient) DeleteHTTPServer(ctx context.Context, upstream string
809818
// Servers that aren't in the slice, but exist in NGINX, will be removed from NGINX.
810819
// Servers that are in the slice and exist in NGINX, but have different parameters, will be updated.
811820
// The client will attempt to update all servers, returning all the errors that occurred.
821+
// If there are duplicate servers with equivalent parameters, the duplicates will be ignored.
822+
// If there are duplicate servers with different parameters, those server entries will be ignored and an error returned.
812823
func (client *NginxClient) UpdateHTTPServers(ctx context.Context, upstream string, servers []UpstreamServer) (added []UpstreamServer, deleted []UpstreamServer, updated []UpstreamServer, err error) {
813824
serversInNginx, err := client.GetHTTPServers(ctx, upstream)
814825
if err != nil {
@@ -822,10 +833,12 @@ func (client *NginxClient) UpdateHTTPServers(ctx context.Context, upstream strin
822833
formattedServers = append(formattedServers, server)
823834
}
824835

836+
formattedServers, err = deduplicateServers(upstream, formattedServers)
837+
825838
toAdd, toDelete, toUpdate := determineUpdates(formattedServers, serversInNginx)
826839

827840
for _, server := range toAdd {
828-
addErr := client.AddHTTPServer(ctx, upstream, server)
841+
addErr := client.addHTTPServer(ctx, upstream, server)
829842
if addErr != nil {
830843
err = errors.Join(err, addErr)
831844
continue
@@ -834,7 +847,7 @@ func (client *NginxClient) UpdateHTTPServers(ctx context.Context, upstream strin
834847
}
835848

836849
for _, server := range toDelete {
837-
deleteErr := client.DeleteHTTPServer(ctx, upstream, server.Server)
850+
deleteErr := client.deleteHTTPServer(ctx, upstream, server.Server, server.ID)
838851
if deleteErr != nil {
839852
err = errors.Join(err, deleteErr)
840853
continue
@@ -858,46 +871,82 @@ func (client *NginxClient) UpdateHTTPServers(ctx context.Context, upstream strin
858871
return added, deleted, updated, err
859872
}
860873

861-
// haveSameParameters checks if a given server has the same parameters as a server already present in NGINX. Order matters.
862-
func haveSameParameters(newServer UpstreamServer, serverNGX UpstreamServer) bool {
863-
newServer.ID = serverNGX.ID
874+
func deduplicateServers(upstream string, servers []UpstreamServer) ([]UpstreamServer, error) {
875+
type serverCheck struct {
876+
server UpstreamServer
877+
valid bool
878+
}
864879

865-
if serverNGX.MaxConns != nil && newServer.MaxConns == nil {
866-
newServer.MaxConns = &defaultMaxConns
880+
serverMap := make(map[string]*serverCheck, len(servers))
881+
var err error
882+
for _, server := range servers {
883+
if prev, ok := serverMap[server.Server]; ok {
884+
if !prev.valid {
885+
continue
886+
}
887+
if !server.hasSameParametersAs(prev.server) {
888+
prev.valid = false
889+
err = errors.Join(err, fmt.Errorf(
890+
"failed to update %s server to %s upstream: %w",
891+
server.Server, upstream, ErrParameterMismatch))
892+
}
893+
continue
894+
}
895+
serverMap[server.Server] = &serverCheck{server, true}
867896
}
897+
retServers := make([]UpstreamServer, 0, len(serverMap))
898+
for _, server := range servers {
899+
if check, ok := serverMap[server.Server]; ok && check.valid {
900+
retServers = append(retServers, server)
901+
delete(serverMap, server.Server)
902+
}
903+
}
904+
return retServers, err
905+
}
868906

869-
if serverNGX.MaxFails != nil && newServer.MaxFails == nil {
870-
newServer.MaxFails = &defaultMaxFails
907+
// hasSameParametersAs checks if a given server has the same parameters.
908+
func (s UpstreamServer) hasSameParametersAs(compareServer UpstreamServer) bool {
909+
s.ID = compareServer.ID
910+
s.applyDefaults()
911+
compareServer.applyDefaults()
912+
return reflect.DeepEqual(s, compareServer)
913+
}
914+
915+
func (s *UpstreamServer) applyDefaults() {
916+
if s.MaxConns == nil {
917+
s.MaxConns = &defaultMaxConns
871918
}
872919

873-
if serverNGX.FailTimeout != "" && newServer.FailTimeout == "" {
874-
newServer.FailTimeout = defaultFailTimeout
920+
if s.MaxFails == nil {
921+
s.MaxFails = &defaultMaxFails
875922
}
876923

877-
if serverNGX.SlowStart != "" && newServer.SlowStart == "" {
878-
newServer.SlowStart = defaultSlowStart
924+
if s.FailTimeout == "" {
925+
s.FailTimeout = defaultFailTimeout
879926
}
880927

881-
if serverNGX.Backup != nil && newServer.Backup == nil {
882-
newServer.Backup = &defaultBackup
928+
if s.SlowStart == "" {
929+
s.SlowStart = defaultSlowStart
883930
}
884931

885-
if serverNGX.Down != nil && newServer.Down == nil {
886-
newServer.Down = &defaultDown
932+
if s.Backup == nil {
933+
s.Backup = &defaultBackup
887934
}
888935

889-
if serverNGX.Weight != nil && newServer.Weight == nil {
890-
newServer.Weight = &defaultWeight
936+
if s.Down == nil {
937+
s.Down = &defaultDown
891938
}
892939

893-
return reflect.DeepEqual(newServer, serverNGX)
940+
if s.Weight == nil {
941+
s.Weight = &defaultWeight
942+
}
894943
}
895944

896945
func determineUpdates(updatedServers []UpstreamServer, nginxServers []UpstreamServer) (toAdd []UpstreamServer, toRemove []UpstreamServer, toUpdate []UpstreamServer) {
897946
for _, server := range updatedServers {
898947
updateFound := false
899948
for _, serverNGX := range nginxServers {
900-
if server.Server == serverNGX.Server && !haveSameParameters(server, serverNGX) {
949+
if server.Server == serverNGX.Server && !server.hasSameParametersAs(serverNGX) {
901950
server.ID = serverNGX.ID
902951
updateFound = true
903952
break
@@ -1089,9 +1138,13 @@ func (client *NginxClient) AddStreamServer(ctx context.Context, upstream string,
10891138
if id != -1 {
10901139
return fmt.Errorf("failed to add %v stream server to %v upstream: %w", server.Server, upstream, ErrServerExists)
10911140
}
1141+
err = client.addStreamServer(ctx, upstream, server)
1142+
return err
1143+
}
10921144

1145+
func (client *NginxClient) addStreamServer(ctx context.Context, upstream string, server StreamUpstreamServer) error {
10931146
path := fmt.Sprintf("stream/upstreams/%v/servers/", upstream)
1094-
err = client.post(ctx, path, &server)
1147+
err := client.post(ctx, path, &server)
10951148
if err != nil {
10961149
return fmt.Errorf("failed to add %v stream server to %v upstream: %w", server.Server, upstream, err)
10971150
}
@@ -1107,9 +1160,13 @@ func (client *NginxClient) DeleteStreamServer(ctx context.Context, upstream stri
11071160
if id == -1 {
11081161
return fmt.Errorf("failed to remove %v stream server from %v upstream: %w", server, upstream, ErrServerNotFound)
11091162
}
1163+
err = client.deleteStreamServer(ctx, upstream, server, id)
1164+
return err
1165+
}
11101166

1111-
path := fmt.Sprintf("stream/upstreams/%v/servers/%v", upstream, id)
1112-
err = client.delete(ctx, path, http.StatusOK)
1167+
func (client *NginxClient) deleteStreamServer(ctx context.Context, upstream, server string, serverID int) error {
1168+
path := fmt.Sprintf("stream/upstreams/%v/servers/%v", upstream, serverID)
1169+
err := client.delete(ctx, path, http.StatusOK)
11131170
if err != nil {
11141171
return fmt.Errorf("failed to remove %v stream server from %v upstream: %w", server, upstream, err)
11151172
}
@@ -1121,6 +1178,8 @@ func (client *NginxClient) DeleteStreamServer(ctx context.Context, upstream stri
11211178
// Servers that aren't in the slice, but exist in NGINX, will be removed from NGINX.
11221179
// Servers that are in the slice and exist in NGINX, but have different parameters, will be updated.
11231180
// The client will attempt to update all servers, returning all the errors that occurred.
1181+
// If there are duplicate servers with equivalent parameters, the duplicates will be ignored.
1182+
// If there are duplicate servers with different parameters, those server entries will be ignored and an error returned.
11241183
func (client *NginxClient) UpdateStreamServers(ctx context.Context, upstream string, servers []StreamUpstreamServer) (added []StreamUpstreamServer, deleted []StreamUpstreamServer, updated []StreamUpstreamServer, err error) {
11251184
serversInNginx, err := client.GetStreamServers(ctx, upstream)
11261185
if err != nil {
@@ -1133,10 +1192,12 @@ func (client *NginxClient) UpdateStreamServers(ctx context.Context, upstream str
11331192
formattedServers = append(formattedServers, server)
11341193
}
11351194

1195+
formattedServers, err = deduplicateStreamServers(upstream, formattedServers)
1196+
11361197
toAdd, toDelete, toUpdate := determineStreamUpdates(formattedServers, serversInNginx)
11371198

11381199
for _, server := range toAdd {
1139-
addErr := client.AddStreamServer(ctx, upstream, server)
1200+
addErr := client.addStreamServer(ctx, upstream, server)
11401201
if addErr != nil {
11411202
err = errors.Join(err, addErr)
11421203
continue
@@ -1145,7 +1206,7 @@ func (client *NginxClient) UpdateStreamServers(ctx context.Context, upstream str
11451206
}
11461207

11471208
for _, server := range toDelete {
1148-
deleteErr := client.DeleteStreamServer(ctx, upstream, server.Server)
1209+
deleteErr := client.deleteStreamServer(ctx, upstream, server.Server, server.ID)
11491210
if deleteErr != nil {
11501211
err = errors.Join(err, deleteErr)
11511212
continue
@@ -1184,45 +1245,82 @@ func (client *NginxClient) getIDOfStreamServer(ctx context.Context, upstream str
11841245
return -1, nil
11851246
}
11861247

1187-
// haveSameParametersForStream checks if a given server has the same parameters as a server already present in NGINX. Order matters.
1188-
func haveSameParametersForStream(newServer StreamUpstreamServer, serverNGX StreamUpstreamServer) bool {
1189-
newServer.ID = serverNGX.ID
1190-
if serverNGX.MaxConns != nil && newServer.MaxConns == nil {
1191-
newServer.MaxConns = &defaultMaxConns
1248+
func deduplicateStreamServers(upstream string, servers []StreamUpstreamServer) ([]StreamUpstreamServer, error) {
1249+
type serverCheck struct {
1250+
server StreamUpstreamServer
1251+
valid bool
1252+
}
1253+
1254+
serverMap := make(map[string]*serverCheck, len(servers))
1255+
var err error
1256+
for _, server := range servers {
1257+
if prev, ok := serverMap[server.Server]; ok {
1258+
if !prev.valid {
1259+
continue
1260+
}
1261+
if !server.hasSameParametersAs(prev.server) {
1262+
prev.valid = false
1263+
err = errors.Join(err, fmt.Errorf(
1264+
"failed to update stream %s server to %s upstream: %w",
1265+
server.Server, upstream, ErrParameterMismatch))
1266+
}
1267+
continue
1268+
}
1269+
serverMap[server.Server] = &serverCheck{server, true}
1270+
}
1271+
retServers := make([]StreamUpstreamServer, 0, len(serverMap))
1272+
for _, server := range servers {
1273+
if check, ok := serverMap[server.Server]; ok && check.valid {
1274+
retServers = append(retServers, server)
1275+
delete(serverMap, server.Server)
1276+
}
11921277
}
1278+
return retServers, err
1279+
}
1280+
1281+
// hasSameParametersAs checks if a given server has the same parameters.
1282+
func (s StreamUpstreamServer) hasSameParametersAs(compareServer StreamUpstreamServer) bool {
1283+
s.ID = compareServer.ID
1284+
s.applyDefaults()
1285+
compareServer.applyDefaults()
1286+
return reflect.DeepEqual(s, compareServer)
1287+
}
11931288

1194-
if serverNGX.MaxFails != nil && newServer.MaxFails == nil {
1195-
newServer.MaxFails = &defaultMaxFails
1289+
func (s *StreamUpstreamServer) applyDefaults() {
1290+
if s.MaxConns == nil {
1291+
s.MaxConns = &defaultMaxConns
11961292
}
11971293

1198-
if serverNGX.FailTimeout != "" && newServer.FailTimeout == "" {
1199-
newServer.FailTimeout = defaultFailTimeout
1294+
if s.MaxFails == nil {
1295+
s.MaxFails = &defaultMaxFails
12001296
}
12011297

1202-
if serverNGX.SlowStart != "" && newServer.SlowStart == "" {
1203-
newServer.SlowStart = defaultSlowStart
1298+
if s.FailTimeout == "" {
1299+
s.FailTimeout = defaultFailTimeout
12041300
}
12051301

1206-
if serverNGX.Backup != nil && newServer.Backup == nil {
1207-
newServer.Backup = &defaultBackup
1302+
if s.SlowStart == "" {
1303+
s.SlowStart = defaultSlowStart
12081304
}
12091305

1210-
if serverNGX.Down != nil && newServer.Down == nil {
1211-
newServer.Down = &defaultDown
1306+
if s.Backup == nil {
1307+
s.Backup = &defaultBackup
12121308
}
12131309

1214-
if serverNGX.Weight != nil && newServer.Weight == nil {
1215-
newServer.Weight = &defaultWeight
1310+
if s.Down == nil {
1311+
s.Down = &defaultDown
12161312
}
12171313

1218-
return reflect.DeepEqual(newServer, serverNGX)
1314+
if s.Weight == nil {
1315+
s.Weight = &defaultWeight
1316+
}
12191317
}
12201318

12211319
func determineStreamUpdates(updatedServers []StreamUpstreamServer, nginxServers []StreamUpstreamServer) (toAdd []StreamUpstreamServer, toRemove []StreamUpstreamServer, toUpdate []StreamUpstreamServer) {
12221320
for _, server := range updatedServers {
12231321
updateFound := false
12241322
for _, serverNGX := range nginxServers {
1225-
if server.Server == serverNGX.Server && !haveSameParametersForStream(server, serverNGX) {
1323+
if server.Server == serverNGX.Server && !server.hasSameParametersAs(serverNGX) {
12261324
server.ID = serverNGX.ID
12271325
updateFound = true
12281326
break
@@ -1950,9 +2048,13 @@ func (client *NginxClient) deleteKeyValPairs(ctx context.Context, zone string, s
19502048
return nil
19512049
}
19522050

1953-
// UpdateHTTPServer updates the server of the upstream.
2051+
// UpdateHTTPServer updates the server of the upstream with the matching server ID.
19542052
func (client *NginxClient) UpdateHTTPServer(ctx context.Context, upstream string, server UpstreamServer) error {
19552053
path := fmt.Sprintf("http/upstreams/%v/servers/%v", upstream, server.ID)
2054+
// The server ID is expected in the URI, but not expected in the body.
2055+
// The NGINX API will return
2056+
// {"error":{"status":400,"text":"unknown parameter \"id\"","code":"UpstreamConfFormatError"}
2057+
// if the ID field is present.
19562058
server.ID = 0
19572059
err := client.patch(ctx, path, &server, http.StatusOK)
19582060
if err != nil {
@@ -1962,9 +2064,13 @@ func (client *NginxClient) UpdateHTTPServer(ctx context.Context, upstream string
19622064
return nil
19632065
}
19642066

1965-
// UpdateStreamServer updates the stream server of the upstream.
2067+
// UpdateStreamServer updates the stream server of the upstream with the matching server ID.
19662068
func (client *NginxClient) UpdateStreamServer(ctx context.Context, upstream string, server StreamUpstreamServer) error {
19672069
path := fmt.Sprintf("stream/upstreams/%v/servers/%v", upstream, server.ID)
2070+
// The server ID is expected in the URI, but not expected in the body.
2071+
// The NGINX API will return
2072+
// {"error":{"status":400,"text":"unknown parameter \"id\"","code":"UpstreamConfFormatError"}
2073+
// if the ID field is present.
19682074
server.ID = 0
19692075
err := client.patch(ctx, path, &server, http.StatusOK)
19702076
if err != nil {

0 commit comments

Comments
 (0)