package gardener

import (
	"context"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"time"

	gcpiam "cloud.google.com/go/iam"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/iam"
	iamtypes "github.com/aws/aws-sdk-go-v2/service/iam/types"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	"github.com/pingcap/errors"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
	"golang.org/x/sync/errgroup"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/utils/pointer"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/common"
	reconciler "github.com/tidbcloud/infra-api/controllers/common/error"
	"github.com/tidbcloud/infra-api/controllers/provider"
	"github.com/tidbcloud/infra-api/controllers/utils"
	"github.com/tidbcloud/infra-api/pkg/consts"
	"github.com/tidbcloud/infra-api/pkg/gardener"
	awsv1alpha1 "github.com/tidbcloud/infra-api/pkg/gardener/aws/apis/v1alpha1"
	gardencorev1beta1 "github.com/tidbcloud/infra-api/pkg/gardener/core/apis/v1beta1"
	gcpv1alpha1 "github.com/tidbcloud/infra-api/pkg/gardener/gcp/apis/v1alpha1"
	"github.com/tidbcloud/infra-api/pkg/k8s"
	"github.com/tidbcloud/infra-api/pkg/o11y"
	"github.com/tidbcloud/infra-api/pkg/progress"
	awsapi "github.com/tidbcloud/infra-api/pkg/providers/aws"
	"github.com/tidbcloud/infra-api/pkg/providers/gcp"
	tracerutil "github.com/tidbcloud/infra-api/pkg/tracer"
	"github.com/tidbcloud/infra-api/pkg/util"
	"github.com/tidbcloud/infra-api/pkg/vpc_endpoint"
)

const (
	NetworkFinalizerName          = "network.cloud.pingcap.com/finalizer"
	defaultRequeueAfter           = 10 * time.Second
	hibernatedNetworkRequeueAfter = 1 * time.Minute
	gcpNetworkResyncInterval      = 5 * time.Minute
)

var (
	WaitBucketReady = errors.New("Wait bucket ready")

	tracer = otel.Tracer("NetworkReconciler")
)

type NetworkReconciler struct {
	client.Client

	awsClient awsapi.Client
	env       gardener.GardenerAPI
	cfg       *cloudv1.InfraAPIConfig
	serverCfg cloudv1.Server

	gcpapi gcp.GCPAPI
	record record.EventRecorder

	// Shoot log agent dependencies introduced by new o11y infra.
	o11yNewAgentEnabled     bool
	manifestDirectoryPrefix string
	o11yConfig              *cloudv1.ObservabilityInfraConfig
	o11yApi                 o11y.Api
}

var _ provider.NetworkReconciler = &NetworkReconciler{}

func NewNetworkReconciler(
	client client.Client,
	env gardener.GardenerAPI,
	cfg *cloudv1.InfraAPIConfig,
	awsClient awsapi.Client,
	gcpapi gcp.GCPAPI,
	record record.EventRecorder,
	o11yNewAgentEnabled bool,
	manifestDirectoryPrefix string,
	o11yConfig *cloudv1.ObservabilityInfraConfig,
	o11yApi o11y.Api,
) *NetworkReconciler {
	r := &NetworkReconciler{
		Client:                  client,
		env:                     env,
		serverCfg:               cfg.Server,
		cfg:                     cfg,
		awsClient:               awsClient,
		gcpapi:                  gcpapi,
		record:                  record,
		o11yNewAgentEnabled:     o11yNewAgentEnabled,
		manifestDirectoryPrefix: manifestDirectoryPrefix,
		o11yConfig:              o11yConfig,
		o11yApi:                 o11yApi,
	}

	return r
}

func (r *NetworkReconciler) Reconcile(
	ctx context.Context,
	providerType *cloudv1.ProviderType,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
) (ctrlResult ctrl.Result, errResult error) {
	start := time.Now()
	log := log.FromContext(ctx)
	log.Info("start reconcile")

	var span trace.Span
	ctx, span = tracer.Start(ctx, "Network Reconcile")
	span.SetAttributes(
		attribute.String("project_id", network.Labels[consts.ProjectTagKey]),
		attribute.String("tenant_id", network.Labels[consts.TenantTagKey]),
		attribute.String("network_namespace", network.Namespace),
		attribute.String("network_name", network.Name),
		attribute.String("network_phase", network.Status.Phase),
		attribute.String("provider", providerType.Name),
	)

	defer func() {
		info := []interface{}{"take", time.Since(start), "network", network.Name}
		if errResult != nil {
			info = append(info, "err", fmt.Sprintf("%T", errors.Cause(errResult)))
			span.RecordError(errResult)
			span.SetStatus(codes.Error, errResult.Error())
		}

		if errResult == nil {
			info = append(info, "result", ctrlResult)
		}
		log.Info("finish reconcile", info...)
		span.End()
	}()

	if network.ObjectMeta.DeletionTimestamp.IsZero() {
		span.AddEvent("Add finalizer")
		if err := common.EnsureFinalizers(ctx, r.Client, network, NetworkFinalizerName); err != nil {
			return ctrl.Result{}, err
		}

		switch network.Status.Phase {
		case "":
			network.Status.Phase = cloudv1.NetworkPhaseCreating
			log.Info("update status to creating", "status", network.Status)
			err := r.Status().Update(ctx, network)
			if err != nil {
				return ctrl.Result{}, err
			}
		case cloudv1.NetworkPhaseCreating:
		case cloudv1.NetworkPhaseNormal:
		case cloudv1.NetworkPhasePendingDelete, cloudv1.NetworkPhaseDeleting:
			return ctrl.Result{}, errors.Errorf("impossible phase %s", network.Status.Phase)
		}

		// Encapsulates the `setUp` function to instrument
		err := func(ctx context.Context) error {
			spanCtx, span := tracer.Start(ctx, "setUp network")
			defer span.End()

			e := r.setUp(spanCtx, providerType, provider, network)
			if e != nil {
				span.RecordError(e)
				span.SetStatus(codes.Error, e.Error())
			}
			return e
		}(ctx)
		if err != nil {
			// Note the shoot need about more than 10min to become ready
			if errors.Cause(err) == gardener.ErrShootNotReady {
				span.AddEvent("Wait for gardener shoot to become ready")
				log.Info("wait for gardener shoot to become ready", "err", errors.Cause(err).Error())
				return ctrl.Result{RequeueAfter: defaultRequeueAfter}, nil
			}
			return ctrl.Result{}, err
		}

		updated, err := r.updateStatus(ctx, provider, network)
		if err != nil {
			if err == WaitBucketReady {
				span.AddEvent("Wait bucket ready")
				r.record.Event(network, corev1.EventTypeNormal, "WaitBucketReady", "Wait bucket ready")
				return ctrl.Result{RequeueAfter: defaultRequeueAfter}, nil
			}
			return ctrl.Result{}, errors.Annotate(err, "update network status")
		}
		if updated {
			log.Info("update status", "status", network.Status)
			if err := r.Status().Update(ctx, network); err != nil {
				// if the status update failed, we should requeue immediate
				return ctrl.Result{RequeueAfter: defaultRequeueAfter}, nil
			}
			r.record.Event(network, corev1.EventTypeNormal, "SuccessfulReconcile", fmt.Sprintf("Successful reconcile network %s", network.Name))
		}
	} else {
		switch network.Status.Phase {
		case "", cloudv1.NetworkPhaseNormal, cloudv1.NetworkPhaseCreating:
			network.Status.Phase = cloudv1.NetworkPhasePendingDelete
			log.Info("update status to pending delete", "status", network.Status)
			err := r.Status().Update(ctx, network)
			if err != nil {
				return ctrl.Result{}, err
			}

			return ctrl.Result{Requeue: true}, nil
		case cloudv1.NetworkPhasePendingDelete:
			// check all clusters belong to this network is deleted.
			clusters, err := getNetworkClusters(ctx, r.Client, network.Namespace, network.Name)
			if err != nil {
				return ctrl.Result{}, err
			}

			if len(clusters) != 0 {
				var names []string
				for _, c := range clusters {
					names = append(names, c.Name)
				}
				r.record.Event(network, corev1.EventTypeNormal, "DeleteNetwork", fmt.Sprintf("waiting clusters to be deleted, still existing: %s", names))
				return ctrl.Result{RequeueAfter: time.Second * 10}, nil
			}

			backupNames, err := getNetworkBackupNames(ctx, r.Client, network)
			if err != nil {
				return ctrl.Result{}, err
			}
			if len(backupNames) > 0 {
				r.record.Event(network, corev1.EventTypeNormal, "DeleteNetwork", fmt.Sprintf("wait backups to be deleted, still existing: %s", backupNames))
				return ctrl.Result{RequeueAfter: time.Second * 10}, nil
			}

			npNames, err := getNetworkPeerNames(ctx, r.Client, network)
			if err != nil {
				return ctrl.Result{}, err
			}
			if len(npNames) > 0 {
				r.record.Event(network, corev1.EventTypeNormal, "DeleteNetwork", fmt.Sprintf("wait network peer to be deleted, still existing: %s", npNames))
				return ctrl.Result{RequeueAfter: time.Second * 10}, nil
			}

			network.Status.Phase = cloudv1.NetworkPhaseDeleting
			log.Info("update status to deleting", "status", network.Status)
			err = r.Status().Update(ctx, network)
			if err != nil {
				return ctrl.Result{}, err
			}

			return ctrl.Result{Requeue: true}, nil
		case cloudv1.NetworkPhaseDeleting:
			// just let it handle deleting below
		}

		info, err := r.buildContextInfo(provider, network)
		if err != nil {
			return ctrl.Result{}, err
		}

		// Encapsulates the `cleanUp` function to instrument
		err = func(ctx context.Context) error {
			spanCtx, span := tracer.Start(ctx, "cleanUp network")
			defer span.End()

			e := r.cleanUp(spanCtx, provider, network, info)
			if e != nil {
				span.RecordError(e)
				span.SetStatus(codes.Error, e.Error())
			}
			return e
		}(ctx)
		if err != nil {
			return ctrl.Result{}, err
		}

		shootName, err := shootName(network, provider.Spec.Type, network.GetProjectId())
		if err != nil {
			return ctrl.Result{}, err
		}

		// make sure shoot being deleted before we remove finalizer.
		// TODO: watch shoot CRD and reconcile the relate network to avoid check in tick.
		exist, err := r.env.ShootClusterIsExist(ctx, shootName)
		if err != nil {
			return ctrl.Result{}, err
		}

		if exist {
			log.Info("wait for shoot deletion", "shoot", shootName)
			span.AddEvent("wait for shoot deletion", trace.WithAttributes(attribute.String("shoot", shootName)))
			return ctrl.Result{RequeueAfter: time.Second * 5}, nil
		}

		span.AddEvent("Remove finalizer")
		controllerutil.RemoveFinalizer(network, NetworkFinalizerName)
		if err := r.Update(ctx, network); err != nil {
			return ctrl.Result{}, err
		}
	}

	if provider.Spec.Type == consts.ProviderTypeGCP {
		return ctrl.Result{RequeueAfter: gcpNetworkResyncInterval}, nil
	}

	return ctrl.Result{}, nil
}

