feat: Network basics (#4)

* Parse ingresses - Add graph and stateful support

* Print topologies for debugging - Tag duplicated services managed - fixed deps update

* Refactoring

* Virtual service and destination rule

* Fixes

* Linting

.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

@@ -4,6 +4,7 @@ go 1.22.0
 
 require (
 	github.com/brunoga/deep v1.2.4
+	github.com/dominikbraun/graph v0.23.0
 	github.com/kurtosis-tech/stacktrace v0.0.0-20211028211901-1c67a77b5409
 	github.com/onsi/ginkgo/v2 v2.19.0
 	github.com/onsi/gomega v1.33.1
@@ -89,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

@@ -18,6 +18,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dominikbraun/graph v0.23.0 h1:TdZB4pPqCLFxYhdyMFb1TBdFxp8XLcJfTTBQucVPgCo=
+github.com/dominikbraun/graph v0.23.0/go.mod h1:yOjYyogZLY1LSG9E33JWZJiq5k83Qy2C6POAuiViluc=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanphx/json-patch v0.5.2 h1:xVCHIVMUu1wtM/VkR9jVZ45N3FhZfYMMYGorLCR8P3k=
@@ -231,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.

internal/controller/core/suite_test.go

@@ -26,6 +26,7 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
+	istioclient "istio.io/client-go/pkg/apis/networking/v1alpha3"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -79,6 +80,8 @@ var _ = BeforeSuite(func() {
 
 	err = corev1.AddToScheme(scheme.Scheme)
 	Expect(err).NotTo(HaveOccurred())
+	err = istioclient.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
 
 	// +kubebuilder:scaffold:scheme
 

kardinal/reconciler/reconciler.go

@@ -3,6 +3,7 @@ package reconciler
 import (
 	"context"
 
+	"github.com/brunoga/deep"
 	"github.com/kurtosis-tech/stacktrace"
 	"github.com/sirupsen/logrus"
 	"kardinal.dev/kardinal-operator/kardinal/resources"
@@ -21,39 +22,71 @@ func Reconcile(ctx context.Context, cl client.Client) error {
 	}
 	// Generate base cluster topology
 	logrus.Info("Generate base cluster topology")
-	version := "baseline"
-	baseClusterTopology, err := topology.NewClusterTopologyFromResources(clusterResources, version)
+	baseClusterTopology, err := topology.NewClusterTopologyFromResources(clusterResources)
 	if err != nil {
 		return stacktrace.Propagate(err, "An error occurred generating the base cluster topology")
 	}
+	baseClusterTopology.Print()
 
-	// Update base cluster topology with flows
+	// Generate flow topologies
+	flowTopologies := []*topology.ClusterTopology{}
 	for _, namespace := range clusterResources.Namespaces {
 		for _, flow := range namespace.Flows {
 			logrus.Infof("Processing flow %s", flow.Name)
-			service, err := baseClusterTopology.GetService(flow.Spec.Service, namespace.Name)
-			if err != nil {
+
+			flowTopology := baseClusterTopology.Copy(flow.Name)
+			clusterGraph := flowTopology.GetGraph()
+
+			service := baseClusterTopology.GetServiceByName(namespace.Name, flow.Spec.Service)
+			if service == nil {
 				return stacktrace.Propagate(err, "An error occurred retrieving base cluster topology service %s in namespace %s", flow.Spec.Service, namespace.Name)
 			}
 			deployment := resources.GetDeploymentFromName(service.ServiceID, namespace.Deployments)
-			deployment.Spec.Template.Spec.Containers[0].Image = flow.Spec.Image
-			patch := &topology.ServicePatch{
-				Namespace:      namespace.Name,
-				Service:        flow.Spec.Service,
-				DeploymentSpec: &deployment.Spec,
+			deploymentCopy := deep.MustCopy(deployment)
+			deploymentCopy.Spec.Template.Spec.Containers[0].Image = flow.Spec.Image
+			err = flowTopology.UpdateWithFlow(clusterGraph, flow.Name, service, &deploymentCopy.Spec)
+			if err != nil {
+				return stacktrace.NewError("An error occurred updating the base cluster topology with flow %s", flow.Name)
 			}
-			patches := []*topology.ServicePatch{patch}
-			flowPatch := &topology.FlowPatch{
-				FlowId:         flow.GetObjectMeta().GetName(),
-				ServicePatches: patches,
+
+			// Replace "baseline" version services with baseClusterTopology versions
+			for idx, service := range flowTopology.Services {
+				if !service.IsManaged {
+					baseService := baseClusterTopology.GetServiceByName(service.Namespace, service.ServiceID)
+					if baseService == nil {
+						return stacktrace.NewError("An error occurred retrieving the baseline service %s", service.ServiceID)
+					}
+					flowTopology.Services[idx] = baseService
+				}
 			}
-			err = baseClusterTopology.UpdateWithFlow(flowPatch)
-			if err != nil {
-				return stacktrace.Propagate(err, "An error occurred updating the base cluster topology with flow %s", flowPatch.FlowId)
+
+			// Update service dependencies
+			for idx, dependency := range flowTopology.ServiceDependencies {
+				if !dependency.Service.IsManaged {
+					baseService := baseClusterTopology.GetServiceByName(dependency.Service.Namespace, dependency.Service.ServiceID)
+					if baseService == nil {
+						return stacktrace.NewError("An error occurred retrieving the baseline service %s for dependency %s", service.ServiceID, dependency.Service.ServiceID)
+					}
+					flowTopology.ServiceDependencies[idx].Service = baseService
+				}
+				if !dependency.DependsOnService.IsManaged {
+					baseDependsOnService := baseClusterTopology.GetServiceByName(dependency.DependsOnService.Namespace, dependency.DependsOnService.ServiceID)
+					if baseDependsOnService == nil {
+						return stacktrace.NewError("An error occurred retrieving the baseline service %s for depends on", dependency.DependsOnService.ServiceID)
+					}
+					flowTopology.ServiceDependencies[idx].DependsOnService = baseDependsOnService
+				}
 			}
+
+			flowTopology.Print()
+			flowTopologies = append(flowTopologies, flowTopology)
 		}
 	}
 
+	// Merge flow topologies with base topology
+	baseClusterTopology = baseClusterTopology.Merge(flowTopologies)
+	baseClusterTopology.Print()
+
 	// Reconcile
 	err = baseClusterTopology.ApplyResources(ctx, clusterResources, cl)
 	if err != nil {

kardinal/resources/namespace.go

@@ -1,16 +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"`
-	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 {
@@ -32,3 +37,33 @@ 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
+}
+
+func (namespace *Namespace) GetIngress(name string) *net.Ingress {
+	for _, ingress := range namespace.Ingresses {
+		if ingress.Name == name {
+			return ingress
+		}
+	}
+
+	return nil
+}