task(GcpVpcPeering): watching operations

api/cloud-control/v1beta1/vpcpeering_types.go

@@ -148,6 +148,13 @@ type VpcPeeringStatus struct {
 	// +listType=map
 	// +listMapKey=type
 	Conditions []metav1.Condition `json:"conditions,omitempty"`
+
+	//GCP creates operations when creating peerings, since we are creating two peerings (one in each direction), we store the operation names here to track them if needed
+	//One example is the quota exceeded error that can happen on either side if one of the vpcs is close to the quota limit
+	// +optional
+	Operation string `json:"operation,omitempty"`
+	// +optional
+	RemoteOperation string `json:"remoteOperation,omitempty"`
 }
 
 //+kubebuilder:object:root=true

config/crd/bases/cloud-control.kyma-project.io_vpcpeerings.yaml

@@ -237,8 +237,15 @@ spec:
                 x-kubernetes-list-type: map
               id:
                 type: string
+              operation:
+                description: |-
+                  GCP creates operations when creating peerings, since we are creating two peerings (one in each direction), we store the operation names here to track them if needed
+                  One example is the quota exceeded error that can happen on either side if one of the vpcs is close to the quota limit
+                type: string
               remoteId:
                 type: string
+              remoteOperation:
+                type: string
               state:
                 type: string
               vpcId:

config/dist/kcp/crd/bases/cloud-control.kyma-project.io_vpcpeerings.yaml

@@ -237,8 +237,15 @@ spec:
                 x-kubernetes-list-type: map
               id:
                 type: string
+              operation:
+                description: |-
+                  GCP creates operations when creating peerings, since we are creating two peerings (one in each direction), we store the operation names here to track them if needed
+                  One example is the quota exceeded error that can happen on either side if one of the vpcs is close to the quota limit
+                type: string
               remoteId:
                 type: string
+              remoteOperation:
+                type: string
               state:
                 type: string
               vpcId:

pkg/kcp/provider/gcp/client/gcpClients.go