type contextInfo struct {
	imageRepoPrefix    string
	reporterSQSRoleARN string

	pcaRoleARN string
	// infra project id, not gcp project id
	projectId string
}

func (r *NetworkReconciler) buildContextInfo(provider *cloudv1.Provider, network *cloudv1.Network) (*contextInfo, error) {
	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return nil, errors.Annotate(err, "gardener spec")
	}

	info := &contextInfo{
		projectId: network.GetProjectId(),
	}
	accountID := spec.AWS.AccountID
	project := r.cfg.Gardener.GardenerProjectName
	shootName, err := shootName(network, provider.Spec.Type, network.GetProjectId())
	if err != nil {
		return nil, errors.Annotate(err, "shoot name")
	}

	info.pcaRoleARN = fmt.Sprintf("arn:aws:iam::%s:role/%s", accountID, getGCPPCARoleName(project, network.GetProjectId(), shootName))

	switch provider.Spec.Type {
	case consts.ProviderTypeAWS:
		info.imageRepoPrefix = fmt.Sprintf("%s.dkr.ecr.%s.amazonaws.com", accountID, spec.AWS.ECRRegion)
		info.reporterSQSRoleARN = fmt.Sprintf("arn:aws:iam::%s:role/dbaas-%s-%s-%s-sqs-writer", accountID, project, network.GetProjectId(), shootName)
	case consts.ProviderTypeGCP:
		info.imageRepoPrefix = "gcr.io/" + spec.GCP.ProjectID
		info.reporterSQSRoleARN = fmt.Sprintf("arn:aws:iam::%s:role/%s", accountID, getGCPReporterAWSRoleName(project, network.GetProjectId(), shootName))
	default:
		return nil, errors.Errorf("unknown provider type: %s", provider.Spec.Type)
	}

	return info, nil
}

func (r *NetworkReconciler) progress(ctx context.Context, network *cloudv1.Network, name string) {
	if network.Status.Phase != cloudv1.NetworkPhaseCreating {
		return
	}

	updated := progress.UpdateAnnForFirstPassIfNeed(network, name)
	if !updated {
		return
	}

	logger := log.FromContext(ctx)
	logger.Info("pass progress when creating network", "name", name, "sinceCreate", time.Since(network.CreationTimestamp.Time))
	err := r.Client.Update(ctx, network)
	if err != nil {
		log.FromContext(ctx).Error(err, "failed to update network")
	}
}

