Code cleanup in trigger and broker controller (#1792)

* Code cleanup in trigger and broker controller

Remove duplicate metadataclient query in both broker and trigger
controllers. Use single ProjectIDEnvConfig for parsing project
id from env, to avoid inconsistency in the future.

* Avoid calling NewClient when project id is not resolved

* Resolve comments

pkg/reconciler/broker/controller.go

@@ -18,9 +18,9 @@ package broker
 
 import (
 	"context"
-	"os"
 
 	"github.com/google/knative-gcp/pkg/apis/configs/dataresidency"
+	"github.com/kelseyhightower/envconfig"
 
 	"cloud.google.com/go/pubsub"
 	"go.uber.org/zap"
@@ -50,6 +50,10 @@ const (
 	controllerAgentName = "broker-controller"
 )
 
+type envConfig struct {
+	utils.ProjectIDEnvConfig
+}
+
 type Constructor injection.ControllerConstructor
 
 // NewConstructor creates a constructor to make a Broker controller.
@@ -60,21 +64,29 @@ func NewConstructor(dataresidencyss *dataresidency.StoreSingleton) Constructor {
 }
 
 func newController(ctx context.Context, cmw configmap.Watcher, drs *dataresidency.Store) *controller.Impl {
+	// Parse all env vars of interest
+	var env envConfig
+	if err := envconfig.Process("", &env); err != nil {
+		logging.FromContext(ctx).Fatal("Failed to process env var", zap.Error(err))
+	}
+
 	brokerInformer := brokerinformer.Get(ctx)
 	bcInformer := brokercellinformer.Get(ctx)
 
+	var client *pubsub.Client
 	// If there is an error, the projectID will be empty. The reconciler will retry
 	// to get the projectID during reconciliation.
-	projectID, err := utils.ProjectID(os.Getenv(utils.ProjectIDEnvKey), metadataClient.NewDefaultMetadataClient())
+	projectID, err := utils.ProjectID(env.ProjectID, metadataClient.NewDefaultMetadataClient())
 	if err != nil {
 		logging.FromContext(ctx).Error("Failed to get project ID", zap.Error(err))
-	}
-	// Attempt to create a pubsub client for all worker threads to use. If this
-	// fails, pass a nil value to the Reconciler. They will attempt to
-	// create a client on reconcile.
-	client, err := newPubsubClient(ctx, projectID)
-	if err != nil {
-		logging.FromContext(ctx).Error("Failed to create controller-wide Pub/Sub client", zap.Error(err))
+	} else {
+		// Attempt to create a pubsub client for all worker threads to use. If this
+		// fails, pass a nil value to the Reconciler. They will attempt to
+		// create a client on reconcile.
+		if client, err = pubsub.NewClient(ctx, projectID); err != nil {
+			client = nil
+			logging.FromContext(ctx).Error("Failed to create controller-wide Pub/Sub client", zap.Error(err))
+		}
 	}
 
 	if client != nil {
@@ -124,16 +136,3 @@ func newController(ctx context.Context, cmw configmap.Watcher, drs *dataresidenc
 
 	return impl
 }
-
-func newPubsubClient(ctx context.Context, projectID string) (*pubsub.Client, error) {
-	projectID, err := utils.ProjectID(projectID, metadataClient.NewDefaultMetadataClient())
-	if err != nil {
-		return nil, err
-	}
-
-	client, err := pubsub.NewClient(ctx, projectID)
-	if err != nil {
-		return nil, err
-	}
-	return client, nil
-}

pkg/reconciler/trigger/controller.go

@@ -60,7 +60,7 @@ const (
 var filterBroker = pkgreconciler.AnnotationFilterFunc(eventingv1beta1.BrokerClassAnnotationKey, brokerv1beta1.BrokerClass, false /*allowUnset*/)
 
 type envConfig struct {
-	ProjectID string `envconfig:"PROJECT_ID"`
+	utils.ProjectIDEnvConfig
 }
 
 type Constructor injection.ControllerConstructor
@@ -73,26 +73,28 @@ func NewConstructor(dataresidencyss *dataresidency.StoreSingleton) Constructor {
 }
 
 func newController(ctx context.Context, cmw configmap.Watcher, drs *dataresidency.Store) *controller.Impl {
+	// Parse all env vars of interest
 	var env envConfig
 	if err := envconfig.Process("", &env); err != nil {
 		logging.FromContext(ctx).Fatal("Failed to process env var", zap.Error(err))
 	}
 
 	triggerInformer := triggerinformer.Get(ctx)
 
+	var client *pubsub.Client
 	// If there is an error, the projectID will be empty. The reconciler will retry
 	// to get the projectID during reconciliation.
 	projectID, err := utils.ProjectID(env.ProjectID, metadataClient.NewDefaultMetadataClient())
 	if err != nil {
 		logging.FromContext(ctx).Error("Failed to get project ID", zap.Error(err))
-	}
-
-	// Attempt to create a pubsub client for all worker threads to use. If this
-	// fails, pass a nil value to the Reconciler. They will attempt to
-	// create a client on reconcile.
-	client, err := newPubsubClient(ctx, projectID)
-	if err != nil {
-		logging.FromContext(ctx).Error("Failed to create controller-wide Pub/Sub client", zap.Error(err))
+	} else {
+		// Attempt to create a pubsub client for all worker threads to use. If this
+		// fails, pass a nil value to the Reconciler. They will attempt to
+		// create a client on reconcile.
+		if client, err = pubsub.NewClient(ctx, projectID); err != nil {
+			client = nil
+			logging.FromContext(ctx).Error("Failed to create controller-wide Pub/Sub client", zap.Error(err))
+		}
 	}
 
 	if client != nil {
@@ -141,19 +143,6 @@ func newController(ctx context.Context, cmw configmap.Watcher, drs *dataresidenc
 	return impl
 }
 
-func newPubsubClient(ctx context.Context, projectID string) (*pubsub.Client, error) {
-	projectID, err := utils.ProjectID(projectID, metadataClient.NewDefaultMetadataClient())
-	if err != nil {
-		return nil, err
-	}
-
-	client, err := pubsub.NewClient(ctx, projectID)
-	if err != nil {
-		return nil, err
-	}
-	return client, nil
-}
-
 func withAgentAndFinalizer(impl *pkgcontroller.Impl) pkgcontroller.Options {
 	return pkgcontroller.Options{
 		FinalizerName: finalizerName,

pkg/utils/metadata.go

@@ -25,6 +25,11 @@ const (
 	ProjectIDEnvKey = "PROJECT_ID"
 )
 
+// ProjectIDEnvConfig is a struct to parse project ID from env var
+type ProjectIDEnvConfig struct {
+	ProjectID string `envconfig:"PROJECT_ID"`
+}
+
 // ProjectID returns the project ID for a particular resource.
 func ProjectID(project string, client metadataClient.Client) (string, error) {
 	// If project is set, then return that one.