@@ -32,6 +32,7 @@ type GcpClients struct {
 
 // The VpcPeeringClients uses a different service account than the other clients and has different permissions as well.
 type VpcPeeringClients struct {
+	ComputeGlobalOperations    *compute.GlobalOperationsClient
 	ComputeNetworks            *compute.NetworksClient
 	ResourceManagerTagBindings *resourcemanager.TagBindingsClient
 }
@@ -126,8 +127,11 @@ func NewGcpClients(ctx context.Context, saJsonKeyPath string, vpcPeeringSaJsonKe
 	if err != nil {
 		return nil, fmt.Errorf("error creating vpc peering compute networks client: %w", err)
 	}
+	vpcPeeringComputeGlobalOperations, err := compute.NewGlobalOperationsRESTClient(ctx, option.WithTokenSource(vpcPeeringComputeNetworksTokenSource))
+	if err != nil {
+		return nil, fmt.Errorf("error creating vpc peering compute operations client: %w", err)
+	}
 	// resource manager client for VPC peering, uses a different service account----------------
-
 	vpcPeeringResourceManagerTokenProvider, err := vpcPeeringClientBuilder.WithScopes(resourcemanager.DefaultAuthScopes()).BuildTokenProvider()
 	if err != nil {
 		return nil, fmt.Errorf("failed to build vpc peering resource manager token provider: %w", err)
@@ -148,6 +152,7 @@ func NewGcpClients(ctx context.Context, saJsonKeyPath string, vpcPeeringSaJsonKe
 		RedisCluster:  redisCluster,
 		RedisInstance: redisInstance,
 		VpcPeeringClients: &VpcPeeringClients{
+			ComputeGlobalOperations:    vpcPeeringComputeGlobalOperations,
 			ComputeNetworks:            vpcPeeringComputeNetworks,
 			ResourceManagerTagBindings: vpcPeeringresourceManagerTagBindings,
 		},

pkg/kcp/provider/gcp/meta/metadata.go

@@ -2,6 +2,7 @@ package meta
 
 import (
 	"errors"
+	"strings"
 
 	"github.com/googleapis/gax-go/v2/apierror"
 	"google.golang.org/api/googleapi"
@@ -74,3 +75,21 @@ func IsTooManyRequests(err error) bool {
 
 	return false
 }
+
+func IsOperationInProgressError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	var googleapierror *googleapi.Error
+	if ok := errors.As(err, &googleapierror); ok {
+		if googleapierror.Code == 400 {
+			for _, e := range googleapierror.Errors {
+				if strings.Contains(e.Message, "peering operation in progress") {
+					return true
+				}
+			}
+		}
+	}
+	return false
+}

pkg/kcp/provider/gcp/mock/vpcPeeringStore.go

@@ -13,10 +13,11 @@ type vpcPeeringEntry struct {
 	peering *pb.NetworkPeering
 }
 type vpcPeeringStore struct {
-	m        sync.Mutex
-	items    map[string]*vpcPeeringEntry
-	errorMap map[string]error
-	tags     map[string][]string
+	m          sync.Mutex
+	items      map[string]*vpcPeeringEntry
+	operations map[string]*pb.Operation
+	errorMap   map[string]error
+	tags       map[string][]string
 }
 
 type VpcPeeringMockClientUtils interface {
@@ -30,7 +31,7 @@ func getFullNetworkUrl(project, vpc string) string {
 	return fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/%s/global/networks/%s", project, vpc)
 }
 
-func (s *vpcPeeringStore) CreateRemoteVpcPeering(ctx context.Context, remotePeeringName string, remoteVpc string, remoteProject string, customRoutes bool, kymaProject string, kymaVpc string) error {
+func (s *vpcPeeringStore) CreateRemoteVpcPeering(ctx context.Context, remotePeeringName string, remoteVpc string, remoteProject string, customRoutes bool, kymaProject string, kymaVpc string) (*pb.Operation, error) {
 	s.m.Lock()
 	defer s.m.Unlock()
 	remoteNetwork := getFullNetworkUrl(remoteProject, remoteVpc)
@@ -42,9 +43,21 @@ func (s *vpcPeeringStore) CreateRemoteVpcPeering(ctx context.Context, remotePeer
 		exportCustomRoutes = true
 	}
 
+	if s.operations == nil {
+		s.operations = make(map[string]*pb.Operation)
+	}
+
+	if s.operations[remoteVpc] == nil {
+		operationpb := &pb.Operation{
+			Status: ptr.To(pb.Operation_DONE),
+			Name:   ptr.To(remoteVpc),
+		}
+		s.operations[remoteVpc] = operationpb
+	}
+
 	_, peeringExists := s.items[remoteNetwork]
 	if peeringExists {
-		return nil
+		return s.operations[remoteVpc], nil
 	}
 
 	item := &vpcPeeringEntry{
@@ -59,10 +72,10 @@ func (s *vpcPeeringStore) CreateRemoteVpcPeering(ctx context.Context, remotePeer
 	}
 	s.items[remoteNetwork] = item
 
-	return nil
+	return s.operations[remoteVpc], nil
 }
 
-func (s *vpcPeeringStore) CreateKymaVpcPeering(ctx context.Context, remotePeeringName string, remoteVpc string, remoteProject string, customRoutes bool, kymaProject string, kymaVpc string) error {
+func (s *vpcPeeringStore) CreateLocalVpcPeering(ctx context.Context, remotePeeringName string, remoteVpc string, remoteProject string, customRoutes bool, kymaProject string, kymaVpc string) (*pb.Operation, error) {
 	s.m.Lock()
 	defer s.m.Unlock()
 
@@ -72,9 +85,21 @@ func (s *vpcPeeringStore) CreateKymaVpcPeering(ctx context.Context, remotePeerin
 	exportCustomRoutes := false
 	importCustomRoutes := customRoutes
 
+	if s.operations == nil {
+		s.operations = make(map[string]*pb.Operation)
+	}
+
+	if s.operations[kymaVpc] == nil {
+		operationpb := &pb.Operation{
+			Status: ptr.To(pb.Operation_DONE),
+			Name:   ptr.To(kymaVpc),
+		}
+		s.operations[kymaVpc] = operationpb
+	}
+
 	_, peeringExists := s.items[kymaNetwork]
 	if peeringExists {
-		return nil
+		return s.operations[kymaVpc], nil
 	}
 
 	item := &vpcPeeringEntry{
@@ -90,7 +115,7 @@ func (s *vpcPeeringStore) CreateKymaVpcPeering(ctx context.Context, remotePeerin
 
 	s.items[kymaNetwork] = item
 
-	return nil
+	return s.operations[kymaVpc], nil
 }
 
 func (s *vpcPeeringStore) GetRemoteNetworkTags(_ context.Context, vpc string, project string) ([]string, error) {
@@ -122,6 +147,10 @@ func (s *vpcPeeringStore) GetVpcPeering(ctx context.Context, remotePeeringName s
 		s.tags = make(map[string][]string)
 	}
 
+	if s.operations == nil {
+		s.operations = make(map[string]*pb.Operation)
+	}
+
 	_, peeringExists := s.items[network]
 	if !peeringExists {
 		return nil, nil
@@ -143,6 +172,17 @@ func (s *vpcPeeringStore) DeleteVpcPeering(ctx context.Context, remotePeeringNam
 	return nil
 }
 
+func (s *vpcPeeringStore) GetOperation(ctx context.Context, project string, operationId string) (*pb.Operation, error) {
+	s.m.Lock()
+	defer s.m.Unlock()
+
+	_, operationExists := s.operations[operationId]
+	if !operationExists {
+		return nil, fmt.Errorf("operation %s not found", operationId)
+	}
+	return s.operations[operationId], nil
+}
+
 // Fake client implementations to mimic google API calls
 func (s *vpcPeeringStore) SetMockVpcPeeringLifeCycleState(project string, vpc string, state pb.NetworkPeering_State) {
 	stateString := state.String()

pkg/kcp/provider/gcp/vpcpeering/checkLocalOperation.go

@@ -0,0 +1,83 @@
+package vpcpeering
+
+import (
+	"context"
+	"strings"
+
+	pb "cloud.google.com/go/compute/apiv1/computepb"
+	cloudcontrolv1beta1 "github.com/kyma-project/cloud-manager/api/cloud-control/v1beta1"
+	"github.com/kyma-project/cloud-manager/api/cloud-resources/v1beta1"
+	"github.com/kyma-project/cloud-manager/pkg/util"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	"github.com/kyma-project/cloud-manager/pkg/composed"
+)
+
+func checkLocalOperation(ctx context.Context, st composed.State) (error, context.Context) {
+	state := st.(*State)
+	logger := composed.LoggerFromCtx(ctx)
+
+	if composed.MarkedForDeletionPredicate(ctx, state) || (state.localOperation != nil && state.localOperation.GetStatus() != pb.Operation_PENDING) {
+		return nil, ctx
+	}
+
+	if state.ObjAsVpcPeering().Status.Operation != "" {
+		op, err := state.client.GetOperation(ctx, state.LocalNetwork().Spec.Network.Reference.Gcp.GcpProject, state.ObjAsVpcPeering().Status.Operation)
+		if err != nil {
+			logger.Error(err, "[KCP GCP VpcPeering checkLocalOperation] Error getting local operation")
+			meta.SetStatusCondition(state.ObjAsVpcPeering().Conditions(), metav1.Condition{
+				Type:    v1beta1.ConditionTypeError,
+				Status:  "True",
+				Reason:  v1beta1.ConditionReasonError,
+				Message: "Error loading Local Vpc Peering Operation: " + state.ObjAsVpcPeering().Status.Operation,
+			})
+			err = state.PatchObjStatus(ctx)
+			if err != nil {
+				return composed.LogErrorAndReturn(err,
+					"Error updating status since it was not possible to load the local Vpc Peering operation.",
+					composed.StopWithRequeueDelay(util.Timing.T10000ms()),
+					ctx,
+				)
+			}
+			return composed.StopWithRequeueDelay(util.Timing.T60000ms()), nil
+		}
+		state.localOperation = op
+		if op != nil {
+			if op.GetStatus() != pb.Operation_DONE {
+				logger.Info("[KCP GCP VpcPeering checkLocalOperation] Local operation still in progress", "localOperation", ptr.Deref(op.Name, "OperationUnknown"))
+				return composed.StopWithRequeueDelay(util.Timing.T60000ms()), nil
+			}
+			if op.GetError() != nil {
+				logger.Error(err, "[KCP GCP VpcPeering checkRemoteOperation] Local operation error "+op.GetName())
+				state.ObjAsVpcPeering().Status.State = cloudcontrolv1beta1.VirtualNetworkPeeringStateDisconnected
+				if strings.Contains(op.GetError().String(), "QUOTA_EXCEEDED") {
+					return composed.UpdateStatus(state.ObjAsVpcPeering()).SetExclusiveConditions(metav1.Condition{
+						Type:    v1beta1.ConditionTypeQuotaExceeded,
+						Status:  "True",
+						Reason:  v1beta1.ConditionTypeQuotaExceeded,
+						Message: "Error creating Local Vpc Peering due to quota limits, please check if your vpc quota limits are not exceeded.",
+					}).
+						ErrorLogMessage("Error creating Local VpcPeering due to quota exceeded").
+						FailedError(composed.StopWithRequeue).
+						SuccessError(composed.StopWithRequeueDelay(util.Timing.T300000ms())).
+						Run(ctx, state)
+				}
+
+				return composed.UpdateStatus(state.ObjAsVpcPeering()).SetExclusiveConditions(metav1.Condition{
+					Type:    v1beta1.ConditionTypeError,
+					Status:  "True",
+					Reason:  v1beta1.ConditionTypeError,
+					Message: "The cloud provider had an error while creating Local Vpc Peering",
+				}).
+					ErrorLogMessage("The cloud provider had an error while creating Local Vpc Peering").
+					FailedError(composed.StopWithRequeue).
+					SuccessError(composed.StopWithRequeueDelay(util.Timing.T300000ms())).
+					Run(ctx, state)
+			}
+		}
+	}
+
+	return nil, nil
+}

pkg/kcp/provider/gcp/vpcpeering/checkRemoteOperation.go

@@ -0,0 +1,55 @@
+package vpcpeering
+
+import (
+	"context"
+
+	pb "cloud.google.com/go/compute/apiv1/computepb"
+	"github.com/kyma-project/cloud-manager/api/cloud-resources/v1beta1"
+	"github.com/kyma-project/cloud-manager/pkg/util"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
+
+	"github.com/kyma-project/cloud-manager/pkg/composed"
+)
+
+func checkRemoteOperation(ctx context.Context, st composed.State) (error, context.Context) {
+	state := st.(*State)
+	logger := composed.LoggerFromCtx(ctx)
+
+	if composed.MarkedForDeletionPredicate(ctx, state) || (state.remoteOperation != nil && state.remoteOperation.GetStatus() != pb.Operation_PENDING) {
+		return nil, ctx
+	}
+
+	if state.ObjAsVpcPeering().Status.RemoteOperation != "" {
+		op, err := state.client.GetOperation(ctx, state.RemoteNetwork().Spec.Network.Reference.Gcp.GcpProject, state.ObjAsVpcPeering().Status.RemoteOperation)
+		if err != nil {
+			logger.Error(err, "[KCP GCP VpcPeering checkRemoteOperation] Error getting remote operation")
+			meta.SetStatusCondition(state.ObjAsVpcPeering().Conditions(), metav1.Condition{
+				Type:    v1beta1.ConditionTypeError,
+				Status:  "True",
+				Reason:  v1beta1.ConditionReasonError,
+				Message: "Error loading Remote Vpc Peering Operation: " + state.ObjAsVpcPeering().Status.RemoteOperation,
+			})
+			err = state.PatchObjStatus(ctx)
+			if err != nil {
+				return composed.LogErrorAndReturn(err,
+					"Error updating status since it was not possible to load the remote Vpc Peering operation.",
+					composed.StopWithRequeueDelay(util.Timing.T10000ms()),
+					ctx,
+				)
+			}
+			return composed.StopWithRequeueDelay(util.Timing.T60000ms()), nil
+		}
+		state.remoteOperation = op
+		if op != nil {
+			if op.GetStatus() != pb.Operation_DONE {
+				logger.Info("[KCP GCP VpcPeering checkRemoteOperation] Remote operation still in progress", "remoteOperation", ptr.Deref(op.Name, "OperationUnknown"))
+				return composed.StopWithRequeueDelay(util.Timing.T60000ms()), nil
+			}
+			return nil, ctx
+		}
+	}
+
+	return nil, nil
+}