/* Copyright 2022. Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License. */ package endpointservice import ( "context" "errors" "fmt" "strings" "time" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/log" "github.com/Azure/azure-sdk-for-go/sdk/azidentity" openapi "github.com/alibabacloud-go/darabonba-openapi/v2/client" "github.com/alibabacloud-go/tea/tea" "github.com/aliyun/credentials-go/credentials" awsconfig "github.com/aws/aws-sdk-go-v2/config" elbtypes "github.com/aws/aws-sdk-go-v2/service/elasticloadbalancingv2/types" "github.com/go-logr/logr" observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1" "github.com/tidbcloud/observability/pkg/common/constants" "github.com/tidbcloud/observability/pkg/common/metrics" "github.com/tidbcloud/observability/pkg/common/structs" "github.com/tidbcloud/observability/pkg/common/utils" "github.com/tidbcloud/observability/pkg/resourceworker/controllers/reconciler" "github.com/tidbcloud/observability/pkg/resourceworker/providers" alicloud "github.com/tidbcloud/observability/pkg/resourceworker/providers/alicloud" epsaws "github.com/tidbcloud/observability/pkg/resourceworker/providers/aws" "github.com/tidbcloud/observability/pkg/resourceworker/providers/azure" pscgcp "github.com/tidbcloud/observability/pkg/resourceworker/providers/gcp" computepb "google.golang.org/genproto/googleapis/cloud/compute/v1" k8sapiv1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/types" ) // EndpointServiceReconciler reconciles a EndpointService object type EndpointServiceReconciler struct { reconciler *reconciler.Reconciler Log logr.Logger } const ( endpointServicefinalizerName = "observability.pingcap.com/endpointService" errNotEndpointService = "not EndpointService resource" errNoServiceHostName = "service not have host name" errNoUserArnInSharedMode = "not found user arns" errELBCreateFailed = "elb create failed" labelID = "observability.pingcap.com/endpointid" labelShared = "observability.pingcap.com/shared" pollInterval = 3 * time.Second errNotSupportProvidedVendor = "not support provided vendor" projectConnectionLimit = 1000 ) func NewEndpointServiceReconciler(mgr ctrl.Manager, log logr.Logger, serviceInfo structs.ServiceInfo) *EndpointServiceReconciler { return &EndpointServiceReconciler{ Log: log, reconciler: &reconciler.Reconciler{ Client: mgr.GetClient(), ExternalObj: endpointServiceConnector{ cli: mgr.GetClient(), serviceInfo: serviceInfo, }, NewResource: func() client.Object { return &observabilitypingcapcomv1.EndpointService{} }, }, } } type endpointServiceConnector struct { serviceInfo structs.ServiceInfo cli client.Client } func (edpc endpointServiceConnector) getAWSClient() (providers.EndpointServiceClient, providers.ELBClient, error) { if edpc.serviceInfo.Vendor != string(constants.AWS) { return nil, nil, nil } // we should init aws config and client before start Observe cfg, err := awsconfig.LoadDefaultConfig(context.TODO()) if err != nil { return nil, nil, errors.New("unable to get aws config") } cfg.Region = edpc.serviceInfo.Region // get iam client return epsaws.NewEndpointService(cfg), epsaws.NewELB(cfg), nil } func (edpc endpointServiceConnector) getGCPClient() (providers.PSCClient, error) { if edpc.serviceInfo.Vendor != string(constants.GCP) { return nil, nil } // get iam client return pscgcp.NewPSC(edpc.serviceInfo.AccountInfo.Project, edpc.serviceInfo.Region) } func (edpc endpointServiceConnector) getAzureClient() (providers.PrivateLinkSvcClient, error) { if edpc.serviceInfo.Vendor != string(constants.Azure) { return nil, nil } cred, err := azidentity.NewDefaultAzureCredential(nil) if err != nil { return nil, err } return azure.NewPLSClient(edpc.serviceInfo.AccountInfo.SubscriptionID, edpc.serviceInfo.AccountInfo.ResourceGroup, cred) } func (edpc endpointServiceConnector) getAlicloudClient() (providers.AliEndpointServiceClient, error) { if edpc.serviceInfo.Vendor != string(constants.AliCloud) { return nil, nil } stsCredential, err := credentials.NewCredential(nil) if err != nil { return nil, err } config := &openapi.Config{} config.Credential = stsCredential config.Endpoint = tea.String(fmt.Sprintf("privatelink.%s.aliyuncs.com", edpc.serviceInfo.Region)) config.RegionId = &(edpc.serviceInfo.Region) return alicloud.NewAliEndpointService(config) } func (edpc endpointServiceConnector) Connect(ctx context.Context, obj client.Object) (_ reconciler.ExternalObjClient, errResult error) { epscli, elbcli, err := edpc.getAWSClient() if err != nil { return nil, err } gcpcli, err := edpc.getGCPClient() if err != nil { return nil, err } azCli, err := edpc.getAzureClient() if err != nil { return nil, err } aliCli, err := edpc.getAlicloudClient() if err != nil { return nil, err } return &endpointService{ epscli: epscli, elbcli: elbcli, gcpcli: gcpcli, azcli: azCli, alicli: aliCli, cli: edpc.cli, serviceInfo: edpc.serviceInfo, }, nil } type endpointService struct { epscli providers.EndpointServiceClient elbcli providers.ELBClient gcpcli providers.PSCClient azcli providers.PrivateLinkSvcClient alicli providers.AliEndpointServiceClient cli client.Client serviceInfo structs.ServiceInfo } // skipEndpointService reports whether this shared endpoint service does not need a // backing cloud VPC endpoint service. This is the AWS BYOC VPC-peering mode: the // region reaches the o11y write NLB over VPC peering instead of PrivateLink, so no // endpoint service is provisioned and the CR can go straight to ACTIVE. func (eps *endpointService) skipEndpointService(epsv1 *observabilitypingcapcomv1.EndpointService) bool { return epsv1.Spec.Shared && eps.serviceInfo.Vendor == string(constants.AWS) && len(eps.serviceInfo.BYOCVPCPeering.RouteTableIDs) > 0 && strings.TrimSpace(epsv1.Spec.EndpointServiceID) == "" } func (eps *endpointService) observeSharedEndpointService(ctx context.Context, log logr.Logger, epsv1 *observabilitypingcapcomv1.EndpointService) (_ reconciler.ExternalObservation, errResult error) { if eps.skipEndpointService(epsv1) { // no cloud endpoint service needed; drive Create to mark it ACTIVE return reconciler.ExternalObservation{ResourceExists: false}, nil } if epsv1.Spec.EndpointServiceID == "" { //wait EndpointServiceID update eps.reconcilerRepoter(epsv1.Name, pollInterval.String(), "") return reconciler.ExternalObservation{ResourceExists: true, ResourceUpToDate: false, PollInterval: pollInterval}, nil } else if epsv1.Status.ServiceID == "" { return reconciler.ExternalObservation{ResourceExists: false}, nil } // exists but Phase not get final state so we should update state return reconciler.ExternalObservation{ResourceExists: true}, nil } func (eps *endpointService) waitPromesvc(ctx context.Context, log logr.Logger, epsv1 *observabilitypingcapcomv1.EndpointService) (promsvc k8sapiv1.Service, needWait bool, err error) { //get service we should wait elb created err = eps.cli.Get(ctx, types.NamespacedName{Namespace: epsv1.Namespace, Name: epsv1.Spec.ServiceName}, &promsvc) if err != nil { if client.IgnoreNotFound(err) != nil { return promsvc, false, err } //wait service create needWait = true } if len(promsvc.Status.LoadBalancer.Ingress) == 0 { //wait needWait = true } return } func (eps *endpointService) observeEndpointService(ctx context.Context, log logr.Logger, epsv1 *observabilitypingcapcomv1.EndpointService) (_ reconciler.ExternalObservation, errResult error) { promsvc, needWait, err := eps.waitPromesvc(ctx, log, epsv1) if err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return reconciler.ExternalObservation{}, err } if needWait { eps.reconcilerRepoter(epsv1.Name, pollInterval.String(), "") return reconciler.ExternalObservation{ResourceExists: false, ResourceUpToDate: false, PollInterval: pollInterval}, nil } if len(epsv1.Status.ServiceID) == 0 { //if not exists then we should create //observe elb status elbname := strings.Split(promsvc.Status.LoadBalancer.Ingress[0].Hostname, "-")[0] lb, err := eps.elbcli.GetLoadBalancer(ctx, elbname) if err != nil { return reconciler.ExternalObservation{}, err } if lb.State.Code != elbtypes.LoadBalancerStateEnumActive { if lb.State.Code == elbtypes.LoadBalancerStateEnumFailed { //lb create failed log.V(1).Error(errors.New(errELBCreateFailed), "elb create failed", fmt.Sprintf("%v", lb)) //TODO now we do noting return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil } //wait return reconciler.ExternalObservation{ResourceExists: false, ResourceUpToDate: false, PollInterval: pollInterval}, nil } return reconciler.ExternalObservation{ResourceExists: false}, nil } // exists but Phase not get final state so we should update state return reconciler.ExternalObservation{ResourceExists: true}, nil } func (eps *endpointService) Observe(ctx context.Context, obj client.Object) (_ reconciler.ExternalObservation, errResult error) { log := log.FromContext(ctx).WithName("Observe") epsv1, ok := obj.(*observabilitypingcapcomv1.EndpointService) if !ok { eps.reconcilerRepoter(obj.GetName(), "", errNotEndpointService) return reconciler.ExternalObservation{Droped: true}, errors.New(errNotEndpointService) } log.V(1).Info("endpointService crd info", "crd", fmt.Sprintf("%v", epsv1)) if epsv1.Status.Phase == constants.ACTIVE || epsv1.Status.Phase == constants.CREATION_FAILED { // final state, nothing to do eps.reconcilerRepoter(epsv1.Name, "", "") return reconciler.ExternalObservation{ResourceUpToDate: true, ResourceExists: true}, nil } if epsv1.Spec.Shared { return eps.observeSharedEndpointService(ctx, log, epsv1) } return eps.observeEndpointService(ctx, log, epsv1) } func processARN(arn string) string { //make arn can apply to lable arn = strings.ReplaceAll(arn, ":", "") arn = strings.ReplaceAll(arn, "/", "") arn = strings.ReplaceAll(arn, "*", "") arn = strings.TrimPrefix(arn, "acsram") return strings.TrimPrefix(arn, "arnawsiam") } type endpointServiceWarp struct { ServiceId *string ServiceName *string } func (eps *endpointService) getEndpointService(ctx context.Context, epsv1 *observabilitypingcapcomv1.EndpointService) (*endpointServiceWarp, error) { switch eps.serviceInfo.Vendor { case string(constants.AWS): out, err := eps.epscli.GetEndpointService(ctx, epsv1.Spec.EndpointServiceID) return func() *endpointServiceWarp { if out == nil { return nil } return &endpointServiceWarp{ ServiceId: out.ServiceId, ServiceName: out.ServiceName, } }(), err case string(constants.GCP): out, err := eps.gcpcli.GetPSCService(ctx, epsv1.Spec.EndpointServiceID) return func() *endpointServiceWarp { if out == nil { return nil } return &endpointServiceWarp{ ServiceId: out.Name, ServiceName: utils.StringPtr(strings.TrimPrefix(*out.SelfLink, "https://www.googleapis.com/compute/v1/")), } }(), err case string(constants.Azure): pls, err := eps.azcli.Get(ctx, epsv1.Spec.EndpointServiceID) return func() *endpointServiceWarp { if err != nil { return nil } return &endpointServiceWarp{ // it's intended to reverse the name and id, just like what we do in gcp. // since we use serviceId as serviceName later to operate with plsClient.AddUserPermissions ServiceId: pls.Name, ServiceName: pls.ID, } }(), err case string(constants.AliCloud): out, err := eps.alicli.GetAliEndpointService(ctx, epsv1.Spec.EndpointServiceID) return func() *endpointServiceWarp { if out == nil { return nil } return &endpointServiceWarp{ ServiceId: out.ServiceId, ServiceName: out.ServiceName, } }(), err } return nil, errors.New(errNotSupportProvidedVendor) } func (eps *endpointService) createSharedEndpointService(ctx context.Context, epsv1 *observabilitypingcapcomv1.EndpointService) (*observabilitypingcapcomv1.EndpointService, error) { // if shared we use exist endpoint service out, err := eps.getEndpointService(ctx, epsv1) if err != nil { return nil, err } if len(epsv1.Spec.UserARNs) == 0 { return nil, errors.New(errNoUserArnInSharedMode) } if epsv1.Labels == nil { epsv1.Labels = make(map[string]string) } epsv1.Labels[labelID] = *out.ServiceId epsv1.Labels[labelShared] = processARN(epsv1.Spec.UserARNs[0]) if err := eps.cli.Update(ctx, epsv1); err != nil { return nil, err } epsv1.Status.ServiceID = *out.ServiceId epsv1.Status.ServiceName = *out.ServiceName return epsv1, nil } func (eps *endpointService) createEndpointService(ctx context.Context, epsv1 *observabilitypingcapcomv1.EndpointService) (*observabilitypingcapcomv1.EndpointService, error) { //this function use endpoint service crd name to do idempotent var promsvc k8sapiv1.Service err := eps.cli.Get(ctx, types.NamespacedName{Namespace: epsv1.Namespace, Name: epsv1.Spec.ServiceName}, &promsvc) if err != nil { return nil, err } if len(promsvc.Status.LoadBalancer.Ingress) == 0 { return nil, errors.New(errNoServiceHostName) } //get elb elbname := strings.Split(promsvc.Status.LoadBalancer.Ingress[0].Hostname, "-")[0] lb, err := eps.elbcli.GetLoadBalancer(ctx, elbname) if err != nil { return nil, err } out, err := eps.epscli.CreateEndpointService(ctx, epsv1.Name, []string{*lb.LoadBalancerArn}) if err != nil { return nil, err } epsv1.Status.ServiceID = *out.ServiceConfiguration.ServiceId epsv1.Status.ServiceName = *out.ServiceConfiguration.ServiceName return epsv1, nil } func (eps *endpointService) Create(ctx context.Context, obj client.Object) (errResult error) { epsv1, ok := obj.(*observabilitypingcapcomv1.EndpointService) if !ok { eps.reconcilerRepoter(obj.GetName(), "", errNotEndpointService) return errors.New(errNotEndpointService) } if eps.skipEndpointService(epsv1) { // AWS BYOC VPC-peering mode: no cloud endpoint service to create, mark ACTIVE directly. epsv1.Status.Phase = constants.ACTIVE if err := eps.cli.Status().Update(ctx, epsv1); err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return err } eps.reconcilerRepoter(epsv1.Name, "", "") return nil } var err error if epsv1.Spec.Shared { epsv1, err = eps.createSharedEndpointService(ctx, epsv1) } else { epsv1, err = eps.createEndpointService(ctx, epsv1) } if err != nil { eps.reconcilerRepoter(obj.GetName(), "", err.Error()) return err } epsv1.Status.Phase = constants.CREATING if err := eps.cli.Status().Update(ctx, epsv1); err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return err } eps.reconcilerRepoter(epsv1.Name, "", "") return nil } func isIncludeProject(projectlist []*computepb.ServiceAttachmentConsumerProjectLimit, project string) bool { for _, v := range projectlist { if v.ProjectIdOrNum == &project { return true } } return false } func (eps *endpointService) addUserPermissionsGCP(ctx context.Context, serviceID string, userARNs []string) error { //get psc config config, err := eps.gcpcli.GetPSCService(ctx, serviceID) if err != nil { return err } changed := false //add project if not in config for i, v := range userARNs { if !isIncludeProject(config.ConsumerAcceptLists, v) { changed = true config.ConsumerAcceptLists = append(config.ConsumerAcceptLists, &computepb.ServiceAttachmentConsumerProjectLimit{ ProjectIdOrNum: &userARNs[i], ConnectionLimit: utils.Uint32Ptr(projectConnectionLimit), }) } } if changed { err = eps.gcpcli.SetPSCService(ctx, serviceID, config) } return err } func (eps *endpointService) addUserPermissions(ctx context.Context, serviceID string, userARNs []string) error { switch eps.serviceInfo.Vendor { case string(constants.AWS): return eps.epscli.AddUserPermissions(ctx, serviceID, userARNs) case string(constants.GCP): return eps.addUserPermissionsGCP(ctx, serviceID, userARNs) case string(constants.Azure): return eps.azcli.AddUserPermissions(ctx, serviceID, userARNs) case string(constants.AliCloud): return eps.alicli.AddUserPermissions(ctx, serviceID, userARNs) } return errors.New(errNotSupportProvidedVendor) } func (eps *endpointService) Update(ctx context.Context, obj client.Object) (errResult error) { epsv1, ok := obj.(*observabilitypingcapcomv1.EndpointService) if !ok { eps.reconcilerRepoter(obj.GetName(), "", errNotEndpointService) return errors.New(errNotEndpointService) } if epsv1.Status.Phase == constants.CREATING { //modify user access permissions err := eps.addUserPermissions(ctx, epsv1.Status.ServiceID, epsv1.Spec.UserARNs) if err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return err } epsv1.Status.Phase = constants.ACTIVE if err := eps.cli.Status().Update(ctx, epsv1); err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return err } } eps.reconcilerRepoter(epsv1.Name, "", "") return nil } func (eps *endpointService) deleteUserPermissionsGCP(ctx context.Context, serviceID string, userARNs []string) error { log := log.FromContext(ctx).WithName("Delete.deleteSharedEndpointService.deleteUserPermissionsGCP") //get config config, err := eps.gcpcli.GetPSCService(ctx, serviceID) if err != nil { return err } //remove access policy and set var saved []*computepb.ServiceAttachmentConsumerProjectLimit for i, v := range config.ConsumerAcceptLists { if !utils.IsIncludeString(userARNs, *v.ProjectIdOrNum) { //save all project if not be deleted saved = append(saved, config.ConsumerAcceptLists[i]) } } log.V(1).Info("saved accept list", "saved", saved) config.ConsumerAcceptLists = saved return eps.gcpcli.SetPSCService(ctx, serviceID, config) } func (eps *endpointService) deleteUserPermissions(ctx context.Context, serviceID string, userARNs []string) error { switch eps.serviceInfo.Vendor { case string(constants.AWS): return eps.epscli.DeleteUserPermissions(ctx, serviceID, userARNs) case string(constants.GCP): return eps.deleteUserPermissionsGCP(ctx, serviceID, userARNs) case string(constants.Azure): return eps.azcli.DeleteUserPermissions(ctx, serviceID, userARNs) case string(constants.AliCloud): return eps.alicli.DeleteUserPermissions(ctx, serviceID, userARNs) } return errors.New(errNotSupportProvidedVendor) } func (eps *endpointService) deleteSharedEndpointService(ctx context.Context, epsv1 *observabilitypingcapcomv1.EndpointService) error { log := log.FromContext(ctx).WithName("Delete.deleteSharedEndpointService") // if userARN is empty we do nothing if len(epsv1.Spec.UserARNs) == 0 { return nil } //check Ref count var epsv1list observabilitypingcapcomv1.EndpointServiceList err := eps.cli.List(ctx, &epsv1list, &client.ListOptions{ LabelSelector: labels.SelectorFromSet(map[string]string{labelShared: processARN(epsv1.Spec.UserARNs[0]), labelID: epsv1.Spec.EndpointServiceID}), }) if err != nil { return err } //if shared and rc <=1 we only remove user access permission if len(epsv1list.Items) <= 1 { log.V(1).Info("only one ref,we will delete access permission in cloud", "id", epsv1.Status.ServiceID, "users", epsv1.Spec.UserARNs) err = eps.deleteUserPermissions(ctx, epsv1.Status.ServiceID, epsv1.Spec.UserARNs) } return err } func (eps *endpointService) Delete(ctx context.Context, obj client.Object) (_ reconciler.ExternalDeletion, errResult error) { epsv1, ok := obj.(*observabilitypingcapcomv1.EndpointService) if !ok { eps.reconcilerRepoter(obj.GetName(), "", errNotEndpointService) return reconciler.ExternalDeletion{}, errors.New(errNotEndpointService) } if len(epsv1.Status.ServiceID) != 0 { var err error if epsv1.Spec.Shared { err = eps.deleteSharedEndpointService(ctx, epsv1) } else { err = eps.epscli.DeleteEndpointService(ctx, epsv1.Status.ServiceID) } if err != nil { eps.reconcilerRepoter(epsv1.Name, "", err.Error()) return reconciler.ExternalDeletion{ResourceDeleted: false}, err } } eps.reconcilerRepoter(epsv1.Name, "", "") return reconciler.ExternalDeletion{ResourceDeleted: true}, nil } // Reconcile is part of the main kubernetes reconciliation loop which aims to // move the current state of the cluster closer to the desired state. // TODO(user): Modify the Reconcile function to compare the state specified by // the EndpointService object against the actual cluster state, and then // perform operations to make the cluster state reflect the state specified by // the user. // // For more details, check Reconcile and its Result here: // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.2/pkg/reconcile func (r *EndpointServiceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { logger := r.Log.WithValues("EndpointService", req.NamespacedName) ctx = log.IntoContext(ctx, logger) return r.reconciler.Reconcile(ctx, req, endpointServicefinalizerName) } // SetupWithManager sets up the controller with the Manager. func (r *EndpointServiceReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). For(&observabilitypingcapcomv1.EndpointService{}). Complete(r) } func (eps *endpointService) reconcilerRepoter(name string, interval string, err string) { metrics.HandleRegionalControllerRunsMetrics(eps.serviceInfo.Vendor, eps.serviceInfo.Region, "EndpointServiceController", name, interval, err) }