proxy/pkg/infras/istio/xds.go - servicecomb-mesher - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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 pilotv2

 import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"strings"

 	k8sinfra "github.com/apache/servicecomb-mesher/proxy/pkg/infras/k8s"
 	apiv2 "github.com/envoyproxy/go-control-plane/envoy/api/v2"
 	apiv2core "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
 	apiv2endpoint "github.com/envoyproxy/go-control-plane/envoy/api/v2/endpoint"
 	apiv2route "github.com/envoyproxy/go-control-plane/envoy/api/v2/route"

 	v2 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
 	"github.com/go-mesh/openlogging"
 	"github.com/gogo/protobuf/proto"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"k8s.io/client-go/rest"
 )

 //XdsClient provides the XDS API calls.
 type XdsClient struct {
 	PilotAddr   string
 	TlsConfig   *tls.Config
 	ReqCaches   map[XdsType]*XdsReqCache
 	nodeInfo    *NodeInfo
 	NodeID      string
 	NodeCluster string
 	k8sClient   *rest.RESTClient
 }

 //XdsType is the wrapper of string, the wrapper type should be "cds", "eds", "lds" or "rds"
 type XdsType string

 const (
 	TypeCds XdsType = "cds"
 	TypeEds XdsType = "eds"
 	TypeLds XdsType = "lds"
 	TypeRds XdsType = "rds"
 )

 //XdsReqCache stores the VersionInfo and Nonce for the XDS calls
 type XdsReqCache struct {
 	Nonce       string
 	VersionInfo string
 }

 //NodeInfo stores the info of the node, which will be used to make a
 //XDS call
 type NodeInfo struct {
 	PodName    string
 	Namespace  string
 	InstanceIP string
 }

 //XdsClusterInfo stores all the infos from a cluster name, which is in
 //the format direction|port|subset|hostName
 type XdsClusterInfo struct {
 	ClusterName  string
 	Direction    string
 	Port         string
 	Subset       string
 	HostName     string
 	ServiceName  string
 	Namespace    string
 	DomainSuffix string // DomainSuffix might not be used
 	Tags         map[string]string
 	Addrs        []string // The accessible addresses of the endpoints
 }

 //NewXdsClient returns the new XDS client.
 func NewXdsClient(pilotAddr string, tlsConfig *tls.Config, nodeInfo *NodeInfo, kubeconfigPath string) (*XdsClient, error) {
 	// TODO Handle the array
 	xdsClient := &XdsClient{
 		PilotAddr: pilotAddr,
 		nodeInfo:  nodeInfo,
 	}
 	xdsClient.NodeID = "sidecar~" + nodeInfo.InstanceIP + "~" + nodeInfo.PodName + "~" + nodeInfo.Namespace
 	xdsClient.NodeCluster = nodeInfo.PodName

 	xdsClient.ReqCaches = map[XdsType]*XdsReqCache{
 		TypeCds: {},
 		TypeEds: {},
 		TypeLds: {},
 		TypeRds: {},
 	}

 	if k8sClient, err := k8sinfra.CreateK8SRestClient(kubeconfigPath, "apis", "networking.istio.io", "v1alpha3"); err != nil {
 		return nil, err
 	} else {
 		xdsClient.k8sClient = k8sClient
 	}

 	return xdsClient, nil
 }

 //GetSubsetTags returns the tags of the specified subset.
 func (client *XdsClient) GetSubsetTags(namespace, hostName, subsetName string) (map[string]string, error) {
 	req := client.k8sClient.Get()
 	req.Resource("destinationrules")
 	req.Namespace(namespace)

 	result := req.Do()
 	rawBody, err := result.Raw()
 	if err != nil {
 		return nil, err
 	}

 	var drResult k8sinfra.DestinationRuleResult
 	if err := json.Unmarshal(rawBody, &drResult); err != nil {
 		return nil, err
 	}

 	// Find the subset
 	tags := map[string]string{}
 	for _, dr := range drResult.Items {
 		if dr.Spec.Host == hostName {
 			for _, subset := range dr.Spec.Subsets {
 				if subset.Name == subsetName {
 					for k, v := range subset.Labels {
 						tags[k] = v
 					}
 					break
 				}
 			}
 			break
 		}
 	}

 	return tags, nil
 }

 func (client *XdsClient) getGrpcConn() (*grpc.ClientConn, error) {
 	var conn *grpc.ClientConn
 	var err error
 	if client.TlsConfig != nil {
 		creds := credentials.NewTLS(client.TlsConfig)
 		conn, err = grpc.Dial(client.PilotAddr, grpc.WithTransportCredentials(creds))
 	} else {
 		conn, err = grpc.Dial(client.PilotAddr, grpc.WithInsecure())
 	}

 	return conn, err
 }

 func getAdsResClient(client *XdsClient) (v2.AggregatedDiscoveryService_StreamAggregatedResourcesClient, *grpc.ClientConn, error) {
 	conn, err := client.getGrpcConn()
 	if err != nil {
 		return nil, nil, err
 	}

 	adsClient := v2.NewAggregatedDiscoveryServiceClient(conn)
 	adsResClient, err := adsClient.StreamAggregatedResources(context.Background())
 	if err != nil {
 		return nil, nil, err
 	}

 	return adsResClient, conn, nil
 }

 func (client *XdsClient) getRouterClusters(clusterName string) ([]string, error) {
 	virtualHosts, err := client.RDS(clusterName)
 	if err != nil {
 		return nil, err
 	}

 	routerClusters := []string{}
 	for _, h := range virtualHosts {
 		for _, r := range h.Routes {
 			routerClusters = append(routerClusters, r.GetRoute().GetCluster())
 		}
 	}

 	return routerClusters, nil
 }

 func (client *XdsClient) getVersionInfo(resType XdsType) string {
 	return client.ReqCaches[resType].VersionInfo
 }
 func (client *XdsClient) getNonce(resType XdsType) string {
 	return client.ReqCaches[resType].Nonce
 }

 func (client *XdsClient) setVersionInfo(resType XdsType, versionInfo string) {
 	client.ReqCaches[resType].VersionInfo = versionInfo
 }

 func (client *XdsClient) setNonce(resType XdsType, nonce string) {
 	client.ReqCaches[resType].Nonce = nonce
 }

 //CDS s the Clsuter Discovery Service API, which fetches all the clusters from istio pilot
 func (client *XdsClient) CDS() ([]apiv2.Cluster, error) {
 	adsResClient, conn, err := getAdsResClient(client)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close()

 	req := &apiv2.DiscoveryRequest{
 		TypeUrl:       "type.googleapis.com/envoy.api.v2.Cluster",
 		VersionInfo:   client.getVersionInfo(TypeCds),
 		ResponseNonce: client.getNonce(TypeCds),
 	}
 	req.Node = &apiv2core.Node{
 		// Sample taken from istio: router~172.30.77.6~istio-egressgateway-84b4d947cd-rqt45.istio-system~istio-system.svc.cluster.local-2
 		// The Node.Id should be in format {nodeType}~{ipAddr}~{serviceId~{domain}, splitted by '~'
 		// The format is required by pilot
 		Id:      client.NodeID,
 		Cluster: client.NodeCluster,
 	}

 	if err := adsResClient.Send(req); err != nil {
 		return nil, err
 	}

 	resp, err := adsResClient.Recv()
 	if err != nil {
 		return nil, err
 	}

 	client.setNonce(TypeCds, resp.GetNonce())
 	client.setVersionInfo(TypeCds, resp.GetVersionInfo())
 	resources := resp.GetResources()

 	var cluster apiv2.Cluster
 	clusters := []apiv2.Cluster{}
 	for _, res := range resources {
 		if err := proto.Unmarshal(res.GetValue(), &cluster); err != nil {
 			openlogging.GetLogger().Warnf("Failed to unmarshal cluster resource: %s", err.Error())
 		} else {
 			clusters = append(clusters, cluster)
 		}
 	}
 	return clusters, nil
 }

 //EDS is the Endpoint Discovery Service API, the API takes the cluster's name and return all its endpoints(which provide address and port)
 func (client *XdsClient) EDS(clusterName string) (*apiv2.ClusterLoadAssignment, error) {
 	adsResClient, conn, err := getAdsResClient(client)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close()

 	req := &apiv2.DiscoveryRequest{
 		TypeUrl:       "type.googleapis.com/envoy.api.v2.ClusterLoadAssignment",
 		VersionInfo:   client.getVersionInfo(TypeEds),
 		ResponseNonce: client.getNonce(TypeEds),
 	}

 	req.Node = &apiv2core.Node{
 		Id:      client.NodeID,
 		Cluster: client.NodeCluster,
 	}
 	req.ResourceNames = []string{clusterName}
 	if err := adsResClient.Send(req); err != nil {
 		return nil, err
 	}

 	resp, err := adsResClient.Recv()
 	if err != nil {
 		return nil, err
 	}

 	resources := resp.GetResources()
 	client.setNonce(TypeEds, resp.GetNonce())
 	client.setVersionInfo(TypeEds, resp.GetVersionInfo())

 	var loadAssignment apiv2.ClusterLoadAssignment
 	var e error
 	// endpoints := []apiv2.ClusterLoadAssignment{}

 	for _, res := range resources {
 		if err := proto.Unmarshal(res.GetValue(), &loadAssignment); err != nil {
 			e = err
 		} else {
 			// The cluster's LoadAssignment will always be ONE, with Endpoints as its field
 			break
 		}
 	}
 	return &loadAssignment, e
 }

 //GetEndpointsByTags fetches the cluster's endpoints with tags. The tags is usually specified in a DestinationRule.
 func (client *XdsClient) GetEndpointsByTags(serviceName string, tags map[string]string) ([]*apiv2endpoint.LbEndpoint, string, error) {
 	clusters, err := client.CDS()
 	if err != nil {
 		return nil, "", err
 	}

 	lbendpoints := []*apiv2endpoint.LbEndpoint{}
 	clusterName := ""
 	for _, cluster := range clusters {
 		clusterInfo := ParseClusterName(cluster.Name)
 		if clusterInfo == nil || clusterInfo.Subset == "" || clusterInfo.ServiceName != serviceName {
 			continue
 		}
 		// So clusterInfo is not nil and subset is not empty
 		if subsetTags, err := client.GetSubsetTags(clusterInfo.Namespace, clusterInfo.ServiceName, clusterInfo.Subset); err == nil {
 			// filter with tags
 			matched := true
 			for k, v := range tags {
 				if subsetTagValue, exists := subsetTags[k]; exists == false || subsetTagValue != v {
 					matched = false
 					break
 				}
 			}

 			if matched { // We got the cluster!
 				clusterName = cluster.Name
 				loadAssignment, err := client.EDS(cluster.Name)
 				if err != nil {
 					return nil, clusterName, err
 				}

 				for _, item := range loadAssignment.Endpoints {
 					lbendpoints = append(lbendpoints, item.LbEndpoints...)
 				}

 				return lbendpoints, clusterName, nil
 			}
 		}
 	}

 	return lbendpoints, clusterName, nil
 }

 //RDS is the Router Discovery Service API, it returns the virtual hosts which contains Routes
 func (client *XdsClient) RDS(clusterName string) ([]*apiv2route.VirtualHost, error) {
 	clusterInfo := ParseClusterName(clusterName)
 	if clusterInfo == nil {
 		return nil, fmt.Errorf("Invalid clusterName for routers: %s", clusterName)
 	}

 	adsResClient, conn, err := getAdsResClient(client)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close()

 	req := &apiv2.DiscoveryRequest{
 		TypeUrl:       "type.googleapis.com/envoy.api.v2.RouteConfiguration",
 		VersionInfo:   client.getVersionInfo(TypeRds),
 		ResponseNonce: client.getNonce(TypeRds),
 	}

 	req.Node = &apiv2core.Node{
 		Id:      client.NodeID,
 		Cluster: client.NodeCluster,
 	}
 	req.ResourceNames = []string{clusterName}
 	if err := adsResClient.Send(req); err != nil {
 		return nil, err
 	}

 	resp, err := adsResClient.Recv()
 	if err != nil {
 		return nil, err
 	}

 	resources := resp.GetResources()
 	client.setNonce(TypeRds, resp.GetNonce())
 	client.setVersionInfo(TypeRds, resp.GetVersionInfo())

 	var route apiv2.RouteConfiguration
 	virtualHosts := []*apiv2route.VirtualHost{}

 	for _, res := range resources {
 		if err := proto.Unmarshal(res.GetValue(), &route); err != nil {
 			openlogging.GetLogger().Warnf("Failed to unmarshal router resource: ", err.Error())
 		} else {
 			vhosts := route.GetVirtualHosts()
 			for _, vhost := range vhosts {
 				if vhost.Name == clusterInfo.ServiceName+":"+clusterInfo.Port {
 					virtualHosts = append(virtualHosts, vhost)
 				}
 			}
 		}
 	}
 	return virtualHosts, nil
 }

 //LDS is the Listener Discovery Service API, which returns all the listerns
 func (client *XdsClient) LDS() ([]apiv2.Listener, error) {
 	adsResClient, conn, err := getAdsResClient(client)
 	if err != nil {
 		return nil, err
 	}
 	defer conn.Close()

 	req := &apiv2.DiscoveryRequest{
 		TypeUrl:       "type.googleapis.com/envoy.api.v2.Listener",
 		VersionInfo:   client.getVersionInfo(TypeLds),
 		ResponseNonce: client.getNonce(TypeLds),
 	}

 	req.Node = &apiv2core.Node{
 		Id:      client.NodeID,
 		Cluster: client.NodeCluster,
 	}
 	if err := adsResClient.Send(req); err != nil {
 		return nil, err
 	}

 	resp, err := adsResClient.Recv()
 	if err != nil {
 		return nil, err
 	}

 	resources := resp.GetResources()
 	client.setNonce(TypeLds, resp.GetNonce())
 	client.setVersionInfo(TypeLds, resp.GetVersionInfo())

 	var listener apiv2.Listener
 	listeners := []apiv2.Listener{}

 	for _, res := range resources {
 		if err := proto.Unmarshal(res.GetValue(), &listener); err != nil {
 			openlogging.GetLogger().Warnf("Failed to unmarshal listener resource: ", err.Error())
 		} else {
 			listeners = append(listeners, listener)
 		}
 	}
 	return listeners, nil
 }

 //ParseClusterName parse the cluster's name, which is in the format direction|port|subset|hostName, the 4 items will be parsed into different fields. The hostName item will also be parsed into ServcieName, Namespace etc.
 func ParseClusterName(clusterName string) *XdsClusterInfo {
 	// clusterName format: direction|port|subset|hostName
 	// hostName format: |svc.namespace.svc.cluster.local

 	parts := strings.Split(clusterName, "|")
 	if len(parts) != 4 {
 		return nil
 	}

 	hostnameParts := strings.Split(parts[3], ".")
 	if len(hostnameParts) < 2 {
 		return nil
 	}

 	cluster := &XdsClusterInfo{
 		Direction:    parts[0],
 		Port:         parts[1],
 		Subset:       parts[2],
 		HostName:     parts[3],
 		ServiceName:  hostnameParts[0],
 		Namespace:    hostnameParts[1],
 		DomainSuffix: strings.Join(hostnameParts[2:], "."),
 		ClusterName:  clusterName,
 	}

 	return cluster
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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 pilotv2

	import (
	"context"
	"crypto/tls"
	"encoding/json"
	"fmt"
	"strings"

	k8sinfra "github.com/apache/servicecomb-mesher/proxy/pkg/infras/k8s"
	apiv2 "github.com/envoyproxy/go-control-plane/envoy/api/v2"
	apiv2core "github.com/envoyproxy/go-control-plane/envoy/api/v2/core"
	apiv2endpoint "github.com/envoyproxy/go-control-plane/envoy/api/v2/endpoint"
	apiv2route "github.com/envoyproxy/go-control-plane/envoy/api/v2/route"

	v2 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v2"
	"github.com/go-mesh/openlogging"
	"github.com/gogo/protobuf/proto"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"k8s.io/client-go/rest"
	)

	//XdsClient provides the XDS API calls.
	type XdsClient struct {
	PilotAddr string
	TlsConfig *tls.Config
	ReqCaches map[XdsType]*XdsReqCache
	nodeInfo *NodeInfo
	NodeID string
	NodeCluster string
	k8sClient *rest.RESTClient
	}

	//XdsType is the wrapper of string, the wrapper type should be "cds", "eds", "lds" or "rds"
	type XdsType string

	const (
	TypeCds XdsType = "cds"
	TypeEds XdsType = "eds"
	TypeLds XdsType = "lds"
	TypeRds XdsType = "rds"
	)

	//XdsReqCache stores the VersionInfo and Nonce for the XDS calls
	type XdsReqCache struct {
	Nonce string
	VersionInfo string
	}

	//NodeInfo stores the info of the node, which will be used to make a
	//XDS call
	type NodeInfo struct {
	PodName string
	Namespace string
	InstanceIP string
	}

	//XdsClusterInfo stores all the infos from a cluster name, which is in
	//the format direction\|port\|subset\|hostName
	type XdsClusterInfo struct {
	ClusterName string
	Direction string
	Port string
	Subset string
	HostName string
	ServiceName string
	Namespace string
	DomainSuffix string // DomainSuffix might not be used
	Tags map[string]string
	Addrs []string // The accessible addresses of the endpoints
	}

	//NewXdsClient returns the new XDS client.
	func NewXdsClient(pilotAddr string, tlsConfig tls.Config, nodeInfo NodeInfo, kubeconfigPath string) (*XdsClient, error) {
	// TODO Handle the array
	xdsClient := &XdsClient{
	PilotAddr: pilotAddr,
	nodeInfo: nodeInfo,
	}
	xdsClient.NodeID = "sidecar~" + nodeInfo.InstanceIP + "~" + nodeInfo.PodName + "~" + nodeInfo.Namespace
	xdsClient.NodeCluster = nodeInfo.PodName

	xdsClient.ReqCaches = map[XdsType]*XdsReqCache{
	TypeCds: {},
	TypeEds: {},
	TypeLds: {},
	TypeRds: {},
	}

	if k8sClient, err := k8sinfra.CreateK8SRestClient(kubeconfigPath, "apis", "networking.istio.io", "v1alpha3"); err != nil {
	return nil, err
	} else {
	xdsClient.k8sClient = k8sClient
	}

	return xdsClient, nil
	}

	//GetSubsetTags returns the tags of the specified subset.
	func (client *XdsClient) GetSubsetTags(namespace, hostName, subsetName string) (map[string]string, error) {
	req := client.k8sClient.Get()
	req.Resource("destinationrules")
	req.Namespace(namespace)

	result := req.Do()
	rawBody, err := result.Raw()
	if err != nil {
	return nil, err
	}

	var drResult k8sinfra.DestinationRuleResult
	if err := json.Unmarshal(rawBody, &drResult); err != nil {
	return nil, err
	}

	// Find the subset
	tags := map[string]string{}
	for _, dr := range drResult.Items {
	if dr.Spec.Host == hostName {
	for _, subset := range dr.Spec.Subsets {
	if subset.Name == subsetName {
	for k, v := range subset.Labels {
	tags[k] = v
	}
	break
	}
	}
	break
	}
	}

	return tags, nil
	}

	func (client XdsClient) getGrpcConn() (grpc.ClientConn, error) {
	var conn *grpc.ClientConn
	var err error
	if client.TlsConfig != nil {
	creds := credentials.NewTLS(client.TlsConfig)
	conn, err = grpc.Dial(client.PilotAddr, grpc.WithTransportCredentials(creds))
	} else {
	conn, err = grpc.Dial(client.PilotAddr, grpc.WithInsecure())
	}

	return conn, err
	}

	func getAdsResClient(client XdsClient) (v2.AggregatedDiscoveryService_StreamAggregatedResourcesClient, grpc.ClientConn, error) {
	conn, err := client.getGrpcConn()
	if err != nil {
	return nil, nil, err
	}

	adsClient := v2.NewAggregatedDiscoveryServiceClient(conn)
	adsResClient, err := adsClient.StreamAggregatedResources(context.Background())
	if err != nil {
	return nil, nil, err
	}

	return adsResClient, conn, nil
	}

	func (client *XdsClient) getRouterClusters(clusterName string) ([]string, error) {
	virtualHosts, err := client.RDS(clusterName)
	if err != nil {
	return nil, err
	}

	routerClusters := []string{}
	for _, h := range virtualHosts {
	for _, r := range h.Routes {
	routerClusters = append(routerClusters, r.GetRoute().GetCluster())
	}
	}

	return routerClusters, nil
	}

	func (client *XdsClient) getVersionInfo(resType XdsType) string {
	return client.ReqCaches[resType].VersionInfo
	}
	func (client *XdsClient) getNonce(resType XdsType) string {
	return client.ReqCaches[resType].Nonce
	}

	func (client *XdsClient) setVersionInfo(resType XdsType, versionInfo string) {
	client.ReqCaches[resType].VersionInfo = versionInfo
	}

	func (client *XdsClient) setNonce(resType XdsType, nonce string) {
	client.ReqCaches[resType].Nonce = nonce
	}

	//CDS s the Clsuter Discovery Service API, which fetches all the clusters from istio pilot
	func (client *XdsClient) CDS() ([]apiv2.Cluster, error) {
	adsResClient, conn, err := getAdsResClient(client)
	if err != nil {
	return nil, err
	}
	defer conn.Close()

	req := &apiv2.DiscoveryRequest{
	TypeUrl: "type.googleapis.com/envoy.api.v2.Cluster",
	VersionInfo: client.getVersionInfo(TypeCds),
	ResponseNonce: client.getNonce(TypeCds),
	}
	req.Node = &apiv2core.Node{
	// Sample taken from istio: router~172.30.77.6~istio-egressgateway-84b4d947cd-rqt45.istio-system~istio-system.svc.cluster.local-2
	// The Node.Id should be in format {nodeType}~{ipAddr}~{serviceId~{domain}, splitted by '~'
	// The format is required by pilot
	Id: client.NodeID,
	Cluster: client.NodeCluster,
	}

	if err := adsResClient.Send(req); err != nil {
	return nil, err
	}

	resp, err := adsResClient.Recv()
	if err != nil {
	return nil, err
	}

	client.setNonce(TypeCds, resp.GetNonce())
	client.setVersionInfo(TypeCds, resp.GetVersionInfo())
	resources := resp.GetResources()

	var cluster apiv2.Cluster
	clusters := []apiv2.Cluster{}
	for _, res := range resources {
	if err := proto.Unmarshal(res.GetValue(), &cluster); err != nil {
	openlogging.GetLogger().Warnf("Failed to unmarshal cluster resource: %s", err.Error())
	} else {
	clusters = append(clusters, cluster)
	}
	}
	return clusters, nil
	}

	//EDS is the Endpoint Discovery Service API, the API takes the cluster's name and return all its endpoints(which provide address and port)
	func (client XdsClient) EDS(clusterName string) (apiv2.ClusterLoadAssignment, error) {
	adsResClient, conn, err := getAdsResClient(client)
	if err != nil {
	return nil, err
	}
	defer conn.Close()

	req := &apiv2.DiscoveryRequest{
	TypeUrl: "type.googleapis.com/envoy.api.v2.ClusterLoadAssignment",
	VersionInfo: client.getVersionInfo(TypeEds),
	ResponseNonce: client.getNonce(TypeEds),
	}

	req.Node = &apiv2core.Node{
	Id: client.NodeID,
	Cluster: client.NodeCluster,
	}
	req.ResourceNames = []string{clusterName}
	if err := adsResClient.Send(req); err != nil {
	return nil, err
	}

	resp, err := adsResClient.Recv()
	if err != nil {
	return nil, err
	}

	resources := resp.GetResources()
	client.setNonce(TypeEds, resp.GetNonce())
	client.setVersionInfo(TypeEds, resp.GetVersionInfo())

	var loadAssignment apiv2.ClusterLoadAssignment
	var e error
	// endpoints := []apiv2.ClusterLoadAssignment{}

	for _, res := range resources {
	if err := proto.Unmarshal(res.GetValue(), &loadAssignment); err != nil {
	e = err
	} else {
	// The cluster's LoadAssignment will always be ONE, with Endpoints as its field
	break
	}
	}
	return &loadAssignment, e
	}

	//GetEndpointsByTags fetches the cluster's endpoints with tags. The tags is usually specified in a DestinationRule.
	func (client XdsClient) GetEndpointsByTags(serviceName string, tags map[string]string) ([]apiv2endpoint.LbEndpoint, string, error) {
	clusters, err := client.CDS()
	if err != nil {
	return nil, "", err
	}

	lbendpoints := []*apiv2endpoint.LbEndpoint{}
	clusterName := ""
	for _, cluster := range clusters {
	clusterInfo := ParseClusterName(cluster.Name)
	if clusterInfo == nil \|\| clusterInfo.Subset == "" \|\| clusterInfo.ServiceName != serviceName {
	continue
	}
	// So clusterInfo is not nil and subset is not empty
	if subsetTags, err := client.GetSubsetTags(clusterInfo.Namespace, clusterInfo.ServiceName, clusterInfo.Subset); err == nil {
	// filter with tags
	matched := true
	for k, v := range tags {
	if subsetTagValue, exists := subsetTags[k]; exists == false \|\| subsetTagValue != v {
	matched = false
	break
	}
	}

	if matched { // We got the cluster!
	clusterName = cluster.Name
	loadAssignment, err := client.EDS(cluster.Name)
	if err != nil {
	return nil, clusterName, err
	}

	for _, item := range loadAssignment.Endpoints {
	lbendpoints = append(lbendpoints, item.LbEndpoints...)
	}

	return lbendpoints, clusterName, nil
	}
	}
	}

	return lbendpoints, clusterName, nil
	}

	//RDS is the Router Discovery Service API, it returns the virtual hosts which contains Routes
	func (client XdsClient) RDS(clusterName string) ([]apiv2route.VirtualHost, error) {
	clusterInfo := ParseClusterName(clusterName)
	if clusterInfo == nil {
	return nil, fmt.Errorf("Invalid clusterName for routers: %s", clusterName)
	}

	adsResClient, conn, err := getAdsResClient(client)
	if err != nil {
	return nil, err
	}
	defer conn.Close()

	req := &apiv2.DiscoveryRequest{
	TypeUrl: "type.googleapis.com/envoy.api.v2.RouteConfiguration",
	VersionInfo: client.getVersionInfo(TypeRds),
	ResponseNonce: client.getNonce(TypeRds),
	}

	req.Node = &apiv2core.Node{
	Id: client.NodeID,
	Cluster: client.NodeCluster,
	}
	req.ResourceNames = []string{clusterName}
	if err := adsResClient.Send(req); err != nil {
	return nil, err
	}

	resp, err := adsResClient.Recv()
	if err != nil {
	return nil, err
	}

	resources := resp.GetResources()
	client.setNonce(TypeRds, resp.GetNonce())
	client.setVersionInfo(TypeRds, resp.GetVersionInfo())

	var route apiv2.RouteConfiguration
	virtualHosts := []*apiv2route.VirtualHost{}

	for _, res := range resources {
	if err := proto.Unmarshal(res.GetValue(), &route); err != nil {
	openlogging.GetLogger().Warnf("Failed to unmarshal router resource: ", err.Error())
	} else {
	vhosts := route.GetVirtualHosts()
	for _, vhost := range vhosts {
	if vhost.Name == clusterInfo.ServiceName+":"+clusterInfo.Port {
	virtualHosts = append(virtualHosts, vhost)
	}
	}
	}
	}
	return virtualHosts, nil
	}

	//LDS is the Listener Discovery Service API, which returns all the listerns
	func (client *XdsClient) LDS() ([]apiv2.Listener, error) {
	adsResClient, conn, err := getAdsResClient(client)
	if err != nil {
	return nil, err
	}
	defer conn.Close()

	req := &apiv2.DiscoveryRequest{
	TypeUrl: "type.googleapis.com/envoy.api.v2.Listener",
	VersionInfo: client.getVersionInfo(TypeLds),
	ResponseNonce: client.getNonce(TypeLds),
	}

	req.Node = &apiv2core.Node{
	Id: client.NodeID,
	Cluster: client.NodeCluster,
	}
	if err := adsResClient.Send(req); err != nil {
	return nil, err
	}

	resp, err := adsResClient.Recv()
	if err != nil {
	return nil, err
	}

	resources := resp.GetResources()
	client.setNonce(TypeLds, resp.GetNonce())
	client.setVersionInfo(TypeLds, resp.GetVersionInfo())

	var listener apiv2.Listener
	listeners := []apiv2.Listener{}

	for _, res := range resources {
	if err := proto.Unmarshal(res.GetValue(), &listener); err != nil {
	openlogging.GetLogger().Warnf("Failed to unmarshal listener resource: ", err.Error())
	} else {
	listeners = append(listeners, listener)
	}
	}
	return listeners, nil
	}

	//ParseClusterName parse the cluster's name, which is in the format direction\|port\|subset\|hostName, the 4 items will be parsed into different fields. The hostName item will also be parsed into ServcieName, Namespace etc.
	func ParseClusterName(clusterName string) *XdsClusterInfo {
	// clusterName format: direction\|port\|subset\|hostName
	// hostName format: \|svc.namespace.svc.cluster.local

	parts := strings.Split(clusterName, "\|")
	if len(parts) != 4 {
	return nil
	}

	hostnameParts := strings.Split(parts[3], ".")
	if len(hostnameParts) < 2 {
	return nil
	}

	cluster := &XdsClusterInfo{
	Direction: parts[0],
	Port: parts[1],
	Subset: parts[2],
	HostName: parts[3],
	ServiceName: hostnameParts[0],
	Namespace: hostnameParts[1],
	DomainSuffix: strings.Join(hostnameParts[2:], "."),
	ClusterName: clusterName,
	}

	return cluster
	}