add nodeSelector and replicas to eventlistener

docs/eventlisteners.md

@@ -16,6 +16,7 @@ using [Event Interceptors](#Interceptors).
 
 - [Syntax](#syntax)
   - [ServiceAccountName](#serviceAccountName)
+  - [Replicas](#replicas)
   - [PodTemplate](#podTemplate)
   - [Triggers](#triggers)
     - [Interceptors](#interceptors)
@@ -45,6 +46,7 @@ the following fields:
 - Optional:
   - [`serviceType`](#serviceType) - Specifies what type of service the sink pod
     is exposed as
+  - [`replicas`](#replicas) - Specifies the number of EventListener pods
   - [`podTemplate`](#podTemplate) - Specifies the PodTemplate
     for your EventListener pod
 
@@ -150,16 +152,24 @@ documentations for details.
 For external services to connect to your cluster (e.g. GitHub sending webhooks),
 check out the guide on [exposing EventListeners](./exposing-eventlisteners.md).
 
-## PodTemplate
+### Replicas
+
+The `replicas` field is optional. By default, the number of replicas of EventListener is 1.
+If you want to deploy more than one pod, you can specify the number to this field. 
+
+### PodTemplate
 
 The `podTemplate` field is optional. A PodTemplate is specifications for 
 creating EventListener pod. A PodTemplate consists of:
 - `tolerations` - list of toleration which allows pods to schedule onto the nodes with matching taints.
 This is needed only if you want to schedule EventListener pod to a tainted node.
+- `nodeSelector` - key-value labels the node has which an EventListener pod should be scheduled on. 
 
 ```yaml
 spec:
   podTemplate:
+    nodeSelector:
+      app: test
     tolerations:
     - key: key
       value: value

examples/eventlisteners/eventlistener-podtemplate.yaml

@@ -0,0 +1,17 @@
+---
+apiVersion: triggers.tekton.dev/v1alpha1
+kind: EventListener
+metadata:
+  name: listener-podtemplate
+spec:
+  serviceAccountName: tekton-triggers-example-sa
+  podTemplate:
+    nodeSelector:
+      app: test
+  triggers:
+    - name: foo-trig
+      bindings:
+        - ref: pipeline-binding
+        - ref: message-binding
+      template:
+        name: pipeline-template

examples/eventlisteners/eventlistener-replicas.yaml

@@ -0,0 +1,15 @@
+---
+apiVersion: triggers.tekton.dev/v1alpha1
+kind: EventListener
+metadata:
+  name: listener-replicas
+spec:
+  serviceAccountName: tekton-triggers-example-sa
+  replicas: 3
+  triggers:
+    - name: foo-trig
+      bindings:
+        - ref: pipeline-binding
+        - ref: message-binding
+      template:
+        name: pipeline-template

pkg/apis/triggers/v1alpha1/event_listener_types.go

@@ -57,11 +57,20 @@ type EventListenerSpec struct {
 	ServiceAccountName string                 `json:"serviceAccountName"`
 	Triggers           []EventListenerTrigger `json:"triggers"`
 	ServiceType        corev1.ServiceType     `json:"serviceType,omitempty"`
+	Replicas           int32                  `json:"replicas"`
 	PodTemplate        PodTemplate            `json:"podTemplate,omitempty"`
 }
 
 type PodTemplate struct {
+	// If specified, the pod's tolerations.
+	// +optional
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
+
+	// NodeSelector is a selector which must be true for the pod to fit on a node.
+	// Selector which must match a node's labels for the pod to be scheduled on that node.
+	// More info: https://kubernetes.io/docs/concepts/configuration/assign-pod-node/
+	// +optional
+	NodeSelector map[string]string `json:"nodeSelector,omitempty"`
 }
 
 // EventListenerTrigger represents a connection between TriggerBinding, Params,

pkg/reconciler/v1alpha1/eventlistener/eventlistener.go

@@ -260,6 +260,9 @@ func (c *Reconciler) reconcileDeployment(el *v1alpha1.EventListener) error {
 
 	labels := mergeLabels(el.Labels, GenerateResourceLabels(el.Name))
 	var replicas int32 = 1
+	if el.Spec.Replicas != 0 {
+		replicas = el.Spec.Replicas
+	}
 	container := corev1.Container{
 		Name:  "event-listener",
 		Image: *elImage,
@@ -320,6 +323,7 @@ func (c *Reconciler) reconcileDeployment(el *v1alpha1.EventListener) error {
 				},
 				Spec: corev1.PodSpec{
 					Tolerations:        el.Spec.PodTemplate.Tolerations,
+					NodeSelector:       el.Spec.PodTemplate.NodeSelector,
 					ServiceAccountName: el.Spec.ServiceAccountName,
 					Containers:         []corev1.Container{container},
 
@@ -363,6 +367,10 @@ func (c *Reconciler) reconcileDeployment(el *v1alpha1.EventListener) error {
 			existingDeployment.Spec.Template.Spec.Tolerations = deployment.Spec.Template.Spec.Tolerations
 			updated = true
 		}
+		if !reflect.DeepEqual(existingDeployment.Spec.Template.Spec.NodeSelector, deployment.Spec.Template.Spec.NodeSelector) {
+			existingDeployment.Spec.Template.Spec.NodeSelector = deployment.Spec.Template.Spec.NodeSelector
+			updated = true
+		}
 		if len(existingDeployment.Spec.Template.Spec.Containers) == 0 ||
 			len(existingDeployment.Spec.Template.Spec.Containers) > 1 {
 			existingDeployment.Spec.Template.Spec.Containers = []corev1.Container{container}

pkg/reconciler/v1alpha1/eventlistener/eventlistener_test.go

@@ -92,6 +92,7 @@ var (
 			Effect:   "NoSchedule",
 		},
 	}
+	updateNodeSelector           = map[string]string{"app": "test"}
 	deploymentAvailableCondition = appsv1.DeploymentCondition{
 		Type:    appsv1.DeploymentAvailable,
 		Status:  corev1.ConditionTrue,
@@ -282,6 +283,9 @@ func Test_reconcileDeployment(t *testing.T) {
 	eventListener5 := eventListener1.DeepCopy()
 	eventListener5.Spec.PodTemplate.Tolerations = updateTolerations
 
+	eventListener6 := eventListener1.DeepCopy()
+	eventListener6.Spec.PodTemplate.NodeSelector = updateNodeSelector
+
 	var replicas int32 = 1
 	// deployment1 == initial deployment
 	deployment1 := &appsv1.Deployment{
@@ -393,6 +397,9 @@ func Test_reconcileDeployment(t *testing.T) {
 	deployment5 := deployment1.DeepCopy()
 	deployment5.Spec.Template.Spec.Tolerations = updateTolerations
 
+	deployment6 := deployment1.DeepCopy()
+	deployment6.Spec.Template.Spec.NodeSelector = updateNodeSelector
+
 	deploymentMissingVolumes := deployment1.DeepCopy()
 	deploymentMissingVolumes.Spec.Template.Spec.Volumes = nil
 	deploymentMissingVolumes.Spec.Template.Spec.Containers[0].VolumeMounts = nil
@@ -492,6 +499,19 @@ func Test_reconcileDeployment(t *testing.T) {
 				Deployments:    []*appsv1.Deployment{deployment5},
 			},
 		},
+		{
+			name: "eventlistener-nodeSelector-update",
+			startResources: test.Resources{
+				Namespaces:     []*corev1.Namespace{namespaceResource},
+				EventListeners: []*v1alpha1.EventListener{eventListener6},
+				Deployments:    []*appsv1.Deployment{deployment1},
+			},
+			endResources: test.Resources{
+				Namespaces:     []*corev1.Namespace{namespaceResource},
+				EventListeners: []*v1alpha1.EventListener{eventListener6},
+				Deployments:    []*appsv1.Deployment{deployment6},
+			},
+		},
 		{
 			name: "eventlistener-config-volume-mount-update",
 			startResources: test.Resources{
@@ -592,6 +612,9 @@ func TestReconcile(t *testing.T) {
 	eventListener5 := eventListener2.DeepCopy()
 	eventListener5.Spec.PodTemplate.Tolerations = updateTolerations
 
+	eventListener6 := eventListener2.DeepCopy()
+	eventListener6.Spec.PodTemplate.NodeSelector = updateNodeSelector
+
 	var replicas int32 = 1
 	deployment1 := &appsv1.Deployment{
 		ObjectMeta: generateObjectMeta(eventListener0),
@@ -606,6 +629,7 @@ func TestReconcile(t *testing.T) {
 				},
 				Spec: corev1.PodSpec{
 					Tolerations:        eventListener0.Spec.PodTemplate.Tolerations,
+					NodeSelector:       eventListener0.Spec.PodTemplate.NodeSelector,
 					ServiceAccountName: eventListener0.Spec.ServiceAccountName,
 					Containers: []corev1.Container{{
 						Name:  "event-listener",
@@ -686,6 +710,9 @@ func TestReconcile(t *testing.T) {
 	deployment4 := deployment2.DeepCopy()
 	deployment4.Spec.Template.Spec.Tolerations = updateTolerations
 
+	deployment5 := deployment2.DeepCopy()
+	deployment5.Spec.Template.Spec.NodeSelector = updateNodeSelector
+
 	service1 := &corev1.Service{
 		ObjectMeta: generateObjectMeta(eventListener0),
 		Spec: corev1.ServiceSpec{
@@ -787,6 +814,22 @@ func TestReconcile(t *testing.T) {
 			Services:       []*corev1.Service{service2},
 			ConfigMaps:     []*corev1.ConfigMap{loggingConfigMap},
 		},
+	}, {
+		name: "update-eventlistener-nodeSelector",
+		key:  reconcileKey,
+		startResources: test.Resources{
+			Namespaces:     []*corev1.Namespace{namespaceResource},
+			EventListeners: []*v1alpha1.EventListener{eventListener6},
+			Deployments:    []*appsv1.Deployment{deployment2},
+			Services:       []*corev1.Service{service2},
+		},
+		endResources: test.Resources{
+			Namespaces:     []*corev1.Namespace{namespaceResource},
+			EventListeners: []*v1alpha1.EventListener{eventListener6},
+			Deployments:    []*appsv1.Deployment{deployment5},
+			Services:       []*corev1.Service{service2},
+			ConfigMaps:     []*corev1.ConfigMap{loggingConfigMap},
+		},
 	}, {
 		name: "update-eventlistener-servicetype",
 		key:  reconcileKey,

test/builder/eventlistener.go

@@ -92,6 +92,13 @@ func EventListenerServiceAccount(saName string) EventListenerSpecOp {
 	}
 }
 
+// EventListenerReplicas sets the specified Replicas of the EventListener.
+func EventListenerReplicas(replicas int32) EventListenerSpecOp {
+	return func(spec *v1alpha1.EventListenerSpec) {
+		spec.Replicas = replicas
+	}
+}
+
 // EventListenerPodTemplate sets the specified pod template of the EventListener.
 func EventListenerPodTemplate(podTemplate v1alpha1.PodTemplate) EventListenerSpecOp {
 	return func(spec *v1alpha1.EventListenerSpec) {
@@ -117,6 +124,13 @@ func EventListenerPodTemplateTolerations(tolerations []corev1.Toleration) EventL
 	}
 }
 
+// EventListenerPodTemplateNodeSelector sets the specified NodeSelector of the EventListener PodTemplate.
+func EventListenerPodTemplateNodeSelector(nodeSelector map[string]string) EventListenerPodTemplateOp {
+	return func(pt *v1alpha1.PodTemplate) {
+		pt.NodeSelector = nodeSelector
+	}
+}
+
 // EventListenerTrigger adds an EventListenerTrigger to the EventListenerSpec Triggers.
 // Any number of EventListenerTriggerOp modifiers can be passed to create/modify it.
 func EventListenerTrigger(ttName, apiVersion string, ops ...EventListenerTriggerOp) EventListenerSpecOp {

test/eventlistener_scale_test.go

@@ -48,8 +48,10 @@ func TestEventListenerScale(t *testing.T) {
 	var err error
 	el := bldr.EventListener("my-eventlistener", namespace, bldr.EventListenerSpec(
 		bldr.EventListenerServiceAccount(saName),
+		bldr.EventListenerReplicas(3),
 		bldr.EventListenerPodTemplate(
 			bldr.EventListenerPodTemplateSpec(
+				bldr.EventListenerPodTemplateNodeSelector(map[string]string{"beta.kubernetes.io/arch": "amd64"}),
 				bldr.EventListenerPodTemplateTolerations([]corev1.Toleration{
 					{
 						Key:      "key",

test/eventlistener_test.go

@@ -230,8 +230,10 @@ func TestEventListenerCreate(t *testing.T) {
 			),
 			bldr.EventListenerSpec(
 				bldr.EventListenerServiceAccount(sa.Name),
+				bldr.EventListenerReplicas(3),
 				bldr.EventListenerPodTemplate(
 					bldr.EventListenerPodTemplateSpec(
+						bldr.EventListenerPodTemplateNodeSelector(map[string]string{"beta.kubernetes.io/arch": "amd64"}),
 						bldr.EventListenerPodTemplateTolerations([]corev1.Toleration{
 							{
 								Key:      "key",

third_party/github.com/hashicorp/golang-lru/simplelru/lru_interface.go

@@ -34,6 +34,6 @@ type LRUCache interface {
 	// Clears all cache entries.
 	Purge()
 
-  // Resizes cache, returning number evicted
-  Resize(int) int
+	// Resizes cache, returning number evicted
+	Resize(int) int
 }