added deployment resources support (#423)

* added deployment resources support

* hack/update-codegen.sh

* implemented deployment spec template for dispatcher

* code style fix

* added tests to cover deploymentSpecTemplate

config/jetstream/302-jsm-channel.yaml

@@ -49,6 +49,25 @@ spec:
               description: Spec defines the desired state of the Channel.
               type: object
               properties:
+                deploymentSpecTemplate:
+                  description: Deployment configurations to apply to programmatically created dispatcher.
+                  type: object
+                  properties:
+                    resources:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    affinity:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    nodeSelector:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    labels:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
+                    annotations:
+                      type: object
+                      x-kubernetes-preserve-unknown-fields: true
                 consumerConfigTemplate:
                   type: object
                   properties:

config/webhook/500-webhook-deployment.yaml

@@ -104,7 +104,7 @@ spec:
             httpHeaders:
               - name: k-kubelet-probe
                 value: "webhook"
-          initialDelaySeconds: 20
+          initialDelaySeconds: 120
 
       # Our webhook should gracefully terminate by lame ducking first, set this to a sufficiently
       # high value that we respect whatever value it has configured for the lame duck grace period.

examples/config-br-default-channel-jsm-with-deployment.yaml

@@ -0,0 +1,44 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: config-br-default-channel
+  namespace: knative-eventing
+  labels:
+    eventing.knative.dev/release: devel
+data:
+  channelTemplateSpec: |
+    apiVersion: messaging.knative.dev/v1alpha1
+    kind: NatsJetStreamChannel
+    spec:
+      stream:
+        config:
+          retention: Limits
+          maxBytes: 1000000000
+          replicas: 1
+      consumerConfigTemplate:
+        deliverPolicy: New
+        maxDeliver: 1
+      deploymentSpecTemplate:
+        labels:
+          mykey: myvalue
+        annotations:
+          amykey: amyvalue
+        nodeSelector:
+          myselector: qwerty
+        affinity:
+          nodeAffinity:
+            preferredDuringSchedulingIgnoredDuringExecution:
+            - preference:
+                matchExpressions:
+                - key: high-memory
+                  operator: In
+                  values:
+                  - "true"
+              weight: 1
+        resources:
+          requests:
+            cpu: 20m
+            memory: 20Mi
+          limits:
+            cpu: 200m
+            memory: 200Mi

pkg/apis/messaging/v1alpha1/nats_jetstream_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -142,6 +143,9 @@ type NatsJetStreamChannelSpec struct {
 
 	// +optional
 	ConsumerConfigTemplate *ConsumerConfigTemplate `json:"consumerConfigTemplate,omitempty"`
+
+	// +optional
+	DeploymentSpecTemplate *JetStreamDispatcherDeploymentTemplate `json:"deploymentSpecTemplate,omitempty"`
 }
 
 // Stream provides customization options to how the eventing-jetstream dispatcher creates streams.
@@ -301,6 +305,14 @@ type ConsumerConfigTemplate struct {
 	MaxAckPending int `json:"maxAckPending,omitempty"`
 }
 
+type JetStreamDispatcherDeploymentTemplate struct {
+	Annotations  map[string]string           `json:"annotations,omitempty"`
+	Labels       map[string]string           `json:"labels,omitempty"`
+	NodeSelector map[string]string           `json:"nodeSelector,omitempty"`
+	Resources    corev1.ResourceRequirements `json:"resources,omitempty"`
+	Affinity     *corev1.Affinity            `json:"affinity,omitempty"`
+}
+
 // NatsJetStreamChannelStatus represents the current state of a NatssChannel.
 type NatsJetStreamChannelStatus struct {
 	// inherits duck/v1 ChannelableStatus, which currently provides:

pkg/channel/jetstream/controller/natsjetstreamchannel_test.go

@@ -227,6 +227,35 @@ func TestAllCases(t *testing.T) {
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
+		}, {
+			Name: "Works, channel with deployment spec",
+			Key:  ncKey,
+			Objects: []runtime.Object{
+				reconciletesting.NewNatsJetStreamChannel(ncName, testNS,
+					reconciletesting.WithNatsJetStreamDeploymentSpecTemplate),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+				Object: reconciletesting.NewNatsJetStreamChannel(ncName, testNS,
+					reconciletesting.WithNatsJetStreamInitChannelConditions,
+					reconciletesting.WithNatsJetStreamChannelEndpointsNotReady(dispatcherEndpointsNotFound, "Dispatcher Endpoints does not exist"),
+					reconciletesting.WithNatsJetStreamChannelServiceReady(),
+					reconciletesting.WithNatsJetStreamDeploymentSpecTemplate,
+				),
+			}},
+			WantCreates: []runtime.Object{
+				makeDispatcherDeploymentWithSpec(),
+				makeDispatcherService(),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{{
+				Name:      ncName,
+				Patch:     []byte(`{"metadata":{"finalizers":["natsjetstreamchannels.messaging.knative.dev"],"resourceVersion":""}}`),
+				PatchType: types.MergePatchType,
+			}},
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(v1.EventTypeNormal, dispatcherDeploymentCreated, "Dispatcher deployment created"),
+				Eventf(v1.EventTypeNormal, dispatcherServiceCreated, "Dispatcher service created"),
+			},
 		},
 	}
 
@@ -263,6 +292,23 @@ func makeDispatcherDeployment() *appsv1.Deployment {
 	}).Build()
 }
 
