Skip to content

Commit

Permalink
WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
@panslava
panslava committed Sep 22, 2022
1 parent 0e32aca commit 9de54d5
Show file tree
Hide file tree
Showing 12 changed files with 999 additions and 224 deletions.
25 changes: 13 additions & 12 deletions cmd/glbc/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -188,18 +188,19 @@ func main() {
cloud := app.NewGCEClient()
defaultBackendServicePort := app.DefaultBackendServicePort(kubeClient)
ctxConfig := ingctx.ControllerContextConfig{
Namespace: flags.F.WatchNamespace,
ResyncPeriod: flags.F.ResyncPeriod,
NumL4Workers: flags.F.NumL4Workers,
DefaultBackendSvcPort: defaultBackendServicePort,
HealthCheckPath: flags.F.HealthCheckPath,
FrontendConfigEnabled: flags.F.EnableFrontendConfig,
EnableASMConfigMap: flags.F.EnableASMConfigMapBasedConfig,
ASMConfigMapNamespace: flags.F.ASMConfigMapBasedConfigNamespace,
ASMConfigMapName: flags.F.ASMConfigMapBasedConfigCMName,
EndpointSlicesEnabled: flags.F.EnableEndpointSlices,
MaxIGSize: flags.F.MaxIGSize,
EnableL4ILBDualStack: flags.F.EnableL4ILBDualStack,
Namespace: flags.F.WatchNamespace,
ResyncPeriod: flags.F.ResyncPeriod,
NumL4Workers: flags.F.NumL4Workers,
DefaultBackendSvcPort: defaultBackendServicePort,
HealthCheckPath: flags.F.HealthCheckPath,
FrontendConfigEnabled: flags.F.EnableFrontendConfig,
EnableASMConfigMap: flags.F.EnableASMConfigMapBasedConfig,
ASMConfigMapNamespace: flags.F.ASMConfigMapBasedConfigNamespace,
ASMConfigMapName: flags.F.ASMConfigMapBasedConfigCMName,
EndpointSlicesEnabled: flags.F.EnableEndpointSlices,
MaxIGSize: flags.F.MaxIGSize,
EnableL4ILBDualStack: flags.F.EnableL4ILBDualStack,
EnableL4NetLBDualStack: flags.F.EnableL4NetLBDualStack,
}
ctx := ingctx.NewControllerContext(kubeConfig, kubeClient, backendConfigClient, frontendConfigClient, svcNegClient, ingParamsClient, svcAttachmentClient, cloud, namer, kubeSystemUID, ctxConfig)
go app.RunHTTPServer(ctx.HealthCheck)
Expand Down
19 changes: 10 additions & 9 deletions pkg/context/context.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -127,15 +127,16 @@ type ControllerContextConfig struct {
ResyncPeriod time.Duration
NumL4Workers int
// DefaultBackendSvcPortID is the ServicePort for the system default backend.
DefaultBackendSvcPort utils.ServicePort
HealthCheckPath string
FrontendConfigEnabled bool
EnableASMConfigMap bool
ASMConfigMapNamespace string
ASMConfigMapName string
EndpointSlicesEnabled bool
MaxIGSize int
EnableL4ILBDualStack bool
DefaultBackendSvcPort utils.ServicePort
HealthCheckPath string
FrontendConfigEnabled bool
EnableASMConfigMap bool
ASMConfigMapNamespace string
ASMConfigMapName string
EndpointSlicesEnabled bool
MaxIGSize int
EnableL4ILBDualStack bool
EnableL4NetLBDualStack bool
}

// NewControllerContext returns a new shared set of informers.
Expand Down
2 changes: 2 additions & 0 deletions pkg/flags/flags.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ var (
EnableEndpointSlices bool
EnablePinhole bool
EnableL4ILBDualStack bool
EnableL4NetLBDualStack bool
EnableMultipleIGs bool
MaxIGSize int
}{
Expand Down Expand Up @@ -251,6 +252,7 @@ L7 load balancing. CSV values accepted. Example: -node-port-ranges=80,8080,400-5
flag.BoolVar(&F.EnableEndpointSlices, "enable-endpoint-slices", false, "Enable using Endpoint Slices API instead of Endpoints API")
flag.BoolVar(&F.EnablePinhole, "enable-pinhole", false, "Enable Pinhole firewall feature")
flag.BoolVar(&F.EnableL4ILBDualStack, "enable-l4ilb-dual-stack", false, "Enable Dual-Stack handling for L4 Internal Load Balancers")
flag.BoolVar(&F.EnableL4NetLBDualStack, "enable-l4netlb-dual-stack", false, "Enable Dual-Stack handling for L4 External load balancers")
flag.BoolVar(&F.EnableMultipleIGs, "enable-multiple-igs", false, "Enable using multiple unmanaged instance groups")
flag.IntVar(&F.MaxIGSize, "max-ig-size", 1000, "Max number of instances in Instance Group")
flag.DurationVar(&F.MetricsExportInterval, "metrics-export-interval", 10*time.Minute, `Period for calculating and exporting metrics related to state of managed objects.`)
Expand Down
14 changes: 11 additions & 3 deletions pkg/healthchecksl4/healthchecksl4.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ const (
// Defaults to 3 * 8 = 24 seconds before the LB will steer traffic away.
gceHcUnhealthyThreshold = int64(3)
L4ILBIPv6HCRange = "2600:2d00:1:b029::/64"
L4NetLBIPv6HCRange = "2600:1901:8001::/48"
)

var (
Expand Down Expand Up @@ -154,7 +155,7 @@ func (l4hc *l4HealthChecks) EnsureHealthCheckWithDualStackFirewalls(svc *corev1.
if needsIPv6 {
ipv6HCFWName := namer.L4IPv6HealthCheckFirewall(svc.Namespace, svc.Name, sharedHC)
klog.V(3).Infof("Ensuring IPv6 firewall rule %s for health check %s for service %s", ipv6HCFWName, hcName, namespacedName.String())
err = l4hc.ensureIPv6Firewall(svc, ipv6HCFWName, hcPort, sharedHC, nodeNames)
err = l4hc.ensureIPv6Firewall(svc, ipv6HCFWName, hcPort, sharedHC, nodeNames, l4Type)
if err != nil {
return &EnsureHealthCheckResult{
GceResourceInError: annotations.FirewallForHealthcheckIPv6Resource,
Expand Down Expand Up @@ -222,10 +223,10 @@ func (l4hc *l4HealthChecks) ensureIPv4Firewall(svc *corev1.Service, hcFwName str
return firewalls.EnsureL4LBFirewallForHc(svc, sharedHC, &hcFWRParams, l4hc.cloud, l4hc.recorderFactory.Recorder(svc.Namespace))
}

func (l4hc *l4HealthChecks) ensureIPv6Firewall(svc *corev1.Service, ipv6HCFWName string, hcPort int32, isSharedHC bool, nodeNames []string) error {
func (l4hc *l4HealthChecks) ensureIPv6Firewall(svc *corev1.Service, ipv6HCFWName string, hcPort int32, isSharedHC bool, nodeNames []string, l4Type utils.L4LBType) error {
hcFWRParams := firewalls.FirewallParams{
PortRanges: []string{strconv.Itoa(int(hcPort))},
SourceRanges: []string{L4ILBIPv6HCRange},
SourceRanges: getIPv6HCFirewallSourceRanges(l4Type),
Protocol: string(corev1.ProtocolTCP),
Name: ipv6HCFWName,
NodeNames: nodeNames,
Expand Down Expand Up @@ -413,3 +414,10 @@ func needToUpdateHealthChecks(hc, newHC *composite.HealthCheck) bool {
hc.UnhealthyThreshold < newHC.UnhealthyThreshold ||
hc.HealthyThreshold < newHC.HealthyThreshold
}

func getIPv6HCFirewallSourceRanges(l4Type utils.L4LBType) []string {
if l4Type == utils.XLB {
return []string{L4NetLBIPv6HCRange}
}
return []string{L4ILBIPv6HCRange}
}
10 changes: 10 additions & 0 deletions pkg/l4lb/l4lbcommon.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,16 @@ func deleteL4RBSAnnotations(ctx *context.ControllerContext, svc *v1.Service) err
return patch.PatchServiceObjectMetadata(ctx.KubeClient.CoreV1(), svc, *newObjectMeta)
}

// deleteL4RBSDualStackAnnotations deletes all annotations which could be added by L4 ELB RBS controller
func deleteL4RBSDualStackAnnotations(ctx *context.ControllerContext, svc *v1.Service) error {
newObjectMeta := computeNewAnnotationsIfNeeded(svc, nil, loadbalancers.L4DualStackResourceRBSAnnotationKeys)
if newObjectMeta == nil {
return nil
}
klog.V(3).Infof("Deleting all DualStack annotations for L4 ELB RBS service %v/%v", svc.Namespace, svc.Name)
return patch.PatchServiceObjectMetadata(ctx.KubeClient.CoreV1(), svc, *newObjectMeta)
}

func deleteAnnotation(ctx *context.ControllerContext, svc *v1.Service, annotationKey string) error {
newObjectMeta := svc.ObjectMeta.DeepCopy()
if _, ok := newObjectMeta.Annotations[annotationKey]; !ok {
Expand Down
74 changes: 54 additions & 20 deletions pkg/l4lb/l4netlbcontroller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ package l4lb
import (
"fmt"
"reflect"
"strings"

"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud"
"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud/meta"
Expand Down Expand Up @@ -60,6 +61,7 @@ type L4NetLBController struct {
instancePool instances.NodePool
igLinker *backends.RegionalInstanceGroupLinker
forwardingRules ForwardingRulesGetter
enableDualStack bool
}

// NewL4NetLBController creates a controller for l4 external loadbalancer.
Expand Down Expand Up @@ -414,10 +416,11 @@ func (lc *L4NetLBController) sync(key string) error {
// Returns an error if processing the service update failed.
func (lc *L4NetLBController) syncInternal(service *v1.Service) *loadbalancers.L4NetLBSyncResult {
l4NetLBParams := &loadbalancers.L4NetLBParams{
Service: service,
Cloud: lc.ctx.Cloud,
Namer: lc.namer,
Recorder: lc.ctx.Recorder(service.Namespace),
Service: service,
Cloud: lc.ctx.Cloud,
Namer: lc.namer,
Recorder: lc.ctx.Recorder(service.Namespace),
DualStackEnabled: lc.enableDualStack,
}
l4netlb := loadbalancers.NewL4NetLB(l4NetLBParams)
// check again that rbs is enabled.
Expand Down Expand Up @@ -464,18 +467,39 @@ func (lc *L4NetLBController) syncInternal(service *v1.Service) *loadbalancers.L4
syncResult.Error = err
return syncResult
}
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeNormal, "SyncLoadBalancerSuccessful",
"Successfully ensured L4 External LoadBalancer resources")
if err = updateL4ResourcesAnnotations(lc.ctx, service, syncResult.Annotations); err != nil {
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeWarning, "SyncExternalLoadBalancerFailed",
"Failed to update annotations for load balancer, err: %v", err)
syncResult.Error = fmt.Errorf("failed to set resource annotations, err: %w", err)
return syncResult
if lc.enableDualStack {
lc.emitEnsuredDualStackEvent(service)

if err = updateL4DualStackResourcesAnnotations(lc.ctx, service, syncResult.Annotations); err != nil {
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeWarning, "SyncExternalLoadBalancerFailed",
"Failed to update annotations for load balancer, err: %v", err)
syncResult.Error = fmt.Errorf("failed to set resource annotations, err: %w", err)
return syncResult
}
} else {
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeNormal, "SyncLoadBalancerSuccessful",
"Successfully ensured L4 External LoadBalancer resources")

if err = updateL4ResourcesAnnotations(lc.ctx, service, syncResult.Annotations); err != nil {
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeWarning, "SyncExternalLoadBalancerFailed",
"Failed to update annotations for load balancer, err: %v", err)
syncResult.Error = fmt.Errorf("failed to set resource annotations, err: %w", err)
return syncResult
}
}
syncResult.SetMetricsForSuccessfulServiceSync()
return syncResult
}

func (lc *L4NetLBController) emitEnsuredDualStackEvent(service *v1.Service) {
var ipFamilies []string
for _, ipFamily := range service.Spec.IPFamilies {
ipFamilies = append(ipFamilies, string(ipFamily))
}
lc.ctx.Recorder(service.Namespace).Eventf(service, v1.EventTypeNormal, "SyncLoadBalancerSuccessful",
"Successfully ensured %v External LoadBalancer resources", strings.Join(ipFamilies, " "))
}

func (lc *L4NetLBController) ensureBackendLinking(port utils.ServicePort) error {
zones, err := lc.translator.ListZones(utils.CandidateNodesPredicate)
if err != nil {
Expand All @@ -498,10 +522,11 @@ func (lc *L4NetLBController) ensureInstanceGroups(service *v1.Service, nodeNames
// garbageCollectRBSNetLB cleans-up all gce resources related to service and removes NetLB finalizer
func (lc *L4NetLBController) garbageCollectRBSNetLB(key string, svc *v1.Service) *loadbalancers.L4NetLBSyncResult {
l4NetLBParams := &loadbalancers.L4NetLBParams{
Service: svc,
Cloud: lc.ctx.Cloud,
Namer: lc.namer,
Recorder: lc.ctx.Recorder(svc.Namespace),
Service: svc,
Cloud: lc.ctx.Cloud,
Namer: lc.namer,
Recorder: lc.ctx.Recorder(svc.Namespace),
DualStackEnabled: lc.enableDualStack,
}
l4netLB := loadbalancers.NewL4NetLB(l4NetLBParams)
lc.ctx.Recorder(svc.Namespace).Eventf(svc, v1.EventTypeNormal, "DeletingLoadBalancer",
Expand Down Expand Up @@ -538,11 +563,20 @@ func (lc *L4NetLBController) garbageCollectRBSNetLB(key string, svc *v1.Service)
}

// IMPORTANT: Remove LB annotations from the Service LAST, cause changing service fields are emitted as service update to other controllers
if err := deleteL4RBSAnnotations(lc.ctx, svc); err != nil {
lc.ctx.Recorder(svc.Namespace).Eventf(svc, v1.EventTypeWarning, "DeleteLoadBalancer",
"Error removing resource annotations: %v: %v", err)
result.Error = fmt.Errorf("Failed to reset resource annotations, err: %w", err)
return result
if lc.enableDualStack {
if err := deleteL4RBSDualStackAnnotations(lc.ctx, svc); err != nil {
lc.ctx.Recorder(svc.Namespace).Eventf(svc, v1.EventTypeWarning, "DeleteLoadBalancer",
"Error removing Dual Stack resource annotations: %v: %v", err)
result.Error = fmt.Errorf("failed to reset Dual Stack resource annotations, err: %w", err)
return result
}
} else {
if err := deleteL4RBSAnnotations(lc.ctx, svc); err != nil {
lc.ctx.Recorder(svc.Namespace).Eventf(svc, v1.EventTypeWarning, "DeleteLoadBalancer",
"Error removing resource annotations: %v: %v", err)
result.Error = fmt.Errorf("failed to reset resource annotations, err: %w", err)
return result
}
}

lc.ctx.Recorder(svc.Namespace).Eventf(svc, v1.EventTypeNormal, "DeletedLoadBalancer", "Deleted L4 External LoadBalancer")
Expand Down
106 changes: 99 additions & 7 deletions pkg/l4lb/l4netlbcontroller_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,13 +62,20 @@ const (
)

var (
serviceAnnotationKeys = []string{
annotations.FirewallRuleKey,
netLBCommonAnnotationKeys = []string{
annotations.BackendServiceKey,
annotations.HealthcheckKey,
}
netLBIPv4AnnotationKeys = []string{
annotations.FirewallRuleKey,
annotations.TCPForwardingRuleKey,
annotations.FirewallRuleForHealthcheckKey,
}
netLBIPv6AnnotationKeys = []string{
annotations.FirewallRuleIPv6Key,
annotations.TCPForwardingRuleIPv6Key,
annotations.FirewallRuleForHealthcheckIPv6Key,
}
)

func getExternalIPS() []string {
Expand Down Expand Up @@ -155,7 +162,6 @@ func createAndSyncLegacyNetLBSvc(t *testing.T) (svc *v1.Service, lc *L4NetLBCont
}

func checkBackendService(lc *L4NetLBController, svc *v1.Service) error {

backendServiceLink, bs, err := getBackend(lc, svc)
if err != nil {
return fmt.Errorf("Failed to fetch backend service, err %v", err)
Expand Down Expand Up @@ -250,9 +256,22 @@ func validateNetLBSvcStatus(svc *v1.Service, t *testing.T) {
}
}

func calculateNetLBExpectedAnnotationsKeys(svc *v1.Service) []string {
expectedAnnotations := netLBCommonAnnotationKeys
if utils.NeedsIPv4(svc) {
expectedAnnotations = append(expectedAnnotations, netLBIPv4AnnotationKeys...)
}
if utils.NeedsIPv6(svc) {
expectedAnnotations = append(expectedAnnotations, netLBIPv6AnnotationKeys...)
}
return expectedAnnotations
}

func validateAnnotations(svc *v1.Service) error {
missingKeys := []string{}
for _, key := range serviceAnnotationKeys {
expectedAnnotationsKeys := calculateNetLBExpectedAnnotationsKeys(svc)

var missingKeys []string
for _, key := range expectedAnnotationsKeys {
if _, ok := svc.Annotations[key]; !ok {
missingKeys = append(missingKeys, key)
}
Expand All @@ -264,8 +283,10 @@ func validateAnnotations(svc *v1.Service) error {
}

func validateAnnotationsDeleted(svc *v1.Service) error {
unexpectedKeys := []string{}
for _, key := range serviceAnnotationKeys {
expectedAnnotationsKeys := calculateNetLBExpectedAnnotationsKeys(svc)

var unexpectedKeys []string
for _, key := range expectedAnnotationsKeys {
if _, exists := svc.Annotations[key]; exists {
unexpectedKeys = append(unexpectedKeys, key)
}
Expand Down Expand Up @@ -1264,3 +1285,74 @@ func TestIsRBSBasedServiceForNonLoadBalancersType(t *testing.T) {
})
}
}

func TestCreateDeleteDualStackNetLBService(t *testing.T) {
testCases := []struct {
ipFamilies []v1.IPFamily
desc string
}{
{
ipFamilies: []v1.IPFamily{v1.IPv4Protocol},
desc: "Create and delete IPv4 NetLB",
},
{
ipFamilies: []v1.IPFamily{v1.IPv4Protocol, v1.IPv6Protocol},
desc: "Create and delete IPv4 IPv6 NetLB",
},
{
ipFamilies: []v1.IPFamily{v1.IPv6Protocol},
desc: "Create and delete IPv6 NetLB",
},
{
ipFamilies: []v1.IPFamily{v1.IPv6Protocol, v1.IPv4Protocol},
desc: "Create and delete IPv6 IPv4 NetLB",
},
{
ipFamilies: []v1.IPFamily{},
desc: "Create and delete NetLB with empty IP families",
},
}

for _, tc := range testCases {
t.Run(tc.desc, func(t *testing.T) {
controller := newL4NetLBServiceController()
controller.enableDualStack = true
svc := test.NewL4NetLBRBSService(8080)
svc.Spec.IPFamilies = tc.ipFamilies
addNetLBService(controller, svc)

prevMetrics, err := test.GetL4NetLBLatencyMetric()
if err != nil {
t.Errorf("Error getting L4 NetLB latency metrics err: %v", err)
}
if prevMetrics == nil {
t.Fatalf("Cannot get prometheus metrics for L4NetLB latency")
}

key, _ := common.KeyFunc(svc)
err = controller.sync(key)
if err != nil {
t.Errorf("Failed to sync newly added service %s, err %v", svc.Name, err)
}
svc, err = controller.ctx.KubeClient.CoreV1().Services(svc.Namespace).Get(context.TODO(), svc.Name, metav1.GetOptions{})
if err != nil {
t.Errorf("Failed to lookup service %s, err %v", svc.Name, err)
}

expectedIngressLength := len(tc.ipFamilies)
// For empty IP Families we should provide ipv4 address
if expectedIngressLength == 0 {
expectedIngressLength = 1
}
if len(svc.Status.LoadBalancer.Ingress) != expectedIngressLength {
t.Errorf("expectedIngressLength = %d, got %d", expectedIngressLength, len(svc.Status.LoadBalancer.Ingress))
}

err = validateAnnotations(svc)
if err != nil {
t.Errorf("validateAnnotations(%+v) returned error %v, want nil", svc, err)
}
deleteNetLBService(controller, svc)
})
}
}
Loading

0 comments on commit 9de54d5

Please sign in to comment.