func (r *NetworkReconciler) setUp(ctx context.Context, providerType *cloudv1.ProviderType, provider *cloudv1.Provider, network *cloudv1.Network) error {
	logger := log.FromContext(ctx)

	// create shoot
	shootName, err := shootName(network, provider.Spec.Type, network.GetProjectId())
	if err != nil {
		return errors.Annotate(err, "shootName")
	}

	if _, err := r.updateGardenerStatus(ctx, network, func(g *cloudv1.GardenerNetworkStatus) {
		if g.ShootName == "" {
			g.ShootName = shootName
		}
	}); err != nil {
		return errors.Annotate(err, "update Gardener Status")
	}

	region, err := provider.FindRegion(network.Spec.Region)
	if err != nil {
		return errors.Annotate(err, "FindRegion")
	}

	var vpcCidrBlock string

	var workerCIDR []string
	var publicCIDR []string
	var internalCIDR []string
	switch provider.Spec.Type {
	case consts.ProviderTypeAWS:
		workerCIDR, publicCIDR, internalCIDR, err = GetAWSCIDR(network.Spec.CIDR, region.Name, len(region.Zones))
		if err != nil {
			return errors.Annotate(err, "get AWS CIDR")
		}
		vpcCidrBlock = network.Spec.CIDR

	case consts.ProviderTypeGCP:
		workerCIDR, internalCIDR, err = getGCPCIDR(network.Spec.CIDR, region.Name)
		if err != nil {
			return errors.Annotate(err, "get GCP CIDR")
		}
		vpcCidrBlock = internalCIDR[0]
	}

	op, err := r.env.CreateOrUpdateShoot(ctx, &gardener.ShootArgs{
		Name:           shootName,
		ProviderType:   provider.Spec.Type,
		Region:         network.Spec.Region,
		Zones:          region.Zones,
		Labels:         network.Labels,
		Cidr:           network.Spec.CIDR,
		WorkerCidr:     workerCIDR,
		PublicLBCidr:   publicCIDR,
		InternalLBCidr: internalCIDR,
	})
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedSetUpShoot", err.Error())
		return errors.Annotate(err, "create/update shoot")
	} else if op == controllerutil.OperationResultCreated {
		r.record.Event(network, corev1.EventTypeNormal, "SetUpShoot", fmt.Sprintf("Setup shoot %s successful", shootName))
	}

	op, err = r.updateGardenerStatus(ctx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.VPCCidrBlock = vpcCidrBlock
	})
	if err != nil {
		return errors.Annotate(err, "update Gardener status")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureShootCidr", fmt.Sprintf("Ensure shoot cidr %s successful", vpcCidrBlock))
	}

	// Maintain GCP Firewall when waiting shoot waking up, GCP shoot NAT address may change in waking up
	shoot, err := r.getShoot(ctx, shootName)
	if err != nil {
		return errors.Annotate(err, "getShoot")
	}
	err = r.MaintainFirewall(ctx, &shoot, network, provider)
	if err != nil {
		return errors.Annotate(err, "maintainFirewall")
	}

	sctx, err := r.env.GetShootContext(ctx, shootName)
	if err != nil {
		// skip reconcile when shoot already hibernation
		if err == gardener.ErrShootAlreadyHibernation {
			if r.o11yNewAgentEnabled && utils.MatchRegion(network.Spec.Provider, network.Spec.Region, shootName, r.o11yConfig.EffectiveRegion) {
				// If shoot is hibernated, o11y infra should be deleted.
				// Please refer to https://gardener.cloud/docs/gardener/usage/shoot_hibernate/ for more details.
				// Why we hibernate shoot rather than delete it? Because our customers rarely delete project manually.
				providerSpec, err := cloudv1.GardenerProviderSpecForProvider(provider)
				if err != nil {
					return errors.Annotate(err, "failed to get gardener provider spec")
				}
				err = vpc_endpoint.CleanupEndpoint(ctx, r.Client, shoot.Status.TechnicalID, network, r.gcpapi, providerSpec, shootName)
				if err != nil {
					return errors.Annotate(err, "failed to cleanup endpoint")
				}
			}

			new := network.DeepCopy()
			if new.Status.Phase != cloudv1.NetworkPhaseHibernated {
				new.Status.Phase = cloudv1.NetworkPhaseHibernated
				if err := r.Status().Update(ctx, new); err != nil {
					return fmt.Errorf("failed to update network status to hibernated, error: %v", err)
				}
			}
			return &reconciler.RequeueAfterError{
				RequeueAfter: hibernatedNetworkRequeueAfter,
				Caused:       err,
			}
		}

		if err == gardener.ErrShootWakingUp {
			new := network.DeepCopy()
			if new.Status.Phase != cloudv1.NetworkPhaseWakeUp {
				new.Status.Phase = cloudv1.NetworkPhaseWakeUp
				if err := r.Status().Update(ctx, new); err != nil {
					return fmt.Errorf("failed to update network status to waking up, error: %v", err)
				}
			}

			return &reconciler.RequeueAfterError{
				RequeueAfter: defaultRequeueAfter,
				Caused:       err,
			}
		}
		errAnn := errors.Annotate(err, "Get Shoot Context")
		if apierrors.IsNotFound(errors.Cause(err)) || errors.Cause(err) == gardener.ErrShootNotReady {
			return &reconciler.RequeueAfterError{
				RequeueAfter: defaultRequeueAfter,
				Caused:       errAnn,
			}
		}
		return errAnn
	}

	info, err := r.buildContextInfo(provider, network)
	if err != nil {
		return errors.Annotate(err, "buildContextInfo")
	}
	r.progress(ctx, network, "Shoot")

	shootCli := sctx.Manager.GetClient()

	group, _ := errgroup.WithContext(ctx)
	group.Go(func() error {
		op, err := r.ensureCertManager(ctx, shootCli, sctx.Kubeconfig, provider, info)
		if err != nil {
			if _, ok := errors.Cause(err).(WaitWebhookReadyError); ok {
				r.record.Event(network, corev1.EventTypeNormal, "EnsureCertManager", err.Error())
				logger.Info("wait for webhook to become ready", "err", errors.Cause(err).Error())
				return &reconciler.RequeueAfterError{
					RequeueAfter: defaultRequeueAfter,
					Caused:       nil,
				}
			} else {
				r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureCertManager", err.Error())
			}
			return errors.Annotate(err, "ensureCertManager")
		} else if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureCertManager", "Ensure cert-manager successful")
		}
		return nil
	})

	group.Go(func() error {
		op, err := r.ensureTiDBOperator(ctx, shootCli, sctx.Kubeconfig, shoot.Spec.Kubernetes.Version, shoot.Spec.Region)
		if err != nil {
			r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureTiDBOperator", err.Error())
			return errors.Annotate(err, "ensureTiDBOperator")
		} else if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureTiDBOperator", "Ensure tidb operator successful")
		}

		return nil
	})

	err = group.Wait()
	if err != nil {
		return errors.Annotate(err, "setup group")
	}
	r.progress(ctx, network, "TiDBOperator-CertManager")

	op, err = r.ensureStorageClass(ctx, shootCli, sctx.Kubeconfig, provider.Spec.Type, shoot.Spec.Kubernetes.Version)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureStorageClass", err.Error())
		return errors.Annotate(err, "ensureStorageClass")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureStorageClass", "Ensure storage class successful")
	}
	r.progress(ctx, network, "StorageClass")

	op, err = r.ensureLocalProxy(ctx, shootCli, provider, info, network.Spec.Region)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureLocalProxy", err.Error())
		return errors.Annotate(err, "ensureLocalProxy")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureLocalProxy", "Ensure local proxy successful")
	}
	r.progress(ctx, network, "LocalProxy")

	op, err = r.ensureTiDBClusterSLIReporter(ctx, shootCli, info, provider)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureTiDBClusterSLIReporter", err.Error())
		return errors.Annotate(err, "ensureTiDBClusterSLIReporter")
	} else if op == controllerutil.OperationResultCreated {
		// Deployment reporter always update, we only reporter event when created
		r.record.Event(network, corev1.EventTypeNormal, "EnsureTiDBClusterSLIReporter", "Ensure tidb cluster SLI reporter successful")
	}
	r.progress(ctx, network, "SLIReporter")

	op, err = r.ensureGracefulShutdownWebhook(ctx, shootCli, provider)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedGracefulShutdownWebhook", err.Error())
		return errors.Annotate(err, "ensureGracefulShutdownWebhook")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureGracefulShutdownWebhook", "Ensure graceful shutdown webhook successful")
	}
	r.progress(ctx, network, "GracefulShutdownWebhook")

	op, err = r.ensureBucket(ctx, provider, network, info)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureBucket", err.Error())
		return errors.Annotate(err, "ensureBucket")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureBucket", "Ensure bucket successful")
	}
	r.progress(ctx, network, "Bucket")

	err = r.ensureRoles(ctx, provider, network, shootName, info)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureRoles", err.Error())
		return errors.Annotate(err, "ensureRoles")
	}
	r.progress(ctx, network, "Roles")

	err = r.ensureIAM(ctx, provider, network, info)
	if err != nil {
		return errors.Annotate(err, "ensureIAM")
	}
	logger.Info("ensure IAM")
	r.progress(ctx, network, "IAM")

	// gcp agent needs iam role to access gcs/s3 so `ensureDataflowAgent` must be called after `ensureIAM`
	op, err = r.ensureDataflowAgent(ctx, shootCli, provider, network, info)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureDataflowAgent", err.Error())
		return errors.Annotate(err, "ensureDataflowAgent")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureDataflowAgent", "Ensure dataflow agent successful")
	}
	r.progress(ctx, network, "DataFlowAgent")

	// ref https://github.com/tidbcloud/dbaas/blob/f1d424a7a2352979add9b7181a684b8105390413/central/runtime/deployer/gardener/shoot_reconciler.go#L260
	// Prepare KMS and GCS credentials in tidb-admin namespace, which will be shared by the tenant's clusters
	// TODO: migrate to GCP KMS and ADC to eliminate static credentials
	if provider.Spec.Type == consts.ProviderTypeGCP {
		err = r.prepareGCPCredentials(ctx, shootCli, shootName, network, provider, info.projectId)
		if err != nil {
			r.record.Event(network, corev1.EventTypeWarning, "FailedPrepareGCPCredentials", err.Error())
			return errors.Annotate(err, "prepare GCP Credentials")
		}
		r.progress(ctx, network, "GCPCredentials")
	}

	providerSpec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return errors.Annotate(err, "failed to get gardener provider spec")
	}

	// ensure GCP network dns zone and update status
	if provider.Spec.Type == consts.ProviderTypeGCP {
		gcpInfraStatus, err := gcpv1alpha1.InfraStatusFromShoot(&shoot)
		if err != nil {
			return errors.Annotate(err, "GCP Infra Status")
		}
		if gcpInfraStatus.Networks.VPC.Name == "" {
			return gardener.ErrShootNotReady
		}
		if err := r.ensureGCPDnsZone(ctx, provider, network, gcpInfraStatus.Networks.VPC.Name); err != nil {
			r.record.Event(network, corev1.EventTypeWarning, "FailedEnsureGCPDnsZone", err.Error())
			return errors.Annotate(err, "GCP DNS Zone")
		}
		op, err := r.updateGardenerStatus(ctx, network, func(g *cloudv1.GardenerNetworkStatus) {
			g.GCP.NetworkName = pointer.String(gcpInfraStatus.Networks.VPC.Name)
			g.GCP.ProjectID = pointer.String(providerSpec.GCP.ProjectID)
		})
		if err != nil {
			return errors.Annotate(err, "GCP Gardener Status")
		} else if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureGCPVPCNetwork", "Ensure gcp vpc network successful")
		}
	}

	// ensure AWS VPC dns zone, account ID and update status
	if provider.Spec.Type == consts.ProviderTypeAWS {
		awsInfraStatus, err := awsv1alpha1.InfraStatusFromShoot(&shoot)
		if err != nil {
			return errors.Annotate(err, "AWS Infra Status")
		}
		if awsInfraStatus.VPC.ID == "" {
			return gardener.ErrShootNotReady
		}
		op, err := r.updateGardenerStatus(ctx, network, func(g *cloudv1.GardenerNetworkStatus) {
			g.AWS.VpcID = pointer.String(awsInfraStatus.VPC.ID)
			g.AWS.AccountID = pointer.String(providerSpec.AWS.AccountID)
			var sgs []cloudv1.SecurityGroup
			for _, sg := range awsInfraStatus.VPC.SecurityGroups {
				item := cloudv1.SecurityGroup{
					Purpose: sg.Purpose,
					ID:      sg.ID,
				}
				sgs = append(sgs, item)
			}
			g.AWS.SecurityGroups = sgs
		})
		if err != nil {
			return errors.Annotate(err, "AWS update Gardener Status")
		} else if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureAWSVPCNetwork", "Ensure aws vpc network successful")
		}
	}
	r.progress(ctx, network, "DNSZone")

	if r.o11yNewAgentEnabled && utils.MatchRegion(network.Spec.Provider, network.Spec.Region, shootName, r.o11yConfig.EffectiveRegion) {
		// Deploy s3 gateway endpoint for the whole vpc
		if provider.Spec.Type == consts.ProviderTypeAWS {
			awsInfraStatus, err := awsv1alpha1.InfraStatusFromShoot(&shoot)
			if err != nil {
				return errors.Annotate(err, "AWS Infra Status")
			}
			if awsInfraStatus.VPC.ID == "" {
				return gardener.ErrShootNotReady
			}
			// create s3 gateway endpoint
			op, _, err := vpc_endpoint.DeployS3GatewayEndpoint(ctx, r.Client, r.awsClient.Ec2(), awsInfraStatus.VPC.ID, shoot.Status.TechnicalID, network.Labels, network, providerSpec.AWS.CrossplaneProvider, r.o11yConfig.Version)
			if err != nil {
				return errors.Annotate(err, "failed to deploy S3 gateway endpoint")
			}
			if op != controllerutil.OperationResultNone {
				r.record.Event(network, corev1.EventTypeNormal, "DeployS3GatewayEndpoint", "Deploy s3 gateway endpoint successful")
			}
		}

		status, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
		if err != nil {
			return errors.Annotate(err, "failed to get gardener network status")
		}

		// Deploy gcs psc endpoint for the whole vpc.
		// Cross-plane provider gcp lacks forwarding rule crd, so we create psc via api directly.
		if provider.Spec.Type == consts.ProviderTypeGCP {
			if err := vpc_endpoint.DeployGApiPscEndpoint(ctx, r.Client, r.gcpapi, network, r.o11yConfig.Version, providerSpec.GCP.ProjectID, *status.GCP.NetworkName, shootName); err != nil {
				return errors.Annotate(err, "failed to deploy google api psc endpoint")
			}
		}

		// Deploy shoot log agent.
		op, err = r.ensureLogAgent(ctx, network, sctx.Kubeconfig, &shoot, status.NodeRole, shootCli)
		if err != nil {
			return errors.Annotate(err, "failed to ensure log agent")
		}
		if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureNetwork log agent", "Ensure network log agent successful")
		}

		r.progress(ctx, network, "O11Y")
	}

	// wait local proxy ready and update the status
	op, err = r.observeLocalProxy(ctx, shootCli, provider, network)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailObserveLocalProxy", err.Error())
		return errors.Annotate(err, "observeLocalProxy")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "ObserveLocalProxy", "Observe local proxy successful")
	}
	r.progress(ctx, network, "OBservedLocalProxy")

	// wait dataflow agent ready and update the status
	op, err = r.observeDataflowAgent(ctx, shootCli, provider, network)
	if err != nil {
		r.record.Event(network, corev1.EventTypeWarning, "FailObserveDataflowAgent", err.Error())
		return errors.Annotate(err, "observeLocalProxy")
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "ObserveDataflowAgent", "Observe dataflow dataflow agent successful")
	}
	r.progress(ctx, network, "ObserveDataflowAgent")

	return nil
}

