Feat: support resource topology for endpoints (#4362)

* Feat: support resource topology for endpoints

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

* Fix: add and refactor the test for endpoint

Signed-off-by: Jianbo Sun <jianbo.sjb@alibaba-inc.com>

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"
+)

pkg/stdlib/pkgs/query.cue

@@ -50,6 +50,9 @@
 		deployVersion?:   string
 		revision?:        string
 		latest?:          bool
+		resourceTree?: {
+			...
+		}
 	}]
 	...
 }

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
 }
 

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 ""
+	}
+}

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"}, &gtapp)).Should(BeNil())
+			gtapp.Status = sts
+			Expect(k8sClient.Status().Update(ctx, &gtapp)).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))
+		})
+	})
+})

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 ""
-	}
-}

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"}, &gtapp)).Should(BeNil())
+		gtapp.Status = appsts
+		Expect(k8sClient.Status().Update(ctx, &gtapp)).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))
 	})
 })
 

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 {

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)
+}

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()
 }

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",
 					},
 				},
 				{

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())