Add expectations

Need to go back and look at a simpler pattern - have the index report
trigger the reaction?

Author: smarterclayton

pkg/image/controller/trigger/customtrigger.go

@@ -67,7 +67,7 @@ type BuildConfigReaction struct {
 	Instantiator BuildConfigInstantiator
 }
 
-func (r *BuildConfigReaction) ImageChanged(obj interface{}, is *imageapi.ImageStream, operations []TriggerOperation) error {
+func (r *BuildConfigReaction) ImageChanged(obj interface{}, operations []TriggerOperation) error {
 	bc := obj.(*buildapi.BuildConfig)
 	var (
 		changed bool

pkg/image/controller/trigger/expectations.go

@@ -0,0 +1,70 @@
+package trigger
+
+import "sync"
+
+// Expectations track the tags that have been requested for resync on image streams.
+// Using expectations allows fan out of trigger checks to be minimized when only a
+// few tags have changed
+type Expectations interface {
+	// Expect indicates that a change to the image stream is expected. If tags is empty,
+	// a full image stream resync is expected. If tags are specified, a resync may not
+	// be necessary.
+	Expect(imageStreamKey string, tags ...ExpectedTags)
+	// Satisfy returns the list of tags that are expected to change. If no expectations
+	// have been registered, all tags should be checked.
+	Satisfy(imageStreamKey string) []ExpectedTags
+}
+
+type ExpectedTags struct {
+	Tag        string
+	TriggerKey string
+}
+
+func newExpectations() Expectations {
+	return &expectedImageStreams{
+		streams: make(map[string][]ExpectedTags),
+		limit:   10,
+	}
+}
+
+type expectedImageStreams struct {
+	lock    sync.Mutex
+	streams map[string][]ExpectedTags
+	limit   int
+}
+
+func (e *expectedImageStreams) Expect(key string, expected ...ExpectedTags) {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+
+	if len(expected) == 0 {
+		e.streams[key] = nil
+		return
+	}
+	tags, ok := e.streams[key]
+	if !ok {
+		e.streams[key] = expected
+		return
+	}
+	if len(tags) == 0 {
+		return
+	}
+	if (len(tags) + len(expected)) > e.limit {
+		e.streams[key] = nil
+		return
+	}
+	for _, tag := range expected {
+		tags = append(tags, tag)
+	}
+	e.streams[key] = tags
+}
+
+func (e *expectedImageStreams) Satisfy(key string) []ExpectedTags {
+	e.lock.Lock()
+	defer e.lock.Unlock()
+	tags, ok := e.streams[key]
+	if ok {
+		delete(e.streams, key)
+	}
+	return tags
+}

pkg/image/controller/trigger/trigger.go

@@ -29,7 +29,7 @@ const (
 	maxRetries = 5
 )
 
-type ReactionFunc func(obj interface{}, is *imageapi.ImageStream, operations []TriggerOperation) error
+type ReactionFunc func(obj interface{}, operations []TriggerOperation) error
 
 type TriggerController struct {
 	client        kcoreclient.CoreInterface
@@ -46,8 +46,10 @@ type TriggerController struct {
 	// lister can list/get image streams from the shared informer's store
 	lister ImageStreamLister
 
-	// Deployments that need to be synced
+	// queue is the list of image stream keys that must be synced.
 	queue workqueue.RateLimitingInterface
+	// expectations indicate the set of tags that may be changing.
+	expectations Expectations
 
 	// syncs are the items that must return true before the queue can be processed
 	syncs []cache.InformerSynced
@@ -95,6 +97,7 @@ func NewTriggerController(client kcoreclient.CoreInterface, is ImageStreamInform
 		client:        client,
 		eventRecorder: eventBroadcaster.NewRecorder(api.EventSource{Component: "image-trigger-controller"}),
 		queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "image-trigger"),
+		expectations:  newExpectations(),
 	}
 
 	c.syncHandler = c.syncImageStream
@@ -146,6 +149,7 @@ func (c *TriggerController) enqueue(is *imageapi.ImageStream) {
 		utilruntime.HandleError(fmt.Errorf("Couldn't get key for object %#v: %v", is, err))
 		return
 	}
+	c.expectations.Expect(key)
 	c.queue.Add(key)
 }
 
@@ -214,6 +218,28 @@ type TriggerOperation struct {
 	Trigger   *trigger.ObjectFieldTrigger
 }
 
+type operationMap map[string][]TriggerOperation
+
+func applyTriggersToOperations(operations operationMap, entry *TriggerCacheEntry, name, tag, ref, id, namespace string) operationMap {
+	for i, trigger := range entry.Triggers {
+		if trigger.From.Name != name || trigger.From.Kind != "ImageStreamTag" || trigger.From.APIVersion != "" || trigger.Paused {
+			continue
+		}
+		if operations == nil {
+			operations = make(map[string][]TriggerOperation)
+		}
+		operations[entry.Key] = append(operations[entry.Key], TriggerOperation{
+			Name:      name,
+			Tag:       tag,
+			Ref:       ref,
+			ID:        id,
+			Namespace: namespace,
+			Trigger:   &entry.Triggers[i],
+		})
+	}
+	return operations
+}
+
 // syncImageStream will sync the image stream with the given key.
 // This function is not meant to be invoked concurrently with the same key.
 func (c *TriggerController) syncImageStream(key string) error {
@@ -223,19 +249,12 @@ func (c *TriggerController) syncImageStream(key string) error {
 		glog.V(4).Infof("Finished syncing image stream %q (%v)", key, time.Now().Sub(startTime))
 	}()
 
-	namespace, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		return err
-	}
+	tags := c.expectations.Satisfy(key)
 
-	triggered, err := c.triggerCache.ByIndex("images", key)
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
 		return err
 	}
-	if len(triggered) == 0 {
-		return nil
-	}
-
 	is, err := c.lister.ImageStreams(namespace).Get(name)
 	if errors.IsNotFound(err) {
 		glog.V(4).Infof("Image stream %v has been deleted", key)
@@ -246,38 +265,61 @@ func (c *TriggerController) syncImageStream(key string) error {
 		return err
 	}
 
-	var operations map[string][]TriggerOperation
-	for tag, v := range is.Status.Tags {
-		if len(v.Items) == 0 {
-			continue
+	// the operations we're expected to perform
+	var operations operationMap
+
+	if len(tags) > 0 {
+		// check only the provided tags against the latest value of the trigger cache
+		for _, expected := range tags {
+			tag := expected.Tag
+			v := is.Status.Tags[tag]
+			if len(v.Items) == 0 {
+				continue
+			}
+
+			// TODO: is this possibly a race?
+			item, ok := c.triggerCache.Get(expected.TriggerKey)
+			if !ok {
+				continue
+			}
+
+			ref, ok := imageapi.ResolveTagReference(is, tag, &v.Items[0])
+			if !ok {
+				continue
+			}
+			name := imageapi.JoinImageStreamTag(name, tag)
+
+			entry := item.(*TriggerCacheEntry)
+			operations = applyTriggersToOperations(operations, entry, name, tag, ref, v.Items[0].Image, namespace)
 		}
-		ref, ok := imageapi.ResolveTagReference(is, tag, &v.Items[0])
-		if !ok {
-			continue
+
+	} else {
+		// check everything against all triggers
+		triggered, err := c.triggerCache.ByIndex("images", key)
+		if err != nil {
+			return err
 		}
-		name := imageapi.JoinImageStreamTag(name, tag)
-		for _, t := range triggered {
-			entry := t.(*TriggerCacheEntry)
-			for i, trigger := range entry.Triggers {
-				if trigger.From.Name != name || trigger.From.Kind != "ImageStreamTag" || trigger.From.APIVersion != "" || trigger.Paused {
-					continue
-				}
-				if operations == nil {
-					operations = make(map[string][]TriggerOperation)
-				}
-				operations[entry.Key] = append(operations[entry.Key], TriggerOperation{
-					Name:      name,
-					Tag:       tag,
-					Ref:       ref,
-					ID:        v.Items[0].Image,
-					Namespace: namespace,
-					Trigger:   &entry.Triggers[i],
-				})
+		if len(triggered) == 0 {
+			return nil
+		}
+
+		for tag, v := range is.Status.Tags {
+			if len(v.Items) == 0 {
+				continue
+			}
+			ref, ok := imageapi.ResolveTagReference(is, tag, &v.Items[0])
+			if !ok {
+				continue
+			}
+			name := imageapi.JoinImageStreamTag(name, tag)
+			for _, t := range triggered {
+				entry := t.(*TriggerCacheEntry)
+				operations = applyTriggersToOperations(operations, entry, name, tag, ref, v.Items[0].Image, namespace)
 			}
 		}
 	}
 
-	// TODO: separate worker pool
+	// TODO: separate worker pool?
 	var failures bool
 	for key, ops := range operations {
 		parts := strings.SplitN(key, "/", 2)
@@ -292,7 +334,7 @@ func (c *TriggerController) syncImageStream(key string) error {
 			// TODO: record expectation?
 			continue
 		}
-		if err := source.ReactionFn(obj, is, ops); err != nil {
+		if err := source.ReactionFn(obj, ops); err != nil {
 			// TODO: enqueue again
 			utilruntime.HandleError(fmt.Errorf("unable to react to new image for %s %s from store: %v", parts[0], parts[1], err))
 			failures = true
@@ -305,118 +347,4 @@ func (c *TriggerController) syncImageStream(key string) error {
 	}
 
 	return nil
-
-	// // Deep-copy otherwise we are mutating our cache.
-	// // TODO: Deep-copy only when needed.
-	// d, err := util.DeploymentDeepCopy(deployment)
-	// if err != nil {
-	// 	return err
-	// }
-
-	// everything := metav1.LabelSelector{}
-	// if reflect.DeepEqual(d.Spec.Selector, &everything) {
-	// 	c.eventRecorder.Eventf(d, v1.EventTypeWarning, "SelectingAll", "This deployment is selecting all pods. A non-empty selector is required.")
-	// 	if d.Status.ObservedGeneration < d.Generation {
-	// 		d.Status.ObservedGeneration = d.Generation
-	// 		c.client.Extensions().Deployments(d.Namespace).UpdateStatus(d)
-	// 	}
-	// 	return nil
-	// }
-
-	// deployments, err := c.dLister.Deployments(d.Namespace).List(labels.Everything())
-	// if err != nil {
-	// 	return fmt.Errorf("error listing deployments in namespace %s: %v", d.Namespace, err)
-	// }
-
-	// // Handle overlapping deployments by deterministically avoid syncing deployments that fight over ReplicaSets.
-	// overlaps, err := c.handleOverlap(d, deployments)
-	// if err != nil {
-	// 	if overlaps {
-	// 		// Emit an event and return a nil error for overlapping deployments so we won't resync them again.
-	// 		c.eventRecorder.Eventf(d, v1.EventTypeWarning, "SelectorOverlap", err.Error())
-	// 		return nil
-	// 	}
-	// 	// For any other failure, we should retry the deployment.
-	// 	return err
-	// }
-
-	// if d.DeletionTimestamp != nil {
-	// 	return c.syncStatusOnly(d)
-	// }
-
-	// // Why run the cleanup policy only when there is no rollback request?
-	// // The thing with the cleanup policy currently is that it is far from smart because it takes into account
-	// // the latest replica sets while it should instead retain the latest *working* replica sets. This means that
-	// // you can have a cleanup policy of 1 but your last known working replica set may be 2 or 3 versions back
-	// // in the history.
-	// // Eventually we will want to find a way to recognize replica sets that have worked at some point in time
-	// // (and chances are higher that they will work again as opposed to others that didn't) for candidates to
-	// // automatically roll back to (#23211) and the cleanup policy should help.
-	// if d.Spec.RollbackTo == nil {
-	// 	_, oldRSs, err := c.getAllReplicaSetsAndSyncRevision(d, false)
-	// 	if err != nil {
-	// 		return err
-	// 	}
-	// 	// So far the cleanup policy was executed once a deployment was paused, scaled up/down, or it
-	// 	// succesfully completed deploying a replica set. Decouple it from the strategies and have it
-	// 	// run almost unconditionally - cleanupDeployment is safe by default.
-	// 	c.cleanupDeployment(oldRSs, d)
-	// }
-
-	// err = c.claimReplicaSets(d)
-	// if err != nil {
-	// 	return err
-	// }
-
-	// // Update deployment conditions with an Unknown condition when pausing/resuming
-	// // a deployment. In this way, we can be sure that we won't timeout when a user
-	// // resumes a Deployment with a set progressDeadlineSeconds.
-	// if err = c.checkPausedConditions(d); err != nil {
-	// 	return err
-	// }
-
-	// _, err = c.hasFailed(d)
-	// if err != nil {
-	// 	return err
-	// }
-	// // TODO: Automatically rollback here if we failed above. Locate the last complete
-	// // revision and populate the rollback spec with it.
-	// // See https://github.com/kubernetes/kubernetes/issues/23211.
-
-	// if d.Spec.Paused {
-	// 	return c.sync(d)
-	// }
-
-	// if d.Spec.RollbackTo != nil {
-	// 	revision := d.Spec.RollbackTo.Revision
-	// 	if d, err = c.rollback(d, &revision); err != nil {
-	// 		return err
-	// 	}
-	// }
-
-	// scalingEvent, err := c.isScalingEvent(d)
-	// if err != nil {
-	// 	return err
-	// }
-	// if scalingEvent {
-	// 	return c.sync(d)
-	// }
-
-	// switch d.Spec.Strategy.Type {
-	// case extensions.RecreateDeploymentStrategyType:
-	// 	return c.rolloutRecreate(d)
-	// case extensions.RollingUpdateDeploymentStrategyType:
-	// 	return c.rolloutRolling(d)
-	// }
-	// return fmt.Errorf("unexpected deployment strategy type: %s", d.Spec.Strategy.Type)
 }
-
-// func startReplicationController(ctx ControllerContext) (bool, error) {
-// 	go replicationcontroller.NewReplicationManager(
-// 		ctx.InformerFactory.Core().V1().Pods(),
-// 		ctx.InformerFactory.Core().V1().ReplicationControllers(),
-// 		ctx.ClientBuilder.ClientOrDie("replication-controller"),
-// 		replicationcontroller.BurstReplicas,
-// 	).Run(int(ctx.Options.ConcurrentRCSyncs), ctx.Stop)
-// 	return true, nil
-// }