func (r *NetworkReconciler) MaintainFirewall(ctx context.Context, shoot *gardencorev1beta1.Shoot, network *cloudv1.Network, provider *cloudv1.Provider) (errResult error) {
	_, span := tracer.Start(ctx, "MaintainFirewall")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	var sourceRangeIps []string
	if r.cfg.CentralCIDR == nil {
		// default to whitelist all if not configured
		sourceRangeIps = append(sourceRangeIps, "0.0.0.0/0")
	} else {
		sourceRangeIps = append(sourceRangeIps, r.cfg.CentralCIDR...)
	}
	if provider.Spec.Type == consts.ProviderTypeGCP {
		gcpInfraStatus, err := gcpv1alpha1.InfraStatusFromShoot(shoot)
		if err != nil {
			return errors.Annotate(err, "GCP Infra Status")
		}
		if gcpInfraStatus.Networks.VPC.Name == "" {
			return gardener.ErrShootNotReady
		}
		spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
		if err != nil {
			return err
		}
		ips, err := r.gcpapi.ListNat(spec.GCP.ProjectID, gcpInfraStatus.Networks.VPC.Name, network.Spec.Region)
		if err != nil {
			return errors.Annotatef(err, fmt.Sprintf("list nat ips for shoot %s failed", shoot.Name))
		}

		// For gcp, Some clients that reside access api-server will not go through NAT, so for general cases we have to add "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16".
		ips = append(ips, "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")
		sourceRangeIps = append(sourceRangeIps, ips...)
	}

	if err := r.env.UpdateShootFirewall(ctx, shoot.Name, sourceRangeIps); err != nil {
		return errors.Annotatef(err, "Update shoot firewall")
	}
	return nil
}

func (r *NetworkReconciler) getShoot(
	ctx context.Context,
	shootName string,
) (gardencorev1beta1.Shoot, error) {
	shoot := new(gardencorev1beta1.Shoot)
	gardenerCli := r.env.Manager().GetClient()
	err := gardenerCli.Get(ctx, types.NamespacedName{Namespace: r.env.GetShootNamespace(), Name: shootName}, shoot)
	return *shoot, errors.Annotate(err, "getShoot")
}

func (r *NetworkReconciler) cleanUp(
	ctx context.Context,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
	info *contextInfo,
) error {
	logger := log.FromContext(ctx)

	shootName, err := shootName(network, provider.Spec.Type, info.projectId)
	if err != nil {
		return errors.Annotate(err, "shootName")
	}
	providerSpec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return errors.Annotate(err, "failed to get gardener provider spec")
	}

	if shoot, err := r.getShoot(ctx, shootName); err == nil {
		if r.o11yNewAgentEnabled && utils.MatchRegion(network.Spec.Provider, network.Spec.Region, shootName, r.o11yConfig.EffectiveRegion) {
			// If shoot exist, try to delete s3 gateway endpoint and o11y infra.
			err = vpc_endpoint.CleanupEndpoint(ctx, r.Client, shoot.Status.TechnicalID, network, r.gcpapi, providerSpec, shootName)
			if err != nil {
				return errors.Annotate(err, "failed to cleanup endpoint")
			}
			err = r.cleanupLogAgent(ctx, network, r.o11yApi)
			if err != nil {
				return errors.Annotate(err, "failed to cleanup log agent")
			}
		}
	}

	alreadyDeleted, err := r.env.DeleteShoot(ctx, shootName)
	if apierrors.IsNotFound(err) {
		// Wait for the shoot to be deleted before exit the node protector controller
		// Node protector controller will remove instance protector, then gardener machine-controller can terminating instance succeful
		r.env.CancelShootContext(shootName)
	}
	if client.IgnoreNotFound(errors.Cause(err)) != nil {
		annErr := errors.Annotatef(err, "DeleteShoot")
		if apierrors.IsConflict(errors.Cause(err)) {
			return &reconciler.RequeueAfterError{
				RequeueAfter: defaultRequeueAfter,
				Caused:       annErr,
			}
		}
		return annErr
	}
	if !alreadyDeleted {
		logger.Info("deleted shoot")
	}

	bucketName := getBucketName(r.env.ProjectName(), network.Spec.Region, info.projectId)
	err = cleanupBucket(ctx, r.Client, r.Scheme(), provider.Spec.Type, bucketName)
	if client.IgnoreNotFound(errors.Cause(err)) != nil {
		return err
	}

	err = r.cleanUpRoles(ctx, provider, network, shootName, info)
	if err != nil {
		return errors.Annotate(err, "cleanup Roles")
	}
	err = r.cleanUpIAM(ctx, provider, network, info)
	if err != nil {
		return errors.Annotate(err, "cleanUp IAM")
	}

	return nil
}

