Virtual service and destination rule

.golangci.yml

@@ -14,6 +14,7 @@ issues:
         - lll
     - path: "kardinal/*"
       linters:
+        - dupl
         - lll
     - path: "internal/*"
       linters:

cmd/main.go

@@ -25,6 +25,7 @@ import (
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
+	istioclient "istio.io/client-go/pkg/apis/networking/v1alpha3"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -48,6 +49,7 @@ var (
 func init() {
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	utilruntime.Must(corev1.AddToScheme(scheme))
+	utilruntime.Must(istioclient.AddToScheme(scheme))
 	// +kubebuilder:scaffold:scheme
 }
 

config/rbac/role.yaml

@@ -8,7 +8,9 @@ rules:
   - ""
   resources:
   - deployments
+  - destinationrules
   - services
+  - virtualservices
   verbs:
   - create
   - delete

go.mod

@@ -90,6 +90,8 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	istio.io/api v1.23.2 // indirect
+	istio.io/client-go v1.23.2 // indirect
 	k8s.io/apiextensions-apiserver v0.31.0 // indirect
 	k8s.io/apiserver v0.31.0 // indirect
 	k8s.io/component-base v0.31.0 // indirect

go.sum

@@ -233,6 +233,10 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+istio.io/api v1.23.2 h1:FvWi7GC+rWD60/ZFPuulX/h3k+f2Q9qot3dP8CIL8Ss=
+istio.io/api v1.23.2/go.mod h1:QPSTGXuIQdnZFEm3myf9NZ5uBMwCdJWUvfj9ZZ+2oBM=
+istio.io/client-go v1.23.2 h1:BIt6A+KaUOFin3SzXiDq2Fr/TMBev1+c836R0BfUfhU=
+istio.io/client-go v1.23.2/go.mod h1:E08wpMtUulJk2tlWOCUVakjy1bKFxUNm22tM1R1QY0Y=
 k8s.io/api v0.31.0 h1:b9LiSjR2ym/SzTOlfMHm1tr7/21aD7fSkqgD/CVJBCo=
 k8s.io/api v0.31.0/go.mod h1:0YiFF+JfFxMM6+1hQei8FY8M7s1Mth+z/q7eF1aJkTE=
 k8s.io/apiextensions-apiserver v0.31.0 h1:fZgCVhGwsclj3qCw1buVXCV6khjRzKC5eCFt24kyLSk=

internal/controller/core/flow_controller.go

@@ -40,6 +40,8 @@ type FlowReconciler struct {
 // +kubebuilder:rbac:groups=core.kardinal.dev,resources=flows/finalizers,verbs=update
 // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=core,resources=deployments,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=core,resources=virtualservices,verbs=get;list;watch;create;update;patch;delete
+// +kubebuilder:rbac:groups=core,resources=destinationrules,verbs=get;list;watch;create;update;patch;delete
 
 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.

kardinal/reconciler/reconciler.go

@@ -49,10 +49,9 @@ func Reconcile(ctx context.Context, cl client.Client) error {
 				return stacktrace.Propagate(err, "An error occurred updating the base cluster topology with flow %s", flow.Name)
 			}
 
-			baselineFlowVersion := namespace.Name
 			// Replace "baseline" version services with baseClusterTopology versions
 			for idx, service := range flowTopology.Services {
-				if service.Version == baselineFlowVersion {
+				if !service.IsManaged {
 					baseService, err := baseClusterTopology.GetService(service.ServiceID, service.Namespace)
 					if err != nil {
 						return stacktrace.Propagate(err, "An error occurred retrieving the baseline service %s", service.ServiceID)
@@ -63,14 +62,14 @@ func Reconcile(ctx context.Context, cl client.Client) error {
 
 			// Update service dependencies
 			for idx, dependency := range flowTopology.ServiceDependencies {
-				if dependency.Service.Version == baselineFlowVersion {
+				if !dependency.Service.IsManaged {
 					baseService, err := baseClusterTopology.GetService(dependency.Service.ServiceID, dependency.Service.Namespace)
 					if err != nil {
 						return stacktrace.Propagate(err, "An error occurred retrieving the baseline service %s for dependency %s", service.ServiceID, dependency.Service.ServiceID)
 					}
 					flowTopology.ServiceDependencies[idx].Service = baseService
 				}
-				if dependency.DependsOnService.Version == baselineFlowVersion {
+				if !dependency.DependsOnService.IsManaged {
 					baseDependsOnService, err := baseClusterTopology.GetService(dependency.DependsOnService.ServiceID, dependency.DependsOnService.Namespace)
 					if err != nil {
 						return stacktrace.Propagate(err, "An error occurred retrieving the baseline service %s for depends on", dependency.DependsOnService.ServiceID)
@@ -85,7 +84,7 @@ func Reconcile(ctx context.Context, cl client.Client) error {
 	}
 
 	// Merge flow topologies with base topology
-	baseClusterTopology.Merge(flowTopologies)
+	baseClusterTopology = baseClusterTopology.Merge(flowTopologies)
 	baseClusterTopology.Print()
 
 	// Reconcile

kardinal/resources/namespace.go

@@ -1,18 +1,21 @@
 package resources
 
 import (
+	istioclient "istio.io/client-go/pkg/apis/networking/v1alpha3"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	net "k8s.io/api/networking/v1"
 	kardinalcorev1 "kardinal.dev/kardinal-operator/api/core/v1"
 )
 
 type Namespace struct {
-	Name        string
-	Services    []*corev1.Service      `json:"services"`
-	Deployments []*appsv1.Deployment   `json:"deployments"`
-	Ingresses   []*net.Ingress         `json:"ingresses"`
-	Flows       []*kardinalcorev1.Flow `json:"flows"`
+	Name             string
+	Services         []*corev1.Service    `json:"services"`
+	Deployments      []*appsv1.Deployment `json:"deployments"`
+	Ingresses        []*net.Ingress       `json:"ingresses"`
+	VirtualServices  []*istioclient.VirtualService
+	DestinationRules []*istioclient.DestinationRule
+	Flows            []*kardinalcorev1.Flow `json:"flows"`
 }
 
 func (namespace *Namespace) GetService(name string) *corev1.Service {
@@ -34,3 +37,23 @@ func (namespace *Namespace) GetDeployment(name string) *appsv1.Deployment {
 
 	return nil
 }
+
+func (namespace *Namespace) GetVirtualService(name string) *istioclient.VirtualService {
+	for _, virtualService := range namespace.VirtualServices {
+		if virtualService.Name == name {
+			return virtualService
+		}
+	}
+
+	return nil
+}
+
+func (namespace *Namespace) GetDestinationRule(name string) *istioclient.DestinationRule {
+	for _, destinationRule := range namespace.DestinationRules {
+		if destinationRule.Name == name {
+			return destinationRule
+		}
+	}
+
+	return nil
+}

kardinal/resources/resources.go

@@ -2,11 +2,13 @@ package resources
 
 import (
 	"context"
+	"reflect"
 	"strings"
 
 	"github.com/kurtosis-tech/stacktrace"
 	"github.com/samber/lo"
 	"github.com/sirupsen/logrus"
+	istioclient "istio.io/client-go/pkg/apis/networking/v1alpha3"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	net "k8s.io/api/networking/v1"
@@ -16,8 +18,9 @@ import (
 )
 
 const (
-	BaselineNamespace = "baseline"
-	trueStr           = "true"
+	BaselineNamespace       = "baseline"
+	trueStr                 = "true"
+	kardinalManagedLabelKey = "kardinal-managed"
 )
 
 type Resources struct {
@@ -80,18 +83,32 @@ func getNamespaceResources(ctx context.Context, namespace string, cl client.Clie
 		return nil, stacktrace.Propagate(err, "An error occurred retrieving the list of ingresses for namespace %s", namespace)
 	}
 
+	virtualServices := &istioclient.VirtualServiceList{}
+	err = cl.List(ctx, virtualServices, client.InNamespace(namespace))
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "An error occurred retrieving the list of virtual services for namespace %s", namespace)
+	}
+
+	destinationRules := &istioclient.DestinationRuleList{}
+	err = cl.List(ctx, destinationRules, client.InNamespace(namespace))
+	if err != nil {
+		return nil, stacktrace.Propagate(err, "An error occurred retrieving the list of destination rules for namespace %s", namespace)
+	}
+
 	flows := &kardinalcorev1.FlowList{}
 	err = cl.List(ctx, flows, client.InNamespace(namespace))
 	if err != nil {
 		return nil, stacktrace.Propagate(err, "An error occurred retrieving the list of flows for namespace %s", namespace)
 	}
 
 	return &Namespace{
-		Name:        namespace,
-		Services:    lo.Map(services.Items, func(service corev1.Service, _ int) *corev1.Service { return &service }),
-		Deployments: lo.Map(deployments.Items, func(deployment appsv1.Deployment, _ int) *appsv1.Deployment { return &deployment }),
-		Ingresses:   lo.Map(ingresses.Items, func(ingress net.Ingress, _ int) *net.Ingress { return &ingress }),
-		Flows:       lo.Map(flows.Items, func(flow kardinalcorev1.Flow, _ int) *kardinalcorev1.Flow { return &flow }),
+		Name:             namespace,
+		Services:         lo.Map(services.Items, func(service corev1.Service, _ int) *corev1.Service { return &service }),
+		Deployments:      lo.Map(deployments.Items, func(deployment appsv1.Deployment, _ int) *appsv1.Deployment { return &deployment }),
+		Ingresses:        lo.Map(ingresses.Items, func(ingress net.Ingress, _ int) *net.Ingress { return &ingress }),
+		VirtualServices:  virtualServices.Items,
+		DestinationRules: destinationRules.Items,
+		Flows:            lo.Map(flows.Items, func(flow kardinalcorev1.Flow, _ int) *kardinalcorev1.Flow { return &flow }),
 	}, nil
 }
 
@@ -132,25 +149,39 @@ func AddAnnotations(obj *metav1.ObjectMeta, annotations map[string]string) {
 	}
 }
 
-// TODO: mode thos apply helper functions to the resources package
+// TODO: Add create, update and delete global options
+
 func ApplyServiceResources(ctx context.Context, clusterResources *Resources, clusterTopologyResources *Resources, cl client.Client) error {
 	for _, namespace := range clusterResources.Namespaces {
 		clusterTopologyNamespace := clusterTopologyResources.GetNamespaceByName(namespace.Name)
 		if clusterTopologyNamespace != nil {
 			for _, service := range clusterTopologyNamespace.Services {
-				if namespace.GetService(service.Name) == nil {
+				namespaceService := namespace.GetService(service.Name)
+				if namespaceService == nil {
 					logrus.Infof("Creating service %s", service.Name)
 					err := cl.Create(ctx, service)
 					if err != nil {
 						return stacktrace.Propagate(err, "An error occurred creating service %s", service.Name)
 					}
+				} else {
+					serviceLabels := service.Labels
+					isManaged, found := serviceLabels[kardinalManagedLabelKey]
+					if found && isManaged == trueStr {
+						if !reflect.DeepEqual(namespaceService.Spec, service.Spec) {
+							service.ResourceVersion = namespaceService.ResourceVersion
+							err := cl.Update(ctx, service)
+							if err != nil {
+								return stacktrace.Propagate(err, "An error occurred updating service %s", service.Name)
+							}
+						}
+					}
 				}
 			}
 		}
 
 		for _, service := range namespace.Services {
-			serviceAnnotations := service.Annotations
-			isManaged, found := serviceAnnotations["kardinal.dev/managed"]
+			serviceLabels := service.Labels
+			isManaged, found := serviceLabels[kardinalManagedLabelKey]
 			if found && isManaged == trueStr {
 				if clusterTopologyNamespace == nil || clusterTopologyNamespace.GetService(service.Name) == nil {
 					logrus.Infof("Deleting service %s", service.Name)
@@ -171,19 +202,32 @@ func ApplyDeploymentResources(ctx context.Context, clusterResources *Resources,
 		clusterTopologyNamespace := clusterTopologyResources.GetNamespaceByName(namespace.Name)
 		if clusterTopologyNamespace != nil {
 			for _, deployment := range clusterTopologyNamespace.Deployments {
-				if namespace.GetDeployment(deployment.Name) == nil {
+				namespaceDeployment := namespace.GetDeployment(deployment.Name)
+				if namespaceDeployment == nil {
 					logrus.Infof("Creating deployment %s", deployment.Name)
 					err := cl.Create(ctx, deployment)
 					if err != nil {
 						return stacktrace.Propagate(err, "An error occurred creating deployment %s", deployment.Name)
 					}
+				} else {
+					deploymentLabels := deployment.Labels
+					isManaged, found := deploymentLabels[kardinalManagedLabelKey]
+					if found && isManaged == trueStr {
+						if !reflect.DeepEqual(namespaceDeployment.Spec, deployment.Spec) {
+							deployment.ResourceVersion = namespaceDeployment.ResourceVersion
+							err := cl.Update(ctx, deployment)
+							if err != nil {
+								return stacktrace.Propagate(err, "An error occurred updating deployment %s", deployment.Name)
+							}
+						}
+					}
 				}
 			}
 		}
 
 		for _, deployment := range namespace.Deployments {
-			deploymentAnnotations := deployment.Annotations
-			isManaged, found := deploymentAnnotations["kardinal.dev/managed"]
+			deploymentLabels := deployment.Labels
+			isManaged, found := deploymentLabels[kardinalManagedLabelKey]
 			if found && isManaged == trueStr {
 				if clusterTopologyNamespace == nil || clusterTopologyNamespace.GetDeployment(deployment.Name) == nil {
 					logrus.Infof("Deleting deployment %s", deployment.Name)
@@ -192,7 +236,8 @@ func ApplyDeploymentResources(ctx context.Context, clusterResources *Resources,
 						return stacktrace.Propagate(err, "An error occurred deleting deployment %s", deployment.Name)
 					}
 				}
-			} else {
+			}
+			/* else {
 				annotationsToAdd := map[string]string{
 					"sidecar.istio.io/inject": "true",
 					// TODO: make this a flag to help debugging
@@ -204,6 +249,82 @@ func ApplyDeploymentResources(ctx context.Context, clusterResources *Resources,
 				if err != nil {
 					return stacktrace.Propagate(err, "An error occurred updating deployment %s", deployment.Name)
 				}
+			} */
+		}
+	}
+
+	return nil
+}
+
+func ApplyVirtualServiceResources(ctx context.Context, clusterResources *Resources, clusterTopologyResources *Resources, cl client.Client) error {
+	for _, namespace := range clusterResources.Namespaces {
+		clusterTopologyNamespace := clusterTopologyResources.GetNamespaceByName(namespace.Name)
+		if clusterTopologyNamespace != nil {
+			for _, virtualService := range clusterTopologyNamespace.VirtualServices {
+				namespaceVirtualService := namespace.GetVirtualService(virtualService.Name)
+				if namespaceVirtualService == nil {
+					logrus.Infof("Creating virtual service %s", virtualService.Name)
+					err := cl.Create(ctx, virtualService)
+					if err != nil {
+						return stacktrace.Propagate(err, "An error occurred creating virtual service %s", virtualService.Name)
+					}
+				} else {
+					if !reflect.DeepEqual(&namespaceVirtualService.Spec, &virtualService.Spec) {
+						virtualService.ResourceVersion = namespaceVirtualService.ResourceVersion
+						err := cl.Update(ctx, virtualService)
+						if err != nil {
+							return stacktrace.Propagate(err, "An error occurred updating virtual service %s", virtualService.Name)
+						}
+					}
+				}
+			}
+		}
+
+		for _, virtualService := range namespace.VirtualServices {
+			if clusterTopologyNamespace == nil || clusterTopologyNamespace.GetVirtualService(virtualService.Name) == nil {
+				logrus.Infof("Deleting virtual service %s", virtualService.Name)
+				err := cl.Delete(ctx, virtualService)
+				if err != nil {
+					return stacktrace.Propagate(err, "An error occurred deleting virtual service %s", virtualService.Name)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func ApplyDestinationRuleResources(ctx context.Context, clusterResources *Resources, clusterTopologyResources *Resources, cl client.Client) error {
+	for _, namespace := range clusterResources.Namespaces {
+		clusterTopologyNamespace := clusterTopologyResources.GetNamespaceByName(namespace.Name)
+		if clusterTopologyNamespace != nil {
+			for _, destinationRule := range clusterTopologyNamespace.DestinationRules {
+				namespaceDestinationRule := namespace.GetDestinationRule(destinationRule.Name)
+				if namespaceDestinationRule == nil {
+					logrus.Infof("Creating destination rule %s", destinationRule.Name)
+					err := cl.Create(ctx, destinationRule)
+					if err != nil {
+						return stacktrace.Propagate(err, "An error occurred creating destination rule %s", destinationRule.Name)
+					}
+				} else {
+					if !reflect.DeepEqual(&namespaceDestinationRule.Spec, &destinationRule.Spec) {
+						destinationRule.ResourceVersion = namespaceDestinationRule.ResourceVersion
+						err := cl.Update(ctx, destinationRule)
+						if err != nil {
+							return stacktrace.Propagate(err, "An error occurred updating destination rule %s", destinationRule.Name)
+						}
+					}
+				}
+			}
+		}
+
+		for _, destinationRule := range namespace.DestinationRules {
+			if clusterTopologyNamespace == nil || clusterTopologyNamespace.GetDestinationRule(destinationRule.Name) == nil {
+				logrus.Infof("Deleting destination rule %s", destinationRule.Name)
+				err := cl.Delete(ctx, destinationRule)
+				if err != nil {
+					return stacktrace.Propagate(err, "An error occurred deleting destination rule %s", destinationRule.Name)
+				}
 			}
 		}
 	}