Merge branch 'master' into 905-new-reconcilers

cmd/broker/fanout/main.go

@@ -36,8 +36,8 @@ import (
 
 	"github.com/google/knative-gcp/pkg/broker/config/volume"
 	"github.com/google/knative-gcp/pkg/broker/handler/pool"
-	"github.com/google/knative-gcp/pkg/broker/handler/pool/fanout"
 	"github.com/google/knative-gcp/pkg/observability"
+	"github.com/google/knative-gcp/pkg/utils"
 	"github.com/google/knative-gcp/pkg/utils/appcredentials"
 )
 
@@ -112,17 +112,24 @@ func main() {
 	// Give the signal channel some buffer so that reconciling handlers won't
 	// block the targets config update?
 	targetsUpdateCh := make(chan struct{})
-	targetsConifg, err := volume.NewTargetsFromFile(
-		volume.WithPath(env.TargetsConfigPath),
-		volume.WithNotifyChan(targetsUpdateCh))
-	if err != nil {
-		logger.Fatalw("Failed to load targets config", zap.String("path", env.TargetsConfigPath), zap.Error(err))
-	}
 
 	logger.Info("Starting the broker fanout")
 
+	projectID, err := utils.ProjectID(env.ProjectID)
+	if err != nil {
+		log.Fatalf("failed to get default ProjectID: %v", err)
+	}
+
 	syncSignal := poolSyncSignal(ctx, targetsUpdateCh)
-	syncPool, err := fanout.NewSyncPool(ctx, targetsConifg, buildPoolOptions(env)...)
+	syncPool, err := InitializeSyncPool(
+		ctx,
+		pool.ProjectID(projectID),
+		[]volume.Option{
+			volume.WithPath(env.TargetsConfigPath),
+			volume.WithNotifyChan(targetsUpdateCh),
+		},
+		buildPoolOptions(env)...,
+	)
 	if err != nil {
 		logger.Fatal("Failed to create fanout sync pool", zap.Error(err))
 	}
@@ -168,9 +175,6 @@ func buildPoolOptions(env envConfig) []pool.Option {
 	if env.MaxConcurrencyPerEvent > 0 {
 		opts = append(opts, pool.WithMaxConcurrentPerEvent(env.MaxConcurrencyPerEvent))
 	}
-	if env.ProjectID != "" {
-		opts = append(opts, pool.WithProjectID(env.ProjectID))
-	}
 	if env.TimeoutPerEvent > 0 {
 		opts = append(opts, pool.WithTimeoutPerEvent(env.TimeoutPerEvent))
 	}

cmd/broker/fanout/wire.go

@@ -0,0 +1,35 @@
+// +build wireinject
+
+/*
+Copyright 2020 Google LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+
+	"github.com/google/knative-gcp/pkg/broker/config/volume"
+	"github.com/google/knative-gcp/pkg/broker/handler/pool"
+	"github.com/google/wire"
+)
+
+// InitializeSyncPool initializes the fanout sync pool. Uses the given projectID to initialize the
+// retry pool's pubsub client and uses targetsVolumeOpts to initialize the targets volume watcher.
+func InitializeSyncPool(ctx context.Context, projectID pool.ProjectID, targetsVolumeOpts []volume.Option, opts ...pool.Option) (*pool.FanoutPool, error) {
+	// Implementation generated by wire. Providers for required FanoutPool dependencies should be
+	// added here.
+	panic(wire.Build(pool.ProviderSet, volume.NewTargetsFromFile))
+}

cmd/broker/ingress/main.go

@@ -21,6 +21,7 @@ import (
 	"log"
 
 	"github.com/google/knative-gcp/pkg/broker/ingress"
+	"github.com/google/knative-gcp/pkg/metrics"
 	"github.com/google/knative-gcp/pkg/observability"
 	"github.com/google/knative-gcp/pkg/utils"
 	"github.com/google/knative-gcp/pkg/utils/appcredentials"
@@ -91,8 +92,8 @@ func main() {
 		ctx,
 		ingress.Port(env.Port),
 		ingress.ProjectID(projectID),
-		ingress.PodName(env.PodName),
-		ingress.ContainerName(containerName),
+		metrics.PodName(env.PodName),
+		metrics.ContainerName(containerName),
 	)
 	if err != nil {
 		logger.Desugar().Fatal("Unable to create ingress handler: ", zap.Error(err))

cmd/broker/ingress/wire.go

@@ -23,15 +23,16 @@ import (
 
 	"github.com/google/knative-gcp/pkg/broker/config/volume"
 	"github.com/google/knative-gcp/pkg/broker/ingress"
+	"github.com/google/knative-gcp/pkg/metrics"
 	"github.com/google/wire"
 )
 
 func InitializeHandler(
 	ctx context.Context,
 	port ingress.Port,
 	projectID ingress.ProjectID,
-	podName ingress.PodName,
-	containerName ingress.ContainerName,
+	podName metrics.PodName,
+	containerName metrics.ContainerName,
 ) (*ingress.Handler, error) {
 	panic(wire.Build(
 		ingress.HandlerSet,

cmd/broker/retry/main.go

@@ -38,8 +38,8 @@ import (
 
 	"github.com/google/knative-gcp/pkg/broker/config/volume"
 	"github.com/google/knative-gcp/pkg/broker/handler/pool"
-	"github.com/google/knative-gcp/pkg/broker/handler/pool/retry"
 	"github.com/google/knative-gcp/pkg/observability"
+	"github.com/google/knative-gcp/pkg/utils"
 	"github.com/google/knative-gcp/pkg/utils/appcredentials"
 )
 
@@ -112,17 +112,23 @@ func main() {
 	// Give the signal channel some buffer so that reconciling handlers won't
 	// block the targets config update?
 	targetsUpdateCh := make(chan struct{})
-	targetsConifg, err := volume.NewTargetsFromFile(
-		volume.WithPath(env.TargetsConfigPath),
-		volume.WithNotifyChan(targetsUpdateCh))
+	logger.Info("Starting the broker retry")
+
+	projectID, err := utils.ProjectID(env.ProjectID)
 	if err != nil {
-		logger.Fatal("Failed to load targets config", zap.String("path", env.TargetsConfigPath), zap.Error(err))
+		log.Fatalf("failed to get default ProjectID: %v", err)
 	}
 
-	logger.Info("Starting the broker retry")
-
 	syncSignal := poolSyncSignal(ctx, targetsUpdateCh)
-	syncPool, err := retry.NewSyncPool(targetsConifg, buildPoolOptions(env)...)
+	syncPool, err := InitializeSyncPool(
+		ctx,
+		pool.ProjectID(projectID),
+		[]volume.Option{
+			volume.WithPath(env.TargetsConfigPath),
+			volume.WithNotifyChan(targetsUpdateCh),
+		},
+		buildPoolOptions(env)...,
+	)
 	if err != nil {
 		logger.Fatal("Failed to get retry sync pool", zap.Error(err))
 	}
@@ -167,9 +173,6 @@ func buildPoolOptions(env envConfig) []pool.Option {
 		opts = append(opts, pool.WithHandlerConcurrency(env.HandlerConcurrency))
 		rs.NumGoroutines = env.HandlerConcurrency
 	}
-	if env.ProjectID != "" {
-		opts = append(opts, pool.WithProjectID(env.ProjectID))
-	}
 	if env.TimeoutPerEvent > 0 {
 		opts = append(opts, pool.WithTimeoutPerEvent(env.TimeoutPerEvent))
 	}

cmd/broker/retry/wire.go

@@ -0,0 +1,35 @@
+// +build wireinject
+
+/*
+Copyright 2020 Google LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+
+	"github.com/google/knative-gcp/pkg/broker/config/volume"
+	"github.com/google/knative-gcp/pkg/broker/handler/pool"
+	"github.com/google/wire"
+)
+
+// InitializeSyncPool initializes the retry sync pool. Uses the given projectID to initialize the
+// retry pool's pubsub client and uses targetsVolumeOpts to initialize the targets volume watcher.
+func InitializeSyncPool(ctx context.Context, projectID pool.ProjectID, targetsVolumeOpts []volume.Option, opts ...pool.Option) (*pool.RetryPool, error) {
+	// Implementation generated by wire. Providers for required RetryPool dependencies should be
+	// added here.
+	panic(wire.Build(pool.ProviderSet, volume.NewTargetsFromFile))
+}