Adds Kafka Channel Provisioner Controllers (#468)

* Kafka Channel Provisioner Controllers

* Remove controllerRuntimeStart and address PR comments

* Remove fetching configmap in controller

* Add more tests, improv coverage

* Add ChannelStatus.IsReady

* Address PR comments

* Remove unfinished README

* Switch from logr to zap

* Provision Channel as Kafka Topic

* Deprovision Channel

* Merge matzew/try_kafka_provisioner into try_kafka_provisioner

* Fix few more after pr 562

* Fix tests and imports

* PR feedback for removing ClusterChannelProvisioner name from configmap

There were some concerns in fetching the provisioner name from a config map.

* Adding instructions for Channel provisioner

* short cut code for missing ...

* Updating to latest Kafka client, and fixing idle bug

* Fix conflicts and unit tests

* Address PR feedback

* Updating docs, based on feedback

Gopkg.toml

@@ -79,6 +79,10 @@ required = [
   name = "github.com/knative/serving"
   version = "v0.1.1"
 
+[[override]]
+  name = "github.com/Shopify/sarama"
+  version = "1.19.0"
+
 [[constraint]]
   name = "sigs.k8s.io/controller-runtime"
   # HEAD as of 2018-09-19

config/provisioners/kafka/README.md

@@ -0,0 +1,78 @@
+# Apache Kafka Channels
+
+Deployment steps:
+1. Setup [Knative Eventing](../../../DEVELOPMENT.md)
+1. If not done already, install an Apache Kafka cluster. There are two choices:
+   * Simple installation of [Apache Kafka](broker).
+   * A production grade installation using the [Strimzi Kafka Operator](strimzi).
+   Installation [guides](http://strimzi.io/quickstarts/) are provided for
+   kubernetes and Openshift.
+
+1. Now that Apache Kafka is installed, you need to configure the
+`bootstrap_servers` value  in the `kafka-channel-controller-config` ConfigMap,
+located inside the `config/provisioners/kafka/kafka-provisioner.yaml` file:
+    ```
+    ...
+    apiVersion: v1
+    kind: ConfigMap
+    metadata:
+      name: kafka-channel-controller-config
+      namespace: knative-eventing
+    data:
+      # Broker URL's for the provisioner
+      bootstrap_servers: kafkabroker.kafka:9092
+    ...
+    ```
+    > Note: The `bootstrap_servers` needs to contain the address of at least
+    one broker of your Apache Kafka cluster. If you are using Strimzi, you need
+    to update the `bootstrap_servers` value to
+    `my-cluster-kafka-bootstrap.mynamespace:9092`.
+1. Apply the 'Kafka' ClusterChannelProvisioner, Controller, and Dispatcher:
+    ```
+    ko apply -f config/provisioners/kafka/kafka-provisioner.yaml
+    ```
+1. Create Channels that reference the 'kafka-channel'.
+
+    ```yaml
+    apiVersion: eventing.knative.dev/v1alpha1
+    kind: Channel
+    metadata:
+      name: my-kafka-channel
+    spec:
+      provisioner:
+        apiVersion: eventing.knative.dev/v1alpha1
+        kind: ClusterChannelProvisioner
+        name: kafka-channel
+    ```
+
+## Components
+
+The major components are:
+* ClusterChannelProvisioner Controller
+* Channel Controller
+* Channel Controller Config Map.
+* Channel Dispatcher
+* Channel Dispatcher Config Map.
+
+The ClusterChannelProvisioner Controller and the Channel Controller are colocated
+in one Pod:
+```shell
+kubectl get deployment -n knative-eventing kafka-channel-controller
+```
+
+The Channel Controller Config Map is used to configure the `bootstrap_servers`
+of your Apache Kafka installation:
+```shell
+kubectl get configmap -n knative-eventing kafka-channel-dispatcher-config-map
+```
+
+The Channel Dispatcher receives and distributes all events:
+```shell
+kubectl get statefulset -n knative-eventing kafka-channel-dispatcher
+```
+
+The Channel Dispatcher Config Map is used to send information about Channels and
+Subscriptions from the Channel Controller to the Channel Dispatcher:
+```shell
+kubectl get configmap -n knative-eventing kafka-channel-dispatcher-config-map
+```

config/provisioners/kafka/broker/README.md

@@ -0,0 +1,13 @@
+# Apache Kafka - simple installation
+
+1. For an installation of a simple (**non production**) Apache Kafka cluster, a setup is provided:
+    ```
+    kubectl create namespace kafka
+    kubectl apply -n kafka -f kafka-broker.yaml
+    ```
+    > Note: If you are running Knative on OpenShift you will need to run the following command first to allow the Kafka broker to run as root:
+      ```
+      oc adm policy add-scc-to-user anyuid -z default -n kafka
+      ```
+
+Continue the configuration of Knative Eventing with [step `3`](../).

config/provisioners/kafka/broker/kafka-broker.yaml

@@ -0,0 +1,87 @@
+########################################## KAFKA BROKER ######################################
+# The following does not need to live in the same namespace as the bus.
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: kafka-broker
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: kafka-broker
+    spec:
+      containers:
+      - name: kafka-broker
+        image: wurstmeister/kafka:1.1.0
+        ports:
+        - containerPort: 9092
+        env:
+          - name: MY_POD_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
+          - name: KAFKA_BROKER_ID
+            value: "0"
+          - name: KAFKA_LISTENERS
+            value: "INTERNAL://:9093,EXTERNAL://:9092"
+          - name: KAFKA_ADVERTISED_LISTENERS
+            value: "INTERNAL://:9093,EXTERNAL://kafkabroker.$(MY_POD_NAMESPACE):9092"
+          - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
+            value: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
+          - name: KAFKA_INTER_BROKER_LISTENER_NAME
+            value: "INTERNAL"
+          - name: KAFKA_ZOOKEEPER_CONNECT
+            value: "zookeeper.$(MY_POD_NAMESPACE):2181"
+          - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
+            value: "false"
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: kafkabroker
+spec:
+  type: NodePort
+  selector:
+    app: kafka-broker
+  ports:
+  - port: 9092
+    name: kafka
+    protocol: TCP
+---
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: zookeeper
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: zookeeper
+    spec:
+      containers:
+      - name: zookeeper
+        image: wurstmeister/zookeeper:3.4.6
+        ports:
+        - containerPort: 2181
+        env:
+        - name: ZOOKEEPER_ID
+          value: "1"
+        - name: ZOOKEEPER_SERVER_1
+          value: zookeeper
+
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: zookeeper
+spec:
+  selector:
+    app: zookeeper
+  ports:
+  - port: 2181
+    name: zookeeper
+    protocol: TCP
+

config/provisioners/kafka/kafka-provisioner.yaml

@@ -0,0 +1,85 @@
+# Copyright 2018 The Knative Authors
+#
+# 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
+#
+#     https://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.
+
+apiVersion: eventing.knative.dev/v1alpha1
+kind: ClusterChannelProvisioner
+metadata:
+  name: kafka-channel
+spec: {}
+---
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: kafka-channel-controller
+  namespace: knative-eventing
+---
+
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: kafka-channel-controller
+rules:
+- apiGroups: ["eventing.knative.dev"]
+  resources: ["clusterchannelprovisioners", "channels"]
+  verbs: ["get", "watch", "list", "update"]
+---
+
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: ClusterRoleBinding
+metadata:
+  name: kafka-channel-controller-manage
+subjects:
+  - kind: ServiceAccount
+    name: kafka-channel-controller
+    namespace: knative-eventing
+roleRef:
+  kind: ClusterRole
+  name: kafka-channel-controller
+  apiGroup: rbac.authorization.k8s.io
+---
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kafka-channel-controller-config
+  namespace: knative-eventing
+data:
+  # Broker URL's for the provisioner
+  bootstrap_servers: kafkabroker.kafka:9092
+---
+
+apiVersion: apps/v1beta1
+kind: Deployment
+metadata:
+  name: kafka-channel-controller
+  namespace: knative-eventing
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: kafka-channel-controller
+    spec:
+      serviceAccountName: kafka-channel-controller
+      containers:
+      - name: kafka-channel-controller-controller
+        image: github.com/knative/eventing/pkg/provisioners/kafka
+        volumeMounts:
+          - name: kafka-channel-controller-config
+            mountPath: /etc/config-provisioner
+      volumes:
+        - name: kafka-channel-controller-config
+          configMap:
+            name: kafka-channel-controller-config

pkg/apis/eventing/v1alpha1/channel_types.go

@@ -136,6 +136,11 @@ func (cs *ChannelStatus) MarkProvisioned() {
 	chanCondSet.Manage(cs).MarkTrue(ChannelConditionProvisioned)
 }
 
+// MarkNotProvisioned sets ChannelConditionProvisioned condition to False state.
+func (cs *ChannelStatus) MarkNotProvisioned(reason, messageFormat string, messageA ...interface{}) {
+	chanCondSet.Manage(cs).MarkFalse(ChannelConditionProvisioned, reason, messageFormat, messageA...)
+}
+
 // SetAddress makes this Channel addressable by setting the hostname. It also
 // sets the ChannelConditionAddressable to true.
 func (cs *ChannelStatus) SetAddress(hostname string) {