// RAINBOND, Application Management Platform // Copyright (C) 2014-2017 Goodrain Co., Ltd. // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. For any non-GPL usage of Rainbond, // one or multiple Commercial Licenses authorized by Goodrain Co., Ltd. // must be obtained first. // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // You should have received a copy of the GNU General Public License // along with this program. If not, see . package handler import ( "context" "crypto/sha256" "crypto/x509" "encoding/json" "encoding/pem" "fmt" "os" "sort" "strconv" "strings" "time" v2 "github.com/apache/apisix-ingress-controller/pkg/kube/apisix/apis/config/v2" apisixversioned "github.com/apache/apisix-ingress-controller/pkg/kube/apisix/client/clientset/versioned" apimodel "github.com/goodrain/rainbond/api/model" apiutil "github.com/goodrain/rainbond/api/util" "github.com/goodrain/rainbond/api/util/bcode" "github.com/goodrain/rainbond/db" "github.com/goodrain/rainbond/db/model" "github.com/goodrain/rainbond/mq/client" "github.com/goodrain/rainbond/pkg/component/k8s" "github.com/goodrain/rainbond/pkg/component/mq" "github.com/goodrain/rainbond/util" "github.com/goodrain/rainbond/worker/appm/controller" "github.com/jinzhu/gorm" "github.com/pkg/errors" "github.com/sirupsen/logrus" corev1 "k8s.io/api/core/v1" k8serror "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" k8svalidation "k8s.io/apimachinery/pkg/util/validation" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" v1 "sigs.k8s.io/gateway-api/apis/v1" "sigs.k8s.io/gateway-api/apis/v1beta1" gateway "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned/typed/apis/v1beta1" ) // GatewayAction - type GatewayAction struct { dbmanager db.Manager mqclient client.MQClient gatewayClient *gateway.GatewayV1beta1Client kubeClient kubernetes.Interface kubeClientset *kubernetes.Clientset config *rest.Config apisixClient apisixversioned.Interface domainConflictChecker func(context.Context, []v2.HostType, string, string) error } const ( gatewayClientCASecretLabel = "rainbond.io/certificate-type" gatewayClientCASecretLabelValue = "client-ca" gatewayMTLSManagedLabel = "rainbond.io/mtls-managed" gatewayMTLSDomainAnnotation = "rainbond.io/mtls-domain" gatewayMTLSParentAnnotation = "rainbond.io/mtls-parent" gatewayMTLSDetachedAnnotation = "rainbond.io/mtls-host-detached" gatewayMTLSInPlaceAnnotation = "rainbond.io/mtls-in-place" gatewayClientCASecretDataKey = corev1.ServiceAccountRootCAKey gatewayMTLSCertificateChainDepth = 1 ) // CreateGatewayManager creates gateway manager. func CreateGatewayManager() *GatewayAction { return &GatewayAction{ dbmanager: db.GetManager(), mqclient: mq.Default().MqClient, gatewayClient: k8s.Default().GatewayClient, kubeClient: k8s.Default().Clientset, kubeClientset: k8s.Default().Clientset, config: k8s.Default().RestConfig, apisixClient: k8s.Default().ApiSixClient, } } // GetClient - func (g *GatewayAction) GetClient() apisixversioned.Interface { return g.apisixClient } // GetK8sClient - func (g *GatewayAction) GetK8sClient() kubernetes.Interface { return g.kubeClient } // CreateCert - func (g *GatewayAction) CreateCert(namespace, domain string) error { secretName := strings.Replace(domain, ".", "-", -1) // Generate self-signed certificate cert, certKey, err := generateSelfSignedCertificate(domain) if err != nil { logrus.Errorf("Error generating self-signed certificate: %v", err) return err } // Create Kubernetes Secret return createK8sSecret(g.kubeClientset, namespace, secretName, cert, certKey) } // BatchGetGatewayHTTPRoute batch get gateway http route func (g *GatewayAction) BatchGetGatewayHTTPRoute(namespace, appID string) ([]*apimodel.GatewayHTTPRouteConcise, error) { var httpRoutes []v1beta1.HTTPRoute if appID != "" { gatewayHTTPRoutes, err := g.gatewayClient.HTTPRoutes(namespace).List(context.Background(), metav1.ListOptions{LabelSelector: "app_id=" + appID}) if err != nil { logrus.Errorf("list http route by app_id = %v failure: %v", appID, err) return nil, err } httpRoutes = gatewayHTTPRoutes.Items } else { gatewayHTTPRoutes, err := g.gatewayClient.HTTPRoutes(namespace).List(context.Background(), metav1.ListOptions{}) if err != nil { logrus.Errorf("list http route failure: %v", err) return nil, err } httpRoutes = gatewayHTTPRoutes.Items } var HTTPRouteConcise []*apimodel.GatewayHTTPRouteConcise for _, httpRoute := range httpRoutes { gatewayName, gatewayNamespace, _ := gatewayParentRefValues(httpRoute.Spec.ParentRefs, namespace) var hosts []string if httpRoute.Spec.Hostnames != nil { for _, hostname := range httpRoute.Spec.Hostnames { hosts = append(hosts, string(hostname)) } } var id string if httpRoute.Labels != nil { id = httpRoute.Labels["app_id"] } HTTPRouteConcise = append(HTTPRouteConcise, &apimodel.GatewayHTTPRouteConcise{ Name: httpRoute.Name, Hosts: hosts, GatewayName: gatewayName, GatewayNamespace: gatewayNamespace, AppID: id, }) } return HTTPRouteConcise, nil } // AddGatewayCertificate create gateway certificate func (g *GatewayAction) AddGatewayCertificate(req *apimodel.GatewayCertificate) error { return g.reconcileGatewayCertificate(req) } // UpdateGatewayCertificate update gateway certificate func (g *GatewayAction) UpdateGatewayCertificate(req *apimodel.GatewayCertificate) error { return g.reconcileGatewayCertificate(req) } func validateGatewayClientCA(certificate string) error { rest := []byte(certificate) foundCertificate := false now := time.Now() for len(rest) > 0 { block, remaining := pem.Decode(rest) if block == nil { if strings.TrimSpace(string(rest)) != "" { return fmt.Errorf("client CA must contain only PEM encoded certificates") } break } rest = remaining if block.Type != "CERTIFICATE" { return fmt.Errorf("client CA contains unsupported PEM block %q", block.Type) } cert, err := x509.ParseCertificate(block.Bytes) if err != nil { return fmt.Errorf("parse client CA certificate: %w", err) } if now.Before(cert.NotBefore) || now.After(cert.NotAfter) { return fmt.Errorf("client CA certificate is not currently valid") } if !cert.BasicConstraintsValid || !cert.IsCA { return fmt.Errorf("client CA certificate must have CA basic constraints") } foundCertificate = true } if !foundCertificate { return fmt.Errorf("client CA certificate is required") } return nil } func validateGatewayResourceName(name string) error { if errs := k8svalidation.IsDNS1123Subdomain(name); len(errs) > 0 { return fmt.Errorf("invalid Kubernetes resource name: %s", strings.Join(errs, "; ")) } return nil } // AddGatewayClientCA creates a Kubernetes Secret containing a validated client CA certificate. func (g *GatewayAction) AddGatewayClientCA(namespace string, req *apimodel.GatewayClientCA) error { if req == nil { return fmt.Errorf("client CA request is required") } if err := validateGatewayResourceName(req.Name); err != nil { return err } if err := validateGatewayClientCA(req.Certificate); err != nil { return err } _, err := g.kubeClient.CoreV1().Secrets(namespace).Create(context.Background(), &corev1.Secret{ ObjectMeta: metav1.ObjectMeta{ Name: req.Name, Namespace: namespace, Labels: map[string]string{ gatewayClientCASecretLabel: gatewayClientCASecretLabelValue, }, }, Type: corev1.SecretTypeOpaque, Data: map[string][]byte{ gatewayClientCASecretDataKey: []byte(req.Certificate), }, }, metav1.CreateOptions{}) if err != nil { return fmt.Errorf("create gateway client CA secret: %w", err) } return nil } // UpdateGatewayClientCA updates a Kubernetes Secret containing a validated client CA certificate. func (g *GatewayAction) UpdateGatewayClientCA(namespace string, req *apimodel.GatewayClientCA) error { if req == nil { return fmt.Errorf("client CA request is required") } if err := validateGatewayResourceName(req.Name); err != nil { return err } if err := validateGatewayClientCA(req.Certificate); err != nil { return err } secrets := g.kubeClient.CoreV1().Secrets(namespace) secret, err := secrets.Get(context.Background(), req.Name, metav1.GetOptions{}) if err != nil { return fmt.Errorf("get gateway client CA secret: %w", err) } updated := secret.DeepCopy() if updated.Labels == nil { updated.Labels = make(map[string]string) } updated.Labels[gatewayClientCASecretLabel] = gatewayClientCASecretLabelValue updated.Type = corev1.SecretTypeOpaque updated.Data = map[string][]byte{gatewayClientCASecretDataKey: []byte(req.Certificate)} if _, err := secrets.Update(context.Background(), updated, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("update gateway client CA secret: %w", err) } return nil } // ListGatewayClientCAs lists Rainbond-managed client CA Secrets and their bound domains. func (g *GatewayAction) ListGatewayClientCAs(namespace string) ([]*apimodel.GatewayClientCAStatus, error) { secrets, err := g.kubeClient.CoreV1().Secrets(namespace).List(context.Background(), metav1.ListOptions{ LabelSelector: gatewayClientCASecretLabel + "=" + gatewayClientCASecretLabelValue, }) if err != nil { return nil, fmt.Errorf("list gateway client CA secrets: %w", err) } tlsList, err := g.apisixClient.ApisixV2().ApisixTlses(namespace).List(context.Background(), metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("list ApisixTls resources: %w", err) } boundDomains := make(map[string]map[string]struct{}) for i := range tlsList.Items { tls := &tlsList.Items[i] if tls.Spec == nil || tls.Spec.Client == nil || tls.Spec.Client.CASecret.Namespace != namespace { continue } name := tls.Spec.Client.CASecret.Name if boundDomains[name] == nil { boundDomains[name] = make(map[string]struct{}) } for _, host := range tls.Spec.Hosts { boundDomains[name][string(host)] = struct{}{} } } statuses := make([]*apimodel.GatewayClientCAStatus, 0, len(secrets.Items)) for i := range secrets.Items { secret := &secrets.Items[i] domains := make([]string, 0, len(boundDomains[secret.Name])) for domain := range boundDomains[secret.Name] { domains = append(domains, domain) } sort.Strings(domains) statuses = append(statuses, &apimodel.GatewayClientCAStatus{Name: secret.Name, BoundDomains: domains}) } sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name }) return statuses, nil } // DeleteGatewayClientCA deletes an unreferenced Rainbond-managed client CA Secret. func (g *GatewayAction) DeleteGatewayClientCA(namespace, name string) error { if err := validateGatewayResourceName(name); err != nil { return err } tlsList, err := g.apisixClient.ApisixV2().ApisixTlses(namespace).List(context.Background(), metav1.ListOptions{}) if err != nil { return fmt.Errorf("list ApisixTls resources: %w", err) } for i := range tlsList.Items { tls := &tlsList.Items[i] if tls.Spec != nil && tls.Spec.Client != nil && tls.Spec.Client.CASecret.Name == name && tls.Spec.Client.CASecret.Namespace == namespace { return fmt.Errorf("client CA is still used by domain %s", strings.Join(hostTypesToStrings(tls.Spec.Hosts), ",")) } } err = g.kubeClient.CoreV1().Secrets(namespace).Delete(context.Background(), name, metav1.DeleteOptions{}) if err != nil && !k8serror.IsNotFound(err) { return fmt.Errorf("delete gateway client CA secret: %w", err) } return nil } func hostTypesToStrings(hosts []v2.HostType) []string { result := make([]string, 0, len(hosts)) for _, host := range hosts { result = append(result, string(host)) } return result } func gatewayMTLSResourceName(domain string) string { sum := sha256.Sum256([]byte(domain)) return fmt.Sprintf("rbd-mtls-%x", sum[:8]) } func validateGatewayMTLSDomain(domain string) error { var validationErrors []string if strings.HasPrefix(domain, "*.") { validationErrors = k8svalidation.IsWildcardDNS1123Subdomain(domain) } else { validationErrors = k8svalidation.IsDNS1123Subdomain(domain) } if len(validationErrors) > 0 { return fmt.Errorf("a valid gateway domain is required") } return nil } func gatewayDomainMatchesHost(domain string, host v2.HostType) bool { pattern := string(host) if pattern == domain { return true } if !strings.HasPrefix(pattern, "*.") { return false } suffix := pattern[1:] if !strings.HasSuffix(domain, suffix) { return false } prefix := strings.TrimSuffix(domain, suffix) return prefix != "" && !strings.Contains(prefix, ".") } func gatewayTLSHasExactHost(tls *v2.ApisixTls, domain string) bool { if tls == nil || tls.Spec == nil { return false } for _, host := range tls.Spec.Hosts { if string(host) == domain { return true } } return false } func gatewayTLSManagedForDomain(tls *v2.ApisixTls, domain string) bool { return tls != nil && tls.Labels[gatewayMTLSManagedLabel] == "true" && tls.Annotations[gatewayMTLSDomainAnnotation] == domain } func selectGatewayTLSForDomain(items []v2.ApisixTls, domain string) *v2.ApisixTls { for i := range items { if gatewayTLSManagedForDomain(&items[i], domain) { return &items[i] } } for i := range items { if gatewayTLSHasExactHost(&items[i], domain) { return &items[i] } } for i := range items { tls := &items[i] if tls.Spec == nil { continue } for _, host := range tls.Spec.Hosts { if gatewayDomainMatchesHost(domain, host) { return tls } } } return nil } func gatewayMTLSClient(namespace, caSecretName string) *v2.ApisixMutualTlsClientConfig { return &v2.ApisixMutualTlsClientConfig{ CASecret: v2.ApisixSecret{Name: caSecretName, Namespace: namespace}, Depth: gatewayMTLSCertificateChainDepth, } } func ensureGatewayMTLSMetadata(tls *v2.ApisixTls, domain, parent string, detached, inPlace bool) { if tls.Labels == nil { tls.Labels = make(map[string]string) } if tls.Annotations == nil { tls.Annotations = make(map[string]string) } tls.Labels[gatewayMTLSManagedLabel] = "true" tls.Annotations[gatewayMTLSDomainAnnotation] = domain tls.Annotations[gatewayMTLSParentAnnotation] = parent tls.Annotations[gatewayMTLSDetachedAnnotation] = strconv.FormatBool(detached) tls.Annotations[gatewayMTLSInPlaceAnnotation] = strconv.FormatBool(inPlace) } func removeGatewayMTLSMetadata(tls *v2.ApisixTls) { delete(tls.Labels, gatewayMTLSManagedLabel) delete(tls.Annotations, gatewayMTLSDomainAnnotation) delete(tls.Annotations, gatewayMTLSParentAnnotation) delete(tls.Annotations, gatewayMTLSDetachedAnnotation) delete(tls.Annotations, gatewayMTLSInPlaceAnnotation) } // ConfigureGatewayDomainMTLS enables or updates inbound mTLS for one HTTPS domain. func (g *GatewayAction) ConfigureGatewayDomainMTLS(namespace string, req *apimodel.GatewayDomainMTLS) error { if req == nil { return fmt.Errorf("a valid gateway domain is required") } if err := validateGatewayMTLSDomain(req.Domain); err != nil { return err } if err := validateGatewayResourceName(req.ClientCASecretName); err != nil { return err } caSecret, err := g.kubeClient.CoreV1().Secrets(namespace).Get(context.Background(), req.ClientCASecretName, metav1.GetOptions{}) if err != nil { return fmt.Errorf("get gateway client CA secret: %w", err) } if caSecret.Labels[gatewayClientCASecretLabel] != gatewayClientCASecretLabelValue { return fmt.Errorf("client CA secret is not managed by Rainbond") } if err := validateGatewayClientCA(string(caSecret.Data[gatewayClientCASecretDataKey])); err != nil { return fmt.Errorf("validate gateway client CA secret: %w", err) } tlses := g.apisixClient.ApisixV2().ApisixTlses(namespace) list, err := tlses.List(context.Background(), metav1.ListOptions{}) if err != nil { return fmt.Errorf("list ApisixTls resources: %w", err) } source := selectGatewayTLSForDomain(list.Items, req.Domain) if source == nil || source.Spec == nil { return fmt.Errorf("domain %s has no HTTPS server certificate", req.Domain) } if gatewayTLSManagedForDomain(source, req.Domain) { updated := source.DeepCopy() updated.Spec.Client = gatewayMTLSClient(namespace, req.ClientCASecretName) if _, err := tlses.Update(context.Background(), updated, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("update domain mTLS configuration: %w", err) } return nil } if len(source.Spec.Hosts) == 1 && gatewayTLSHasExactHost(source, req.Domain) { updated := source.DeepCopy() updated.Spec.Client = gatewayMTLSClient(namespace, req.ClientCASecretName) ensureGatewayMTLSMetadata(updated, req.Domain, source.Name, false, true) if _, err := tlses.Update(context.Background(), updated, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("update domain mTLS configuration: %w", err) } return nil } detached := gatewayTLSHasExactHost(source, req.Domain) previousSource := source.DeepCopy() if detached { updatedSource := source.DeepCopy() updatedSource.Spec.Hosts = make([]v2.HostType, 0, len(source.Spec.Hosts)-1) for _, host := range source.Spec.Hosts { if string(host) != req.Domain { updatedSource.Spec.Hosts = append(updatedSource.Spec.Hosts, host) } } persistedSource, updateErr := tlses.Update(context.Background(), updatedSource, metav1.UpdateOptions{}) if updateErr != nil { return fmt.Errorf("detach domain from server TLS configuration: %w", updateErr) } previousSource.ResourceVersion = persistedSource.ResourceVersion } child := &v2.ApisixTls{ TypeMeta: source.TypeMeta, ObjectMeta: metav1.ObjectMeta{ Name: gatewayMTLSResourceName(req.Domain), Namespace: namespace, }, Spec: &v2.ApisixTlsSpec{ IngressClassName: source.Spec.IngressClassName, Hosts: []v2.HostType{v2.HostType(req.Domain)}, Secret: source.Spec.Secret, Client: gatewayMTLSClient(namespace, req.ClientCASecretName), }, } ensureGatewayMTLSMetadata(child, req.Domain, source.Name, detached, false) if _, err := tlses.Create(context.Background(), child, metav1.CreateOptions{}); err != nil { if detached { if _, rollbackErr := tlses.Update(context.Background(), previousSource, metav1.UpdateOptions{}); rollbackErr != nil { return fmt.Errorf("create domain mTLS configuration: %v; restore server TLS configuration: %v", err, rollbackErr) } } return fmt.Errorf("create domain mTLS configuration: %w", err) } return nil } // DisableGatewayDomainMTLS disables inbound mTLS for one HTTPS domain. func (g *GatewayAction) DisableGatewayDomainMTLS(namespace, domain string) error { if err := validateGatewayMTLSDomain(domain); err != nil { return err } tlses := g.apisixClient.ApisixV2().ApisixTlses(namespace) list, err := tlses.List(context.Background(), metav1.ListOptions{}) if err != nil { return fmt.Errorf("list ApisixTls resources: %w", err) } configured := selectGatewayTLSForDomain(list.Items, domain) if configured == nil || configured.Spec == nil || configured.Spec.Client == nil { return nil } if !gatewayTLSManagedForDomain(configured, domain) || configured.Annotations[gatewayMTLSInPlaceAnnotation] == "true" { updated := configured.DeepCopy() updated.Spec.Client = nil removeGatewayMTLSMetadata(updated) if _, err := tlses.Update(context.Background(), updated, metav1.UpdateOptions{}); err != nil { return fmt.Errorf("disable domain mTLS configuration: %w", err) } return nil } child := configured.DeepCopy() parentName := child.Annotations[gatewayMTLSParentAnnotation] detached := child.Annotations[gatewayMTLSDetachedAnnotation] == "true" if err := tlses.Delete(context.Background(), child.Name, metav1.DeleteOptions{}); err != nil && !k8serror.IsNotFound(err) { return fmt.Errorf("delete domain mTLS configuration: %w", err) } if !detached { return nil } parent, err := tlses.Get(context.Background(), parentName, metav1.GetOptions{}) if err != nil { if _, recreateErr := tlses.Create(context.Background(), child, metav1.CreateOptions{}); recreateErr != nil { return fmt.Errorf("restore domain to server TLS configuration: %v; recreate mTLS configuration: %v", err, recreateErr) } return fmt.Errorf("restore domain to server TLS configuration: %w", err) } updatedParent := parent.DeepCopy() if !gatewayTLSHasExactHost(updatedParent, domain) { updatedParent.Spec.Hosts = append(updatedParent.Spec.Hosts, v2.HostType(domain)) } if _, err := tlses.Update(context.Background(), updatedParent, metav1.UpdateOptions{}); err != nil { child.ResourceVersion = "" if _, recreateErr := tlses.Create(context.Background(), child, metav1.CreateOptions{}); recreateErr != nil { return fmt.Errorf("restore domain to server TLS configuration: %v; recreate mTLS configuration: %v", err, recreateErr) } return fmt.Errorf("restore domain to server TLS configuration: %w", err) } return nil } // GetGatewayDomainMTLS returns inbound mTLS status for the requested domains in input order. func (g *GatewayAction) GetGatewayDomainMTLS(namespace string, domains []string) ([]*apimodel.GatewayDomainMTLS, error) { list, err := g.apisixClient.ApisixV2().ApisixTlses(namespace).List(context.Background(), metav1.ListOptions{}) if err != nil { return nil, fmt.Errorf("list ApisixTls resources: %w", err) } statuses := make([]*apimodel.GatewayDomainMTLS, 0, len(domains)) for _, domain := range domains { status := &apimodel.GatewayDomainMTLS{Domain: domain} tls := selectGatewayTLSForDomain(list.Items, domain) if tls != nil && tls.Spec != nil && tls.Spec.Client != nil { status.Enabled = true status.ClientCASecretName = tls.Spec.Client.CASecret.Name } statuses = append(statuses, status) } return statuses, nil } func (g *GatewayAction) reconcileGatewayCertificate(req *apimodel.GatewayCertificate) (retErr error) { ctx := context.Background() secrets := g.kubeClient.CoreV1().Secrets(req.Namespace) desiredData := map[string][]byte{ corev1.TLSCertKey: []byte(req.Certificate), corev1.TLSPrivateKeyKey: []byte(req.PrivateKey), } currentSecret, err := secrets.Get(ctx, req.Name, metav1.GetOptions{}) var previousSecret *corev1.Secret var persistedSecret *corev1.Secret secretCreated := false switch { case k8serror.IsNotFound(err): persistedSecret, err = secrets.Create(ctx, &corev1.Secret{ TypeMeta: metav1.TypeMeta{ Kind: apimodel.Secret, APIVersion: controller.APIVersionSecret, }, ObjectMeta: metav1.ObjectMeta{ Name: req.Name, Namespace: req.Namespace, }, Data: desiredData, Type: corev1.SecretTypeTLS, }, metav1.CreateOptions{}) if err != nil { logrus.Errorf("create gateway certificate secret failure: %v", err) return err } secretCreated = true case err != nil: logrus.Errorf("get gateway certificate secret failure: %v", err) return err default: previousSecret = currentSecret.DeepCopy() updatedSecret := currentSecret.DeepCopy() updatedSecret.Data = desiredData updatedSecret.Type = corev1.SecretTypeTLS persistedSecret, err = secrets.Update(ctx, updatedSecret, metav1.UpdateOptions{}) if err != nil { logrus.Errorf("update gateway certificate secret failure: %v", err) return err } } defer func() { if retErr == nil { return } var rollbackErr error if secretCreated { rollbackErr = secrets.Delete(ctx, req.Name, metav1.DeleteOptions{}) if k8serror.IsNotFound(rollbackErr) { rollbackErr = nil } } else { previousSecret.ResourceVersion = persistedSecret.ResourceVersion _, rollbackErr = secrets.Update(ctx, previousSecret, metav1.UpdateOptions{}) } if rollbackErr != nil { logrus.Errorf("rollback gateway certificate secret failure: %v", rollbackErr) retErr = fmt.Errorf("%w; rollback gateway certificate secret: %v", retErr, rollbackErr) } }() hosts, err := apiutil.GetCertificateDomains(persistedSecret) if err != nil { logrus.Errorf("get certificate domains failure: %v", err) return err } apisixTlses := g.apisixClient.ApisixV2().ApisixTlses(req.Namespace) tlsList, err := apisixTlses.List(ctx, metav1.ListOptions{}) if err != nil { logrus.Errorf("list gateway certificate apisix tls resources failure: %v", err) return err } detachedDomains := make(map[string]struct{}) for i := range tlsList.Items { tls := &tlsList.Items[i] if tls.Labels[gatewayMTLSManagedLabel] == "true" && tls.Annotations[gatewayMTLSParentAnnotation] == req.Name && tls.Annotations[gatewayMTLSDetachedAnnotation] == "true" { detachedDomains[tls.Annotations[gatewayMTLSDomainAnnotation]] = struct{}{} } } if len(detachedDomains) > 0 { baseHosts := make([]v2.HostType, 0, len(hosts)) for _, host := range hosts { if _, detached := detachedDomains[string(host)]; !detached { baseHosts = append(baseHosts, host) } } hosts = baseHosts } domainConflictChecker := g.domainConflictChecker if domainConflictChecker == nil { domainConflictChecker = apiutil.CheckDomainConflict } if err := domainConflictChecker(ctx, hosts, req.Namespace, req.Name); err != nil { logrus.Errorf("domain conflict detected: %s", err.Error()) return err } desiredSpec := &v2.ApisixTlsSpec{ IngressClassName: "apisix", Hosts: hosts, Secret: v2.ApisixSecret{ Name: req.Name, Namespace: req.Namespace, }, } apisixTls, err := apisixTlses.Get(ctx, req.Name, metav1.GetOptions{}) switch { case k8serror.IsNotFound(err): _, err = apisixTlses.Create(ctx, &v2.ApisixTls{ TypeMeta: metav1.TypeMeta{ Kind: apiutil.ApisixTLS, APIVersion: apiutil.APIVersion, }, ObjectMeta: metav1.ObjectMeta{ Name: req.Name, Namespace: req.Namespace, }, Spec: desiredSpec, }, metav1.CreateOptions{}) if err != nil { logrus.Errorf("create gateway certificate apisix tls failure: %v", err) return err } case err != nil: logrus.Errorf("get gateway certificate apisix tls failure: %v", err) return err default: if apisixTls.Spec == nil { apisixTls.Spec = desiredSpec } else { apisixTls.Spec.IngressClassName = desiredSpec.IngressClassName apisixTls.Spec.Hosts = desiredSpec.Hosts apisixTls.Spec.Secret = desiredSpec.Secret } if _, err = apisixTlses.Update(ctx, apisixTls, metav1.UpdateOptions{}); err != nil { logrus.Errorf("update gateway certificate apisix tls failure: %v", err) return err } } return nil } // DeleteGatewayCertificate delete gateway certificate func (g *GatewayAction) DeleteGatewayCertificate(name, namespace string) error { tlses := g.apisixClient.ApisixV2().ApisixTlses(namespace) list, err := tlses.List(context.Background(), metav1.ListOptions{}) if err != nil { logrus.Errorf("list gateway certificate apisix tls resources failure: %v", err) return err } for i := range list.Items { tls := &list.Items[i] if tls.Name == name || tls.Labels[gatewayMTLSManagedLabel] != "true" || tls.Annotations[gatewayMTLSParentAnnotation] != name { continue } if err := tlses.Delete(context.Background(), tls.Name, metav1.DeleteOptions{}); err != nil && !k8serror.IsNotFound(err) { logrus.Errorf("delete managed mTLS apisix tls failure: %v", err) return err } } err = tlses.Delete(context.Background(), name, metav1.DeleteOptions{}) if err != nil && !k8serror.IsNotFound(err) { logrus.Errorf("delete gateway certificate apisix tls failure: %v", err) return err } err = g.kubeClient.CoreV1().Secrets(namespace).Delete(context.Background(), name, metav1.DeleteOptions{}) if err != nil && !k8serror.IsNotFound(err) { logrus.Errorf("delete gateway certificate secret failure: %v", err) return err } return nil } func handleGatewayRules(req *apimodel.GatewayHTTPRouteStruct) []v1.HTTPRouteRule { var rules []v1.HTTPRouteRule for _, rule := range req.Rules { var ( backendRefs []v1.HTTPBackendRef matches []v1.HTTPRouteMatch filters []v1.HTTPRouteFilter ) if rule.MatchesRules != nil { for _, match := range rule.MatchesRules { var httpRouteMatch v1.HTTPRouteMatch if path := match.Path; path != nil { pathType := v1.PathMatchType(path.Type) value := path.Value httpRouteMatch.Path = &v1.HTTPPathMatch{ Type: &pathType, Value: &value, } } if headers := match.Headers; headers != nil { for _, header := range headers { headerType := v1.HeaderMatchType(header.Type) httpRouteMatch.Headers = append(httpRouteMatch.Headers, v1.HTTPHeaderMatch{ Name: v1.HTTPHeaderName(header.Name), Type: &headerType, Value: header.Value, }) } } matches = append(matches, httpRouteMatch) } } if rule.BackendRefsRules != nil { for _, backendRef := range rule.BackendRefsRules { var group v1.Group if backendRef.Kind == apimodel.HTTPRoute { group = v1.GroupName } kind := v1.Kind(backendRef.Kind) namespace := v1.Namespace(backendRef.Namespace) var port *v1.PortNumber if backendRef.Port != 0 { p := v1.PortNumber(backendRef.Port) port = &p } weight := int32(backendRef.Weight) backendRefs = append(backendRefs, v1.HTTPBackendRef{ BackendRef: v1.BackendRef{ BackendObjectReference: v1.BackendObjectReference{ Group: &group, Kind: &kind, Name: v1.ObjectName(backendRef.Name), Namespace: &namespace, Port: port, }, Weight: &weight, }, }) } } if rule.FiltersRules != nil { for _, filter := range rule.FiltersRules { var httpRoutefilter v1.HTTPRouteFilter if filter.RequestHeaderModifier != nil { var setHTTPHeader []v1.HTTPHeader var addHTTPHeader []v1.HTTPHeader if filter.RequestHeaderModifier.Set != nil { for _, set := range filter.RequestHeaderModifier.Set { setHTTPHeader = append(setHTTPHeader, v1.HTTPHeader{ Name: v1.HTTPHeaderName(set.Name), Value: set.Value, }) } } if filter.RequestHeaderModifier.Add != nil { for _, add := range filter.RequestHeaderModifier.Add { addHTTPHeader = append(addHTTPHeader, v1.HTTPHeader{ Name: v1.HTTPHeaderName(add.Name), Value: add.Value, }) } } httpRoutefilter.RequestHeaderModifier = &v1.HTTPHeaderFilter{ Set: setHTTPHeader, Add: addHTTPHeader, Remove: filter.RequestHeaderModifier.Remove, } } if filter.RequestRedirect != nil { scheme := filter.RequestRedirect.Scheme hostname := v1.PreciseHostname(filter.RequestRedirect.Hostname) var port *v1.PortNumber var sc *int if v1.PortNumber(filter.RequestRedirect.Port) != 0 { p := v1.PortNumber(filter.RequestRedirect.Port) port = &p } if filter.RequestRedirect.StatusCode != 0 { s := filter.RequestRedirect.StatusCode sc = &s } httpRoutefilter.RequestRedirect = &v1.HTTPRequestRedirectFilter{ Scheme: &scheme, Hostname: &hostname, Port: port, StatusCode: sc, } } httpRoutefilter.Type = v1.HTTPRouteFilterType(filter.Type) filters = append(filters, httpRoutefilter) } } rule := v1.HTTPRouteRule{ Matches: matches, BackendRefs: backendRefs, Filters: filters, } rules = append(rules, rule) } return rules } // AddGatewayHTTPRoute create gateway http route func (g *GatewayAction) AddGatewayHTTPRoute(req *apimodel.GatewayHTTPRouteStruct) (*model.K8sResource, error) { gatewayNamespace := v1.Namespace(req.GatewayNamespace) var hosts []v1.Hostname for _, host := range req.Hosts { hosts = append(hosts, v1.Hostname(host)) } rules := handleGatewayRules(req) labels := make(map[string]string) labels["app_id"] = req.AppID var sectionName *v1.SectionName if req.SectionName != "" { sn := v1.SectionName(req.SectionName) sectionName = &sn } httpRoute, err := g.gatewayClient.HTTPRoutes(req.Namespace).Create(context.Background(), &v1beta1.HTTPRoute{ TypeMeta: metav1.TypeMeta{ Kind: apimodel.HTTPRoute, APIVersion: controller.APIVersionHTTPRoute, }, ObjectMeta: metav1.ObjectMeta{ Name: req.Name, Namespace: req.Namespace, Labels: labels, }, Spec: v1.HTTPRouteSpec{ CommonRouteSpec: v1.CommonRouteSpec{ ParentRefs: []v1.ParentReference{{ Name: v1.ObjectName(req.GatewayName), Namespace: &gatewayNamespace, SectionName: sectionName, }}, }, Hostnames: hosts, Rules: rules, }, }, metav1.CreateOptions{}) if err != nil { logrus.Errorf("create gateway http route %v failure: %v", req.Name, err) return nil, err } httpRoute.Kind = apimodel.HTTPRoute httpRoute.APIVersion = controller.APIVersionHTTPRoute httpRouteYaml, err := ObjectToJSONORYaml("yaml", &httpRoute) if err != nil { logrus.Errorf("create gateway http route object to yaml failure: %v", err) return nil, err } k8sresource := []*model.K8sResource{{ AppID: req.AppID, Name: req.Name, Kind: apimodel.HTTPRoute, Content: httpRouteYaml, ErrorOverview: "创建成功", State: apimodel.CreateSuccess, }} err = db.GetManager().K8sResourceDao().CreateK8sResource(k8sresource) if err != nil { logrus.Errorf("database operation gateway http route create k8s resource failure: %v", err) return nil, err } return k8sresource[0], nil } // GetGatewayHTTPRoute get gateway http route func (g *GatewayAction) GetGatewayHTTPRoute(name, namespace string) (*apimodel.GatewayHTTPRouteStruct, error) { var req apimodel.GatewayHTTPRouteStruct route, err := g.gatewayClient.HTTPRoutes(namespace).Get(context.Background(), name, metav1.GetOptions{}) if err != nil { logrus.Errorf("get gateway route failure: %v", err) return nil, err } gatewayName, gatewayNamespace, sectionName := gatewayParentRefValues(route.Spec.ParentRefs, namespace) var hosts []string if route.Spec.Hostnames != nil { for _, hostname := range route.Spec.Hostnames { hosts = append(hosts, string(hostname)) } } var rules []*apimodel.Rules for _, rule := range route.Spec.Rules { var matchesRules []*apimodel.MatchesRule var backendRefsRules []*apimodel.BackendRefsRule var filtersRules []*apimodel.FiltersRule if rule.Matches != nil { for _, match := range rule.Matches { var path apimodel.MatchesRulePath var headers []*apimodel.MatchesRuleHeader if match.Headers != nil { for _, header := range match.Headers { var ty string if header.Type != nil { ty = string(*header.Type) } headers = append(headers, &apimodel.MatchesRuleHeader{ Name: string(header.Name), Type: ty, Value: header.Value, }) } } if match.Path != nil { var ty, value string if match.Path.Type != nil { ty = string(*match.Path.Type) } if match.Path.Value != nil { value = *match.Path.Value } path.Type = ty path.Value = value } matchesRules = append(matchesRules, &apimodel.MatchesRule{ Path: &path, Headers: headers, }) } } if rule.Filters != nil { for _, filter := range rule.Filters { var filterRule apimodel.FiltersRule filterRule.Type = string(filter.Type) if filter.RequestHeaderModifier != nil { var setHTTPHeader []*apimodel.HTTPHeader var addHTTPHeader []*apimodel.HTTPHeader var remove []string if filter.RequestHeaderModifier.Add != nil { for _, add := range filter.RequestHeaderModifier.Add { addHTTPHeader = append(addHTTPHeader, &apimodel.HTTPHeader{ Name: string(add.Name), Value: add.Value, }) } } if filter.RequestHeaderModifier.Set != nil { for _, set := range filter.RequestHeaderModifier.Set { setHTTPHeader = append(setHTTPHeader, &apimodel.HTTPHeader{ Name: string(set.Name), Value: set.Value, }) } } if filter.RequestHeaderModifier.Remove != nil { for _, re := range filter.RequestHeaderModifier.Remove { remove = append(remove, re) } } filterRule.RequestHeaderModifier = &apimodel.HTTPHeaderFilter{ Set: setHTTPHeader, Add: addHTTPHeader, Remove: remove, } } if filter.RequestRedirect != nil { var hostname, scheme string var statusCode, port int if filter.RequestRedirect.Hostname != nil { hostname = string(*filter.RequestRedirect.Hostname) } if filter.RequestRedirect.Scheme != nil { scheme = *filter.RequestRedirect.Scheme } if filter.RequestRedirect.StatusCode != nil { statusCode = *filter.RequestRedirect.StatusCode } if filter.RequestRedirect.Port != nil { port = int(*filter.RequestRedirect.Port) } filterRule.RequestRedirect = &apimodel.HTTPRequestRedirectFilter{ Scheme: scheme, Hostname: hostname, Port: port, StatusCode: statusCode, } } filtersRules = append(filtersRules, &filterRule) } } if rule.BackendRefs != nil { for _, backendRef := range rule.BackendRefs { weight := 100 kind := apimodel.Service if backendRef.Weight != nil { weight = int(*backendRef.Weight) } if backendRef.Kind != nil { kind = string(*backendRef.Kind) } namespace := namespace if backendRef.Namespace != nil { namespace = string(*backendRef.Namespace) } var port int if backendRef.Port != nil { port = int(*backendRef.Port) } backendRefsRules = append(backendRefsRules, &apimodel.BackendRefsRule{ Name: string(backendRef.Name), Weight: weight, Kind: kind, Namespace: namespace, Port: port, }) } } rules = append(rules, &apimodel.Rules{ MatchesRules: matchesRules, BackendRefsRules: backendRefsRules, FiltersRules: filtersRules, }) } var id string if route.Labels != nil { id = route.Labels["app_id"] } req.Hosts = hosts req.AppID = id req.GatewayName = gatewayName req.GatewayNamespace = gatewayNamespace req.Name = name req.SectionName = sectionName req.Namespace = namespace req.Rules = rules return &req, nil } func gatewayParentRefValues(parentRefs []v1.ParentReference, defaultNamespace string) (gatewayName, gatewayNamespace, sectionName string) { gatewayNamespace = defaultNamespace if len(parentRefs) == 0 { return gatewayName, gatewayNamespace, sectionName } parentRef := parentRefs[0] gatewayName = string(parentRef.Name) if parentRef.Namespace != nil { gatewayNamespace = string(*parentRef.Namespace) } if parentRef.SectionName != nil { sectionName = string(*parentRef.SectionName) } return gatewayName, gatewayNamespace, sectionName } // UpdateGatewayHTTPRoute update gateway http route func (g *GatewayAction) UpdateGatewayHTTPRoute(req *apimodel.GatewayHTTPRouteStruct) (*model.K8sResource, error) { rules := handleGatewayRules(req) gatewayNamespace := v1.Namespace(req.GatewayNamespace) var hosts []v1.Hostname for _, host := range req.Hosts { hosts = append(hosts, v1.Hostname(host)) } httpRoute, err := g.gatewayClient.HTTPRoutes(req.Namespace).Get(context.Background(), req.Name, metav1.GetOptions{}) if err != nil { logrus.Errorf("update gateway http route get failure: %v", err) return nil, err } var sectionName *v1.SectionName if req.SectionName != "" { sn := v1.SectionName(req.SectionName) sectionName = &sn } httpRoute.Spec.Hostnames = hosts httpRoute.Spec.ParentRefs = []v1.ParentReference{{ Name: v1.ObjectName(req.GatewayName), Namespace: &gatewayNamespace, SectionName: sectionName, }} httpRoute.Spec.Rules = rules newHTTPRoute, err := g.gatewayClient.HTTPRoutes(req.Namespace).Update(context.Background(), httpRoute, metav1.UpdateOptions{}) if err != nil { logrus.Errorf("update gateway http route update failure: %v", err) return nil, err } newHTTPRoute.Kind = apimodel.HTTPRoute newHTTPRoute.APIVersion = controller.APIVersionHTTPRoute httpRouteYaml, err := ObjectToJSONORYaml("yaml", &newHTTPRoute) if err != nil { logrus.Errorf("update gateway http route object to yaml failure: %v", err) return nil, err } res, err := db.GetManager().K8sResourceDao().GetK8sResourceByName(req.AppID, req.Name, apimodel.HTTPRoute) res.ErrorOverview = "更新成功" res.Content = httpRouteYaml res.State = apimodel.UpdateSuccess err = db.GetManager().K8sResourceDao().UpdateModel(&res) if err != nil { logrus.Errorf("database operation gateway http route update k8s resource failure: %v", err) return nil, err } return &res, nil } func (g *GatewayAction) createGatewayHTTPRouteDeleteEvents(route *apimodel.GatewayHTTPRouteStruct, operator string) ([]*model.ServiceEvent, error) { if route == nil { return nil, nil } serviceNames := gatewayHTTPRouteBackendServiceNames(route) if len(serviceNames) == 0 { return nil, nil } ports, err := db.GetManager().TenantServicesPortDao().ListByK8sServiceNames(serviceNames) if err != nil { return nil, err } if len(ports) == 0 { return nil, nil } reqBody, err := json.Marshal(map[string]interface{}{ "name": route.Name, "hosts": route.Hosts, "backend_service_names": serviceNames, }) if err != nil { return nil, err } var events []*model.ServiceEvent seenTargets := make(map[string]struct{}) for _, port := range ports { if port == nil || port.ServiceID == "" || port.TenantID == "" { continue } if _, ok := seenTargets[port.ServiceID]; ok { continue } seenTargets[port.ServiceID] = struct{}{} event, err := apiutil.CreateEvent(model.TargetTypeService, "delete-gateway-http-route", port.ServiceID, port.TenantID, string(reqBody), operator, "", "", model.SYNEVENTTYPE) if err != nil { return events, err } events = append(events, event) } return events, nil } func gatewayHTTPRouteBackendServiceNames(route *apimodel.GatewayHTTPRouteStruct) []string { seen := make(map[string]struct{}) var serviceNames []string for _, rule := range route.Rules { if rule == nil { continue } for _, backendRef := range rule.BackendRefsRules { if backendRef == nil || backendRef.Name == "" { continue } if backendRef.Kind != "" && backendRef.Kind != apimodel.Service { continue } if _, ok := seen[backendRef.Name]; ok { continue } seen[backendRef.Name] = struct{}{} serviceNames = append(serviceNames, backendRef.Name) } } return serviceNames } func updateGatewayHTTPRouteDeleteEvents(events []*model.ServiceEvent, statusCode int) { for _, event := range events { if event != nil { apiutil.UpdateEvent(event.EventID, statusCode) } } } // DeleteGatewayHTTPRoute delete gateway http route func (g *GatewayAction) DeleteGatewayHTTPRoute(name, namespace, appID, operator string) error { route, err := g.GetGatewayHTTPRoute(name, namespace) if err != nil { logrus.Errorf("get gateway http route before delete failure: %v", err) return err } events, err := g.createGatewayHTTPRouteDeleteEvents(route, operator) if err != nil { logrus.Errorf("create gateway http route delete events failure: %v", err) return err } err = g.gatewayClient.HTTPRoutes(namespace).Delete(context.Background(), name, metav1.DeleteOptions{}) if err != nil { updateGatewayHTTPRouteDeleteEvents(events, 500) logrus.Errorf("delete gateway http route failure: %v", err) return err } err = db.GetManager().K8sResourceDao().DeleteK8sResource(appID, name, apimodel.HTTPRoute) if err != nil { updateGatewayHTTPRouteDeleteEvents(events, 500) logrus.Errorf("database operation gateway http route delete k8s resource failure: %v", err) return err } updateGatewayHTTPRouteDeleteEvents(events, 200) return nil } // AddHTTPRule adds http rule to db if it doesn't exists. func (g *GatewayAction) AddHTTPRule(req *apimodel.AddHTTPRuleStruct) error { return db.GetManager().DB().Transaction(func(tx *gorm.DB) error { if err := g.CreateHTTPRule(tx, req); err != nil { return err } // Effective immediately err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": req.ServiceID, "action": "add-http-rule", "limit": map[string]string{"domain": req.Domain}, }) if err != nil { return fmt.Errorf("send http rule task: %v", err) } return nil }) } // CreateHTTPRule Create http rules through transactions func (g *GatewayAction) CreateHTTPRule(tx *gorm.DB, req *apimodel.AddHTTPRuleStruct) error { httpRule := &model.HTTPRule{ UUID: req.HTTPRuleID, ServiceID: req.ServiceID, ContainerPort: req.ContainerPort, Domain: req.Domain, Path: func() string { if !strings.HasPrefix(req.Path, "/") { return "/" + req.Path } return req.Path }(), Header: req.Header, Cookie: req.Cookie, Weight: req.Weight, IP: req.IP, CertificateID: req.CertificateID, PathRewrite: req.PathRewrite, } if err := db.GetManager().HTTPRuleDaoTransactions(tx).AddModel(httpRule); err != nil { return fmt.Errorf("create http rule: %v", err) } if len(req.Rewrites) > 0 { for _, rewrite := range req.Rewrites { r := &model.HTTPRuleRewrite{ UUID: util.NewUUID(), HTTPRuleID: httpRule.UUID, Regex: rewrite.Regex, Replacement: rewrite.Replacement, Flag: rewrite.Flag, } if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).AddModel(r); err != nil { return fmt.Errorf("create http rule rewrite: %v", err) } } } if strings.Replace(req.CertificateID, " ", "", -1) != "" { cert := &model.Certificate{ UUID: req.CertificateID, CertificateName: fmt.Sprintf("cert-%s", util.NewUUID()[0:8]), Certificate: req.Certificate, PrivateKey: req.PrivateKey, } if err := db.GetManager().CertificateDaoTransactions(tx).AddOrUpdate(cert); err != nil { return fmt.Errorf("create or update http rule: %v", err) } } for _, ruleExtension := range req.RuleExtensions { re := &model.RuleExtension{ UUID: util.NewUUID(), RuleID: httpRule.UUID, Key: ruleExtension.Key, Value: ruleExtension.Value, } if err := db.GetManager().RuleExtensionDaoTransactions(tx).AddModel(re); err != nil { return fmt.Errorf("create rule extensions: %v", err) } } return nil } // UpdateHTTPRule updates http rule func (g *GatewayAction) UpdateHTTPRule(req *apimodel.UpdateHTTPRuleStruct) error { tx := db.GetManager().Begin() defer func() { if r := recover(); r != nil { logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r) tx.Rollback() } }() rule, err := g.dbmanager.HTTPRuleDaoTransactions(tx).GetHTTPRuleByID(req.HTTPRuleID) if err != nil { tx.Rollback() return err } if rule == nil || rule.UUID == "" { // rule won't be nil tx.Rollback() return fmt.Errorf("HTTPRule dosen't exist based on uuid(%s)", req.HTTPRuleID) } // delete old http rule rewrites if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).DeleteByHTTPRuleID(rule.UUID); err != nil { tx.Rollback() return err } if len(req.Rewrites) > 0 { // add new http rule rewrites for _, rewrite := range req.Rewrites { r := &model.HTTPRuleRewrite{ UUID: util.NewUUID(), HTTPRuleID: rule.UUID, Regex: rewrite.Regex, Replacement: rewrite.Replacement, Flag: rewrite.Flag, } if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).AddModel(r); err != nil { tx.Rollback() return err } } } if strings.Replace(req.CertificateID, " ", "", -1) != "" { // add new certificate cert := &model.Certificate{ UUID: req.CertificateID, Certificate: req.Certificate, PrivateKey: req.PrivateKey, } if err := g.dbmanager.CertificateDaoTransactions(tx).AddOrUpdate(cert); err != nil { tx.Rollback() return err } rule.CertificateID = req.CertificateID } else { rule.CertificateID = "" } if len(req.RuleExtensions) > 0 { // delete old RuleExtensions if err := g.dbmanager.RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(rule.UUID); err != nil { tx.Rollback() return err } // add new rule extensions for _, ruleExtension := range req.RuleExtensions { re := &model.RuleExtension{ UUID: util.NewUUID(), RuleID: rule.UUID, Key: ruleExtension.Key, Value: ruleExtension.Value, } if err := db.GetManager().RuleExtensionDaoTransactions(tx).AddModel(re); err != nil { tx.Rollback() return err } } } // update http rule if req.ServiceID != "" { rule.ServiceID = req.ServiceID } if req.ContainerPort != 0 { rule.ContainerPort = req.ContainerPort } if req.Domain != "" { rule.Domain = req.Domain } rule.Path = func() string { if !strings.HasPrefix(req.Path, "/") { return "/" + req.Path } return req.Path }() rule.Header = req.Header rule.Cookie = req.Cookie rule.Weight = req.Weight rule.PathRewrite = req.PathRewrite if req.IP != "" { rule.IP = req.IP } if err := db.GetManager().HTTPRuleDaoTransactions(tx).UpdateModel(rule); err != nil { tx.Rollback() return err } // end transaction if err := tx.Commit().Error; err != nil { tx.Rollback() return err } if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": rule.ServiceID, "action": "update-http-rule", "limit": map[string]string{"domain": req.Domain}, }); err != nil { logrus.Errorf("send runtime message about gateway failure %s", err.Error()) } return nil } // DeleteHTTPRule deletes http rule, including certificate and rule extensions func (g *GatewayAction) DeleteHTTPRule(req *apimodel.DeleteHTTPRuleStruct) error { // begin transaction tx := db.GetManager().Begin() defer func() { if r := recover(); r != nil { logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r) tx.Rollback() } }() // delete http rule httpRule, err := g.dbmanager.HTTPRuleDaoTransactions(tx).GetHTTPRuleByID(req.HTTPRuleID) if err != nil { tx.Rollback() return err } svcID := httpRule.ServiceID if err := g.dbmanager.HTTPRuleDaoTransactions(tx).DeleteHTTPRuleByID(httpRule.UUID); err != nil { tx.Rollback() return err } // delete http rule rewrites if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).DeleteByHTTPRuleID(httpRule.UUID); err != nil { tx.Rollback() return err } // delete rule extension if err := g.dbmanager.RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(httpRule.UUID); err != nil { tx.Rollback() return err } // end transaction if err := tx.Commit().Error; err != nil { return err } if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": svcID, "action": "delete-http-rule", "limit": map[string]string{"domain": httpRule.Domain}, }); err != nil { logrus.Errorf("send runtime message about gateway failure %s", err.Error()) } return nil } // DeleteHTTPRuleByServiceIDWithTransaction deletes http rule, including certificate and rule extensions func (g *GatewayAction) DeleteHTTPRuleByServiceIDWithTransaction(sid string, tx *gorm.DB) error { // delete http rule rules, err := g.dbmanager.HTTPRuleDaoTransactions(tx).ListByServiceID(sid) if err != nil { return err } for _, rule := range rules { if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).DeleteByHTTPRuleID(rule.UUID); err != nil { return err } if err := g.dbmanager.RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(rule.UUID); err != nil { return err } if err := g.dbmanager.GwRuleConfigDaoTransactions(tx).DeleteByRuleID(rule.UUID); err != nil { return err } if err := g.dbmanager.HTTPRuleDaoTransactions(tx).DeleteHTTPRuleByID(rule.UUID); err != nil { return err } } return nil } // AddCertificate adds certificate to db if it doesn't exists func (g *GatewayAction) AddCertificate(req *apimodel.AddHTTPRuleStruct, tx *gorm.DB) error { cert := &model.Certificate{ UUID: req.CertificateID, CertificateName: fmt.Sprintf("cert-%s", util.NewUUID()[0:8]), Certificate: req.Certificate, PrivateKey: req.PrivateKey, } return g.dbmanager.CertificateDaoTransactions(tx).AddModel(cert) } // UpdateCertificate updates certificate for http rule func (g *GatewayAction) UpdateCertificate(req apimodel.AddHTTPRuleStruct, httpRule *model.HTTPRule, tx *gorm.DB) error { // delete old certificate cert, err := g.dbmanager.CertificateDaoTransactions(tx).GetCertificateByID(req.CertificateID) if err != nil { return err } if cert == nil { return fmt.Errorf("certificate doesn't exist based on certificateID(%s)", req.CertificateID) } cert.CertificateName = fmt.Sprintf("cert-%s", util.NewUUID()[0:8]) cert.Certificate = req.Certificate cert.PrivateKey = req.PrivateKey return g.dbmanager.CertificateDaoTransactions(tx).UpdateModel(cert) } // AddTCPRule adds tcp rule. func (g *GatewayAction) AddTCPRule(req *apimodel.AddTCPRuleStruct) error { return g.dbmanager.DB().Transaction(func(tx *gorm.DB) error { if err := g.CreateTCPRule(tx, req); err != nil { return err } err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": req.ServiceID, "action": "add-tcp-rule", "limit": map[string]string{"tcp-address": fmt.Sprintf("%s:%d", req.IP, req.Port)}, }) if err != nil { return fmt.Errorf("send tcp rule task: %v", err) } return nil }) } // CreateTCPRule Create tcp rules through transactions func (g *GatewayAction) CreateTCPRule(tx *gorm.DB, req *apimodel.AddTCPRuleStruct) error { // add tcp rule tcpRule := &model.TCPRule{ UUID: req.TCPRuleID, Protocol: req.Protocol, ServiceID: req.ServiceID, ContainerPort: req.ContainerPort, IP: req.IP, Port: req.Port, } if err := g.dbmanager.TCPRuleDaoTransactions(tx).AddModel(tcpRule); err != nil { return err } // add rule extensions for _, ruleExtension := range req.RuleExtensions { re := &model.RuleExtension{ UUID: util.NewUUID(), RuleID: tcpRule.UUID, Value: ruleExtension.Value, } if err := g.dbmanager.RuleExtensionDaoTransactions(tx).AddModel(re); err != nil { return err } } return nil } // UpdateTCPRule updates a tcp rule func (g *GatewayAction) UpdateTCPRule(req *apimodel.UpdateTCPRuleStruct, minPort int) error { // begin transaction tx := db.GetManager().Begin() defer func() { if r := recover(); r != nil { logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r) tx.Rollback() } }() // get old tcp rule tcpRule, err := g.dbmanager.TCPRuleDaoTransactions(tx).GetTCPRuleByID(req.TCPRuleID) if err != nil { tx.Rollback() return err } if len(req.RuleExtensions) > 0 { // delete old rule extensions if err := g.dbmanager.RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(tcpRule.UUID); err != nil { logrus.Debugf("TCP rule id: %s;error delete rule extension: %v", tcpRule.UUID, err) tx.Rollback() return err } // add new rule extensions for _, ruleExtension := range req.RuleExtensions { re := &model.RuleExtension{ UUID: util.NewUUID(), RuleID: tcpRule.UUID, Value: ruleExtension.Value, } if err := g.dbmanager.RuleExtensionDaoTransactions(tx).AddModel(re); err != nil { tx.Rollback() logrus.Debugf("TCP rule id: %s;error add rule extension: %v", tcpRule.UUID, err) return err } } } // update tcp rule if req.ContainerPort != 0 { tcpRule.ContainerPort = req.ContainerPort } if req.IP != "" { tcpRule.IP = req.IP } tcpRule.Port = req.Port if req.ServiceID != "" { tcpRule.ServiceID = req.ServiceID } if err := g.dbmanager.TCPRuleDaoTransactions(tx).UpdateModel(tcpRule); err != nil { logrus.Debugf("TCP rule id: %s;error updating tcp rule: %v", tcpRule.UUID, err) tx.Rollback() return err } // end transaction if err := tx.Commit().Error; err != nil { tx.Rollback() logrus.Debugf("TCP rule id: %s;error end transaction %v", tcpRule.UUID, err) return err } if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": tcpRule.ServiceID, "action": "update-tcp-rule", "limit": map[string]string{"tcp-address": fmt.Sprintf("%s:%d", tcpRule.IP, tcpRule.Port)}, }); err != nil { logrus.Errorf("send runtime message about gateway failure %s", err.Error()) } return nil } // DeleteTCPRule deletes a tcp rule func (g *GatewayAction) DeleteTCPRule(req *apimodel.DeleteTCPRuleStruct) error { // begin transaction tx := db.GetManager().Begin() defer func() { if r := recover(); r != nil { logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r) tx.Rollback() } }() tcpRule, err := db.GetManager().TCPRuleDaoTransactions(tx).GetTCPRuleByID(req.TCPRuleID) if err != nil { tx.Rollback() return err } // delete rule extensions if err := db.GetManager().RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(tcpRule.UUID); err != nil { tx.Rollback() return err } // delete tcp rule if err := db.GetManager().TCPRuleDaoTransactions(tx).DeleteByID(tcpRule.UUID); err != nil { tx.Rollback() return err } // delete LBMappingPort err = db.GetManager().TenantServiceLBMappingPortDaoTransactions(tx).DELServiceLBMappingPortByServiceIDAndPort( tcpRule.ServiceID, tcpRule.Port) if err != nil { tx.Rollback() return err } // end transaction if err := tx.Commit().Error; err != nil { tx.Rollback() return err } if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": tcpRule.ServiceID, "action": "delete-tcp-rule", "limit": map[string]string{"tcp-address": fmt.Sprintf("%s:%d", tcpRule.IP, tcpRule.Port)}, }); err != nil { logrus.Errorf("send runtime message about gateway failure %s", err.Error()) } return nil } // DeleteTCPRuleByServiceIDWithTransaction deletes a tcp rule func (g *GatewayAction) DeleteTCPRuleByServiceIDWithTransaction(sid string, tx *gorm.DB) error { rules, err := db.GetManager().TCPRuleDaoTransactions(tx).GetTCPRuleByServiceID(sid) if err != nil { return err } for _, rule := range rules { // delete rule extensions if err := db.GetManager().RuleExtensionDaoTransactions(tx).DeleteRuleExtensionByRuleID(rule.UUID); err != nil { return err } // delete tcp rule if err := db.GetManager().TCPRuleDaoTransactions(tx).DeleteByID(rule.UUID); err != nil { return err } } return nil } // AddRuleExtensions adds rule extensions to db if any of they doesn't exists func (g *GatewayAction) AddRuleExtensions(ruleID string, ruleExtensions []*apimodel.RuleExtensionStruct, tx *gorm.DB) error { for _, ruleExtension := range ruleExtensions { re := &model.RuleExtension{ UUID: util.NewUUID(), RuleID: ruleID, Value: ruleExtension.Value, } err := g.dbmanager.RuleExtensionDaoTransactions(tx).AddModel(re) if err != nil { return err } } return nil } // GetAvailablePort returns a available port func (g *GatewayAction) GetAvailablePort(ip string, lock bool) (int, error) { roles, err := g.dbmanager.TCPRuleDao().GetUsedPortsByIP(ip) if err != nil { return 0, err } var ports []int for _, p := range roles { ports = append(ports, p.Port) } //resp, err := clientv3.KV(g.etcdCli).Get(context.TODO(), "/rainbond/gateway/lockports", clientv3.WithPrefix()) //if err != nil { // logrus.Info("get lock ports failed") //} //for _, etcdValue := range resp.Kvs { // port, err := strconv.Atoi(string(etcdValue.Value)) // if err != nil { // continue // } // ports = append(ports, port) //} port := selectAvailablePort(ports) if port != 0 { // if lock { // lease := clientv3.NewLease(g.etcdCli) // leaseResp, err := lease.Grant(context.Background(), 120) // if err != nil { // logrus.Info("set lease failed") // return port, nil // } // lockPortKey := fmt.Sprintf("/rainbond/gateway/lockports/%d", port) // _, err = g.etcdCli.Put(context.Background(), lockPortKey, fmt.Sprintf("%d", port), clientv3.WithLease(leaseResp.ID)) // if err != nil { // logrus.Infof("set lock port key %s failed", lockPortKey) // return port, nil // } // logrus.Infof("select gateway port %d, lock it 2 min", port) // } return port, nil } return 0, fmt.Errorf("no more lb port can be use with ip %s", ip) } func selectAvailablePort(used []int) int { maxPort, _ := strconv.Atoi(os.Getenv("MAX_LB_PORT")) minPort, _ := strconv.Atoi(os.Getenv("MIN_LB_PORT")) if minPort == 0 { minPort = 30000 } if maxPort == 0 { maxPort = 65535 } if len(used) == 0 { return minPort } sort.Ints(used) // 优先从 minPort 开始找空闲端口（捡漏） selectPort := minPort for _, p := range used { // 跳过小于最小端口的记录 if p < minPort { continue } // 如果当前端口已被使用，尝试下一个 if p == selectPort { selectPort = selectPort + 1 continue } // 找到间隙，返回空闲端口 if p > selectPort { if selectPort <= maxPort { return selectPort } break } } // 如果没有间隙，使用最大端口 + 1 if selectPort <= maxPort { return selectPort } // 无可用端口 return 0 } func reassignConflictingTCPRulePorts(existingRules, incomingRules []*model.TCPRule) error { usedPorts := make(map[int]struct{}, len(existingRules)+len(incomingRules)) for _, rule := range existingRules { if rule == nil || rule.Port <= 0 { continue } usedPorts[rule.Port] = struct{}{} } for _, rule := range incomingRules { if rule == nil { continue } if _, exists := usedPorts[rule.Port]; rule.Port <= 0 || exists { originalPort := rule.Port nextPort := selectAvailablePortFromSet(usedPorts) if nextPort == 0 { return fmt.Errorf("no more lb port can be used for service %s", rule.ServiceID) } rule.Port = nextPort if originalPort > 0 { logrus.Infof("tcp rule port %d for service %s is already used, reassign to %d", originalPort, rule.ServiceID, nextPort) } } usedPorts[rule.Port] = struct{}{} } return nil } func selectAvailablePortFromSet(usedPorts map[int]struct{}) int { used := make([]int, 0, len(usedPorts)) for port := range usedPorts { used = append(used, port) } return selectAvailablePort(used) } // TCPIPPortExists returns if the port exists func (g *GatewayAction) TCPIPPortExists(host string, port int) bool { roles, _ := db.GetManager().TCPRuleDao().GetUsedPortsByIP(host) for _, role := range roles { if role.Port == port { return true } } return false } // SendTaskDeprecated sends apply rules task func (g *GatewayAction) SendTaskDeprecated(in map[string]interface{}) error { sid := in["service_id"].(string) service, err := db.GetManager().TenantServiceDao().GetServiceByID(sid) if err != nil { return fmt.Errorf("unexpected error occurred while getting Service by ServiceID(%s): %v", sid, err) } body := make(map[string]interface{}) body["deploy_version"] = service.DeployVersion for k, v := range in { body[k] = v } err = g.mqclient.SendBuilderTopic(client.TaskStruct{ Topic: client.WorkerTopic, TaskType: "apply_rule", TaskBody: body, }) if err != nil { return fmt.Errorf("unexpected error occurred while sending task: %v", err) } return nil } // SendTask sends apply rules task func (g *GatewayAction) SendTask(task *ComponentIngressTask) error { err := g.mqclient.SendBuilderTopic(client.TaskStruct{ Topic: client.WorkerTopic, TaskType: "apply_rule", TaskBody: task, }) if err != nil { return errors.WithMessage(err, "send gateway task") } return nil } // RuleConfig - func (g *GatewayAction) RuleConfig(req *apimodel.RuleConfigReq) error { var configs []*model.GwRuleConfig // TODO: use reflect to read the field of req, huangrh configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-connect-timeout", Value: strconv.Itoa(req.Body.ProxyConnectTimeout), }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-send-timeout", Value: strconv.Itoa(req.Body.ProxySendTimeout), }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-read-timeout", Value: strconv.Itoa(req.Body.ProxyReadTimeout), }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-body-size", Value: strconv.Itoa(req.Body.ProxyBodySize), }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-buffer-size", Value: strconv.Itoa(req.Body.ProxyBufferSize) + "k", }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-buffer-numbers", Value: strconv.Itoa(req.Body.ProxyBufferNumbers), }) configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: "proxy-buffering", Value: req.Body.ProxyBuffering, }) setheaders := make(map[string]string) for _, item := range req.Body.SetHeaders { if strings.TrimSpace(item.Key) == "" { continue } if strings.TrimSpace(item.Value) == "" { item.Value = "empty" } // filter same key setheaders["set-header-"+item.Key] = item.Value } for k, v := range setheaders { configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: k, Value: v, }) } // response headers responseHeaders := make(map[string]string) for _, item := range req.Body.ResponseHeaders { if strings.TrimSpace(item.Key) == "" { continue } if strings.TrimSpace(item.Value) == "" { item.Value = "empty" } // filter same key responseHeaders["resp-header-"+item.Key] = item.Value } for k, v := range responseHeaders { configs = append(configs, &model.GwRuleConfig{ RuleID: req.RuleID, Key: k, Value: v, }) } rule, err := g.dbmanager.HTTPRuleDao().GetHTTPRuleByID(req.RuleID) if err != nil { return err } tx := db.GetManager().Begin() defer func() { if r := recover(); r != nil { logrus.Errorf("Unexpected panic occurred, rollback transaction: %v", r) tx.Rollback() } }() if err := g.dbmanager.GwRuleConfigDaoTransactions(tx).DeleteByRuleID(req.RuleID); err != nil { tx.Rollback() return err } for _, cfg := range configs { if err := g.dbmanager.GwRuleConfigDaoTransactions(tx).AddModel(cfg); err != nil { tx.Rollback() return err } } if err := tx.Commit().Error; err != nil { tx.Rollback() return err } if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": req.ServiceID, "action": "update-rule-config", "event_id": req.EventID, "limit": map[string]string{"domain": rule.Domain}, }); err != nil { logrus.Errorf("send runtime message about gateway failure %s", err.Error()) } return nil } // UpdCertificate - func (g *GatewayAction) UpdCertificate(req *apimodel.UpdCertificateReq) error { cert, err := db.GetManager().CertificateDao().GetCertificateByID(req.CertificateID) if err != nil { msg := "retrieve certificate: %v" return fmt.Errorf(msg, err) } if cert == nil { // cert do not exists in region db, create it cert = &model.Certificate{ UUID: req.CertificateID, CertificateName: req.CertificateName, Certificate: req.Certificate, PrivateKey: req.PrivateKey, } if err := db.GetManager().CertificateDao().AddModel(cert); err != nil { msg := "update cert error :%s" return fmt.Errorf(msg, err.Error()) } return nil } cert.CertificateName = req.CertificateName cert.Certificate = req.Certificate cert.PrivateKey = req.PrivateKey if err := db.GetManager().CertificateDao().UpdateModel(cert); err != nil { msg := "update certificate: %v" return fmt.Errorf(msg, err) } // list related http rules rules, err := g.ListHTTPRulesByCertID(req.CertificateID) if err != nil { msg := "certificate id: %s; list http rules: %v" return fmt.Errorf(msg, req.CertificateID, err) } for _, rule := range rules { eventID := util.NewUUID() if err := g.SendTaskDeprecated(map[string]interface{}{ "service_id": rule.ServiceID, "action": "update-rule-config", "event_id": eventID, "limit": map[string]string{"domain": rule.Domain}, }); err != nil { logrus.Warningf("send runtime message about gateway failure %v", err) } } return nil } // ListHTTPRulesByCertID - func (g *GatewayAction) ListHTTPRulesByCertID(certID string) ([]*model.HTTPRule, error) { return db.GetManager().HTTPRuleDao().ListByCertID(certID) } // IPAndAvailablePort ip and advice available port type IPAndAvailablePort struct { IP string `json:"ip"` AvailablePort int `json:"available_port"` } // GetGatewayIPs get all gateway node ips func (g *GatewayAction) GetGatewayIPs() []IPAndAvailablePort { defaultAvailablePort, _ := g.GetAvailablePort("0.0.0.0", false) defaultIps := []IPAndAvailablePort{{ IP: "0.0.0.0", AvailablePort: defaultAvailablePort, }} res, err := db.GetManager().KeyValueDao().WithPrefix("/rainbond/gateway/ips") if err != nil || len(res) == 0 { return defaultIps } var gatewayIps = make([]string, 0) for _, v := range res { gatewayIps = append(gatewayIps, v.V) } sort.Strings(gatewayIps) for _, v := range gatewayIps { availablePort, _ := g.GetAvailablePort(v, false) defaultIps = append(defaultIps, IPAndAvailablePort{ IP: v, AvailablePort: availablePort, }) } return defaultIps } // DeleteIngressRulesByComponentPort deletes ingress rules, including http rules and tcp rules, based on the given componentID and port. func (g *GatewayAction) DeleteIngressRulesByComponentPort(tx *gorm.DB, componentID string, port int) error { httpRuleIDs, err := g.listHTTPRuleIDs(componentID, port) if err != nil { return err } // delete rule configs if err := db.GetManager().GwRuleConfigDaoTransactions(tx).DeleteByRuleIDs(httpRuleIDs); err != nil { return err } // delete rule extentions if err := db.GetManager().RuleExtensionDaoTransactions(tx).DeleteByRuleIDs(httpRuleIDs); err != nil { return err } // delete http rules if err := db.GetManager().HTTPRuleDaoTransactions(tx).DeleteByComponentPort(componentID, port); err != nil { if !errors.Is(err, bcode.ErrIngressHTTPRuleNotFound) { return err } } // delete tcp rules if err := db.GetManager().TCPRuleDaoTransactions(tx).DeleteByComponentPort(componentID, port); err != nil { if !errors.Is(err, bcode.ErrIngressTCPRuleNotFound) { return err } } return nil } func (g *GatewayAction) listHTTPRuleIDs(componentID string, port int) ([]string, error) { httpRules, err := db.GetManager().HTTPRuleDao().ListByComponentPort(componentID, port) if err != nil { return nil, err } var ruleIDs []string for _, rule := range httpRules { ruleIDs = append(ruleIDs, rule.UUID) } return ruleIDs, nil } // SyncHTTPRules - func (g *GatewayAction) SyncHTTPRules(tx *gorm.DB, components []*apimodel.Component) error { var ( componentIDs []string httpRules []*model.HTTPRule ruleExtensions []*model.RuleExtension httpRuleRewrites []*model.HTTPRuleRewrite ) for _, component := range components { if len(component.HTTPRules) == 0 { continue } componentIDs = append(componentIDs, component.ComponentBase.ComponentID) for _, httpRule := range component.HTTPRules { httpRules = append(httpRules, httpRule.DbModel(component.ComponentBase.ComponentID)) for _, rewrite := range httpRule.Rewrites { httpRuleRewrites = append(httpRuleRewrites, &model.HTTPRuleRewrite{ UUID: util.NewUUID(), HTTPRuleID: httpRule.HTTPRuleID, Regex: rewrite.Regex, Replacement: rewrite.Replacement, Flag: rewrite.Flag, }) } for _, ext := range httpRule.RuleExtensions { ruleExtensions = append(ruleExtensions, &model.RuleExtension{ UUID: util.NewUUID(), RuleID: httpRule.HTTPRuleID, Key: ext.Key, Value: ext.Value, }) } } } if err := g.syncHTTPRuleRewrites(tx, httpRules, httpRuleRewrites); err != nil { return err } if err := g.syncRuleExtensions(tx, httpRules, ruleExtensions); err != nil { return err } if err := db.GetManager().HTTPRuleDaoTransactions(tx).DeleteByComponentIDs(componentIDs); err != nil { return err } return db.GetManager().HTTPRuleDaoTransactions(tx).CreateOrUpdateHTTPRuleInBatch(httpRules) } func (g *GatewayAction) syncHTTPRuleRewrites(tx *gorm.DB, httpRules []*model.HTTPRule, rewrites []*model.HTTPRuleRewrite) error { var ruleIDs []string for _, hr := range httpRules { ruleIDs = append(ruleIDs, hr.UUID) } if err := db.GetManager().HTTPRuleRewriteDaoTransactions(tx).DeleteByHTTPRuleIDs(ruleIDs); err != nil { return err } return db.GetManager().HTTPRuleRewriteDaoTransactions(tx).CreateOrUpdateHTTPRuleRewriteInBatch(rewrites) } func (g *GatewayAction) syncRuleExtensions(tx *gorm.DB, httpRules []*model.HTTPRule, exts []*model.RuleExtension) error { var ruleIDs []string for _, hr := range httpRules { ruleIDs = append(ruleIDs, hr.UUID) } if err := db.GetManager().RuleExtensionDaoTransactions(tx).DeleteByRuleIDs(ruleIDs); err != nil { return err } return db.GetManager().RuleExtensionDaoTransactions(tx).CreateOrUpdateRuleExtensionsInBatch(exts) } // SyncTCPRules - func (g *GatewayAction) SyncTCPRules(tx *gorm.DB, components []*apimodel.Component) error { var ( componentIDs []string tcpRules []*model.TCPRule ) for _, component := range components { if len(component.TCPRules) == 0 { continue } componentIDs = append(componentIDs, component.ComponentBase.ComponentID) for _, tcpRule := range component.TCPRules { tcpRules = append(tcpRules, tcpRule.DbModel(component.ComponentBase.ComponentID)) } } if len(tcpRules) == 0 { return nil } if err := db.GetManager().TCPRuleDaoTransactions(tx).DeleteByComponentIDs(componentIDs); err != nil { return err } existingTCPRules, err := db.GetManager().TCPRuleDaoTransactions(tx).GetUsedPortsByIP("0.0.0.0") if err != nil { return err } if err := reassignConflictingTCPRulePorts(existingTCPRules, tcpRules); err != nil { return err } return db.GetManager().TCPRuleDaoTransactions(tx).CreateOrUpdateTCPRuleInBatch(tcpRules) } // SyncRuleConfigs - func (g *GatewayAction) SyncRuleConfigs(tx *gorm.DB, components []*apimodel.Component) error { var configs []*model.GwRuleConfig var componentIDs []string for _, component := range components { componentIDs = append(componentIDs, component.ComponentBase.ComponentID) if len(component.HTTPRuleConfigs) == 0 { continue } for _, httpRuleConfig := range component.HTTPRuleConfigs { configs = append(configs, httpRuleConfig.DbModel()...) } } // http rule ids rules, err := db.GetManager().HTTPRuleDao().ListByComponentIDs(componentIDs) if err != nil { return err } var ruleIDs []string for _, rule := range rules { ruleIDs = append(ruleIDs, rule.UUID) } if err := db.GetManager().GwRuleConfigDaoTransactions(tx).DeleteByRuleIDs(ruleIDs); err != nil { return err } return db.GetManager().GwRuleConfigDaoTransactions(tx).CreateOrUpdateGwRuleConfigsInBatch(configs) }