// updateGardenerStatus checkpoint the current reconcile state to the status
func (r *NetworkReconciler) updateGardenerStatus(ctx context.Context, network *cloudv1.Network, mutateFn func(g *cloudv1.GardenerNetworkStatus)) (controllerutil.OperationResult, error) {
	status, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
	if err != nil {
		return controllerutil.OperationResultNone, err
	}
	old := status.DeepCopy()
	mutateFn(status)

	// skip updating if there is no mutation
	if reflect.DeepEqual(old, status) {
		return controllerutil.OperationResultNone, nil
	}

	cloudv1.SetGardenerNetworkStatus(network, status)
	log.FromContext(ctx).Info("update status gardener", "status", network.Status)
	return controllerutil.OperationResultUpdated, r.Status().Update(ctx, network)
}

func (r *NetworkReconciler) ensureTiDBOperator(ctx context.Context, cli client.Client, kubeconfig []byte, k8sVersion string, region string) (controllerutil.OperationResult, error) {
	return deployOperator(ctx, cli, kubeconfig, consts.LocalOperatorNamespace, r.cfg.Server.ManifestDirectoryPrefix, k8sVersion, region)
}

func (r *NetworkReconciler) ensureStorageClass(
	ctx context.Context,
	cli client.Client,
	kubeconfig []byte,
	providerType string,
	k8sVersion string,
) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureStorageClass")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	var (
		op  controllerutil.OperationResult
		err error
	)

	err = ensureStorageClasses(spanCtx, cli, providerType, k8sVersion)
	if err != nil {
		return op, err
	}

	kubeconfigFilename, err := k8s.WriteTempFile(kubeconfig)
	if err != nil {
		return op, err
	}
	defer os.Remove(kubeconfigFilename)

	var storageClass string
	switch providerType {
	case consts.ProviderTypeAWS:
		storageClass = filepath.Join(r.serverCfg.ManifestDirectoryPrefix, "/manifests/storage-classes.yaml")
	case consts.ProviderTypeGCP:
		storageClass = filepath.Join(r.serverCfg.ManifestDirectoryPrefix, "/manifests/gcp-storage-classes.yaml")
	default:
		return op, errors.Errorf("unknown provider type: %s", providerType)
	}

	kubectl := k8s.NewKubectl().SetDir(r.serverCfg.ManifestDirectoryPrefix).WithKubeConfig(kubeconfigFilename)

	op, out, err := kubectl.ApplyWithFile(storageClass)
	if err != nil {
		return op, errors.Errorf("failed to install storageclass: output: %s, err: %s", out, err.Error())
	}

	return op, nil
}

func (r *NetworkReconciler) ensureLocalProxy(ctx context.Context, cli client.Client, provider *cloudv1.Provider, info *contextInfo, region string) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureLocalProxy")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener spec")
	}

	image := fmt.Sprintf("%s/%s", info.imageRepoPrefix, spec.Proxy.Image)
	return deployLocalProxy(
		spanCtx,
		cli,
		image,
		region,
		provider.Spec.Type,
		spec.Proxy.LoadBalancerSourceRanges,
		spec.Proxy.Auth0Audience,
		spec.Proxy.Auth0Domain,
	)
}

func (r *NetworkReconciler) observeLocalProxy(ctx context.Context, cli client.Client, provider *cloudv1.Provider, network *cloudv1.Network) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "observeLocalProxy")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	proxySvc := &corev1.Service{}
	if err := cli.Get(spanCtx, types.NamespacedName{Namespace: consts.LocalOperatorNamespace, Name: localProxyName}, proxySvc); err != nil {
		if apierrors.IsNotFound(errors.Cause(err)) {
			return controllerutil.OperationResultNone, &reconciler.RequeueAfterError{Caused: err, RequeueAfter: defaultRequeueAfter}
		}
		return controllerutil.OperationResultNone, errors.Annotate(err, "get proxy service")
	}
	if inIntegrationTest() {
		// mock the lb of proxy ready
		return r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
			g.ProxyBaseUrl = "mock.pingcap.com"
		})
	}
	ings := proxySvc.Status.LoadBalancer.Ingress
	if len(ings) < 1 {
		return controllerutil.OperationResultNone, &reconciler.RequeueAfterError{
			RequeueAfter: defaultRequeueAfter,
			Caused:       errors.Errorf("Waiting for Ingress Load Balancer"),
		}
	}
	var proxyEp string
	switch provider.Spec.Type {
	case consts.ProviderTypeGCP:
		proxyEp = ings[0].IP
	case consts.ProviderTypeAWS:
		proxyEp = ings[0].Hostname
	default:
		return controllerutil.OperationResultNone, errors.Errorf("unsupported provider %s", provider.Spec.Type)
	}
	return r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.ProxyBaseUrl = proxyEp
	})
}

func (r *NetworkReconciler) ensureDataflowAgent(
	ctx context.Context,
	shootCli client.Client,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
	info *contextInfo) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureDataflowAgent")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener spec")
	}

	// NOTE: we reuse local proxy's tls certs to save money, so dataflow agent must be deployed after local proxy
	image := spec.DataFlowAgent.Image
	return deployDataflowAgent(
		spanCtx,
		r.Client,
		shootCli,
		image,
		network,
		spec.Proxy.LoadBalancerSourceRanges,
	)
}

func (r *NetworkReconciler) observeDataflowAgent(ctx context.Context, cli client.Client, provider *cloudv1.Provider, network *cloudv1.Network) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "observeDataflowAgent")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	svc := &corev1.Service{}
	if err := cli.Get(spanCtx, types.NamespacedName{Namespace: consts.LocalOperatorNamespace, Name: dataflowAgentSvcName}, svc); err != nil {
		if apierrors.IsNotFound(errors.Cause(err)) {
			return controllerutil.OperationResultNone, &reconciler.RequeueAfterError{Caused: err, RequeueAfter: defaultRequeueAfter}
		}
		return controllerutil.OperationResultNone, errors.Annotate(err, "get dataflow agent service")
	}
	// TODO add new field
	if inIntegrationTest() {
		// mock the lb of proxy ready
		return r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
			g.DataFlowAgentURL = "mock.pingcap.com"
		})
	}
	ings := svc.Status.LoadBalancer.Ingress
	if len(ings) < 1 {
		return controllerutil.OperationResultNone, &reconciler.RequeueAfterError{
			RequeueAfter: defaultRequeueAfter,
			Caused:       errors.Errorf("Waiting for Ingress Load Balancer"),
		}
	}
	var proxyEp string
	switch provider.Spec.Type {
	case consts.ProviderTypeGCP:
		proxyEp = ings[0].IP
	case consts.ProviderTypeAWS:
		proxyEp = ings[0].Hostname
	default:
		return controllerutil.OperationResultNone, errors.Errorf("unsupported provider %s", provider.Spec.Type)
	}
	return r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.DataFlowAgentURL = proxyEp
	})
}

// ensureTiDBClusterSLIReporter deploys the SLI Reporter to the shoot cluster.
func (r *NetworkReconciler) ensureTiDBClusterSLIReporter(ctx context.Context, cli client.Client, info *contextInfo, provider *cloudv1.Provider) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureTiDBClusterSLIReporter")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener spec")
	}

	return deployTiDBClusterSLIReporter(
		spanCtx,
		cli,
		info.imageRepoPrefix,
		spec.SLIReporter.Image,
		provider.Spec.Type,
		info.reporterSQSRoleARN,
		spec.SLIReporter.SQSEndpoint,
		spec.SLIReporter.Region,
		r.cfg.IdentityImage)
}

