Merge pull request #51562 from nicksardo/gce-attempt-firewall

pkg/cloudprovider/providers/gce/BUILD

@@ -77,7 +77,10 @@ go_library(
         "//vendor/k8s.io/apiserver/pkg/server/options/encryptionconfig:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/storage/value/encrypt/envelope:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes/scheme:go_default_library",
+        "//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
         "//vendor/k8s.io/client-go/tools/cache:go_default_library",
+        "//vendor/k8s.io/client-go/tools/record:go_default_library",
         "//vendor/k8s.io/client-go/util/flowcontrol:go_default_library",
     ],
 )

pkg/cloudprovider/providers/gce/gce.go

@@ -30,10 +30,15 @@ import (
 
 	"cloud.google.com/go/compute/metadata"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/server/options/encryptionconfig"
 	"k8s.io/apiserver/pkg/storage/value/encrypt/envelope"
+	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/controller"
@@ -100,7 +105,10 @@ type GCECloud struct {
 	serviceAlpha             *computealpha.Service
 	containerService         *container.Service
 	cloudkmsService          *cloudkms.Service
+	client                   clientset.Interface
 	clientBuilder            controller.ControllerClientBuilder
+	eventBroadcaster         record.EventBroadcaster
+	eventRecorder            record.EventRecorder
 	projectID                string
 	region                   string
 	localZone                string   // The zone in which we are running
@@ -547,6 +555,14 @@ func tryConvertToProjectNames(configProject, configNetworkProject string, servic
 // This must be called before utilizing the funcs of gce.ClusterID
 func (gce *GCECloud) Initialize(clientBuilder controller.ControllerClientBuilder) {
 	gce.clientBuilder = clientBuilder
+	gce.client = clientBuilder.ClientOrDie("cloud-provider")
+
+	if gce.OnXPN() {
+		gce.eventBroadcaster = record.NewBroadcaster()
+		gce.eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(gce.client.Core().RESTClient()).Events("")})
+		gce.eventRecorder = gce.eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "gce-cloudprovider"})
+	}
+
 	go gce.watchClusterID()
 }
 

pkg/cloudprovider/providers/gce/gce_clusterid.go

