lookup buildconfigs in shared indexed cache instead of listing

pkg/build/controller/factory/factory.go

@@ -25,12 +25,18 @@ import (
 	strategy "github.com/openshift/origin/pkg/build/controller/strategy"
 	buildutil "github.com/openshift/origin/pkg/build/util"
 	osclient "github.com/openshift/origin/pkg/client"
+	oscache "github.com/openshift/origin/pkg/client/cache"
 	controller "github.com/openshift/origin/pkg/controller"
 	imageapi "github.com/openshift/origin/pkg/image/api"
 	errors "github.com/openshift/origin/pkg/util/errors"
 )
 
-const maxRetries = 60
+const (
+	// We must avoid creating processing imagestream changes until the build config store has synced.
+	// If it hasn't synced, to avoid a hot loop, we'll wait this long between checks.
+	StoreSyncedPollPeriod = 100 * time.Millisecond
+	maxRetries            = 60
+)
 
 // limitedLogAndRetry stops retrying after maxTimeout, failing the build.
 func limitedLogAndRetry(buildupdater buildclient.BuildUpdater, maxTimeout time.Duration) controller.RetryFunc {
@@ -274,6 +280,8 @@ func (factory *BuildPodControllerFactory) CreateDeleteController() controller.Ru
 type ImageChangeControllerFactory struct {
 	Client                  osclient.Interface
 	BuildConfigInstantiator buildclient.BuildConfigInstantiator
+	BuildConfigIndex        oscache.StoreToBuildConfigLister
+	BuildConfigIndexSynced  func() bool
 	// Stop may be set to allow controllers created by this factory to be terminated.
 	Stop <-chan struct{}
 }
@@ -284,14 +292,14 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl
 	queue := cache.NewResyncableFIFO(cache.MetaNamespaceKeyFunc)
 	cache.NewReflector(&imageStreamLW{factory.Client}, &imageapi.ImageStream{}, queue, 2*time.Minute).RunUntil(factory.Stop)
 
-	store := cache.NewStore(cache.MetaNamespaceKeyFunc)
-	cache.NewReflector(&buildConfigLW{client: factory.Client}, &buildapi.BuildConfig{}, store, 2*time.Minute).RunUntil(factory.Stop)
-
 	imageChangeController := &buildcontroller.ImageChangeController{
-		BuildConfigStore:        store,
+		BuildConfigIndex:        factory.BuildConfigIndex,
 		BuildConfigInstantiator: factory.BuildConfigInstantiator,
 	}
 
+	// Wait for the bc store to sync before starting any work in this controller.
+	factory.waitForSyncedStores()
+
 	return &controller.RetryController{
 		Queue: queue,
 		RetryManager: controller.NewQueueRetryManager(
@@ -305,11 +313,20 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl
 		),
 		Handle: func(obj interface{}) error {
 			imageRepo := obj.(*imageapi.ImageStream)
-			return imageChangeController.HandleImageRepo(imageRepo)
+			return imageChangeController.HandleImageStream(imageRepo)
 		},
 	}
 }
 