// return updated as true if we change the network status, it's caller's responsibility to update the status to api server if need.
func (r *NetworkReconciler) updateStatus(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network) (updated bool, err error) {
	info, err := r.buildContextInfo(provider, network)
	if err != nil {
		return updated, err
	}

	// currently we creating a network, we will change update status to normal when status changed and bucket is ready
	if network.Status.Phase != cloudv1.NetworkPhaseNormal {
		var bucketReady bool

		region := network.Spec.Region
		bucketName := getBucketName(r.env.ProjectName(), region, info.projectId)
		bs, err := GetBucketStatus(ctx, r.Client, provider.Spec.Type, network.GetNamespace(), bucketName)
		if err != nil {
			return updated, err
		}

		logger := log.FromContext(ctx)

		c := bs.GetCondition(xpv1.TypeReady)
		logger.Info("bucket ready status", "bucket", bucketName,
			"status", c.Status)

		// https://github.com/crossplane-contrib/provider-aws/issues/1302
		// Some times although 'Synced' is true, but it take long time to get 'Ready' as true.
		// I think when 'Synced' is true we can make sure bucket is created, so  we check 'Synced' to workaround the issue.
		syncedStatus := bs.GetCondition(xpv1.TypeSynced)
		logger.Info("bucket synced status", "bucket", bucketName,
			"status", syncedStatus.Status)

		if c.Status == corev1.ConditionTrue || syncedStatus.Status == corev1.ConditionTrue {
			bucketReady = true
		} else {
			return false, WaitBucketReady
		}
		if bucketReady {
			r.progress(ctx, network, "All")

			network.Status.Phase = cloudv1.NetworkPhaseNormal
			network.Status.Peering.Allowed = true
			updated = true

		}
	}

	return updated, nil
}

func (r *NetworkReconciler) ensureGracefulShutdownWebhook(ctx context.Context, cli client.Client, provider *cloudv1.Provider) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureGracefulShutdownWebhook")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener spec")
	}

	if !spec.GetEnableGracefulShutdownWebhook() {
		log.FromContext(ctx).Info("skip deploy graceful web hook")
		return controllerutil.OperationResultNone, nil
	}

	itest := inIntegrationTest()

	return deployGracefulShutdownWebhook(spanCtx, consts.LocalOperatorNamespace, cli, !itest)
}

func (r *NetworkReconciler) ensureCertManager(ctx context.Context, cli client.Client, kubeconfig []byte, provider *cloudv1.Provider, info *contextInfo) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureCertManager")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "Gardener Provider")
	}

	return deployCertManager(
		spanCtx,
		provider.Spec.Type,
		cli,
		kubeconfig,
		r.cfg.Server.ManifestDirectoryPrefix,
		info.pcaRoleARN,
		spec.AWS.PcaArn,
		spec.AWS.PcaRegion,
		r.cfg.IdentityImage,
	)
}

func (r *NetworkReconciler) ensureGCPDnsZone(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, vpcName string) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureGCPDnsZone")
	defer func() { tracerutil.EndSpan(span, errResult) }()
	gardenNetStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
	if err != nil {
		return err
	}
	if gardenNetStatus.GCP.HostedZoneID != nil {
		// already created
		return nil
	}
	gardenProviderSpec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return err
	}
	region := network.Spec.Region
	dnsName := fmt.Sprintf("%s.%s.", region, gardenProviderSpec.GCP.PublicHostedZoneDomain)

	zoneId, err := r.gcpapi.CreateDnsZone(gardenProviderSpec.GCP.ProjectID, vpcName, vpcName, dnsName)
	if err != nil {
		return err
	}

	op, err := r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.GCP.HostedZoneID = pointer.String(zoneId)
	})
	if err != nil {
		return err
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureGCPDnsZone", "Ensure gcp dns zone successful")
	}
	return nil
}

func (r *NetworkReconciler) ensureBucket(ctx context.Context, provider *cloudv1.Provider, network *cloudv1.Network, info *contextInfo) (_ controllerutil.OperationResult, errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureBucket")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	region := network.Spec.Region
	bucketName := getBucketName(r.env.ProjectName(), region, info.projectId)
	labels := cloudv1.TagsFromLabels(network.Labels)

	if _, err := r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.BucketName = bucketName
	}); err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener status")
	}

	gp, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return controllerutil.OperationResultNone, errors.Annotate(err, "gardener spec")
	}

	return ensureBucket(
		spanCtx,
		r.Client,
		r.Scheme(),
		provider.Spec.Type,
		gp,
		network,
		network.GetNamespace(),
		bucketName,
		region,
		labels,
	)
}

func (r *NetworkReconciler) ensureRoles(
	ctx context.Context,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
	shootName string,
	info *contextInfo,
) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureRoles")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return errors.Annotate(err, "gardener spec")
	}
	gardenProject := r.env.ProjectName()
	iamClient := r.awsClient.IAM()

	tags := cloudv1.TagsFromLabels(network.Labels)
	switch provider.Spec.Type {
	case consts.ProviderTypeAWS:
		accountID := spec.AWS.AccountID
		queueName := getQueueName(spec.Reporter.SQSEndpoint)
		sliQueueName := getQueueName(spec.SLIReporter.SQSEndpoint)

		err := putPoliciesToNodeRole(spanCtx, iamClient, shootName, gardenProject, accountID,
			NewSqsPolicy("billing-sqs-policy", accountID, queueName),  // allow the shoot node to send messages to the billing SQS.
			NewSqsPolicy("dbaas-sqs-policy", accountID, sliQueueName), // allow the shoot node to send messages to the SLI SQS.
			NewS3Policy(gardenProject, info.projectId, getBucketName(gardenProject, network.Spec.Region, info.projectId)),
			NewPCAPolicy(),
			NewCertManagerPolicy(),
			NewEC2VolumeModifyPolicy(),
		)
		if err != nil {
			return errors.Annotate(err, "failed put policies to node role")
		}

		nodeARN, err := getShootWorkerIamIdentity(provider.Spec.Type, r.env.ProjectName(), shootName, spec.AWS.AccountID)
		if err != nil {
			return errors.Annotate(err, "worker iam")
		}

		role := getImportRoleName(gardenProject, shootName, info.projectId)

		err = EnsureImportRole(
			spanCtx,
			iamClient,
			accountID,
			role,
			"",
			nodeARN,
			r.cfg.Import.CentralRoleARN,
			r.cfg.Import.SkipBanMyBuckets,
			tags,
		)
		if err != nil {
			return errors.Annotate(err, "import role")
		}

		roleARN, err := GetImportIamIdentity(provider.Spec.Type, r.env.ProjectName(), shootName, info.projectId, spec.AWS.AccountID)
		if err != nil {
			return errors.Annotate(err, "import iam")
		}

		if network.Status.Import.RoleARN != roleARN {
			network.Status.Import.RoleARN = roleARN
			err := r.Status().Update(ctx, network)
			if err != nil {
				return errors.Annotate(err, "roleARN")
			}
			r.record.Event(network, corev1.EventTypeNormal, "EnsureImportRoles", "Ensure import roles successful")
		}

		externalId := getExternalId(roleARN)
		if network.Status.IAM.ExternalId != externalId {
			network.Status.IAM.ExternalId = externalId
			err := r.Status().Update(ctx, network)
			if err != nil {
				return errors.Annotate(err, "externalId")
			}
		}

		op, err := r.updateGardenerStatus(ctx, network, func(g *cloudv1.GardenerNetworkStatus) {
			if g.NodeRole != nodeARN {
				g.NodeRole = nodeARN
			}
		})
		if err != nil {
			return errors.Annotate(err, "gardener status update")
		} else if op != controllerutil.OperationResultNone {
			r.record.Event(network, corev1.EventTypeNormal, "EnsureNodeRoles", "Ensure node roles successful")
		}
	case consts.ProviderTypeGCP:
		err := r.ensureGCPShootWorkerPolicies(spanCtx, iamClient, shootName, provider, network)
		if err != nil {
			return errors.Annotate(err, "shoot worker policies")
		}

		// ref https://github.com/tidbcloud/dbaas/blob/master/central/runtime/deployer/gardener/shoot_cluster.go#L577
		//
		// Add storage admin permission to node service account, so BR and TiKV can access the bucket using ADC
		// However this two components don't support Application Default Credentials yet
		// See https://github.com/pingcap/br/issues/633
		// So currently we create a key for the node service account
		// and copy it to every tidb cluster namespace for BR during cluster creation
		// BR should only have objectAdmin permission, but it needs bucket list permission which is improper
		if provider.Spec.Type == consts.ProviderTypeGCP {
			gcpNodeSa, err := getShootWorkerIamIdentity(
				consts.ProviderTypeGCP,
				r.env.ProjectName(),
				shootName,
				spec.GCP.ProjectID,
			)
			if err != nil {
				return err
			}
			var roleName gcpiam.RoleName = "roles/storage.admin"
			member := fmt.Sprintf("serviceAccount:%s", gcpNodeSa)
			bucketName := getBucketName(r.env.ProjectName(), network.Spec.Region, info.projectId)
			err = r.gcpapi.AddBucketIAMMember(bucketName, roleName, member)
			if err != nil {
				return errors.Annotatef(err, "failed to configure policy for the bucket %q", bucketName)
			}
		}
	default:
		return errors.Errorf("unknown provider type: %s", provider.Spec.Type)
	}

	return nil
}

