From cde76989a0eefaef4de4c3198a29efb5becca594 Mon Sep 17 00:00:00 2001 From: Jianbo Sun Date: Wed, 13 Jul 2022 14:51:06 +0800 Subject: [PATCH] Feat: support resource topology for endpoints (#4362) * Feat: support resource topology for endpoints Signed-off-by: Jianbo Sun * Fix: add and refactor the test for endpoint Signed-off-by: Jianbo Sun --- pkg/oam/labels.go | 11 + pkg/stdlib/pkgs/query.cue | 3 + pkg/velaql/providers/query/collector.go | 54 ++- pkg/velaql/providers/query/endpoint.go | 358 ++++++++++++++++++ pkg/velaql/providers/query/endpoint_test.go | 251 +++++++++++++ pkg/velaql/providers/query/handler.go | 383 +------------------- pkg/velaql/providers/query/handler_test.go | 157 ++++---- pkg/velaql/providers/query/tree.go | 27 +- pkg/velaql/providers/query/types/type.go | 7 +- pkg/velaql/providers/query/utils.go | 6 +- references/cli/velaql_test.go | 14 +- test/e2e-apiserver-test/velaql_test.go | 2 +- 12 files changed, 806 insertions(+), 467 deletions(-) create mode 100644 pkg/velaql/providers/query/endpoint.go create mode 100644 pkg/velaql/providers/query/endpoint_test.go diff --git a/pkg/oam/labels.go b/pkg/oam/labels.go index f3d296748..d208d3fd1 100644 --- a/pkg/oam/labels.go +++ b/pkg/oam/labels.go @@ -97,8 +97,12 @@ const ( // LabelProject recorde the project the resource belong to LabelProject = "core.oam.dev/project" + // LabelResourceRules defines the configmap is representing the resource topology rules LabelResourceRules = "rules.oam.dev/resources" + // LabelResourceRuleFormat defines the resource format of the resource topology rules + LabelResourceRuleFormat = "rules.oam.dev/resource-format" + // LabelControllerName indicates the controller name LabelControllerName = "controller.oam.dev/name" ) @@ -220,3 +224,10 @@ const ( // AnnotationResourceURL records the source url of the Kubernetes object AnnotationResourceURL = "app.oam.dev/resource-url" ) + +const ( + // ResourceTopologyFormatYAML mark the format of resource topology is yaml, by default, it's yaml. + ResourceTopologyFormatYAML = "yaml" + // ResourceTopologyFormatJSON mark the format of resource topology is json. + ResourceTopologyFormatJSON = "json" +) diff --git a/pkg/stdlib/pkgs/query.cue b/pkg/stdlib/pkgs/query.cue index 2119d6340..3ad295a6d 100644 --- a/pkg/stdlib/pkgs/query.cue +++ b/pkg/stdlib/pkgs/query.cue @@ -50,6 +50,9 @@ deployVersion?: string revision?: string latest?: bool + resourceTree?: { + ... + } }] ... } diff --git a/pkg/velaql/providers/query/collector.go b/pkg/velaql/providers/query/collector.go index 6ab11e0e9..ad901b85c 100644 --- a/pkg/velaql/providers/query/collector.go +++ b/pkg/velaql/providers/query/collector.go @@ -96,7 +96,6 @@ func (c *AppCollector) ListApplicationResources(app *v1beta1.Application) ([]*ty if err != nil { return nil, err } - var managedResources []*types.AppliedResource for _, rt := range append(historyRTs, rootRT, currentRT) { if rt != nil { @@ -129,6 +128,59 @@ func (c *AppCollector) ListApplicationResources(app *v1beta1.Application) ([]*ty } } } + + // merge user defined customize rule before every request. + err = mergeCustomRules(ctx, c.k8sClient) + if err != nil { + return managedResources, err + } + + // error from leaf nodes won't block the results + for _, resource := range managedResources { + root := types.ResourceTreeNode{ + Cluster: resource.Cluster, + APIVersion: resource.APIVersion, + Kind: resource.Kind, + Namespace: resource.Namespace, + Name: resource.Name, + UID: resource.UID, + } + root.LeafNodes, err = iteratorChildResources(ctx, resource.Cluster, c.k8sClient, root, 1) + if err != nil { + // if the resource has been deleted, continue access next appliedResource don't break the whole request + if kerrors.IsNotFound(err) { + continue + } + klog.Errorf("query leaf node resource apiVersion=%s kind=%s namespace=%s name=%s failure %s, skip this resource", root.APIVersion, root.Kind, root.Namespace, root.Name, err.Error()) + continue + } + rootObject, err := fetchObjectWithResourceTreeNode(ctx, resource.Cluster, c.k8sClient, root) + if err != nil { + // if the resource has been deleted, continue access next appliedResource don't break the whole request + if kerrors.IsNotFound(err) { + continue + } + klog.Errorf("fetch object for resource apiVersion=%s kind=%s namespace=%s name=%s failure %s, skip this resource", root.APIVersion, root.Kind, root.Namespace, root.Name, err.Error()) + continue + } + rootStatus, err := checkResourceStatus(*rootObject) + if err != nil { + klog.Errorf("check status for resource apiVersion=%s kind=%s namespace=%s name=%s failure %s, skip this resource", root.APIVersion, root.Kind, root.Namespace, root.Name, err.Error()) + continue + } + root.HealthStatus = *rootStatus + addInfo, err := additionalInfo(*rootObject) + if err != nil { + klog.Errorf("check additionalInfo for resource apiVersion=%s kind=%s namespace=%s name=%s failure %s, skip this resource", root.APIVersion, root.Kind, root.Namespace, root.Name, err.Error()) + continue + } + root.AdditionalInfo = addInfo + root.CreationTimestamp = rootObject.GetCreationTimestamp().Time + if !rootObject.GetDeletionTimestamp().IsZero() { + root.DeletionTimestamp = rootObject.GetDeletionTimestamp().Time + } + resource.ResourceTree = &root + } return managedResources, nil } diff --git a/pkg/velaql/providers/query/endpoint.go b/pkg/velaql/providers/query/endpoint.go new file mode 100644 index 000000000..1ab0354e8 --- /dev/null +++ b/pkg/velaql/providers/query/endpoint.go @@ -0,0 +1,358 @@ +/* + Copyright 2022 The KubeVela Authors. + + 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 query + +import ( + "context" + "fmt" + "strconv" + "time" + + corev1 "k8s.io/api/core/v1" + networkv1beta1 "k8s.io/api/networking/v1beta1" + kerrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/klog" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + apis "github.com/oam-dev/kubevela/apis/types" + helmapi "github.com/oam-dev/kubevela/pkg/appfile/helm/flux2apis" + "github.com/oam-dev/kubevela/pkg/cue/model/value" + "github.com/oam-dev/kubevela/pkg/multicluster" + "github.com/oam-dev/kubevela/pkg/utils" + querytypes "github.com/oam-dev/kubevela/pkg/velaql/providers/query/types" + wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" + "github.com/oam-dev/kubevela/pkg/workflow/types" +) + +// GeneratorServiceEndpoints generator service endpoints is available for common component type, +// such as webservice or helm +// it can not support the cloud service component currently +func (h *provider) GeneratorServiceEndpoints(wfctx wfContext.Context, v *value.Value, act types.Action) error { + ctx := context.Background() + + val, err := v.LookupValue("app") + if err != nil { + return err + } + opt := Option{} + if err = val.UnmarshalTo(&opt); err != nil { + return err + } + app := new(v1beta1.Application) + err = findResource(ctx, h.cli, app, opt.Name, opt.Namespace, "") + if err != nil { + return fmt.Errorf("query app failure %w", err) + } + serviceEndpoints := make([]querytypes.ServiceEndpoint, 0) + var clusterGatewayNodeIP = make(map[string]string) + collector := NewAppCollector(h.cli, opt) + resources, err := collector.ListApplicationResources(app) + if err != nil { + return err + } + for i, resource := range resources { + cluster := resources[i].Cluster + cachedSelectorNodeIP := func() string { + if ip, exist := clusterGatewayNodeIP[cluster]; exist { + return ip + } + ip := selectorNodeIP(ctx, cluster, h.cli) + if ip != "" { + clusterGatewayNodeIP[cluster] = ip + } + return ip + } + serviceEndpoints = append(serviceEndpoints, getEndpointFromNode(ctx, h.cli, resource.ResourceTree, resource.Component, cachedSelectorNodeIP)...) + } + return fillQueryResult(v, serviceEndpoints, "list") +} + +func getEndpointFromNode(ctx context.Context, cli client.Client, node *querytypes.ResourceTreeNode, component string, cachedSelectorNodeIP func() string) []querytypes.ServiceEndpoint { + if node == nil { + return nil + } + var serviceEndpoints []querytypes.ServiceEndpoint + serviceEndpoints = append(serviceEndpoints, getServiceEndpoints(ctx, cli, node.GroupVersionKind(), node.Name, node.Namespace, node.Cluster, component, cachedSelectorNodeIP)...) + for _, child := range node.LeafNodes { + serviceEndpoints = append(serviceEndpoints, getEndpointFromNode(ctx, cli, child, component, cachedSelectorNodeIP)...) + } + return serviceEndpoints +} + +func getServiceEndpoints(ctx context.Context, cli client.Client, gvk schema.GroupVersionKind, name, namespace, cluster, component string, cachedSelectorNodeIP func() string) []querytypes.ServiceEndpoint { + var serviceEndpoints []querytypes.ServiceEndpoint + switch gvk.Kind { + case "Ingress": + if gvk.Group == networkv1beta1.GroupName && (gvk.Version == "v1beta1" || gvk.Version == "v1") { + var ingress networkv1beta1.Ingress + ingress.SetGroupVersionKind(gvk) + if err := findResource(ctx, cli, &ingress, name, namespace, cluster); err != nil { + klog.Error(err, fmt.Sprintf("find v1 Ingress %s/%s from cluster %s failure", name, namespace, cluster)) + return nil + } + serviceEndpoints = append(serviceEndpoints, generatorFromIngress(ingress, cluster, component)...) + } else { + klog.Warning("not support ingress version", "version", gvk) + } + case "Service": + var service corev1.Service + service.SetGroupVersionKind(gvk) + if err := findResource(ctx, cli, &service, name, namespace, cluster); err != nil { + klog.Error(err, fmt.Sprintf("find v1 Service %s/%s from cluster %s failure", name, namespace, cluster)) + return nil + } + serviceEndpoints = append(serviceEndpoints, generatorFromService(service, cachedSelectorNodeIP, cluster, component, "")...) + case helmapi.HelmReleaseGVK.Kind: + obj := new(unstructured.Unstructured) + obj.SetNamespace(namespace) + obj.SetName(name) + hc := NewHelmReleaseCollector(cli, obj) + services, err := hc.CollectServices(ctx, cluster) + if err != nil { + klog.Error(err, "collect service by helm release failure", "helmRelease", name, "namespace", namespace, "cluster", cluster) + return nil + } + for _, service := range services { + serviceEndpoints = append(serviceEndpoints, generatorFromService(service, cachedSelectorNodeIP, cluster, component, "")...) + } + ingresses, err := hc.CollectIngress(ctx, cluster) + if err != nil { + klog.Error(err, "collect ingres by helm release failure", "helmRelease", name, "namespace", namespace, "cluster", cluster) + return nil + } + for _, uns := range ingresses { + var ingress networkv1beta1.Ingress + if err := runtime.DefaultUnstructuredConverter.FromUnstructured(uns.UnstructuredContent(), &ingress); err != nil { + klog.Errorf("fail to convert unstructured to ingress %s", err.Error()) + continue + } + serviceEndpoints = append(serviceEndpoints, generatorFromIngress(ingress, cluster, component)...) + } + case "SeldonDeployment": + obj := new(unstructured.Unstructured) + obj.SetGroupVersionKind(gvk) + if err := findResource(ctx, cli, obj, name, namespace, cluster); err != nil { + klog.Error(err, fmt.Sprintf("find v1 Seldon Deployment %s/%s from cluster %s failure", name, namespace, cluster)) + return nil + } + anno := obj.GetAnnotations() + serviceName := "ambassador" + serviceNS := apis.DefaultKubeVelaNS + if anno != nil { + if anno[annoAmbassadorServiceName] != "" { + serviceName = anno[annoAmbassadorServiceName] + } + if anno[annoAmbassadorServiceNamespace] != "" { + serviceNS = anno[annoAmbassadorServiceNamespace] + } + } + var service corev1.Service + if err := findResource(ctx, cli, &service, serviceName, serviceNS, cluster); err != nil { + klog.Error(err, fmt.Sprintf("find v1 Service %s/%s from cluster %s failure", serviceName, serviceNS, cluster)) + return nil + } + serviceEndpoints = append(serviceEndpoints, generatorFromService(service, cachedSelectorNodeIP, cluster, component, fmt.Sprintf("/seldon/%s/%s", namespace, name))...) + } + return serviceEndpoints +} + +func findResource(ctx context.Context, cli client.Client, obj client.Object, name, namespace, cluster string) error { + obj.SetNamespace(namespace) + obj.SetName(name) + gctx, cancel := context.WithTimeout(ctx, time.Second*10) + defer cancel() + if err := cli.Get(multicluster.ContextWithClusterName(gctx, cluster), + client.ObjectKeyFromObject(obj), obj); err != nil { + if kerrors.IsNotFound(err) { + return nil + } + return err + } + return nil +} + +func generatorFromService(service corev1.Service, selectorNodeIP func() string, cluster, component, path string) []querytypes.ServiceEndpoint { + var serviceEndpoints []querytypes.ServiceEndpoint + + var objRef = corev1.ObjectReference{ + Kind: "Service", + Namespace: service.ObjectMeta.Namespace, + Name: service.ObjectMeta.Name, + UID: service.UID, + APIVersion: service.APIVersion, + ResourceVersion: service.ResourceVersion, + } + + formatEndpoint := func(host, appProtocol string, portProtocol corev1.Protocol, portNum int32) querytypes.ServiceEndpoint { + return querytypes.ServiceEndpoint{ + Endpoint: querytypes.Endpoint{ + Protocol: portProtocol, + AppProtocol: &appProtocol, + Host: host, + Port: int(portNum), + Path: path, + }, + Ref: objRef, + Cluster: cluster, + Component: component, + } + } + switch service.Spec.Type { + case corev1.ServiceTypeLoadBalancer: + for _, port := range service.Spec.Ports { + appp := judgeAppProtocol(port.Port) + for _, ingress := range service.Status.LoadBalancer.Ingress { + if ingress.Hostname != "" { + serviceEndpoints = append(serviceEndpoints, formatEndpoint(ingress.Hostname, appp, port.Protocol, port.Port)) + } + if ingress.IP != "" { + serviceEndpoints = append(serviceEndpoints, formatEndpoint(ingress.IP, appp, port.Protocol, port.Port)) + } + } + } + case corev1.ServiceTypeNodePort: + for _, port := range service.Spec.Ports { + appp := judgeAppProtocol(port.Port) + serviceEndpoints = append(serviceEndpoints, formatEndpoint(selectorNodeIP(), appp, port.Protocol, port.NodePort)) + } + case corev1.ServiceTypeClusterIP, corev1.ServiceTypeExternalName: + for _, port := range service.Spec.Ports { + appp := judgeAppProtocol(port.Port) + serviceEndpoints = append(serviceEndpoints, formatEndpoint(fmt.Sprintf("%s.%s", service.Name, service.Namespace), appp, port.Protocol, port.Port)) + } + } + return serviceEndpoints +} + +func generatorFromIngress(ingress networkv1beta1.Ingress, cluster, component string) (serviceEndpoints []querytypes.ServiceEndpoint) { + getAppProtocol := func(host string) string { + if len(ingress.Spec.TLS) > 0 { + for _, tls := range ingress.Spec.TLS { + if len(tls.Hosts) > 0 && utils.StringsContain(tls.Hosts, host) { + return querytypes.HTTPS + } + if len(tls.Hosts) == 0 { + return querytypes.HTTPS + } + } + } + return "http" + } + // It depends on the Ingress Controller + getEndpointPort := func(appProtocol string) int { + if appProtocol == querytypes.HTTPS { + if port, err := strconv.Atoi(ingress.Annotations[apis.AnnoIngressControllerHTTPSPort]); port > 0 && err == nil { + return port + } + return 443 + } + if port, err := strconv.Atoi(ingress.Annotations[apis.AnnoIngressControllerHTTPPort]); port > 0 && err == nil { + return port + } + return 80 + } + for _, rule := range ingress.Spec.Rules { + var appProtocol = getAppProtocol(rule.Host) + var appPort = getEndpointPort(appProtocol) + if rule.HTTP != nil { + for _, path := range rule.HTTP.Paths { + serviceEndpoints = append(serviceEndpoints, querytypes.ServiceEndpoint{ + Endpoint: querytypes.Endpoint{ + Protocol: corev1.ProtocolTCP, + AppProtocol: &appProtocol, + Host: rule.Host, + Path: path.Path, + Port: appPort, + }, + Ref: corev1.ObjectReference{ + Kind: "Ingress", + Namespace: ingress.ObjectMeta.Namespace, + Name: ingress.ObjectMeta.Name, + UID: ingress.UID, + APIVersion: ingress.APIVersion, + ResourceVersion: ingress.ResourceVersion, + }, + Cluster: cluster, + Component: component, + }) + } + } + } + return serviceEndpoints +} + +func selectorNodeIP(ctx context.Context, clusterName string, client client.Client) string { + ctx, cancel := context.WithTimeout(ctx, time.Second*10) + defer cancel() + var nodes corev1.NodeList + if err := client.List(multicluster.ContextWithClusterName(ctx, clusterName), &nodes); err != nil { + return "" + } + if len(nodes.Items) == 0 { + return "" + } + var gatewayNode *corev1.Node + var workerNodes []corev1.Node + for i, node := range nodes.Items { + if _, exist := node.Labels[apis.LabelNodeRoleGateway]; exist { + gatewayNode = &nodes.Items[i] + break + } else if _, exist := node.Labels[apis.LabelNodeRoleWorker]; exist { + workerNodes = append(workerNodes, nodes.Items[i]) + } + } + if gatewayNode == nil && len(workerNodes) > 0 { + gatewayNode = &workerNodes[0] + } + if gatewayNode == nil { + gatewayNode = &nodes.Items[0] + } + if gatewayNode != nil { + var addressMap = make(map[corev1.NodeAddressType]string) + for _, address := range gatewayNode.Status.Addresses { + addressMap[address.Type] = address.Address + } + // first get external ip + if ip, exist := addressMap[corev1.NodeExternalIP]; exist { + return ip + } + if ip, exist := addressMap[corev1.NodeInternalIP]; exist { + return ip + } + } + return "" +} + +// judgeAppProtocol RFC-6335 and http://www.iana.org/assignments/service-names). +func judgeAppProtocol(port int32) string { + switch port { + case 80, 8080: + return querytypes.HTTP + case 443: + return querytypes.HTTPS + case 3306: + return querytypes.Mysql + case 6379: + return querytypes.Redis + default: + return "" + } +} diff --git a/pkg/velaql/providers/query/endpoint_test.go b/pkg/velaql/providers/query/endpoint_test.go new file mode 100644 index 000000000..17f8eb18e --- /dev/null +++ b/pkg/velaql/providers/query/endpoint_test.go @@ -0,0 +1,251 @@ +/* + Copyright 2022 The KubeVela Authors. + + 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 query + +import ( + "context" + + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/intstr" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/common" + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + "github.com/oam-dev/kubevela/apis/types" + "github.com/oam-dev/kubevela/pkg/cue/model/value" + "github.com/oam-dev/kubevela/pkg/oam" + "github.com/oam-dev/kubevela/pkg/oam/util" + querytypes "github.com/oam-dev/kubevela/pkg/velaql/providers/query/types" +) + +var _ = Describe("Test Query Provider", func() { + + BeforeEach(func() { + }) + + Context("Test Generate Endpoints", func() { + It("Test endpoints with additional rules", func() { + err := k8sClient.Create(context.TODO(), &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "vela-system", + }, + }) + Expect(err).Should(SatisfyAny(BeNil(), util.AlreadyExistMatcher{})) + sts := common.AppStatus{ + AppliedResources: []common.ClusterObjectReference{ + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + APIVersion: "machinelearning.seldon.io/v1", + Kind: "SeldonDeployment", + Namespace: "default", + Name: "sdep2", + }, + }, + }, + } + testApp := &v1beta1.Application{ + ObjectMeta: metav1.ObjectMeta{ + Name: "endpoints-app-2", + Namespace: "default", + }, + Spec: v1beta1.ApplicationSpec{ + Components: []common.ApplicationComponent{ + { + Name: "endpoints-test-2", + Type: "webservice", + }, + }, + }, + Status: sts, + } + Expect(k8sClient.Create(context.TODO(), testApp)).Should(BeNil()) + var gtapp v1beta1.Application + Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "endpoints-app-2", Namespace: "default"}, >app)).Should(BeNil()) + gtapp.Status = sts + Expect(k8sClient.Status().Update(ctx, >app)).Should(BeNil()) + var mr []v1beta1.ManagedResource + for _, ar := range sts.AppliedResources { + smr := v1beta1.ManagedResource{ + ClusterObjectReference: ar, + } + smr.Component = "endpoints-test-2" + mr = append(mr, smr) + } + rt := &v1beta1.ResourceTracker{ + ObjectMeta: metav1.ObjectMeta{ + Name: "endpoints-app-2", + Namespace: "default", + Labels: map[string]string{ + oam.LabelAppName: testApp.Name, + oam.LabelAppNamespace: testApp.Namespace, + }, + }, + Spec: v1beta1.ResourceTrackerSpec{ + Type: v1beta1.ResourceTrackerTypeRoot, + ManagedResources: mr, + }, + } + err = k8sClient.Create(context.TODO(), rt) + Expect(err).Should(BeNil()) + + By("Prepare configmap for relationship") + + err = k8sClient.Create(context.TODO(), &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Name: "rule-for-seldon-test", + Namespace: types.DefaultKubeVelaNS, + Labels: map[string]string{ + oam.LabelResourceRules: "true", + oam.LabelResourceRuleFormat: oam.ResourceTopologyFormatJSON, + }, + }, + Data: map[string]string{ + "rules": `[ + { + "parentResourceType": { + "group": "machinelearning.seldon.io", + "kind": "SeldonDeployment" + }, + "childrenResourceType": [ + { + "apiVersion": "v1", + "kind": "Service" + } + ] + } +]`, + }, + }) + Expect(err).Should(BeNil()) + testServicelist := []map[string]interface{}{ + { + "name": "clusterip-2", + "ports": []corev1.ServicePort{ + {Port: 80, TargetPort: intstr.FromInt(80), Name: "80port"}, + {Port: 81, TargetPort: intstr.FromInt(81), Name: "81port"}, + }, + "type": corev1.ServiceTypeClusterIP, + }, + { + "name": "seldon-ambassador-2", + "ports": []corev1.ServicePort{ + {Port: 80, TargetPort: intstr.FromInt(80), Name: "80port"}, + }, + "type": corev1.ServiceTypeLoadBalancer, + "status": corev1.ServiceStatus{ + LoadBalancer: corev1.LoadBalancerStatus{ + Ingress: []corev1.LoadBalancerIngress{ + { + IP: "1.1.1.1", + }, + }, + }, + }, + }, + } + + abgvk := schema.GroupVersionKind{ + Group: "machinelearning.seldon.io", + Version: "v1", + Kind: "SeldonDeployment", + } + obj := &unstructured.Unstructured{} + obj.SetName("sdep2") + obj.SetNamespace("default") + obj.SetAnnotations(map[string]string{ + annoAmbassadorServiceName: "seldon-ambassador-2", + annoAmbassadorServiceNamespace: "default", + }) + obj.SetGroupVersionKind(abgvk) + err = k8sClient.Create(context.TODO(), obj) + Expect(err).Should(BeNil()) + abobj := &unstructured.Unstructured{} + abobj.SetGroupVersionKind(abgvk) + Expect(k8sClient.Get(ctx, client.ObjectKey{Name: "sdep2", Namespace: "default"}, abobj)).Should(BeNil()) + + for _, s := range testServicelist { + ns := "default" + if s["namespace"] != nil { + ns = s["namespace"].(string) + } + service := &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: s["name"].(string), + Namespace: ns, + OwnerReferences: []metav1.OwnerReference{ + {APIVersion: "machinelearning.seldon.io/v1", Kind: "SeldonDeployment", Name: "sdep2", UID: abobj.GetUID()}, + }, + }, + Spec: corev1.ServiceSpec{ + Ports: s["ports"].([]corev1.ServicePort), + Type: s["type"].(corev1.ServiceType), + }, + } + + if s["labels"] != nil { + service.Labels = s["labels"].(map[string]string) + } + err := k8sClient.Create(context.TODO(), service) + Expect(err).Should(BeNil()) + if s["status"] != nil { + service.Status = s["status"].(corev1.ServiceStatus) + err := k8sClient.Status().Update(context.TODO(), service) + Expect(err).Should(BeNil()) + } + } + + opt := `app: { + name: "endpoints-app-2" + namespace: "default" + filter: { + cluster: "", + clusterNamespace: "default", + } + }` + v, err := value.NewValue(opt, nil, "") + Expect(err).Should(BeNil()) + pr := &provider{ + cli: k8sClient, + } + err = pr.GeneratorServiceEndpoints(nil, v, nil) + Expect(err).Should(BeNil()) + + urls := []string{ + "http://1.1.1.1/seldon/default/sdep2", + "http://clusterip-2.default", + "clusterip-2.default:81", + "http://1.1.1.1", + } + endValue, err := v.Field("list") + Expect(err).Should(BeNil()) + var endpoints []querytypes.ServiceEndpoint + err = endValue.Decode(&endpoints) + Expect(err).Should(BeNil()) + var edps []string + for _, e := range endpoints { + edps = append(edps, e.String()) + } + Expect(edps).Should(BeEquivalentTo(urls)) + }) + }) +}) diff --git a/pkg/velaql/providers/query/handler.go b/pkg/velaql/providers/query/handler.go index 38aa7bbf4..4db255f3f 100644 --- a/pkg/velaql/providers/query/handler.go +++ b/pkg/velaql/providers/query/handler.go @@ -23,16 +23,12 @@ import ( "encoding/base64" "fmt" "io" - "strconv" "time" "github.com/pkg/errors" corev1 "k8s.io/api/core/v1" - networkv1beta1 "k8s.io/api/networking/v1beta1" - kerrors "k8s.io/apimachinery/pkg/api/errors" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" - "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -40,11 +36,8 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" - apis "github.com/oam-dev/kubevela/apis/types" - helmapi "github.com/oam-dev/kubevela/pkg/appfile/helm/flux2apis" "github.com/oam-dev/kubevela/pkg/cue/model/value" "github.com/oam-dev/kubevela/pkg/multicluster" - "github.com/oam-dev/kubevela/pkg/utils" querytypes "github.com/oam-dev/kubevela/pkg/velaql/providers/query/types" wfContext "github.com/oam-dev/kubevela/pkg/workflow/context" "github.com/oam-dev/kubevela/pkg/workflow/providers" @@ -124,12 +117,12 @@ func (h *provider) ListAppliedResources(ctx wfContext.Context, v *value.Value, a } opt := Option{} if err = val.UnmarshalTo(&opt); err != nil { - return err + return v.FillObject(err.Error(), "err") } collector := NewAppCollector(h.cli, opt) app := new(v1beta1.Application) appKey := client.ObjectKey{Name: opt.Name, Namespace: opt.Namespace} - if err := h.cli.Get(context.Background(), appKey, app); err != nil { + if err = h.cli.Get(context.Background(), appKey, app); err != nil { return v.FillObject(err.Error(), "err") } appResList, err := collector.ListApplicationResources(app) @@ -146,64 +139,22 @@ func (h *provider) ListAppliedResources(ctx wfContext.Context, v *value.Value, a func (h *provider) GetApplicationResourceTree(ctx wfContext.Context, v *value.Value, act types.Action) error { val, err := v.LookupValue("app") if err != nil { - return err + return v.FillObject(err.Error(), "err") } opt := Option{} if err = val.UnmarshalTo(&opt); err != nil { - return err + return v.FillObject(err.Error(), "err") } collector := NewAppCollector(h.cli, opt) app := new(v1beta1.Application) appKey := client.ObjectKey{Name: opt.Name, Namespace: opt.Namespace} - if err := h.cli.Get(context.Background(), appKey, app); err != nil { + if err = h.cli.Get(context.Background(), appKey, app); err != nil { return v.FillObject(err.Error(), "err") } appResList, err := collector.ListApplicationResources(app) if err != nil { return v.FillObject(err.Error(), "err") } - // merge user defined customize rule before every request. - err = mergeCustomRules(context.Background(), h.cli) - if err != nil { - return err - } - for _, resource := range appResList { - root := querytypes.ResourceTreeNode{ - APIVersion: resource.APIVersion, - Kind: resource.Kind, - Cluster: resource.Cluster, - Namespace: resource.Namespace, - Name: resource.Name, - UID: resource.UID, - } - root.LeafNodes, err = iteratorChildResources(context.Background(), resource.Cluster, h.cli, root, 1) - if err != nil { - // if the resource has been deleted, continue access next appliedResource don't break the whole request - if kerrors.IsNotFound(err) { - continue - } - return v.FillObject(err.Error(), "err") - } - rootObject, err := fetchObjectWithResourceTreeNode(context.Background(), resource.Cluster, h.cli, root) - if err != nil { - return v.FillObject(err.Error(), "err") - } - rootStatus, err := checkResourceStatus(*rootObject) - if err != nil { - return v.FillObject(err.Error(), "err") - } - root.HealthStatus = *rootStatus - addInfo, err := additionalInfo(*rootObject) - if err != nil { - return err - } - root.AdditionalInfo = addInfo - root.CreationTimestamp = rootObject.GetCreationTimestamp().Time - if !rootObject.GetDeletionTimestamp().IsZero() { - root.DeletionTimestamp = rootObject.GetDeletionTimestamp().Time - } - resource.ResourceTree = &root - } if appResList == nil { appResList = []*querytypes.AppliedResource{} } @@ -270,136 +221,6 @@ func (h *provider) SearchEvents(ctx wfContext.Context, v *value.Value, act types return fillQueryResult(v, eventList.Items, "list") } -// GeneratorServiceEndpoints generator service endpoints is available for common component type, -// such as webservice or helm -// it can not support the cloud service component currently -func (h *provider) GeneratorServiceEndpoints(wfctx wfContext.Context, v *value.Value, act types.Action) error { - ctx := context.Background() - findResource := func(obj client.Object, name, namespace, cluster string) error { - obj.SetNamespace(namespace) - obj.SetName(name) - gctx, cancel := context.WithTimeout(ctx, time.Second*10) - defer cancel() - if err := h.cli.Get(multicluster.ContextWithClusterName(gctx, cluster), - client.ObjectKeyFromObject(obj), obj); err != nil { - if kerrors.IsNotFound(err) { - return nil - } - return err - } - return nil - } - val, err := v.LookupValue("app") - if err != nil { - return err - } - opt := Option{} - if err = val.UnmarshalTo(&opt); err != nil { - return err - } - app := new(v1beta1.Application) - err = findResource(app, opt.Name, opt.Namespace, "") - if err != nil { - return fmt.Errorf("query app failure %w", err) - } - serviceEndpoints := make([]querytypes.ServiceEndpoint, 0) - var clusterGatewayNodeIP = make(map[string]string) - collector := NewAppCollector(h.cli, opt) - resources, err := collector.ListApplicationResources(app) - if err != nil { - return err - } - - for i, resource := range resources { - cluster := resources[i].Cluster - selectorNodeIP := func() string { - if ip, exist := clusterGatewayNodeIP[cluster]; exist { - return ip - } - ip := selectorNodeIP(ctx, cluster, h.cli) - if ip != "" { - clusterGatewayNodeIP[cluster] = ip - } - return ip - } - switch resource.Kind { - case "Ingress": - if resource.GroupVersionKind().Group == networkv1beta1.GroupName && (resource.GroupVersionKind().Version == "v1beta1" || resource.GroupVersionKind().Version == "v1") { - var ingress networkv1beta1.Ingress - ingress.SetGroupVersionKind(resource.GroupVersionKind()) - if err := findResource(&ingress, resource.Name, resource.Namespace, resource.Cluster); err != nil { - klog.Error(err, fmt.Sprintf("find v1 Ingress %s/%s from cluster %s failure", resource.Name, resource.Namespace, resource.Cluster)) - continue - } - serviceEndpoints = append(serviceEndpoints, generatorFromIngress(ingress, cluster, resource.Component)...) - } else { - klog.Warning("not support ingress version", "version", resource.GroupVersionKind()) - } - case "Service": - var service corev1.Service - service.SetGroupVersionKind(resource.GroupVersionKind()) - if err := findResource(&service, resource.Name, resource.Namespace, resource.Cluster); err != nil { - klog.Error(err, fmt.Sprintf("find v1 Service %s/%s from cluster %s failure", resource.Name, resource.Namespace, resource.Cluster)) - continue - } - serviceEndpoints = append(serviceEndpoints, generatorFromService(service, selectorNodeIP, cluster, resource.Component, "")...) - case helmapi.HelmReleaseGVK.Kind: - obj := new(unstructured.Unstructured) - obj.SetNamespace(resource.Namespace) - obj.SetName(resource.Name) - hc := NewHelmReleaseCollector(h.cli, obj) - services, err := hc.CollectServices(ctx, resource.Cluster) - if err != nil { - klog.Error(err, "collect service by helm release failure", "helmRelease", resource.Name, "namespace", resource.Namespace, "cluster", resource.Cluster) - } - for _, service := range services { - serviceEndpoints = append(serviceEndpoints, generatorFromService(service, selectorNodeIP, cluster, resource.Component, "")...) - } - ingress, err := hc.CollectIngress(ctx, resource.Cluster) - if err != nil { - klog.Error(err, "collect ingres by helm release failure", "helmRelease", resource.Name, "namespace", resource.Namespace, "cluster", resource.Cluster) - } - for _, uns := range ingress { - var ingress networkv1beta1.Ingress - if err := runtime.DefaultUnstructuredConverter.FromUnstructured(uns.UnstructuredContent(), &ingress); err != nil { - klog.Errorf("fail to convert unstructured to ingress %s", err.Error()) - continue - } - serviceEndpoints = append(serviceEndpoints, generatorFromIngress(ingress, cluster, resource.Component)...) - } - case "SeldonDeployment": - obj := new(unstructured.Unstructured) - obj.SetGroupVersionKind(schema.GroupVersionKind{ - Group: "machinelearning.seldon.io", - Version: "v1", - Kind: "SeldonDeployment", - }) - if err := findResource(obj, resource.Name, resource.Namespace, resource.Cluster); err != nil { - klog.Error(err, fmt.Sprintf("find v1 Seldon Deployment %s/%s from cluster %s failure", resource.Name, resource.Namespace, resource.Cluster)) - continue - } - anno := obj.GetAnnotations() - serviceName := "ambassador" - serviceNS := "vela-system" - if anno != nil { - if anno[annoAmbassadorServiceName] != "" { - serviceName = anno[annoAmbassadorServiceName] - } - if anno[annoAmbassadorServiceNamespace] != "" { - serviceNS = anno[annoAmbassadorServiceNamespace] - } - } - var service corev1.Service - if err := findResource(&service, serviceName, serviceNS, resource.Cluster); err != nil { - klog.Error(err, fmt.Sprintf("find v1 Service %s/%s from cluster %s failure", serviceName, serviceNS, resource.Cluster)) - continue - } - serviceEndpoints = append(serviceEndpoints, generatorFromService(service, selectorNodeIP, cluster, resource.Component, fmt.Sprintf("/seldon/%s/%s", resource.Namespace, resource.Name))...) - } - } - return fillQueryResult(v, serviceEndpoints, "list") -} - func (h *provider) CollectLogsInPod(ctx wfContext.Context, v *value.Value, act types.Action) error { cluster, err := v.GetString("cluster") if err != nil { @@ -501,197 +322,3 @@ func Install(p providers.Providers, cli client.Client, cfg *rest.Config) { "getApplicationTree": prd.GetApplicationResourceTree, }) } - -func generatorFromService(service corev1.Service, selectorNodeIP func() string, cluster, component, path string) []querytypes.ServiceEndpoint { - var serviceEndpoints []querytypes.ServiceEndpoint - switch service.Spec.Type { - case corev1.ServiceTypeLoadBalancer: - for _, port := range service.Spec.Ports { - judgeAppProtocol := judgeAppProtocol(port.Port) - for _, ingress := range service.Status.LoadBalancer.Ingress { - if ingress.Hostname != "" { - serviceEndpoints = append(serviceEndpoints, querytypes.ServiceEndpoint{ - Endpoint: querytypes.Endpoint{ - Protocol: port.Protocol, - AppProtocol: &judgeAppProtocol, - Host: ingress.Hostname, - Port: int(port.Port), - Path: path, - }, - Ref: corev1.ObjectReference{ - Kind: "Service", - Namespace: service.ObjectMeta.Namespace, - Name: service.ObjectMeta.Name, - UID: service.UID, - APIVersion: service.APIVersion, - ResourceVersion: service.ResourceVersion, - }, - Cluster: cluster, - Component: component, - }) - } - if ingress.IP != "" { - serviceEndpoints = append(serviceEndpoints, querytypes.ServiceEndpoint{ - Endpoint: querytypes.Endpoint{ - Protocol: port.Protocol, - AppProtocol: &judgeAppProtocol, - Host: ingress.IP, - Port: int(port.Port), - Path: path, - }, - Ref: corev1.ObjectReference{ - Kind: "Service", - Namespace: service.ObjectMeta.Namespace, - Name: service.ObjectMeta.Name, - UID: service.UID, - APIVersion: service.APIVersion, - ResourceVersion: service.ResourceVersion, - }, - Cluster: cluster, - Component: component, - }) - } - } - } - case corev1.ServiceTypeNodePort: - for _, port := range service.Spec.Ports { - judgeAppProtocol := judgeAppProtocol(port.Port) - serviceEndpoints = append(serviceEndpoints, querytypes.ServiceEndpoint{ - Endpoint: querytypes.Endpoint{ - Protocol: port.Protocol, - Port: int(port.NodePort), - AppProtocol: &judgeAppProtocol, - Host: selectorNodeIP(), - Path: path, - }, - Ref: corev1.ObjectReference{ - Kind: "Service", - Namespace: service.ObjectMeta.Namespace, - Name: service.ObjectMeta.Name, - UID: service.UID, - APIVersion: service.APIVersion, - ResourceVersion: service.ResourceVersion, - }, - Cluster: cluster, - Component: component, - }) - } - case corev1.ServiceTypeClusterIP, corev1.ServiceTypeExternalName: - } - return serviceEndpoints -} - -func generatorFromIngress(ingress networkv1beta1.Ingress, cluster, component string) (serviceEndpoints []querytypes.ServiceEndpoint) { - getAppProtocol := func(host string) string { - if len(ingress.Spec.TLS) > 0 { - for _, tls := range ingress.Spec.TLS { - if len(tls.Hosts) > 0 && utils.StringsContain(tls.Hosts, host) { - return querytypes.HTTPS - } - if len(tls.Hosts) == 0 { - return querytypes.HTTPS - } - } - } - return "http" - } - // It depends on the Ingress Controller - getEndpointPort := func(appProtocol string) int { - if appProtocol == querytypes.HTTPS { - if port, err := strconv.Atoi(ingress.Annotations[apis.AnnoIngressControllerHTTPSPort]); port > 0 && err == nil { - return port - } - return 443 - } - if port, err := strconv.Atoi(ingress.Annotations[apis.AnnoIngressControllerHTTPPort]); port > 0 && err == nil { - return port - } - return 80 - } - for _, rule := range ingress.Spec.Rules { - var appProtocol = getAppProtocol(rule.Host) - var appPort = getEndpointPort(appProtocol) - if rule.HTTP != nil { - for _, path := range rule.HTTP.Paths { - serviceEndpoints = append(serviceEndpoints, querytypes.ServiceEndpoint{ - Endpoint: querytypes.Endpoint{ - Protocol: corev1.ProtocolTCP, - AppProtocol: &appProtocol, - Host: rule.Host, - Path: path.Path, - Port: appPort, - }, - Ref: corev1.ObjectReference{ - Kind: "Ingress", - Namespace: ingress.ObjectMeta.Namespace, - Name: ingress.ObjectMeta.Name, - UID: ingress.UID, - APIVersion: ingress.APIVersion, - ResourceVersion: ingress.ResourceVersion, - }, - Cluster: cluster, - Component: component, - }) - } - } - } - return serviceEndpoints -} - -func selectorNodeIP(ctx context.Context, clusterName string, client client.Client) string { - ctx, cancel := context.WithTimeout(ctx, time.Second*10) - defer cancel() - var nodes corev1.NodeList - if err := client.List(multicluster.ContextWithClusterName(ctx, clusterName), &nodes); err != nil { - return "" - } - if len(nodes.Items) == 0 { - return "" - } - var gatewayNode *corev1.Node - var workerNodes []corev1.Node - for i, node := range nodes.Items { - if _, exist := node.Labels[apis.LabelNodeRoleGateway]; exist { - gatewayNode = &nodes.Items[i] - break - } else if _, exist := node.Labels[apis.LabelNodeRoleWorker]; exist { - workerNodes = append(workerNodes, nodes.Items[i]) - } - } - if gatewayNode == nil && len(workerNodes) > 0 { - gatewayNode = &workerNodes[0] - } - if gatewayNode == nil { - gatewayNode = &nodes.Items[0] - } - if gatewayNode != nil { - var addressMap = make(map[corev1.NodeAddressType]string) - for _, address := range gatewayNode.Status.Addresses { - addressMap[address.Type] = address.Address - } - // first get external ip - if ip, exist := addressMap[corev1.NodeExternalIP]; exist { - return ip - } - if ip, exist := addressMap[corev1.NodeInternalIP]; exist { - return ip - } - } - return "" -} - -// judgeAppProtocol RFC-6335 and http://www.iana.org/assignments/service-names). -func judgeAppProtocol(port int32) string { - switch port { - case 80, 8080: - return querytypes.HTTP - case 443: - return querytypes.HTTPS - case 3306: - return querytypes.Mysql - case 6379: - return querytypes.Redis - default: - return "" - } -} diff --git a/pkg/velaql/providers/query/handler_test.go b/pkg/velaql/providers/query/handler_test.go index 4e19101ea..82e670781 100644 --- a/pkg/velaql/providers/query/handler_test.go +++ b/pkg/velaql/providers/query/handler_test.go @@ -569,6 +569,73 @@ options: { }) It("Test generator service endpoints", func() { + appsts := common.AppStatus{ + AppliedResources: []common.ClusterObjectReference{ + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + Kind: "Ingress", + Namespace: "default", + Name: "ingress-http", + APIVersion: "networking.k8s.io/v1beta1", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + Kind: "Ingress", + Namespace: "default", + Name: "ingress-https", + APIVersion: "networking.k8s.io/v1beta1", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + Kind: "Ingress", + Namespace: "default", + Name: "ingress-paths", + APIVersion: "networking.k8s.io/v1beta1", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + APIVersion: "v1", + Kind: "Service", + Namespace: "default", + Name: "nodeport", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + APIVersion: "v1", + Kind: "Service", + Namespace: "default", + Name: "loadbalancer", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + APIVersion: "helm.toolkit.fluxcd.io/v2beta1", + Kind: helmapi.HelmReleaseGVK.Kind, + Namespace: "default", + Name: "helmRelease", + }, + }, + { + Cluster: "", + ObjectReference: corev1.ObjectReference{ + APIVersion: "machinelearning.seldon.io/v1", + Kind: "SeldonDeployment", + Namespace: "default", + Name: "sdep", + }, + }, + }, + } testApp := &v1beta1.Application{ ObjectMeta: metav1.ObjectMeta{ Name: "endpoints-app", @@ -582,77 +649,22 @@ options: { }, }, }, - Status: common.AppStatus{ - AppliedResources: []common.ClusterObjectReference{ - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "Ingress", - Namespace: "default", - Name: "ingress-http", - APIVersion: "networking.k8s.io/v1beta1", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "Ingress", - Namespace: "default", - Name: "ingress-https", - APIVersion: "networking.k8s.io/v1", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "Ingress", - Namespace: "default", - Name: "ingress-paths", - APIVersion: "networking.k8s.io/v1", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "Service", - Namespace: "default", - Name: "nodeport", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "Service", - Namespace: "default", - Name: "loadbalancer", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: helmapi.HelmReleaseGVK.Kind, - Namespace: "default", - Name: "helmRelease", - }, - }, - { - Cluster: "", - ObjectReference: corev1.ObjectReference{ - Kind: "SeldonDeployment", - Namespace: "default", - Name: "sdep", - }, - }, - }, - }, + Status: appsts, } err := k8sClient.Create(context.TODO(), testApp) Expect(err).Should(BeNil()) + + var gtapp v1beta1.Application + Expect(k8sClient.Get(context.TODO(), client.ObjectKey{Name: "endpoints-app", Namespace: "default"}, >app)).Should(BeNil()) + gtapp.Status = appsts + Expect(k8sClient.Status().Update(ctx, >app)).Should(BeNil()) var mr []v1beta1.ManagedResource - for i := range testApp.Status.AppliedResources { - mr = append(mr, v1beta1.ManagedResource{ - ClusterObjectReference: testApp.Status.AppliedResources[i], - }) + for _, ar := range appsts.AppliedResources { + smr := v1beta1.ManagedResource{ + ClusterObjectReference: ar, + } + smr.Component = "endpoints-test" + mr = append(mr, smr) } rt := &v1beta1.ResourceTracker{ ObjectMeta: metav1.ObjectMeta{ @@ -962,19 +974,18 @@ options: { "http://text.example.com", "10.10.10.10:81", "text.example.com:81", - // helmRelease - fmt.Sprintf("http://%s:30002", gatewayIP), - "http://ingress.domain.helm", - "1.1.1.1:80/seldon/test", + "http://1.1.1.1/seldon/default/sdep", } endValue, err := v.Field("list") Expect(err).Should(BeNil()) var endpoints []querytypes.ServiceEndpoint err = endValue.Decode(&endpoints) Expect(err).Should(BeNil()) - for i, endpoint := range endpoints { - Expect(endpoint.String()).Should(BeEquivalentTo(urls[i])) + var edps []string + for _, e := range endpoints { + edps = append(edps, e.String()) } + Expect(edps).Should(BeEquivalentTo(urls)) }) }) diff --git a/pkg/velaql/providers/query/tree.go b/pkg/velaql/providers/query/tree.go index 11e78b178..3bfacee38 100644 --- a/pkg/velaql/providers/query/tree.go +++ b/pkg/velaql/providers/query/tree.go @@ -18,6 +18,7 @@ package query import ( "context" + "encoding/json" "fmt" "time" @@ -29,6 +30,7 @@ import ( "k8s.io/apimachinery/pkg/runtime" types2 "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/duration" + "k8s.io/klog" "k8s.io/kubectl/pkg/util/podutils" "sigs.k8s.io/controller-runtime/pkg/client" @@ -696,7 +698,7 @@ func iteratorChildResources(ctx context.Context, cluster string, k8sClient clien items, err := listItemByRule(clusterCTX, k8sClient, resource, *parentObject, specifiedFunc, rules.DefaultGenListOptionFunc, rules.DisableFilterByOwnerReference) if err != nil { if meta.IsNoMatchError(err) || runtime.IsNotRegisteredError(err) { - log.Logger.Errorf("error to list subresources: %s err: %v", resource.Kind, err) + klog.Errorf("error to list sub-resources: %s err: %v", resource.Kind, err) continue } return nil, err @@ -739,19 +741,32 @@ func iteratorChildResources(ctx context.Context, cluster string, k8sClient clien return nil, nil } -// mergeCustomRules merge the customize +// mergeCustomRules merge user defined resource topology rules with the system ones func mergeCustomRules(ctx context.Context, k8sClient client.Client) error { rulesList := v12.ConfigMapList{} if err := k8sClient.List(ctx, &rulesList, client.InNamespace(velatypes.DefaultKubeVelaNS), client.HasLabels{oam.LabelResourceRules}); err != nil { - return err + return client.IgnoreNotFound(err) } for _, item := range rulesList.Items { ruleStr := item.Data[relationshipKey] - var customRules []*customRule - err := yaml.Unmarshal([]byte(ruleStr), &customRules) + var ( + customRules []*customRule + format string + err error + ) + if item.Annotations != nil { + format = item.Annotations[oam.LabelResourceRuleFormat] + } + switch format { + case oam.ResourceTopologyFormatJSON: + err = json.Unmarshal([]byte(ruleStr), &customRules) + case oam.ResourceTopologyFormatYAML, "": + err = yaml.Unmarshal([]byte(ruleStr), &customRules) + } if err != nil { // don't let one miss-config configmap brake whole process - log.Logger.Errorf("relationship rule configamp %s miss config %v", item.Name, err) + klog.Errorf("relationship rule configmap %s miss config %v", item.Name, err) + continue } for _, rule := range customRules { if cResource, ok := globalRule[*rule.ParentResourceType]; ok { diff --git a/pkg/velaql/providers/query/types/type.go b/pkg/velaql/providers/query/types/type.go index 22b0b4ddd..610b9eff6 100644 --- a/pkg/velaql/providers/query/types/type.go +++ b/pkg/velaql/providers/query/types/type.go @@ -122,7 +122,12 @@ type ResourceTreeNode struct { AdditionalInfo map[string]interface{} `json:"additionalInfo,omitempty"` } -// GroupVersionKind returns the stored group, version, and kind of an object +// GroupVersionKind returns the stored group, version, and kind from AppliedResource func (obj *AppliedResource) GroupVersionKind() schema.GroupVersionKind { return schema.FromAPIVersionAndKind(obj.APIVersion, obj.Kind) } + +// GroupVersionKind returns the stored group, version, and kind from ResourceTreeNode +func (rtn *ResourceTreeNode) GroupVersionKind() schema.GroupVersionKind { + return schema.FromAPIVersionAndKind(rtn.APIVersion, rtn.Kind) +} diff --git a/pkg/velaql/providers/query/utils.go b/pkg/velaql/providers/query/utils.go index fc05c1c5b..2c0fa2aba 100644 --- a/pkg/velaql/providers/query/utils.go +++ b/pkg/velaql/providers/query/utils.go @@ -34,5 +34,9 @@ func fillQueryResult(v *value.Value, res interface{}, paths ...string) error { if err != nil { return v.FillObject(err, "err") } - return v.FillObject(expr, paths...) + err = v.FillObject(expr, paths...) + if err != nil { + return err + } + return v.Error() } diff --git a/references/cli/velaql_test.go b/references/cli/velaql_test.go index 557bcd29f..a409f89ba 100644 --- a/references/cli/velaql_test.go +++ b/references/cli/velaql_test.go @@ -141,17 +141,19 @@ var _ = Describe("Test velaQL", func() { { Cluster: "", ObjectReference: corev1.ObjectReference{ - Kind: "Service", - Namespace: "default", - Name: "nodeport", + Kind: "Service", + Namespace: "default", + Name: "nodeport", + APIVersion: "v1", }, }, { Cluster: "", ObjectReference: corev1.ObjectReference{ - Kind: "Service", - Namespace: "default", - Name: "loadbalancer", + Kind: "Service", + Namespace: "default", + Name: "loadbalancer", + APIVersion: "v1", }, }, { diff --git a/test/e2e-apiserver-test/velaql_test.go b/test/e2e-apiserver-test/velaql_test.go index bd2b44b40..dfd47b3bb 100644 --- a/test/e2e-apiserver-test/velaql_test.go +++ b/test/e2e-apiserver-test/velaql_test.go @@ -400,7 +400,7 @@ var _ = Describe("Test velaQL rest api", func() { return err } if len(status.Resources) != 1 { - return fmt.Errorf("applied resource velaql error") + return fmt.Errorf("applied resource velaql error, expect to be 1 but %d", len(status.Resources)) } return nil }, 30*time.Second, 300*time.Millisecond).Should(BeNil())