@@ -62,7 +62,7 @@ type ClusterID struct {
 func (gce *GCECloud) watchClusterID() {
 	gce.ClusterID = ClusterID{
 		cfgMapKey: fmt.Sprintf("%v/%v", UIDNamespace, UIDConfigMapName),
-		client:    gce.clientBuilder.ClientOrDie("cloud-provider"),
+		client:    gce.client,
 	}
 
 	mapEventHandler := cache.ResourceEventHandlerFuncs{

pkg/cloudprovider/providers/gce/gce_loadbalancer_internal.go

@@ -34,8 +34,6 @@ const (
 	allInstances = "ALL"
 )
 
-type lbBalancingMode string
-
 func (gce *GCECloud) ensureInternalLoadBalancer(clusterName, clusterID string, svc *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
 	nm := types.NamespacedName{Name: svc.Name, Namespace: svc.Namespace}
 	ports, protocol := getPortsAndProtocol(svc.Spec.Ports)
@@ -90,12 +88,8 @@ func (gce *GCECloud) ensureInternalLoadBalancer(clusterName, clusterID string, s
 	glog.V(2).Infof("ensureInternalLoadBalancer(%v): reserved IP %q for the forwarding rule", loadBalancerName, ipToUse)
 
 	// Ensure firewall rules if necessary
-	if gce.OnXPN() {
-		glog.V(2).Infof("ensureInternalLoadBalancer: cluster is on a cross-project network (XPN) network project %v, compute project %v - skipping firewall creation", gce.networkProjectID, gce.projectID)
-	} else {
-		if err = gce.ensureInternalFirewalls(loadBalancerName, ipToUse, clusterID, nm, svc, strconv.Itoa(int(hcPort)), sharedHealthCheck, nodes); err != nil {
-			return nil, err
-		}
+	if err = gce.ensureInternalFirewalls(loadBalancerName, ipToUse, clusterID, nm, svc, strconv.Itoa(int(hcPort)), sharedHealthCheck, nodes); err != nil {
+		return nil, err
 	}
 
 	expectedFwdRule := &compute.ForwardingRule{
@@ -141,7 +135,7 @@ func (gce *GCECloud) ensureInternalLoadBalancer(clusterName, clusterID string, s
 
 	// Delete the previous internal load balancer resources if necessary
 	if existingBackendService != nil {
-		gce.clearPreviousInternalResources(loadBalancerName, existingBackendService, backendServiceName, hcName)
+		gce.clearPreviousInternalResources(svc, loadBalancerName, existingBackendService, backendServiceName, hcName)
 	}
 
 	// Now that the controller knows the forwarding rule exists, we can release the address.
@@ -154,7 +148,7 @@ func (gce *GCECloud) ensureInternalLoadBalancer(clusterName, clusterID string, s
 	return status, nil
 }
 
-func (gce *GCECloud) clearPreviousInternalResources(loadBalancerName string, existingBackendService *compute.BackendService, expectedBSName, expectedHCName string) {
+func (gce *GCECloud) clearPreviousInternalResources(svc *v1.Service, loadBalancerName string, existingBackendService *compute.BackendService, expectedBSName, expectedHCName string) {
 	// If a new backend service was created, delete the old one.
 	if existingBackendService.Name != expectedBSName {
 		glog.V(2).Infof("clearPreviousInternalResources(%v): expected backend service %q does not match previous %q - deleting backend service", loadBalancerName, expectedBSName, existingBackendService.Name)
@@ -168,7 +162,7 @@ func (gce *GCECloud) clearPreviousInternalResources(loadBalancerName string, exi
 		existingHCName := getNameFromLink(existingBackendService.HealthChecks[0])
 		if existingHCName != expectedHCName {
 			glog.V(2).Infof("clearPreviousInternalResources(%v): expected health check %q does not match previous %q - deleting health check", loadBalancerName, expectedHCName, existingHCName)
-			if err := gce.teardownInternalHealthCheckAndFirewall(existingHCName); err != nil {
+			if err := gce.teardownInternalHealthCheckAndFirewall(svc, existingHCName); err != nil {
 				glog.Warningf("clearPreviousInternalResources: could not delete existing healthcheck: %v, err: %v", existingHCName, err)
 			}
 		}
@@ -224,12 +218,17 @@ func (gce *GCECloud) ensureInternalLoadBalancerDeleted(clusterName, clusterID st
 
 	glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting firewall for traffic", loadBalancerName)
 	if err := gce.DeleteFirewall(loadBalancerName); err != nil {
-		return err
+		if isForbidden(err) && gce.OnXPN() {
+			glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): could not delete traffic firewall on XPN cluster. Raising event.", loadBalancerName)
+			gce.raiseFirewallChangeNeededEvent(svc, FirewallToGCloudDeleteCmd(loadBalancerName, gce.NetworkProjectID()))
+		} else {
+			return err
+		}
 	}
 
 	hcName := makeHealthCheckName(loadBalancerName, clusterID, sharedHealthCheck)
 	glog.V(2).Infof("ensureInternalLoadBalancerDeleted(%v): deleting health check %v and its firewall", loadBalancerName, hcName)
-	if err := gce.teardownInternalHealthCheckAndFirewall(hcName); err != nil {
+	if err := gce.teardownInternalHealthCheckAndFirewall(svc, hcName); err != nil {
 		return err
 	}
 
@@ -258,7 +257,7 @@ func (gce *GCECloud) teardownInternalBackendService(bsName string) error {
 	return nil
 }
 
-func (gce *GCECloud) teardownInternalHealthCheckAndFirewall(hcName string) error {
+func (gce *GCECloud) teardownInternalHealthCheckAndFirewall(svc *v1.Service, hcName string) error {
 	if err := gce.DeleteHealthCheck(hcName); err != nil {
 		if isNotFound(err) {
 			glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check does not exist.", hcName)
@@ -274,13 +273,19 @@ func (gce *GCECloud) teardownInternalHealthCheckAndFirewall(hcName string) error
 
 	hcFirewallName := makeHealthCheckFirewallNameFromHC(hcName)
 	if err := gce.DeleteFirewall(hcFirewallName); err != nil && !isNotFound(err) {
+		if isForbidden(err) && gce.OnXPN() {
+			glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): could not delete health check traffic firewall on XPN cluster. Raising Event.", hcName)
+			gce.raiseFirewallChangeNeededEvent(svc, FirewallToGCloudDeleteCmd(hcFirewallName, gce.NetworkProjectID()))
+			return nil
+		}
+
 		return fmt.Errorf("failed to delete health check firewall: %v, err: %v", hcFirewallName, err)
 	}
 	glog.V(2).Infof("teardownInternalHealthCheckAndFirewall(%v): health check firewall deleted", hcFirewallName)
 	return nil
 }
 
-func (gce *GCECloud) ensureInternalFirewall(fwName, fwDesc string, sourceRanges []string, ports []string, protocol v1.Protocol, nodes []*v1.Node) error {
+func (gce *GCECloud) ensureInternalFirewall(svc *v1.Service, fwName, fwDesc string, sourceRanges []string, ports []string, protocol v1.Protocol, nodes []*v1.Node) error {
 	glog.V(2).Infof("ensureInternalFirewall(%v): checking existing firewall", fwName)
 	targetTags, err := gce.GetNodeTags(nodeNames(nodes))
 	if err != nil {
@@ -308,7 +313,13 @@ func (gce *GCECloud) ensureInternalFirewall(fwName, fwDesc string, sourceRanges
 
 	if existingFirewall == nil {
 		glog.V(2).Infof("ensureInternalFirewall(%v): creating firewall", fwName)
-		return gce.CreateFirewall(expectedFirewall)
+		err = gce.CreateFirewall(expectedFirewall)
+		if err != nil && isForbidden(err) && gce.OnXPN() {
+			glog.V(2).Infof("ensureInternalFirewall(%v): do not have permission to create firewall rule (on XPN). Raising event.", fwName)
+			gce.raiseFirewallChangeNeededEvent(svc, FirewallToGCloudCreateCmd(expectedFirewall, gce.NetworkProjectID()))
+			return nil
+		}
+		return err
 	}
 
 	if firewallRuleEqual(expectedFirewall, existingFirewall) {
@@ -316,7 +327,13 @@ func (gce *GCECloud) ensureInternalFirewall(fwName, fwDesc string, sourceRanges
 	}
 
 	glog.V(2).Infof("ensureInternalFirewall(%v): updating firewall", fwName)
-	return gce.UpdateFirewall(expectedFirewall)
+	err = gce.UpdateFirewall(expectedFirewall)
+	if err != nil && isForbidden(err) && gce.OnXPN() {
+		glog.V(2).Infof("ensureInternalFirewall(%v): do not have permission to update firewall rule (on XPN). Raising event.", fwName)
+		gce.raiseFirewallChangeNeededEvent(svc, FirewallToGCloudUpdateCmd(expectedFirewall, gce.NetworkProjectID()))
+		return nil
+	}
+	return err
 }
 
 func (gce *GCECloud) ensureInternalFirewalls(loadBalancerName, ipAddress, clusterID string, nm types.NamespacedName, svc *v1.Service, healthCheckPort string, sharedHealthCheck bool, nodes []*v1.Node) error {
@@ -327,7 +344,7 @@ func (gce *GCECloud) ensureInternalFirewalls(loadBalancerName, ipAddress, cluste
 	if err != nil {
 		return err
 	}
-	err = gce.ensureInternalFirewall(loadBalancerName, fwDesc, sourceRanges.StringSlice(), ports, protocol, nodes)
+	err = gce.ensureInternalFirewall(svc, loadBalancerName, fwDesc, sourceRanges.StringSlice(), ports, protocol, nodes)
 	if err != nil {
 		return err
 	}
@@ -335,7 +352,7 @@ func (gce *GCECloud) ensureInternalFirewalls(loadBalancerName, ipAddress, cluste
 	// Second firewall is for health checking nodes / services
 	fwHCName := makeHealthCheckFirewallName(loadBalancerName, clusterID, sharedHealthCheck)
 	hcSrcRanges := LoadBalancerSrcRanges()
-	return gce.ensureInternalFirewall(fwHCName, "", hcSrcRanges, []string{healthCheckPort}, v1.ProtocolTCP, nodes)
+	return gce.ensureInternalFirewall(svc, fwHCName, "", hcSrcRanges, []string{healthCheckPort}, v1.ProtocolTCP, nodes)
 }
 
 func (gce *GCECloud) ensureInternalHealthCheck(name string, svcName types.NamespacedName, shared bool, path string, port int32) (*compute.HealthCheck, error) {

pkg/cloudprovider/providers/gce/gce_util.go

@@ -23,6 +23,7 @@ import (
 	"regexp"
 	"strings"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 
@@ -58,6 +59,43 @@ func getProjectAndZone() (string, string, error) {
 	return projectID, zone, nil
 }
 
+func (gce *GCECloud) raiseFirewallChangeNeededEvent(svc *v1.Service, cmd string) {
+	msg := fmt.Sprintf("Firewall change required by network admin: `%v`", cmd)
+	if gce.eventRecorder != nil && svc != nil {
+		gce.eventRecorder.Event(svc, v1.EventTypeNormal, "LoadBalancerManualChange", msg)
+	}
+}
+
+// FirewallToGCloudCreateCmd generates a gcloud command to create a firewall with specified params
+func FirewallToGCloudCreateCmd(fw *compute.Firewall, projectID string) string {
+	args := firewallToGcloudArgs(fw, projectID)
+	return fmt.Sprintf("gcloud compute firewall-rules create %v --network %v %v", fw.Name, getNameFromLink(fw.Network), args)
+}
+
+// FirewallToGCloudCreateCmd generates a gcloud command to update a firewall to specified params
+func FirewallToGCloudUpdateCmd(fw *compute.Firewall, projectID string) string {
+	args := firewallToGcloudArgs(fw, projectID)
+	return fmt.Sprintf("gcloud compute firewall-rules update %v %v", fw.Name, args)
+}
+
+// FirewallToGCloudCreateCmd generates a gcloud command to delete a firewall to specified params
+func FirewallToGCloudDeleteCmd(fwName, projectID string) string {
+	return fmt.Sprintf("gcloud compute firewall-rules delete %v --project %v", fwName, projectID)
+}
+
+func firewallToGcloudArgs(fw *compute.Firewall, projectID string) string {
+	var allPorts []string
+	for _, a := range fw.Allowed {
+		for _, p := range a.Ports {
+			allPorts = append(allPorts, fmt.Sprintf("%v:%v", a.IPProtocol, p))
+		}
+	}
+	allow := strings.Join(allPorts, ",")
+	srcRngs := strings.Join(fw.SourceRanges, ",")
+	targets := strings.Join(fw.TargetTags, ",")
+	return fmt.Sprintf("--description %q --allow %v --source-ranges %v --target-tags %v --project %v", fw.Description, allow, srcRngs, targets, projectID)
+}
+
 // Take a GCE instance 'hostname' and break it down to something that can be fed
 // to the GCE API client library.  Basically this means reducing 'kubernetes-
 // node-2.c.my-proj.internal' to 'kubernetes-node-2' if necessary.
@@ -150,6 +188,10 @@ func isNotFoundOrInUse(err error) bool {
 	return isNotFound(err) || isInUsedByError(err)
 }
 
+func isForbidden(err error) bool {
+	return isHTTPErrorCode(err, http.StatusForbidden)
+}
+
 func makeGoogleAPINotFoundError(message string) error {
 	return &googleapi.Error{Code: http.StatusNotFound, Message: message}
 }
@@ -158,10 +200,6 @@ func makeGoogleAPIError(code int, message string) error {
 	return &googleapi.Error{Code: code, Message: message}
 }
 
-func isForbidden(err error) bool {
-	return isHTTPErrorCode(err, http.StatusForbidden)
-}
-
 // TODO(#51665): Remove this once Network Tiers becomes Beta in GCP.
 func handleAlphaNetworkTierGetError(err error) (string, error) {
 	if isForbidden(err) {

test/e2e/framework/util.go

@@ -4802,7 +4802,7 @@ func CleanupGCEResources(c clientset.Interface, loadBalancerName, zone string) (
 		retErr = fmt.Errorf("%v\n%v", retErr, err)
 		return
 	}
-	if err := gceCloud.DeleteExternalTargetPoolAndChecks(loadBalancerName, region, clusterID, hcNames...); err != nil &&
+	if err := gceCloud.DeleteExternalTargetPoolAndChecks(nil, loadBalancerName, region, clusterID, hcNames...); err != nil &&
 		!IsGoogleAPIHTTPErrorCode(err, http.StatusNotFound) {
 		retErr = fmt.Errorf("%v\n%v", retErr, err)
 	}