Create grpc connection only on invocation of the CRUD events for the installation controller

controllers/generate.go

@@ -1,3 +1,4 @@
 package controllers
 
 //go:generate mockery --name=PorterClient --filename=grpc_mocks.go --outpkg=mocks --output=../mocks/grpc
+//go:generate mockery --name=ClientConn --filename=clientconn_mocks.go --outpkg=mocks --output=../mocks/grpc

controllers/installation_controller.go

@@ -9,8 +9,11 @@ import (
 
 	v1 "get.porter.sh/operator/api/v1"
 	installationv1 "get.porter.sh/porter/gen/proto/go/porterapis/installation/v1alpha1"
+	porterv1alpha1 "get.porter.sh/porter/gen/proto/go/porterapis/porter/v1alpha1"
 	"github.com/go-logr/logr"
 	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,9 +35,9 @@ const (
 type InstallationReconciler struct {
 	client.Client
 	Log              logr.Logger
-	PorterGRPCClient PorterClient
 	Recorder         record.EventRecorder
 	Scheme           *runtime.Scheme
+	CreateGRPCClient func(ctx context.Context) (porterv1alpha1.PorterClient, ClientConn, error)
 }
 
 // +kubebuilder:rbac:groups=getporter.org,resources=agentconfigs,verbs=get;list;watch;create;update;patch;delete
@@ -121,10 +124,8 @@ func (r *InstallationReconciler) Reconcile(ctx context.Context, req ctrl.Request
 
 		// Nothing for us to do at this point
 		log.V(Log4Debug).Info("Reconciliation complete: A porter agent has already been dispatched.")
-		if r.PorterGRPCClient != nil {
-			return r.CheckOrCreateInstallationOutputsCR(ctx, log, inst)
-		}
-		return ctrl.Result{}, nil
+		log.V(Log4Debug).Info(fmt.Sprintf("performing installation outputs for %s", inst.Name))
+		return r.CheckOrCreateInstallationOutputsCR(ctx, log, inst)
 	}
 
 	// Should we uninstall the bundle?
@@ -165,21 +166,26 @@ func (r *InstallationReconciler) Reconcile(ctx context.Context, req ctrl.Request
 	}
 
 	log.V(Log4Debug).Info("Reconciliation complete: A porter agent has been dispatched to apply changes to the installation.")
-	if r.PorterGRPCClient != nil {
-		return r.CheckOrCreateInstallationOutputsCR(ctx, log, inst)
-	}
-	return ctrl.Result{}, nil
+	log.V(Log4Debug).Info(fmt.Sprintf("performing installation outputs for %s", inst.Name))
+	return r.CheckOrCreateInstallationOutputsCR(ctx, log, inst)
 }
 
 func (r *InstallationReconciler) CheckOrCreateInstallationOutputsCR(ctx context.Context, log logr.Logger, inst *v1.Installation) (ctrl.Result, error) {
 	// NOTE: May not want to requeue if this fails
+	porterGRPCClient, conn, err := r.CreateGRPCClient(ctx)
+	if err != nil {
+		log.V(Log4Debug).Info("no grpc client... Not performing installation outputs")
+		r.Recorder.Event(inst, "Warning", "CreateInstallationOutputs", "no grpc client not creating installation outputs")
+		return ctrl.Result{}, nil
+	}
+	defer conn.Close()
 	installCr := &v1.InstallationOutput{}
-	err := r.Get(ctx, types.NamespacedName{Name: inst.Spec.Name, Namespace: inst.Spec.Namespace}, installCr)
+	err = r.Get(ctx, types.NamespacedName{Name: inst.Spec.Name, Namespace: inst.Spec.Namespace}, installCr)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			log.V(Log4Debug).Info("installation output cr doesn't exist, seeing if we should create")
 			in := &installationv1.ListInstallationLatestOutputRequest{Name: inst.Spec.Name, Namespace: ptr.To(inst.Spec.Namespace)}
-			resp, err := r.PorterGRPCClient.ListInstallationLatestOutputs(ctx, in)
+			resp, err := porterGRPCClient.ListInstallationLatestOutputs(ctx, in)
 			if err != nil {
 				log.V(Log4Debug).Info(fmt.Sprintf("failed to get output from grpc server for: %s:%s installation error: %s", inst.Spec.Name, inst.Spec.Namespace, err.Error()))
 				// NOTE: Stop installation output cr creation
@@ -436,3 +442,16 @@ func (r *InstallationReconciler) applyDeletionPolicy(ctx context.Context, log lo
 	inst.SetAnnotations(annotations)
 	return r.Update(ctx, inst)
 }
+
+func CreatePorterGRPCClient(ctx context.Context) (porterv1alpha1.PorterClient, ClientConn, error) {
+	// TODO: Make this not a hard coded value of grpc deployment/service.
+	// Have a controller create deployment of grpc server and service
+	conn, err := grpc.DialContext(ctx, "porter-grpc-service:3001", grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		return nil, nil, fmt.Errorf("error setting up listener for porter grpc client")
+	}
+	if conn != nil {
+		return porterv1alpha1.NewPorterClient(conn), conn, nil
+	}
+	return nil, nil, fmt.Errorf("error creating porter grpc client")
+}

controllers/installation_controller_test.go

@@ -9,6 +9,7 @@ import (
 	v1 "get.porter.sh/operator/api/v1"
 	mocks "get.porter.sh/operator/mocks/grpc"
 	installationv1 "get.porter.sh/porter/gen/proto/go/porterapis/installation/v1alpha1"
+	porterv1alpha1 "get.porter.sh/porter/gen/proto/go/porterapis/porter/v1alpha1"
 	"github.com/go-logr/logr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -109,6 +110,11 @@ func TestInstallationReconciler_Reconcile(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name, Generation: 1}}
 
 	controller := setupInstallationController(testdata)
+	clientConn := &mocks.ClientConn{}
+	clientConn.On("Close").Return(nil)
+	controller.CreateGRPCClient = func(ctx context.Context) (porterv1alpha1.PorterClient, ClientConn, error) {
+		return nil, clientConn, fmt.Errorf("this is not needed for this test")
+	}
 
 	var inst v1.Installation
 	triggerReconcile := func() {
@@ -416,7 +422,6 @@ func TestCheckOrCreateInstallationOutputsCRCreate(t *testing.T) {
 	listInstallationRequest := &installationv1.ListInstallationLatestOutputRequest{Name: "fake-install", Namespace: ptr.To("fake-ns")}
 	grpcClient.On("ListInstallationLatestOutputs", ctx, listInstallationRequest).Return(outputs, nil)
 	rec := setupInstallationController()
-	rec.PorterGRPCClient = grpcClient
 	install := &v1.Installation{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "fake-install",
@@ -427,6 +432,11 @@ func TestCheckOrCreateInstallationOutputsCRCreate(t *testing.T) {
 			Namespace: "fake-ns",
 		},
 	}
+	clientConn := &mocks.ClientConn{}
+	clientConn.On("Close").Return(nil)
+	rec.CreateGRPCClient = func(ctx context.Context) (porterv1alpha1.PorterClient, ClientConn, error) {
+		return grpcClient, clientConn, nil
+	}
 	_, err := rec.CheckOrCreateInstallationOutputsCR(ctx, logr.Discard(), install)
 	// NOTE: This errors because of the limitation we have with fake in
 	// controller-runtime. https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.15.0/pkg/client/fake
@@ -444,7 +454,6 @@ func TestCheckOrCreateInstallationOutputsCRCreateFail(t *testing.T) {
 	listInstallationRequest := &installationv1.ListInstallationLatestOutputRequest{Name: "fake-install", Namespace: ptr.To("fake-ns")}
 	grpcClient.On("ListInstallationLatestOutputs", ctx, listInstallationRequest).Return(nil, fmt.Errorf("this is an error"))
 	rec := setupInstallationController()
-	rec.PorterGRPCClient = grpcClient
 	install := &v1.Installation{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "fake-install",
@@ -483,11 +492,16 @@ func setupInstallationController(objs ...client.Object) *InstallationReconciler
 	fakeBuilder.WithObjects(objs...).WithStatusSubresource(objs...)
 	fakeClient := fakeBuilder.Build()
 
+	clientConn := &mocks.ClientConn{}
+	clientConn.On("Close").Return(nil)
 	return &InstallationReconciler{
 		Log:      logr.Discard(),
 		Client:   fakeClient,
 		Recorder: record.NewFakeRecorder(42),
 		Scheme:   scheme,
+		CreateGRPCClient: func(ctx context.Context) (porterv1alpha1.PorterClient, ClientConn, error) {
+			return &mocks.PorterClient{}, clientConn, fmt.Errorf("error with grpc")
+		},
 	}
 }
 

controllers/types.go

@@ -11,3 +11,7 @@ type PorterClient interface {
 	ListInstallations(ctx context.Context, in *installationv1.ListInstallationsRequest, opts ...grpc.CallOption) (*installationv1.ListInstallationsResponse, error)
 	ListInstallationLatestOutputs(ctx context.Context, in *installationv1.ListInstallationLatestOutputRequest, opts ...grpc.CallOption) (*installationv1.ListInstallationLatestOutputResponse, error)
 }
+
+type ClientConn interface {
+	Close() error
+}

main.go

@@ -1,14 +1,12 @@
 package main
 
 import (
-	"context"
 	"flag"
 	"os"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials/insecure"
+
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -21,7 +19,6 @@ import (
 
 	v1 "get.porter.sh/operator/api/v1"
 	"get.porter.sh/operator/controllers"
-	porterv1alpha1 "get.porter.sh/porter/gen/proto/go/porterapis/porter/v1alpha1"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -66,22 +63,12 @@ func main() {
 		setupLog.Error(err, "unable to start manager")
 		os.Exit(1)
 	}
-	// NOTE: Pass in nil client if connection isn't established
-	var client controllers.PorterClient
-	conn, err := grpc.DialContext(context.Background(), "porter-grpc-service:3001", grpc.WithTransportCredentials(insecure.NewCredentials()))
-	if err != nil {
-		setupLog.Error(err, "unable to set up listener")
-	}
-	defer conn.Close()
-	if conn != nil {
-		client = porterv1alpha1.NewPorterClient(conn)
-	}
 	if err = (&controllers.InstallationReconciler{
 		Client:           mgr.GetClient(),
-		PorterGRPCClient: client,
 		Recorder:         mgr.GetEventRecorderFor("installation"),
 		Log:              ctrl.Log.WithName("controllers").WithName("Installation"),
 		Scheme:           mgr.GetScheme(),
+		CreateGRPCClient: controllers.CreatePorterGRPCClient,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Installation")
 		os.Exit(1)