Add KafkaSource v1 API (#4130)

* Add KafkaSource v1 types

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Add conversions functions

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Update CRD resource

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Add conversion webhook

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Fix tests

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Generate sources v1 clients

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Add permissions to webhook for read + write CRDs

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

---------

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

control-plane/cmd/webhook-kafka/main.go

@@ -30,12 +30,14 @@ import (
 	"knative.dev/pkg/webhook"
 	"knative.dev/pkg/webhook/certificates"
 	"knative.dev/pkg/webhook/resourcesemantics"
+	"knative.dev/pkg/webhook/resourcesemantics/conversion"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
 	eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	"knative.dev/eventing/pkg/apis/feature"
 
+	sourcesv1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1"
 	sourcesv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1"
@@ -53,6 +55,7 @@ const (
 var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"):    &eventingv1alpha1.KafkaSink{},
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):    &sourcesv1beta1.KafkaSource{},
+	sourcesv1.SchemeGroupVersion.WithKind("KafkaSource"):         &sourcesv1.KafkaSource{},
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
 	eventingcorev1.SchemeGroupVersion.WithKind("Broker"):         &eventingv1.BrokerStub{},
 	kafkainternals.SchemeGroupVersion.WithKind("ConsumerGroup"):  &kafkainternals.ConsumerGroup{},
@@ -144,6 +147,33 @@ func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher
 	)
 }
 
+func NewConversionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+
+	ctxFunc := func(ctx context.Context) context.Context {
+		return ctx
+	}
+
+	return conversion.NewConversionController(
+		ctx,
+
+		// The path on which to serve the webhook
+		"/resource-conversion",
+
+		map[schema.GroupKind]conversion.GroupKindConversion{
+			sourcesv1.Kind("KafkaSource"): {
+				DefinitionName: "kafkasources.sources.knative.dev",
+				HubVersion:     sourcesv1beta1.SchemeGroupVersion.Version,
+				Zygotes: map[string]conversion.ConvertibleObject{
+					sourcesv1beta1.SchemeGroupVersion.Version: &sourcesv1beta1.KafkaSource{},
+					sourcesv1.SchemeGroupVersion.Version:      &sourcesv1.KafkaSource{},
+				},
+			},
+		},
+		// A function that infuses the context passed to ConvertTo/ConvertFrom/SetDefaults with custom metadata.
+		ctxFunc,
+	)
+}
+
 func main() {
 
 	// Set up a signal context with our webhook options
@@ -159,5 +189,6 @@ func main() {
 		NewDefaultingAdmissionController,
 		NewPodDefaultingAdmissionController,
 		NewValidationAdmissionController,
+		NewConversionController,
 	)
 }

control-plane/config/eventing-kafka-broker/200-webhook/100-webhook-cluster-role.yaml

@@ -108,3 +108,9 @@ rules:
     verbs:
       - get
       - list
+
+  # Necessary for conversion webhook.
+  - apiGroups: ["apiextensions.k8s.io"]
+    resources: ["customresourcedefinitions"]
+    verbs: ["get", "list", "create", "update", "patch", "watch"]
+

control-plane/pkg/apis/bindings/v1/doc.go

@@ -0,0 +1,22 @@
+/*
+Copyright 2020 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
+
+    http://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.
+*/
+
+// Package v1 contains API Schema definitions for the sources v1 API group
+// +k8s:openapi-gen=true
+// +k8s:deepcopy-gen=package,register
+// +k8s:defaulter-gen=TypeMeta
+// +groupName=bindings.knative.dev
+package v1

control-plane/pkg/apis/bindings/v1/fuzzer.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2020 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
+
+    http://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.
+*/
+
+package v1
+
+import (
+	fuzz "github.com/google/gofuzz"
+	"k8s.io/apimachinery/pkg/api/apitesting/fuzzer"
+	"k8s.io/apimachinery/pkg/runtime/serializer"
+	pkgfuzzer "knative.dev/pkg/apis/testing/fuzzer"
+)
+
+// FuzzerFuncs includes fuzzing funcs for bindings.knative.dev v1 types
+//
+// For other examples see
+// https://github.com/kubernetes/apimachinery/blob/master/pkg/apis/meta/fuzzer/fuzzer.go
+var FuzzerFuncs = fuzzer.MergeFuzzerFuncs(
+	func(codecs serializer.CodecFactory) []interface{} {
+		return []interface{}{
+			func(s *KafkaBindingStatus, c fuzz.Continue) {
+				c.FuzzNoCustom(s) // fuzz the status object
+
+				// Clear the random fuzzed condition
+				s.Status.SetConditions(nil)
+
+				// Fuzz the known conditions except their type value
+				s.InitializeConditions()
+				pkgfuzzer.FuzzConditions(&s.Status, c)
+			},
+		}
+	},
+)

control-plane/pkg/apis/bindings/v1/kafka_conversion.go

@@ -0,0 +1,44 @@
+/*
+Copyright 2020 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
+
+    http://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.
+*/
+
+package v1
+
+import (
+	"context"
+	"fmt"
+
+	"knative.dev/pkg/apis"
+)
+
+// ConvertTo implements apis.Convertible
+func (source *KafkaBinding) ConvertTo(_ context.Context, sink apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", sink)
+}
+
+// ConvertFrom implements apis.Convertible
+func (sink *KafkaBinding) ConvertFrom(_ context.Context, source apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", source)
+}
+
+// ConvertTo implements apis.Convertible
+func (source *KafkaAuthSpec) ConvertTo(_ context.Context, sink apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", sink)
+}
+
+// ConvertFrom implements apis.Convertible
+func (sink *KafkaAuthSpec) ConvertFrom(_ context.Context, source apis.Convertible) error {
+	return fmt.Errorf("v1 is the highest known version, got: %T", source)
+}

control-plane/pkg/apis/bindings/v1/kafka_defaults.go

@@ -0,0 +1,29 @@
+/*
+Copyright 2020 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
+
+    http://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.
+*/
+
+package v1
+
+import (
+	"context"
+)
+
+// SetDefaults ensures KafkaBinding reflects the default values.
+func (r *KafkaBinding) SetDefaults(ctx context.Context) {
+	if r.Spec.Subject.Namespace == "" {
+		// Default the subject's namespace to our namespace.
+		r.Spec.Subject.Namespace = r.Namespace
+	}
+}