diff --git a/cmd/controller/main.go b/cmd/controller/main.go index 8bbdc8d81c3..c6f0c7b7a55 100644 --- a/cmd/controller/main.go +++ b/cmd/controller/main.go @@ -21,6 +21,8 @@ import ( "log" "os" + "github.com/knative/eventing/pkg/reconciler/eventtype" + kubeinformers "k8s.io/client-go/informers" "k8s.io/client-go/rest" @@ -67,7 +69,7 @@ func main() { logger.Info("Starting the controller") - const numControllers = 5 + const numControllers = 6 cfg.QPS = numControllers * rest.DefaultQPS cfg.Burst = numControllers * rest.DefaultBurst opt := reconciler.NewOptionsOrDie(cfg, logger, stopCh) @@ -80,6 +82,7 @@ func main() { channelInformer := eventingInformerFactory.Eventing().V1alpha1().Channels() subscriptionInformer := eventingInformerFactory.Eventing().V1alpha1().Subscriptions() brokerInformer := eventingInformerFactory.Eventing().V1alpha1().Brokers() + eventTypeInformer := eventingInformerFactory.Eventing().V1alpha1().EventTypes() // Kube serviceInformer := kubeInformerFactory.Core().V1().Services() @@ -125,6 +128,11 @@ func main() { FilterServiceAccountName: getRequiredEnv("BROKER_FILTER_SERVICE_ACCOUNT"), }, ), + eventtype.NewController( + opt, + eventTypeInformer, + brokerInformer, + ), } if len(controllers) != numControllers { logger.Fatalf("Number of controllers and QPS settings mismatch: %d != %d", len(controllers), numControllers) @@ -147,6 +155,7 @@ func main() { channelInformer.Informer(), subscriptionInformer.Informer(), triggerInformer.Informer(), + eventTypeInformer.Informer(), // Kube configMapInformer.Informer(), serviceInformer.Informer(), @@ -158,8 +167,7 @@ func main() { // Start all of the controllers. logger.Info("Starting controllers.") - go kncontroller.StartAll(stopCh, controllers...) - <-stopCh + kncontroller.StartAll(stopCh, controllers...) } func init() { diff --git a/cmd/sources-controller/main.go b/cmd/sources-controller/main.go index 000cbf04c4a..2f5055a5b04 100644 --- a/cmd/sources-controller/main.go +++ b/cmd/sources-controller/main.go @@ -121,9 +121,7 @@ func main() { // Start all of the controllers. logger.Info("Starting controllers.") - go kncontroller.StartAll(stopCh, controllers...) - - <-stopCh + kncontroller.StartAll(stopCh, controllers...) } func setupLogger() (*zap.SugaredLogger, zap.AtomicLevel) { diff --git a/cmd/webhook/main.go b/cmd/webhook/main.go index b746e9ca24a..429ca099c03 100644 --- a/cmd/webhook/main.go +++ b/cmd/webhook/main.go @@ -101,6 +101,7 @@ func main() { eventingv1alpha1.SchemeGroupVersion.WithKind("ClusterChannelProvisioner"): &eventingv1alpha1.ClusterChannelProvisioner{}, eventingv1alpha1.SchemeGroupVersion.WithKind("Subscription"): &eventingv1alpha1.Subscription{}, eventingv1alpha1.SchemeGroupVersion.WithKind("Trigger"): &eventingv1alpha1.Trigger{}, + eventingv1alpha1.SchemeGroupVersion.WithKind("EventType"): &eventingv1alpha1.EventType{}, }, Logger: logger, } diff --git a/config/200-controller-clusterrole.yaml b/config/200-controller-clusterrole.yaml index 5be03648e78..60afacc783d 100644 --- a/config/200-controller-clusterrole.yaml +++ b/config/200-controller-clusterrole.yaml @@ -70,6 +70,8 @@ rules: - "subscriptions/status" - "triggers" - "triggers/status" + - "eventtypes" + - "eventtypes/status" verbs: *everything # Source resources and statuses we care about. diff --git a/config/200-webhook-clusterrole.yaml b/config/200-webhook-clusterrole.yaml index 89823ca6cb1..53176c1013a 100644 --- a/config/200-webhook-clusterrole.yaml +++ b/config/200-webhook-clusterrole.yaml @@ -79,6 +79,8 @@ rules: - "subscriptions/status" - "triggers" - "triggers/status" + - "eventtypes" + - "eventtypes/status" verbs: - "get" - "list" diff --git a/config/300-eventtype.yaml b/config/300-eventtype.yaml new file mode 100644 index 00000000000..1b83fd8cd11 --- /dev/null +++ b/config/300-eventtype.yaml @@ -0,0 +1,53 @@ +# Copyright 2019 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. +apiVersion: apiextensions.k8s.io/v1beta1 +kind: CustomResourceDefinition +metadata: + name: eventtypes.eventing.knative.dev +spec: + group: eventing.knative.dev + version: v1alpha1 + names: + kind: EventType + plural: eventtypes + singular: eventtype + categories: + - all + - knative + - eventing + scope: Namespaced + subresources: + status: {} + additionalPrinterColumns: + - name: Type + type: string + JSONPath: ".spec.type" + - name: Source + type: string + JSONPath: ".spec.source" + - name: Schema + type: string + JSONPath: ".spec.schema" + - name: Broker + type: string + JSONPath: ".spec.broker" + - name: Description + type: string + JSONPath: ".spec.description" + - name: Ready + type: string + JSONPath: ".status.conditions[?(@.type==\"Ready\")].status" + - name: Reason + type: string + JSONPath: ".status.conditions[?(@.type==\"Ready\")].reason" diff --git a/docs/registry/example_eventtype.yaml b/docs/registry/example_eventtype.yaml new file mode 100644 index 00000000000..17b12a44545 --- /dev/null +++ b/docs/registry/example_eventtype.yaml @@ -0,0 +1,49 @@ +# Copyright 2019 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. + +--- + +# This EventType creates an event type with type 'com.github.pull_request', source +# 'github.com', for the 'default' Broker. + +apiVersion: eventing.knative.dev/v1alpha1 +kind: EventType +metadata: + name: com.github.pullrequest +spec: + type: com.github.pull_request + source: github.com + broker: default + description: "GitHub Pull Request" + +--- + +# This Trigger matches all events of type 'com.github.pull_request' and source +# 'github.com', that are sent to the 'default' Broker. + +apiVersion: eventing.knative.dev/v1alpha1 +kind: Trigger +metadata: + name: filtering-event-type +spec: + filter: + sourceAndType: + type: com.github.pull_request + source: github.com + broker: default + subscriber: + ref: + apiVersion: serving.knative.dev/v1alpha1 + kind: Service + name: message-dumper diff --git a/pkg/apis/eventing/v1alpha1/eventtype_defaults.go b/pkg/apis/eventing/v1alpha1/eventtype_defaults.go new file mode 100644 index 00000000000..eee2cec8c46 --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_defaults.go @@ -0,0 +1,29 @@ +/* +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 + + 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" + +func (et *EventType) SetDefaults(ctx context.Context) { + et.Spec.SetDefaults(ctx) +} + +func (ets *EventTypeSpec) SetDefaults(ctx context.Context) { + if ets.Broker == "" { + ets.Broker = "default" + } +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_defaults_test.go b/pkg/apis/eventing/v1alpha1/eventtype_defaults_test.go new file mode 100644 index 00000000000..6ac447be495 --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_defaults_test.go @@ -0,0 +1,83 @@ +/* +Copyright 2019 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 v1alpha1 + +import ( + "context" + "testing" + + "github.com/google/go-cmp/cmp" +) + +func TestEventTypeDefaults(t *testing.T) { + testCases := map[string]struct { + initial EventType + expected EventType + }{ + "nil spec": { + initial: EventType{}, + expected: EventType{ + Spec: EventTypeSpec{ + Broker: "default", + }, + }, + }, + "broker empty": { + initial: EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "", + Schema: "test-schema", + }, + }, + expected: EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "default", + Schema: "test-schema", + }, + }, + }, + "broker not set": { + initial: EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Schema: "test-schema", + }, + }, + expected: EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "default", + Schema: "test-schema", + }, + }, + }, + } + for n, tc := range testCases { + t.Run(n, func(t *testing.T) { + tc.initial.SetDefaults(context.TODO()) + if diff := cmp.Diff(tc.expected, tc.initial); diff != "" { + t.Fatalf("Unexpected defaults (-want, +got): %s", diff) + } + }) + } +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_lifecycle.go b/pkg/apis/eventing/v1alpha1/eventtype_lifecycle.go new file mode 100644 index 00000000000..5cef6862a6c --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_lifecycle.go @@ -0,0 +1,58 @@ +/* +Copyright 2019 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 v1alpha1 + +import duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1" + +var eventTypeCondSet = duckv1alpha1.NewLivingConditionSet(EventTypeConditionBrokerExists, EventTypeConditionBrokerReady) + +const ( + EventTypeConditionReady = duckv1alpha1.ConditionReady + EventTypeConditionBrokerExists duckv1alpha1.ConditionType = "BrokerExists" + EventTypeConditionBrokerReady duckv1alpha1.ConditionType = "BrokerReady" +) + +// GetCondition returns the condition currently associated with the given type, or nil. +func (et *EventTypeStatus) GetCondition(t duckv1alpha1.ConditionType) *duckv1alpha1.Condition { + return eventTypeCondSet.Manage(et).GetCondition(t) +} + +// IsReady returns true if the resource is ready overall. +func (et *EventTypeStatus) IsReady() bool { + return eventTypeCondSet.Manage(et).IsHappy() +} + +// InitializeConditions sets relevant unset conditions to Unknown state. +func (et *EventTypeStatus) InitializeConditions() { + eventTypeCondSet.Manage(et).InitializeConditions() +} + +func (et *EventTypeStatus) MarkBrokerExists() { + eventTypeCondSet.Manage(et).MarkTrue(EventTypeConditionBrokerExists) +} + +func (et *EventTypeStatus) MarkBrokerDoesNotExist() { + eventTypeCondSet.Manage(et).MarkFalse(EventTypeConditionBrokerExists, "BrokerDoesNotExist", "Broker does not exist") +} + +func (et *EventTypeStatus) MarkBrokerReady() { + eventTypeCondSet.Manage(et).MarkTrue(EventTypeConditionBrokerReady) +} + +func (et *EventTypeStatus) MarkBrokerNotReady() { + eventTypeCondSet.Manage(et).MarkFalse(EventTypeConditionBrokerReady, "BrokerNotReady", "Broker is not ready") +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_lifecycle_test.go b/pkg/apis/eventing/v1alpha1/eventtype_lifecycle_test.go new file mode 100644 index 00000000000..d88c3fcb308 --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_lifecycle_test.go @@ -0,0 +1,247 @@ +/* +Copyright 2019 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 v1alpha1 + +import ( + "testing" + + "github.com/google/go-cmp/cmp" + duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1" + corev1 "k8s.io/api/core/v1" +) + +var ( + trueValue = true + falseValue = false +) + +var ( + eventTypeConditionReady = duckv1alpha1.Condition{ + Type: EventTypeConditionReady, + Status: corev1.ConditionTrue, + } + + eventTypeConditionBrokerExists = duckv1alpha1.Condition{ + Type: EventTypeConditionBrokerExists, + Status: corev1.ConditionTrue, + } + + eventTypeConditionBrokerReady = duckv1alpha1.Condition{ + Type: EventTypeConditionBrokerReady, + Status: corev1.ConditionTrue, + } +) + +func TestEventTypeGetCondition(t *testing.T) { + tests := []struct { + name string + ets *EventTypeStatus + condQuery duckv1alpha1.ConditionType + want *duckv1alpha1.Condition + }{{ + name: "single condition", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{ + eventTypeConditionReady, + }, + }, + }, + condQuery: duckv1alpha1.ConditionReady, + want: &eventTypeConditionReady, + }, { + name: "broker exists condition", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{ + eventTypeConditionBrokerExists, + }, + }, + }, + condQuery: EventTypeConditionBrokerExists, + want: &eventTypeConditionBrokerExists, + }, { + name: "multiple conditions, condition true", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{ + eventTypeConditionBrokerExists, + eventTypeConditionBrokerReady, + }, + }, + }, + condQuery: EventTypeConditionBrokerReady, + want: &eventTypeConditionBrokerReady, + }, { + name: "unknown condition", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{ + eventTypeConditionBrokerReady, + eventTypeConditionReady, + }, + }, + }, + condQuery: duckv1alpha1.ConditionType("foo"), + want: nil, + }} + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := test.ets.GetCondition(test.condQuery) + if diff := cmp.Diff(test.want, got); diff != "" { + t.Errorf("unexpected condition (-want, +got) = %v", diff) + } + }) + } +} + +func TestEventTypeInitializeConditions(t *testing.T) { + tests := []struct { + name string + ets *EventTypeStatus + want *EventTypeStatus + }{{ + name: "empty", + ets: &EventTypeStatus{}, + want: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{{ + Type: EventTypeConditionBrokerExists, + Status: corev1.ConditionUnknown, + }, { + Type: EventTypeConditionBrokerReady, + Status: corev1.ConditionUnknown, + }, { + Type: EventTypeConditionReady, + Status: corev1.ConditionUnknown, + }, + }, + }, + }, + }, { + name: "one false", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{{ + Type: EventTypeConditionBrokerExists, + Status: corev1.ConditionFalse, + }}, + }, + }, + want: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{{ + Type: EventTypeConditionBrokerExists, + Status: corev1.ConditionFalse, + }, { + Type: EventTypeConditionBrokerReady, + Status: corev1.ConditionUnknown, + }, { + Type: EventTypeConditionReady, + Status: corev1.ConditionUnknown, + }}, + }, + }, + }, { + name: "one true", + ets: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{{ + Type: EventTypeConditionBrokerReady, + Status: corev1.ConditionTrue, + }}, + }, + }, + want: &EventTypeStatus{ + Status: duckv1alpha1.Status{ + Conditions: []duckv1alpha1.Condition{{ + Type: EventTypeConditionBrokerExists, + Status: corev1.ConditionUnknown, + }, { + Type: EventTypeConditionBrokerReady, + Status: corev1.ConditionTrue, + }, { + Type: EventTypeConditionReady, + Status: corev1.ConditionUnknown, + }}, + }, + }}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + test.ets.InitializeConditions() + if diff := cmp.Diff(test.want, test.ets, ignoreAllButTypeAndStatus); diff != "" { + t.Errorf("unexpected conditions (-want, +got) = %v", diff) + } + }) + } +} + +func TestEventTypeIsReady(t *testing.T) { + tests := []struct { + name string + markBrokerExists *bool + markBrokerReady *bool + wantReady bool + }{{ + name: "all happy", + markBrokerExists: &trueValue, + markBrokerReady: &trueValue, + wantReady: true, + }, { + name: "broker exist sad", + markBrokerExists: &falseValue, + markBrokerReady: &trueValue, + wantReady: false, + }, { + name: "broker ready sad", + markBrokerExists: &trueValue, + markBrokerReady: &falseValue, + wantReady: false, + }, { + name: "all sad", + markBrokerExists: &falseValue, + markBrokerReady: &falseValue, + wantReady: false, + }} + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + ets := &EventTypeStatus{} + if test.markBrokerExists != nil { + if *test.markBrokerExists { + ets.MarkBrokerExists() + } else { + ets.MarkBrokerDoesNotExist() + } + } + if test.markBrokerReady != nil { + if *test.markBrokerReady { + ets.MarkBrokerReady() + } else { + ets.MarkBrokerNotReady() + } + } + + got := ets.IsReady() + if test.wantReady != got { + t.Errorf("unexpected readiness: want %v, got %v", test.wantReady, got) + } + }) + } +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_types.go b/pkg/apis/eventing/v1alpha1/eventtype_types.go new file mode 100644 index 00000000000..51e3ac5c4fd --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_types.go @@ -0,0 +1,83 @@ +/* + * Copyright 2019 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 v1alpha1 + +import ( + "github.com/knative/pkg/apis" + duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1" + "github.com/knative/pkg/webhook" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" +) + +// +genclient +// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + +type EventType struct { + metav1.TypeMeta `json:",inline"` + // +optional + metav1.ObjectMeta `json:"metadata,omitempty"` + + // Spec defines the desired state of the EventType. + Spec EventTypeSpec `json:"spec,omitempty"` + + // Status represents the current state of the EventType. + // This data may be out of date. + // +optional + Status EventTypeStatus `json:"status,omitempty"` +} + +// Check that EventType can be validated, can be defaulted, and has immutable fields. +var _ apis.Validatable = (*EventType)(nil) +var _ apis.Defaultable = (*EventType)(nil) +var _ apis.Immutable = (*EventType)(nil) +var _ runtime.Object = (*EventType)(nil) +var _ webhook.GenericCRD = (*EventType)(nil) + +type EventTypeSpec struct { + // Type represents the CloudEvents type. It is authoritative. + Type string `json:"type"` + // Source is a URI, it represents the CloudEvents source. + Source string `json:"source"` + // Schema is a URI, it represents the CloudEvents schemaurl extension attribute. + // It may be a JSON schema, a protobuf schema, etc. It is optional. + // +optional + Schema string `json:"schema,omitempty"` + // Broker refers to the Broker that can provide the EventType. + Broker string `json:"broker"` + // Description is an optional field used to describe the EventType, in any meaningful way. + // +optional + Description string `json:description,omitempty` +} + +// EventTypeStatus represents the current state of a EventType. +type EventTypeStatus struct { + // inherits duck/v1alpha1 Status, which currently provides: + // * ObservedGeneration - the 'Generation' of the Service that was last processed by the controller. + // * Conditions - the latest available observations of a resource's current state. + duckv1alpha1.Status `json:",inline"` +} + +// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object + +// EventTypeList is a collection of EventTypes. +type EventTypeList struct { + metav1.TypeMeta `json:",inline"` + // +optional + metav1.ListMeta `json:"metadata,omitempty"` + Items []EventType `json:"items"` +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_validation.go b/pkg/apis/eventing/v1alpha1/eventtype_validation.go new file mode 100644 index 00000000000..a162f7dc920 --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_validation.go @@ -0,0 +1,77 @@ +/* +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 + + 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" + + "github.com/google/go-cmp/cmp/cmpopts" + + "github.com/knative/pkg/apis" + "github.com/knative/pkg/kmp" +) + +func (et *EventType) Validate(ctx context.Context) *apis.FieldError { + return et.Spec.Validate(ctx).ViaField("spec") +} + +func (ets *EventTypeSpec) Validate(ctx context.Context) *apis.FieldError { + var errs *apis.FieldError + if ets.Type == "" { + fe := apis.ErrMissingField("type") + errs = errs.Also(fe) + } + if ets.Source == "" { + // TODO validate is a valid URI. + fe := apis.ErrMissingField("source") + errs = errs.Also(fe) + } + if ets.Broker == "" { + fe := apis.ErrMissingField("broker") + errs = errs.Also(fe) + } + // TODO validate Schema is a valid URI. + return errs +} + +func (et *EventType) CheckImmutableFields(ctx context.Context, og apis.Immutable) *apis.FieldError { + if og == nil { + return nil + } + + original, ok := og.(*EventType) + if !ok { + return &apis.FieldError{Message: "The provided original was not an EventType"} + } + + // All but Description field immutable. + ignoreArguments := cmpopts.IgnoreFields(EventTypeSpec{}, "Description") + if diff, err := kmp.ShortDiff(original.Spec, et.Spec, ignoreArguments); err != nil { + return &apis.FieldError{ + Message: "Failed to diff EventType", + Paths: []string{"spec"}, + Details: err.Error(), + } + } else if diff != "" { + return &apis.FieldError{ + Message: "Immutable fields changed (-old +new)", + Paths: []string{"spec"}, + Details: diff, + } + } + return nil +} diff --git a/pkg/apis/eventing/v1alpha1/eventtype_validation_test.go b/pkg/apis/eventing/v1alpha1/eventtype_validation_test.go new file mode 100644 index 00000000000..142429fc826 --- /dev/null +++ b/pkg/apis/eventing/v1alpha1/eventtype_validation_test.go @@ -0,0 +1,279 @@ +/* +Copyright 2019 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 v1alpha1 + +import ( + "context" + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/knative/pkg/apis" +) + +func TestEventTypeValidation(t *testing.T) { + name := "invalid type and source and broker" + broker := &EventType{Spec: EventTypeSpec{}} + + want := &apis.FieldError{ + Paths: []string{"spec.type", "spec.source", "spec.broker"}, + Message: "missing field(s)", + } + + t.Run(name, func(t *testing.T) { + got := broker.Validate(context.TODO()) + if diff := cmp.Diff(want.Error(), got.Error()); diff != "" { + t.Errorf("EventType.Validate (-want, +got) = %v", diff) + } + }) +} + +func TestEventTypeSpecValidation(t *testing.T) { + tests := []struct { + name string + ets *EventTypeSpec + want *apis.FieldError + }{{ + name: "invalid eventtype spec", + ets: &EventTypeSpec{}, + want: func() *apis.FieldError { + fe := apis.ErrMissingField("type", "source", "broker") + return fe + }(), + }, { + name: "invalid eventtype type", + ets: &EventTypeSpec{ + Source: "test-source", + Broker: "test-broker", + }, + want: func() *apis.FieldError { + fe := apis.ErrMissingField("type") + return fe + }(), + }, { + name: "invalid eventtype source", + ets: &EventTypeSpec{ + Type: "test-type", + Broker: "test-broker", + }, + want: func() *apis.FieldError { + fe := apis.ErrMissingField("source") + return fe + }(), + }, { + name: "invalid eventtype broker", + ets: &EventTypeSpec{ + Type: "test-type", + Source: "test-source", + }, + want: func() *apis.FieldError { + fe := apis.ErrMissingField("broker") + return fe + }(), + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := test.ets.Validate(context.TODO()) + if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" { + t.Errorf("%s: Validate EventTypeSpec (-want, +got) = %v", test.name, diff) + } + }) + } +} + +func TestEventTypeImmutableFields(t *testing.T) { + tests := []struct { + name string + current apis.Immutable + original apis.Immutable + want *apis.FieldError + }{{ + name: "good (no change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + }, + }, + want: nil, + }, { + name: "new nil is ok", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + }, + }, + original: nil, + want: nil, + }, { + name: "invalid type", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + }, + }, + original: &Trigger{}, + want: &apis.FieldError{ + Message: "The provided original was not an EventType", + }, + }, { + name: "bad (broker change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "original-broker", + }, + }, + want: &apis.FieldError{ + Message: "Immutable fields changed (-old +new)", + Paths: []string{"spec"}, + Details: `{v1alpha1.EventTypeSpec}.Broker: + -: "original-broker" + +: "test-broker" +`, + }, + }, { + name: "bad (type change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "original-type", + Source: "test-source", + Broker: "test-broker", + }, + }, + want: &apis.FieldError{ + Message: "Immutable fields changed (-old +new)", + Paths: []string{"spec"}, + Details: `{v1alpha1.EventTypeSpec}.Type: + -: "original-type" + +: "test-type" +`, + }, + }, { + name: "bad (source change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "original-source", + Broker: "test-broker", + }, + }, + want: &apis.FieldError{ + Message: "Immutable fields changed (-old +new)", + Paths: []string{"spec"}, + Details: `{v1alpha1.EventTypeSpec}.Source: + -: "original-source" + +: "test-source" +`, + }, + }, { + name: "bad (schema change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "original-schema", + }, + }, + want: &apis.FieldError{ + Message: "Immutable fields changed (-old +new)", + Paths: []string{"spec"}, + Details: `{v1alpha1.EventTypeSpec}.Schema: + -: "original-schema" + +: "test-schema" +`, + }, + }, { + name: "good (description change)", + current: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + Description: "test-description", + }, + }, + original: &EventType{ + Spec: EventTypeSpec{ + Type: "test-type", + Source: "test-source", + Broker: "test-broker", + Schema: "test-schema", + Description: "original-description", + }, + }, + want: nil, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + got := test.current.CheckImmutableFields(context.TODO(), test.original) + if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" { + t.Errorf("CheckImmutableFields (-want, +got) = %v", diff) + } + }) + } +} diff --git a/pkg/apis/eventing/v1alpha1/register.go b/pkg/apis/eventing/v1alpha1/register.go index fb3a5292623..977cad56aaf 100644 --- a/pkg/apis/eventing/v1alpha1/register.go +++ b/pkg/apis/eventing/v1alpha1/register.go @@ -55,6 +55,8 @@ func addKnownTypes(scheme *runtime.Scheme) error { &SubscriptionList{}, &Trigger{}, &TriggerList{}, + &EventType{}, + &EventTypeList{}, ) metav1.AddToGroupVersion(scheme, SchemeGroupVersion) return nil diff --git a/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go index 9807e6caf2a..efddd131c82 100644 --- a/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go +++ b/pkg/apis/eventing/v1alpha1/zz_generated.deepcopy.go @@ -342,6 +342,100 @@ func (in *ClusterChannelProvisionerStatus) DeepCopy() *ClusterChannelProvisioner return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EventType) DeepCopyInto(out *EventType) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + out.Spec = in.Spec + in.Status.DeepCopyInto(&out.Status) + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EventType. +func (in *EventType) DeepCopy() *EventType { + if in == nil { + return nil + } + out := new(EventType) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EventType) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EventTypeList) DeepCopyInto(out *EventTypeList) { + *out = *in + out.TypeMeta = in.TypeMeta + out.ListMeta = in.ListMeta + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]EventType, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EventTypeList. +func (in *EventTypeList) DeepCopy() *EventTypeList { + if in == nil { + return nil + } + out := new(EventTypeList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EventTypeList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EventTypeSpec) DeepCopyInto(out *EventTypeSpec) { + *out = *in + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EventTypeSpec. +func (in *EventTypeSpec) DeepCopy() *EventTypeSpec { + if in == nil { + return nil + } + out := new(EventTypeSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EventTypeStatus) DeepCopyInto(out *EventTypeStatus) { + *out = *in + in.Status.DeepCopyInto(&out.Status) + return +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EventTypeStatus. +func (in *EventTypeStatus) DeepCopy() *EventTypeStatus { + if in == nil { + return nil + } + out := new(EventTypeStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *ReplyStrategy) DeepCopyInto(out *ReplyStrategy) { *out = *in diff --git a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventing_client.go b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventing_client.go index 68722de4481..4af33411504 100644 --- a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventing_client.go +++ b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventing_client.go @@ -30,6 +30,7 @@ type EventingV1alpha1Interface interface { BrokersGetter ChannelsGetter ClusterChannelProvisionersGetter + EventTypesGetter SubscriptionsGetter TriggersGetter } @@ -51,6 +52,10 @@ func (c *EventingV1alpha1Client) ClusterChannelProvisioners() ClusterChannelProv return newClusterChannelProvisioners(c) } +func (c *EventingV1alpha1Client) EventTypes(namespace string) EventTypeInterface { + return newEventTypes(c, namespace) +} + func (c *EventingV1alpha1Client) Subscriptions(namespace string) SubscriptionInterface { return newSubscriptions(c, namespace) } diff --git a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventtype.go b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventtype.go new file mode 100644 index 00000000000..ff71fb35749 --- /dev/null +++ b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/eventtype.go @@ -0,0 +1,174 @@ +/* +Copyright 2019 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. +*/ + +// Code generated by client-gen. DO NOT EDIT. + +package v1alpha1 + +import ( + v1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + scheme "github.com/knative/eventing/pkg/client/clientset/versioned/scheme" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + types "k8s.io/apimachinery/pkg/types" + watch "k8s.io/apimachinery/pkg/watch" + rest "k8s.io/client-go/rest" +) + +// EventTypesGetter has a method to return a EventTypeInterface. +// A group's client should implement this interface. +type EventTypesGetter interface { + EventTypes(namespace string) EventTypeInterface +} + +// EventTypeInterface has methods to work with EventType resources. +type EventTypeInterface interface { + Create(*v1alpha1.EventType) (*v1alpha1.EventType, error) + Update(*v1alpha1.EventType) (*v1alpha1.EventType, error) + UpdateStatus(*v1alpha1.EventType) (*v1alpha1.EventType, error) + Delete(name string, options *v1.DeleteOptions) error + DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error + Get(name string, options v1.GetOptions) (*v1alpha1.EventType, error) + List(opts v1.ListOptions) (*v1alpha1.EventTypeList, error) + Watch(opts v1.ListOptions) (watch.Interface, error) + Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.EventType, err error) + EventTypeExpansion +} + +// eventTypes implements EventTypeInterface +type eventTypes struct { + client rest.Interface + ns string +} + +// newEventTypes returns a EventTypes +func newEventTypes(c *EventingV1alpha1Client, namespace string) *eventTypes { + return &eventTypes{ + client: c.RESTClient(), + ns: namespace, + } +} + +// Get takes name of the eventType, and returns the corresponding eventType object, and an error if there is any. +func (c *eventTypes) Get(name string, options v1.GetOptions) (result *v1alpha1.EventType, err error) { + result = &v1alpha1.EventType{} + err = c.client.Get(). + Namespace(c.ns). + Resource("eventtypes"). + Name(name). + VersionedParams(&options, scheme.ParameterCodec). + Do(). + Into(result) + return +} + +// List takes label and field selectors, and returns the list of EventTypes that match those selectors. +func (c *eventTypes) List(opts v1.ListOptions) (result *v1alpha1.EventTypeList, err error) { + result = &v1alpha1.EventTypeList{} + err = c.client.Get(). + Namespace(c.ns). + Resource("eventtypes"). + VersionedParams(&opts, scheme.ParameterCodec). + Do(). + Into(result) + return +} + +// Watch returns a watch.Interface that watches the requested eventTypes. +func (c *eventTypes) Watch(opts v1.ListOptions) (watch.Interface, error) { + opts.Watch = true + return c.client.Get(). + Namespace(c.ns). + Resource("eventtypes"). + VersionedParams(&opts, scheme.ParameterCodec). + Watch() +} + +// Create takes the representation of a eventType and creates it. Returns the server's representation of the eventType, and an error, if there is any. +func (c *eventTypes) Create(eventType *v1alpha1.EventType) (result *v1alpha1.EventType, err error) { + result = &v1alpha1.EventType{} + err = c.client.Post(). + Namespace(c.ns). + Resource("eventtypes"). + Body(eventType). + Do(). + Into(result) + return +} + +// Update takes the representation of a eventType and updates it. Returns the server's representation of the eventType, and an error, if there is any. +func (c *eventTypes) Update(eventType *v1alpha1.EventType) (result *v1alpha1.EventType, err error) { + result = &v1alpha1.EventType{} + err = c.client.Put(). + Namespace(c.ns). + Resource("eventtypes"). + Name(eventType.Name). + Body(eventType). + Do(). + Into(result) + return +} + +// UpdateStatus was generated because the type contains a Status member. +// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus(). + +func (c *eventTypes) UpdateStatus(eventType *v1alpha1.EventType) (result *v1alpha1.EventType, err error) { + result = &v1alpha1.EventType{} + err = c.client.Put(). + Namespace(c.ns). + Resource("eventtypes"). + Name(eventType.Name). + SubResource("status"). + Body(eventType). + Do(). + Into(result) + return +} + +// Delete takes name of the eventType and deletes it. Returns an error if one occurs. +func (c *eventTypes) Delete(name string, options *v1.DeleteOptions) error { + return c.client.Delete(). + Namespace(c.ns). + Resource("eventtypes"). + Name(name). + Body(options). + Do(). + Error() +} + +// DeleteCollection deletes a collection of objects. +func (c *eventTypes) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error { + return c.client.Delete(). + Namespace(c.ns). + Resource("eventtypes"). + VersionedParams(&listOptions, scheme.ParameterCodec). + Body(options). + Do(). + Error() +} + +// Patch applies the patch and returns the patched eventType. +func (c *eventTypes) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.EventType, err error) { + result = &v1alpha1.EventType{} + err = c.client.Patch(pt). + Namespace(c.ns). + Resource("eventtypes"). + SubResource(subresources...). + Name(name). + Body(data). + Do(). + Into(result) + return +} diff --git a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventing_client.go b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventing_client.go index 52759cda929..75da888a5e6 100644 --- a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventing_client.go +++ b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventing_client.go @@ -40,6 +40,10 @@ func (c *FakeEventingV1alpha1) ClusterChannelProvisioners() v1alpha1.ClusterChan return &FakeClusterChannelProvisioners{c} } +func (c *FakeEventingV1alpha1) EventTypes(namespace string) v1alpha1.EventTypeInterface { + return &FakeEventTypes{c, namespace} +} + func (c *FakeEventingV1alpha1) Subscriptions(namespace string) v1alpha1.SubscriptionInterface { return &FakeSubscriptions{c, namespace} } diff --git a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventtype.go b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventtype.go new file mode 100644 index 00000000000..4958426561e --- /dev/null +++ b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/fake/fake_eventtype.go @@ -0,0 +1,140 @@ +/* +Copyright 2019 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. +*/ + +// Code generated by client-gen. DO NOT EDIT. + +package fake + +import ( + v1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + labels "k8s.io/apimachinery/pkg/labels" + schema "k8s.io/apimachinery/pkg/runtime/schema" + types "k8s.io/apimachinery/pkg/types" + watch "k8s.io/apimachinery/pkg/watch" + testing "k8s.io/client-go/testing" +) + +// FakeEventTypes implements EventTypeInterface +type FakeEventTypes struct { + Fake *FakeEventingV1alpha1 + ns string +} + +var eventtypesResource = schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1alpha1", Resource: "eventtypes"} + +var eventtypesKind = schema.GroupVersionKind{Group: "eventing.knative.dev", Version: "v1alpha1", Kind: "EventType"} + +// Get takes name of the eventType, and returns the corresponding eventType object, and an error if there is any. +func (c *FakeEventTypes) Get(name string, options v1.GetOptions) (result *v1alpha1.EventType, err error) { + obj, err := c.Fake. + Invokes(testing.NewGetAction(eventtypesResource, c.ns, name), &v1alpha1.EventType{}) + + if obj == nil { + return nil, err + } + return obj.(*v1alpha1.EventType), err +} + +// List takes label and field selectors, and returns the list of EventTypes that match those selectors. +func (c *FakeEventTypes) List(opts v1.ListOptions) (result *v1alpha1.EventTypeList, err error) { + obj, err := c.Fake. + Invokes(testing.NewListAction(eventtypesResource, eventtypesKind, c.ns, opts), &v1alpha1.EventTypeList{}) + + if obj == nil { + return nil, err + } + + label, _, _ := testing.ExtractFromListOptions(opts) + if label == nil { + label = labels.Everything() + } + list := &v1alpha1.EventTypeList{ListMeta: obj.(*v1alpha1.EventTypeList).ListMeta} + for _, item := range obj.(*v1alpha1.EventTypeList).Items { + if label.Matches(labels.Set(item.Labels)) { + list.Items = append(list.Items, item) + } + } + return list, err +} + +// Watch returns a watch.Interface that watches the requested eventTypes. +func (c *FakeEventTypes) Watch(opts v1.ListOptions) (watch.Interface, error) { + return c.Fake. + InvokesWatch(testing.NewWatchAction(eventtypesResource, c.ns, opts)) + +} + +// Create takes the representation of a eventType and creates it. Returns the server's representation of the eventType, and an error, if there is any. +func (c *FakeEventTypes) Create(eventType *v1alpha1.EventType) (result *v1alpha1.EventType, err error) { + obj, err := c.Fake. + Invokes(testing.NewCreateAction(eventtypesResource, c.ns, eventType), &v1alpha1.EventType{}) + + if obj == nil { + return nil, err + } + return obj.(*v1alpha1.EventType), err +} + +// Update takes the representation of a eventType and updates it. Returns the server's representation of the eventType, and an error, if there is any. +func (c *FakeEventTypes) Update(eventType *v1alpha1.EventType) (result *v1alpha1.EventType, err error) { + obj, err := c.Fake. + Invokes(testing.NewUpdateAction(eventtypesResource, c.ns, eventType), &v1alpha1.EventType{}) + + if obj == nil { + return nil, err + } + return obj.(*v1alpha1.EventType), err +} + +// UpdateStatus was generated because the type contains a Status member. +// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus(). +func (c *FakeEventTypes) UpdateStatus(eventType *v1alpha1.EventType) (*v1alpha1.EventType, error) { + obj, err := c.Fake. + Invokes(testing.NewUpdateSubresourceAction(eventtypesResource, "status", c.ns, eventType), &v1alpha1.EventType{}) + + if obj == nil { + return nil, err + } + return obj.(*v1alpha1.EventType), err +} + +// Delete takes name of the eventType and deletes it. Returns an error if one occurs. +func (c *FakeEventTypes) Delete(name string, options *v1.DeleteOptions) error { + _, err := c.Fake. + Invokes(testing.NewDeleteAction(eventtypesResource, c.ns, name), &v1alpha1.EventType{}) + + return err +} + +// DeleteCollection deletes a collection of objects. +func (c *FakeEventTypes) DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error { + action := testing.NewDeleteCollectionAction(eventtypesResource, c.ns, listOptions) + + _, err := c.Fake.Invokes(action, &v1alpha1.EventTypeList{}) + return err +} + +// Patch applies the patch and returns the patched eventType. +func (c *FakeEventTypes) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *v1alpha1.EventType, err error) { + obj, err := c.Fake. + Invokes(testing.NewPatchSubresourceAction(eventtypesResource, c.ns, name, data, subresources...), &v1alpha1.EventType{}) + + if obj == nil { + return nil, err + } + return obj.(*v1alpha1.EventType), err +} diff --git a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/generated_expansion.go b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/generated_expansion.go index 00ba65313fd..c33c3d484f4 100644 --- a/pkg/client/clientset/versioned/typed/eventing/v1alpha1/generated_expansion.go +++ b/pkg/client/clientset/versioned/typed/eventing/v1alpha1/generated_expansion.go @@ -24,6 +24,8 @@ type ChannelExpansion interface{} type ClusterChannelProvisionerExpansion interface{} +type EventTypeExpansion interface{} + type SubscriptionExpansion interface{} type TriggerExpansion interface{} diff --git a/pkg/client/informers/externalversions/eventing/v1alpha1/eventtype.go b/pkg/client/informers/externalversions/eventing/v1alpha1/eventtype.go new file mode 100644 index 00000000000..52b41ce918e --- /dev/null +++ b/pkg/client/informers/externalversions/eventing/v1alpha1/eventtype.go @@ -0,0 +1,89 @@ +/* +Copyright 2019 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. +*/ + +// Code generated by informer-gen. DO NOT EDIT. + +package v1alpha1 + +import ( + time "time" + + eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + versioned "github.com/knative/eventing/pkg/client/clientset/versioned" + internalinterfaces "github.com/knative/eventing/pkg/client/informers/externalversions/internalinterfaces" + v1alpha1 "github.com/knative/eventing/pkg/client/listers/eventing/v1alpha1" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + runtime "k8s.io/apimachinery/pkg/runtime" + watch "k8s.io/apimachinery/pkg/watch" + cache "k8s.io/client-go/tools/cache" +) + +// EventTypeInformer provides access to a shared informer and lister for +// EventTypes. +type EventTypeInformer interface { + Informer() cache.SharedIndexInformer + Lister() v1alpha1.EventTypeLister +} + +type eventTypeInformer struct { + factory internalinterfaces.SharedInformerFactory + tweakListOptions internalinterfaces.TweakListOptionsFunc + namespace string +} + +// NewEventTypeInformer constructs a new informer for EventType type. +// Always prefer using an informer factory to get a shared informer instead of getting an independent +// one. This reduces memory footprint and number of connections to the server. +func NewEventTypeInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers) cache.SharedIndexInformer { + return NewFilteredEventTypeInformer(client, namespace, resyncPeriod, indexers, nil) +} + +// NewFilteredEventTypeInformer constructs a new informer for EventType type. +// Always prefer using an informer factory to get a shared informer instead of getting an independent +// one. This reduces memory footprint and number of connections to the server. +func NewFilteredEventTypeInformer(client versioned.Interface, namespace string, resyncPeriod time.Duration, indexers cache.Indexers, tweakListOptions internalinterfaces.TweakListOptionsFunc) cache.SharedIndexInformer { + return cache.NewSharedIndexInformer( + &cache.ListWatch{ + ListFunc: func(options v1.ListOptions) (runtime.Object, error) { + if tweakListOptions != nil { + tweakListOptions(&options) + } + return client.EventingV1alpha1().EventTypes(namespace).List(options) + }, + WatchFunc: func(options v1.ListOptions) (watch.Interface, error) { + if tweakListOptions != nil { + tweakListOptions(&options) + } + return client.EventingV1alpha1().EventTypes(namespace).Watch(options) + }, + }, + &eventingv1alpha1.EventType{}, + resyncPeriod, + indexers, + ) +} + +func (f *eventTypeInformer) defaultInformer(client versioned.Interface, resyncPeriod time.Duration) cache.SharedIndexInformer { + return NewFilteredEventTypeInformer(client, f.namespace, resyncPeriod, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc}, f.tweakListOptions) +} + +func (f *eventTypeInformer) Informer() cache.SharedIndexInformer { + return f.factory.InformerFor(&eventingv1alpha1.EventType{}, f.defaultInformer) +} + +func (f *eventTypeInformer) Lister() v1alpha1.EventTypeLister { + return v1alpha1.NewEventTypeLister(f.Informer().GetIndexer()) +} diff --git a/pkg/client/informers/externalversions/eventing/v1alpha1/interface.go b/pkg/client/informers/externalversions/eventing/v1alpha1/interface.go index 004212a1b29..614c6d2e90a 100644 --- a/pkg/client/informers/externalversions/eventing/v1alpha1/interface.go +++ b/pkg/client/informers/externalversions/eventing/v1alpha1/interface.go @@ -30,6 +30,8 @@ type Interface interface { Channels() ChannelInformer // ClusterChannelProvisioners returns a ClusterChannelProvisionerInformer. ClusterChannelProvisioners() ClusterChannelProvisionerInformer + // EventTypes returns a EventTypeInformer. + EventTypes() EventTypeInformer // Subscriptions returns a SubscriptionInformer. Subscriptions() SubscriptionInformer // Triggers returns a TriggerInformer. @@ -62,6 +64,11 @@ func (v *version) ClusterChannelProvisioners() ClusterChannelProvisionerInformer return &clusterChannelProvisionerInformer{factory: v.factory, tweakListOptions: v.tweakListOptions} } +// EventTypes returns a EventTypeInformer. +func (v *version) EventTypes() EventTypeInformer { + return &eventTypeInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions} +} + // Subscriptions returns a SubscriptionInformer. func (v *version) Subscriptions() SubscriptionInformer { return &subscriptionInformer{factory: v.factory, namespace: v.namespace, tweakListOptions: v.tweakListOptions} diff --git a/pkg/client/informers/externalversions/generic.go b/pkg/client/informers/externalversions/generic.go index 48066085062..fbcb3b3bcba 100644 --- a/pkg/client/informers/externalversions/generic.go +++ b/pkg/client/informers/externalversions/generic.go @@ -60,6 +60,8 @@ func (f *sharedInformerFactory) ForResource(resource schema.GroupVersionResource return &genericInformer{resource: resource.GroupResource(), informer: f.Eventing().V1alpha1().Channels().Informer()}, nil case v1alpha1.SchemeGroupVersion.WithResource("clusterchannelprovisioners"): return &genericInformer{resource: resource.GroupResource(), informer: f.Eventing().V1alpha1().ClusterChannelProvisioners().Informer()}, nil + case v1alpha1.SchemeGroupVersion.WithResource("eventtypes"): + return &genericInformer{resource: resource.GroupResource(), informer: f.Eventing().V1alpha1().EventTypes().Informer()}, nil case v1alpha1.SchemeGroupVersion.WithResource("subscriptions"): return &genericInformer{resource: resource.GroupResource(), informer: f.Eventing().V1alpha1().Subscriptions().Informer()}, nil case v1alpha1.SchemeGroupVersion.WithResource("triggers"): diff --git a/pkg/client/listers/eventing/v1alpha1/eventtype.go b/pkg/client/listers/eventing/v1alpha1/eventtype.go new file mode 100644 index 00000000000..ad549bd981b --- /dev/null +++ b/pkg/client/listers/eventing/v1alpha1/eventtype.go @@ -0,0 +1,94 @@ +/* +Copyright 2019 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. +*/ + +// Code generated by lister-gen. DO NOT EDIT. + +package v1alpha1 + +import ( + v1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/client-go/tools/cache" +) + +// EventTypeLister helps list EventTypes. +type EventTypeLister interface { + // List lists all EventTypes in the indexer. + List(selector labels.Selector) (ret []*v1alpha1.EventType, err error) + // EventTypes returns an object that can list and get EventTypes. + EventTypes(namespace string) EventTypeNamespaceLister + EventTypeListerExpansion +} + +// eventTypeLister implements the EventTypeLister interface. +type eventTypeLister struct { + indexer cache.Indexer +} + +// NewEventTypeLister returns a new EventTypeLister. +func NewEventTypeLister(indexer cache.Indexer) EventTypeLister { + return &eventTypeLister{indexer: indexer} +} + +// List lists all EventTypes in the indexer. +func (s *eventTypeLister) List(selector labels.Selector) (ret []*v1alpha1.EventType, err error) { + err = cache.ListAll(s.indexer, selector, func(m interface{}) { + ret = append(ret, m.(*v1alpha1.EventType)) + }) + return ret, err +} + +// EventTypes returns an object that can list and get EventTypes. +func (s *eventTypeLister) EventTypes(namespace string) EventTypeNamespaceLister { + return eventTypeNamespaceLister{indexer: s.indexer, namespace: namespace} +} + +// EventTypeNamespaceLister helps list and get EventTypes. +type EventTypeNamespaceLister interface { + // List lists all EventTypes in the indexer for a given namespace. + List(selector labels.Selector) (ret []*v1alpha1.EventType, err error) + // Get retrieves the EventType from the indexer for a given namespace and name. + Get(name string) (*v1alpha1.EventType, error) + EventTypeNamespaceListerExpansion +} + +// eventTypeNamespaceLister implements the EventTypeNamespaceLister +// interface. +type eventTypeNamespaceLister struct { + indexer cache.Indexer + namespace string +} + +// List lists all EventTypes in the indexer for a given namespace. +func (s eventTypeNamespaceLister) List(selector labels.Selector) (ret []*v1alpha1.EventType, err error) { + err = cache.ListAllByNamespace(s.indexer, s.namespace, selector, func(m interface{}) { + ret = append(ret, m.(*v1alpha1.EventType)) + }) + return ret, err +} + +// Get retrieves the EventType from the indexer for a given namespace and name. +func (s eventTypeNamespaceLister) Get(name string) (*v1alpha1.EventType, error) { + obj, exists, err := s.indexer.GetByKey(s.namespace + "/" + name) + if err != nil { + return nil, err + } + if !exists { + return nil, errors.NewNotFound(v1alpha1.Resource("eventtype"), name) + } + return obj.(*v1alpha1.EventType), nil +} diff --git a/pkg/client/listers/eventing/v1alpha1/expansion_generated.go b/pkg/client/listers/eventing/v1alpha1/expansion_generated.go index 4bd2b3c78eb..5f34cbf332d 100644 --- a/pkg/client/listers/eventing/v1alpha1/expansion_generated.go +++ b/pkg/client/listers/eventing/v1alpha1/expansion_generated.go @@ -38,6 +38,14 @@ type ChannelNamespaceListerExpansion interface{} // ClusterChannelProvisionerLister. type ClusterChannelProvisionerListerExpansion interface{} +// EventTypeListerExpansion allows custom methods to be added to +// EventTypeLister. +type EventTypeListerExpansion interface{} + +// EventTypeNamespaceListerExpansion allows custom methods to be added to +// EventTypeNamespaceLister. +type EventTypeNamespaceListerExpansion interface{} + // SubscriptionListerExpansion allows custom methods to be added to // SubscriptionLister. type SubscriptionListerExpansion interface{} diff --git a/pkg/reconciler/eventtype/eventtype.go b/pkg/reconciler/eventtype/eventtype.go new file mode 100644 index 00000000000..a1aa28113a8 --- /dev/null +++ b/pkg/reconciler/eventtype/eventtype.go @@ -0,0 +1,211 @@ +/* +Copyright 2019 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 eventtype + +import ( + "context" + "fmt" + "reflect" + "time" + + "github.com/knative/eventing/pkg/utils" + "github.com/knative/pkg/tracker" + + "k8s.io/client-go/tools/cache" + + "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + eventinginformers "github.com/knative/eventing/pkg/client/informers/externalversions/eventing/v1alpha1" + listers "github.com/knative/eventing/pkg/client/listers/eventing/v1alpha1" + "github.com/knative/eventing/pkg/logging" + "github.com/knative/eventing/pkg/reconciler" + "github.com/knative/pkg/controller" + "go.uber.org/zap" + corev1 "k8s.io/api/core/v1" + apierrs "k8s.io/apimachinery/pkg/api/errors" +) + +const ( + // ReconcilerName is the name of the reconciler. + ReconcilerName = "EventTypes" + // controllerAgentName is the string used by this controller to identify + // itself when creating events. + controllerAgentName = "eventtype-controller" + + // Name of the corev1.Events emitted from the reconciliation process. + eventTypeReadinessChanged = "EventTypeReadinessChanged" + eventTypeReconcileFailed = "EventTypeReconcileFailed" + eventTypeUpdateStatusFailed = "EventTypeUpdateStatusFailed" +) + +type Reconciler struct { + *reconciler.Base + + // listers index properties about resources + eventTypeLister listers.EventTypeLister + brokerLister listers.BrokerLister + tracker tracker.Interface +} + +var brokerGVK = v1alpha1.SchemeGroupVersion.WithKind("Broker") + +// Check that our Reconciler implements controller.Reconciler +var _ controller.Reconciler = (*Reconciler)(nil) + +// NewController initializes the controller and is called by the generated code +// Registers event handlers to enqueue events +func NewController( + opt reconciler.Options, + eventTypeInformer eventinginformers.EventTypeInformer, + brokerInformer eventinginformers.BrokerInformer, +) *controller.Impl { + + r := &Reconciler{ + Base: reconciler.NewBase(opt, controllerAgentName), + eventTypeLister: eventTypeInformer.Lister(), + brokerLister: brokerInformer.Lister(), + } + impl := controller.NewImpl(r, r.Logger, ReconcilerName, reconciler.MustNewStatsReporter(ReconcilerName, r.Logger)) + + r.Logger.Info("Setting up event handlers") + eventTypeInformer.Informer().AddEventHandler(reconciler.Handler(impl.Enqueue)) + + // Tracker is used to notify us that a EventType's Broker has changed so that + // we can reconcile. + r.tracker = tracker.New(impl.EnqueueKey, opt.GetTrackerLease()) + brokerInformer.Informer().AddEventHandler(reconciler.Handler( + controller.EnsureTypeMeta( + r.tracker.OnChanged, + v1alpha1.SchemeGroupVersion.WithKind("Broker"), + ), + )) + + return impl +} + +// Reconcile compares the actual state with the desired, and attempts to +// converge the two. It then updates the Status block of the EventType resource +// with the current status of the resource. +func (r *Reconciler) Reconcile(ctx context.Context, key string) error { + // Convert the namespace/name string into a distinct namespace and name + namespace, name, err := cache.SplitMetaNamespaceKey(key) + if err != nil { + r.Logger.Errorf("invalid resource key: %s", key) + return nil + } + ctx = logging.WithLogger(ctx, r.Logger.Desugar().With(zap.String("key", key))) + + // Get the EventType resource with this namespace/name + original, err := r.eventTypeLister.EventTypes(namespace).Get(name) + if apierrs.IsNotFound(err) { + // The resource may no longer exist, in which case we stop processing. + logging.FromContext(ctx).Info("eventType key in work queue no longer exists") + return nil + } else if err != nil { + return err + } + + // Don't modify the informers copy + eventType := original.DeepCopy() + + // Reconcile this copy of the EventType and then write back any status + // updates regardless of whether the reconcile error out. + reconcileErr := r.reconcile(ctx, eventType) + if reconcileErr != nil { + logging.FromContext(ctx).Warn("Error reconciling Broker", zap.Error(err)) + r.Recorder.Eventf(eventType, corev1.EventTypeWarning, eventTypeReconcileFailed, fmt.Sprintf("EventType reconcile error: %v", reconcileErr)) + } else { + logging.FromContext(ctx).Debug("EventType reconciled") + } + + if _, err = r.updateStatus(ctx, eventType); err != nil { + logging.FromContext(ctx).Warn("Failed to update the EventType status", zap.Error(err)) + r.Recorder.Eventf(eventType, corev1.EventTypeWarning, eventTypeUpdateStatusFailed, "Failed to update Broker's status: %v", err) + return err + } + + // Requeue if the resource is not ready: + return reconcileErr +} + +func (r *Reconciler) reconcile(ctx context.Context, et *v1alpha1.EventType) error { + et.Status.InitializeConditions() + + // 1. Verify the Broker exists. + // 2. Verify the Broker is ready. + + if et.DeletionTimestamp != nil { + // Everything is cleaned up by the garbage collector. + return nil + } + + b, err := r.getBroker(ctx, et) + if err != nil { + logging.FromContext(ctx).Error("Unable to get the Broker", zap.Error(err)) + et.Status.MarkBrokerDoesNotExist() + return err + } + et.Status.MarkBrokerExists() + + // Tell tracker to reconcile this EventType whenever the Broker changes. + if err = r.tracker.Track(utils.ObjectRef(b, brokerGVK), et); err != nil { + logging.FromContext(ctx).Error("Unable to track changes to Broker", zap.Error(err)) + return err + } + + if !b.Status.IsReady() { + logging.FromContext(ctx).Error("Broker is not ready", zap.String("broker", b.Name)) + et.Status.MarkBrokerNotReady() + return nil + } + et.Status.MarkBrokerReady() + + return nil +} + +// updateStatus updates the EventType's status. +func (r *Reconciler) updateStatus(ctx context.Context, desired *v1alpha1.EventType) (*v1alpha1.EventType, error) { + eventType, err := r.eventTypeLister.EventTypes(desired.Namespace).Get(desired.Name) + if err != nil { + return nil, err + } + + // If there's nothing to update, just return. + if reflect.DeepEqual(eventType.Status, desired.Status) { + return eventType, nil + } + + becomesReady := desired.Status.IsReady() && !eventType.Status.IsReady() + + // Don't modify the informers copy. + existing := eventType.DeepCopy() + existing.Status = desired.Status + + et, err := r.EventingClientSet.EventingV1alpha1().EventTypes(desired.Namespace).UpdateStatus(existing) + if err == nil && becomesReady { + duration := time.Since(et.ObjectMeta.CreationTimestamp.Time) + logging.FromContext(ctx).Sugar().Infof("EventType %q became ready after %v", eventType.Name, duration) + r.Recorder.Event(eventType, corev1.EventTypeNormal, eventTypeReadinessChanged, fmt.Sprintf("EventType %q became ready", eventType.Name)) + //r.StatsReporter.ReportServiceReady(eventType.Namespace, eventType.Name, duration) // TODO: stats + } + + return et, err +} + +// getBroker returns the Broker for EventType 'et' if it exists, otherwise it returns an error. +func (r *Reconciler) getBroker(ctx context.Context, et *v1alpha1.EventType) (*v1alpha1.Broker, error) { + return r.brokerLister.Brokers(et.Namespace).Get(et.Spec.Broker) +} diff --git a/pkg/reconciler/eventtype/eventtype_test.go b/pkg/reconciler/eventtype/eventtype_test.go new file mode 100644 index 00000000000..1ea039e539f --- /dev/null +++ b/pkg/reconciler/eventtype/eventtype_test.go @@ -0,0 +1,165 @@ +/* +Copyright 2019 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 eventtype + +import ( + "fmt" + "testing" + + "github.com/knative/pkg/tracker" + + "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + "github.com/knative/eventing/pkg/reconciler" + . "github.com/knative/eventing/pkg/reconciler/testing" + "github.com/knative/pkg/controller" + logtesting "github.com/knative/pkg/logging/testing" + . "github.com/knative/pkg/reconciler/testing" + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/scheme" + clientgotesting "k8s.io/client-go/testing" +) + +const ( + testNS = "test-namespace" + eventTypeName = "test-eventtype" + eventTypeType = "test-type" + eventTypeBroker = "test-broker" + eventTypeSource = "/test-source" +) + +var ( + trueVal = true + + testKey = fmt.Sprintf("%s/%s", testNS, eventTypeName) +) + +func init() { + // Add types to scheme + _ = v1alpha1.AddToScheme(scheme.Scheme) +} + +func TestReconcile(t *testing.T) { + table := TableTest{ + { + Name: "bad workqueue key", + // Make sure Reconcile handles bad keys. + Key: "too/many/parts", + }, { + Name: "key not found", + // Make sure Reconcile handles good keys that don't exist. + Key: "foo/not-found", + }, + { + Name: "EventType not found", + Key: testKey, + }, + { + Name: "EventType being deleted", + Key: testKey, + Objects: []runtime.Object{ + NewEventType(eventTypeName, testNS, + WithInitEventTypeConditions, + WithEventTypeDeletionTimestamp), + }, + }, + { + Name: "Broker not found", + Key: testKey, + Objects: []runtime.Object{ + NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + ), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + WithInitEventTypeConditions, + WithEventTypeBrokerDoesNotExist, + ), + }}, + WantErr: true, + WantEvents: []string{ + Eventf(corev1.EventTypeWarning, eventTypeReconcileFailed, "EventType reconcile error: broker.eventing.knative.dev %q not found", eventTypeBroker), + }, + }, + { + Name: "Broker not ready", + Key: testKey, + Objects: []runtime.Object{ + NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + ), + NewBroker(eventTypeBroker, testNS, + WithInitBrokerConditions, + ), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + WithEventTypeBrokerExists, + WithEventTypeBrokerNotReady, + ), + }}, + }, + { + Name: "Successful reconcile, became ready", + Key: testKey, + Objects: []runtime.Object{ + NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + ), + NewBroker(eventTypeBroker, testNS, + WithBrokerReady, + ), + }, + WantStatusUpdates: []clientgotesting.UpdateActionImpl{{ + Object: NewEventType(eventTypeName, testNS, + WithEventTypeType(eventTypeType), + WithEventTypeSource(eventTypeSource), + WithEventTypeBroker(eventTypeBroker), + WithEventTypeBrokerExists, + WithEventTypeBrokerReady, + ), + }}, + WantEvents: []string{ + Eventf(corev1.EventTypeNormal, eventTypeReadinessChanged, "EventType %q became ready", eventTypeName), + }, + }, + } + + defer logtesting.ClearAll() + table.Test(t, MakeFactory(func(listers *Listers, opt reconciler.Options) controller.Reconciler { + return &Reconciler{ + Base: reconciler.NewBase(opt, controllerAgentName), + eventTypeLister: listers.GetEventTypeLister(), + brokerLister: listers.GetBrokerLister(), + tracker: tracker.New(func(string) {}, 0), + } + })) + +} diff --git a/pkg/reconciler/testing/eventtype.go b/pkg/reconciler/testing/eventtype.go new file mode 100644 index 00000000000..828659684f6 --- /dev/null +++ b/pkg/reconciler/testing/eventtype.go @@ -0,0 +1,91 @@ +/* +Copyright 2019 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 testing + +import ( + "context" + "time" + + "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// EventTypeOption enables further configuration of an EventType. +type EventTypeOption func(*v1alpha1.EventType) + +// NewEventType creates a EventType with EventTypeOptions. +func NewEventType(name, namespace string, o ...EventTypeOption) *v1alpha1.EventType { + et := &v1alpha1.EventType{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: namespace, + Name: name, + }, + } + for _, opt := range o { + opt(et) + } + et.SetDefaults(context.Background()) + return et +} + +// WithInitEventTypeConditions initializes the EventType's conditions. +func WithInitEventTypeConditions(et *v1alpha1.EventType) { + et.Status.InitializeConditions() +} + +func WithEventTypeSource(source string) EventTypeOption { + return func(et *v1alpha1.EventType) { + et.Spec.Source = source + } +} + +func WithEventTypeType(t string) EventTypeOption { + return func(et *v1alpha1.EventType) { + et.Spec.Type = t + } +} + +func WithEventTypeBroker(broker string) EventTypeOption { + return func(et *v1alpha1.EventType) { + et.Spec.Broker = broker + } +} + +func WithEventTypeDeletionTimestamp(et *v1alpha1.EventType) { + t := metav1.NewTime(time.Unix(1e9, 0)) + et.ObjectMeta.SetDeletionTimestamp(&t) +} + +// WithEventTypeBrokerNotFound calls .Status.MarkFilterFailed on the EventType. +func WithEventTypeBrokerDoesNotExist(et *v1alpha1.EventType) { + et.Status.MarkBrokerDoesNotExist() +} + +// WithEventTypeBrokerExists calls .Status.MarkBrokerExists on the EventType. +func WithEventTypeBrokerExists(et *v1alpha1.EventType) { + et.Status.MarkBrokerExists() +} + +// WithEventTypeBrokerNotReady calls .Status.MarkBrokerNotReady on the EventType. +func WithEventTypeBrokerNotReady(et *v1alpha1.EventType) { + et.Status.MarkBrokerNotReady() +} + +// WithEventTypeBrokerReady calls .Status.MarkBrokerReady on the EventType. +func WithEventTypeBrokerReady(et *v1alpha1.EventType) { + et.Status.MarkBrokerReady() +} diff --git a/pkg/reconciler/testing/listers.go b/pkg/reconciler/testing/listers.go index bf17f6dd67f..d84b97755ba 100644 --- a/pkg/reconciler/testing/listers.go +++ b/pkg/reconciler/testing/listers.go @@ -108,6 +108,10 @@ func (l *Listers) GetBrokerLister() eventinglisters.BrokerLister { return eventinglisters.NewBrokerLister(l.indexerFor(&eventingv1alpha1.Broker{})) } +func (l *Listers) GetEventTypeLister() eventinglisters.EventTypeLister { + return eventinglisters.NewEventTypeLister(l.indexerFor(&eventingv1alpha1.EventType{})) +} + func (l *Listers) GetChannelLister() eventinglisters.ChannelLister { return eventinglisters.NewChannelLister(l.indexerFor(&eventingv1alpha1.Channel{})) } diff --git a/pkg/reconciler/trigger/trigger.go b/pkg/reconciler/trigger/trigger.go index 473ce7c8e4f..d046ce10a9f 100644 --- a/pkg/reconciler/trigger/trigger.go +++ b/pkg/reconciler/trigger/trigger.go @@ -23,6 +23,8 @@ import ( "reflect" "time" + "github.com/knative/eventing/pkg/utils" + "github.com/knative/eventing/pkg/apis/eventing/v1alpha1" eventinginformers "github.com/knative/eventing/pkg/client/informers/externalversions/eventing/v1alpha1" listers "github.com/knative/eventing/pkg/client/listers/eventing/v1alpha1" @@ -206,7 +208,7 @@ func (r *Reconciler) reconcile(ctx context.Context, t *v1alpha1.Trigger) error { t.Status.PropagateBrokerStatus(&b.Status) // Tell tracker to reconcile this Trigger whenever the Broker changes. - if err = r.tracker.Track(objectRef(b, brokerGVK), t); err != nil { + if err = r.tracker.Track(utils.ObjectRef(b, brokerGVK), t); err != nil { logging.FromContext(ctx).Error("Unable to track changes to Broker", zap.Error(err)) return err } @@ -406,23 +408,3 @@ func (r *Reconciler) getSubscription(ctx context.Context, t *v1alpha1.Trigger) ( return nil, apierrs.NewNotFound(schema.GroupResource{}, "") } - -type accessor interface { - GroupVersionKind() schema.GroupVersionKind - GetNamespace() string - GetName() string -} - -func objectRef(a accessor, gvk schema.GroupVersionKind) corev1.ObjectReference { - // We can't always rely on the TypeMeta being populated. - // See: https://github.com/knative/serving/issues/2372 - // Also: https://github.com/kubernetes/apiextensions-apiserver/issues/29 - // gvk := a.GroupVersionKind() - apiVersion, kind := gvk.ToAPIVersionAndKind() - return corev1.ObjectReference{ - APIVersion: apiVersion, - Kind: kind, - Namespace: a.GetNamespace(), - Name: a.GetName(), - } -} diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go index 33482fc0523..1f9bc2ec1e8 100644 --- a/pkg/utils/utils.go +++ b/pkg/utils/utils.go @@ -22,6 +22,10 @@ import ( "os" "strings" "sync" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" ) const ( @@ -67,3 +71,16 @@ func getClusterDomainName(r io.Reader) string { // For all abnormal cases return default domain name return defaultDomainName } + +func ObjectRef(obj metav1.Object, gvk schema.GroupVersionKind) corev1.ObjectReference { + // We can't always rely on the TypeMeta being populated. + // See: https://github.com/knative/serving/issues/2372 + // Also: https://github.com/kubernetes/apiextensions-apiserver/issues/29 + apiVersion, kind := gvk.ToAPIVersionAndKind() + return corev1.ObjectReference{ + APIVersion: apiVersion, + Kind: kind, + Namespace: obj.GetNamespace(), + Name: obj.GetName(), + } +}