func (r *NetworkReconciler) cleanUpRoles(
	ctx context.Context,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
	shootName string,
	info *contextInfo,
) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "cleanUpRoles")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	switch provider.Spec.Type {
	case consts.ProviderTypeAWS:
		return r.cleanUpAWSRoles(spanCtx, provider, network, shootName, info)
	case consts.ProviderTypeGCP:
		return r.cleanUpGCPRoles(spanCtx, shootName, info.projectId)
	default:
		return errors.Errorf("unknown provider type: %s", provider.Spec.Type)
	}
}

func (r *NetworkReconciler) cleanUpAWSRoles(
	ctx context.Context,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
	shootName string,
	info *contextInfo,
) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "cleanUpAWSRoles")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	iamClient := r.awsClient.IAM()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return errors.Annotate(err, "GardenerProvider")
	}
	project := r.env.ProjectName()

	role := getImportRoleName(project, shootName, info.projectId)
	err = util.DeleteRoleIgnoreNotFound(spanCtx, iamClient, role)
	if err != nil {
		return errors.Annotate(err, "DeleteRole")
	}

	// It should be no need to do this, since we never attach any policy to node role.
	// the node role should be delete by gardener.
	// keep this like central anyway...
	err = detachPolicy(spanCtx, iamClient, shootName, r.env.ProjectName(), spec.AWS.AccountID)
	if err != nil && !util.IsIAMNoSuchEntityException(err) {
		return errors.Annotate(err, "detachPolicy")
	}

	return nil
}

func shootName(network *cloudv1.Network, providerType string, projectId string) (string, error) {
	status, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
	if err != nil {
		return "", errors.Annotate(err, "gardener status from network")
	}

	if status != nil && status.ShootName != "" {
		return status.ShootName, nil
	}

	// use the same format as dbaas to easy migrate
	// this assume user will not use the same (tenantID, providerType, region) to create
	// multi network at infra-api
	// after migration, we can change it to generate the name by (ns, name) of network.
	rawName := fmt.Sprintf("%s-%s-%s", projectId, providerType, network.Spec.Region)
	name := util.FnvHash(rawName)
	return name, nil
}

func inIntegrationTest() bool {
	v := os.Getenv("KUBEBUILDER_ASSETS")
	return v != ""
}

// getQueueName parses the queue name from the SQS endpoint.
func getQueueName(sqsEndpoint string) string {
	parts := strings.Split(strings.TrimSpace(sqsEndpoint), "/")
	return parts[len(parts)-1]
}

func (r *NetworkReconciler) cleanUpGCPRoles(ctx context.Context, shootName string, project string) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "cleanUpGCPRoles")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	gardenProject := r.env.ProjectName()
	iamClient := r.awsClient.IAM()

	var roles []string

	roleName := getGCPReporterAWSRoleName(gardenProject, project, shootName)
	roles = append(roles, roleName)

	roleName = getGCPPCARoleName(gardenProject, project, shootName)
	roles = append(roles, roleName)

	for _, role := range roles {
		err := util.DeleteRoleIgnoreNotFound(spanCtx, iamClient, role)
		if err != nil {
			return errors.Annotatef(err, "failed to delete GCP role: %s", role)
		}
	}

	user := getGCPTiKVAWSUserName(gardenProject, project, shootName)

	err := util.DeleteUserAccessKeys(spanCtx, iamClient, user)
	if err != nil {
		// we may never create the user, or already been deleted
		// just ignore it and still try to delete the user
		if util.IsIAMNoSuchEntityException(err) {
			log.FromContext(ctx).Info("user already been deleted", "user", user)
		} else {
			return errors.Annotatef(err, "failed to delete GCP user access key: %s", user)
		}
	}

	err = util.DeleteUserIgnoreNotFound(spanCtx, iamClient, user)
	if err != nil {
		return errors.Annotatef(err, "failed to delete GCP user: %s", user)
	}

	return nil
}

// Note: any roles created in this function must be deleted in `cleanUpGCPRoles()`.
func (r *NetworkReconciler) ensureGCPShootWorkerPolicies(
	ctx context.Context,
	iamClient awsapi.IAMClient,
	shootName string,
	provider *cloudv1.Provider,
	network *cloudv1.Network,
) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "ensureGCPShootWorkerPolicies")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	gardenProject := r.env.ProjectName()
	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return err
	}

	gcpNodeSAName, err := getShootWorkerIamIdentity(consts.ProviderTypeGCP, gardenProject, shootName, spec.GCP.ProjectID)
	if err != nil {
		return errors.Annotatef(err, "get shoot worker service account")
	}
	// Note: So the "gcp" provider also need the AccountID
	accountID := spec.AWS.AccountID

	// Note: This get default credentials, so DO NOT support multi gcp type provider with different account.
	gcpNodeSA, err := r.gcpapi.QueryServiceAccount(gcpNodeSAName, spec.GCP.ProjectID)
	if err != nil {
		return errors.Annotatef(err, "get shoot worker service account")
	}
	arp, err := awsapi.NewGoogleIdentityPolicy(gcpNodeSA.UniqueId).ToString()
	if err != nil {
		return errors.Annotatef(err, "constructing reporter assume role policy")
	}

	var awsTags []iamtypes.Tag
	for k, v := range cloudv1.TagsFromLabels(network.Labels) {
		awsTags = append(awsTags, iamtypes.Tag{
			Key:   aws.String(k),
			Value: aws.String(v),
		})
	}
	awsTags = append(awsTags, iamtypes.Tag{
		Key:   aws.String("Provider"),
		Value: aws.String("gcp"),
	})

	// ensure AWS IAM role for reporters
	role := &iam.CreateRoleInput{
		RoleName:                 aws.String(getGCPReporterAWSRoleName(gardenProject, network.GetProjectId(), shootName)),
		AssumeRolePolicyDocument: aws.String(arp),
		Path:                     aws.String("/"),
		Tags:                     awsTags,
	}
	// the billing reporter
	billingQueueName := getQueueName(spec.Reporter.SQSEndpoint)
	billingPolicyDoc, err := NewSqsPolicy("billing-sqs-policy", accountID, billingQueueName).PolicyDocument()
	if err != nil {
		return errors.Annotatef(err, "constructing the billing reporter role policy")
	}
	billingRolePolicy := &iam.PutRolePolicyInput{
		PolicyDocument: aws.String(billingPolicyDoc),
		PolicyName:     aws.String("gcp-reporter-policy"),
		RoleName:       role.RoleName,
	}
	// the SLI reporter
	sliQueueName := getQueueName(spec.SLIReporter.SQSEndpoint)
	sliPolicyDoc, err := NewSqsPolicy("dbaas-sqs-policy", accountID, sliQueueName).PolicyDocument()
	if err != nil {
		return errors.Annotatef(err, "constructing the SLI reporter role policy")
	}
	sliRolePolicy := &iam.PutRolePolicyInput{
		PolicyDocument: aws.String(sliPolicyDoc),
		PolicyName:     aws.String("gcp-sli-reporter-policy"),
		RoleName:       role.RoleName,
	}
	if err := util.EnsureRole(spanCtx, iamClient, role, billingRolePolicy, sliRolePolicy); err != nil {
		return errors.Annotatef(err, "Ensure iam role for reporter")
	}

	// ensure AWS IAM role for PCA
	pca := &iam.CreateRoleInput{
		RoleName:                 aws.String(getGCPPCARoleName(gardenProject, network.GetProjectId(), shootName)),
		AssumeRolePolicyDocument: aws.String(arp),
		Path:                     aws.String("/"),
		Tags:                     awsTags,
	}
	pcaPolicy, err := NewPCAPolicy().PolicyDocument()
	if err != nil {
		return errors.Annotatef(err, "constructing pca role policy")
	}
	pcaRolePolicy := &iam.PutRolePolicyInput{
		PolicyDocument: aws.String(pcaPolicy),
		PolicyName:     aws.String("gcp-pca-policy"),
		RoleName:       pca.RoleName,
	}
	if err := util.EnsureRole(spanCtx, iamClient, pca, pcaRolePolicy); err != nil {
		return errors.Annotatef(err, "Ensure iam role for reporter")
	}

	// ensure AWS user for GCP TiKV
	kmsUser := &iam.CreateUserInput{
		UserName: aws.String(getGCPTiKVAWSUserName(gardenProject, network.GetProjectId(), shootName)),
		Path:     aws.String("/"),
		Tags:     awsTags,
	}
	// Grant kms key access separately when create TiDB cluster in favor of fine-grained access control
	if err := util.EnsureUser(spanCtx, iamClient, kmsUser); err != nil {
		return errors.Annotatef(err, "Ensure iam role for gcp kms reader")
	}

	if err := attachPolicyToNodeSa(r.gcpapi, gardenProject, shootName, network.Spec.Region, network.GetProjectId(), spec.GCP.ProjectID, spec.GCP.CertManagerRole); err != nil {
		return errors.Annotate(err, "Attach policy to node service account")
	}

	op, err := r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.NodeRole = gcpNodeSAName
	})
	if err != nil {
		return err
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureGCPNodeRoles", "Ensure gcp node roles successful")
	}

	return nil
}

