Report ingress and service to support cross cluster visibility (#220)

Author: NetanelBollag

src/go.mod

@@ -1,3 +1,3 @@
 package istiowatcher
 
-//go:generate go run github.com/golang/mock/mockgen@v1.6.0 -source=watcher.go -destination=mocks/mocks.go
+//go:generate go run go.uber.org/mock/mockgen@v0.2.0 -source=watcher.go -destination=mocks/mocks.go

src/go.sum

@@ -3,10 +3,10 @@ package istiowatcher
 import (
 	"context"
 	"fmt"
-	"github.com/golang/mock/gomock"
 	mock_istiowatcher "github.com/otterize/network-mapper/src/istio-watcher/pkg/watcher/mocks"
 	"github.com/otterize/network-mapper/src/mapper/pkg/graph/model"
 	"github.com/stretchr/testify/suite"
+	"go.uber.org/mock/gomock"
 	"testing"
 	"time"
 )

src/istio-watcher/pkg/watcher/generate.go

@@ -1,4 +1,4 @@
 package mapperclient
 
-//go:generate go run github.com/golang/mock/mockgen@v1.6.0 -source=client.go -destination=mockclient/mocks.go
+//go:generate go run go.uber.org/mock/mockgen@v0.2.0 -source=client.go -destination=mockclient/mocks.go
 //go:generate go run github.com/Khan/genqlient@v0.7.0 ./genqlient.yaml

src/istio-watcher/pkg/watcher/mocks/mocks.go

@@ -20,6 +20,7 @@ import (
 	"github.com/otterize/network-mapper/src/mapper/pkg/incomingtrafficholder"
 	"github.com/otterize/network-mapper/src/mapper/pkg/mapperwebhooks"
 	"github.com/otterize/network-mapper/src/mapper/pkg/pod_webhook"
+	"github.com/otterize/network-mapper/src/mapper/pkg/resourcevisibility"
 	"github.com/otterize/network-mapper/src/shared/echologrus"
 	"golang.org/x/sync/errgroup"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
@@ -247,6 +248,16 @@ func main() {
 		awsIntentsHolder.RegisterNotifyIntents(cloudUploader.NotifyAWSIntents)
 
 		go cloudUploader.PeriodicStatusReport(errGroupCtx)
+
+		ingressReconciler := resourcevisibility.NewIngressReconciler(mgr.GetClient(), cloudClient)
+		if err := ingressReconciler.SetupWithManager(mgr); err != nil {
+			logrus.WithError(err).Panic("unable to create ingress reconciler")
+		}
+
+		serviceReconciler := resourcevisibility.NewServiceReconciler(mgr.GetClient(), cloudClient, kubeFinder)
+		if err := serviceReconciler.SetupWithManager(mgr); err != nil {
+			logrus.WithError(err).Panic("unable to create service reconciler")
+		}
 	}
 
 	if viper.GetBool(config.OTelEnabledKey) {

src/istio-watcher/pkg/watcher/watcher_test.go

@@ -3,3 +3,4 @@ package mapper
 //go:generate go run github.com/99designs/gqlgen@v0.17.44
 //go:generate ./fix-errors-import.sh
 //go:generate go run go.uber.org/mock/mockgen@v0.2.0 -destination=./pkg/mocks/mock_k8s_client.go -package=mocks -mock_names Client=K8sClient,SubResourceWriter=K8sStatus sigs.k8s.io/controller-runtime/pkg/client Client,SubResourceWriter
+//go:generate go run go.uber.org/mock/mockgen@v0.2.0 -destination=./pkg/mocks/mock_kubefinder.go -package=mocks -source=./pkg/resourcevisibility/svc_reconciler.go KubeFinder

src/kafka-watcher/pkg/mapperclient/generate.go

@@ -13,6 +13,8 @@ type CloudClient interface {
 	ReportComponentStatus(ctx context.Context, component ComponentType) error
 	ReportExternalTrafficDiscoveredIntents(ctx context.Context, intents []ExternalTrafficDiscoveredIntentInput) error
 	ReportIncomingTrafficDiscoveredIntents(ctx context.Context, intents []IncomingTrafficDiscoveredIntentInput) error
+	ReportK8sServices(ctx context.Context, namespace string, services []K8sServiceInput) error
+	ReportK8sIngresses(ctx context.Context, namespace string, ingresses []K8sIngressInput) error
 }
 
 type CloudClientImpl struct {
@@ -34,7 +36,6 @@ func (c *CloudClientImpl) ReportDiscoveredIntents(ctx context.Context, intents [
 	logrus.Debug("Uploading intents to cloud, count: ", len(intents))
 
 	_, err := ReportDiscoveredIntents(ctx, c.client, intents)
-
 	if err != nil {
 		return errors.Wrap(err)
 	}
@@ -60,6 +61,7 @@ func (c *CloudClientImpl) ReportComponentStatus(ctx context.Context, component C
 	if err != nil {
 		return errors.Wrap(err)
 	}
+
 	return nil
 }
 
@@ -73,3 +75,25 @@ func (c *CloudClientImpl) ReportIncomingTrafficDiscoveredIntents(ctx context.Con
 
 	return nil
 }
+
+func (c *CloudClientImpl) ReportK8sServices(ctx context.Context, namespace string, services []K8sServiceInput) error {
+	logrus.Debug("Uploading k8s services to cloud, count: ", len(services))
+
+	_, err := ReportK8sServices(ctx, c.client, namespace, services)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	return nil
+}
+
+func (c *CloudClientImpl) ReportK8sIngresses(ctx context.Context, namespace string, ingresses []K8sIngressInput) error {
+	logrus.Debug("Uploading k8s ingresses to cloud, count: ", len(ingresses))
+
+	_, err := ReportK8sIngresses(ctx, c.client, namespace, ingresses)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	return nil
+}