Allow for http and broker sinks (knative#574)

* Introduce Broker as a kind of sink, genericize sinks in general

* loltest

* Fix tests for mergeout

* the test file I forgot

* docstrings

* Move fake dynamic client into seprate file

* Pass whole client, not raw client, to resolve sinks

* one more test

pkg/dynamic/client.go

@@ -42,6 +42,9 @@ type KnDynamicClient interface {
 
 	// ListSourceCRDs returns list of eventing sources CRDs
 	ListSourcesTypes() (*unstructured.UnstructuredList, error)
+
+	// RawClient returns the raw dynamic client interface
+	RawClient() dynamic.Interface
 }
 
 // knDynamicClient is a combination of client-go Dynamic client interface and namespace
@@ -87,3 +90,7 @@ func (c *knDynamicClient) ListSourcesTypes() (*unstructured.UnstructuredList, er
 	options.LabelSelector = sourcesLabels.String()
 	return c.ListCRDs(options)
 }
+
+func (c knDynamicClient) RawClient() dynamic.Interface {
+	return c.client
+}

pkg/dynamic/client_test.go

@@ -22,7 +22,10 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/dynamic/fake"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8s_fake "k8s.io/client-go/dynamic/fake"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 )
 
 const testNamespace = "testns"
@@ -43,18 +46,14 @@ func newUnstructured(name string) *unstructured.Unstructured {
 	}
 }
 
-func createFakeKnDynamicClient(objects ...runtime.Object) KnDynamicClient {
-	client := fake.NewSimpleDynamicClient(runtime.NewScheme(), objects...)
-	return NewKnDynamicClient(client, testNamespace)
-}
-
 func TestNamespace(t *testing.T) {
-	client := createFakeKnDynamicClient(newUnstructured("foo"))
+	client := createFakeKnDynamicClient(testNamespace, newUnstructured("foo"))
 	assert.Equal(t, client.Namespace(), testNamespace)
 }
 
 func TestListCRDs(t *testing.T) {
 	client := createFakeKnDynamicClient(
+		testNamespace,
 		newUnstructured("foo"),
 		newUnstructured("bar"),
 	)
@@ -84,6 +83,7 @@ func TestListCRDs(t *testing.T) {
 
 func TestListSourceTypes(t *testing.T) {
 	client := createFakeKnDynamicClient(
+		testNamespace,
 		newUnstructured("foo"),
 		newUnstructured("bar"),
 	)
@@ -99,3 +99,13 @@ func TestListSourceTypes(t *testing.T) {
 		assert.Equal(t, uList.Items[1].GetName(), "bar")
 	})
 }
+
+// createFakeKnDynamicClient gives you a dynamic client for testing contianing the given objects.
+// See also the one in the fake package. Duplicated here to avoid a dependency loop.
+func createFakeKnDynamicClient(testNamespace string, objects ...runtime.Object) KnDynamicClient {
+	scheme := runtime.NewScheme()
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "serving.knative.dev", Version: "v1alpha1", Kind: "Service"}, &serving_v1alpha1.Service{})
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "eventing.knative.dev", Version: "v1alpha1", Kind: "Broker"}, &eventing_v1alpha1.Broker{})
+	client := k8s_fake.NewSimpleDynamicClient(scheme, objects...)
+	return NewKnDynamicClient(client, testNamespace)
+}

pkg/dynamic/fake/fake.go

@@ -0,0 +1,34 @@
+// 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 fake
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8s_fake "k8s.io/client-go/dynamic/fake"
+
+	"knative.dev/client/pkg/dynamic"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
+)
+
+// CreateFakeKnDynamicClient gives you a dynamic client for testing contianing the given objects.
+func CreateFakeKnDynamicClient(testNamespace string, objects ...runtime.Object) dynamic.KnDynamicClient {
+	scheme := runtime.NewScheme()
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "serving.knative.dev", Version: "v1alpha1", Kind: "Service"}, &serving_v1alpha1.Service{})
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "eventing.knative.dev", Version: "v1alpha1", Kind: "Broker"}, &eventing_v1alpha1.Broker{})
+	client := k8s_fake.NewSimpleDynamicClient(scheme, objects...)
+	return dynamic.NewKnDynamicClient(client, testNamespace)
+}

