mfordjody commented on code in PR #839:
URL: https://github.com/apache/dubbo-go-pixiu/pull/839#discussion_r2616475791
##########
controllers/internal/controller/gateway_controller.go:
##########
@@ -735,3 +949,203 @@ func (r *GatewayReconciler) listGatewaysForService(ctx
context.Context, obj clie
return nil
}
+
+func (r *GatewayReconciler) resolveClusterEndpoints(ctx context.Context,
namespace string, cluster *converter.Cluster, clusterConfig
*v1alpha1.ClusterConfig) error {
+ if len(clusterConfig.Endpoints) == 0 {
+ return nil
+ }
+
+ resolvedEndpoints := []*converter.Endpoint{}
+ for _, epConfig := range clusterConfig.Endpoints {
+ if !isIPAddress(epConfig.Address) {
+ serviceName, serviceNamespace :=
parseServiceAddress(epConfig.Address, namespace)
+
+ clusterIP, svcPort, err :=
r.resolveServiceClusterIP(ctx, serviceNamespace, serviceName, epConfig.Port)
+ if err == nil && clusterIP != "" {
+ resolvedEndpoints = append(resolvedEndpoints,
&converter.Endpoint{
+ ID: func() int {
+ if epConfig.ID != nil {
+ return int(*epConfig.ID)
+ }
+ return len(resolvedEndpoints) +
1
+ }(),
+ SocketAddress: converter.SocketAddress{
+ Address: clusterIP,
+ Port: int(svcPort),
+ },
+ })
+ continue
+ }
+
+ resolvedEndpoints = append(resolvedEndpoints,
&converter.Endpoint{
+ ID: func() int {
+ if epConfig.ID != nil {
+ return int(*epConfig.ID)
+ }
+ return len(resolvedEndpoints) + 1
+ }(),
+ SocketAddress: converter.SocketAddress{
+ Address: epConfig.Address,
+ Port: int(epConfig.Port),
+ },
+ })
+ continue
+ }
+
+ resolvedEndpoints = append(resolvedEndpoints,
&converter.Endpoint{
+ ID: func() int {
+ if epConfig.ID != nil {
+ return int(*epConfig.ID)
+ }
+ return len(resolvedEndpoints) + 1
+ }(),
+ SocketAddress: converter.SocketAddress{
+ Address: epConfig.Address,
+ Port: int(epConfig.Port),
+ },
+ })
+ }
+
+ if len(resolvedEndpoints) > 0 {
+ cluster.Endpoints = resolvedEndpoints
+ }
+ return nil
+}
+
+func (r *GatewayReconciler) resolveServiceEndpoints(ctx context.Context,
namespace, serviceName string, port int32) ([]*ir.Endpoint, error) {
+ var service corev1.Service
+ if err := r.Client.Get(ctx, client.ObjectKey{
+ Namespace: namespace,
+ Name: serviceName,
+ }, &service); err != nil {
+ return nil, fmt.Errorf("failed to get service %s/%s: %w",
namespace, serviceName, err)
+ }
+
+ var endpointSliceList discoveryv1.EndpointSliceList
+ if err := r.Client.List(ctx, &endpointSliceList, client.MatchingLabels{
+ discoveryv1.LabelServiceName: serviceName,
+ }, client.InNamespace(namespace)); err != nil {
+ return nil, fmt.Errorf("failed to list endpoint slices: %w",
err)
+ }
+
+ endpoints := []*ir.Endpoint{}
+ for _, endpointSlice := range endpointSliceList.Items {
+ for _, endpoint := range endpointSlice.Endpoints {
+ if endpoint.Conditions.Ready != nil &&
!*endpoint.Conditions.Ready {
+ continue
+ }
+ for _, address := range endpoint.Addresses {
+ targetPort := port
+ if len(endpointSlice.Ports) > 0 {
+ for _, endpointPort := range
endpointSlice.Ports {
+ if endpointPort.Port != nil &&
endpointPort.Name != nil {
+ targetPort =
int32(*endpointPort.Port)
+ break
+ } else if endpointPort.Port !=
nil {
+ targetPort =
int32(*endpointPort.Port)
+ break
Review Comment:
done
##########
controllers/internal/controller/gateway_controller.go:
##########
@@ -397,13 +482,91 @@ func (r *GatewayReconciler) ensureGatewayConfigMap(ctx
context.Context, gateway
pixiuConfig, err := conv.ConvertIRToPixiuConfig(xds, allClusters)
if err != nil {
- return fmt.Errorf("failed to convert IR to Pixiu config: %w",
err)
+ return "", fmt.Errorf("failed to convert IR to Pixiu config:
%w", err)
}
+ policyLoader := NewPolicyLoader(r.Client, r.Log)
+ gatewayPolicy, err := policyLoader.LoadGatewayPolicy(ctx, gateway)
+ if err != nil {
+ r.Log.Error(err, "failed to load gateway policy", "gateway",
gateway.Name)
+ } else if gatewayPolicy != nil {
+ converter.ApplyGatewayPolicy(pixiuConfig, gatewayPolicy)
+ }
+
+ clusterPolicies, err := policyLoader.LoadAllClusterPolicies(ctx,
gateway.Namespace)
+ if err != nil {
+ r.Log.Error(err, "failed to load cluster policies")
+ } else {
+ clusterConfigMap := make(map[string]*v1alpha1.ClusterConfig)
+ for i := range clusterPolicies {
+ policy := &clusterPolicies[i]
+ if policy.Spec.TargetRef.Kind != "" &&
policy.Spec.TargetRef.Kind == "Gateway" && string(policy.Spec.TargetRef.Name)
== gateway.Name {
+ for j := range policy.Spec.ClusterRef {
+ clusterConfig :=
&policy.Spec.ClusterRef[j]
+ clusterConfigMap[clusterConfig.Name] =
clusterConfig
+ }
+ }
+ }
+
+ serviceConfigMap :=
make(map[string]*v1alpha1.ServiceClusterConfig)
+ for i := range clusterPolicies {
+ policy := &clusterPolicies[i]
+ if policy.Spec.TargetRef.Kind == "" ||
policy.Spec.TargetRef.Kind != "Gateway" || string(policy.Spec.TargetRef.Name)
!= gateway.Name {
+ for j := range policy.Spec.ServiceRef {
+ serviceConfig :=
&policy.Spec.ServiceRef[j]
+ serviceConfigMap[serviceConfig.Name] =
serviceConfig
+ }
+ }
+ }
Review Comment:
done
##########
controllers/internal/controller/gateway_controller.go:
##########
@@ -397,13 +482,91 @@ func (r *GatewayReconciler) ensureGatewayConfigMap(ctx
context.Context, gateway
pixiuConfig, err := conv.ConvertIRToPixiuConfig(xds, allClusters)
if err != nil {
- return fmt.Errorf("failed to convert IR to Pixiu config: %w",
err)
+ return "", fmt.Errorf("failed to convert IR to Pixiu config:
%w", err)
}
+ policyLoader := NewPolicyLoader(r.Client, r.Log)
+ gatewayPolicy, err := policyLoader.LoadGatewayPolicy(ctx, gateway)
+ if err != nil {
+ r.Log.Error(err, "failed to load gateway policy", "gateway",
gateway.Name)
+ } else if gatewayPolicy != nil {
+ converter.ApplyGatewayPolicy(pixiuConfig, gatewayPolicy)
+ }
+
+ clusterPolicies, err := policyLoader.LoadAllClusterPolicies(ctx,
gateway.Namespace)
+ if err != nil {
+ r.Log.Error(err, "failed to load cluster policies")
+ } else {
+ clusterConfigMap := make(map[string]*v1alpha1.ClusterConfig)
+ for i := range clusterPolicies {
+ policy := &clusterPolicies[i]
+ if policy.Spec.TargetRef.Kind != "" &&
policy.Spec.TargetRef.Kind == "Gateway" && string(policy.Spec.TargetRef.Name)
== gateway.Name {
+ for j := range policy.Spec.ClusterRef {
+ clusterConfig :=
&policy.Spec.ClusterRef[j]
+ clusterConfigMap[clusterConfig.Name] =
clusterConfig
+ }
+ }
+ }
+
+ serviceConfigMap :=
make(map[string]*v1alpha1.ServiceClusterConfig)
+ for i := range clusterPolicies {
+ policy := &clusterPolicies[i]
+ if policy.Spec.TargetRef.Kind == "" ||
policy.Spec.TargetRef.Kind != "Gateway" || string(policy.Spec.TargetRef.Name)
!= gateway.Name {
+ for j := range policy.Spec.ServiceRef {
+ serviceConfig :=
&policy.Spec.ServiceRef[j]
+ serviceConfigMap[serviceConfig.Name] =
serviceConfig
+ }
+ }
+ }
+
+ for i := range pixiuConfig.StaticResources.Clusters {
+ cluster := pixiuConfig.StaticResources.Clusters[i]
+ if clusterConfig, found :=
clusterConfigMap[cluster.Name]; found {
+ if err := r.resolveClusterEndpoints(ctx,
gateway.Namespace, cluster, clusterConfig); err != nil {
+ r.Log.Error(err, "failed to resolve
cluster endpoints", "cluster", cluster.Name)
+ }
+ converter.ApplyClusterConfig(cluster,
clusterConfig)
+ } else if serviceConfig, found :=
serviceConfigMap[cluster.Name]; found {
+ converter.ApplyClusterPolicy(cluster,
serviceConfig)
+ } else {
+ parts := splitClusterName(cluster.Name)
+ if len(parts) == 2 {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