func (r *NetworkReconciler) prepareGCPCredentials(
	ctx context.Context,
	shootCli client.Client,
	shootName string,
	network *cloudv1.Network,
	provider *cloudv1.Provider,
	projectId string,
) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "prepareGCPCredentials")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	spec, err := cloudv1.GardenerProviderSpecForProvider(provider)
	if err != nil {
		return err
	}

	nodeSA, err := getShootWorkerIamIdentity(consts.ProviderTypeGCP, r.env.ProjectName(), shootName, spec.GCP.ProjectID)
	if err != nil {
		return err
	}

	err = r.prepareSecretForBR(spanCtx, shootCli, spec.GCP.ProjectID, nodeSA)
	if err != nil {
		return err
	}

	tikvUserName := getGCPTiKVAWSUserName(r.env.ProjectName(), projectId, shootName)
	err = prepareSecretForKMS(spanCtx, shootCli, r.awsClient.IAM(), tikvUserName)
	if err != nil {
		return err
	}

	op, err := r.updateGardenerStatus(spanCtx, network, func(g *cloudv1.GardenerNetworkStatus) {
		g.AdminNamespace = consts.LocalOperatorNamespace
		g.GCP.BrSecretName = pointer.String(consts.BRGCPSecretName)
		g.GCP.KmsSecretName = pointer.String(gcpKMSSecretName)
		g.GCP.TiKVAWSUserArn = pointer.String(fmt.Sprintf("arn:aws:iam::%s:user/%s", spec.AWS.AccountID, tikvUserName))
	})
	if err != nil {
		return err
	} else if op != controllerutil.OperationResultNone {
		r.record.Event(network, corev1.EventTypeNormal, "EnsureKMS", "Ensure kms successful")
	}

	return nil
}

func (r *NetworkReconciler) prepareSecretForBR(ctx context.Context, kubecli client.Client, projectID string, nodeSa string) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "prepareSecretForBR")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	secretName := consts.BRGCPSecretName
	secret := &corev1.Secret{}
	namespacedName := types.NamespacedName{
		Namespace: consts.LocalOperatorNamespace,
		Name:      secretName,
	}
	err := kubecli.Get(spanCtx, namespacedName, secret)
	if err != nil {
		if !apierrors.IsNotFound(errors.Cause(err)) {
			return errors.Annotate(err, "get service account key secret from tidb operator namespace")
		}
		key, err := r.gcpapi.CreateServiceAccountKey(projectID, nodeSa)
		if err != nil {
			return errors.Annotate(err, "create service account key for node sa")
		}
		secret = &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      secretName,
				Namespace: consts.LocalOperatorNamespace,
			},
			Data: map[string][]byte{
				"credentials": key,
			},
			Type: corev1.SecretTypeOpaque,
		}
		if err := kubecli.Create(spanCtx, secret); err != nil {
			return errors.Annotate(err, "create service account key secret in tidb operator namespace")
		}
	}
	return nil
}

func prepareSecretForKMS(ctx context.Context, shootCli client.Client, iamClient awsapi.IAMClient, kmsUser string) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "prepareSecretForKMS")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	secretName := gcpKMSSecretName
	secret := &corev1.Secret{}
	namespacedName := types.NamespacedName{
		Namespace: consts.LocalOperatorNamespace,
		Name:      secretName,
	}
	err := shootCli.Get(spanCtx, namespacedName, secret)
	if err != nil {
		if !apierrors.IsNotFound(errors.Cause(err)) {
			return errors.Annotate(err, "Get")
		}
		if err = cleanUpAccessKeyIfExceedQuota(spanCtx, iamClient, kmsUser); err != nil {
			return err
		}

		key, err := ensureAccessKeyForUser(spanCtx, iamClient, kmsUser)
		if err != nil {
			return errors.Annotatef(err, "create access key for user %s", kmsUser)
		}
		secret = &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      secretName,
				Namespace: consts.LocalOperatorNamespace,
			},
			Data: map[string][]byte{
				consts.AWS_ACCESS_KEY_ENV_VAR: []byte(*key.AccessKeyId),
				consts.AWS_SECRET_KEY_ENV_VAR: []byte(*key.SecretAccessKey),
			},
			Type: corev1.SecretTypeOpaque,
		}
		if err := shootCli.Create(spanCtx, secret); err != nil {
			return errors.Annotate(err, "create KMS access key secret in tidb-admin namespace")
		}
	}
	return nil
}

func cleanUpAccessKeyIfExceedQuota(ctx context.Context, iamClient awsapi.IAMClient, username string) (errResult error) {
	spanCtx, span := tracer.Start(ctx, "cleanUpAccessKeyIfExceedQuota")
	defer func() { tracerutil.EndSpan(span, errResult) }()

	output, err := iamClient.ListAccessKeys(spanCtx, &iam.ListAccessKeysInput{
		UserName: aws.String(username)})
	if err != nil {
		return err
	}
	// AWS cannot exceed quota for AccessKeysPerUser: 2
	if len(output.AccessKeyMetadata) >= 2 {
		for _, key := range output.AccessKeyMetadata {
			_, err = iamClient.DeleteAccessKey(spanCtx, &iam.DeleteAccessKeyInput{AccessKeyId: key.AccessKeyId, UserName: aws.String(username)})
			if err != nil {
				return err
			}
		}
	}
	return nil
}

func ensureAccessKeyForUser(ctx context.Context, iamClient awsapi.IAMClient, username string) (*iamtypes.AccessKey, error) {
	key, err := iamClient.CreateAccessKey(ctx, &iam.CreateAccessKeyInput{UserName: aws.String(username)})
	if err != nil && !util.IsIAMEntityAlreadyExists(err) {
		return nil, errors.Annotate(err, "create access key")
	}

	return key.AccessKey, nil
}

// getNetworkClusters get all clusters belong to the specific network.
func getNetworkClusters(ctx context.Context, c client.Client, ns string, name string) ([]*cloudv1.Cluster, error) {
	list := &cloudv1.ClusterList{}
	err := c.List(ctx, list, client.InNamespace(ns))
	if err != nil {
		return nil, errors.Annotate(err, "list clusters")
	}

	var clusters []*cloudv1.Cluster
	for _, c := range list.Items {
		if c.Spec.NetworkName != name {
			continue
		}

		c := c
		clusters = append(clusters, &c)
	}

	return clusters, nil
}

func getNetworkBackupNames(ctx context.Context, c client.Client, network *cloudv1.Network) ([]string, error) {
	networkStatus, err := cloudv1.GardenerNetworkStatusFromNetwork(network)
	if err != nil {
		return nil, err
	}
	networkShootName := networkStatus.ShootName

	var backupList cloudv1.BackupList
	err = c.List(ctx, &backupList, client.InNamespace(network.Namespace))
	if err != nil {
		return nil, err
	}

	var backupNames []string
	for _, b := range backupList.Items {
		backupStatus, err := cloudv1.ShootBackupStatusFromBackup(&b)
		if err != nil {
			return nil, err
		}
		if backupStatus.ShootName == networkShootName {
			backupNames = append(backupNames, b.Name)
		}
	}
	return backupNames, nil
}

func getNetworkPeerNames(ctx context.Context, c client.Client, network *cloudv1.Network) ([]string, error) {
	nps := new(cloudv1.NetworkPeerList)
	err := c.List(context.TODO(), nps, client.InNamespace(network.Namespace))
	if err != nil {
		return nil, err
	}

	var names []string
	for _, np := range nps.Items {
		if np.Spec.Source == network.Name {
			names = append(names, np.Name)
		}
	}
	return names, nil
}