pkg/kn/commands/flags/sink.go

@@ -20,9 +20,11 @@ import (
 
 	"github.com/spf13/cobra"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	kn_dynamic "knative.dev/client/pkg/dynamic"
+	"knative.dev/pkg/apis"
 	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
-
-	"knative.dev/client/pkg/serving/v1alpha1"
 )
 
 type SinkFlags struct {
@@ -33,26 +35,73 @@ func (i *SinkFlags) Add(cmd *cobra.Command) {
 	cmd.Flags().StringVarP(&i.sink, "sink", "s", "", "Addressable sink for events")
 }
 
-func (i *SinkFlags) ResolveSink(client v1alpha1.KnServingClient) (*duckv1beta1.Destination, error) {
+// SinkPrefixes maps prefixes used for sinks to their GroupVersionResources.
+var SinkPrefixes = map[string]schema.GroupVersionResource{
+	"broker": {
+		Resource: "brokers",
+		Group:    "eventing.knative.dev",
+		Version:  "v1alpha1",
+	},
+	"service": {
+		Resource: "services",
+		Group:    "serving.knative.dev",
+		Version:  "v1alpha1",
+	},
+	// Shorthand alias for service
+	"svc": {
+		Resource: "services",
+		Group:    "serving.knative.dev",
+		Version:  "v1alpha1",
+	},
+}
+
+// ResolveSink returns the Destination referred to by the flags in the acceptor.
+// It validates that any object the user is referring to exists.
+func (i *SinkFlags) ResolveSink(knclient kn_dynamic.KnDynamicClient, namespace string) (*duckv1beta1.Destination, error) {
+	client := knclient.RawClient()
 	if i.sink == "" {
 		return nil, nil
 	}
 
-	if strings.HasPrefix(i.sink, "svc:") {
-		serviceName := i.sink[4:]
-		service, err := client.GetService(serviceName)
+	prefix, name := parseSink(i.sink)
+	if prefix == "" {
+		// URI target
+		uri, err := apis.ParseURL(name)
 		if err != nil {
 			return nil, err
 		}
-		return &duckv1beta1.Destination{
-			Ref: &v1.ObjectReference{
-				Kind:       service.Kind,
-				APIVersion: service.APIVersion,
-				Name:       service.Name,
-				Namespace:  service.Namespace,
-			},
-		}, nil
+		return &duckv1beta1.Destination{URI: uri}, nil
+	}
+	typ, ok := SinkPrefixes[prefix]
+	if !ok {
+		return nil, fmt.Errorf("Not supported sink type: %s", i.sink)
+	}
+	obj, err := client.Resource(typ).Namespace(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
 	}
 
-	return nil, fmt.Errorf("Not supported sink type: %s", i.sink)
+	return &duckv1beta1.Destination{
+		Ref: &v1.ObjectReference{
+			Kind:       obj.GetKind(),
+			APIVersion: obj.GetAPIVersion(),
+			Name:       obj.GetName(),
+			Namespace:  namespace,
+		},
+	}, nil
+
+}
+
+// parseSink takes the string given by the user into the prefix and the name of
+// the object. If the user put a URI instead, the prefix is empty and the name
+// is the whole URI.
+func parseSink(sink string) (string, string) {
+	parts := strings.SplitN(sink, ":", 2)
+	if len(parts) == 1 {
+		return "svc", parts[0]
+	} else if parts[0] == "http" || parts[0] == "https" {
+		return "", sink
+	} else {
+		return parts[0], parts[1]
+	}
 }

pkg/kn/commands/flags/sink_test.go

@@ -0,0 +1,81 @@
+// 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 flags
+
+import (
+	"testing"
+
+	"gotest.tools/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	"knative.dev/pkg/apis"
+	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
+)
+
+type resolveCase struct {
+	sink        string
+	destination *duckv1beta1.Destination
+	errContents string
+}
+
+func TestResolve(t *testing.T) {
+	targetExampleCom, err := apis.ParseURL("http://target.example.com")
+	mysvc := &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	}
+	defaultBroker := &eventing_v1alpha1.Broker{
+		TypeMeta:   metav1.TypeMeta{Kind: "Broker", APIVersion: "eventing.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "default", Namespace: "default"},
+	}
+
+	assert.NilError(t, err)
+	cases := []resolveCase{
+		{"svc:mysvc", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Service",
+				APIVersion: "serving.knative.dev/v1alpha1",
+				Name:       "mysvc",
+				Namespace:  "default"}}, ""},
+		{"service:mysvc", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Service",
+				APIVersion: "serving.knative.dev/v1alpha1",
+				Name:       "mysvc",
+				Namespace:  "default"}}, ""},
+		{"svc:absent", nil, "\"absent\" not found"},
+		{"broker:default", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Broker",
+				APIVersion: "eventing.knative.dev/v1alpha1",
+				Name:       "default",
+				Namespace:  "default",
+			}}, ""},
+		{"http://target.example.com", &duckv1beta1.Destination{
+			URI: targetExampleCom,
+		}, ""},
+	}
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", mysvc, defaultBroker)
+	for _, c := range cases {
+		i := &SinkFlags{c.sink}
+		result, err := i.ResolveSink(dynamicClient, "default")
+		if c.destination != nil {
+			assert.DeepEqual(t, result, c.destination)
+			assert.NilError(t, err)
+		} else {
+			assert.ErrorContains(t, err, c.errContents)
+		}
+	}
+}

