Skip to content

Commit 3c4f1ee

Browse files
justinsbjustinsb
committed
refactor: introduce loadBalancer sync operation
This clarifies the flow, laying the groundwork for multiple protocols.
1 parent b54f8b1 commit 3c4f1ee

File tree

7 files changed

+123
-46
lines changed

7 files changed

+123
-46
lines changed

‎providers/gce/BUILD‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ go_library(
2626
"gce_loadbalancer_internal.go",
2727
"gce_loadbalancer_metrics.go",
2828
"gce_loadbalancer_naming.go",
29+
"gce_loadbalancer_sync.go",
2930
"gce_networkendpointgroup.go",
3031
"gce_networks.go",
3132
"gce_routes.go",

‎providers/gce/gce_loadbalancer.go‎

Lines changed: 8 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -169,13 +169,15 @@ func (g *Cloud) EnsureLoadBalancer(ctx context.Context, clusterName string, svc
169169

170170
klog.V(4).Infof("EnsureLoadBalancer(%v, %v, %v, %v, %v): ensure %v loadbalancer", clusterName, svc.Namespace, svc.Name, loadBalancerName, g.region, desiredScheme)
171171

172-
existingFwdRule, err := g.GetRegionForwardingRule(loadBalancerName, g.region)
172+
actualForwardingRule, err := g.GetRegionForwardingRule(loadBalancerName, g.region)
173173
if err != nil && !isNotFound(err) {
174174
return nil, err
175175
}
176+
op := &loadBalancerSync{}
177+
op.actualForwardingRule = actualForwardingRule
176178

177-
if existingFwdRule != nil {
178-
existingScheme := cloud.LbScheme(strings.ToUpper(existingFwdRule.LoadBalancingScheme))
179+
if op.actualForwardingRule != nil {
180+
existingScheme := cloud.LbScheme(strings.ToUpper(op.actualForwardingRule.LoadBalancingScheme))
179181

180182
// If the loadbalancer type changes between INTERNAL and EXTERNAL, the old load balancer should be deleted.
181183
if existingScheme != desiredScheme {
@@ -192,16 +194,16 @@ func (g *Cloud) EnsureLoadBalancer(ctx context.Context, clusterName string, svc
192194
}
193195

194196
// Assume the ensureDeleted function successfully deleted the forwarding rule.
195-
existingFwdRule = nil
197+
op.actualForwardingRule = nil
196198
}
197199
}
198200

199201
var status *v1.LoadBalancerStatus
200202
switch desiredScheme {
201203
case cloud.SchemeInternal:
202-
status, err = g.ensureInternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
204+
status, err = g.ensureInternalLoadBalancer(clusterName, clusterID, svc, op, nodes)
203205
default:
204-
status, err = g.ensureExternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
206+
status, err = g.ensureExternalLoadBalancer(clusterName, clusterID, svc, op, nodes)
205207
}
206208
if err != nil {
207209
klog.Errorf("Failed to EnsureLoadBalancer(%s, %s, %s, %s, %s), err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, g.region, err)

‎providers/gce/gce_loadbalancer_external.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,11 +55,11 @@ const (
5555
// Due to an interesting series of design decisions, this handles both creating
5656
// new load balancers and updating existing load balancers, recognizing when
5757
// each is needed.
58-
func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string, apiService *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
58+
func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string, apiService *v1.Service, op *loadBalancerSync, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
5959
// Process services with LoadBalancerClass "networking.gke.io/l4-regional-external-legacy" used for this controller.
6060
// LoadBalancerClass can't be updated so we know this controller should process the NetLB.
6161
// Skip service handling if it uses Regional Backend Services and handled by other controllers
62-
if !shouldProcessNetLB(apiService, existingFwdRule) {
62+
if !shouldProcessNetLB(apiService, op.actualForwardingRule) {
6363
return nil, cloudprovider.ImplementedElsewhere
6464
}
6565

‎providers/gce/gce_loadbalancer_external_test.go‎

Lines changed: 42 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -483,11 +483,14 @@ func createExternalLoadBalancer(gce *Cloud, svc *v1.Service, nodeNames []string,
483483
return nil, err
484484
}
485485

486+
op := &loadBalancerSync{}
487+
op.actualForwardingRule = nil
488+
486489
return gce.ensureExternalLoadBalancer(
487490
clusterName,
488491
clusterID,
489492
svc,
490-
nil,
493+
op,
491494
nodes,
492495
)
493496
}
@@ -554,7 +557,11 @@ func TestShouldNotRecreateLBWhenNetworkTiersMismatch(t *testing.T) {
554557
},
555558
} {
556559
tc.mutateSvc(svc)
557-
status, err := gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
560+
561+
op := &loadBalancerSync{}
562+
op.actualForwardingRule = nil
563+
564+
status, err := gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
558565
if tc.expectError {
559566
if err == nil {
560567
t.Errorf("for test case %q, expect errror != nil, but got %v", tc.desc, err)
@@ -782,7 +789,10 @@ func TestEnsureExternalLoadBalancerFailsIfInvalidNetworkTier(t *testing.T) {
782789
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
783790
require.NoError(t, err)
784791

785-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
792+
op := &loadBalancerSync{}
793+
op.actualForwardingRule = nil
794+
795+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
786796
require.Error(t, err)
787797
assert.EqualError(t, err, errStrUnsupportedTier)
788798
}
@@ -799,7 +809,10 @@ func TestEnsureExternalLoadBalancerFailsWithNoNodes(t *testing.T) {
799809
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
800810
require.NoError(t, err)
801811

802-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, []*v1.Node{})
812+
op := &loadBalancerSync{}
813+
op.actualForwardingRule = nil
814+
815+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, []*v1.Node{})
803816
require.Error(t, err)
804817
assert.EqualError(t, err, errStrLbNoHosts)
805818
}
@@ -843,7 +856,10 @@ func TestEnsureExternalLoadBalancerRBSAnnotation(t *testing.T) {
843856
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
844857
require.NoError(t, err)
845858

846-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
859+
op := &loadBalancerSync{}
860+
op.actualForwardingRule = nil
861+
862+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
847863
if tc.wantError != nil {
848864
assert.EqualError(t, err, (*tc.wantError).Error())
849865
} else {
@@ -911,7 +927,10 @@ func TestEnsureExternalLoadBalancerRBSFinalizer(t *testing.T) {
911927
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
912928
require.NoError(t, err)
913929

914-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
930+
op := &loadBalancerSync{}
931+
op.actualForwardingRule = nil
932+
933+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
915934
if tc.wantError != nil {
916935
assert.EqualError(t, err, (*tc.wantError).Error())
917936
} else {
@@ -991,7 +1010,10 @@ func TestDeleteExternalLoadBalancerWithFinalizer(t *testing.T) {
9911010
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
9921011
require.NoError(t, err)
9931012

994-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
1013+
op := &loadBalancerSync{}
1014+
op.actualForwardingRule = nil
1015+
1016+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
9951017
if err != nil {
9961018
assert.NoError(t, err, "Should not return an error "+desc)
9971019
}
@@ -1055,7 +1077,10 @@ func TestEnsureExternalLoadBalancerExistingFwdRule(t *testing.T) {
10551077
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
10561078
require.NoError(t, err)
10571079

1058-
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, tc.existingForwardingRule, nodes)
1080+
op := &loadBalancerSync{}
1081+
op.actualForwardingRule = tc.existingForwardingRule
1082+
1083+
_, err = gce.ensureExternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
10591084
if tc.wantError != nil {
10601085
assert.EqualError(t, err, (*tc.wantError).Error())
10611086
} else {
@@ -1868,11 +1893,14 @@ func TestEnsureTargetPoolAndHealthCheck(t *testing.T) {
18681893
svc, err = gce.client.CoreV1().Services(svc.Namespace).Create(context.TODO(), svc, metav1.CreateOptions{})
18691894
require.NoError(t, err)
18701895

1896+
op := &loadBalancerSync{}
1897+
op.actualForwardingRule = nil
1898+
18711899
status, err := gce.ensureExternalLoadBalancer(
18721900
vals.ClusterName,
18731901
vals.ClusterID,
18741902
svc,
1875-
nil,
1903+
op,
18761904
nodes,
18771905
)
18781906
require.NotNil(t, status)
@@ -2121,11 +2149,15 @@ func TestEnsureExternalLoadBalancerErrors(t *testing.T) {
21212149
if tc.injectMock != nil {
21222150
tc.injectMock(gce.c.(*cloud.MockGCE))
21232151
}
2152+
2153+
op := &loadBalancerSync{}
2154+
op.actualForwardingRule = params.existingFwdRule
2155+
21242156
status, err := gce.ensureExternalLoadBalancer(
21252157
params.clusterName,
21262158
params.clusterID,
21272159
params.service,
2128-
params.existingFwdRule,
2160+
op,
21292161
params.nodes,
21302162
)
21312163
assert.Error(t, err, "Should return an error when "+desc)

‎providers/gce/gce_loadbalancer_internal.go‎

Lines changed: 22 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -54,10 +54,10 @@ const (
5454
labelGKESubnetworkName = "cloud.google.com/gke-node-pool-subnet"
5555
)
5656

57-
func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
57+
func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, op *loadBalancerSync, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
5858
// Process services with LoadBalancerClass "networking.gke.io/l4-regional-internal-legacy" used for this controller.
5959
// LoadBalancerClass can't be updated so we know this controller should process the ILB.
60-
if existingFwdRule == nil && !hasFinalizer(svc, ILBFinalizerV1) && !hasLoadBalancerClass(svc, LegacyRegionalInternalLoadBalancerClass) {
60+
if op.actualForwardingRule == nil && !hasFinalizer(svc, ILBFinalizerV1) && !hasLoadBalancerClass(svc, LegacyRegionalInternalLoadBalancerClass) {
6161
// Neither the forwarding rule nor the V1 finalizer exists. This is most likely a new service.
6262
if svc.Spec.LoadBalancerClass != nil && !hasLoadBalancerClass(svc, LegacyRegionalInternalLoadBalancerClass) {
6363
klog.V(2).Infof("Skipped ensureInternalLoadBalancer for service %s/%s, as service contains %q loadBalancerClass.", svc.Namespace, svc.Name, *svc.Spec.LoadBalancerClass)
@@ -118,8 +118,8 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
118118

119119
// Get existing backend service (if exists)
120120
var existingBackendService *compute.BackendService
121-
if existingFwdRule != nil && existingFwdRule.BackendService != "" {
122-
existingBSName := getNameFromLink(existingFwdRule.BackendService)
121+
if op.actualForwardingRule != nil && op.actualForwardingRule.BackendService != "" {
122+
existingBSName := getNameFromLink(op.actualForwardingRule.BackendService)
123123
if existingBackendService, err = g.GetRegionBackendService(existingBSName, g.region); err != nil && !isNotFound(err) {
124124
return nil, err
125125
}
@@ -153,7 +153,7 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
153153
}
154154
// Determine IP which will be used for this LB. If no forwarding rule has been established
155155
// or specified in the Service spec, then requestedIP = "".
156-
ipToUse := ilbIPToUse(svc, existingFwdRule, subnetworkURL)
156+
ipToUse := ilbIPToUse(svc, op.actualForwardingRule, subnetworkURL)
157157

158158
klog.V(2).Infof("ensureInternalLoadBalancer(%v): Using subnet %s for LoadBalancer IP %s", loadBalancerName, options.SubnetName, ipToUse)
159159

@@ -199,20 +199,20 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
199199
newFwdRule.Ports = nil
200200
newFwdRule.AllPorts = true
201201
}
202+
op.desiredForwardingRule = newFwdRule
202203

203-
fwdRuleDeleted := false
204-
if existingFwdRule != nil && !forwardingRulesEqual(existingFwdRule, newFwdRule) {
204+
if op.actualForwardingRule != nil && !forwardingRulesEqual(op.actualForwardingRule, op.desiredForwardingRule) {
205205
// Delete existing forwarding rule before making changes to the backend service. For example - changing protocol
206206
// of backend service without first deleting forwarding rule will throw an error since the linked forwarding
207207
// rule would show the old protocol.
208208
if klogV := klog.V(2); klogV.Enabled() {
209-
frDiff := cmp.Diff(existingFwdRule, newFwdRule)
210-
klogV.Infof("ensureInternalLoadBalancer(%v): forwarding rule changed - Existing - %+v\n, New - %+v\n, Diff(-existing, +new) - %s\n. Deleting existing forwarding rule.", loadBalancerName, existingFwdRule, newFwdRule, frDiff)
209+
frDiff := cmp.Diff(op.actualForwardingRule, op.desiredForwardingRule)
210+
klogV.Infof("ensureInternalLoadBalancer(%v): forwarding rule changed - Existing - %+v\n, New - %+v\n, Diff(-existing, +new) - %s\n. Deleting existing forwarding rule.", loadBalancerName, op.actualForwardingRule, op.desiredForwardingRule, frDiff)
211211
}
212212
if err = ignoreNotFound(g.DeleteRegionForwardingRule(loadBalancerName, g.region)); err != nil {
213213
return nil, err
214214
}
215-
fwdRuleDeleted = true
215+
op.actualForwardingRule = nil
216216
}
217217

218218
bsDescription := makeBackendServiceDescription(nm, sharedBackend)
@@ -221,9 +221,9 @@ func (g *Cloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v
221221
return nil, err
222222
}
223223

224-
if fwdRuleDeleted || existingFwdRule == nil {
225-
// existing rule has been deleted, pass in nil
226-
if err := g.ensureInternalForwardingRule(nil, newFwdRule); err != nil {
224+
if op.actualForwardingRule == nil {
225+
// existing rule has been deleted
226+
if err := g.ensureInternalForwardingRule(op); err != nil {
227227
return nil, err
228228
}
229229
}
@@ -1084,24 +1084,24 @@ func getFwdRuleAPIVersion(rule *compute.ForwardingRule) (meta.Version, error) {
10841084
return d.APIVersion, nil
10851085
}
10861086

1087-
func (g *Cloud) ensureInternalForwardingRule(existingFwdRule, newFwdRule *compute.ForwardingRule) (err error) {
1088-
if existingFwdRule != nil {
1089-
if forwardingRulesEqual(existingFwdRule, newFwdRule) {
1090-
klog.V(4).Infof("existingFwdRule == newFwdRule, no updates needed (existingFwdRule == %+v)", existingFwdRule)
1087+
func (g *Cloud) ensureInternalForwardingRule(op *loadBalancerSync) (err error) {
1088+
if op.actualForwardingRule != nil {
1089+
if forwardingRulesEqual(op.actualForwardingRule, op.desiredForwardingRule) {
1090+
klog.V(4).Infof("actualForwardingRule == desiredForwardingRule, no updates needed (actualForwardingRule == %+v)", op.actualForwardingRule)
10911091
return nil
10921092
}
1093-
klog.V(2).Infof("ensureInternalLoadBalancer(%v): deleting existing forwarding rule with IP address %v", existingFwdRule.Name, existingFwdRule.IPAddress)
1094-
if err = ignoreNotFound(g.DeleteRegionForwardingRule(existingFwdRule.Name, g.region)); err != nil {
1093+
klog.V(2).Infof("ensureInternalLoadBalancer(%v): deleting existing forwarding rule with IP address %v", op.actualForwardingRule.Name, op.actualForwardingRule.IPAddress)
1094+
if err = ignoreNotFound(g.DeleteRegionForwardingRule(op.actualForwardingRule.Name, g.region)); err != nil {
10951095
return err
10961096
}
10971097
}
10981098
// At this point, the existing rule has been deleted if required.
10991099
// Create the rule based on the api version determined
1100-
klog.V(2).Infof("ensureInternalLoadBalancer(%v): creating forwarding rule", newFwdRule.Name)
1101-
if err = g.CreateRegionForwardingRule(newFwdRule, g.region); err != nil {
1100+
klog.V(2).Infof("ensureInternalLoadBalancer(%v): creating forwarding rule", op.desiredForwardingRule.Name)
1101+
if err = g.CreateRegionForwardingRule(op.desiredForwardingRule, g.region); err != nil {
11021102
return err
11031103
}
1104-
klog.V(2).Infof("ensureInternalLoadBalancer(%v): created forwarding rule", newFwdRule.Name)
1104+
klog.V(2).Infof("ensureInternalLoadBalancer(%v): created forwarding rule", op.desiredForwardingRule.Name)
11051105
return nil
11061106
}
11071107

‎providers/gce/gce_loadbalancer_internal_test.go‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -48,11 +48,14 @@ func createInternalLoadBalancer(gce *Cloud, svc *v1.Service, existingFwdRule *co
4848
return nil, err
4949
}
5050

51+
op := &loadBalancerSync{}
52+
op.actualForwardingRule = existingFwdRule
53+
5154
return gce.ensureInternalLoadBalancer(
5255
clusterName,
5356
clusterID,
5457
svc,
55-
existingFwdRule,
58+
op,
5659
nodes,
5760
)
5861
}
@@ -660,7 +663,10 @@ func TestUpdateInternalLoadBalancerNodes(t *testing.T) {
660663
nodes, err := createAndInsertNodes(gce, node1Name, vals.ZoneName)
661664
require.NoError(t, err)
662665

663-
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
666+
op := &loadBalancerSync{}
667+
op.actualForwardingRule = nil
668+
669+
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
664670
assert.NoError(t, err)
665671

666672
// Replace the node in initial zone; add new node in a new zone.
@@ -729,7 +735,10 @@ func TestUpdateInternalLoadBalancerNodesWithEmptyZone(t *testing.T) {
729735
nodes, err := createAndInsertNodes(gce, node1Name, vals.ZoneName)
730736
require.NoError(t, err)
731737

732-
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, nil, nodes)
738+
op := &loadBalancerSync{}
739+
op.actualForwardingRule = nil
740+
741+
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
733742
assert.NoError(t, err)
734743

735744
// Ensure Node has been added to instance group
@@ -747,7 +756,7 @@ func TestUpdateInternalLoadBalancerNodesWithEmptyZone(t *testing.T) {
747756
nodes[0].Labels[v1.LabelTopologyZone] = "" // empty zone
748757

749758
lbName := gce.GetLoadBalancerName(context.TODO(), "", svc)
750-
existingFwdRule := &compute.ForwardingRule{
759+
op.actualForwardingRule = &compute.ForwardingRule{
751760
Name: lbName,
752761
IPAddress: "",
753762
Ports: []string{"123"},
@@ -756,7 +765,7 @@ func TestUpdateInternalLoadBalancerNodesWithEmptyZone(t *testing.T) {
756765
Description: fmt.Sprintf(`{"kubernetes.io/service-name":"%s"}`, types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}.String()),
757766
}
758767

759-
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, existingFwdRule, nodes)
768+
_, err = gce.ensureInternalLoadBalancer(vals.ClusterName, vals.ClusterID, svc, op, nodes)
760769
assert.NoError(t, err)
761770

762771
// Expect load balancer to not have deleted node test-node-1
@@ -1232,11 +1241,15 @@ func TestEnsureInternalLoadBalancerErrors(t *testing.T) {
12321241
}
12331242
_, err = gce.client.CoreV1().Services(params.service.Namespace).Create(context.TODO(), params.service, metav1.CreateOptions{})
12341243
require.NoError(t, err)
1244+
1245+
op := &loadBalancerSync{}
1246+
op.actualForwardingRule = params.existingFwdRule
1247+
12351248
status, err := gce.ensureInternalLoadBalancer(
12361249
params.clusterName,
12371250
params.clusterID,
12381251
params.service,
1239-
params.existingFwdRule,
1252+
op,
12401253
params.nodes,
12411254
)
12421255
assert.Error(t, err, "Should return an error when "+desc)

0 commit comments

Comments
 (0)