APIServerSource with selector to target namespaces (#6665)

Fixes #6644 

## Proposed Changes

<!-- Please categorize your changes:
- 🎁 Add new feature
- 🐛 Fix bug
- 🧹 Update or clean up current behavior
- 🗑️ Remove feature or internal logic
-->

- 🎁 ApiServerSource can specify a selector to target one or more
namespaces. If the selector is missing, it will default to targeting the
namespace in which the source resides

API changes (additive):
- `spec.namespaceSelector`: A label selector to identify namespaces to
track
- `status.namespaces`: A list of namespaces currently being tracked by
the source


### Pre-review Checklist

<!-- If these boxes are not checked, you will be asked to complete these
requirements or explain why they do not apply to your PR. -->

- [ ] **At least 80% unit test coverage**
- [ ] **E2E tests** for any new behavior
- [ ] **Docs PR** for any user-facing impact
- [ ] **Spec PR** for any new API feature
- [ ] **Conformance test** for any change to the spec

**Release Note**

<!--
📄 If this change has user-visible impact, write a release
note in the block
below. Include the string "action required" if additional action is
required of
users switching to the new release, for example in case of a breaking
change.

Write as if you are speaking to users, not other Knative contributors.
If this
change has no user-visible impact, no release note is needed.
-->

```release-note
📄 ApiServerSource can specify a selector to target one or more namespaces. If the selector is missing, it will default to targeting the namespace in which the source resides
```


**Docs**

Docs to follow once the behaviour is finalized and approved.

config/core/resources/apiserversource.yaml

@@ -137,6 +137,32 @@ spec:
                   uri:
                     description: URI can be an absolute URL(non-empty scheme and non-empty host) pointing to the target or a relative URI. Relative URIs will be resolved using the base URI retrieved from Ref.
                     type: string
+              namespaceSelector:
+                description: NamespaceSelector is a label selector to capture the namespaces that should be watched by the source.
+                type: object
+                properties:
+                  matchExpressions:
+                    description: matchExpressions is a list of label selector requirements. The requirements are ANDed.
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        key:
+                          description: key is the label key that the selector applies to.
+                          type: string
+                        operator:
+                          description: operator represents a key's relationship to a set of values. Valid operators are In, NotIn, Exists and DoesNotExist.
+                          type: string
+                        values:
+                          description: values is an array of string values. If the operator is In or NotIn, the values array must be non-empty. If the operator is Exists or DoesNotExist, the values array must be empty. This array is replaced during a strategic merge patch.
+                          type: array
+                          items:
+                            type: string
+                  matchLabels:
+                    description: matchLabels is a map of {key,value} pairs. A single {key,value} in the matchLabels map is equivalent to an element of matchExpressions, whose key field is "key", the operator is "In", and the values array contains only "value". The requirements are ANDed.
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+
           status:
             type: object
             properties:
@@ -190,6 +216,11 @@ spec:
               sinkUri:
                 description: SinkURI is the current active sink URI that has been configured for the Source.
                 type: string
+              namespaces:
+                description: Namespaces show the namespaces currently watched by the ApiServerSource
+                type: array
+                items:
+                  type: string
     additionalPrinterColumns:
     - name: Sink
       type: string

docs/eventing-api.md

@@ -4410,6 +4410,21 @@ string
 source. Defaults to default if not set.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>namespaceSelector</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.21/#labelselector-v1-meta">
+Kubernetes meta/v1.LabelSelector
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>NamespaceSelector is a label selector to capture the namespaces that
+should be watched by the source.</p>
+</td>
+</tr>
 </table>
 </td>
 </tr>
@@ -5002,6 +5017,21 @@ string
 source. Defaults to default if not set.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>namespaceSelector</code><br/>
+<em>
+<a href="https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.21/#labelselector-v1-meta">
+Kubernetes meta/v1.LabelSelector
+</a>
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>NamespaceSelector is a label selector to capture the namespaces that
+should be watched by the source.</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sources.knative.dev/v1.ApiServerSourceStatus">ApiServerSourceStatus
@@ -5042,6 +5072,17 @@ state.
 Source.</p>
 </td>
 </tr>
+<tr>
+<td>
+<code>namespaces</code><br/>
+<em>
+[]string
+</em>
+</td>
+<td>
+<p>Namespaces show the namespaces currently watched by the ApiServerSource</p>
+</td>
+</tr>
 </tbody>
 </table>
 <h3 id="sources.knative.dev/v1.ContainerSourceSpec">ContainerSourceSpec

pkg/adapter/apiserver/adapter.go

@@ -95,21 +95,25 @@ func (a *apiServerAdapter) start(ctx context.Context, stopCh <-chan struct{}) er
 		exists := false
 		for _, apires := range resources.APIResources {
 			if apires.Name == configRes.GVR.Resource {
-
-				var res dynamic.ResourceInterface
-				if apires.Namespaced {
-					res = a.k8s.Resource(configRes.GVR).Namespace(a.config.Namespace)
+				var resources []dynamic.ResourceInterface
+				if apires.Namespaced && !a.config.AllNamespaces {
+					for _, ns := range a.config.Namespaces {
+						resources = append(resources, a.k8s.Resource(configRes.GVR).Namespace(ns))
+					}
 				} else {
-					res = a.k8s.Resource(configRes.GVR)
+					resources = append(resources, a.k8s.Resource(configRes.GVR))
 				}
 
-				lw := &cache.ListWatch{
-					ListFunc:  asUnstructuredLister(ctx, res.List, configRes.LabelSelector),
-					WatchFunc: asUnstructuredWatcher(ctx, res.Watch, configRes.LabelSelector),
+				for _, res := range resources {
+					lw := &cache.ListWatch{
+						ListFunc:  asUnstructuredLister(ctx, res.List, configRes.LabelSelector),
+						WatchFunc: asUnstructuredWatcher(ctx, res.Watch, configRes.LabelSelector),
+					}
+
+					reflector := cache.NewReflector(lw, &unstructured.Unstructured{}, delegate, resyncPeriod)
+					go reflector.Run(stop)
 				}
 
-				reflector := cache.NewReflector(lw, &unstructured.Unstructured{}, delegate, resyncPeriod)
-				go reflector.Run(stop)
 				exists = true
 				break
 			}

pkg/adapter/apiserver/adapter_test.go

@@ -45,7 +45,7 @@ func TestAdapter_StartRef(t *testing.T) {
 	ce := adaptertest.NewTestClient()
 
 	config := Config{
-		Namespace: "default",
+		Namespaces: []string{"default"},
 		Resources: []ResourceWatch{{
 			GVR: schema.GroupVersionResource{
 				Version:  "v1",
@@ -93,7 +93,7 @@ func TestAdapter_StartResource(t *testing.T) {
 	ce := adaptertest.NewTestClient()
 
 	config := Config{
-		Namespace: "default",
+		Namespaces: []string{"default"},
 		Resources: []ResourceWatch{{
 			GVR: schema.GroupVersionResource{
 				Version:  "v1",
@@ -140,7 +140,7 @@ func TestAdapter_StartNonNamespacedResource(t *testing.T) {
 	ce := adaptertest.NewTestClient()
 
 	config := Config{
-		Namespace: "default",
+		Namespaces: []string{"default"},
 		Resources: []ResourceWatch{{
 			GVR: schema.GroupVersionResource{
 				Version:  "v1",

pkg/adapter/apiserver/config.go

@@ -31,9 +31,13 @@ type ResourceWatch struct {
 }
 
 type Config struct {
-	// Namespace specifies the namespace that Resources[] exist.
+	// Namespaces specifies the namespaces where Resources[] exist.
 	// +required
-	Namespace string `json:"namespace"`
+	Namespaces []string `json:"namespaces"`
+
+	// AllNamespaces indicates whether this source is watching all
+	// existing namespaces
+	AllNamespaces bool `json:"allNamespaces"`
 
 	// Resource is the resource this source will track and send related
 	// lifecycle events from the Kubernetes ApiServer.

pkg/apis/sources/v1/apiserver_types.go

@@ -80,6 +80,11 @@ type ApiServerSourceSpec struct {
 	// source. Defaults to default if not set.
 	// +optional
 	ServiceAccountName string `json:"serviceAccountName,omitempty"`
+
+	// NamespaceSelector is a label selector to capture the namespaces that
+	// should be watched by the source.
+	// +optional
+	NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"`
 }
 
 // ApiServerSourceStatus defines the observed state of ApiServerSource
@@ -92,6 +97,9 @@ type ApiServerSourceStatus struct {
 	// * SinkURI - the current active sink URI that has been configured for the
 	//   Source.
 	duckv1.SourceStatus `json:",inline"`
+
+	// Namespaces show the namespaces currently watched by the ApiServerSource
+	Namespaces []string `json:"namespaces"`
 }
 
 // APIVersionKind is an APIVersion and Kind tuple.