pkg/kn/commands/source/apiserver/apiserver_test.go

@@ -19,12 +19,12 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/clientcmd"
+	kn_dynamic "knative.dev/client/pkg/dynamic"
 	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
 	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
 
 	knsource_v1alpha1 "knative.dev/client/pkg/eventing/sources/v1alpha1"
 	"knative.dev/client/pkg/kn/commands"
-	knserving_v1alpha1 "knative.dev/client/pkg/serving/v1alpha1"
 )
 
 const testNamespace = "default"
@@ -55,14 +55,14 @@ current-context: x
 	}
 }
 
-func executeAPIServerSourceCommand(apiServerSourceClient knsource_v1alpha1.KnAPIServerSourcesClient, servingClient knserving_v1alpha1.KnServingClient, args ...string) (string, error) {
+func executeAPIServerSourceCommand(apiServerSourceClient knsource_v1alpha1.KnAPIServerSourcesClient, dynamicClient kn_dynamic.KnDynamicClient, args ...string) (string, error) {
 	knParams := &commands.KnParams{}
 	knParams.ClientConfig = blankConfig
 
 	output := new(bytes.Buffer)
 	knParams.Output = output
-	knParams.NewServingClient = func(namespace string) (knserving_v1alpha1.KnServingClient, error) {
-		return servingClient, nil
+	knParams.NewDynamicClient = func(namespace string) (kn_dynamic.KnDynamicClient, error) {
+		return dynamicClient, nil
 	}
 
 	cmd := NewAPIServerCommand(knParams)
@@ -92,8 +92,10 @@ func createAPIServerSource(name, resourceKind, resourceVersion, serviceAccount,
 
 	sink := &duckv1beta1.Destination{
 		Ref: &corev1.ObjectReference{
-			Kind: "Service",
-			Name: service,
+			Kind:       "Service",
+			Name:       service,
+			APIVersion: "serving.knative.dev/v1alpha1",
+			Namespace:  "default",
 		}}
 
 	return knsource_v1alpha1.NewAPIServerSourceBuilder(name).

pkg/kn/commands/source/apiserver/create.go

@@ -51,13 +51,11 @@ func NewAPIServerCreateCommand(p *commands.KnParams) *cobra.Command {
 
 			namespace := apiSourceClient.Namespace()
 
-			// create Serving client for resolving service sink
-			servingClient, err := p.NewServingClient(namespace)
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
-
-			objectRef, err := sinkFlags.ResolveSink(servingClient)
+			objectRef, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 			if err != nil {
 				return fmt.Errorf(
 					"cannot create ApiServerSource '%s' in namespace '%s' "+