Pubsub v1beta1 Conversions

cmd/webhook/main.go

@@ -22,7 +22,9 @@ import (
 	configvalidation "github.com/google/knative-gcp/pkg/apis/configs/validation"
 	eventsv1alpha1 "github.com/google/knative-gcp/pkg/apis/events/v1alpha1"
 	messagingv1alpha1 "github.com/google/knative-gcp/pkg/apis/messaging/v1alpha1"
+	"github.com/google/knative-gcp/pkg/apis/pubsub"
 	pubsubv1alpha1 "github.com/google/knative-gcp/pkg/apis/pubsub/v1alpha1"
+	pubsubv1beta1 "github.com/google/knative-gcp/pkg/apis/pubsub/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"knative.dev/eventing/pkg/logconfig"
 	"knative.dev/pkg/configmap"
@@ -36,6 +38,7 @@ import (
 	"knative.dev/pkg/webhook/certificates"
 	"knative.dev/pkg/webhook/configmaps"
 	"knative.dev/pkg/webhook/resourcesemantics"
+	"knative.dev/pkg/webhook/resourcesemantics/conversion"
 	"knative.dev/pkg/webhook/resourcesemantics/defaulting"
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 )
@@ -56,7 +59,7 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	pubsubv1alpha1.SchemeGroupVersion.WithKind("Topic"):            &pubsubv1alpha1.Topic{},
 }
 
-func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+func NewDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	// Decorate contexts with the current state of the config.
 	ctxFunc := func(ctx context.Context) context.Context {
 		return ctx
@@ -81,7 +84,7 @@ func NewDefaultingAdmissionController(ctx context.Context, cmw configmap.Watcher
 	)
 }
 
-func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	return validation.NewAdmissionController(ctx,
 
 		// Name of the validation webhook.
@@ -104,7 +107,7 @@ func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher
 	)
 }
 
-func NewConfigValidationController(ctx context.Context, cmw configmap.Watcher) *controller.Impl {
+func NewConfigValidationController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
 	return configmaps.NewAdmissionController(ctx,
 
 		// Name of the configmap webhook.
@@ -123,6 +126,41 @@ func NewConfigValidationController(ctx context.Context, cmw configmap.Watcher) *
 	)
 }
 
