diff --git a/internal/controller/cache/cache.go b/internal/controller/cache/cache.go
new file mode 100644
index 00000000..5cb504f6
--- /dev/null
+++ b/internal/controller/cache/cache.go
@@ -0,0 +1,38 @@
+package cache
+
+import (
+	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kyma-project/nats-manager/internal/label"
+)
+
+// New returns a cache with the cache-options applied, generade form the rest-config.
+func New(config *rest.Config, options cache.Options) (cache.Cache, error) {
+	return cache.New(config, applySelectors(options))
+}
+
+func applySelectors(options cache.Options) cache.Options {
+	// The only objects we allow are the ones with the 'created-by: nats-manager' label applied.
+	createdByNATSManager := fromLabelSelector(label.SelectorCreatedByNATS())
+
+	// Apply the label selector to all relevant objects.
+	options.ByObject = map[client.Object]cache.ByObject{
+		&appsv1.Deployment{}:                     createdByNATSManager,
+		&autoscalingv1.HorizontalPodAutoscaler{}: createdByNATSManager,
+		&corev1.ServiceAccount{}:                 createdByNATSManager,
+		&rbacv1.ClusterRole{}:                    createdByNATSManager,
+		&rbacv1.ClusterRoleBinding{}:             createdByNATSManager,
+	}
+	return options
+}
+
+func fromLabelSelector(selector labels.Selector) cache.ByObject {
+	return cache.ByObject{Label: selector}
+}
diff --git a/internal/controller/cache/cache_test.go b/internal/controller/cache/cache_test.go
new file mode 100644
index 00000000..9c14991d
--- /dev/null
+++ b/internal/controller/cache/cache_test.go
@@ -0,0 +1,158 @@
+package cache
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/kyma-project/nats-manager/internal/label"
+	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func Test_applySelectors(t *testing.T) {
+	// given
+	syncPeriod := 30 * time.Second
+	selector := cache.ByObject{
+		Label: labels.SelectorFromSet(
+			map[string]string{
+				label.KeyCreatedBy: label.ValueNATSManager,
+			},
+		),
+	}
+
+	type args struct {
+		options cache.Options
+	}
+	testCases := []struct {
+		name string
+		args args
+		want cache.Options
+	}{
+		{
+			name: "should apply the correct selectors",
+			args: args{
+				options: cache.Options{},
+			},
+			want: cache.Options{
+				ByObject: map[client.Object]cache.ByObject{
+					&appsv1.Deployment{}:                     selector,
+					&corev1.ServiceAccount{}:                 selector,
+					&rbacv1.ClusterRole{}:                    selector,
+					&rbacv1.ClusterRoleBinding{}:             selector,
+					&autoscalingv1.HorizontalPodAutoscaler{}: selector,
+				},
+			},
+		},
+		{
+			name: "should not remove existing options",
+			args: args{
+				options: cache.Options{
+					SyncPeriod: &syncPeriod,
+				},
+			},
+			want: cache.Options{
+				SyncPeriod: &syncPeriod,
+				ByObject: map[client.Object]cache.ByObject{
+					&appsv1.Deployment{}:                     selector,
+					&corev1.ServiceAccount{}:                 selector,
+					&rbacv1.ClusterRole{}:                    selector,
+					&rbacv1.ClusterRoleBinding{}:             selector,
+					&autoscalingv1.HorizontalPodAutoscaler{}: selector,
+				},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			// when
+			got := applySelectors(tc.args.options)
+
+			// then
+			require.True(t, deepEqualOptions(tc.want, got))
+		})
+	}
+}
+
+func deepEqualOptions(a, b cache.Options) bool {
+	// we only care about the ByObject comparison
+	o := deepEqualByObject(a.ByObject, b.ByObject)
+	s := *a.SyncPeriod == *b.SyncPeriod
+	return o && s
+}
+
+func deepEqualByObject(a, b map[client.Object]cache.ByObject) bool {
+	if len(a) != len(b) {
+		return false
+	}
+
+	aTypeMap := make(map[string]cache.ByObject, len(a))
+	bTypeMap := make(map[string]cache.ByObject, len(a))
+	computeTypeMap(a, aTypeMap)
+	computeTypeMap(b, bTypeMap)
+	return reflect.DeepEqual(aTypeMap, bTypeMap)
+}
+
+func computeTypeMap(byObjectMap map[client.Object]cache.ByObject, typeMap map[string]cache.ByObject) {
+	keyOf := func(i interface{}) string { return fmt.Sprintf(">>> %T", i) }
+	for k, v := range byObjectMap {
+		if obj, ok := k.(*appsv1.Deployment); ok {
+			key := keyOf(obj)
+			typeMap[key] = v
+		}
+		if obj, ok := k.(*corev1.ServiceAccount); ok {
+			key := keyOf(obj)
+			typeMap[key] = v
+		}
+		if obj, ok := k.(*rbacv1.ClusterRole); ok {
+			key := keyOf(obj)
+			typeMap[key] = v
+		}
+		if obj, ok := k.(*rbacv1.ClusterRoleBinding); ok {
+			key := keyOf(obj)
+			typeMap[key] = v
+		}
+		if obj, ok := k.(*autoscalingv1.HorizontalPodAutoscaler); ok {
+			key := keyOf(obj)
+			typeMap[key] = v
+		}
+	}
+}
+
+func Test_fromLabelSelector(t *testing.T) {
+	// given
+	type args struct {
+		label labels.Selector
+	}
+	tests := []struct {
+		name string
+		args args
+		want cache.ByObject
+	}{
+		{
+			name: "should return the correct selector",
+			args: args{
+				label: labels.SelectorFromSet(map[string]string{"key": "value"}),
+			},
+			want: cache.ByObject{
+				Label: labels.SelectorFromSet(map[string]string{"key": "value"}),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// when
+			got := fromLabelSelector(tt.args.label)
+
+			// then
+			require.Equal(t, tt.want, got)
+		})
+	}
+}