package privatelinkservice import ( "context" "fmt" "sync" "github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/network/armnetwork/v6" ec2api "github.com/aws/aws-sdk-go-v2/service/ec2" ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1" route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1" "github.com/crossplane/crossplane-runtime/pkg/meta" "github.com/pingcap/log" regionalapiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/regional-server/v1" "github.com/tidbcloud/infra-provider/pkg/cloud/aws" "github.com/tidbcloud/infra-provider/pkg/consts" "github.com/tidbcloud/infra-provider/pkg/util" "github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/azure" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli" bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/privatelinkgroup" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils" "github.com/tidbcloud/infra-provider/regional-server/pkg/config" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers" "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/approute" "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/privatelinkservice" "github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/vpc" "go.uber.org/zap" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" ) type PrivateLinkServiceService struct { cli client.Client privatelinkservicerw privatelinkservice.PrivateLinkServiceReaderWriter approuter approute.APPRouteReader vpcr vpc.VPCReader config config.RootConfig plsClients sync.Map awsCli aws.Client groupManager *privatelinkgroup.Manager groupClusterClient *clustercli.ClusterCli } func NewPrivateLinkServiceService( cli client.Client, privatelinkservicerw privatelinkservice.PrivateLinkServiceReaderWriter, approuter approute.APPRouteReader, vpcr vpc.VPCReader, config config.RootConfig, ) *PrivateLinkServiceService { return &PrivateLinkServiceService{ cli: cli, privatelinkservicerw: privatelinkservicerw, approuter: approuter, vpcr: vpcr, config: config, plsClients: sync.Map{}, groupManager: privatelinkgroup.NewManager(cli, nil), groupClusterClient: clustercli.NewClusterCli(config.ClusterControlServiceEndpoint), } } func (s *PrivateLinkServiceService) WithAWSClient(cli aws.Client) *PrivateLinkServiceService { s.awsCli = cli s.groupManager = privatelinkgroup.NewManager(s.cli, cli) return s } func (s *PrivateLinkServiceService) CreatePrivateLinkService(ctx context.Context, req *regionalapiv1.CreatePrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) { // check approute is already existed and ready. approutes := &v1alpha1.APPRouteList{} if err := s.approuter.List(ctx, approutes, client.MatchingLabels{ v1alpha1.ClusterIDKey: req.PrivateLinkService.ClusterId, }); err != nil { log.Error("failed to list approutes", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to list cluster") } if len(approutes.Items) != 1 { log.Error("failed to find approute", zap.Any("approutes", approutes.Items)) return nil, bizerr.New(bizerr.K8SClientError, "failed to find cluster") } approute := approutes.Items[0] pls := &v1alpha1.PrivateLinkService{ ObjectMeta: metav1.ObjectMeta{ Name: req.PrivateLinkService.Id, Namespace: approute.Namespace, Labels: map[string]string{ v1alpha1.VPCIDKey: approute.GetLabels()[v1alpha1.VPCIDKey], v1alpha1.VPCNameKey: approute.GetLabels()[v1alpha1.VPCNameKey], v1alpha1.ClusterIDKey: req.PrivateLinkService.ClusterId, v1alpha1.LoadBalancerIDKey: req.PrivateLinkService.TidbGroupId, // tags for billing v1alpha1.TagTenantKey: approute.GetLabels()[v1alpha1.TagTenantKey], v1alpha1.TagProjectKey: approute.GetLabels()[v1alpha1.TagProjectKey], v1alpha1.TagClusterKey: approute.GetLabels()[v1alpha1.TagClusterKey], }, }, Spec: v1alpha1.PrivateLinkServiceSpec{ Provider: s.config.Provider, Region: s.config.RegionConfig.Name, VPCRef: v1alpha1.Reference{ Name: approute.GetLabels()[v1alpha1.VPCNameKey], Namespace: approute.Namespace, }, LoadBalancers: []v1alpha1.LoadBalancer{ {ID: req.PrivateLinkService.TidbGroupId, ClusterID: req.PrivateLinkService.ClusterId}, }, }, } for _, c := range req.PrivateLinkService.Connections { conn := v1alpha1.Connection{} if c.GetAws() != nil { conn.Endpoint = c.GetAws().VpcEndpointId } if c.GetGcp() != nil { conn.Endpoint = c.GetGcp().PrivateServiceConnectEndpoint } if c.GetAzure() != nil { conn = v1alpha1.Connection{ Endpoint: c.GetAzure().PrivateEndpointResourceId, IPs: []string{c.GetAzure().PrivateIpAddress}, } } pls.Spec.Connections = append(pls.Spec.Connections, conn) } pls.ObjectMeta.Labels = util.MergeMaps(pls.ObjectMeta.Labels, s.config.ResourceLabels) if err := s.privatelinkservicerw.Create(ctx, pls); err != nil { if apierrors.IsAlreadyExists(err) { log.Info("privatelinkservice already exists", zap.String("id", req.PrivateLinkService.Id)) return req.PrivateLinkService, bizerr.New(bizerr.ResourceConflict, "privatelinkservice already exists") } log.Error("failed to create privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to create privatelinkservice") } return req.PrivateLinkService, nil } func (s *PrivateLinkServiceService) GetPrivateLinkService(ctx context.Context, req *regionalapiv1.GetPrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) { namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env) pls := &v1alpha1.PrivateLinkService{} if err := s.privatelinkservicerw.Get(ctx, client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}, pls); err != nil { if client.IgnoreNotFound(err) != nil { log.Error("failed to get privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice") } log.Error("privatelinkservice not found", zap.String("id", req.PrivateLinkServiceId)) return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId { log.Error("privatelinkservice cluster id not match", zap.String("id", req.PrivateLinkServiceId), zap.String("clusterId", req.ClusterId)) return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if s.config.Provider == consts.ProviderTypeAKS { if err := s.getPrivateLinkConnectionsForAzure(ctx, pls); err != nil { log.Error("failed to get privatelinkservice connections from azure", zap.Error(err)) } } plsPB, err := genPrivateLinkServicePB(pls) if err != nil { log.Error("failed to fill privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkservice") } return plsPB, nil } func (s *PrivateLinkServiceService) getPrivateLinkConnectionsForAzure(ctx context.Context, pls *v1alpha1.PrivateLinkService) error { conns, err := s.getPrivateLinkServiceConnectionsFromAzure(ctx, pls) if err != nil { return err } existedConnsMap := make(map[string]v1alpha1.ConnectionStatus) for _, conn := range pls.Status.Connections { existedConnsMap[conn.Endpoint] = conn } for _, conn := range conns { state := ptr.Deref(conn.Properties.PrivateLinkServiceConnectionState.Status, "") if state != azure.ConnectionStatePending { continue } endpintID := ptr.Deref(conn.Properties.PrivateEndpoint.ID, "") if endpintID == "" { continue } if _, ok := existedConnsMap[endpintID]; ok { continue } pls.Status.Connections = append(pls.Status.Connections, v1alpha1.ConnectionStatus{ Endpoint: endpintID, Phase: v1alpha1.ConnectionPhasePending, }) } return nil } func (s *PrivateLinkServiceService) ListPrivateLinkService(ctx context.Context, req *regionalapiv1.ListPrivateLinkServicesRequest) (*regionalapiv1.ListPrivateLinkServicesResponse, error) { plsList := &v1alpha1.PrivateLinkServiceList{} labelSelector := client.MatchingLabels{ v1alpha1.ClusterIDKey: req.ClusterId, } if req.TidbGroupId != "" { labelSelector[v1alpha1.LoadBalancerIDKey] = req.TidbGroupId } if err := s.privatelinkservicerw.List(ctx, plsList, labelSelector); err != nil { log.Error("failed to list privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to list privatelinkservice") } plsPBList := make([]*regionalapiv1.PrivateLinkService, 0, len(plsList.Items)) for _, pls := range plsList.Items { if s.config.Provider == consts.ProviderTypeAKS { if err := s.getPrivateLinkConnectionsForAzure(ctx, &pls); err != nil { log.Error("failed to get privatelinkservice connections from azure", zap.Error(err)) } } plsPB, err := genPrivateLinkServicePB(&pls) if err != nil { log.Error("failed to fill privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkservice") } plsPBList = append(plsPBList, plsPB) } return ®ionalapiv1.ListPrivateLinkServicesResponse{ PrivateLinkServices: plsPBList, }, nil } func (s *PrivateLinkServiceService) DeletePrivateLinkService(ctx context.Context, req *regionalapiv1.DeletePrivateLinkServiceRequest) error { namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env) pls := &v1alpha1.PrivateLinkService{ ObjectMeta: metav1.ObjectMeta{ Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace(), }, } if err := s.privatelinkservicerw.Get(ctx, client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}, pls); err != nil { if client.IgnoreNotFound(err) != nil { log.Error("failed to get privatelinkservice", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice") } return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId { log.Error("privatelinkservice cluster id not match", zap.String("id", req.PrivateLinkServiceId), zap.String("clusterId", req.ClusterId)) return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if err := s.privatelinkservicerw.Delete(ctx, pls); err != nil { if apierrors.IsNotFound(err) { log.Info("privatelinkservice not found", zap.String("id", req.PrivateLinkServiceId)) return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } log.Error("failed to delete privatelinkservice", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to delete privatelinkservice") } return nil } func (s *PrivateLinkServiceService) RejectVpcEndpointConnections(ctx context.Context, req *regionalapiv1.RejectVpcEndpointConnectionsRequest) error { if s.config.Provider != consts.ProviderTypeEKS { return bizerr.New(bizerr.ParameterInvalid, "RejectVpcEndpointConnections is only supported for AWS") } if s.awsCli == nil { return bizerr.New(bizerr.InternalServerError, "AWS client is not initialized") } namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env) pls := &v1alpha1.PrivateLinkService{} if err := s.privatelinkservicerw.Get(ctx, client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}, pls); err != nil { if client.IgnoreNotFound(err) != nil { log.Error("failed to get privatelinkservice", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice") } return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId { log.Error("privatelinkservice cluster id not match", zap.String("id", req.PrivateLinkServiceId), zap.String("clusterId", req.ClusterId)) return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } svcID := pls.Status.EndpointService.ServiceID if svcID == "" { return bizerr.New(bizerr.ParameterInvalid, "VPC endpoint service is not ready") } // Get the VPC to obtain the AWS account role ARN vpcObj := &v1alpha1.VPC{} if err := s.vpcr.Get(ctx, client.ObjectKey{Namespace: pls.Spec.VPCRef.Namespace, Name: pls.Spec.VPCRef.Name}, vpcObj); err != nil { log.Error("failed to get vpc", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get vpc") } roleARN := "" if vpcObj.Spec.AccountRef != nil { roleARN = "arn:aws:iam::" + vpcObj.Spec.AccountRef.AccountID + ":role/" + vpcObj.Spec.AccountRef.Role } ec2Cli := s.awsCli.EC2(&aws.AssumedRole{RoleARN: roleARN}) res, err := ec2Cli.RejectVpcEndpointConnections(ctx, &ec2api.RejectVpcEndpointConnectionsInput{ ServiceId: &svcID, VpcEndpointIds: req.VpcEndpointIds, }, func(opts *ec2api.Options) { opts.Region = pls.Spec.Region }) if err != nil { log.Error("failed to reject vpc endpoint connections", zap.Error(err)) return bizerr.New(bizerr.AWSClientError, "failed to reject vpc endpoint connections") } if len(res.Unsuccessful) != 0 { log.Error("failed to reject some vpc endpoints", zap.Any("unsuccessful", res.Unsuccessful)) return bizerr.New(bizerr.AWSClientError, fmt.Sprintf("failed to reject some vpc endpoints: %v", res.Unsuccessful)) } return nil } func (s *PrivateLinkServiceService) PausePrivateLinkService(ctx context.Context, req *regionalapiv1.PausePrivateLinkServiceRequest) error { if s.config.Provider != consts.ProviderTypeEKS { return bizerr.New(bizerr.ParameterInvalid, "pause private link service is only supported for AWS") } namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env) pls := &v1alpha1.PrivateLinkService{} if err := s.privatelinkservicerw.Get(ctx, client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}, pls); err != nil { if client.IgnoreNotFound(err) != nil { log.Error("failed to get privatelinkservice", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice") } return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId { log.Error("privatelinkservice cluster id not match", zap.String("id", req.PrivateLinkServiceId), zap.String("clusterId", req.ClusterId)) return bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } // Pause the PrivateLinkService CR (idempotent - skip if already paused) if !utils.IsPaused(pls) { utils.Pause(pls) if err := s.privatelinkservicerw.Update(ctx, pls); err != nil { log.Error("failed to pause privatelinkservice", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to pause privatelinkservice") } } // Pause the crossplane VPCEndpointServiceConfiguration (idempotent) vpceSvcName := utils.GetEndpointServiceName(pls) ves := &ec2v1alpha1.VPCEndpointServiceConfiguration{} if err := s.cli.Get(ctx, types.NamespacedName{Name: vpceSvcName}, ves); err != nil { if apierrors.IsNotFound(err) { log.Info("crossplane VPCEndpointServiceConfiguration not found, skip pausing", zap.String("name", vpceSvcName)) return nil } log.Error("failed to get crossplane VPCEndpointServiceConfiguration", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get crossplane VPCEndpointServiceConfiguration") } if ves.GetAnnotations()[meta.AnnotationKeyReconciliationPaused] != "true" { patchBase := ves.DeepCopy() anno := ves.GetAnnotations() if anno == nil { anno = map[string]string{} } anno[meta.AnnotationKeyReconciliationPaused] = "true" ves.SetAnnotations(anno) if err := s.cli.Patch(ctx, ves, client.MergeFrom(patchBase)); err != nil { log.Error("failed to pause crossplane VPCEndpointServiceConfiguration", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to pause crossplane VPCEndpointServiceConfiguration") } } // Pause the crossplane Route53 ResourceRecordSet (idempotent) recordName := pls.Status.EndpointService.PrivateDNSName if name := pls.Annotations[v1alpha1.AnnotationKeyOwnCrossplaneNameResourceRecordSet]; name != "" { recordName = name } if recordName == "" { log.Info("PrivateDNSName is not set, skip pausing crossplane ResourceRecordSet") return nil } rrs := &route53v1alpha1.ResourceRecordSet{} if err := s.cli.Get(ctx, types.NamespacedName{Name: recordName}, rrs); err != nil { if apierrors.IsNotFound(err) { log.Info("crossplane ResourceRecordSet not found, skip pausing", zap.String("name", recordName)) return nil } log.Error("failed to get crossplane ResourceRecordSet", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to get crossplane ResourceRecordSet") } if rrs.GetAnnotations()[meta.AnnotationKeyReconciliationPaused] != "true" { patchBase := rrs.DeepCopy() anno := rrs.GetAnnotations() if anno == nil { anno = map[string]string{} } anno[meta.AnnotationKeyReconciliationPaused] = "true" rrs.SetAnnotations(anno) if err := s.cli.Patch(ctx, rrs, client.MergeFrom(patchBase)); err != nil { log.Error("failed to pause crossplane ResourceRecordSet", zap.Error(err)) return bizerr.New(bizerr.K8SClientError, "failed to pause crossplane ResourceRecordSet") } } return nil } func (s *PrivateLinkServiceService) UpdatePrivateLinkService(ctx context.Context, req *regionalapiv1.UpdatePrivateLinkServiceRequest) (*regionalapiv1.PrivateLinkService, error) { namer := utils.NewNamerWithClusterID(req.ClusterId, s.config.RegionConfig.Name, s.config.Env) pls := &v1alpha1.PrivateLinkService{} if err := s.privatelinkservicerw.Get(ctx, client.ObjectKey{Name: req.PrivateLinkServiceId, Namespace: namer.GetNamespace()}, pls); err != nil { if client.IgnoreNotFound(err) != nil { log.Error("failed to get privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to get privatelinkservice") } log.Error("privatelinkservice not found", zap.String("id", req.PrivateLinkServiceId)) return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } if pls.GetLabels()[v1alpha1.ClusterIDKey] != req.ClusterId { log.Error("privatelinkservice cluster id not match", zap.String("id", req.PrivateLinkServiceId), zap.String("clusterId", req.ClusterId)) return nil, bizerr.New(bizerr.ResourceNotFound, "privatelinkservice not found") } // validate only for AKS, ignore for other providers // if validateonly is false, we will ignore the validation and update the connections directly, // because the the update action will contains all the connections, some connections maybe have been deleted by users on cloud console, // this connection will be validated failed, this failure will block this update action. this is not what we want. if req.GetValidateOnly() && s.config.Provider == consts.ProviderTypeAKS { if err := s.validatePrivateLinkServiceConnections(ctx, req, pls); err != nil { log.Info("privatelinkservice connections validation failed", zap.Error(err)) return nil, err } return nil, nil } conns := make([]v1alpha1.Connection, 0, len(req.PrivateLinkConnections)) for _, c := range req.PrivateLinkConnections { conn := v1alpha1.Connection{} switch s.config.Provider { case consts.ProviderTypeEKS: if c.GetAws() == nil { return nil, bizerr.New(bizerr.ParameterInvalid, "aws connection required") } conn.Endpoint = c.GetAws().VpcEndpointId case consts.ProviderTypeGKE: if c.GetGcp() == nil { return nil, bizerr.New(bizerr.ParameterInvalid, "gcp connection required") } conn.Endpoint = c.GetGcp().PrivateServiceConnectEndpoint case consts.ProviderTypeAKS: if c.GetAzure() == nil { return nil, bizerr.New(bizerr.ParameterInvalid, "azure connection required") } conn = v1alpha1.Connection{ Endpoint: c.GetAzure().PrivateEndpointResourceId, IPs: []string{c.GetAzure().PrivateIpAddress}, } default: return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid") } conns = append(conns, conn) } pls.Spec.Connections = conns if err := s.privatelinkservicerw.Update(ctx, pls); err != nil { log.Error("failed to update privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.K8SClientError, "failed to update privatelinkservice") } plsPB, err := genPrivateLinkServicePB(pls) if err != nil { log.Error("failed to fill privatelinkservice", zap.Error(err)) return nil, bizerr.New(bizerr.InternalServerError, "failed to fill privatelinkservice") } plsPB.Status = regionalapiv1.PrivateLinkService_CREATING return plsPB, nil } func (s *PrivateLinkServiceService) validatePrivateLinkServiceConnections(ctx context.Context, req *regionalapiv1.UpdatePrivateLinkServiceRequest, pls *v1alpha1.PrivateLinkService) error { conns, err := s.getPrivateLinkServiceConnectionsFromAzure(ctx, pls) if err != nil { return bizerr.New(bizerr.InternalServerError, "failed to get privatelinkservice connections from azure") } connsMap := make(map[string]*armnetwork.PrivateEndpointConnection) for _, conn := range conns { connsMap[ptr.Deref(conn.Properties.PrivateEndpoint.ID, "")] = conn } for _, conn := range req.PrivateLinkConnections { if conn.GetAzure() == nil { continue } if _, ok := connsMap[conn.GetAzure().PrivateEndpointResourceId]; !ok { return bizerr.New(bizerr.ConnectionNotFound, fmt.Sprintf("connection %s not found", conn.GetAzure().PrivateEndpointResourceId)) } } return nil } func (s *PrivateLinkServiceService) getPrivateLinkServiceConnectionsFromAzure(ctx context.Context, pls *v1alpha1.PrivateLinkService) ([]*armnetwork.PrivateEndpointConnection, error) { vpc := &v1alpha1.VPC{} if err := s.vpcr.Get(ctx, client.ObjectKey{Namespace: pls.Spec.VPCRef.Namespace, Name: pls.Spec.VPCRef.Name}, vpc); err != nil { return nil, err } subscriptionID := vpc.Spec.AccountRef.AccountID cli, ok := s.plsClients.Load(subscriptionID) if !ok { cli, err := azure.NewPrivateLinkServiceClient(subscriptionID) if err != nil { return nil, err } s.plsClients.Store(subscriptionID, cli) } var plsCli azure.PrivateLinkServiceClient plsCli, ok = cli.(azure.PrivateLinkServiceClient) if !ok { return nil, fmt.Errorf("failed to convert to azure.PrivateLinkServiceClient") } conns, err := plsCli.ListPrivateLinkEndpointConnections(ctx, controllers.GetResourceGroupName(vpc), pls.Status.AzurePrivateLinkService.Name) if err != nil { return nil, err } return conns, nil } func genPrivateLinkServicePB(pls *v1alpha1.PrivateLinkService) (*regionalapiv1.PrivateLinkService, error) { plsPB := ®ionalapiv1.PrivateLinkService{ Id: pls.Name, ClusterId: pls.GetLabels()[v1alpha1.ClusterIDKey], TidbGroupId: pls.GetLabels()[v1alpha1.LoadBalancerIDKey], } switch pls.Spec.Provider { case consts.ProviderTypeEKS: plsPB.ServiceDnsName = pls.Status.EndpointService.PrivateDNSName plsPB.Host = pls.Status.EndpointService.PrivateDNSName plsPB.Info = ®ionalapiv1.PrivateLinkService_Aws_{ Aws: ®ionalapiv1.PrivateLinkService_Aws{ ServiceName: pls.Status.EndpointService.ServiceName, AzIds: pls.Status.EndpointService.AvailabilityZones, }, } case consts.ProviderTypeGKE: plsPB.ServiceDnsName = pls.Status.ServiceAttachment.ConnectDomain plsPB.Host = pls.Status.ServiceAttachment.ConnectDomain plsPB.Info = ®ionalapiv1.PrivateLinkService_Gcp_{ Gcp: ®ionalapiv1.PrivateLinkService_Gcp{ ServiceAttachment: pls.Status.ServiceAttachment.ServiceAttachment, }, } case consts.ProviderTypeAKS: plsPB.ServiceDnsName = pls.Status.AzurePrivateLinkService.ServiceDNSName plsPB.Host = pls.Status.AzurePrivateLinkService.ServiceDNSName plsPB.Info = ®ionalapiv1.PrivateLinkService_Azure_{ Azure: ®ionalapiv1.PrivateLinkService_Azure{ ServiceResourceId: pls.Status.AzurePrivateLinkService.ID, }, } default: return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid") } endpointMap := make(map[string]v1alpha1.Connection) for _, c := range pls.Spec.Connections { endpointMap[c.Endpoint] = c connInfo := ®ionalapiv1.PrivateLinkService_Connection{ Status: regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED, } connInfo, err := setConnectionProviderInfo(connInfo, pls.Spec.Provider, c, nil) if err != nil { return nil, err } for _, cs := range pls.Status.Connections { if c.Endpoint == cs.Endpoint { connInfo, err = genConnectionPB(pls.Spec.Provider, c, ptr.To(cs)) if err != nil { return nil, err } break } } plsPB.Connections = append(plsPB.Connections, connInfo) } // only AKS need to add pending connections which are not in allowed list if pls.Spec.Provider == consts.ProviderTypeAKS { for _, connStatus := range pls.Status.Connections { if connStatus.Phase != v1alpha1.ConnectionPhasePending { continue } // skip if connection is already in allowed list, because it's already added on above if _, ok := endpointMap[connStatus.Endpoint]; ok { continue } connInfo, err := genConnectionPB(pls.Spec.Provider, v1alpha1.Connection{Endpoint: connStatus.Endpoint}, ptr.To(connStatus)) if err != nil { return nil, err } plsPB.Connections = append(plsPB.Connections, connInfo) } } plsPB.Status = coverToPrivateLinkServicePBStatus(pls.Status.Phase) return plsPB, nil } func coverToPrivateLinkServicePBStatus(phase v1alpha1.PrivateLinkServicePhase) regionalapiv1.PrivateLinkService_Status { status := regionalapiv1.PrivateLinkService_STATE_UNSPECIFIED switch phase { case "", v1alpha1.PrivateLinkServicePhasePending, v1alpha1.PrivateLinkServicePhaseReadyForCreate, v1alpha1.PrivateLinkServicePhaseCreating, v1alpha1.PrivateLinkServicePhaseFailed: status = regionalapiv1.PrivateLinkService_CREATING case v1alpha1.PrivateLinkServicePhaseActive: status = regionalapiv1.PrivateLinkService_ACTIVE case v1alpha1.PrivateLinkServicePhaseDeleting, v1alpha1.PrivateLinkServicePhaseDeleted, v1alpha1.PrivateLinkServicePhaseReadyForDelete: status = regionalapiv1.PrivateLinkService_DELETING } return status } func genConnectionPB(provider string, conn v1alpha1.Connection, connStatus *v1alpha1.ConnectionStatus) (*regionalapiv1.PrivateLinkService_Connection, error) { info := ®ionalapiv1.PrivateLinkService_Connection{ Status: regionalapiv1.PrivateLinkService_Connection_STATE_UNSPECIFIED, Message: connStatus.Message, } if provider == consts.ProviderTypeEKS { addressType := endpointIPAddressTypeToPB(connStatus.IPAddressType) if addressType != regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_UNSPECIFIED { info.IpAddressType = ptr.To(addressType) } } switch connStatus.Phase { case v1alpha1.ConnectionPhasePending: info.Status = regionalapiv1.PrivateLinkService_Connection_PENDING case v1alpha1.ConnectionPhaseActive: info.Status = regionalapiv1.PrivateLinkService_Connection_ACTIVE case v1alpha1.ConnectionPhaseFailed: info.Status = regionalapiv1.PrivateLinkService_Connection_FAILED case v1alpha1.ConnectionPhaseDeleting: info.Status = regionalapiv1.PrivateLinkService_Connection_DELETING } info, err := setConnectionProviderInfo(info, provider, conn, connStatus) if err != nil { return nil, err } return info, nil } func endpointIPAddressTypeToPB(addressType v1alpha1.EndpointIPAddressType) regionalapiv1.PrivateLinkService_Connection_IPAddressType { switch addressType { case v1alpha1.EndpointIPAddressTypeIPv4: return regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_IPV4 case v1alpha1.EndpointIPAddressTypeIPv6: return regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_IPV6 case v1alpha1.EndpointIPAddressTypeDualstack: return regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_DUALSTACK default: return regionalapiv1.PrivateLinkService_Connection_IP_ADDRESS_TYPE_UNSPECIFIED } } func setConnectionProviderInfo(info *regionalapiv1.PrivateLinkService_Connection, provider string, conn v1alpha1.Connection, connStatus *v1alpha1.ConnectionStatus) (*regionalapiv1.PrivateLinkService_Connection, error) { switch provider { case consts.ProviderTypeEKS: info.Info = ®ionalapiv1.PrivateLinkService_Connection_Aws_{ Aws: ®ionalapiv1.PrivateLinkService_Connection_Aws{ VpcEndpointId: conn.Endpoint, }, } case consts.ProviderTypeGKE: host := "" if connStatus != nil { host = connStatus.DNSName } info.Info = ®ionalapiv1.PrivateLinkService_Connection_Gcp_{ Gcp: ®ionalapiv1.PrivateLinkService_Connection_Gcp{ PrivateServiceConnectEndpoint: conn.Endpoint, Host: host, }, } case consts.ProviderTypeAKS: host, address := "", "" if connStatus != nil { host = connStatus.DNSName } if len(conn.IPs) > 0 { address = conn.IPs[0] } info.Info = ®ionalapiv1.PrivateLinkService_Connection_Azure_{ Azure: ®ionalapiv1.PrivateLinkService_Connection_Azure{ PrivateEndpointResourceId: conn.Endpoint, Host: host, PrivateIpAddress: address, }, } default: return nil, bizerr.New(bizerr.InternalServerError, "provider config invalid") } return info, nil }