+func (factory *ImageChangeControllerFactory) waitForSyncedStores() {
+	for !factory.BuildConfigIndexSynced() {
+		glog.V(4).Infof("Waiting for the bc caches to sync before starting the imagechange buildconfig controller worker")
+		select {
+		case <-time.After(StoreSyncedPollPeriod):
+		}
+	}
+}
+
 type BuildConfigControllerFactory struct {
 	Client                  osclient.Interface
 	KubeClient              kclient.Interface

pkg/build/controller/image_change_controller.go

@@ -8,12 +8,12 @@ import (
 	kerrors "k8s.io/kubernetes/pkg/api/errors"
 
 	kapi "k8s.io/kubernetes/pkg/api"
-	"k8s.io/kubernetes/pkg/client/cache"
 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"
 
 	buildapi "github.com/openshift/origin/pkg/build/api"
 	buildclient "github.com/openshift/origin/pkg/build/client"
 	buildutil "github.com/openshift/origin/pkg/build/util"
+	oscache "github.com/openshift/origin/pkg/client/cache"
 	imageapi "github.com/openshift/origin/pkg/image/api"
 )
 
@@ -31,7 +31,7 @@ func (e ImageChangeControllerFatalError) Error() string {
 // builds when a new version of a tag referenced by a BuildConfig
 // is available.
 type ImageChangeController struct {
-	BuildConfigStore        cache.Store
+	BuildConfigIndex        oscache.StoreToBuildConfigLister
 	BuildConfigInstantiator buildclient.BuildConfigInstantiator
 }
 
@@ -42,9 +42,9 @@ func getImageStreamNameFromReference(ref *kapi.ObjectReference) string {
 	return strings.Split(name, "@")[0]
 }
 
-// HandleImageRepo processes the next ImageStream event.
-func (c *ImageChangeController) HandleImageRepo(repo *imageapi.ImageStream) error {
-	glog.V(4).Infof("Build image change controller detected ImageStream change %s", repo.Status.DockerImageRepository)
+// HandleImageStream processes the next ImageStream event.
+func (c *ImageChangeController) HandleImageStream(stream *imageapi.ImageStream) error {
+	glog.V(4).Infof("Build image change controller detected ImageStream change %s", stream.Status.DockerImageRepository)
 
 	// Loop through all build configurations and record if there was an error
 	// instead of breaking the loop. The error will be returned in the end, so the
@@ -53,17 +53,18 @@ func (c *ImageChangeController) HandleImageRepo(repo *imageapi.ImageStream) erro
 	// in a no-op for them.
 	hasError := false
 
-	// TODO: this is inefficient
-	for _, bc := range c.BuildConfigStore.List() {
-		config := bc.(*buildapi.BuildConfig)
-
+	bcs, err := c.BuildConfigIndex.GetConfigsForImageStreamTrigger(stream)
+	if err != nil {
+		return err
+	}
+	for _, config := range bcs {
 		var (
 			from           *kapi.ObjectReference
 			shouldBuild    = false
 			triggeredImage = ""
 			latest         *imageapi.TagEvent
 		)
-		// For every ImageChange trigger find the latest tagged image from the image repository and
+		// For every ImageChange trigger find the latest tagged image from the image stream and
 		// invoke a build using that image id. A new build is triggered only if the latest tagged image id or pull spec
 		// differs from the last triggered build recorded on the build config for that trigger
 		for _, trigger := range config.Spec.Triggers {
@@ -90,21 +91,21 @@ func (c *ImageChangeController) HandleImageRepo(repo *imageapi.ImageStream) erro
 				fromNamespace = config.Namespace
 			}
 
-			// only trigger a build if this image repo matches the name and namespace of the ref in the build trigger
+			// only trigger a build if this image stream matches the name and namespace of the stream ref in the build trigger
 			// also do not trigger if the imagerepo does not have a valid DockerImageRepository value for us to pull
 			// the image from
-			if len(repo.Status.DockerImageRepository) == 0 || fromStreamName != repo.Name || fromNamespace != repo.Namespace {
+			if len(stream.Status.DockerImageRepository) == 0 || fromStreamName != stream.Name || fromNamespace != stream.Namespace {
 				continue
 			}
 
 			// This split is safe because ImageStreamTag names always have the form
 			// name:tag.
-			latest = imageapi.LatestTaggedImage(repo, tag)
+			latest = imageapi.LatestTaggedImage(stream, tag)
 			if latest == nil {
-				glog.V(4).Infof("unable to find tagged image: no image recorded for %s/%s:%s", repo.Namespace, repo.Name, tag)
+				glog.V(4).Infof("unable to find tagged image: no image recorded for %s/%s:%s", stream.Namespace, stream.Name, tag)
 				continue
 			}
-			glog.V(4).Infof("Found ImageStream %s/%s with tag %s", repo.Namespace, repo.Name, tag)
+			glog.V(4).Infof("Found ImageStream %s/%s with tag %s", stream.Namespace, stream.Name, tag)
 
 			// (must be different) to trigger a build
 			last := trigger.ImageChange.LastTriggeredImageID
@@ -155,7 +156,7 @@ func (c *ImageChangeController) HandleImageRepo(repo *imageapi.ImageStream) erro
 		}
 	}
 	if hasError {
-		return fmt.Errorf("an error occurred processing 1 or more build configurations; the image change trigger for image stream %s will be retried", repo.Status.DockerImageRepository)
+		return fmt.Errorf("an error occurred processing 1 or more build configurations; the image change trigger for image stream %s will be retried", stream.Status.DockerImageRepository)
 	}
 	return nil
 }

pkg/build/controller/image_change_controller_test.go

@@ -26,9 +26,9 @@ func TestNewImageID(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Fatalf("Unexpected error %v from HandleImageRepo", err)
+		t.Fatalf("Unexpected error %v from HandleImageStream", err)
 	}
 
 	if len(bcInstantiator.name) == 0 {
@@ -54,9 +54,9 @@ func TestNewImageIDDefaultTag(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Fatalf("Unexpected error %v from HandleImageRepo", err)
+		t.Fatalf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) == 0 {
 		t.Error("Expected build generation when new image was created!")
@@ -82,9 +82,9 @@ func TestNonExistentImageStream(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Fatalf("Unexpected error %v from HandleImageRepo", err)
+		t.Fatalf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -103,9 +103,9 @@ func TestNewImageDifferentTagUpdate(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -126,9 +126,9 @@ func TestNewImageDifferentTagUpdate2(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -147,9 +147,9 @@ func TestNewDifferentImageUpdate(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -170,9 +170,9 @@ func TestSameStreamNameDifferentNamespaces(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -192,9 +192,9 @@ func TestBuildConfigWithDifferentTriggerType(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when a different repository was updated!")
@@ -215,9 +215,9 @@ func TestNoImageIDChange(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when no change happened!")
@@ -237,9 +237,9 @@ func TestBuildConfigInstantiatorError(t *testing.T) {
 	bcInstantiator.err = fmt.Errorf("instantiating error")
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err == nil || !strings.Contains(err.Error(), "will be retried") {
-		t.Fatalf("Expected 'will be retried' from HandleImageRepo, got %s", err.Error())
+		t.Fatalf("Expected 'will be retried' from HandleImageStream, got %s", err.Error())
 	}
 	if actual, expected := bcInstantiator.newBuild.Spec.Strategy.DockerStrategy.From.Name, "registry.com/namespace/imagename:newImageID123"; actual != expected {
 		t.Errorf("Image substitutions not properly setup for new build. Expected %s, got %s |", expected, actual)
@@ -259,12 +259,12 @@ func TestBuildConfigUpdateError(t *testing.T) {
 	bcUpdater := bcInstantiator.buildConfigUpdater
 	bcUpdater.err = kerrors.NewConflict(buildapi.Resource("BuildConfig"), buildcfg.Name, errors.New("foo"))
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if len(bcInstantiator.name) == 0 {
 		t.Error("Expected build generation when new image was created!")
 	}
 	if err == nil || !strings.Contains(err.Error(), "will be retried") {
-		t.Fatalf("Expected 'will be retried' from HandleImageRepo, got %s", err.Error())
+		t.Fatalf("Expected 'will be retried' from HandleImageStream, got %s", err.Error())
 	}
 }
 
@@ -277,9 +277,9 @@ func TestNewImageIDNoDockerRepo(t *testing.T) {
 	bcInstantiator := controller.BuildConfigInstantiator.(*buildConfigInstantiator)
 	bcUpdater := bcInstantiator.buildConfigUpdater
 
-	err := controller.HandleImageRepo(imageStream)
+	err := controller.HandleImageStream(imageStream)
 	if err != nil {
-		t.Errorf("Unexpected error %v from HandleImageRepo", err)
+		t.Errorf("Unexpected error %v from HandleImageStream", err)
 	}
 	if len(bcInstantiator.name) != 0 {
 		t.Error("New build generated when no change happened!")
@@ -417,7 +417,7 @@ func mockBuildConfigInstantiator(buildcfg *buildapi.BuildConfig, imageStream *im
 
 func mockImageChangeController(buildcfg *buildapi.BuildConfig, imageStream *imageapi.ImageStream, image *imageapi.Image) *ImageChangeController {
 	return &ImageChangeController{
-		BuildConfigStore:        buildtest.NewFakeBuildConfigStore(buildcfg),
+		BuildConfigIndex:        buildtest.NewFakeBuildConfigIndex(buildcfg),
 		BuildConfigInstantiator: mockBuildConfigInstantiator(buildcfg, imageStream, image),
 	}
 }

pkg/build/controller/test/fake_build_config_index.go

@@ -0,0 +1,24 @@
+package test
+
+import (
+	buildapi "github.com/openshift/origin/pkg/build/api"
+	oscache "github.com/openshift/origin/pkg/client/cache"
+	imageapi "github.com/openshift/origin/pkg/image/api"
+)
+
+type FakeBuildConfigIndex struct {
+	Build *buildapi.BuildConfig
+	Err   error
+}
+
+func NewFakeBuildConfigIndex(build *buildapi.BuildConfig) oscache.StoreToBuildConfigLister {
+	return &FakeBuildConfigIndex{Build: build}
+}
+
+func (i *FakeBuildConfigIndex) List() ([]*buildapi.BuildConfig, error) {
+	return []*buildapi.BuildConfig{i.Build}, nil
+}
+
+func (i *FakeBuildConfigIndex) GetConfigsForImageStreamTrigger(stream *imageapi.ImageStream) ([]*buildapi.BuildConfig, error) {
+	return []*buildapi.BuildConfig{i.Build}, nil
+}