+func NewConversionController(ctx context.Context, _ configmap.Watcher) *controller.Impl {
+	var (
+		pubsubv1alpha1_ = pubsubv1alpha1.SchemeGroupVersion.Version
+		pubsubv1beta1_  = pubsubv1beta1.SchemeGroupVersion.Version
+	)
+
+	return conversion.NewConversionController(ctx,
+		// The path on which to serve the webhook
+		"/resource-conversion",
+
+		// Specify the types of custom resource definitions that should be converted
+		map[schema.GroupKind]conversion.GroupKindConversion{
+			// pubsub
+			pubsubv1beta1.Kind("PullSubscription"): {
+				DefinitionName: pubsub.PullSubscriptionsResource.String(),
+				HubVersion:     pubsubv1alpha1_,
+				Zygotes: map[string]conversion.ConvertibleObject{
+					pubsubv1alpha1_: &pubsubv1alpha1.PullSubscription{},
+					pubsubv1beta1_:  &pubsubv1beta1.PullSubscription{},
+				},
+			},
+			pubsubv1beta1.Kind("Topic"): {
+				DefinitionName: pubsub.TopicsResource.String(),
+				HubVersion:     pubsubv1alpha1_,
+				Zygotes: map[string]conversion.ConvertibleObject{
+					pubsubv1alpha1_: &pubsubv1alpha1.Topic{},
+					pubsubv1beta1_:  &pubsubv1beta1.Topic{},
+				},
+			},
+		},
+		func(ctx context.Context) context.Context {
+			return ctx
+		},
+	)
+}
 func main() {
 	// Set up a signal context with our webhook options
 	ctx := webhook.WithOptions(signals.NewContext(), webhook.Options{
@@ -137,5 +175,6 @@ func main() {
 		NewConfigValidationController,
 		NewValidationAdmissionController,
 		NewDefaultingAdmissionController,
+		NewConversionController,
 	)
 }

config/core/resources/pullsubscription.yaml

@@ -32,6 +32,12 @@ spec:
   subresources:
     status: {}
   preserveUnknownFields: false
+  conversion:
+    strategy: Webhook
+    webhookClientConfig:
+      service:
+        name: webhook
+        namespace: cloud-run-events
   additionalPrinterColumns:
     - name: Ready
       type: string
@@ -46,6 +52,9 @@ spec:
     - name: v1alpha1
       served: true
       storage: true
+    - name: v1beta1
+      served: true
+      storage: false
   # All versions happen to have the same schema today. They will likely diverge in the future.
   validation:
     openAPIV3Schema:

config/core/resources/topic.yaml

@@ -33,6 +33,12 @@ spec:
   subresources:
     status: {}
   preserveUnknownFields: false
+  conversion:
+    strategy: Webhook
+    webhookClientConfig:
+      service:
+        name: webhook
+        namespace: cloud-run-events
   additionalPrinterColumns:
     - name: Ready
       type: string
@@ -50,6 +56,9 @@ spec:
     - name: v1alpha1
       served: true
       storage: true
+    - name: v1beta1
+      served: true
+      storage: false
   # All versions happen to have the same schema today. They will likely diverge in the future.
   validation:
     openAPIV3Schema:

config/core/roles/webhook-clusterrole.yaml

@@ -70,3 +70,9 @@ rules:
       - "delete"
       - "patch"
       - "watch"
+
+  # Necessary for conversion webhook. These are copied from the serving
+  # TODO: Do we really need all these permissions?
+  - apiGroups: ["apiextensions.k8s.io"]
+    resources: ["customresourcedefinitions"]
+    verbs: ["get", "list", "create", "update", "delete", "patch", "watch"]

pkg/apis/pubsub/register.go

@@ -17,6 +17,21 @@ limitations under the License.
 // Package pubsub contains Cloud Run Events API versions for pubsub components
 package pubsub
 
+import "k8s.io/apimachinery/pkg/runtime/schema"
+
 const (
 	GroupName = "pubsub.cloud.google.com"
 )
+
+var (
+	// PullSubscriptionsResource represents a PubSub PullSubscription.
+	PullSubscriptionsResource = schema.GroupResource{
+		Group:    GroupName,
+		Resource: "pullsubscriptions",
+	}
+	// TopicsResource represents a PubSub Topic.
+	TopicsResource = schema.GroupResource{
+		Group:    GroupName,
+		Resource: "topics",
+	}
+)

pkg/apis/pubsub/v1alpha1/pullsubscription_conversion.go

@@ -0,0 +1,174 @@
+/*
+Copyright 2020 Google LLC
+
+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 v1alpha1
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
+	duckv1beta1 "github.com/google/knative-gcp/pkg/apis/duck/v1beta1"
+	"github.com/google/knative-gcp/pkg/apis/pubsub/v1beta1"
+	"knative.dev/pkg/apis"
+)
+
+// ConvertTo implements apis.Convertible.
+// Converts source (from v1beta1.PullSubscription) into v1alpha1.PullSubscription.
+func (source *PullSubscription) ConvertTo(_ context.Context, to apis.Convertible) error {
+	switch sink := to.(type) {
+	case *v1beta1.PullSubscription:
+		sink.ObjectMeta = source.ObjectMeta
+		sink.Spec.PubSubSpec = convertToV1beta1PubSubSpec(source.Spec.PubSubSpec)
+		sink.Spec.Topic = source.Spec.Topic
+		sink.Spec.AckDeadline = source.Spec.AckDeadline
+		sink.Spec.RetainAckedMessages = source.Spec.RetainAckedMessages
+		sink.Spec.RetentionDuration = source.Spec.RetentionDuration
+		sink.Spec.Transformer = source.Spec.Transformer
+		if mode, err := convertToV1beta1ModeType(source.Spec.Mode); err != nil {
+			return err
+		} else {
+			sink.Spec.Mode = mode
+		}
+		sink.Spec.AdapterType = source.Spec.AdapterType
+		sink.Status.PubSubStatus = convertToV1beta1PubSubStatus(source.Status.PubSubStatus)
+		sink.Status.TransformerURI = source.Status.TransformerURI
+		sink.Status.SubscriptionID = source.Status.SubscriptionID
+		return nil
+	default:
+		return fmt.Errorf("unknown conversion, got: %T", sink)
+
+	}
+}
+
+// ConvertFrom implements apis.Convertible.
+// Converts obj from v1alpha1.PullSubscription into v1beta1.PullSubscription.
+func (sink *PullSubscription) ConvertFrom(_ context.Context, from apis.Convertible) error {
+	switch source := from.(type) {
+	case *v1beta1.PullSubscription:
+		sink.ObjectMeta = source.ObjectMeta
+		sink.Spec.PubSubSpec = convertFromV1beta1PubSubSpec(source.Spec.PubSubSpec)
+		sink.Spec.Topic = source.Spec.Topic
+		sink.Spec.AckDeadline = source.Spec.AckDeadline
+		sink.Spec.RetainAckedMessages = source.Spec.RetainAckedMessages
+		sink.Spec.RetentionDuration = source.Spec.RetentionDuration
+		sink.Spec.Transformer = source.Spec.Transformer
+		if mode, err := convertFromV1beta1ModeType(source.Spec.Mode); err != nil {
+			return err
+		} else {
+			sink.Spec.Mode = mode
+		}
+		sink.Spec.AdapterType = source.Spec.AdapterType
+		sink.Status.PubSubStatus = convertFromV1beta1PubSubStatus(source.Status.PubSubStatus)
+		sink.Status.TransformerURI = source.Status.TransformerURI
+		sink.Status.SubscriptionID = source.Status.SubscriptionID
+		return nil
+	default:
+		return fmt.Errorf("unknown conversion, got: %T", source)
+	}
+}
+
+func convertToV1beta1ModeType(from ModeType) (v1beta1.ModeType, error) {
+	switch from {
+	case ModeCloudEventsBinary:
+		return v1beta1.ModeCloudEventsBinary, nil
+	case ModeCloudEventsStructured:
+		return v1beta1.ModeCloudEventsStructured, nil
+	case ModePushCompatible:
+		return v1beta1.ModePushCompatible, nil
+	case "":
+		return "", nil
+	default:
+		return "unknown", fmt.Errorf("unknown ModeType %v", from)
+	}
+}
+
+func convertFromV1beta1ModeType(from v1beta1.ModeType) (ModeType, error) {
+	switch from {
+	case v1beta1.ModeCloudEventsBinary:
+		return ModeCloudEventsBinary, nil
+	case v1beta1.ModeCloudEventsStructured:
+		return ModeCloudEventsStructured, nil
+	case v1beta1.ModePushCompatible:
+		return ModePushCompatible, nil
+	case "":
+		return "", nil
+	default:
+		return "unknown", fmt.Errorf("unknown ModeType %v", from)
+	}
+}
+
+func convertToV1beta1PubSubSpec(from v1alpha1.PubSubSpec) duckv1beta1.PubSubSpec {
+	to := duckv1beta1.PubSubSpec{}
+	to.SourceSpec = from.SourceSpec
+	to.IdentitySpec = convertToV1beta1IdentitySpec(from.IdentitySpec)
+	to.Secret = from.Secret
+	to.Project = from.Project
+	return to
+}
+func convertFromV1beta1PubSubSpec(from duckv1beta1.PubSubSpec) v1alpha1.PubSubSpec {
+	to := v1alpha1.PubSubSpec{}
+	to.SourceSpec = from.SourceSpec
+	to.IdentitySpec = convertFromV1beta1IdentitySpec(from.IdentitySpec)
+	to.Secret = from.Secret
+	to.Project = from.Project
+	return to
+}
+
+func convertToV1beta1IdentitySpec(from v1alpha1.IdentitySpec) duckv1beta1.IdentitySpec {
+	to := duckv1beta1.IdentitySpec{}
+	to.GoogleServiceAccount = from.GoogleServiceAccount
+	return to
+}
+func convertFromV1beta1IdentitySpec(from duckv1beta1.IdentitySpec) v1alpha1.IdentitySpec {
+	to := v1alpha1.IdentitySpec{}
+	to.GoogleServiceAccount = from.GoogleServiceAccount
+	return to
+}
+
+func convertToV1beta1PubSubStatus(from v1alpha1.PubSubStatus) duckv1beta1.PubSubStatus {
+	to := duckv1beta1.PubSubStatus{}
+	to.IdentityStatus = convertToV1beta1IdentityStatus(from.IdentityStatus)
+	to.SinkURI = from.SinkURI
+	to.CloudEventAttributes = from.CloudEventAttributes
+	to.ProjectID = from.ProjectID
+	to.TopicID = from.TopicID
+	to.SubscriptionID = from.SubscriptionID
+	return to
+}
+func convertFromV1beta1PubSubStatus(from duckv1beta1.PubSubStatus) v1alpha1.PubSubStatus {
+	to := v1alpha1.PubSubStatus{}
+	to.IdentityStatus = convertFromV1beta1IdentityStatus(from.IdentityStatus)
+	to.SinkURI = from.SinkURI
+	to.CloudEventAttributes = from.CloudEventAttributes
+	to.ProjectID = from.ProjectID
+	to.TopicID = from.TopicID
+	to.SubscriptionID = from.SubscriptionID
+	return to
+}
+
+func convertToV1beta1IdentityStatus(from v1alpha1.IdentityStatus) duckv1beta1.IdentityStatus {
+	to := duckv1beta1.IdentityStatus{}
+	to.Status = from.Status
+	to.ServiceAccountName = from.ServiceAccountName
+	return to
+}
+func convertFromV1beta1IdentityStatus(from duckv1beta1.IdentityStatus) v1alpha1.IdentityStatus {
+	to := v1alpha1.IdentityStatus{}
+	to.Status = from.Status
+	to.ServiceAccountName = from.ServiceAccountName
+	return to
+}