+func makeDispatcherDeploymentWithSpec() *appsv1.Deployment {
+	return resources.NewDispatcherDeploymentBuilder().WithArgs(&resources.DispatcherDeploymentArgs{
+		DispatcherScope:     "",
+		DispatcherNamespace: testNS,
+		Image:               dispatcherImage,
+		Replicas:            1,
+		ServiceAccount:      dispatcherServiceAccount,
+		OwnerRef:            metav1.OwnerReference{}, // TODO: Make this work
+		DeploymentLabels: map[string]string{
+			"label1": "value",
+		},
+		DeploymentAnnotations: map[string]string{
+			"annotation1": "value",
+		},
+	}).Build()
+}
+
 func makeEmptyEndpoints() *v1.Endpoints {
 	return &v1.Endpoints{
 		TypeMeta: metav1.TypeMeta{

pkg/channel/jetstream/controller/reconciler.go

@@ -235,6 +235,15 @@ func (r *Reconciler) reconcileDispatcher(ctx context.Context, scope, dispatcherN
 		OwnerRef:            r.controllerRef,
 	}
 
+	if nc.Spec.DeploymentSpecTemplate != nil {
+		logger.Infow("Dispatcher deployment configuration", zap.Any("DispatcherDeployment", nc.Spec.DeploymentSpecTemplate))
+		args.DeploymentAnnotations = nc.Spec.DeploymentSpecTemplate.Annotations
+		args.DeploymentLabels = nc.Spec.DeploymentSpecTemplate.Labels
+		args.DeploymentNodeSelector = nc.Spec.DeploymentSpecTemplate.NodeSelector
+		args.DeploymentAffinity = nc.Spec.DeploymentSpecTemplate.Affinity
+		args.DeploymentResources = nc.Spec.DeploymentSpecTemplate.Resources
+	}
+
 	want := resources.NewDispatcherDeploymentBuilder().WithArgs(&args).Build()
 	d, err := r.deploymentLister.Deployments(dispatcherNamespace).Get(jetstream.DispatcherName)
 	if err != nil {

pkg/channel/jetstream/controller/resources/dispatcher.go

@@ -45,18 +45,21 @@ type DispatcherDeploymentBuilder struct {
 }
 
 type DispatcherDeploymentArgs struct {
-	DispatcherScope       string
-	DispatcherNamespace   string
-	Image                 string
-	Replicas              int32
-	ServiceAccount        string
-	ConfigMapName         string
-	ConfigMapHash         string
-	OwnerRef              metav1.OwnerReference
-	DeploymentAnnotations map[string]string
-	DeploymentLabels      map[string]string
-	PodAnnotations        map[string]string
-	PodLabels             map[string]string
+	DispatcherScope        string
+	DispatcherNamespace    string
+	Image                  string
+	Replicas               int32
+	ServiceAccount         string
+	ConfigMapName          string
+	ConfigMapHash          string
+	OwnerRef               metav1.OwnerReference
+	DeploymentAnnotations  map[string]string
+	DeploymentLabels       map[string]string
+	DeploymentResources    corev1.ResourceRequirements
+	DeploymentNodeSelector map[string]string
+	DeploymentAffinity     *corev1.Affinity
+	PodAnnotations         map[string]string
+	PodLabels              map[string]string
 }
 
 // NewDispatcherDeploymentBuilder returns a builder which builds from scratch a dispatcher deployment.
@@ -94,9 +97,13 @@ func (b *DispatcherDeploymentBuilder) Build() *v1.Deployment {
 	b.deployment.Spec.Template.ObjectMeta.Labels = commonconfig.JoinStringMaps(b.deployment.Spec.Template.ObjectMeta.Labels, b.args.PodLabels)
 	b.deployment.Spec.Template.Spec.ServiceAccountName = b.args.ServiceAccount
 
+	b.deployment.Spec.Template.Spec.NodeSelector = b.args.DeploymentNodeSelector
+	b.deployment.Spec.Template.Spec.Affinity = b.args.DeploymentAffinity
+
 	for i, c := range b.deployment.Spec.Template.Spec.Containers {
 		if c.Name == DispatcherContainerName {
 			container := &b.deployment.Spec.Template.Spec.Containers[i]
+			container.Resources = b.args.DeploymentResources
 			container.Image = b.args.Image
 			if container.Env == nil {
 				container.Env = makeEnv(b.args)

pkg/reconciler/testing/natsjetstreamchannel.go

@@ -79,6 +79,17 @@ func WithNatsJetStreamChannelDeleted(nc *v1alpha1.NatsJetStreamChannel) {
 	nc.ObjectMeta.SetDeletionTimestamp(&deleteTime)
 }
 
+func WithNatsJetStreamDeploymentSpecTemplate(nc *v1alpha1.NatsJetStreamChannel) {
+	nc.Spec.DeploymentSpecTemplate = &v1alpha1.JetStreamDispatcherDeploymentTemplate{
+		Labels: map[string]string{
+			"label1": "value",
+		},
+		Annotations: map[string]string{
+			"annotation1": "value",
+		},
+	}
+}
+
 func WithNatsJetStreamChannelDispatcherReady() NatsJetStreamChannelOption {
 	return func(nc *v1alpha1.NatsJetStreamChannel) {
 		nc.Status.MarkDispatcherTrue()