Repository: Kong/kubernetes-ingress-controller
Pull request #5965 (2024-05-04 01:11:24)

feat(plugins) allow cross-namespace plugin references
**What this PR does / why we need it**:

Add a new colon-delimited value format for the `konghq.com/plugins` CSV annotation. The string before the colon indicates a remote KongPlugin namespace. This namespace must contain a ReferenceGrant that allows this reference.

While the existing `konghq.com/plugins: foo,bar` format loads plugins `foo` and `bar` from the same namespace as the annotated resource, the new `konghq.com/plugins: foo,alpha:bar` format loads `foo` from the local namespace and `bar` from the `alpha` namespace.

[example_cross_plugin.yaml.txt](https://github.com/Kong/kubernetes-ingress-controller/files/15242556/example_cross_plugin.yaml.txt) provides a more complete example. Service `one` in namespace `pwuh` loads the `test` KongPlugin from the `pwuh` namespace as usual. Service `two` in namespace `fwuh` loads the same `test` KongPlugin from `pwuh`. It can do so because a ReferenceGrant in `pwuh` grants KongPlugin access to `fwuh` Services.

**Which issue this PR fixes**:

Fix https://github.com/Kong/kubernetes-ingress-controller/issues/5355. Although the original request concerned KongClusterPlugins, using KongPlugins allows for a simpler security configuration model.

Review https://github.com/Kong/kubernetes-ingress-controller/issues/5355#issuecomment-2075580376 for additional background on why multi-entity plugins require grants to adhere to the Kubernetes namespace security model.

**Special notes for your reviewer**:

There was already a fair amount of code duplication in the plugin relationship builder. Ideally we'd do away with it, but I've left it around for the time being. The `isRemotePluginReferenceAllowed()` helper hopefully negates the worst of it by extracting most of the logic we need to duplicate into a separate function, so the duplicated code is just the function invocation.

**PR Readiness Checklist**:

Complete these before marking the PR as `ready to review`:

- [x] the `CHANGELOG.md` release notes have been updated to reflect any significant (and particularly user-facing) changes introduced by this PR


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) Report
Attention: Patch coverage is `27.65957%` with `68 lines` in your changes are missing coverage. Please review.
> Project coverage is 71.1%. Comparing base [(`04dead8`)](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/commit/04dead87f31c71629a53245ec525d8800d92742b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) to head [(`0cad672`)](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong).
> Report is 20 commits behind head on main.

| [Files](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) | Patch % | Lines |
|---|---|---|
| [internal/dataplane/kongstate/kongstate.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&filepath=internal%2Fdataplane%2Fkongstate%2Fkongstate.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZGF0YXBsYW5lL2tvbmdzdGF0ZS9rb25nc3RhdGUuZ28=) | 16.9% | [45 Missing and 4 partials :warning: ](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) |
| [...al/gatewayapi/references.go/wrappers\_refchecker.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&filepath=internal%2Fgatewayapi%2Freferences.go%2Fwrappers_refchecker.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvZ2F0ZXdheWFwaS9yZWZlcmVuY2VzLmdvL3dyYXBwZXJzX3JlZmNoZWNrZXIuZ28=) | 22.2% | [14 Missing :warning: ](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) |
| [internal/annotations/annotations.go](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&filepath=internal%2Fannotations%2Fannotations.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong#diff-aW50ZXJuYWwvYW5ub3RhdGlvbnMvYW5ub3RhdGlvbnMuZ28=) | 68.7% | [3 Missing and 2 partials :warning: ](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong) |

<details><summary>Additional details and impacted files</summary>


```diff
@@           Coverage Diff           @@
##            main   #5965     +/-   ##
=======================================
+ Coverage   68.9%   71.1%   +2.2%     
=======================================
  Files        179     179             
  Lines      18308   18387     +79     
=======================================
+ Hits       12627   13090    +463     
+ Misses      4693    4270    -423     
- Partials     988    1027     +39     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Kong/kubernetes-ingress-controller/pull/5965?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Kong).

rainest: Some sort of nonsense apparently means the test... passes? but then leaves some of its resources around, furthermore preventing the test suite from finishing :(

Patch:
diff --git a/.golangci.yaml b/.golangci.yaml
index 3238a69f4..82d98e060 100644
--- a/.golangci.yaml
+++ b/.golangci.yaml
@@ -33,7 +33,6 @@ linters:
   - importas
   - ineffassign
   - loggercheck
-  - megacheck
   - misspell
   - nakedret
   - nilerr
@@ -160,6 +159,9 @@ issues:
    - (Expect directory permissions to be 0750 or less|Expect file permissions to be 0600 or less)
    # EXC0010 gosec: False positive is triggered by 'src, err := ioutil.ReadFile(filename)'
    - Potential file inclusion via variable
+   # G601 Implicit memory aliasing in for loop
+   # Irrelevant as of Go 1.22
+   - (G601)
 
   exclude-rules:
   # Ignore insecure TLS in tests and hardcoded credentials
diff --git a/.tools_versions.yaml b/.tools_versions.yaml
index 9dd777b5d..93d67392d 100644
--- a/.tools_versions.yaml
+++ b/.tools_versions.yaml
@@ -7,7 +7,7 @@ kustomize: "5.3.0"
 # renovate: datasource=github-releases depName=golangci/golangci-lint
 golangci-lint: "1.58.1"
 # renovate: datasource=github-releases depName=GoogleContainerTools/skaffold
-skaffold: "2.11.1"
+skaffold: "2.12.0"
 # renovate: datasource=github-releases depName=kubernetes-sigs/controller-runtime
 setup-envtest: "0.18.2"
 # renovate: datasource=github-releases depName=elastic/crd-ref-docs
diff --git a/CHANGELOG.md b/CHANGELOG.md
index cadb722e2..3e651220d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -135,6 +135,7 @@ Adding a new version? You'll need three changes:
 - Added the CRD `KongCustomEntity` to support custom Kong entities that are not
   defined in KIC yet.
   [#5982](https://github.com/Kong/kubernetes-ingress-controller/pull/5982)
+  [#6006](https://github.com/Kong/kubernetes-ingress-controller/pull/6006)
 - Added `FallbackConfiguration` feature gate to enable the controller to generate a fallback configuration
   for Kong when it fails to apply the configuration. The feature gate is disabled by default.
   [#5993](https://github.com/Kong/kubernetes-ingress-controller/pull/5993)
@@ -142,7 +143,16 @@ Adding a new version? You'll need three changes:
   - add a flag `--enable-controller-gwapi-grpcroute` to control whether enable or disable GRPCRoute controller.
   - add support for `GRPCRoute` v1, which requires users to upgrade the Gateway API's CRD to v1.1.
   [#5918](https://github.com/Kong/kubernetes-ingress-controller/pull/5918)
-
+- The `konghq.com/plugins` annotation supports a new `<namespace>:<name>`
+  format. This format requests a KongPlugin from a remote namespace. Binding
+  plugins across namespaces requires a ReferenceGrant from the requesting
+  resource to KongPlugins in the target namespace. This approach is useful for
+  some plugins bound to different types of entities, such as a set of
+  rate-limiting plugins applied to a service and various consumers. The
+  cross-namespace grant allows the service manager to define different limits
+  for consumers managed by other users without requiring those users to create
+  consumers in the Service's namespace.
+  [#5965](https://github.com/Kong/kubernetes-ingress-controller/pull/5965)
 
 ### Fixed
 
diff --git a/Makefile b/Makefile
index c17e6a62c..b80a59117 100644
--- a/Makefile
+++ b/Makefile
@@ -62,7 +62,7 @@ mise:
 	@mise -V >/dev/null || (echo "mise - https://github.com/jdx/mise - not found. Please install it." && exit 1)
 
 .PHONY: tools
-tools: controller-gen kustomize client-gen golangci-lint.download gotestsum crd-ref-docs skaffold looppointer.download staticcheck.download
+tools: controller-gen kustomize client-gen golangci-lint.download gotestsum crd-ref-docs skaffold staticcheck.download
 
 export MISE_DATA_DIR = $(PROJECT_DIR)/bin/
 
@@ -161,11 +161,6 @@ go-junit-report: ## Download go-junit-report locally if necessary.
 	@$(MAKE) mise-plugin-install DEP=go-junit-report URL=https://github.com/pmalek/asdf-go-junit-report.git
 	@$(MISE) install go-junit-report@$(GOJUNIT_REPORT_VERSION)
 
-LOOPPOINTER= $(PROJECT_DIR)/bin/looppointer
-.PHONY: looppointer.download
-looppointer.download: ## Download looppointer locally if necessary.
-	@$(MAKE) _download_tool TOOL=looppointer
-
 # ------------------------------------------------------------------------------
 # Build
 # ------------------------------------------------------------------------------
@@ -216,7 +211,7 @@ fmt:
 	go fmt ./...
 
 .PHONY: lint
-lint: verify.tidy golangci-lint staticcheck looppointer
+lint: verify.tidy golangci-lint staticcheck 
 
 .PHONY: golangci-lint
 golangci-lint: golangci-lint.download
@@ -229,9 +224,6 @@ staticcheck: staticcheck.download
 		grep -F -e internal/konnect/controlplanes -v | \
 		xargs $(STATICCHECK) -tags envtest,e2e_tests,integration_tests,istio_tests,conformance_tests -f stylish
 
-looppointer: looppointer.download
-	$(LOOPPOINTER) -v ./internal/... ./test/...
-
 .PHONY: verify.tidy
 verify.tidy:
 	./scripts/verify-tidy.sh
@@ -289,7 +281,7 @@ manifests.single: kustomize ## Compose single-file deployment manifests from bui
 # ------------------------------------------------------------------------------
 
 .PHONY: generate
-generate: generate.controllers generate.clientsets generate.gateway-api-consts generate.docs generate.store fmt
+generate: generate.controllers generate.clientsets generate.gateway-api-consts generate.docs generate.go fmt
 
 .PHONY: generate.controllers
 generate.controllers: controller-gen
@@ -328,9 +320,10 @@ generate.apidocs: crd-ref-docs
 generate.cli-arguments-docs:
 	go run ./scripts/cli-arguments-docs-gen/main.go > ./docs/cli-arguments.md
 
-.PHONY: generate.store
-generate.store:
-	go generate ./internal/store
+.PHONY: generate.go
+generate.go:
+	go generate ./...
+
 
 # ------------------------------------------------------------------------------
 # Build - Container Images
diff --git a/config/webhook/manifests.yaml b/config/webhook/manifests.yaml
index 3281b2066..71a95d9fc 100644
--- a/config/webhook/manifests.yaml
+++ b/config/webhook/manifests.yaml
@@ -132,6 +132,27 @@ webhooks:
     resources:
     - kongconsumers
   sideEffects: None
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    service:
+      name: webhook-service
+      namespace: system
+      path: /
+  failurePolicy: Fail
+  matchPolicy: Equivalent
+  name: kongcustomentities.validation.ingress-controller.konghq.com
+  rules:
+  - apiGroups:
+    - configuration.konghq.com
+    apiVersions:
+    - v1alpha1
+    operations:
+    - CREATE
+    - UPDATE
+    resources:
+    - kongcustomentities
+  sideEffects: None
 - admissionReviewVersions:
   - v1
   clientConfig:
diff --git a/examples/cross-namespace-plugin-grant.yaml b/examples/cross-namespace-plugin-grant.yaml
new file mode 100644
index 000000000..956c532e7
--- /dev/null
+++ b/examples/cross-namespace-plugin-grant.yaml
@@ -0,0 +1,166 @@
+# Service A, Consumer 1 in Namespace Foo, Service B in Namespace Bar (KongPlugin)
+# With the original behavior, KIC provided no means for the Service in Bar to use the KongPlugin in Foo.
+# With the modified behavior, the Service in Bar can use namespace:plugin_name to request another
+# namespace's plugin. Functionally, the Service in Bar behaves the same as a Service in Foo for the
+# purpose of plugin binding, i.e. the Service in Bar can create a multi-entity plugin with the KongConsumer
+# in Foo same as the Service in Foo can, if it has permission. Permissions are ReferenceGrants to
+# the KongPlugin:
+# - In namespace Foo, grant from Bar:Services to KongPlugins.
+# These grants may optionally use the standard ReferenceGrant specific target syntax to limit which
+# plugins external resources can bind to. Because this KongPlugin inherently has access to only
+# resources in its namespace, permissions to it are a clearer indication of access to those resources
+# than the ambiguous KongClusterPlugin situation, and we do not need bidirectional permissions.
+
+# Service A in Foo, Service B in Bar, Consumer in Baz
+# This doesn't differ much for KongPlugins either, in that you simply need an additional grant from
+# both other resource namespaces. You do need to choose whether the plugin resides in the KongConsumer
+# namespace or the Service namespaces. Functionally this doesn't change much other than the permission
+# configuration, though it may make sense to choose one or the other based other resources in the from
+# namespace. Note that you can still vary configuration if you choose the consumer namespace, you just
+# need multiple KongPlugins to do so.
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: one
+  namespace: pwuh
+  labels:
+    app: httpbin
+  annotations:
+    konghq.com/plugins: test
+spec:
+  ports:
+  - name: http
+    port: 80
+    targetPort: 80
+  selector:
+    app: httpbin
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: two
+  namespace: fwuh
+  labels:
+    app: httpbin
+  annotations:
+    konghq.com/plugins: pwuh:test
+spec:
+  ports:
+  - name: http
+    port: 80
+    targetPort: 80
+  selector:
+    app: httpbin
+---
+apiVersion: configuration.konghq.com/v1
+kind: KongConsumer
+metadata:
+  name: admin
+  namespace: pwuh
+  annotations:
+    kubernetes.io/ingress.class: kong
+    konghq.com/plugins: test
+username: admin
+---
+apiVersion: configuration.konghq.com/v1
+kind: KongPlugin
+metadata:
+  name: test
+  namespace: pwuh
+plugin: udp-log
+config:
+  host: example.com
+  port: 80
+---
+kind: ReferenceGrant
+metadata:
+  name: example
+  namespace: pwuh
+spec:
+  from:
+  - group: ""
+    kind: Service
+    namespace: fwuh
+  to:
+  - group: "configuration.konghq.com"
+    kind: KongPlugin
+    name: test # Optional, to limit external resources to a specific KongPlugin only.
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: httpbin
+  namespace: pwuh
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: httpbin
+  template:
+    metadata:
+      labels:
+        app: httpbin
+    spec:
+      containers:
+      - image: docker.io/kennethreitz/httpbin
+        name: httpbin
+        ports:
+        - containerPort: 80
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: httpbin
+  namespace: pwuh
+spec:
+  ingressClassName: kong
+  rules:
+  - http:
+      paths:
+      - path: /one
+        pathType: ImplementationSpecific
+        backend:
+          service:
+            name: one
+            port:
+              number: 80
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: httpbin
+  namespace: fwuh
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: httpbin
+  template:
+    metadata:
+      labels:
+        app: httpbin
+    spec:
+      containers:
+      - image: docker.io/kennethreitz/httpbin
+        name: httpbin
+        ports:
+        - containerPort: 80
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: httpbin
+  namespace: fwuh
+spec:
+  ingressClassName: kong
+  rules:
+  - http:
+      paths:
+      - path: /two
+        pathType: ImplementationSpecific
+        backend:
+          service:
+            name: two
+            port:
+              number: 80
diff --git a/go.mod b/go.mod
index 96277727b..ad94d0525 100644
--- a/go.mod
+++ b/go.mod
@@ -25,6 +25,7 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/avast/retry-go/v4 v4.6.0
 	github.com/blang/semver/v4 v4.0.0
+	github.com/dominikbraun/graph v0.23.0
 	github.com/go-logr/logr v1.4.1
 	github.com/go-logr/zapr v1.3.0
 	github.com/goccy/go-json v0.10.2
@@ -32,7 +33,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/jpillora/backoff v1.0.0
 	github.com/kong/go-database-reconciler v1.10.0
-	github.com/kong/go-kong v0.54.0
+	github.com/kong/go-kong v0.55.0
 	github.com/kong/kubernetes-telemetry v0.1.3
 	github.com/kong/kubernetes-testing-framework v0.47.0
 	github.com/lithammer/dedent v1.1.0
@@ -51,11 +52,11 @@ require (
 	github.com/testcontainers/testcontainers-go v0.31.0
 	go.uber.org/zap v1.27.0
 	google.golang.org/api v0.180.0
-	k8s.io/api v0.30.0
-	k8s.io/apiextensions-apiserver v0.30.0
-	k8s.io/apimachinery v0.30.0
-	k8s.io/client-go v0.30.0
-	k8s.io/component-base v0.30.0
+	k8s.io/api v0.30.1
+	k8s.io/apiextensions-apiserver v0.30.1
+	k8s.io/apimachinery v0.30.1
+	k8s.io/client-go v0.30.1
+	k8s.io/component-base v0.30.1
 	sigs.k8s.io/controller-runtime v0.18.2
 	sigs.k8s.io/gateway-api v1.1.0
 	sigs.k8s.io/kustomize/api v0.17.1
@@ -94,7 +95,7 @@ require (
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/distribution/reference v0.5.0 // indirect
-	github.com/docker/docker v26.1.2+incompatible
+	github.com/docker/docker v26.1.3+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.12.0 // indirect
@@ -215,7 +216,7 @@ require (
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240506185236-b8a5c65736ae // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240506185236-b8a5c65736ae // indirect
-	google.golang.org/grpc v1.63.2
+	google.golang.org/grpc v1.64.0
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
diff --git a/go.sum b/go.sum
index 8b9aaf37c..3266a5018 100644
--- a/go.sum
+++ b/go.sum
@@ -78,12 +78,14 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
 github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
-github.com/docker/docker v26.1.2+incompatible h1:UVX5ZOrrfTGZZYEP+ZDq3Xn9PdHNXaSYMFPDumMqG2k=
-github.com/docker/docker v26.1.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v26.1.3+incompatible h1:lLCzRbrVZrljpVNobJu1J2FHk8V0s4BawoZippkc+xo=
+github.com/docker/docker v26.1.3+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
 github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
+github.com/dominikbraun/graph v0.23.0 h1:TdZB4pPqCLFxYhdyMFb1TBdFxp8XLcJfTTBQucVPgCo=
+github.com/dominikbraun/graph v0.23.0/go.mod h1:yOjYyogZLY1LSG9E33JWZJiq5k83Qy2C6POAuiViluc=
 github.com/emicklei/go-restful/v3 v3.12.0 h1:y2DdzBAURM29NFF94q6RaY4vjIH1rtwDapwQtU84iWk=
 github.com/emicklei/go-restful/v3 v3.12.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/envoyproxy/go-control-plane v0.9.0/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
@@ -246,8 +248,8 @@ github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJw
 github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/kong/go-database-reconciler v1.10.0 h1:502Nn7CTsUNZyClL5bjrhkFrendVrPxDkP4ZGlBfsdg=
 github.com/kong/go-database-reconciler v1.10.0/go.mod h1:88/u23NIhkQr7SeTP1p4nLcnWCm8AMCwCF1f7Telw+w=
-github.com/kong/go-kong v0.54.0 h1:HZkZJRREJs/azkgJxMWr2jANQQfg8xXsAiFKu+d1nrs=
-github.com/kong/go-kong v0.54.0/go.mod h1:51rSSjgSZKukXgn5nNYbJUx0Et/islqR+LTltwCyjG8=
+github.com/kong/go-kong v0.55.0 h1:lonKRzsDGk12dh9E+y+pWnY2ThXhKuMHjzBHSpCvQLw=
+github.com/kong/go-kong v0.55.0/go.mod h1:i1cMgTu6RYPHSyMpviShddRnc+DML/vlpgKC00hr8kU=
 github.com/kong/kubernetes-telemetry v0.1.3 h1:Hz2tkHGIIUqbn1x46QRDmmNjbEtJyxyOvHSPne3uPto=
 github.com/kong/kubernetes-telemetry v0.1.3/go.mod h1:wB7o8dOKa5R396CyiU0sPa8am/g3c5DKd/qrn/Vmb+k=
 github.com/kong/kubernetes-testing-framework v0.47.0 h1:+9AfTinsGwtxVOcQ/dsDDpwHr4YPpXogKc0THauYJjA=
@@ -597,8 +599,8 @@ google.golang.org/grpc v1.23.0/go.mod h1:Y5yQAOtifL1yxbo5wqy6BxZv8vAUGQwXBOALyac
 google.golang.org/grpc v1.25.1/go.mod h1:c3i+UQWmh7LiEpx4sFZnkU36qjEYZ0imhYfXVyQciAY=
 google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 google.golang.org/grpc v1.33.2/go.mod h1:JMHMWHQWaTccqQQlmk3MJZS+GWXOdAesneDmEnv2fbc=
-google.golang.org/grpc v1.63.2 h1:MUeiw1B2maTVZthpU5xvASfTh3LDbxHd6IJ6QQVU+xM=
-google.golang.org/grpc v1.63.2/go.mod h1:WAX/8DgncnokcFUldAxq7GeB5DXHDbMF+lLvDomNkRA=
+google.golang.org/grpc v1.64.0 h1:KH3VH9y/MgNQg1dE7b3XfVK0GsPSIzJwdF617gUSbvY=
+google.golang.org/grpc v1.64.0/go.mod h1:oxjF8E3FBnjp+/gVFYdWacaLDx9na1aqy9oovLpxQYg=
 google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
 google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
 google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
@@ -635,18 +637,18 @@ gotest.tools/v3 v3.5.0 h1:Ljk6PdHdOhAb5aDMWXjDLMMhph+BpztA4v1QdqEW2eY=
 gotest.tools/v3 v3.5.0/go.mod h1:isy3WKz7GK6uNw/sbHzfKBLvlvXwUyV06n6brMxxopU=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
-k8s.io/api v0.30.0 h1:siWhRq7cNjy2iHssOB9SCGNCl2spiF1dO3dABqZ8niA=
-k8s.io/api v0.30.0/go.mod h1:OPlaYhoHs8EQ1ql0R/TsUgaRPhpKNxIMrKQfWUp8QSE=
-k8s.io/apiextensions-apiserver v0.30.0 h1:jcZFKMqnICJfRxTgnC4E+Hpcq8UEhT8B2lhBcQ+6uAs=
-k8s.io/apiextensions-apiserver v0.30.0/go.mod h1:N9ogQFGcrbWqAY9p2mUAL5mGxsLqwgtUce127VtRX5Y=
-k8s.io/apimachinery v0.30.0 h1:qxVPsyDM5XS96NIh9Oj6LavoVFYff/Pon9cZeDIkHHA=
-k8s.io/apimachinery v0.30.0/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
+k8s.io/api v0.30.1 h1:kCm/6mADMdbAxmIh0LBjS54nQBE+U4KmbCfIkF5CpJY=
+k8s.io/api v0.30.1/go.mod h1:ddbN2C0+0DIiPntan/bye3SW3PdwLa11/0yqwvuRrJM=
+k8s.io/apiextensions-apiserver v0.30.1 h1:4fAJZ9985BmpJG6PkoxVRpXv9vmPUOVzl614xarePws=
+k8s.io/apiextensions-apiserver v0.30.1/go.mod h1:R4GuSrlhgq43oRY9sF2IToFh7PVlF1JjfWdoG3pixk4=
+k8s.io/apimachinery v0.30.1 h1:ZQStsEfo4n65yAdlGTfP/uSHMQSoYzU/oeEbkmF7P2U=
+k8s.io/apimachinery v0.30.1/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
 k8s.io/cli-runtime v0.29.4 h1:QvUrddBxVX6XFJ6z64cGpEk7e4bQduKweqbqq+qBd9g=
 k8s.io/cli-runtime v0.29.4/go.mod h1:NmklYuZ4DLfOr2XEIT8Nzl883KMZUCv7KMj3wMHayCA=
-k8s.io/client-go v0.30.0 h1:sB1AGGlhY/o7KCyCEQ0bPWzYDL0pwOZO4vAtTSh/gJQ=
-k8s.io/client-go v0.30.0/go.mod h1:g7li5O5256qe6TYdAMyX/otJqMhIiGgTapdLchhmOaY=
-k8s.io/component-base v0.30.0 h1:cj6bp38g0ainlfYtaOQuRELh5KSYjhKxM+io7AUIk4o=
-k8s.io/component-base v0.30.0/go.mod h1:V9x/0ePFNaKeKYA3bOvIbrNoluTSG+fSJKjLdjOoeXQ=
+k8s.io/client-go v0.30.1 h1:uC/Ir6A3R46wdkgCV3vbLyNOYyCJ8oZnjtJGKfytl/Q=
+k8s.io/client-go v0.30.1/go.mod h1:wrAqLNs2trwiCH/wxxmT/x3hKVH9PuV0GGW0oDoHVqc=
+k8s.io/component-base v0.30.1 h1:bvAtlPh1UrdaZL20D9+sWxsJljMi0QZ3Lmw+kmZAaxQ=
+k8s.io/component-base v0.30.1/go.mod h1:e/X9kDiOebwlI41AvBHuWdqFriSRrX50CdwA9TFaHLI=
 k8s.io/klog/v2 v2.120.1 h1:QXU6cPEOIslTGvZaXvFWiP9VKyeet3sawzTOvdXb4Vw=
 k8s.io/klog/v2 v2.120.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20240430033511-f0e62f92d13f h1:0LQagt0gDpKqvIkAMPaRGcXawNMouPECM1+F9BVxEaM=
diff --git a/hack/generators/cache-stores-snapshot/main.go b/hack/generators/cache-stores-snapshot/main.go
deleted file mode 100644
index 834ed2d01..000000000
--- a/hack/generators/cache-stores-snapshot/main.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package main
-
-import (
-	"bytes"
-	"os"
-	"reflect"
-	"text/template"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
-)
-
-// This code generates `store.CacheStores.TakeSnapshot` method.
-
-const (
-	outputFile = "zz_generated_cache_stores_snapshot.go"
-
-	outputTemplate = `// Code generated by hack/generators/cache/snapshot.go; DO NOT EDIT.
-package store
-
-import (
-	"fmt"
-
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/tools/cache"
-)
-
-func (c CacheStores) TakeSnapshot() (CacheStores, error) {
-	// Create a fresh CacheStores instance to store the snapshot.
-	snapshot := NewCacheStores()
-
-	// Gather cache store fields for all kinds.
-	allStores := []cache.Store{
-		{{- range . }}
-		c.{{ . }},
-		{{- end }}
-	}
-
-	c.l.RLock()
-	defer c.l.RUnlock()
-
-	// Iterate over all stores and add a deep copy of each object to the snapshot.
-	for _, store := range allStores {
-		for _, item := range store.List() {
-			obj, ok := item.(runtime.Object)
-			if !ok {
-				return CacheStores{}, fmt.Errorf("expected runtime.Object, got %T", item)
-			}
-
-			copiedObj := obj.DeepCopyObject()
-			if err := snapshot.Add(copiedObj); err != nil {
-				return CacheStores{}, err
-			}
-		}
-	}
-
-	return snapshot, nil
-}
-`
-)
-
-func main() {
-	cacheStoresType := reflect.TypeOf(store.CacheStores{})
-	var storesFieldsNames []string
-	for i := 0; i < cacheStoresType.NumField(); i++ {
-		field := cacheStoresType.Field(i)
-		// Only collect fields of `Store` type as `CacheStores` may include other ones (e.g. a mutex).
-		if field.Type.Name() == "Store" {
-			storesFieldsNames = append(storesFieldsNames, field.Name)
-		}
-	}
-
-	tpl, err := template.New("output").Parse(outputTemplate)
-	if err != nil {
-		panic(err)
-	}
-
-	contents := &bytes.Buffer{}
-	if err := tpl.Execute(contents, storesFieldsNames); err != nil {
-		panic(err)
-	}
-
-	if err := os.WriteFile(outputFile, contents.Bytes(), 0o600); err != nil {
-		panic(err)
-	}
-}
diff --git a/hack/generators/cache-stores/main.go b/hack/generators/cache-stores/main.go
new file mode 100644
index 000000000..7d9fb468c
--- /dev/null
+++ b/hack/generators/cache-stores/main.go
@@ -0,0 +1,51 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"text/template"
+
+	"github.com/Masterminds/sprig/v3"
+	"github.com/samber/lo"
+)
+
+type cacheStoreSupportedType struct {
+	// Type is the name of the type that the store supports (e.g. Ingress, Service, etc.).
+	Type string
+
+	// Package is the package name of the type (e.g. gatewayapi, netv1, etc.).
+	Package string
+
+	// KeyFunc is a function to be used as the type's store's KeyFunc.
+	// Defaults to `namespacedKeyFunc` if not provided.
+	KeyFunc string
+
+	// StoreField is the name of the field in the CacheStores struct that holds the cache store.
+	// Optional: if not provided, the field name will be the same as the type name.
+	StoreField string
+}
+
+const (
+	clusterWideKeyFunc string = "clusterWideKeyFunc"
+)
+
+func main() {
+	lo.Must0(renderTemplate(cacheStoresTemplate, cacheStoresOutputFile))
+	lo.Must0(renderTemplate(cacheStoresTestTemplate, cacheStoresTestOutputFile))
+}
+
+func renderTemplate(templateContent string, outputFile string) error {
+	tpl, err := template.New("tpl").Funcs(sprig.TxtFuncMap()).Parse(templateContent)
+	if err != nil {
+		return fmt.Errorf("failed to parse template for %s: %w", outputFile, err)
+	}
+	contents := &bytes.Buffer{}
+	if err := tpl.Execute(contents, supportedTypes); err != nil {
+		return fmt.Errorf("failed to execute template for %s: %w", outputFile, err)
+	}
+	if err := os.WriteFile(outputFile, contents.Bytes(), 0o600); err != nil {
+		return fmt.Errorf("failed to write file %s: %w", outputFile, err)
+	}
+	return nil
+}
diff --git a/hack/generators/cache-stores/spec.go b/hack/generators/cache-stores/spec.go
new file mode 100644
index 000000000..befcbbf53
--- /dev/null
+++ b/hack/generators/cache-stores/spec.go
@@ -0,0 +1,111 @@
+package main
+
+// supportedTypes is a list of types that the cache stores support.
+// To add a new type support, add a new entry to this list.
+var supportedTypes = []cacheStoreSupportedType{
+	// Core Kubernetes types
+	{
+		Type:       "Ingress",
+		Package:    "netv1",
+		StoreField: "IngressV1",
+	},
+	{
+		Type:       "IngressClass",
+		Package:    "netv1",
+		StoreField: "IngressClassV1",
+		KeyFunc:    clusterWideKeyFunc,
+	},
+	{
+		Type:    "Service",
+		Package: "corev1",
+	},
+	{
+		Type:    "Secret",
+		Package: "corev1",
+	},
+	{
+		Type:    "EndpointSlice",
+		Package: "discoveryv1",
+	},
+	// Gateway API types
+	{
+		Type:    "HTTPRoute",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "UDPRoute",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "TCPRoute",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "TLSRoute",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "GRPCRoute",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "ReferenceGrant",
+		Package: "gatewayapi",
+	},
+	{
+		Type:    "Gateway",
+		Package: "gatewayapi",
+	},
+	// Kong types
+	{
+		Type:       "KongPlugin",
+		Package:    "kongv1",
+		StoreField: "Plugin",
+	},
+	{
+		Type:       "KongClusterPlugin",
+		Package:    "kongv1",
+		StoreField: "ClusterPlugin",
+		KeyFunc:    clusterWideKeyFunc,
+	},
+	{
+		Type:       "KongConsumer",
+		Package:    "kongv1",
+		StoreField: "Consumer",
+	},
+	{
+		Type:       "KongConsumerGroup",
+		Package:    "kongv1beta1",
+		StoreField: "ConsumerGroup",
+	},
+	{
+		Type:    "KongIngress",
+		Package: "kongv1",
+	},
+	{
+		Type:    "TCPIngress",
+		Package: "kongv1beta1",
+	},
+	{
+		Type:    "UDPIngress",
+		Package: "kongv1beta1",
+	},
+	{
+		Type:    "KongUpstreamPolicy",
+		Package: "kongv1beta1",
+	},
+	{
+		Type:       "IngressClassParameters",
+		Package:    "kongv1alpha1",
+		StoreField: "IngressClassParametersV1alpha1",
+	},
+	{
+		Type:    "KongServiceFacade",
+		Package: "incubatorv1alpha1",
+	},
+	{
+		Type:    "KongVault",
+		Package: "kongv1alpha1",
+		KeyFunc: clusterWideKeyFunc,
+	},
+}
diff --git a/hack/generators/cache-stores/templates.go b/hack/generators/cache-stores/templates.go
new file mode 100644
index 000000000..2883c9956
--- /dev/null
+++ b/hack/generators/cache-stores/templates.go
@@ -0,0 +1,168 @@
+package main
+
+const (
+	cacheStoresOutputFile = "zz_generated_cache_stores.go"
+	cacheStoresTemplate   = `// Code generated by hack/generators/cache-stores/main.go; DO NOT EDIT.
+// If you want to add a new type to the cache store, you need to add a new entry to the supportedTypes list in spec.go.
+package store
+
+import ( 
+	"fmt"
+	"sync"
+	
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	netv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+	
+)
+
+// CacheStores stores cache.Store for all Kinds of k8s objects that
+// the Ingress Controller reads.
+type CacheStores struct {
+	{{- range . }}
+	{{ .StoreField | default .Type }} cache.Store
+	{{- end }}
+
+	l *sync.RWMutex
+}
+
+// NewCacheStores is a convenience function for CacheStores to initialize all attributes with new cache stores.
+func NewCacheStores() CacheStores {
+	return CacheStores{
+		{{- range . }}
+		{{ .StoreField | default .Type }}: cache.NewStore({{ if eq .KeyFunc "clusterWideKeyFunc" }}clusterWideKeyFunc{{ else }}namespacedKeyFunc{{ end }}),
+		{{- end }}
+
+		l: &sync.RWMutex{},
+	}
+}
+
+// Get checks whether or not there's already some version of the provided object present in the cache.
+func (c CacheStores) Get(obj runtime.Object) (item interface{}, exists bool, err error) {
+	c.l.RLock()
+	defer c.l.RUnlock()
+
+	switch obj := obj.(type) {
+	{{- range . }}
+	case *{{ .Package }}.{{ .Type }}:
+		return c.{{ .StoreField | default .Type }}.Get(obj)
+	{{- end }}
+	}
+	return nil, false, fmt.Errorf("%T is not a supported cache object type", obj)
+}
+
+// Add stores a provided runtime.Object into the CacheStore if it's of a supported type.
+// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
+func (c CacheStores) Add(obj runtime.Object) error {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	switch obj := obj.(type) {
+	{{- range . }}
+	case *{{ .Package }}.{{ .Type }}:
+		return c.{{ .StoreField | default .Type }}.Add(obj)
+	{{- end }}	
+	}
+	return fmt.Errorf("cannot add unsupported kind %q to the store", obj.GetObjectKind().GroupVersionKind())
+}
+
+// Delete removes a provided runtime.Object from the CacheStore if it's of a supported type.
+// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
+func (c CacheStores) Delete(obj runtime.Object) error {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	switch obj := obj.(type) {
+	{{- range . }}
+	case *{{ .Package }}.{{ .Type }}:
+		return c.{{ .StoreField | default .Type }}.Delete(obj)
+	{{- end }}
+	}
+	return fmt.Errorf("cannot delete unsupported kind %q from the store", obj.GetObjectKind().GroupVersionKind())
+}
+
+// ListAllStores returns a list of all cache stores embedded in the struct.
+func (c CacheStores) ListAllStores() []cache.Store {
+	return []cache.Store{
+		{{- range . }}
+		c.{{ .StoreField | default .Type }},
+		{{- end }}
+	}
+}
+
+// SupportedTypes returns a list of supported types for the cache.
+func (c CacheStores) SupportedTypes() []client.Object {
+	return []client.Object{
+		{{- range . }}
+		&{{ .Package }}.{{ .Type }}{},
+		{{- end }}
+	}
+}
+`
+
+	cacheStoresTestOutputFile = "zz_generated_cache_stores_test.go"
+	cacheStoresTestTemplate   = `// Code generated by hack/generators/cache-stores/main.go; DO NOT EDIT.
+// If you want to add a new type to the cache store, you need to add a new entry to the supportedTypes list in spec.go.
+package store_test
+
+import (
+	"testing"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	netv1 "k8s.io/api/networking/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestCacheStores(t *testing.T) {
+	testCases := []struct {
+		name          string
+		objectToStore client.Object
+	}{
+		{{ range . }}
+		{
+			name: "{{ .Type }}",
+			objectToStore: &{{ .Package }}.{{ .Type }}{},
+		},
+		{{ end }}
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := store.NewCacheStores()
+			err := s.Add(tc.objectToStore)
+			require.NoError(t, err)
+
+			storedObj, ok, err := s.Get(tc.objectToStore)
+			require.NoError(t, err)
+			require.True(t, ok)
+			require.Equal(t, tc.objectToStore, storedObj)
+
+			err = s.Delete(tc.objectToStore)
+			require.NoError(t, err)
+
+			_, ok, err = s.Get(tc.objectToStore)
+			require.NoError(t, err, err)
+			require.False(t, ok)
+		})
+	}
+}
+`
+)
diff --git a/internal/admission/adminapi_provider.go b/internal/admission/adminapi_provider.go
index 7868c3d67..9736aafd1 100644
--- a/internal/admission/adminapi_provider.go
+++ b/internal/admission/adminapi_provider.go
@@ -69,6 +69,14 @@ func (p DefaultAdminAPIServicesProvider) GetVaultsService() (kong.AbstractVaultS
 	return c.Vaults, true
 }
 
+func (p DefaultAdminAPIServicesProvider) GetSchemasService() (kong.AbstractSchemaService, bool) {
+	c, ok := p.designatedAdminAPIClient()
+	if !ok {
+		return nil, ok
+	}
+	return c.Schemas, true
+}
+
 func (p DefaultAdminAPIServicesProvider) designatedAdminAPIClient() (*kong.Client, bool) {
 	gwClients := p.gatewayClientsProvider.GatewayClients()
 	if len(gwClients) == 0 {
diff --git a/internal/admission/errors.go b/internal/admission/errors.go
index 33917f8be..83e876af4 100644
--- a/internal/admission/errors.go
+++ b/internal/admission/errors.go
@@ -9,6 +9,8 @@ const (
 	ErrTextConsumerGroupUnsupported           = "consumer group support requires Kong Enterprise"
 	ErrTextConsumerGroupUnlicensed            = "consumer group support requires a valid Kong Enterprise license"
 	ErrTextConsumerGroupUnexpected            = "unexpected error during checking support for consumer group"
+	ErrTextCustomEntityFieldsUnmarshalFailed  = "failed to unmarshal fields of custom entity: %v"
+	ErrTextCustomEntityGetSchemaFailed        = "failed to get schema of Kong entity type '%s': %v"
 	ErrTextFailedToRetrieveSecret             = "could not retrieve secrets from the kubernetes API" //nolint:revive,gosec
 	ErrTextPluginConfigInvalid                = "could not parse plugin configuration"
 	ErrTextPluginConfigValidationFailed       = "unable to validate plugin schema"
diff --git a/internal/admission/handler.go b/internal/admission/handler.go
index 9c77afe37..0b3fd95db 100644
--- a/internal/admission/handler.go
+++ b/internal/admission/handler.go
@@ -103,6 +103,11 @@ var (
 		Version:  kongv1alpha1.SchemeGroupVersion.Version,
 		Resource: "kongvaults",
 	}
+	kongCustomEntityGVResource = metav1.GroupVersionResource{
+		Group:    kongv1alpha1.SchemeGroupVersion.Group,
+		Version:  kongv1alpha1.SchemeGroupVersion.Version,
+		Resource: "kongcustomentities",
+	}
 	secretGVResource = metav1.GroupVersionResource{
 		Group:    corev1.SchemeGroupVersion.Group,
 		Version:  corev1.SchemeGroupVersion.Version,
@@ -144,6 +149,8 @@ func (h RequestHandler) handleValidation(ctx context.Context, request admissionv
 		return h.handleKongIngress(ctx, request, responseBuilder)
 	case kongVaultGVResource:
 		return h.handleKongVault(ctx, request, responseBuilder)
+	case kongCustomEntityGVResource:
+		return h.handleKongCustomEntity(ctx, request, responseBuilder)
 	case serviceGVResource:
 		return h.handleService(ctx, request, responseBuilder)
 	case ingressGVResource:
@@ -492,3 +499,21 @@ func (h RequestHandler) handleKongVault(ctx context.Context, request admissionv1
 
 	return responseBuilder.Allowed(ok).WithMessage(message).Build(), nil
 }
+
+// +kubebuilder:webhook:verbs=create;update,groups=configuration.konghq.com,resources=kongcustomentities,versions=v1alpha1,name=kongcustomentities.validation.ingress-controller.konghq.com,path=/,webhookVersions=v1,matchPolicy=equivalent,mutating=false,failurePolicy=fail,sideEffects=None,admissionReviewVersions=v1
+
+func (h RequestHandler) handleKongCustomEntity(ctx context.Context, request admissionv1.AdmissionRequest, responseBuilder *ResponseBuilder) (*admissionv1.AdmissionResponse, error) {
+	kongCustomEntity := kongv1alpha1.KongCustomEntity{}
+
+	_, _, err := codecs.UniversalDeserializer().Decode(request.Object.Raw, nil, &kongCustomEntity)
+	if err != nil {
+		return nil, err
+	}
+
+	ok, message, err := h.Validator.ValidateCustomEntity(ctx, kongCustomEntity)
+	if err != nil {
+		return nil, err
+	}
+
+	return responseBuilder.Allowed(ok).WithMessage(message).Build(), nil
+}
diff --git a/internal/admission/validator.go b/internal/admission/validator.go
index d376bc87e..bb7f05b10 100644
--- a/internal/admission/validator.go
+++ b/internal/admission/validator.go
@@ -38,6 +38,7 @@ type KongValidator interface {
 	ValidatePlugin(ctx context.Context, plugin kongv1.KongPlugin, overrideSecrets []*corev1.Secret) (bool, string, error)
 	ValidateClusterPlugin(ctx context.Context, plugin kongv1.KongClusterPlugin, overrideSecrets []*corev1.Secret) (bool, string, error)
 	ValidateVault(ctx context.Context, vault kongv1alpha1.KongVault) (bool, string, error)
+	ValidateCustomEntity(ctx context.Context, entity kongv1alpha1.KongCustomEntity) (bool, string, error)
 	ValidateCredential(ctx context.Context, secret corev1.Secret) (bool, string)
 	ValidateGateway(ctx context.Context, gateway gatewayapi.Gateway) (bool, string, error)
 	ValidateHTTPRoute(ctx context.Context, httproute gatewayapi.HTTPRoute) (bool, string, error)
@@ -54,6 +55,7 @@ type AdminAPIServicesProvider interface {
 	GetInfoService() (kong.AbstractInfoService, bool)
 	GetRoutesService() (kong.AbstractRouteService, bool)
 	GetVaultsService() (kong.AbstractVaultService, bool)
+	GetSchemasService() (kong.AbstractSchemaService, bool)
 }
 
 // ConsumerGetter is an interface for retrieving KongConsumers.
@@ -630,3 +632,55 @@ func (m *managerClientConsumerGetter) ListAllConsumers(ctx context.Context) ([]k
 	}
 	return consumers.Items, nil
 }
+
+func (validator KongHTTPValidator) ValidateCustomEntity(ctx context.Context, entity kongv1alpha1.KongCustomEntity) (bool, string, error) {
+	logger := validator.Logger.WithValues("namespace", entity.Namespace, "name", entity.Name, "kind", kongv1alpha1.KongCustomEntityKind)
+	// If the spec.contollerName does not match the ingress class name,
+	// and the ingress class annotation does not match the ingress class name either,
+	// ignore it as it is not controlled by the controller.
+	if (!validator.ingressClassMatcher(&entity.ObjectMeta, annotations.IngressClassKey, annotations.ExactClassMatch)) &&
+		(!validator.ingressClassMatcher(&metav1.ObjectMeta{
+			Annotations: map[string]string{
+				annotations.IngressClassKey: entity.Spec.ControllerName,
+			},
+		}, annotations.IngressClassKey, annotations.ExactClassMatch)) {
+		return true, "", nil
+	}
+
+	fields, err := kongstate.RawConfigToConfiguration(entity.Spec.Fields.Raw)
+	if err != nil {
+		return false, fmt.Sprintf(ErrTextCustomEntityFieldsUnmarshalFailed, err), nil
+	}
+
+	schemaService, hasClient := validator.AdminAPIServicesProvider.GetSchemasService()
+	// Skip validation on Kong gateway if we do not have available client.
+	if !hasClient {
+		logger.V(util.DebugLevel).Info("Skipped because no schema service available")
+		return true, "", nil
+	}
+
+	// Retrieve the schema of the entity from Kong gateway.
+	entityType := entity.Spec.EntityType
+	schema, err := schemaService.Get(ctx, entityType)
+	if err != nil {
+		logger.V(util.DebugLevel).Info("Failed to get schema of entity", "entity_type", entityType, "error", err)
+		return false, fmt.Sprintf(ErrTextCustomEntityGetSchemaFailed, entityType, err), nil
+	}
+
+	// Extract field definitions from the response of getting the schema of the entity.
+	s := kongstate.ExtractEntityFieldDefinitions(schema)
+	// Fill in fields that are "foreign" and required if there are such fields.
+	for fieldName, field := range s.Fields {
+		// Fill in an arbitrary UUID if the entity requires a reference to a foreign key.
+		// This format of foreign reference in entities should satisfy most of foreign fields in entities(services/routes/consumers/...).
+		// The admission webhook needs a placeholder value because the actual value isn't present in the KongCustomEntity spec.
+		// It will be derived from the resource(s) used by the parent KongPlugin during translation.
+		if field.Required && field.Type == kongstate.EntityFieldTypeForeign {
+			fields[fieldName] = map[string]interface{}{
+				"id": util.DefaultUUIDGenerator{}.NewString(),
+			}
+		}
+	}
+	// validate the custom entity against Kong gateway.
+	return schemaService.Validate(ctx, kong.EntityType(entity.Spec.EntityType), fields)
+}
diff --git a/internal/annotations/annotations.go b/internal/annotations/annotations.go
index c5db8a2da..54b4cf400 100644
--- a/internal/annotations/annotations.go
+++ b/internal/annotations/annotations.go
@@ -23,6 +23,7 @@ import (
 	"github.com/samber/lo"
 	netv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
@@ -155,6 +156,34 @@ func ExtractKongPluginsFromAnnotations(anns map[string]string) []string {
 	return kongPluginCRs
 }
 
+type NamespacedKongPlugin k8stypes.NamespacedName
+
+// ExtractNamespacedKongPluginsFromAnnotations extracts a KongPlugin name and optional namespace from an annotation
+// value. Plugins are delimited by ",". Values are either colon-delimited "namespace:name" strings or name-only
+// strings.
+func ExtractNamespacedKongPluginsFromAnnotations(anns map[string]string) []NamespacedKongPlugin {
+	v := pluginsFromAnnotations(anns)
+	if v == "" {
+		return nil
+	}
+	split := strings.Split(v, ",")
+	plugins := make([]NamespacedKongPlugin, 0, len(split))
+	for _, s := range split {
+		if s != "" {
+			plugin := NamespacedKongPlugin{}
+			if strings.Contains(s, ":") {
+				split := strings.Split(s, ":")
+				plugin.Namespace = strings.TrimSpace(split[0])
+				plugin.Name = strings.TrimSpace(split[1])
+			} else {
+				plugin.Name = strings.TrimSpace(s)
+			}
+			plugins = append(plugins, plugin)
+		}
+	}
+	return plugins
+}
+
 // ExtractConfigurationName extracts the name of the KongIngress object that holds
 // information about the configuration to use in Routes, Services and Upstreams.
 func ExtractConfigurationName(anns map[string]string) string {
diff --git a/internal/dataplane/fallback/graph.go b/internal/dataplane/fallback/graph.go
new file mode 100644
index 000000000..595eaae67
--- /dev/null
+++ b/internal/dataplane/fallback/graph.go
@@ -0,0 +1,116 @@
+package fallback
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/dominikbraun/graph"
+	"github.com/samber/lo"
+	k8stypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// ConfigGraph is a graph representation of the Kubernetes resources kept in the cache.
+// Vertices are objects, edges are dependencies between objects (dependency -> dependant).
+// It allows to quickly determine which objects are affected by traversing the graph from
+// the affected object to its dependants.
+//
+// If you want to extend the graph with a new object type, you need to ensure ResolveDependencies
+// function is implemented for that object type. If your object type has no dependencies, you can ignore it.
+type ConfigGraph struct {
+	graph graph.Graph[ObjectHash, client.Object]
+}
+
+// ObjectHash is a unique identifier for a given object that is used as a vertex key in the graph.
+// It could consist of the object's UID only, but we also include the object's kind, namespace and name
+// to make it easier to debug and understand the graph.
+type ObjectHash struct {
+	// UID is the unique identifier of the object.
+	UID k8stypes.UID
+
+	// Kind, Namespace and Name are the object's kind, namespace and name - included for debugging purposes.
+	Kind      string
+	Namespace string
+	Name      string
+}
+
+// String returns a string representation of the ObjectHash. It intentionally does not include the UID
+// as it is not human-readable and is not necessary for debugging purposes.
+func (h ObjectHash) String() string {
+	if h.Namespace == "" {
+		return fmt.Sprintf("%s:%s", h.Kind, h.Name)
+	}
+	return fmt.Sprintf("%s:%s/%s", h.Kind, h.Namespace, h.Name)
+}
+
+// GetObjectHash is a function that returns a unique identifier for a given object that is used as a
+// vertex key in the graph.
+func GetObjectHash(obj client.Object) ObjectHash {
+	return ObjectHash{
+		UID:       obj.GetUID(),
+		Kind:      obj.GetObjectKind().GroupVersionKind().Kind,
+		Namespace: obj.GetNamespace(),
+		Name:      obj.GetName(),
+	}
+}
+
+// NewConfigGraphFromCacheStores creates a new ConfigGraph from the given cache stores. It adds all objects
+// from the cache stores to the graph as vertices as well as edges between objects and their dependencies
+// resolved by the ResolveDependencies function.
+func NewConfigGraphFromCacheStores(c store.CacheStores) (*ConfigGraph, error) {
+	g := graph.New[ObjectHash, client.Object](GetObjectHash, graph.Directed())
+
+	for _, s := range c.ListAllStores() {
+		for _, o := range s.List() {
+			obj, ok := o.(client.Object)
+			if !ok {
+				// Should not happen since all objects in the cache are client.Objects, but better safe than sorry.
+				return nil, fmt.Errorf("expected client.Object, got %T", o)
+			}
+			// Add the object to the graph. It can happen that the object is already in the graph (i.e. was already added
+			// as a dependency of another object), in which case we ignore the error.
+			if err := g.AddVertex(obj); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
+				return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
+			}
+
+			deps, err := ResolveDependencies(c, obj)
+			if err != nil {
+				return nil, fmt.Errorf("failed to resolve dependencies for %s: %w", GetObjectHash(obj), err)
+			}
+			// Add the object's dependencies to the graph.
+			for _, dep := range deps {
+				// Add the dependency to the graph in case it wasn't added before. If it was added before, we ignore the
+				// error.
+				if err := g.AddVertex(dep); err != nil && !errors.Is(err, graph.ErrVertexAlreadyExists) {
+					return nil, fmt.Errorf("failed to add %s to the graph: %w", GetObjectHash(obj), err)
+				}
+
+				// Add an edge from a dependency to the object. If the edge was already added before, we ignore the error.
+				if err := g.AddEdge(GetObjectHash(dep), GetObjectHash(obj)); err != nil && !errors.Is(err, graph.ErrEdgeAlreadyExists) {
+					return nil, fmt.Errorf("failed to add edge from %s to %s: %w", GetObjectHash(obj), GetObjectHash(dep), err)
+				}
+			}
+		}
+	}
+
+	return &ConfigGraph{graph: g}, nil
+}
+
+// AdjacencyMap is a map of object hashes to their neighbours' hashes.
+type AdjacencyMap map[ObjectHash][]ObjectHash
+
+// AdjacencyMap returns a map of object hashes to their neighbours' hashes.
+func (g *ConfigGraph) AdjacencyMap() (AdjacencyMap, error) {
+	am, err := g.graph.AdjacencyMap()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get adjacency map: %w", err)
+	}
+
+	m := make(map[ObjectHash][]ObjectHash)
+	for v, neighbours := range am {
+		m[v] = lo.Keys(neighbours)
+	}
+	return m, nil
+}
diff --git a/internal/dataplane/fallback/graph_dependencies.go b/internal/dataplane/fallback/graph_dependencies.go
new file mode 100644
index 000000000..55476c537
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies.go
@@ -0,0 +1,62 @@
+package fallback
+
+import (
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	netv1 "k8s.io/api/networking/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+// ResolveDependencies resolves dependencies for a given object. Dependencies are all objects referenced by the
+// given object. For example, an Ingress object might refer to an IngressClass, Services, Plugins, etc.
+// Every supported object type should explicitly have a case in this function.
+func ResolveDependencies(cache store.CacheStores, obj client.Object) ([]client.Object, error) {
+	switch obj := obj.(type) {
+	case *netv1.Ingress:
+		return resolveIngressDependencies(cache, obj), nil
+	case *gatewayapi.HTTPRoute:
+		return resolveHTTPRouteDependencies(cache, obj), nil
+	case *gatewayapi.TLSRoute:
+		return resolveTLSRouteDependencies(cache, obj), nil
+	case *gatewayapi.TCPRoute:
+		return resolveTCPRouteDependencies(cache, obj), nil
+	case *gatewayapi.UDPRoute:
+		return resolveUDPRouteDependencies(cache, obj), nil
+	case *gatewayapi.GRPCRoute:
+		return resolveGRPCRouteDependencies(cache, obj), nil
+	case *kongv1.KongPlugin:
+		return resolveKongPluginDependencies(cache, obj), nil
+	case *kongv1.KongClusterPlugin:
+		return resolveKongClusterPluginDependencies(cache, obj), nil
+	case *kongv1beta1.UDPIngress:
+		return resolveUDPIngressDependencies(cache, obj), nil
+	case *kongv1beta1.TCPIngress:
+		return resolveTCPIngressDependencies(cache, obj), nil
+	case *incubatorv1alpha1.KongServiceFacade:
+		return resolveKongServiceFacadeDependencies(cache, obj), nil
+	case *netv1.IngressClass, // Object types that have no dependencies.
+		*corev1.Service,
+		*corev1.Secret,
+		*discoveryv1.EndpointSlice,
+		*gatewayapi.ReferenceGrant,
+		*gatewayapi.Gateway,
+		*kongv1.KongConsumer,
+		*kongv1beta1.KongConsumerGroup,
+		*kongv1.KongIngress,
+		*kongv1beta1.KongUpstreamPolicy,
+		*kongv1alpha1.IngressClassParameters,
+		*kongv1alpha1.KongVault:
+		return nil, nil
+	default:
+		return nil, fmt.Errorf("unsupported object type: %T", obj)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_common.go b/internal/dataplane/fallback/graph_dependencies_common.go
new file mode 100644
index 000000000..5ec408a48
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_common.go
@@ -0,0 +1,28 @@
+package fallback
+
+import (
+	"fmt"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// resolveObjectDependenciesPlugin resolves KongPlugin and KongClusterPlugin dependencies for an arbitrary object
+// that refers them in its annotations.
+func resolveObjectDependenciesPlugin(cache store.CacheStores, obj client.Object) []client.Object {
+	var dependencies []client.Object
+	for _, pluginName := range annotations.ExtractKongPluginsFromAnnotations(obj.GetAnnotations()) {
+		if plugin, exists, err := cache.Plugin.GetByKey(
+			fmt.Sprintf("%s/%s", obj.GetNamespace(), pluginName),
+		); err == nil && exists {
+			dependencies = append(dependencies, plugin.(client.Object))
+		}
+
+		if plugin, exists, err := cache.ClusterPlugin.GetByKey(pluginName); err == nil && exists {
+			dependencies = append(dependencies, plugin.(client.Object))
+		}
+	}
+	return dependencies
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_gatewayapi.go b/internal/dataplane/fallback/graph_dependencies_gatewayapi.go
new file mode 100644
index 000000000..1620b9a63
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_gatewayapi.go
@@ -0,0 +1,137 @@
+package fallback
+
+import (
+	"fmt"
+	"slices"
+
+	"github.com/samber/lo"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
+)
+
+// resolveHTTPRouteDependencies resolves potential dependencies for a given HTTPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveHTTPRouteDependencies(cache store.CacheStores, route *gatewayapi.HTTPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getHTTPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveTCPRouteDependencies resolves potential dependencies for a given TCPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveTCPRouteDependencies(cache store.CacheStores, route *gatewayapi.TCPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getTCPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveUDPRouteDependencies resolves potential dependencies for a given UDPRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveUDPRouteDependencies(cache store.CacheStores, route *gatewayapi.UDPRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getUDPRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveTLSRouteDependencies resolves potential dependencies for a given TLSRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveTLSRouteDependencies(cache store.CacheStores, route *gatewayapi.TLSRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getTLSRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// resolveGRPCRouteDependencies resolves potential dependencies for a given GRPCRoute object:
+// - Service
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveGRPCRouteDependencies(cache store.CacheStores, route *gatewayapi.GRPCRoute) []client.Object {
+	return slices.Concat(
+		resolveGatewayAPIRouteDependenciesBackendRefs(cache, route, getGRPCRouteBackendRefs(route)),
+		resolveObjectDependenciesPlugin(cache, route),
+	)
+}
+
+// gatewayAPIRoute is an interface that represents a GatewayAPI Route object.
+type gatewayAPIRoute interface {
+	client.Object
+	*gatewayapi.HTTPRoute | *gatewayapi.TCPRoute | *gatewayapi.UDPRoute | *gatewayapi.TLSRoute | *gatewayapi.GRPCRoute
+}
+
+// resolveGatewayAPIRouteDependenciesBackendRefs resolves backend references for a given gatewayAPIRoute object.
+func resolveGatewayAPIRouteDependenciesBackendRefs[T gatewayAPIRoute](cache store.CacheStores, route T, backendRefs []gatewayapi.BackendRef) []client.Object {
+	var dependencies []client.Object
+	for _, backendRef := range backendRefs {
+		if !util.IsBackendRefGroupKindSupported(backendRef.Group, backendRef.Kind) {
+			continue
+		}
+		ns := route.GetNamespace()
+		if backendRef.Namespace != nil {
+			ns = string(*backendRef.Namespace)
+		}
+		ingressClass, exists, err := cache.Service.GetByKey(fmt.Sprintf("%s/%s", ns, backendRef.Name))
+		if err == nil && exists {
+			dependencies = append(dependencies, ingressClass.(client.Object))
+		}
+	}
+	return dependencies
+}
+
+func getHTTPRouteBackendRefs(route *gatewayapi.HTTPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, lo.Map(rule.BackendRefs, func(b gatewayapi.HTTPBackendRef, _ int) gatewayapi.BackendRef {
+			return b.BackendRef
+		})...)
+	}
+	return backendRefs
+}
+
+func getTCPRouteBackendRefs(route *gatewayapi.TCPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
+}
+
+func getUDPRouteBackendRefs(route *gatewayapi.UDPRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
+}
+
+func getTLSRouteBackendRefs(route *gatewayapi.TLSRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, rule.BackendRefs...)
+	}
+	return backendRefs
+}
+
+func getGRPCRouteBackendRefs(route *gatewayapi.GRPCRoute) []gatewayapi.BackendRef {
+	var backendRefs []gatewayapi.BackendRef
+	for _, rule := range route.Spec.Rules {
+		backendRefs = append(backendRefs, lo.Map(rule.BackendRefs, func(b gatewayapi.GRPCBackendRef, _ int) gatewayapi.BackendRef {
+			return b.BackendRef
+		})...)
+	}
+	return backendRefs
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_ingress.go b/internal/dataplane/fallback/graph_dependencies_ingress.go
new file mode 100644
index 000000000..21b8f1f7f
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_ingress.go
@@ -0,0 +1,92 @@
+package fallback
+
+import (
+	"fmt"
+	"slices"
+
+	netv1 "k8s.io/api/networking/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/translator/subtranslator"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+)
+
+// resolveIngressDependencies resolves potential dependencies for an Ingress object:
+// - IngressClass
+// - Service
+// - KongServiceFacade
+// - KongUpstreamPolicy
+// - KongPlugin
+// - KongClusterPlugin.
+func resolveIngressDependencies(cache store.CacheStores, ingress *netv1.Ingress) []client.Object {
+	return slices.Concat(
+		resolveIngressDependenciesIngressClass(cache, ingress),
+		resolveIngressDependenciesService(cache, ingress),
+		resolveIngressDependenciesKongUpstreamPolicy(cache, ingress),
+		resolveObjectDependenciesPlugin(cache, ingress),
+	)
+}
+
+// resolveIngressDependenciesIngressClass resolves IngressClass dependencies for an Ingress object.
+func resolveIngressDependenciesIngressClass(cache store.CacheStores, ingress *netv1.Ingress) []client.Object {
+	resolveIngressClassName := func() string {
+		// IngressClass can be referenced by the Ingress object in two ways:
+		// 1. Using the IngressClassName field.
+		if ingress.Spec.IngressClassName != nil && *ingress.Spec.IngressClassName != "" {
+			return *ingress.Spec.IngressClassName
+		}
+		// 2. Using the annotation in the Ingress object.
+		if ingressClassName, ok := ingress.Annotations[annotations.IngressClassKey]; ok && ingressClassName != "" {
+			return ingressClassName
+		}
+		return "" // No IngressClass reference was found.
+	}
+
+	var dependencies []client.Object
+	if ingressClassName := resolveIngressClassName(); ingressClassName != "" {
+		ingressClass, exists, err := cache.IngressClassV1.GetByKey(ingressClassName)
+		if err == nil && exists {
+			dependencies = append(dependencies, ingressClass.(client.Object))
+		}
+	}
+	return dependencies
+}
+
+// resolveIngressDependenciesService resolves Service and KongServiceFacade dependencies for an Ingress object.
+func resolveIngressDependenciesService(cache store.CacheStores, ingress *netv1.Ingress) []client.Object {
+	var dependencies []client.Object
+	for _, rule := range ingress.Spec.Rules {
+		if rule.HTTP == nil {
+			continue
+		}
+		for _, path := range rule.HTTP.Paths {
+			if path.Backend.Service != nil {
+				service, exists, err := cache.Service.GetByKey(fmt.Sprintf("%s/%s", ingress.GetNamespace(), path.Backend.Service.Name))
+				if err == nil && exists {
+					dependencies = append(dependencies, service.(client.Object))
+				}
+			}
+
+			if resource := path.Backend.Resource; resource != nil && subtranslator.IsKongServiceFacade(resource) {
+				kongServiceFacade, exists, err := cache.KongServiceFacade.GetByKey(fmt.Sprintf("%s/%s", ingress.GetNamespace(), resource.Name))
+				if err == nil && exists {
+					dependencies = append(dependencies, kongServiceFacade.(client.Object))
+				}
+			}
+		}
+	}
+	return dependencies
+}
+
+// resolveIngressDependenciesKongUpstreamPolicy resolves KongUpstreamPolicy dependencies for an Ingress object.
+func resolveIngressDependenciesKongUpstreamPolicy(cache store.CacheStores, ingress *netv1.Ingress) []client.Object {
+	upstreamPolicyName, ok := annotations.ExtractUpstreamPolicy(ingress.Annotations)
+	if ok {
+		upstreamPolicy, exists, err := cache.KongUpstreamPolicy.GetByKey(fmt.Sprintf("%s/%s", ingress.GetNamespace(), upstreamPolicyName))
+		if err == nil && exists {
+			return []client.Object{upstreamPolicy.(client.Object)}
+		}
+	}
+	return nil
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_kong.go b/internal/dataplane/fallback/graph_dependencies_kong.go
new file mode 100644
index 000000000..ddcf3f7f5
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_kong.go
@@ -0,0 +1,35 @@
+package fallback
+
+import (
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+func resolveKongPluginDependencies(_ store.CacheStores, _ *kongv1.KongPlugin) []client.Object {
+	return nil
+}
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+func resolveKongClusterPluginDependencies(_ store.CacheStores, _ *kongv1.KongClusterPlugin) []client.Object {
+	return nil
+}
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+func resolveUDPIngressDependencies(_ store.CacheStores, _ *kongv1beta1.UDPIngress) []client.Object {
+	return nil
+}
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+func resolveTCPIngressDependencies(_ store.CacheStores, _ *kongv1beta1.TCPIngress) []client.Object {
+	return nil
+}
+
+// TODO: https://github.com/Kong/kubernetes-ingress-controller/issues/6026
+func resolveKongServiceFacadeDependencies(_ store.CacheStores, _ *incubatorv1alpha1.KongServiceFacade) []client.Object {
+	return nil
+}
diff --git a/internal/dataplane/kongstate/customentity.go b/internal/dataplane/kongstate/customentity.go
new file mode 100644
index 000000000..de9fadf1d
--- /dev/null
+++ b/internal/dataplane/kongstate/customentity.go
@@ -0,0 +1,91 @@
+package kongstate
+
+import "github.com/kong/go-kong/kong"
+
+// EntityFieldType represents type of a Kong entity field.
+// possible field types include boolean, integer, number, string, array, set, map, record, json, foreign.
+type EntityFieldType string
+
+// These types and field properties are defined upstream in the Kong DAO:
+// https://github.com/Kong/kong/blob/3.6.1/kong/db/schema/init.lua#L131-L143
+// https://docs.konghq.com/gateway/latest/plugin-development/custom-entities/#define-a-schema
+
+const (
+	EntityFieldTypeBoolean EntityFieldType = "boolean"
+	EntityFieldTypeInteger EntityFieldType = "integer"
+	EntityFieldTypeNumber  EntityFieldType = "number"
+	EntityFieldTypeString  EntityFieldType = "string"
+	EntityFieldTypeSet     EntityFieldType = "set"
+	EntityFieldTypeArray   EntityFieldType = "array"
+	EntityFieldTypeMap     EntityFieldType = "map"
+	EntityFieldTypeRecord  EntityFieldType = "record"
+	EntityFieldTypeJSON    EntityFieldType = "json"
+	// EntityFieldTypeForeign means that this field refers to another entity by the key (typically ID).
+	EntityFieldTypeForeign EntityFieldType = "foreign"
+)
+
+type EntityField struct {
+	// Name is the name of the field.
+	Name string `json:"name"`
+	// Type stands for the type of the field.
+	Type EntityFieldType `json:"type"`
+	// Required is true means that the field must present in the entity.
+	Required bool `json:"required,omitempty"`
+	// Auto is true means that the field is automatically generated when it is created in Kong gateway.
+	Auto bool `json:"auto,omitempty"`
+	// UUID is true means that the field is in UUID format.
+	UUID bool `json:"uuid,omitempty"`
+	// Default is the default value of the field when it is not given.
+	Default interface{} `json:"default,omitempty"`
+	// Reference is the type referring entity when the field is "foreign" to refer to another entity.
+	Reference string `json:"reference,omitempty"`
+	// Other attributes in field metadata that do not affect validation and translation are omitted.
+}
+
+// EntitySchema is the schema of an entity.
+type EntitySchema struct {
+	Fields map[string]EntityField
+}
+
+// ExtractEntityFieldDefinitions extracts the fields in response of retrieving entity schema from Kong gateway
+// and fill the definition of each field in the `Fields` map of returning value.
+func ExtractEntityFieldDefinitions(schema kong.Schema) EntitySchema {
+	retSchema := EntitySchema{
+		Fields: make(map[string]EntityField),
+	}
+
+	fieldList, ok := schema["fields"].([]interface{})
+	if !ok {
+		return retSchema
+	}
+	for _, item := range fieldList {
+		fieldDef, ok := item.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		for fieldName, fieldAttributes := range fieldDef {
+			fieldAttributesMap, ok := fieldAttributes.(map[string]interface{})
+			if !ok {
+				continue
+			}
+
+			fieldTypeStr, _ := fieldAttributesMap["type"].(string)
+			fieldAuto, _ := fieldAttributesMap["auto"].(bool)
+			fieldIsUUID, _ := fieldAttributesMap["uuid"].(bool)
+			fieldRequired, _ := fieldAttributesMap["required"].(bool)
+			fieldReference, _ := fieldAttributesMap["reference"].(string)
+
+			f := EntityField{
+				Name:      fieldName,
+				Type:      EntityFieldType(fieldTypeStr),
+				Auto:      fieldAuto,
+				UUID:      fieldIsUUID,
+				Required:  fieldRequired,
+				Default:   fieldAttributesMap["default"],
+				Reference: fieldReference,
+			}
+			retSchema.Fields[fieldName] = f
+		}
+	}
+	return retSchema
+}
diff --git a/internal/dataplane/kongstate/kongstate.go b/internal/dataplane/kongstate/kongstate.go
index a2b7d5be6..75169adce 100644
--- a/internal/dataplane/kongstate/kongstate.go
+++ b/internal/dataplane/kongstate/kongstate.go
@@ -10,14 +10,18 @@ import (
 	"github.com/kong/go-kong/kong"
 	"github.com/samber/lo"
 	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/admission/validation/consumers/credentials"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/failures"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
 	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
 )
@@ -330,81 +334,147 @@ func (ks *KongState) FillVaults(
 	}
 }
 
-func (ks *KongState) getPluginRelations() map[string]util.ForeignRelations {
+type NamespacedKongPlugin struct {
+	Namespace string
+	Name      string
+}
+
+func (ks *KongState) getPluginRelations(cacheStore store.Storer, log logr.Logger) map[string]util.ForeignRelations {
 	// KongPlugin key (KongPlugin's name:namespace) to corresponding associations
 	pluginRels := map[string]util.ForeignRelations{}
-	addConsumerRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
+
+	type entityRelationType int
+	const (
+		ConsumerRelation      entityRelationType = iota
+		ConsumerGroupRelation entityRelationType = iota
+		RouteRelation         entityRelationType = iota
+		ServiceRelation       entityRelationType = iota
+	)
+	addRelation := func(referer client.Object, plugin annotations.NamespacedKongPlugin, identifier string, t entityRelationType) {
+		// There are 2 types of KongPlugin references: local and remote.
+		// A local reference is one where the KongPlugin is in the same namespace as the referer.
+		// A remote reference is one where the KongPlugin is in a different namespace.
+		// By default a KongPlugin is considered local.
+		// If the plugin has a namespace specified, it is considered remote.
+		//
+		// The referer is the entity that the KongPlugin is associated with.
+		//
+		// Code in buildPlugins() will combine plugin associations into
+		// multi-entity plugins within the local namespace
+		namespace := referer.GetNamespace()
+		if plugin.Namespace != "" {
+			// remote KongPlugin, permitted if ReferenceGrant allows
+			if err := isRemotePluginReferenceAllowed(
+				cacheStore,
+				pluginReference{
+					Referer:   referer,
+					Namespace: plugin.Namespace,
+					Name:      plugin.Name,
+				},
+			); err != nil {
+				log.Error(err, "could not bind requested plugin", "plugin", plugin.Name, "namespace", plugin.Namespace)
+				return
+			}
+
+			namespace = plugin.Namespace
 		}
-		relations.Consumer = append(relations.Consumer, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addConsumerGroupRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
+
+		pluginKey := namespace + ":" + plugin.Name
 		relations, ok := pluginRels[pluginKey]
 		if !ok {
 			relations = util.ForeignRelations{}
 		}
-		relations.ConsumerGroup = append(relations.ConsumerGroup, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addRouteRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
+		switch t {
+		case ConsumerRelation:
+			relations.Consumer = append(relations.Consumer, identifier)
+		case ConsumerGroupRelation:
+			relations.ConsumerGroup = append(relations.ConsumerGroup, identifier)
+		case RouteRelation:
+			relations.Route = append(relations.Route, identifier)
+		case ServiceRelation:
+			relations.Service = append(relations.Service, identifier)
 		}
-		relations.Route = append(relations.Route, identifier)
-		pluginRels[pluginKey] = relations
-	}
-	addServiceRelation := func(namespace, pluginName, identifier string) {
-		pluginKey := namespace + ":" + pluginName
-		relations, ok := pluginRels[pluginKey]
-		if !ok {
-			relations = util.ForeignRelations{}
-		}
-		relations.Service = append(relations.Service, identifier)
 		pluginRels[pluginKey] = relations
 	}
 
 	for i := range ks.Services {
-		// service
 		for _, svc := range ks.Services[i].K8sServices {
-			pluginList := annotations.ExtractKongPluginsFromAnnotations(svc.GetAnnotations())
-			for _, pluginName := range pluginList {
-				addServiceRelation(svc.Namespace, pluginName, *ks.Services[i].Name)
+			pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(svc.GetAnnotations())
+			for _, plugin := range pluginList {
+				addRelation(svc, plugin, *ks.Services[i].Name, ServiceRelation)
 			}
 		}
-		// route
+
 		for j := range ks.Services[i].Routes {
 			ingress := ks.Services[i].Routes[j].Ingress
-			pluginList := annotations.ExtractKongPluginsFromAnnotations(ingress.Annotations)
-			for _, pluginName := range pluginList {
-				addRouteRelation(ingress.Namespace, pluginName, *ks.Services[i].Routes[j].Name)
+			pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(ingress.Annotations)
+			for _, plugin := range pluginList {
+				// pretend we have a full Ingress struct for reference checks
+				virtualIngress := netv1.Ingress{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: ingress.Namespace,
+						Name:      ingress.Name,
+					},
+				}
+				addRelation(&virtualIngress, plugin, *ks.Services[i].Routes[j].Name, RouteRelation)
 			}
 		}
 	}
-	// consumer
+
 	for _, c := range ks.Consumers {
-		pluginList := annotations.ExtractKongPluginsFromAnnotations(c.K8sKongConsumer.GetAnnotations())
-		for _, pluginName := range pluginList {
-			addConsumerRelation(c.K8sKongConsumer.Namespace, pluginName, *c.Username)
+		pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(c.K8sKongConsumer.GetAnnotations())
+		for _, plugin := range pluginList {
+			addRelation(&c.K8sKongConsumer, plugin, *c.Username, ConsumerRelation)
 		}
 	}
-	// consumer group
+
 	for _, cg := range ks.ConsumerGroups {
-		pluginList := annotations.ExtractKongPluginsFromAnnotations(cg.K8sKongConsumerGroup.GetAnnotations())
-		for _, pluginName := range pluginList {
-			addConsumerGroupRelation(cg.K8sKongConsumerGroup.Namespace, pluginName, *cg.Name)
+		pluginList := annotations.ExtractNamespacedKongPluginsFromAnnotations(cg.K8sKongConsumerGroup.GetAnnotations())
+		for _, plugin := range pluginList {
+			addRelation(&cg.K8sKongConsumerGroup, plugin, *cg.Name, ConsumerGroupRelation)
 		}
 	}
 
 	return pluginRels
 }
 
+type pluginReference struct {
+	Referer   client.Object
+	Namespace string
+	Name      string
+}
+
+func isRemotePluginReferenceAllowed(s store.Storer, r pluginReference) error {
+	// remote plugin. considered part of this namespace if a suitable ReferenceGrant exists
+	grants, err := s.ListReferenceGrants()
+	if err != nil {
+		return fmt.Errorf("could not retrieve ReferenceGrants from store when building plugin relations map: %w", err)
+	}
+	allowed := gatewayapi.GetPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
+		Group:     gatewayapi.Group(r.Referer.GetObjectKind().GroupVersionKind().Group),
+		Kind:      gatewayapi.Kind(r.Referer.GetObjectKind().GroupVersionKind().Kind),
+		Namespace: gatewayapi.Namespace(r.Referer.GetNamespace()),
+	}, grants)
+
+	// we don't have a full plugin resource here for the grant checker, so we build a fake one with the correct
+	// name and namespace
+	virtualReference := gatewayapi.PluginLabelReference{
+		Namespace: lo.ToPtr(r.Referer.GetNamespace()),
+		Name:      r.Name,
+	}
+	virtualPlugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: r.Namespace,
+			Name:      r.Name,
+		},
+	}
+	if !gatewayapi.NewRefCheckerForKongPlugin(virtualPlugin, virtualReference).IsRefAllowedByGrant(allowed) {
+		return fmt.Errorf("no grant found for %s in %s to plugin %s in %s requested remote KongPlugin bind",
+			r.Referer.GetObjectKind().GroupVersionKind().Kind, r.Referer.GetNamespace(), r.Name, r.Namespace)
+	}
+	return nil
+}
+
 func buildPlugins(
 	logger logr.Logger,
 	s store.Storer,
@@ -539,7 +609,7 @@ func (ks *KongState) FillPlugins(
 	s store.Storer,
 	failuresCollector *failures.ResourceFailuresCollector,
 ) {
-	ks.Plugins = buildPlugins(log, s, failuresCollector, ks.getPluginRelations())
+	ks.Plugins = buildPlugins(log, s, failuresCollector, ks.getPluginRelations(s, log))
 }
 
 // FillIDs iterates over the KongState and fills in the ID field for each entity
diff --git a/internal/dataplane/translator/backendref.go b/internal/dataplane/translator/backendref.go
index ef26c24d3..781fb9c2a 100644
--- a/internal/dataplane/translator/backendref.go
+++ b/internal/dataplane/translator/backendref.go
@@ -65,7 +65,7 @@ func backendRefsToKongStateBackends(
 		}
 
 		if !util.IsBackendRefGroupKindSupported(backendRef.Group, backendRef.Kind) ||
-			!newRefCheckerForRoute(route, backendRef).IsRefAllowedByGrant(allowed) {
+			!gatewayapi.NewRefCheckerForRoute(route, backendRef).IsRefAllowedByGrant(allowed) {
 			// we log impermissible refs rather than failing the entire rule. while we cannot actually route to
 			// these, we do not want a single impermissible ref to take the entire rule offline. in the case of edits,
 			// failing the entire rule could potentially delete routes that were previously online and in use, and
diff --git a/internal/dataplane/translator/translate_utils.go b/internal/dataplane/translator/translate_utils.go
index c5d8b5f7d..6a5449885 100644
--- a/internal/dataplane/translator/translate_utils.go
+++ b/internal/dataplane/translator/translate_utils.go
@@ -42,10 +42,10 @@ func convertGatewayMatchHeadersToKongRouteMatchHeaders(headers []gatewayapi.HTTP
 	return convertedHeaders, nil
 }
 
-// getPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
+// GetPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
 // from a namespace to a slice of ReferenceGrant Tos. When a To is included in the slice, the key namespace has a
 // ReferenceGrant with those Tos and the input From.
-func getPermittedForReferenceGrantFrom(
+func GetPermittedForReferenceGrantFrom(
 	from gatewayapi.ReferenceGrantFrom,
 	grants []*gatewayapi.ReferenceGrant,
 ) map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo {
@@ -82,7 +82,7 @@ func generateKongServiceFromBackendRefWithName(
 	if err != nil {
 		return kongstate.Service{}, fmt.Errorf("could not retrieve ReferenceGrants for %s: %w", objName, err)
 	}
-	allowed := getPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
+	allowed := GetPermittedForReferenceGrantFrom(gatewayapi.ReferenceGrantFrom{
 		Group:     gatewayapi.Group(route.GetObjectKind().GroupVersionKind().Group),
 		Kind:      gatewayapi.Kind(route.GetObjectKind().GroupVersionKind().Kind),
 		Namespace: gatewayapi.Namespace(route.GetNamespace()),
diff --git a/internal/dataplane/translator/wrappers_refchecker.go b/internal/dataplane/translator/wrappers_refchecker.go
deleted file mode 100644
index 3492d4aff..000000000
--- a/internal/dataplane/translator/wrappers_refchecker.go
+++ /dev/null
@@ -1,95 +0,0 @@
-package translator
-
-import (
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
-)
-
-// refChecker is a wrapper type that facilitates checking whether a backenRef is allowed
-// by a referenceGrantTo set.
-type refChecker[T gatewayapi.BackendRefT] struct {
-	route      client.Object
-	backendRef T
-}
-
-// newRefCheckerForRoute returns a refChecker for the provided route and backendRef.
-func newRefCheckerForRoute[T gatewayapi.BackendRefT](route client.Object, ref T) refChecker[T] {
-	return refChecker[T]{
-		route:      route,
-		backendRef: ref,
-	}
-}
-
-// IsRefAllowedByGrant is a wrapper on top of isRefAllowedByGrant checks if backendRef (that RefChecker
-// holds) is permitted by the provided namespace-indexed ReferenceGrantTo set: allowedRefs.
-// allowedRefs is assumed to contain Tos that only match the backendRef's parent's From, as returned by
-// getPermittedForReferenceGrantFrom.
-func (rc refChecker[T]) IsRefAllowedByGrant(
-	allowedRefs map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo,
-) bool {
-	switch br := (interface{})(rc.backendRef).(type) {
-	case gatewayapi.BackendRef:
-		if br.Namespace == nil {
-			return true
-		}
-
-		// If the namespace is specified but is the same as the route's namespace, then the ref is allowed.
-		if rc.route.GetNamespace() == string(*br.Namespace) {
-			return true
-		}
-
-		return isRefAllowedByGrant(
-			(*string)(br.Namespace),
-			(string)(br.Name),
-			(string)(*br.Group),
-			(string)(*br.Kind),
-			allowedRefs,
-		)
-
-	case gatewayapi.SecretObjectReference:
-		if br.Namespace == nil {
-			return true
-		}
-
-		return isRefAllowedByGrant(
-			(*string)(br.Namespace),
-			(string)(br.Name),
-			(string)(*br.Group),
-			(string)(*br.Kind),
-			allowedRefs,
-		)
-	}
-
-	return false
-}
-
-// isRefAllowedByGrant checks if backendRef is permitted by the provided namespace-indexed ReferenceGrantTo set: allowed.
-// allowed is assumed to contain Tos that only match the backendRef's parent's From, as returned by
-// getPermittedForReferenceGrantFrom.
-func isRefAllowedByGrant(
-	namespace *string,
-	name string,
-	group string,
-	kind string,
-	allowed map[gatewayapi.Namespace][]gatewayapi.ReferenceGrantTo,
-) bool {
-	if namespace == nil {
-		// local references are always fine
-		return true
-	}
-	for _, to := range allowed[gatewayapi.Namespace(*namespace)] {
-		if string(to.Group) == group && string(to.Kind) == kind {
-			if to.Name != nil {
-				if string(*to.Name) == name {
-					return true
-				}
-			} else {
-				// if no referent name specified, matching group/kind is sufficient
-				return true
-			}
-		}
-	}
-
-	return false
-}
diff --git a/internal/gatewayapi/contraints.go b/internal/gatewayapi/contraints.go
index 0e11b3d6a..c52e0f3ed 100644
--- a/internal/gatewayapi/contraints.go
+++ b/internal/gatewayapi/contraints.go
@@ -16,7 +16,20 @@ type RouteT interface {
 		*GRPCRoute
 }
 
+// TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+// This is currently a "BackendRefT" but it's used as a generic target object.
+// Needs to be renamed as such. It should be able to handle anything that satisfies
+// https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/client#Object
+// ReferenceGrants deal with anything that has a Group, Kind, Namespace, and sometimes a Name.
+// client.Object wraps two interfaces that have methods to surface all of those.
+
 type BackendRefT interface {
 	BackendRef |
-		SecretObjectReference
+		SecretObjectReference |
+		PluginLabelReference
+}
+
+type PluginLabelReference struct {
+	Namespace *string
+	Name      string
 }
diff --git a/internal/gatewayapi/references.go b/internal/gatewayapi/references.go
new file mode 100644
index 000000000..c97112739
--- /dev/null
+++ b/internal/gatewayapi/references.go
@@ -0,0 +1,147 @@
+package gatewayapi
+
+import (
+	"reflect"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// RefChecker is a wrapper type that facilitates checking whether a backendRef is allowed
+// by a referenceGrantTo set.
+type RefChecker[T BackendRefT] struct {
+	target     client.Object
+	backendRef T
+}
+
+// NewRefCheckerForRoute returns a RefChecker for the provided route and backendRef.
+func NewRefCheckerForRoute[T BackendRefT](route client.Object, ref T) RefChecker[T] {
+	return RefChecker[T]{
+		target:     route,
+		backendRef: ref,
+	}
+}
+
+func NewRefCheckerForKongPlugin[T BackendRefT](target client.Object, requester T) RefChecker[T] {
+	return RefChecker[T]{
+		target:     target,
+		backendRef: requester,
+	}
+}
+
+// TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000 this has separate cases for different types,
+// but doesn't do anything meaningfully different for them (it only fills in some default info that should be available
+// from the involved objects' methods). We want a generic utility that handles relationship checks for any
+// client.Object. Any behavior particular to specific combinations of GVK->GVK relationships should be handled in the
+// code that implements those relationships.
+
+// IsRefAllowedByGrant is a wrapper on top of isRefAllowedByGrant checks if backendRef (that RefChecker
+// holds) is permitted by the provided namespace-indexed ReferenceGrantTo set: allowedRefs.
+// allowedRefs is assumed to contain Tos that only match the backendRef's parent's From, as returned by
+// GetPermittedForReferenceGrantFrom.
+func (rc RefChecker[T]) IsRefAllowedByGrant(
+	allowedRefs map[Namespace][]ReferenceGrantTo,
+) bool {
+	switch br := (interface{})(rc.backendRef).(type) {
+	case BackendRef:
+		// NOTE https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+		// This is a catch-all that the plugins technically won't need as-is because they have their own
+		// inherent namespace check: if no namespace specified the ref is assumed allowed because it must be local.
+		if br.Namespace == nil {
+			return true
+		}
+
+		// If the namespace is specified but is the same as the route's namespace, then the ref is allowed.
+		if rc.target.GetNamespace() == string(*br.Namespace) {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(*string)(br.Namespace),
+			(string)(br.Name),
+			(string)(*br.Group),
+			(string)(*br.Kind),
+			allowedRefs,
+		)
+
+	case SecretObjectReference:
+		if br.Namespace == nil {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(*string)(br.Namespace),
+			(string)(br.Name),
+			(string)(*br.Group),
+			(string)(*br.Kind),
+			allowedRefs,
+		)
+
+	case PluginLabelReference:
+		if br.Namespace == nil {
+			return true
+		}
+
+		return isRefAllowedByGrant(
+			(br.Namespace),
+			(br.Name),
+			"configuration.konghq.com", // TODO https://github.com/Kong/kubernetes-ingress-controller/issues/6000
+			"KongPlugin",               // TODO These magic strings should become unnecessary once we work with client.Object
+			allowedRefs,
+		)
+	}
+
+	return false
+}
+
+// isRefAllowedByGrant checks if backendRef is permitted by the provided namespace-indexed ReferenceGrantTo set: allowed.
+// allowed is assumed to contain Tos that only match the backendRef's parent's From, as returned by
+// GetPermittedForReferenceGrantFrom.
+func isRefAllowedByGrant(
+	namespace *string,
+	name string,
+	group string,
+	kind string,
+	allowed map[Namespace][]ReferenceGrantTo,
+) bool {
+	if namespace == nil {
+		// local references are always fine
+		return true
+	}
+	for _, to := range allowed[Namespace(*namespace)] {
+		if string(to.Group) == group && string(to.Kind) == kind {
+			if to.Name != nil {
+				if string(*to.Name) == name {
+					return true
+				}
+			} else {
+				// if no referent name specified, matching group/kind is sufficient
+				return true
+			}
+		}
+	}
+
+	return false
+}
+
+// GetPermittedForReferenceGrantFrom takes a ReferenceGrant From (a namespace, group, and kind) and returns a map
+// from a namespace to a slice of ReferenceGrant Tos. When a To is included in the slice, the key namespace has a
+// ReferenceGrant with those Tos and the input From.
+func GetPermittedForReferenceGrantFrom(
+	from ReferenceGrantFrom,
+	grants []*ReferenceGrant,
+) map[Namespace][]ReferenceGrantTo {
+	allowed := make(map[Namespace][]ReferenceGrantTo)
+	// loop over all From values in all grants. if we find a match, add all Tos to the list of Tos allowed for the
+	// grant namespace. this technically could add duplicate copies of the Tos if there are duplicate Froms (it makes
+	// no sense to add them, but it's allowed), but duplicate Tos are harmless (we only care about having at least one
+	// matching To when checking if a ReferenceGrant allows a reference)
+	for _, grant := range grants {
+		for _, otherFrom := range grant.Spec.From {
+			if reflect.DeepEqual(from, otherFrom) {
+				allowed[Namespace(grant.ObjectMeta.Namespace)] = append(allowed[Namespace(grant.ObjectMeta.Namespace)], grant.Spec.To...)
+			}
+		}
+	}
+
+	return allowed
+}
diff --git a/internal/store/cache_key_funcs.go b/internal/store/cache_key_funcs.go
new file mode 100644
index 000000000..387544332
--- /dev/null
+++ b/internal/store/cache_key_funcs.go
@@ -0,0 +1,17 @@
+package store
+
+import "reflect"
+
+// namespacedKeyFunc returns a key for a namespaced object.
+func namespacedKeyFunc(obj interface{}) (string, error) {
+	v := reflect.Indirect(reflect.ValueOf(obj))
+	name := v.FieldByName("Name")
+	namespace := v.FieldByName("Namespace")
+	return namespace.String() + "/" + name.String(), nil
+}
+
+// clusterWideKeyFunc returns a key for a cluster-wide object.
+func clusterWideKeyFunc(obj interface{}) (string, error) {
+	v := reflect.Indirect(reflect.ValueOf(obj))
+	return v.FieldByName("Name").String(), nil
+}
diff --git a/internal/store/cache_stores.go b/internal/store/cache_stores.go
index 79aa7b3db..befa7a4b8 100644
--- a/internal/store/cache_stores.go
+++ b/internal/store/cache_stores.go
@@ -1,95 +1,13 @@
 package store
 
 import (
-	"fmt"
-	"sync"
-
-	corev1 "k8s.io/api/core/v1"
-	discoveryv1 "k8s.io/api/discovery/v1"
-	netv1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured/unstructuredscheme"
 	"k8s.io/apimachinery/pkg/runtime"
 	serializer "k8s.io/apimachinery/pkg/runtime/serializer/json"
 	yamlserializer "k8s.io/apimachinery/pkg/runtime/serializer/yaml"
-	"k8s.io/client-go/tools/cache"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
-	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
-	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
-	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
-	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 )
 
-//go:generate go run ../../hack/generators/cache-stores-snapshot
-
-// CacheStores stores cache.Store for all Kinds of k8s objects that
-// the Ingress Controller reads.
-type CacheStores struct {
-	// Core Kubernetes Stores
-	IngressV1      cache.Store
-	IngressClassV1 cache.Store
-	Service        cache.Store
-	Secret         cache.Store
-	EndpointSlice  cache.Store
-
-	// Gateway API Stores
-	HTTPRoute      cache.Store
-	UDPRoute       cache.Store
-	TCPRoute       cache.Store
-	TLSRoute       cache.Store
-	GRPCRoute      cache.Store
-	ReferenceGrant cache.Store
-	Gateway        cache.Store
-
-	// Kong Stores
-	Plugin                         cache.Store
-	ClusterPlugin                  cache.Store
-	Consumer                       cache.Store
-	ConsumerGroup                  cache.Store
-	KongIngress                    cache.Store
-	TCPIngress                     cache.Store
-	UDPIngress                     cache.Store
-	KongUpstreamPolicy             cache.Store
-	IngressClassParametersV1alpha1 cache.Store
-	KongServiceFacade              cache.Store
-	KongVault                      cache.Store
-
-	l *sync.RWMutex
-}
-
-// NewCacheStores is a convenience function for CacheStores to initialize all attributes with new cache stores.
-func NewCacheStores() CacheStores {
-	return CacheStores{
-		// Core Kubernetes Stores
-		IngressV1:      cache.NewStore(keyFunc),
-		IngressClassV1: cache.NewStore(clusterResourceKeyFunc),
-		Service:        cache.NewStore(keyFunc),
-		Secret:         cache.NewStore(keyFunc),
-		EndpointSlice:  cache.NewStore(keyFunc),
-		// Gateway API Stores
-		HTTPRoute:      cache.NewStore(keyFunc),
-		UDPRoute:       cache.NewStore(keyFunc),
-		TCPRoute:       cache.NewStore(keyFunc),
-		TLSRoute:       cache.NewStore(keyFunc),
-		GRPCRoute:      cache.NewStore(keyFunc),
-		ReferenceGrant: cache.NewStore(keyFunc),
-		Gateway:        cache.NewStore(keyFunc),
-		// Kong Stores
-		Plugin:                         cache.NewStore(keyFunc),
-		ClusterPlugin:                  cache.NewStore(clusterResourceKeyFunc),
-		Consumer:                       cache.NewStore(keyFunc),
-		ConsumerGroup:                  cache.NewStore(keyFunc),
-		KongIngress:                    cache.NewStore(keyFunc),
-		TCPIngress:                     cache.NewStore(keyFunc),
-		UDPIngress:                     cache.NewStore(keyFunc),
-		KongUpstreamPolicy:             cache.NewStore(keyFunc),
-		IngressClassParametersV1alpha1: cache.NewStore(keyFunc),
-		KongServiceFacade:              cache.NewStore(keyFunc),
-		KongVault:                      cache.NewStore(clusterResourceKeyFunc),
-
-		l: &sync.RWMutex{},
-	}
-}
+//go:generate go run ../../hack/generators/cache-stores
 
 // NewCacheStoresFromObjYAML provides a new CacheStores object given any number of byte arrays containing
 // YAML Kubernetes objects. An error is returned if any provided YAML was not a valid Kubernetes object.
@@ -132,202 +50,3 @@ func NewCacheStoresFromObjs(objs ...runtime.Object) (CacheStores, error) {
 	}
 	return c, nil
 }
-
-// Get checks whether or not there's already some version of the provided object present in the cache.
-func (c CacheStores) Get(obj runtime.Object) (item interface{}, exists bool, err error) {
-	c.l.RLock()
-	defer c.l.RUnlock()
-
-	switch obj := obj.(type) {
-	// ----------------------------------------------------------------------------
-	// Kubernetes Core API Support
-	// ----------------------------------------------------------------------------
-	case *netv1.Ingress:
-		return c.IngressV1.Get(obj)
-	case *netv1.IngressClass:
-		return c.IngressClassV1.Get(obj)
-	case *corev1.Service:
-		return c.Service.Get(obj)
-	case *corev1.Secret:
-		return c.Secret.Get(obj)
-	case *discoveryv1.EndpointSlice:
-		return c.EndpointSlice.Get(obj)
-	// ----------------------------------------------------------------------------
-	// Kubernetes Gateway API Support
-	// ----------------------------------------------------------------------------
-	case *gatewayapi.HTTPRoute:
-		return c.HTTPRoute.Get(obj)
-	case *gatewayapi.UDPRoute:
-		return c.UDPRoute.Get(obj)
-	case *gatewayapi.TCPRoute:
-		return c.TCPRoute.Get(obj)
-	case *gatewayapi.TLSRoute:
-		return c.TLSRoute.Get(obj)
-	case *gatewayapi.GRPCRoute:
-		return c.GRPCRoute.Get(obj)
-	case *gatewayapi.ReferenceGrant:
-		return c.ReferenceGrant.Get(obj)
-	case *gatewayapi.Gateway:
-		return c.Gateway.Get(obj)
-	// ----------------------------------------------------------------------------
-	// Kong API Support
-	// ----------------------------------------------------------------------------
-	case *kongv1.KongPlugin:
-		return c.Plugin.Get(obj)
-	case *kongv1.KongClusterPlugin:
-		return c.ClusterPlugin.Get(obj)
-	case *kongv1.KongConsumer:
-		return c.Consumer.Get(obj)
-	case *kongv1beta1.KongConsumerGroup:
-		return c.ConsumerGroup.Get(obj)
-	case *kongv1.KongIngress:
-		return c.KongIngress.Get(obj)
-	case *kongv1beta1.TCPIngress:
-		return c.TCPIngress.Get(obj)
-	case *kongv1beta1.UDPIngress:
-		return c.UDPIngress.Get(obj)
-	case *kongv1beta1.KongUpstreamPolicy:
-		return c.KongUpstreamPolicy.Get(obj)
-	case *kongv1alpha1.IngressClassParameters:
-		return c.IngressClassParametersV1alpha1.Get(obj)
-	case *incubatorv1alpha1.KongServiceFacade:
-		return c.KongServiceFacade.Get(obj)
-	case *kongv1alpha1.KongVault:
-		return c.KongVault.Get(obj)
-	}
-	return nil, false, fmt.Errorf("%T is not a supported cache object type", obj)
-}
-
-// Add stores a provided runtime.Object into the CacheStore if it's of a supported type.
-// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
-func (c CacheStores) Add(obj runtime.Object) error {
-	c.l.Lock()
-	defer c.l.Unlock()
-
-	switch obj := obj.(type) {
-	// ----------------------------------------------------------------------------
-	// Kubernetes Core API Support
-	// ----------------------------------------------------------------------------
-	case *netv1.Ingress:
-		return c.IngressV1.Add(obj)
-	case *netv1.IngressClass:
-		return c.IngressClassV1.Add(obj)
-	case *corev1.Service:
-		return c.Service.Add(obj)
-	case *corev1.Secret:
-		return c.Secret.Add(obj)
-	case *discoveryv1.EndpointSlice:
-		return c.EndpointSlice.Add(obj)
-	// ----------------------------------------------------------------------------
-	// Kubernetes Gateway API Support
-	// ----------------------------------------------------------------------------
-	case *gatewayapi.HTTPRoute:
-		return c.HTTPRoute.Add(obj)
-	case *gatewayapi.UDPRoute:
-		return c.UDPRoute.Add(obj)
-	case *gatewayapi.TCPRoute:
-		return c.TCPRoute.Add(obj)
-	case *gatewayapi.TLSRoute:
-		return c.TLSRoute.Add(obj)
-	case *gatewayapi.GRPCRoute:
-		return c.GRPCRoute.Add(obj)
-	case *gatewayapi.ReferenceGrant:
-		return c.ReferenceGrant.Add(obj)
-	case *gatewayapi.Gateway:
-		return c.Gateway.Add(obj)
-	// ----------------------------------------------------------------------------
-	// Kong API Support
-	// ----------------------------------------------------------------------------
-	case *kongv1.KongPlugin:
-		return c.Plugin.Add(obj)
-	case *kongv1.KongClusterPlugin:
-		return c.ClusterPlugin.Add(obj)
-	case *kongv1.KongConsumer:
-		return c.Consumer.Add(obj)
-	case *kongv1beta1.KongConsumerGroup:
-		return c.ConsumerGroup.Add(obj)
-	case *kongv1.KongIngress:
-		return c.KongIngress.Add(obj)
-	case *kongv1beta1.TCPIngress:
-		return c.TCPIngress.Add(obj)
-	case *kongv1beta1.UDPIngress:
-		return c.UDPIngress.Add(obj)
-	case *kongv1beta1.KongUpstreamPolicy:
-		return c.KongUpstreamPolicy.Add(obj)
-	case *kongv1alpha1.IngressClassParameters:
-		return c.IngressClassParametersV1alpha1.Add(obj)
-	case *incubatorv1alpha1.KongServiceFacade:
-		return c.KongServiceFacade.Add(obj)
-	case *kongv1alpha1.KongVault:
-		return c.KongVault.Add(obj)
-	default:
-		return fmt.Errorf("cannot add unsupported kind %q to the store", obj.GetObjectKind().GroupVersionKind())
-	}
-}
-
-// Delete removes a provided runtime.Object from the CacheStore if it's of a supported type.
-// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
-func (c CacheStores) Delete(obj runtime.Object) error {
-	c.l.Lock()
-	defer c.l.Unlock()
-
-	switch obj := obj.(type) {
-	// ----------------------------------------------------------------------------
-	// Kubernetes Core API Support
-	// ----------------------------------------------------------------------------
-	case *netv1.Ingress:
-		return c.IngressV1.Delete(obj)
-	case *netv1.IngressClass:
-		return c.IngressClassV1.Delete(obj)
-	case *corev1.Service:
-		return c.Service.Delete(obj)
-	case *corev1.Secret:
-		return c.Secret.Delete(obj)
-	case *discoveryv1.EndpointSlice:
-		return c.EndpointSlice.Delete(obj)
-	// ----------------------------------------------------------------------------
-	// Kubernetes Gateway API Support
-	// ----------------------------------------------------------------------------
-	case *gatewayapi.HTTPRoute:
-		return c.HTTPRoute.Delete(obj)
-	case *gatewayapi.UDPRoute:
-		return c.UDPRoute.Delete(obj)
-	case *gatewayapi.TCPRoute:
-		return c.TCPRoute.Delete(obj)
-	case *gatewayapi.TLSRoute:
-		return c.TLSRoute.Delete(obj)
-	case *gatewayapi.GRPCRoute:
-		return c.GRPCRoute.Delete(obj)
-	case *gatewayapi.ReferenceGrant:
-		return c.ReferenceGrant.Delete(obj)
-	case *gatewayapi.Gateway:
-		return c.Gateway.Delete(obj)
-	// ----------------------------------------------------------------------------
-	// Kong API Support
-	// ----------------------------------------------------------------------------
-	case *kongv1.KongPlugin:
-		return c.Plugin.Delete(obj)
-	case *kongv1.KongClusterPlugin:
-		return c.ClusterPlugin.Delete(obj)
-	case *kongv1.KongConsumer:
-		return c.Consumer.Delete(obj)
-	case *kongv1beta1.KongConsumerGroup:
-		return c.ConsumerGroup.Delete(obj)
-	case *kongv1.KongIngress:
-		return c.KongIngress.Delete(obj)
-	case *kongv1beta1.TCPIngress:
-		return c.TCPIngress.Delete(obj)
-	case *kongv1beta1.UDPIngress:
-		return c.UDPIngress.Delete(obj)
-	case *kongv1beta1.KongUpstreamPolicy:
-		return c.KongUpstreamPolicy.Delete(obj)
-	case *kongv1alpha1.IngressClassParameters:
-		return c.IngressClassParametersV1alpha1.Delete(obj)
-	case *incubatorv1alpha1.KongServiceFacade:
-		return c.KongServiceFacade.Delete(obj)
-	case *kongv1alpha1.KongVault:
-		return c.KongVault.Delete(obj)
-	default:
-		return fmt.Errorf("cannot delete unsupported kind %q from the store", obj.GetObjectKind().GroupVersionKind())
-	}
-}
diff --git a/internal/store/cache_stores_snapshot.go b/internal/store/cache_stores_snapshot.go
new file mode 100644
index 000000000..b58263df3
--- /dev/null
+++ b/internal/store/cache_stores_snapshot.go
@@ -0,0 +1,142 @@
+package store
+
+import (
+	"crypto/sha256"
+	"encoding/base32"
+	"fmt"
+	"hash"
+	"slices"
+
+	"github.com/samber/lo"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+)
+
+// TakeSnapshot takes a snapshot of the CacheStores.
+func (c CacheStores) TakeSnapshot() (CacheStores, error) {
+	// Create a fresh CacheStores instance to store the snapshot
+	// in the c.takeSnapshot method. It happens here because it's
+	// not required to be guarded by a lock.
+	snapshot := NewCacheStores()
+	listOfStores := c.ListAllStores()
+
+	c.l.RLock()
+	defer c.l.RUnlock()
+
+	err := takeSnapshot(&snapshot, listOfStores)
+	return snapshot, err
+}
+
+// SnapshotHash is type that represents a hash of the snapshot.
+// It's a base32 encoded string of the sha256 hash of the snapshot.
+// It's printable and human-readable.
+type SnapshotHash string
+
+// SnapshotHashEmpty is a constant that represents an empty snapshot hash.
+const SnapshotHashEmpty SnapshotHash = ""
+
+func newHashCalculator() hashCalculator {
+	return hashCalculator{calc: sha256.New()}
+}
+
+type hashCalculator struct {
+	calc hash.Hash
+}
+
+func (hc hashCalculator) Write(s string) {
+	hc.calc.Write([]byte(s))
+}
+
+func (hc hashCalculator) Get() SnapshotHash {
+	return SnapshotHash(base32.StdEncoding.EncodeToString(hc.calc.Sum(nil)))
+}
+
+// TakeSnapshotIfChanged takes a snapshot of the CacheStores if the hash of the current state
+// differs from the hash of the previous snapshot supplied as an argument (to make and initial
+// just pass empty string). When error is not nil discard all other return values.
+// When newHash is empty it means that the snapshot hasn't been taken - returned snapshot is
+// meaningless. This is a situation when hash of the current state is the same as the hash of
+// the previous snapshot supplied as an argument.
+func (c CacheStores) TakeSnapshotIfChanged(previousSnapshotHash SnapshotHash) (
+	snapshot CacheStores,
+	newHash SnapshotHash,
+	err error,
+) {
+	// Initialize all variables that don't need to be guarded by a lock.
+	snapshot = NewCacheStores()
+	listOfStores := c.ListAllStores()
+	accessor := meta.NewAccessor()
+	hashCalculator := newHashCalculator()
+
+	c.l.RLock()
+	defer c.l.RUnlock()
+
+	// Compute the hash of the current store.
+	for _, store := range listOfStores {
+		// Underlying store is implemented a thread-safe map so for method List() it doesn't maintain order of items.
+		// To successfully calculate hash we need to sort the items.
+		var capturedErr error
+		valuesForHashComputation := lo.Map(store.List(), func(item interface{}, _ int) string {
+			obj, ok := item.(runtime.Object)
+			if !ok {
+				capturedErr = fmt.Errorf("expected runtime.Object, got %T", item)
+				return ""
+			}
+			uid, err := accessor.UID(obj)
+			if err != nil {
+				capturedErr = fmt.Errorf("failed to get UID: %w", err)
+				return ""
+			}
+			resourceVer, err := accessor.ResourceVersion(obj)
+			if err != nil {
+				capturedErr = fmt.Errorf("failed to get ResourceVersion: %w", err)
+				return ""
+			}
+			// UID is unique for each object in Kubernetes and ResourceVersion reflects the version of the object.
+			return string(uid) + resourceVer
+		})
+		if capturedErr != nil {
+			return CacheStores{}, "", capturedErr
+		}
+		// Strings have to be used instead of byte slices, because Cmp.Ordered has to be satisfied.
+		slices.Sort(valuesForHashComputation)
+		for _, v := range valuesForHashComputation {
+			hashCalculator.Write(v)
+		}
+	}
+	// Encode the hash to base32 string to make it human-readable.
+	newHash = hashCalculator.Get()
+
+	// If the hash of the current state is the same as the hash of the previous snapshot, return an empty snapshot.
+	if newHash == previousSnapshotHash {
+		return CacheStores{}, "", nil
+	}
+
+	// Take a snapshot of the current state as the hash of the current state differs from the previous one.
+	if err := takeSnapshot(&snapshot, listOfStores); err != nil {
+		return CacheStores{}, "", fmt.Errorf("failed to take snapshot: %w", err)
+	}
+	return snapshot, newHash, nil
+}
+
+// takeSnapshot iterates over all stores and add a deep copy of each object to the snapshot.
+// It's up to the caller to ensure that the CacheStore from listOfStores has been derived is
+// not modified while the snapshot is being taken, also supplying a pointer to the properly
+// constructed CacheStore as an argument which when error is nil, will contain the snapshot.
+func takeSnapshot(snapshot *CacheStores, listOfStores []cache.Store) error {
+	for _, store := range listOfStores {
+		for _, item := range store.List() {
+			obj, ok := item.(runtime.Object)
+			if !ok {
+				return fmt.Errorf("expected runtime.Object, got %T", item)
+			}
+
+			copiedObj := obj.DeepCopyObject()
+			if err := snapshot.Add(copiedObj); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
diff --git a/internal/store/fake_store.go b/internal/store/fake_store.go
index 0037c36ca..ba65dacfc 100644
--- a/internal/store/fake_store.go
+++ b/internal/store/fake_store.go
@@ -28,18 +28,6 @@ import (
 	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 )
 
-func keyFunc(obj interface{}) (string, error) {
-	v := reflect.Indirect(reflect.ValueOf(obj))
-	name := v.FieldByName("Name")
-	namespace := v.FieldByName("Namespace")
-	return namespace.String() + "/" + name.String(), nil
-}
-
-func clusterResourceKeyFunc(obj interface{}) (string, error) {
-	v := reflect.Indirect(reflect.ValueOf(obj))
-	return v.FieldByName("Name").String(), nil
-}
-
 // FakeObjects can be used to populate a fake Store.
 type FakeObjects struct {
 	IngressesV1                    []*netv1.Ingress
@@ -73,153 +61,153 @@ func NewFakeStore(
 ) (Storer, error) {
 	var s Storer
 
-	ingressV1Store := cache.NewStore(keyFunc)
+	ingressV1Store := cache.NewStore(namespacedKeyFunc)
 	for _, ingress := range objects.IngressesV1 {
 		err := ingressV1Store.Add(ingress)
 		if err != nil {
 			return nil, err
 		}
 	}
-	ingressClassV1Store := cache.NewStore(clusterResourceKeyFunc)
+	ingressClassV1Store := cache.NewStore(clusterWideKeyFunc)
 	for _, ingress := range objects.IngressClassesV1 {
 		err := ingressClassV1Store.Add(ingress)
 		if err != nil {
 			return nil, err
 		}
 	}
-	IngressClassParametersV1alpha1Store := cache.NewStore(clusterResourceKeyFunc)
+	IngressClassParametersV1alpha1Store := cache.NewStore(clusterWideKeyFunc)
 	for _, IngressClassParametersV1alpha1 := range objects.IngressClassParametersV1alpha1 {
 		err := IngressClassParametersV1alpha1Store.Add(IngressClassParametersV1alpha1)
 		if err != nil {
 			return nil, err
 		}
 	}
-	httprouteStore := cache.NewStore(keyFunc)
+	httprouteStore := cache.NewStore(namespacedKeyFunc)
 	for _, httproute := range objects.HTTPRoutes {
 		if err := httprouteStore.Add(httproute); err != nil {
 			return nil, err
 		}
 	}
-	udprouteStore := cache.NewStore(keyFunc)
+	udprouteStore := cache.NewStore(namespacedKeyFunc)
 	for _, udproute := range objects.UDPRoutes {
 		if err := udprouteStore.Add(udproute); err != nil {
 			return nil, err
 		}
 	}
-	tcprouteStore := cache.NewStore(keyFunc)
+	tcprouteStore := cache.NewStore(namespacedKeyFunc)
 	for _, tcproute := range objects.TCPRoutes {
 		if err := tcprouteStore.Add(tcproute); err != nil {
 			return nil, err
 		}
 	}
-	tlsrouteStore := cache.NewStore(keyFunc)
+	tlsrouteStore := cache.NewStore(namespacedKeyFunc)
 	for _, tlsroute := range objects.TLSRoutes {
 		if err := tlsrouteStore.Add(tlsroute); err != nil {
 			return nil, err
 		}
 	}
-	grpcrouteStore := cache.NewStore(keyFunc)
+	grpcrouteStore := cache.NewStore(namespacedKeyFunc)
 	for _, grpcroute := range objects.GRPCRoutes {
 		if err := grpcrouteStore.Add(grpcroute); err != nil {
 			return nil, err
 		}
 	}
-	referencegrantStore := cache.NewStore(keyFunc)
+	referencegrantStore := cache.NewStore(namespacedKeyFunc)
 	for _, referencegrant := range objects.ReferenceGrants {
 		if err := referencegrantStore.Add(referencegrant); err != nil {
 			return nil, err
 		}
 	}
-	gatewayStore := cache.NewStore(keyFunc)
+	gatewayStore := cache.NewStore(namespacedKeyFunc)
 	for _, gw := range objects.Gateways {
 		if err := gatewayStore.Add(gw); err != nil {
 			return nil, err
 		}
 	}
-	tcpIngressStore := cache.NewStore(keyFunc)
+	tcpIngressStore := cache.NewStore(namespacedKeyFunc)
 	for _, ingress := range objects.TCPIngresses {
 		err := tcpIngressStore.Add(ingress)
 		if err != nil {
 			return nil, err
 		}
 	}
-	udpIngressStore := cache.NewStore(keyFunc)
+	udpIngressStore := cache.NewStore(namespacedKeyFunc)
 	for _, ingress := range objects.UDPIngresses {
 		if err := udpIngressStore.Add(ingress); err != nil {
 			return nil, err
 		}
 	}
-	serviceStore := cache.NewStore(keyFunc)
+	serviceStore := cache.NewStore(namespacedKeyFunc)
 	for _, s := range objects.Services {
 		err := serviceStore.Add(s)
 		if err != nil {
 			return nil, err
 		}
 	}
-	secretsStore := cache.NewStore(keyFunc)
+	secretsStore := cache.NewStore(namespacedKeyFunc)
 	for _, s := range objects.Secrets {
 		err := secretsStore.Add(s)
 		if err != nil {
 			return nil, err
 		}
 	}
-	endpointSliceStore := cache.NewStore(keyFunc)
+	endpointSliceStore := cache.NewStore(namespacedKeyFunc)
 	for _, e := range objects.EndpointSlices {
 		err := endpointSliceStore.Add(e)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongIngressStore := cache.NewStore(keyFunc)
+	kongIngressStore := cache.NewStore(namespacedKeyFunc)
 	for _, k := range objects.KongIngresses {
 		err := kongIngressStore.Add(k)
 		if err != nil {
 			return nil, err
 		}
 	}
-	consumerStore := cache.NewStore(keyFunc)
+	consumerStore := cache.NewStore(namespacedKeyFunc)
 	for _, c := range objects.KongConsumers {
 		err := consumerStore.Add(c)
 		if err != nil {
 			return nil, err
 		}
 	}
-	consumerGroupStore := cache.NewStore(keyFunc)
+	consumerGroupStore := cache.NewStore(namespacedKeyFunc)
 	for _, c := range objects.KongConsumerGroups {
 		err := consumerGroupStore.Add(c)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongPluginsStore := cache.NewStore(keyFunc)
+	kongPluginsStore := cache.NewStore(namespacedKeyFunc)
 	for _, p := range objects.KongPlugins {
 		err := kongPluginsStore.Add(p)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongClusterPluginsStore := cache.NewStore(clusterResourceKeyFunc)
+	kongClusterPluginsStore := cache.NewStore(clusterWideKeyFunc)
 	for _, p := range objects.KongClusterPlugins {
 		err := kongClusterPluginsStore.Add(p)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongUpstreamPolicyStore := cache.NewStore(keyFunc)
+	kongUpstreamPolicyStore := cache.NewStore(namespacedKeyFunc)
 	for _, p := range objects.KongUpstreamPolicies {
 		err := kongUpstreamPolicyStore.Add(p)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongServiceFacade := cache.NewStore(keyFunc)
+	kongServiceFacade := cache.NewStore(namespacedKeyFunc)
 	for _, s := range objects.KongServiceFacades {
 		err := kongServiceFacade.Add(s)
 		if err != nil {
 			return nil, err
 		}
 	}
-	kongVaultStore := cache.NewStore(clusterResourceKeyFunc)
+	kongVaultStore := cache.NewStore(clusterWideKeyFunc)
 	for _, v := range objects.KongVaults {
 		err := kongVaultStore.Add(v)
 		if err != nil {
diff --git a/internal/store/store.go b/internal/store/store.go
index 17c237a49..86972f481 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -711,6 +711,8 @@ func mkObjFromGVK(gvk schema.GroupVersionKind) (runtime.Object, error) {
 	// ----------------------------------------------------------------------------
 	case netv1.SchemeGroupVersion.WithKind("Ingress"):
 		return &netv1.Ingress{}, nil
+	case netv1.SchemeGroupVersion.WithKind("IngressClass"):
+		return &netv1.IngressClass{}, nil
 	case corev1.SchemeGroupVersion.WithKind("Service"):
 		return &corev1.Service{}, nil
 	case corev1.SchemeGroupVersion.WithKind("Secret"):
@@ -725,7 +727,7 @@ func mkObjFromGVK(gvk schema.GroupVersionKind) (runtime.Object, error) {
 	// ----------------------------------------------------------------------------
 	case gatewayv1.SchemeGroupVersion.WithKind("HTTPRoute"):
 		return &gatewayapi.HTTPRoute{}, nil
-	case gatewayv1alpha2.SchemeGroupVersion.WithKind("GRPCRoute"):
+	case gatewayv1.SchemeGroupVersion.WithKind("GRPCRoute"):
 		return &gatewayapi.GRPCRoute{}, nil
 	case gatewayv1alpha2.SchemeGroupVersion.WithKind("TCPRoute"):
 		return &gatewayapi.TCPRoute{}, nil
diff --git a/internal/store/zz_generated_cache_stores.go b/internal/store/zz_generated_cache_stores.go
new file mode 100644
index 000000000..2df2e6615
--- /dev/null
+++ b/internal/store/zz_generated_cache_stores.go
@@ -0,0 +1,310 @@
+// Code generated by hack/generators/cache-stores/main.go; DO NOT EDIT.
+// If you want to add a new type to the cache store, you need to add a new entry to the supportedTypes list in spec.go.
+package store
+
+import (
+	"fmt"
+	"sync"
+
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	netv1 "k8s.io/api/networking/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+// CacheStores stores cache.Store for all Kinds of k8s objects that
+// the Ingress Controller reads.
+type CacheStores struct {
+	IngressV1                      cache.Store
+	IngressClassV1                 cache.Store
+	Service                        cache.Store
+	Secret                         cache.Store
+	EndpointSlice                  cache.Store
+	HTTPRoute                      cache.Store
+	UDPRoute                       cache.Store
+	TCPRoute                       cache.Store
+	TLSRoute                       cache.Store
+	GRPCRoute                      cache.Store
+	ReferenceGrant                 cache.Store
+	Gateway                        cache.Store
+	Plugin                         cache.Store
+	ClusterPlugin                  cache.Store
+	Consumer                       cache.Store
+	ConsumerGroup                  cache.Store
+	KongIngress                    cache.Store
+	TCPIngress                     cache.Store
+	UDPIngress                     cache.Store
+	KongUpstreamPolicy             cache.Store
+	IngressClassParametersV1alpha1 cache.Store
+	KongServiceFacade              cache.Store
+	KongVault                      cache.Store
+
+	l *sync.RWMutex
+}
+
+// NewCacheStores is a convenience function for CacheStores to initialize all attributes with new cache stores.
+func NewCacheStores() CacheStores {
+	return CacheStores{
+		IngressV1:                      cache.NewStore(namespacedKeyFunc),
+		IngressClassV1:                 cache.NewStore(clusterWideKeyFunc),
+		Service:                        cache.NewStore(namespacedKeyFunc),
+		Secret:                         cache.NewStore(namespacedKeyFunc),
+		EndpointSlice:                  cache.NewStore(namespacedKeyFunc),
+		HTTPRoute:                      cache.NewStore(namespacedKeyFunc),
+		UDPRoute:                       cache.NewStore(namespacedKeyFunc),
+		TCPRoute:                       cache.NewStore(namespacedKeyFunc),
+		TLSRoute:                       cache.NewStore(namespacedKeyFunc),
+		GRPCRoute:                      cache.NewStore(namespacedKeyFunc),
+		ReferenceGrant:                 cache.NewStore(namespacedKeyFunc),
+		Gateway:                        cache.NewStore(namespacedKeyFunc),
+		Plugin:                         cache.NewStore(namespacedKeyFunc),
+		ClusterPlugin:                  cache.NewStore(clusterWideKeyFunc),
+		Consumer:                       cache.NewStore(namespacedKeyFunc),
+		ConsumerGroup:                  cache.NewStore(namespacedKeyFunc),
+		KongIngress:                    cache.NewStore(namespacedKeyFunc),
+		TCPIngress:                     cache.NewStore(namespacedKeyFunc),
+		UDPIngress:                     cache.NewStore(namespacedKeyFunc),
+		KongUpstreamPolicy:             cache.NewStore(namespacedKeyFunc),
+		IngressClassParametersV1alpha1: cache.NewStore(namespacedKeyFunc),
+		KongServiceFacade:              cache.NewStore(namespacedKeyFunc),
+		KongVault:                      cache.NewStore(clusterWideKeyFunc),
+
+		l: &sync.RWMutex{},
+	}
+}
+
+// Get checks whether or not there's already some version of the provided object present in the cache.
+func (c CacheStores) Get(obj runtime.Object) (item interface{}, exists bool, err error) {
+	c.l.RLock()
+	defer c.l.RUnlock()
+
+	switch obj := obj.(type) {
+	case *netv1.Ingress:
+		return c.IngressV1.Get(obj)
+	case *netv1.IngressClass:
+		return c.IngressClassV1.Get(obj)
+	case *corev1.Service:
+		return c.Service.Get(obj)
+	case *corev1.Secret:
+		return c.Secret.Get(obj)
+	case *discoveryv1.EndpointSlice:
+		return c.EndpointSlice.Get(obj)
+	case *gatewayapi.HTTPRoute:
+		return c.HTTPRoute.Get(obj)
+	case *gatewayapi.UDPRoute:
+		return c.UDPRoute.Get(obj)
+	case *gatewayapi.TCPRoute:
+		return c.TCPRoute.Get(obj)
+	case *gatewayapi.TLSRoute:
+		return c.TLSRoute.Get(obj)
+	case *gatewayapi.GRPCRoute:
+		return c.GRPCRoute.Get(obj)
+	case *gatewayapi.ReferenceGrant:
+		return c.ReferenceGrant.Get(obj)
+	case *gatewayapi.Gateway:
+		return c.Gateway.Get(obj)
+	case *kongv1.KongPlugin:
+		return c.Plugin.Get(obj)
+	case *kongv1.KongClusterPlugin:
+		return c.ClusterPlugin.Get(obj)
+	case *kongv1.KongConsumer:
+		return c.Consumer.Get(obj)
+	case *kongv1beta1.KongConsumerGroup:
+		return c.ConsumerGroup.Get(obj)
+	case *kongv1.KongIngress:
+		return c.KongIngress.Get(obj)
+	case *kongv1beta1.TCPIngress:
+		return c.TCPIngress.Get(obj)
+	case *kongv1beta1.UDPIngress:
+		return c.UDPIngress.Get(obj)
+	case *kongv1beta1.KongUpstreamPolicy:
+		return c.KongUpstreamPolicy.Get(obj)
+	case *kongv1alpha1.IngressClassParameters:
+		return c.IngressClassParametersV1alpha1.Get(obj)
+	case *incubatorv1alpha1.KongServiceFacade:
+		return c.KongServiceFacade.Get(obj)
+	case *kongv1alpha1.KongVault:
+		return c.KongVault.Get(obj)
+	}
+	return nil, false, fmt.Errorf("%T is not a supported cache object type", obj)
+}
+
+// Add stores a provided runtime.Object into the CacheStore if it's of a supported type.
+// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
+func (c CacheStores) Add(obj runtime.Object) error {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	switch obj := obj.(type) {
+	case *netv1.Ingress:
+		return c.IngressV1.Add(obj)
+	case *netv1.IngressClass:
+		return c.IngressClassV1.Add(obj)
+	case *corev1.Service:
+		return c.Service.Add(obj)
+	case *corev1.Secret:
+		return c.Secret.Add(obj)
+	case *discoveryv1.EndpointSlice:
+		return c.EndpointSlice.Add(obj)
+	case *gatewayapi.HTTPRoute:
+		return c.HTTPRoute.Add(obj)
+	case *gatewayapi.UDPRoute:
+		return c.UDPRoute.Add(obj)
+	case *gatewayapi.TCPRoute:
+		return c.TCPRoute.Add(obj)
+	case *gatewayapi.TLSRoute:
+		return c.TLSRoute.Add(obj)
+	case *gatewayapi.GRPCRoute:
+		return c.GRPCRoute.Add(obj)
+	case *gatewayapi.ReferenceGrant:
+		return c.ReferenceGrant.Add(obj)
+	case *gatewayapi.Gateway:
+		return c.Gateway.Add(obj)
+	case *kongv1.KongPlugin:
+		return c.Plugin.Add(obj)
+	case *kongv1.KongClusterPlugin:
+		return c.ClusterPlugin.Add(obj)
+	case *kongv1.KongConsumer:
+		return c.Consumer.Add(obj)
+	case *kongv1beta1.KongConsumerGroup:
+		return c.ConsumerGroup.Add(obj)
+	case *kongv1.KongIngress:
+		return c.KongIngress.Add(obj)
+	case *kongv1beta1.TCPIngress:
+		return c.TCPIngress.Add(obj)
+	case *kongv1beta1.UDPIngress:
+		return c.UDPIngress.Add(obj)
+	case *kongv1beta1.KongUpstreamPolicy:
+		return c.KongUpstreamPolicy.Add(obj)
+	case *kongv1alpha1.IngressClassParameters:
+		return c.IngressClassParametersV1alpha1.Add(obj)
+	case *incubatorv1alpha1.KongServiceFacade:
+		return c.KongServiceFacade.Add(obj)
+	case *kongv1alpha1.KongVault:
+		return c.KongVault.Add(obj)
+	}
+	return fmt.Errorf("cannot add unsupported kind %q to the store", obj.GetObjectKind().GroupVersionKind())
+}
+
+// Delete removes a provided runtime.Object from the CacheStore if it's of a supported type.
+// The CacheStore must be initialized (see NewCacheStores()) or this will panic.
+func (c CacheStores) Delete(obj runtime.Object) error {
+	c.l.Lock()
+	defer c.l.Unlock()
+
+	switch obj := obj.(type) {
+	case *netv1.Ingress:
+		return c.IngressV1.Delete(obj)
+	case *netv1.IngressClass:
+		return c.IngressClassV1.Delete(obj)
+	case *corev1.Service:
+		return c.Service.Delete(obj)
+	case *corev1.Secret:
+		return c.Secret.Delete(obj)
+	case *discoveryv1.EndpointSlice:
+		return c.EndpointSlice.Delete(obj)
+	case *gatewayapi.HTTPRoute:
+		return c.HTTPRoute.Delete(obj)
+	case *gatewayapi.UDPRoute:
+		return c.UDPRoute.Delete(obj)
+	case *gatewayapi.TCPRoute:
+		return c.TCPRoute.Delete(obj)
+	case *gatewayapi.TLSRoute:
+		return c.TLSRoute.Delete(obj)
+	case *gatewayapi.GRPCRoute:
+		return c.GRPCRoute.Delete(obj)
+	case *gatewayapi.ReferenceGrant:
+		return c.ReferenceGrant.Delete(obj)
+	case *gatewayapi.Gateway:
+		return c.Gateway.Delete(obj)
+	case *kongv1.KongPlugin:
+		return c.Plugin.Delete(obj)
+	case *kongv1.KongClusterPlugin:
+		return c.ClusterPlugin.Delete(obj)
+	case *kongv1.KongConsumer:
+		return c.Consumer.Delete(obj)
+	case *kongv1beta1.KongConsumerGroup:
+		return c.ConsumerGroup.Delete(obj)
+	case *kongv1.KongIngress:
+		return c.KongIngress.Delete(obj)
+	case *kongv1beta1.TCPIngress:
+		return c.TCPIngress.Delete(obj)
+	case *kongv1beta1.UDPIngress:
+		return c.UDPIngress.Delete(obj)
+	case *kongv1beta1.KongUpstreamPolicy:
+		return c.KongUpstreamPolicy.Delete(obj)
+	case *kongv1alpha1.IngressClassParameters:
+		return c.IngressClassParametersV1alpha1.Delete(obj)
+	case *incubatorv1alpha1.KongServiceFacade:
+		return c.KongServiceFacade.Delete(obj)
+	case *kongv1alpha1.KongVault:
+		return c.KongVault.Delete(obj)
+	}
+	return fmt.Errorf("cannot delete unsupported kind %q from the store", obj.GetObjectKind().GroupVersionKind())
+}
+
+// ListAllStores returns a list of all cache stores embedded in the struct.
+func (c CacheStores) ListAllStores() []cache.Store {
+	return []cache.Store{
+		c.IngressV1,
+		c.IngressClassV1,
+		c.Service,
+		c.Secret,
+		c.EndpointSlice,
+		c.HTTPRoute,
+		c.UDPRoute,
+		c.TCPRoute,
+		c.TLSRoute,
+		c.GRPCRoute,
+		c.ReferenceGrant,
+		c.Gateway,
+		c.Plugin,
+		c.ClusterPlugin,
+		c.Consumer,
+		c.ConsumerGroup,
+		c.KongIngress,
+		c.TCPIngress,
+		c.UDPIngress,
+		c.KongUpstreamPolicy,
+		c.IngressClassParametersV1alpha1,
+		c.KongServiceFacade,
+		c.KongVault,
+	}
+}
+
+// SupportedTypes returns a list of supported types for the cache.
+func (c CacheStores) SupportedTypes() []client.Object {
+	return []client.Object{
+		&netv1.Ingress{},
+		&netv1.IngressClass{},
+		&corev1.Service{},
+		&corev1.Secret{},
+		&discoveryv1.EndpointSlice{},
+		&gatewayapi.HTTPRoute{},
+		&gatewayapi.UDPRoute{},
+		&gatewayapi.TCPRoute{},
+		&gatewayapi.TLSRoute{},
+		&gatewayapi.GRPCRoute{},
+		&gatewayapi.ReferenceGrant{},
+		&gatewayapi.Gateway{},
+		&kongv1.KongPlugin{},
+		&kongv1.KongClusterPlugin{},
+		&kongv1.KongConsumer{},
+		&kongv1beta1.KongConsumerGroup{},
+		&kongv1.KongIngress{},
+		&kongv1beta1.TCPIngress{},
+		&kongv1beta1.UDPIngress{},
+		&kongv1beta1.KongUpstreamPolicy{},
+		&kongv1alpha1.IngressClassParameters{},
+		&incubatorv1alpha1.KongServiceFacade{},
+		&kongv1alpha1.KongVault{},
+	}
+}
diff --git a/internal/store/zz_generated_cache_stores_snapshot.go b/internal/store/zz_generated_cache_stores_snapshot.go
deleted file mode 100644
index 300f4f22a..000000000
--- a/internal/store/zz_generated_cache_stores_snapshot.go
+++ /dev/null
@@ -1,61 +0,0 @@
-// Code generated by hack/generators/cache/snapshot.go; DO NOT EDIT.
-package store
-
-import (
-	"fmt"
-
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/tools/cache"
-)
-
-func (c CacheStores) TakeSnapshot() (CacheStores, error) {
-	// Create a fresh CacheStores instance to store the snapshot.
-	snapshot := NewCacheStores()
-
-	// Gather cache store fields for all kinds.
-	allStores := []cache.Store{
-		c.IngressV1,
-		c.IngressClassV1,
-		c.Service,
-		c.Secret,
-		c.EndpointSlice,
-		c.HTTPRoute,
-		c.UDPRoute,
-		c.TCPRoute,
-		c.TLSRoute,
-		c.GRPCRoute,
-		c.ReferenceGrant,
-		c.Gateway,
-		c.Plugin,
-		c.ClusterPlugin,
-		c.Consumer,
-		c.ConsumerGroup,
-		c.KongIngress,
-		c.TCPIngress,
-		c.UDPIngress,
-		c.KongUpstreamPolicy,
-		c.IngressClassParametersV1alpha1,
-		c.KongServiceFacade,
-		c.KongVault,
-	}
-
-	c.l.RLock()
-	defer c.l.RUnlock()
-
-	// Iterate over all stores and add a deep copy of each object to the snapshot.
-	for _, store := range allStores {
-		for _, item := range store.List() {
-			obj, ok := item.(runtime.Object)
-			if !ok {
-				return CacheStores{}, fmt.Errorf("expected runtime.Object, got %T", item)
-			}
-
-			copiedObj := obj.DeepCopyObject()
-			if err := snapshot.Add(copiedObj); err != nil {
-				return CacheStores{}, err
-			}
-		}
-	}
-
-	return snapshot, nil
-}


Test patch:
diff --git a/.github/test_dependencies.yaml b/.github/test_dependencies.yaml
index 6df6298c7..1465228a6 100644
--- a/.github/test_dependencies.yaml
+++ b/.github/test_dependencies.yaml
@@ -1,13 +1,13 @@
 e2e:
   kind:
     # renovate: datasource=docker depName=kindest/node versioning=docker
-    - 'v1.29.2'
+    - 'v1.30.0'
     # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-    - 'v1.28.7'
+    - 'v1.28.9'
     # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-    - 'v1.27.11'
+    - 'v1.27.13'
     # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-    - 'v1.26.14'
+    - 'v1.26.15'
     # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
     - 'v1.25.16'
   gke:
@@ -18,19 +18,19 @@ e2e:
   # used directly in the test matrix `include` section.
   istio:
     - # renovate: datasource=docker depName=kindest/node versioning=docker
-      kind: 'v1.29.2'
+      kind: 'v1.30.0'
       # renovate: datasource=docker depName=istio/istioctl versioning=docker
       istio: '1.21.2'
     - # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-      kind: 'v1.28.7'
+      kind: 'v1.28.9'
       # renovate: datasource=docker depName=istio/istioctl@only-patch packageName=istio/istioctl versioning=docker
       istio: '1.19.10'
     - # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-      kind: 'v1.27.11'
+      kind: 'v1.27.13'
       # renovate: datasource=docker depName=istio/istioctl@only-patch packageName=istio/istioctl versioning=docker
       istio: '1.18.7'
     - # renovate: datasource=docker depName=kindest/node@only-patch packageName=kindest/node versioning=docker
-      kind: 'v1.26.14'
+      kind: 'v1.26.15'
       # renovate: datasource=docker depName=istio/istioctl@only-patch packageName=istio/istioctl versioning=docker
       istio: '1.17.8'
 
@@ -42,11 +42,11 @@ integration:
     # renovate: datasource=helm depName=kong registryUrl=https://charts.konghq.com versioning=helm
     kong: '2.38.0'
   # renovate: datasource=docker depName=kindest/node versioning=docker
-  kind: 'v1.29.2'
+  kind: 'v1.30.0'
   # renovate: datasource=docker depName=kong versioning=docker
   kong-oss: '3.6.1'
   # renovate: datasource=docker depName=kong/kong-gateway versioning=docker
-  kong-ee: '3.6.1.3'
+  kong-ee: '3.6.1.4'
 
 kongintegration:
   # renovate: datasource=docker depName=kong versioning=docker
diff --git a/.github/workflows/_envtest_tests.yaml b/.github/workflows/_envtest_tests.yaml
index 49a76b59c..882522039 100644
--- a/.github/workflows/_envtest_tests.yaml
+++ b/.github/workflows/_envtest_tests.yaml
@@ -20,10 +20,17 @@ jobs:
         with:
           install: false
 
+      - uses: Kong/kong-license@master
+        id: license
+        with:
+          op-token: ${{ secrets.OP_SERVICE_ACCOUNT_TOKEN }}
+
       - name: run envtest tests
         run: make test.envtest.pretty
         env:
           GOTESTSUM_JUNITFILE: envtest-tests.xml
+          TEST_KONG_ENTERPRISE: "true"
+          KONG_LICENSE_DATA: ${{ steps.license.outputs.license }}
 
       - name: collect test coverage
         uses: actions/upload-artifact@v4
diff --git a/internal/admission/server_test.go b/internal/admission/server_test.go
index 01afbf419..076e3c27b 100644
--- a/internal/admission/server_test.go
+++ b/internal/admission/server_test.go
@@ -82,6 +82,10 @@ func (v KongFakeValidator) ValidateVault(_ context.Context, _ kongv1alpha1.KongV
 	return v.Result, v.Message, v.Error
 }
 
+func (v KongFakeValidator) ValidateCustomEntity(_ context.Context, _ kongv1alpha1.KongCustomEntity) (bool, string, error) {
+	return v.Result, v.Message, v.Error
+}
+
 func (v KongFakeValidator) ValidateService(_ context.Context, _ corev1.Service) (bool, string, error) {
 	return v.Result, v.Message, v.Error
 }
diff --git a/internal/admission/validator_test.go b/internal/admission/validator_test.go
index 95aac2362..6a5d83c1d 100644
--- a/internal/admission/validator_test.go
+++ b/internal/admission/validator_test.go
@@ -2,6 +2,7 @@ package admission
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"testing"
@@ -23,6 +24,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/translator"
 	managerscheme "github.com/kong/kubernetes-ingress-controller/v3/internal/manager/scheme"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
@@ -64,6 +66,7 @@ type fakeServicesProvider struct {
 	infoSvc          kong.AbstractInfoService
 	routeSvc         kong.AbstractRouteService
 	vaultSvc         kong.AbstractVaultService
+	schemaSvc        kong.AbstractSchemaService
 }
 
 func (f fakeServicesProvider) GetConsumersService() (kong.AbstractConsumerService, bool) {
@@ -108,6 +111,13 @@ func (f fakeServicesProvider) GetVaultsService() (kong.AbstractVaultService, boo
 	return nil, false
 }
 
+func (f fakeServicesProvider) GetSchemasService() (kong.AbstractSchemaService, bool) {
+	if f.schemaSvc != nil {
+		return f.schemaSvc, true
+	}
+	return nil, false
+}
+
 func TestKongHTTPValidator_ValidatePlugin(t *testing.T) {
 	store, _ := store.NewFakeStore(store.FakeObjects{
 		Secrets: []*corev1.Secret{
@@ -1413,9 +1423,159 @@ type fakeVaultSvc struct {
 	shouldFail bool
 }
 
-func (s fakeVaultSvc) Validate(context.Context, *kong.Vault) (bool, string, error) {
+func (s fakeVaultSvc) Validate(_ context.Context, _ *kong.Vault) (bool, string, error) {
 	if s.shouldFail {
 		return false, "something is wrong with the vault", nil
 	}
 	return true, "", nil
 }
+
+func TestValidator_ValidateCustomEntity(t *testing.T) {
+	testCases := []struct {
+		name            string
+		entity          kongv1alpha1.KongCustomEntity
+		fields          map[string]kongstate.EntityField
+		entityTypeExist bool
+		validateSvcFail bool
+		expectedOK      bool
+		expectedMessage string
+	}{
+		{
+			name: "entity with non-exist type should fail the validation",
+			entity: kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "entity_non_exist",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "non_exist",
+					Fields:     apiextensionsv1.JSON{Raw: []byte("{}")},
+				},
+			},
+			entityTypeExist: false,
+			expectedOK:      false,
+			expectedMessage: "entity type does not exist",
+		},
+		{
+			name: "valid entity",
+			entity: kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "valid_entity",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "entity_type_ok",
+					Fields:     apiextensionsv1.JSON{Raw: []byte(`{"foo":"bar"}`)},
+				},
+			},
+			fields: map[string]kongstate.EntityField{
+				"foo": {
+					Name: "foo",
+					Type: kongstate.EntityFieldTypeString,
+				},
+			},
+			entityTypeExist: true,
+			expectedOK:      true,
+		},
+		{
+			name: "valid entity with foreign key",
+			entity: kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "entity_with_foreign",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "entity_type_with_foreign",
+					Fields:     apiextensionsv1.JSON{Raw: []byte(`{"foo":"bar"}`)},
+				},
+			},
+			fields: map[string]kongstate.EntityField{
+				"foo": {
+					Name: "foo",
+					Type: kongstate.EntityFieldTypeString,
+				},
+				"baz": {
+					Name:      "baz",
+					Type:      kongstate.EntityFieldTypeForeign,
+					Required:  true,
+					Reference: "service",
+				},
+			},
+			entityTypeExist: true,
+			expectedOK:      true,
+		},
+		{
+			name: "validate service fail",
+			entity: kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "validate_service_fail",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "entity_type_ok",
+					Fields:     apiextensionsv1.JSON{Raw: []byte(`{"foo":"bar"}`)},
+				},
+			},
+			fields: map[string]kongstate.EntityField{
+				"foo": {
+					Name: "foo",
+					Type: kongstate.EntityFieldTypeString,
+				},
+			},
+			entityTypeExist: true,
+			validateSvcFail: true,
+			expectedOK:      false,
+			expectedMessage: "something is wrong in the entity",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fieldList := make([]interface{}, 0, len(tc.fields))
+			for _, field := range tc.fields {
+				fieldList = append(fieldList, map[string]interface{}{
+					field.Name: map[string]interface{}{
+						"type":      string(field.Type),
+						"required":  field.Required,
+						"reference": field.Reference,
+					},
+				})
+			}
+
+			validator := KongHTTPValidator{
+				AdminAPIServicesProvider: fakeServicesProvider{
+					schemaSvc: fakeSchemaSvc{
+						entityTypeExist: tc.entityTypeExist,
+						schema: map[string]interface{}{
+							"fields": fieldList,
+						},
+						shouldFail: tc.validateSvcFail,
+					},
+				},
+				ingressClassMatcher: fakeClassMatcher,
+			}
+			ok, msg, err := validator.ValidateCustomEntity(context.Background(), tc.entity)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedOK, ok)
+			assert.Contains(t, msg, tc.expectedMessage)
+		})
+	}
+}
+
+type fakeSchemaSvc struct {
+	schema          map[string]interface{}
+	entityTypeExist bool
+	shouldFail      bool
+}
+
+var _ kong.AbstractSchemaService = fakeSchemaSvc{}
+
+func (s fakeSchemaSvc) Get(_ context.Context, _ string) (kong.Schema, error) {
+	if !s.entityTypeExist {
+		return nil, errors.New("entity type does not exist")
+	}
+	return s.schema, nil
+}
+
+func (s fakeSchemaSvc) Validate(_ context.Context, _ kong.EntityType, _ interface{}) (bool, string, error) {
+	if s.shouldFail {
+		return false, "something is wrong in the entity", nil
+	}
+	return true, "", nil
+}
diff --git a/internal/annotations/annotations_test.go b/internal/annotations/annotations_test.go
index f283c7a48..bb7fdbec2 100644
--- a/internal/annotations/annotations_test.go
+++ b/internal/annotations/annotations_test.go
@@ -157,6 +157,52 @@ func TestExtractKongPluginsFromAnnotations(t *testing.T) {
 	}
 }
 
+func TestExtractNamespacedKongPluginsFromAnnotations(t *testing.T) {
+	type args struct {
+		anns map[string]string
+	}
+	tests := []struct {
+		name string
+		args args
+		want []NamespacedKongPlugin
+	}{
+		{
+			name: "empty",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/nothing": "whatever",
+				},
+			},
+			want: nil,
+		},
+		{
+			name: "basic",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/plugins": "kp-rl, kp-cors",
+				},
+			},
+			want: []NamespacedKongPlugin{{Name: "kp-rl"}, {Name: "kp-cors"}},
+		},
+		{
+			name: "mixed",
+			args: args{
+				anns: map[string]string{
+					"konghq.com/plugins": "default:kp-rl, kp-cors",
+				},
+			},
+			want: []NamespacedKongPlugin{{Name: "kp-rl", Namespace: "default"}, {Name: "kp-cors"}},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := ExtractNamespacedKongPluginsFromAnnotations(tt.args.anns); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ExtractNamespacedKongPluginsFromAnnotations() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestExtractConfigurationName(t *testing.T) {
 	type args struct {
 		anns map[string]string
diff --git a/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go b/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go
new file mode 100644
index 000000000..9b1a6facc
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_gatewayapi_test.go
@@ -0,0 +1,460 @@
+package fallback_test
+
+import (
+	"testing"
+
+	"github.com/samber/lo"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+)
+
+func TestResolveDependencies_HTTPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "HTTPRoute -> Service",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.HTTPRouteSpec{
+					Rules: []gatewayapi.HTTPRouteRule{
+						{
+							BackendRefs: []gatewayapi.HTTPBackendRef{
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "2",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "HTTPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.HTTPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_TLSRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "TLSRoute -> Service",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.TLSRouteSpec{
+					Rules: []gatewayapi.TLSRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "TLSRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.TLSRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_TCPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "TCPRoute -> Service",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.TCPRouteSpec{
+					Rules: []gatewayapi.TCPRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "TCPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.TCPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_UDPRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "UDPRoute -> Service",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.UDPRouteSpec{
+					Rules: []gatewayapi.UDPRouteRule{
+						{
+							BackendRefs: []gatewayapi.BackendRef{
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "1",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+								{
+									BackendObjectReference: gatewayapi.BackendObjectReference{
+										Name: "2",
+										Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "UDPRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.UDPRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
+
+func TestResolveDependencies_GRPCRoute(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "GRPCRoute -> Service",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+				},
+				Spec: gatewayapi.GRPCRouteSpec{
+					Rules: []gatewayapi.GRPCRouteRule{
+						{
+							BackendRefs: []gatewayapi.GRPCBackendRef{
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+								{
+									BackendRef: gatewayapi.BackendRef{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "2",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "GRPCRoute -> KongPlugin, KongClusterPlugin",
+			object: &gatewayapi.GRPCRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-route",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_ingress_test.go b/internal/dataplane/fallback/graph_dependencies_ingress_test.go
new file mode 100644
index 000000000..0e952a8ef
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_ingress_test.go
@@ -0,0 +1,244 @@
+package fallback_test
+
+import (
+	"testing"
+
+	"github.com/samber/lo"
+	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestResolveDependencies_Ingress(t *testing.T) {
+	testCases := []resolveDependenciesTestCase{
+		{
+			name: "no dependencies",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testIngressClass(t, "1"),
+				testService(t, "1"),
+				testKongServiceFacade(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "1"),
+			),
+			expected: []client.Object{},
+		},
+		{
+			name: "Ingress -> IngressClass - annotation",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.IngressClassKey: "1",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testIngressClass(t, "1"),
+				testIngressClass(t, "2"),
+			),
+			expected: []client.Object{testIngressClass(t, "1")},
+		},
+		{
+			name: "Ingress -> IngressClass - field",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+				},
+				Spec: netv1.IngressSpec{
+					IngressClassName: lo.ToPtr("1"),
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testIngressClass(t, "1"),
+				testIngressClass(t, "2"),
+			),
+			expected: []client.Object{testIngressClass(t, "1")},
+		},
+		{
+			name: "Ingress -> Service",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+				},
+				Spec: netv1.IngressSpec{
+					Rules: []netv1.IngressRule{
+						{
+							IngressRuleValue: netv1.IngressRuleValue{
+								HTTP: &netv1.HTTPIngressRuleValue{
+									Paths: []netv1.HTTPIngressPath{
+										{
+											Backend: netv1.IngressBackend{
+												Service: &netv1.IngressServiceBackend{
+													Name: "1",
+												},
+											},
+										},
+										{
+											Backend: netv1.IngressBackend{
+												Service: &netv1.IngressServiceBackend{
+													Name: "2",
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testService(t, "1"),
+				testService(t, "2"),
+				testService(t, "3"),
+			),
+			expected: []client.Object{
+				testService(t, "1"),
+				testService(t, "2"),
+			},
+		},
+		{
+			name: "Ingress -> KongServiceFacade",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+				},
+				Spec: netv1.IngressSpec{
+					Rules: []netv1.IngressRule{
+						{
+							IngressRuleValue: netv1.IngressRuleValue{
+								HTTP: &netv1.HTTPIngressRuleValue{
+									Paths: []netv1.HTTPIngressPath{
+										{
+											Backend: netv1.IngressBackend{
+												Resource: &corev1.TypedLocalObjectReference{
+													Name:     "1",
+													Kind:     "KongServiceFacade",
+													APIGroup: lo.ToPtr(incubatorv1alpha1.SchemeGroupVersion.Group),
+												},
+											},
+										},
+										{
+											Backend: netv1.IngressBackend{
+												Resource: &corev1.TypedLocalObjectReference{
+													Name:     "2",
+													Kind:     "KongServiceFacade",
+													APIGroup: lo.ToPtr(incubatorv1alpha1.SchemeGroupVersion.Group),
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongServiceFacade(t, "1"),
+				testKongServiceFacade(t, "2"),
+				testKongServiceFacade(t, "3"),
+			),
+			expected: []client.Object{
+				testKongServiceFacade(t, "1"),
+				testKongServiceFacade(t, "2"),
+			},
+		},
+		{
+			name: "Ingress -> KongPlugin, KongClusterPlugin",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,2,cluster-1,cluster-2",
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongPlugin(t, "3"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+				testKongClusterPlugin(t, "cluster-3"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongPlugin(t, "2"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testKongClusterPlugin(t, "cluster-2"),
+			},
+		},
+		{
+			name: "Ingress -> all dependencies at once",
+			object: &netv1.Ingress{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-ingress",
+					Namespace: "test-namespace",
+					Annotations: map[string]string{
+						annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+					},
+				},
+				Spec: netv1.IngressSpec{
+					Rules: []netv1.IngressRule{
+						{
+							IngressRuleValue: netv1.IngressRuleValue{
+								HTTP: &netv1.HTTPIngressRuleValue{
+									Paths: []netv1.HTTPIngressPath{
+										{
+											Backend: netv1.IngressBackend{
+												Resource: &corev1.TypedLocalObjectReference{
+													Name:     "1",
+													Kind:     "KongServiceFacade",
+													APIGroup: lo.ToPtr(incubatorv1alpha1.SchemeGroupVersion.Group),
+												},
+											},
+										},
+										{
+											Backend: netv1.IngressBackend{
+												Service: &netv1.IngressServiceBackend{
+													Name: "1",
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			cache: cacheStoresFromObjs(t,
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testService(t, "1"),
+				testKongServiceFacade(t, "1"),
+			),
+			expected: []client.Object{
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+				testService(t, "1"),
+				testKongServiceFacade(t, "1"),
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		runResolveDependenciesTest(t, tc)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_dependencies_test.go b/internal/dataplane/fallback/graph_dependencies_test.go
new file mode 100644
index 000000000..84b65eb0a
--- /dev/null
+++ b/internal/dataplane/fallback/graph_dependencies_test.go
@@ -0,0 +1,62 @@
+package fallback_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	"github.com/kong/kubernetes-ingress-controller/v3/test/helpers"
+)
+
+// resolveDependenciesTestCase is a test case for the ResolveDependencies function.
+type resolveDependenciesTestCase struct {
+	name     string
+	object   client.Object
+	cache    store.CacheStores
+	expected []client.Object
+}
+
+func runResolveDependenciesTest(t *testing.T, tc resolveDependenciesTestCase) {
+	t.Run(tc.name, func(t *testing.T) {
+		t.Run("when cache empty, does not panic and gives no dependencies", func(t *testing.T) {
+			require.NotPanics(t, func() {
+				dependencies, err := fallback.ResolveDependencies(store.NewCacheStores(), tc.object)
+				require.NoError(t, err)
+				require.Empty(t, dependencies, "expect no dependencies found in an empty cache")
+			})
+		})
+		t.Run("when cache has objects, resolves dependencies as expected", func(t *testing.T) {
+			dependencies, err := fallback.ResolveDependencies(tc.cache, tc.object)
+			require.NoError(t, err)
+			require.ElementsMatch(t, tc.expected, dependencies)
+		})
+	})
+}
+
+// cacheStoresFromObjs creates a CacheStores from the given objects.
+// It assigns each object a type meta using the scheme.
+func cacheStoresFromObjs(t *testing.T, objs ...runtime.Object) store.CacheStores {
+	for i := range objs {
+		obj := objs[i].(client.Object)
+		obj = helpers.WithTypeMeta(t, obj)
+		objs[i] = obj
+	}
+	s, err := store.NewCacheStoresFromObjs(objs...)
+	require.NoError(t, err)
+	return s
+}
+
+func TestResolveDependencies_ImplementedForAllSupportedCacheStoresTypes(t *testing.T) {
+	cache := store.NewCacheStores()
+	for _, supportedType := range cache.SupportedTypes() {
+		_, err := fallback.ResolveDependencies(cache, supportedType)
+		assert.NoErrorf(t, err, "ResolveDependencies not implemented for %T. If your object type does not have any dependencies,"+
+			"please add it to the ResolveDependencies' switch-case with object types having no dependencies. Otherwise, please add a new"+
+			"switch-case for your type and implement resolve{YourType}Dependencies function.", supportedType)
+	}
+}
diff --git a/internal/dataplane/fallback/graph_test.go b/internal/dataplane/fallback/graph_test.go
new file mode 100644
index 000000000..9d283147d
--- /dev/null
+++ b/internal/dataplane/fallback/graph_test.go
@@ -0,0 +1,344 @@
+package fallback_test
+
+import (
+	"sort"
+	"testing"
+
+	"github.com/samber/lo"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/fallback"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestNewConfigGraphFromCacheStores(t *testing.T) {
+	// adjacencyGraphStrings returns a map of stringified vertices and their neighbours
+	// in the given graph for easy comparison.
+	adjacencyGraphStrings := func(t *testing.T, g *fallback.ConfigGraph) map[string][]string {
+		am, err := g.AdjacencyMap()
+		require.NoError(t, err)
+		adjacencyMapStrings := make(map[string][]string, len(am))
+		for v, neighbours := range am {
+			neighboursStrings := lo.Map(neighbours, func(n fallback.ObjectHash, _ int) string {
+				return n.String()
+			})
+			sort.Strings(neighboursStrings) // Sort for deterministic output.
+			adjacencyMapStrings[v.String()] = neighboursStrings
+		}
+		return adjacencyMapStrings
+	}
+
+	testCases := []struct {
+		name                 string
+		cache                store.CacheStores
+		expectedAdjacencyMap map[string][]string
+	}{
+		{
+			name:                 "empty cache",
+			cache:                store.NewCacheStores(),
+			expectedAdjacencyMap: map[string][]string{},
+		},
+		{
+			name: "cache with Ingress and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&netv1.Ingress{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-ingress",
+						Namespace: "test-namespace",
+					},
+					Spec: netv1.IngressSpec{
+						IngressClassName: lo.ToPtr("test-ingress-class"),
+						Rules: []netv1.IngressRule{
+							{
+								IngressRuleValue: netv1.IngressRuleValue{
+									HTTP: &netv1.HTTPIngressRuleValue{
+										Paths: []netv1.HTTPIngressPath{
+											{
+												Backend: netv1.IngressBackend{
+													Service: &netv1.IngressServiceBackend{
+														Name: "test-service",
+													},
+												},
+											},
+											{
+												Backend: netv1.IngressBackend{
+													Resource: &corev1.TypedLocalObjectReference{
+														Name:     "test-kong-service-facade",
+														Kind:     "KongServiceFacade",
+														APIGroup: lo.ToPtr(incubatorv1alpha1.GroupVersion.Group),
+													},
+												},
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				&netv1.IngressClass{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "test-ingress-class",
+					},
+				},
+				&corev1.Service{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-service",
+						Namespace: "test-namespace",
+					},
+				},
+				&incubatorv1alpha1.KongServiceFacade{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-kong-service-facade",
+						Namespace: "test-namespace",
+					},
+				},
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"Ingress:test-namespace/test-ingress": {},
+				"IngressClass:test-ingress-class": {
+					"Ingress:test-namespace/test-ingress",
+				},
+				"Service:test-namespace/test-service": {
+					"Ingress:test-namespace/test-ingress",
+				},
+				"KongServiceFacade:test-namespace/test-kong-service-facade": {
+					"Ingress:test-namespace/test-ingress",
+				},
+			},
+		},
+		{
+			name: "cache with HTTPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.HTTPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.HTTPRouteSpec{
+						Rules: []gatewayapi.HTTPRouteRule{
+							{
+								BackendRefs: []gatewayapi.HTTPBackendRef{
+									{
+										BackendRef: gatewayapi.BackendRef{
+											BackendObjectReference: gatewayapi.BackendObjectReference{
+												Name: "1",
+												Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"HTTPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"HTTPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with TLSRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.TLSRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.TLSRouteSpec{
+						Rules: []gatewayapi.TLSRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"TLSRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"TLSRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with TCPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.TCPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.TCPRouteSpec{
+						Rules: []gatewayapi.TCPRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"TCPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"TCPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with UDPRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.UDPRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.UDPRouteSpec{
+						Rules: []gatewayapi.UDPRouteRule{
+							{
+								BackendRefs: []gatewayapi.BackendRef{
+									{
+										BackendObjectReference: gatewayapi.BackendObjectReference{
+											Name: "1",
+											Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"UDPRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"UDPRoute:test-namespace/test-route",
+				},
+			},
+		},
+		{
+			name: "cache with GRPCRoute and its dependencies",
+			cache: cacheStoresFromObjs(t,
+				&gatewayapi.GRPCRoute{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "test-route",
+						Namespace: "test-namespace",
+						Annotations: map[string]string{
+							annotations.AnnotationPrefix + annotations.PluginsKey: "1,cluster-1",
+						},
+					},
+					Spec: gatewayapi.GRPCRouteSpec{
+						Rules: []gatewayapi.GRPCRouteRule{
+							{
+								BackendRefs: []gatewayapi.GRPCBackendRef{
+									{
+										BackendRef: gatewayapi.BackendRef{
+											BackendObjectReference: gatewayapi.BackendObjectReference{
+												Name: "1",
+												Kind: lo.ToPtr(gatewayapi.Kind("Service")),
+											},
+										},
+									},
+								},
+							},
+						},
+					},
+				},
+				testService(t, "1"),
+				testKongPlugin(t, "1"),
+				testKongClusterPlugin(t, "cluster-1"),
+			),
+			expectedAdjacencyMap: map[string][]string{
+				"GRPCRoute:test-namespace/test-route": {},
+				"Service:test-namespace/1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+				"KongPlugin:test-namespace/1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+				"KongClusterPlugin:test-namespace/cluster-1": {
+					"GRPCRoute:test-namespace/test-route",
+				},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			g, err := fallback.NewConfigGraphFromCacheStores(tc.cache)
+			require.NoError(t, err)
+			require.NotNil(t, g)
+			require.Equal(t, tc.expectedAdjacencyMap, adjacencyGraphStrings(t, g))
+		})
+	}
+}
diff --git a/internal/dataplane/fallback/helpers_test.go b/internal/dataplane/fallback/helpers_test.go
new file mode 100644
index 000000000..1e39d28d6
--- /dev/null
+++ b/internal/dataplane/fallback/helpers_test.go
@@ -0,0 +1,61 @@
+package fallback_test
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+	"github.com/kong/kubernetes-ingress-controller/v3/test/helpers"
+)
+
+const (
+	testNamespace = "test-namespace"
+)
+
+func testIngressClass(t *testing.T, name string) *netv1.IngressClass {
+	return helpers.WithTypeMeta(t, &netv1.IngressClass{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	})
+}
+
+func testService(t *testing.T, name string) *corev1.Service {
+	return helpers.WithTypeMeta(t, &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+}
+
+func testKongServiceFacade(t *testing.T, name string) *incubatorv1alpha1.KongServiceFacade {
+	return helpers.WithTypeMeta(t, &incubatorv1alpha1.KongServiceFacade{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+}
+
+func testKongPlugin(t *testing.T, name string) *kongv1.KongPlugin {
+	return helpers.WithTypeMeta(t, &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+}
+
+func testKongClusterPlugin(t *testing.T, name string) *kongv1.KongClusterPlugin {
+	return helpers.WithTypeMeta(t, &kongv1.KongClusterPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: testNamespace,
+		},
+	})
+}
diff --git a/internal/dataplane/kongstate/customentity_test.go b/internal/dataplane/kongstate/customentity_test.go
new file mode 100644
index 000000000..ff754de6d
--- /dev/null
+++ b/internal/dataplane/kongstate/customentity_test.go
@@ -0,0 +1,100 @@
+package kongstate
+
+import (
+	"testing"
+
+	"github.com/kong/go-kong/kong"
+	"github.com/stretchr/testify/require"
+)
+
+func TestExtractEntityFieldDefinitions(t *testing.T) {
+	testCases := []struct {
+		name           string
+		schema         kong.Schema
+		expectedFields map[string]EntityField
+	}{
+		{
+			name: "absent fields should have a nil value",
+			schema: map[string]interface{}{
+				"fields": []interface{}{
+					map[string]interface{}{
+						"foo": map[string]interface{}{
+							"type":     "string",
+							"required": true,
+						},
+					},
+					map[string]interface{}{
+						"bar": map[string]interface{}{
+							"type":      "foreign",
+							"required":  true,
+							"reference": "service",
+						},
+					},
+				},
+			},
+			expectedFields: map[string]EntityField{
+				"foo": {
+					Name:     "foo",
+					Type:     EntityFieldTypeString,
+					Required: true,
+					Auto:     false,
+					UUID:     false,
+				},
+				"bar": {
+					Name:      "bar",
+					Type:      EntityFieldTypeForeign,
+					Required:  true,
+					Reference: "service",
+				},
+			},
+		},
+		{
+			name: "irrelevant fields should be safely ignored",
+			schema: map[string]interface{}{
+				"fields": []interface{}{
+					map[string]interface{}{
+						"protocol": map[string]interface{}{
+							"type":     "string",
+							"required": true,
+							"default":  "http",
+							"one_of":   []string{"http", "https"},
+						},
+						"port": map[string]interface{}{
+							"type":     "integer",
+							"required": true,
+							"default":  80,
+							"min":      1,
+							"max":      65535,
+						},
+					},
+				},
+				"checks": "some_check",
+			},
+			expectedFields: map[string]EntityField{
+				"protocol": {
+					Name:     "protocol",
+					Type:     EntityFieldTypeString,
+					Required: true,
+					Default:  "http",
+				},
+				"port": {
+					Name:     "port",
+					Type:     EntityFieldTypeInteger,
+					Required: true,
+					Default:  int(80),
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			fields := ExtractEntityFieldDefinitions(tc.schema).Fields
+			for fieldName, expectedField := range tc.expectedFields {
+				actualField, ok := fields[fieldName]
+				require.Truef(t, ok, "field %s should exist", fieldName)
+				require.Equalf(t, expectedField, actualField, "field %s should be same as expected", fieldName)
+			}
+		})
+	}
+}
diff --git a/internal/dataplane/kongstate/kongstate_test.go b/internal/dataplane/kongstate/kongstate_test.go
index e87399b3b..b9bbcf21b 100644
--- a/internal/dataplane/kongstate/kongstate_test.go
+++ b/internal/dataplane/kongstate/kongstate_test.go
@@ -458,7 +458,8 @@ func TestGetPluginRelations(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := tt.args.state.getPluginRelations(); !reflect.DeepEqual(got, tt.want) {
+			store, _ := store.NewFakeStore(store.FakeObjects{})
+			if got := tt.args.state.getPluginRelations(store, logr.Discard()); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("getPluginRelations() = %v, want %v", got, tt.want)
 			}
 		})
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
index 4f3152ba6..87bc8c088 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/default_golden.yaml
@@ -29,7 +29,7 @@ services:
     - k8s-namespace:default
     - k8s-kind:GRPCRoute
     - k8s-group:gateway.networking.k8s.io
-    - k8s-version:v1alpha2
+    - k8s-version:v1
   tags:
   - k8s-name:UNKNOWN
   - k8s-namespace:UNKNOWN
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
index 5c585b916..8c9197377 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/expression-routes-on_golden.yaml
@@ -24,7 +24,7 @@ services:
     - k8s-namespace:default
     - k8s-kind:GRPCRoute
     - k8s-group:gateway.networking.k8s.io
-    - k8s-version:v1alpha2
+    - k8s-version:v1
   tags:
   - k8s-name:UNKNOWN
   - k8s-namespace:UNKNOWN
diff --git a/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml b/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
index 45461deb9..be2c54ee9 100644
--- a/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
+++ b/internal/dataplane/translator/testdata/golden/grpcroute-example/in.yaml
@@ -1,4 +1,4 @@
-apiVersion: gateway.networking.k8s.io/v1alpha2
+apiVersion: gateway.networking.k8s.io/v1
 kind: GRPCRoute
 metadata:
   name: grpcbin
@@ -30,4 +30,4 @@ spec:
     port: 443
     targetPort: 9001
   selector:
-    app: grpcbin
\ No newline at end of file
+    app: grpcbin
diff --git a/internal/dataplane/translator/translate_utils_test.go b/internal/dataplane/translator/translate_utils_test.go
index 48d576e4a..9bfb62425 100644
--- a/internal/dataplane/translator/translate_utils_test.go
+++ b/internal/dataplane/translator/translate_utils_test.go
@@ -250,7 +250,7 @@ func TestGetPermittedForReferenceGrantFrom(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.msg, func(t *testing.T) {
-			result := getPermittedForReferenceGrantFrom(tt.from, grants)
+			result := GetPermittedForReferenceGrantFrom(tt.from, grants)
 			assert.Equal(t, tt.result, result)
 		})
 	}
diff --git a/internal/dataplane/translator/wrappers_refchecker_test.go b/internal/dataplane/translator/wrappers_refchecker_test.go
index 233a73b78..9a18525c5 100644
--- a/internal/dataplane/translator/wrappers_refchecker_test.go
+++ b/internal/dataplane/translator/wrappers_refchecker_test.go
@@ -111,7 +111,7 @@ func TestRefChecker_IsRefAllowedByGrant(t *testing.T) {
 
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
-			rc := newRefCheckerForRoute(tc.route, tc.backendRef)
+			rc := gatewayapi.NewRefCheckerForRoute(tc.route, tc.backendRef)
 			result := rc.IsRefAllowedByGrant(tc.allowedRefs)
 			require.Equal(t, tc.expected, result)
 		})
diff --git a/internal/store/cache_key_funcs_test.go b/internal/store/cache_key_funcs_test.go
new file mode 100644
index 000000000..b0096aa0a
--- /dev/null
+++ b/internal/store/cache_key_funcs_test.go
@@ -0,0 +1,72 @@
+package store
+
+import (
+	"testing"
+
+	netv1 "k8s.io/api/networking/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestKeyFunc(t *testing.T) {
+	type args struct {
+		obj interface{}
+	}
+
+	type F struct {
+		Name      string
+		Namespace string
+	}
+	type B struct {
+		F
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    string
+		wantErr bool
+	}{
+		{
+			want: "Bar/Foo",
+			args: args{
+				obj: &F{
+					Name:      "Foo",
+					Namespace: "Bar",
+				},
+			},
+		},
+		{
+			want: "Bar/Fu",
+			args: args{
+				obj: B{
+					F: F{
+						Name:      "Fu",
+						Namespace: "Bar",
+					},
+				},
+			},
+		},
+		{
+			want: "default/foo",
+			args: args{
+				obj: netv1.Ingress{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "foo",
+						Namespace: "default",
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := namespacedKeyFunc(tt.args.obj)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("namespacedKeyFunc() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("namespacedKeyFunc() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/internal/store/cache_stores_snapshot_test.go b/internal/store/cache_stores_snapshot_test.go
index df8a533cd..4ea0d6b2f 100644
--- a/internal/store/cache_stores_snapshot_test.go
+++ b/internal/store/cache_stores_snapshot_test.go
@@ -1,17 +1,61 @@
 package store_test
 
 import (
+	"fmt"
 	"testing"
 
+	"github.com/google/uuid"
+	"github.com/samber/lo"
 	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
 	netv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	k8stypes "k8s.io/apimachinery/pkg/types"
 
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
 )
 
 func TestCacheStores_TakeSnapshot(t *testing.T) {
+	originalStores := getStoresForTests(t)
+	t.Log("Taking a snapshot of the originalStores")
+	snapshot, err := originalStores.TakeSnapshot()
+	require.NoError(t, err)
+	require.NotEqual(t, store.CacheStores{}, snapshot)
+
+	testIfSnapshotIsIndependentFromSource(t, &originalStores, &snapshot)
+}
+
+func TestCacheStores_TakeSnapshotIfChanged(t *testing.T) {
+	originalStores := getStoresForTests(t)
+	t.Log("Taking a snapshot of the originalStores")
+	originalStoresSnapshot, originalStoresHash, err := originalStores.TakeSnapshotIfChanged(store.SnapshotHashEmpty)
+	require.NoError(t, err)
+	require.Equal(t, store.SnapshotHash("4FU3CVGPMPGXTSC2I3L4UCKE6M46LWOJBSQMJ2N7HAYR46IVHNGQ===="), originalStoresHash)
+	require.NotEqual(t, store.CacheStores{}, originalStoresSnapshot)
+
+	t.Log("Taking again a snapshot of the originalStores")
+	originalStoresSnapshot2, originalStoresHash2, err := originalStores.TakeSnapshotIfChanged(originalStoresHash)
+	require.NoError(t, err)
+	require.Empty(t, originalStoresHash2)
+	require.Equal(t, store.CacheStores{}, originalStoresSnapshot2)
+
+	testIfSnapshotIsIndependentFromSource(t, &originalStores, &originalStoresSnapshot) // This modifies the originalStores.
+
+	t.Log("Taking a snapshot of the originalStores after modifying it")
+	modStoresSnapshot, modStoresHash, err := originalStores.TakeSnapshotIfChanged(originalStoresHash)
+	require.NoError(t, err)
+	require.NotEmpty(t, modStoresHash)
+	require.NotEqual(t, originalStoresHash, modStoresSnapshot)
+	require.NotEqual(t, store.CacheStores{}, modStoresSnapshot)
+
+	originalStores = getStoresForTests(t)
+	t.Log("Taking a snapshot of the originalStores after resetting it")
+}
+
+func getStoresForTests(t *testing.T) store.CacheStores {
+	t.Helper()
 	originalStores, err := store.NewCacheStoresFromObjs([]runtime.Object{
 		&netv1.Ingress{
 			TypeMeta: metav1.TypeMeta{
@@ -24,30 +68,47 @@ func TestCacheStores_TakeSnapshot(t *testing.T) {
 				Annotations: map[string]string{
 					"foo": "bar",
 				},
+				ResourceVersion: "123",
+				UID:             "19c1f570-b301-4b09-adcb-a1d29eb0b27e",
 			},
 		},
-	}...)
-	require.NoError(t, err)
-
-	t.Log("Taking a snapshot of the originalStores")
-	snapshot, err := originalStores.TakeSnapshot()
-	require.NoError(t, err)
-
-	t.Log("Adding a new object to the original store")
-	err = originalStores.Add(&netv1.Ingress{
-		TypeMeta: metav1.TypeMeta{
-			APIVersion: "networking.k8s.io/v1",
-			Kind:       "Ingress",
+		&netv1.Ingress{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "networking.k8s.io/v1",
+				Kind:       "Ingress",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "bar",
+				Namespace: "default",
+				Annotations: map[string]string{
+					"foo": "bar",
+				},
+				ResourceVersion: "456",
+				UID:             "3a463a14-59ba-422d-9d7b-02f57ddb2800",
+			},
 		},
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      "bar",
-			Namespace: "default",
-			Annotations: map[string]string{
-				"foo": "baz",
+		&netv1.Ingress{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "networking.k8s.io/v1",
+				Kind:       "Ingress",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "bar",
+				Namespace: "default",
+				Annotations: map[string]string{
+					"foo": "bar",
+				},
+				ResourceVersion: "789",
+				UID:             "2f37d2d3-0e95-40e7-810a-31953df5ee69",
 			},
 		},
-	})
+	}...)
 	require.NoError(t, err)
+	return originalStores
+}
+
+func testIfSnapshotIsIndependentFromSource(t *testing.T, originalStores, snapshot *store.CacheStores) {
+	t.Helper()
 
 	// We'll use ingressMeta in .Get() calls.
 	ingressMeta := &netv1.Ingress{
@@ -69,6 +130,7 @@ func TestCacheStores_TakeSnapshot(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, "bar", ingressFromOriginalStore.Annotations["foo"])
 	ingressFromOriginalStore.Annotations["foo"] = "qux"
+	ingressFromOriginalStore.ResourceVersion = "567"
 
 	t.Log("Checking that the original store returns the modified object")
 	obj, ok, err = originalStores.IngressV1.Get(ingressMeta)
@@ -86,3 +148,179 @@ func TestCacheStores_TakeSnapshot(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, "bar", ingressFromSnapshot.Annotations["foo"], "Snapshot should not be affected by the change in the original store")
 }
+
+func BenchmarkCacheStores_TakeSnapshot(b *testing.B) {
+	smallStores, err := store.NewCacheStoresFromObjs([]runtime.Object{
+		&netv1.Ingress{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "networking.k8s.io/v1",
+				Kind:       "Ingress",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "foo",
+				Namespace: "default",
+				Annotations: map[string]string{
+					"foo": "bar",
+				},
+				ResourceVersion: "123",
+				UID:             k8stypes.UID(uuid.New().String()),
+			},
+		},
+	}...)
+	require.NoError(b, err)
+
+	var k8sObjects []runtime.Object
+	for i := 0; i < 1_000; i++ {
+		route := &gatewayapi.HTTPRoute{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "gateway.networking.k8s.io/v1",
+				Kind:       "HTTPRoute",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            fmt.Sprintf("route-%d", i),
+				Namespace:       "default",
+				ResourceVersion: fmt.Sprintf("12%d", i),
+				UID:             k8stypes.UID(uuid.New().String()),
+			},
+			Spec: gatewayapi.HTTPRouteSpec{
+				Rules: []gatewayapi.HTTPRouteRule{
+					{
+						Matches: []gatewayapi.HTTPRouteMatch{
+							{
+								Path: &gatewayapi.HTTPPathMatch{
+									Type:  lo.ToPtr(gatewayapi.PathMatchExact),
+									Value: lo.ToPtr("/test1"),
+								},
+								Method: lo.ToPtr(gatewayapi.HTTPMethodGet),
+							},
+							{
+								Path: &gatewayapi.HTTPPathMatch{
+									Type:  lo.ToPtr(gatewayapi.PathMatchExact),
+									Value: lo.ToPtr("/test2"),
+								},
+								Method: lo.ToPtr(gatewayapi.HTTPMethodGet),
+							},
+						},
+					},
+				},
+				CommonRouteSpec: gatewayapi.CommonRouteSpec{},
+			},
+		}
+		k8sObjects = append(k8sObjects, route)
+
+		service := &corev1.Service{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "v1",
+				Kind:       "Service",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            fmt.Sprintf("service-%d", i),
+				Namespace:       "default",
+				ResourceVersion: fmt.Sprintf("12%d", i),
+				UID:             k8stypes.UID(uuid.New().String()),
+			},
+			Spec: corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name: "http",
+						Port: int32(i),
+					},
+				},
+			},
+		}
+		k8sObjects = append(k8sObjects, service)
+
+		ingress := &netv1.Ingress{
+			TypeMeta: metav1.TypeMeta{
+				APIVersion: "networking.k8s.io/v1",
+				Kind:       "Ingress",
+			},
+			ObjectMeta: metav1.ObjectMeta{
+				Name:            fmt.Sprintf("ingress-%d", i),
+				Namespace:       "default",
+				ResourceVersion: fmt.Sprintf("12%d", i),
+				UID:             k8stypes.UID(uuid.New().String()),
+			},
+			Spec: netv1.IngressSpec{
+				Rules: []netv1.IngressRule{
+					{
+						Host: fmt.Sprintf("host-%d", i),
+						IngressRuleValue: netv1.IngressRuleValue{
+							HTTP: &netv1.HTTPIngressRuleValue{
+								Paths: []netv1.HTTPIngressPath{
+									{
+										Path:     fmt.Sprintf("/path-%d", i),
+										PathType: lo.ToPtr(netv1.PathTypeExact),
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+		k8sObjects = append(k8sObjects, ingress)
+	}
+	bigStores, err := store.NewCacheStoresFromObjs(k8sObjects...)
+	require.NoError(b, err)
+
+	b.Run("Small_Without_Cache", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			s, err := smallStores.TakeSnapshot()
+			if err != nil || (s == store.CacheStores{}) {
+				b.Fatalf("unexpected error or empty snapshot err: %v, snapshot: %v", err, s)
+			}
+		}
+	})
+	b.Run("Big_Without_Cache", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			s, err := bigStores.TakeSnapshot()
+			if err != nil || (s == store.CacheStores{}) {
+				b.Fatalf("unexpected error or empty snapshot err: %v, snapshot: %v", err, s)
+			}
+		}
+	})
+	b.Run("Small_With_Cache", func(b *testing.B) {
+		s, hash, err := smallStores.TakeSnapshotIfChanged(store.SnapshotHashEmpty)
+		require.NoError(b, err)
+		require.NotEmpty(b, hash)
+		require.NotEqual(b, store.CacheStores{}, s)
+		for i := 0; i < b.N; i++ {
+			s, hashNew, err := smallStores.TakeSnapshotIfChanged(hash)
+			if err != nil || hashNew != store.SnapshotHashEmpty || s != (store.CacheStores{}) {
+				b.Fatalf("unexpected error or non-empty snapshot err: %v, hash: %s, snapshot: %v", err, hashNew, s)
+			}
+		}
+	})
+
+	b.Run("Big_With_Cache", func(b *testing.B) {
+		s, hash, err := bigStores.TakeSnapshotIfChanged(store.SnapshotHashEmpty)
+		require.NoError(b, err)
+		require.NotEmpty(b, hash)
+		require.NotEqual(b, store.CacheStores{}, s)
+		for i := 0; i < b.N; i++ {
+			s, hashNew, err := bigStores.TakeSnapshotIfChanged(hash)
+			if err != nil || hashNew != store.SnapshotHashEmpty || s != (store.CacheStores{}) {
+				b.Fatalf("unexpected error or non-empty snapshot err: %v, hash: %s, snapshot: %v", err, hashNew, s)
+			}
+		}
+	})
+
+	b.Run("Small_With_Missed_Cache", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			s, hashNew, err := smallStores.TakeSnapshotIfChanged(store.SnapshotHashEmpty)
+			if err != nil || hashNew == store.SnapshotHashEmpty || s == (store.CacheStores{}) {
+				b.Fatalf("unexpected error or empty snapshot err: %v, hash: %s, snapshot: %v", err, hashNew, s)
+			}
+		}
+	})
+
+	b.Run("Big_With_Missed_Cache", func(b *testing.B) {
+		for i := 0; i < b.N; i++ {
+			s, hashNew, err := bigStores.TakeSnapshotIfChanged(store.SnapshotHashEmpty)
+			if err != nil || hashNew == store.SnapshotHashEmpty || s == (store.CacheStores{}) {
+				b.Fatalf("unexpected error or empty snapshot err: %v, hash: %s, snapshot: %v", err, hashNew, s)
+			}
+		}
+	})
+}
diff --git a/internal/store/cache_stores_test.go b/internal/store/cache_stores_test.go
deleted file mode 100644
index e00dfa362..000000000
--- a/internal/store/cache_stores_test.go
+++ /dev/null
@@ -1,69 +0,0 @@
-package store_test
-
-import (
-	"testing"
-
-	"github.com/samber/lo"
-	"github.com/stretchr/testify/require"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
-	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
-	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
-)
-
-func TestCacheStores(t *testing.T) {
-	testCases := []struct {
-		name          string
-		objectToStore client.Object
-	}{
-		{
-			name: "KongUpstreamPolicy",
-			objectToStore: &kongv1beta1.KongUpstreamPolicy{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "upstream-policy",
-					Namespace: "default",
-				},
-				Spec: kongv1beta1.KongUpstreamPolicySpec{
-					Algorithm: lo.ToPtr("least-connections"),
-				},
-			},
-		},
-		{
-			name: "KongServiceFacade",
-			objectToStore: &incubatorv1alpha1.KongServiceFacade{
-				ObjectMeta: metav1.ObjectMeta{
-					Name:      "service-facade",
-					Namespace: "default",
-				},
-				Spec: incubatorv1alpha1.KongServiceFacadeSpec{
-					Backend: incubatorv1alpha1.KongServiceFacadeBackend{
-						Name: "backend",
-						Port: 80,
-					},
-				},
-			},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			s := store.NewCacheStores()
-			err := s.Add(tc.objectToStore)
-			require.NoError(t, err)
-
-			storedObj, ok, err := s.Get(tc.objectToStore)
-			require.NoError(t, err)
-			require.True(t, ok)
-			require.Equal(t, tc.objectToStore, storedObj)
-
-			err = s.Delete(tc.objectToStore)
-			require.NoError(t, err)
-
-			_, ok, err = s.Get(tc.objectToStore)
-			require.NoError(t, err, err)
-			require.False(t, ok)
-		})
-	}
-}
diff --git a/internal/store/fake_store_test.go b/internal/store/fake_store_test.go
index cba045816..27a10b28b 100644
--- a/internal/store/fake_store_test.go
+++ b/internal/store/fake_store_test.go
@@ -19,70 +19,6 @@ import (
 	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
 )
 
-func TestKeyFunc(t *testing.T) {
-	type args struct {
-		obj interface{}
-	}
-
-	type F struct {
-		Name      string
-		Namespace string
-	}
-	type B struct {
-		F
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    string
-		wantErr bool
-	}{
-		{
-			want: "Bar/Foo",
-			args: args{
-				obj: &F{
-					Name:      "Foo",
-					Namespace: "Bar",
-				},
-			},
-		},
-		{
-			want: "Bar/Fu",
-			args: args{
-				obj: B{
-					F: F{
-						Name:      "Fu",
-						Namespace: "Bar",
-					},
-				},
-			},
-		},
-		{
-			want: "default/foo",
-			args: args{
-				obj: netv1.Ingress{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      "foo",
-						Namespace: "default",
-					},
-				},
-			},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := keyFunc(tt.args.obj)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("keyFunc() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if got != tt.want {
-				t.Errorf("keyFunc() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
 func TestFakeStoreEmpty(t *testing.T) {
 	assert := assert.New(t)
 	store, err := NewFakeStore(FakeObjects{})
diff --git a/internal/store/zz_generated_cache_stores_test.go b/internal/store/zz_generated_cache_stores_test.go
new file mode 100644
index 000000000..eae0631ae
--- /dev/null
+++ b/internal/store/zz_generated_cache_stores_test.go
@@ -0,0 +1,163 @@
+// Code generated by hack/generators/cache-stores/main.go; DO NOT EDIT.
+// If you want to add a new type to the cache store, you need to add a new entry to the supportedTypes list in spec.go.
+package store_test
+
+import (
+	"testing"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
+	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	discoveryv1 "k8s.io/api/discovery/v1"
+	netv1 "k8s.io/api/networking/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/store"
+	kongv1beta1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1beta1"
+	incubatorv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/incubator/v1alpha1"
+)
+
+func TestCacheStores(t *testing.T) {
+	testCases := []struct {
+		name          string
+		objectToStore client.Object
+	}{
+
+		{
+			name:          "Ingress",
+			objectToStore: &netv1.Ingress{},
+		},
+
+		{
+			name:          "IngressClass",
+			objectToStore: &netv1.IngressClass{},
+		},
+
+		{
+			name:          "Service",
+			objectToStore: &corev1.Service{},
+		},
+
+		{
+			name:          "Secret",
+			objectToStore: &corev1.Secret{},
+		},
+
+		{
+			name:          "EndpointSlice",
+			objectToStore: &discoveryv1.EndpointSlice{},
+		},
+
+		{
+			name:          "HTTPRoute",
+			objectToStore: &gatewayapi.HTTPRoute{},
+		},
+
+		{
+			name:          "UDPRoute",
+			objectToStore: &gatewayapi.UDPRoute{},
+		},
+
+		{
+			name:          "TCPRoute",
+			objectToStore: &gatewayapi.TCPRoute{},
+		},
+
+		{
+			name:          "TLSRoute",
+			objectToStore: &gatewayapi.TLSRoute{},
+		},
+
+		{
+			name:          "GRPCRoute",
+			objectToStore: &gatewayapi.GRPCRoute{},
+		},
+
+		{
+			name:          "ReferenceGrant",
+			objectToStore: &gatewayapi.ReferenceGrant{},
+		},
+
+		{
+			name:          "Gateway",
+			objectToStore: &gatewayapi.Gateway{},
+		},
+
+		{
+			name:          "KongPlugin",
+			objectToStore: &kongv1.KongPlugin{},
+		},
+
+		{
+			name:          "KongClusterPlugin",
+			objectToStore: &kongv1.KongClusterPlugin{},
+		},
+
+		{
+			name:          "KongConsumer",
+			objectToStore: &kongv1.KongConsumer{},
+		},
+
+		{
+			name:          "KongConsumerGroup",
+			objectToStore: &kongv1beta1.KongConsumerGroup{},
+		},
+
+		{
+			name:          "KongIngress",
+			objectToStore: &kongv1.KongIngress{},
+		},
+
+		{
+			name:          "TCPIngress",
+			objectToStore: &kongv1beta1.TCPIngress{},
+		},
+
+		{
+			name:          "UDPIngress",
+			objectToStore: &kongv1beta1.UDPIngress{},
+		},
+
+		{
+			name:          "KongUpstreamPolicy",
+			objectToStore: &kongv1beta1.KongUpstreamPolicy{},
+		},
+
+		{
+			name:          "IngressClassParameters",
+			objectToStore: &kongv1alpha1.IngressClassParameters{},
+		},
+
+		{
+			name:          "KongServiceFacade",
+			objectToStore: &incubatorv1alpha1.KongServiceFacade{},
+		},
+
+		{
+			name:          "KongVault",
+			objectToStore: &kongv1alpha1.KongVault{},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := store.NewCacheStores()
+			err := s.Add(tc.objectToStore)
+			require.NoError(t, err)
+
+			storedObj, ok, err := s.Get(tc.objectToStore)
+			require.NoError(t, err)
+			require.True(t, ok)
+			require.Equal(t, tc.objectToStore, storedObj)
+
+			err = s.Delete(tc.objectToStore)
+			require.NoError(t, err)
+
+			_, ok, err = s.Get(tc.objectToStore)
+			require.NoError(t, err, err)
+			require.False(t, ok)
+		})
+	}
+}
diff --git a/test/envtest/admission_webhook_envtest_test.go b/test/envtest/admission_webhook_envtest_test.go
index fc86115f3..d13deb277 100644
--- a/test/envtest/admission_webhook_envtest_test.go
+++ b/test/envtest/admission_webhook_envtest_test.go
@@ -27,6 +27,7 @@ import (
 	kongv1alpha1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1alpha1"
 	"github.com/kong/kubernetes-ingress-controller/v3/test/helpers"
 	"github.com/kong/kubernetes-ingress-controller/v3/test/helpers/certificate"
+	"github.com/kong/kubernetes-ingress-controller/v3/test/internal/testenv"
 )
 
 func TestAdmissionWebhook_KongVault(t *testing.T) {
@@ -1192,6 +1193,160 @@ func TestAdmissionWebhook_SecretCredentials(t *testing.T) {
 	})
 }
 
+func TestAdmissionWebhook_KongCustomEntities(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var (
+		scheme           = Scheme(t, WithKong)
+		envcfg           = Setup(t, scheme)
+		ctrlClientGlobal = NewControllerClient(t, scheme, envcfg)
+		ns               = CreateNamespace(ctx, t, ctrlClientGlobal)
+		ctrlClient       = client.NewNamespacedClient(ctrlClientGlobal, ns.Name)
+
+		webhookCert, webhookKey = certificate.MustGenerateSelfSignedCertPEMFormat(
+			certificate.WithDNSNames("localhost"),
+		)
+		admissionWebhookPort = helpers.GetFreePort(t)
+
+		kongContainer = runKongEnterprise(ctx, t)
+	)
+
+	_, logs := RunManager(ctx, t, envcfg,
+		AdminAPIOptFns(),
+		WithPublishService(ns.Name),
+		WithAdmissionWebhookEnabled(webhookKey, webhookCert, fmt.Sprintf(":%d", admissionWebhookPort)),
+		WithKongAdminURLs(kongContainer.AdminURL(ctx, t)),
+	)
+	WaitForManagerStart(t, logs)
+	setupValidatingWebhookConfiguration(ctx, t, admissionWebhookPort, webhookCert, ctrlClient)
+
+	testCases := []struct {
+		name                     string
+		requireEnterpriseLicense bool
+		entity                   *kongv1alpha1.KongCustomEntity
+		valid                    bool
+
+		errContains string
+	}{
+		{
+			name: "entity not supported in Kong",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "not-supported-entity",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType:     "invalid_entity",
+					ControllerName: annotations.DefaultIngressClass,
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte(`{"key":"value"}`),
+					},
+				},
+			},
+			valid:       false,
+			errContains: "failed to get schema of Kong entity type 'invalid_entity'",
+		},
+		{
+			name: "valid session entity",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "session-1",
+					Annotations: map[string]string{
+						annotations.IngressClassKey: annotations.DefaultIngressClass,
+					},
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "sessions",
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte(`{"session_id":"session1"}`),
+					},
+				},
+			},
+			valid: true,
+		},
+		{
+			name: "invalid session entity",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "session-1",
+					Annotations: map[string]string{
+						annotations.IngressClassKey: annotations.DefaultIngressClass,
+					},
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "sessions",
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte(`{"session_id":"session2","foo":"bar"}`),
+					},
+				},
+			},
+			valid: false,
+		},
+		{
+			name: "valid degraphql_route entity",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "degraphql-route-1",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType:     "degraphql_routes",
+					ControllerName: annotations.DefaultIngressClass,
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte(`{"uri":"/me","query":"query{ viewer { login}}"}`),
+					},
+				},
+			},
+			valid: true,
+		},
+		{
+			name: "invalid degraphql_route entity",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "degraphql-route-1",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType:     "degraphql_routes",
+					ControllerName: annotations.DefaultIngressClass,
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte(`{"uri":"/me","query":"query{ viewer { login}}","foo":"bar"}`),
+					},
+				},
+			},
+			valid: false,
+		},
+		{
+			name: "KongCustomEntity not controlled by the current controller",
+			entity: &kongv1alpha1.KongCustomEntity{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unrelated-entity",
+				},
+				Spec: kongv1alpha1.KongCustomEntitySpec{
+					EntityType: "unrelated_entity",
+					Fields: apiextensionsv1.JSON{
+						Raw: []byte("{}"),
+					},
+				},
+			},
+			valid: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			if tc.requireEnterpriseLicense && !testenv.KongEnterpriseEnabled() {
+				t.Skip("Skipped because Kong enterprise is not enabled")
+			}
+			err := ctrlClient.Create(ctx, tc.entity)
+			if tc.valid {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.errContains)
+			}
+		})
+	}
+}
+
 // createKongConsumers creates a provider number of consumers on the cluster.
 // Resources will be created in client's default namespace. When using controller-runtime's
 // client you can specify that by calling client.NewNamespacedClient(client, namespace).
diff --git a/test/helpers/k8s_meta.go b/test/helpers/k8s_meta.go
new file mode 100644
index 000000000..04cdb0125
--- /dev/null
+++ b/test/helpers/k8s_meta.go
@@ -0,0 +1,20 @@
+package helpers
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime"
+
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/manager/scheme"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
+)
+
+// WithTypeMeta adds type meta to the given object based on its Go type.
+func WithTypeMeta[T runtime.Object](t *testing.T, obj T) T {
+	s, err := scheme.Get()
+	require.NoError(t, err)
+	err = util.PopulateTypeMeta(obj, s)
+	require.NoError(t, err)
+	return obj
+}
diff --git a/test/integration/plugin_test.go b/test/integration/plugin_test.go
index cbfd9da7e..dc21fd0bf 100644
--- a/test/integration/plugin_test.go
+++ b/test/integration/plugin_test.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/kong/go-kong/kong"
 	"github.com/kong/kubernetes-testing-framework/pkg/clusters"
 	"github.com/kong/kubernetes-testing-framework/pkg/utils/kubernetes/generators"
@@ -19,8 +20,11 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	gatewayclient "sigs.k8s.io/gateway-api/pkg/client/clientset/versioned"
 
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/annotations"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/gatewayapi"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/labels"
 	kongv1 "github.com/kong/kubernetes-ingress-controller/v3/pkg/apis/configuration/v1"
 	"github.com/kong/kubernetes-ingress-controller/v3/pkg/clientset"
 	"github.com/kong/kubernetes-ingress-controller/v3/test"
@@ -472,3 +476,193 @@ func TestPluginOrdering(t *testing.T) {
 	require.NoError(t, clusters.DeleteIngress(ctx, env.Cluster(), ns.Name, ingress))
 	helpers.EventuallyExpectHTTP404WithNoRoute(t, proxyHTTPURL, proxyHTTPURL.Host, "/test_plugin_ordering", ingressWait, waitTick, nil)
 }
+
+// TestPluginCrossNamespaceReference creates an Ingress in one namespace and a KongConsumer in another. It tests that
+// the controller can generate a plugin attached to the generated route and consumer if and only if a ReferenceGrant
+// allows access to the KongPlugin from KongConsumers in a remote namespace.
+func TestPluginCrossNamespaceReference(t *testing.T) {
+	ctx := context.Background()
+
+	t.Parallel()
+	ns, cleaner := helpers.Setup(ctx, t, env)
+	cluster := env.Cluster()
+	remote, err := clusters.GenerateNamespace(ctx, cluster, helpers.LabelValueForTest(t))
+	require.NoError(t, err)
+	cleaner.AddNamespace(remote)
+
+	gatewayClient, err := gatewayclient.NewForConfig(env.Cluster().Config())
+	require.NoError(t, err)
+
+	t.Log("deploying a minimal HTTP container deployment to test Ingress routes")
+	container := generators.NewContainer("httpbin", test.HTTPBinImage, test.HTTPBinPort)
+	deployment := generators.NewDeploymentForContainer(container)
+	deployment, err = env.Cluster().Client().AppsV1().Deployments(ns.Name).Create(ctx, deployment, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(deployment)
+
+	t.Logf("exposing deployment %s via service", deployment.Name)
+	service := generators.NewServiceForDeployment(deployment, corev1.ServiceTypeLoadBalancer)
+	_, err = env.Cluster().Client().CoreV1().Services(ns.Name).Create(ctx, service, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(service)
+
+	t.Logf("creating an ingress for service %s with ingress.class %s", service.Name, consts.IngressClass)
+	ingress := generators.NewIngressForService("/test_plugin_reference", map[string]string{
+		"konghq.com/strip-path": "true",
+	}, service)
+	ingress.Spec.IngressClassName = kong.String(consts.IngressClass)
+	require.NoError(t, clusters.DeployIngress(ctx, env.Cluster(), ns.Name, ingress))
+	cleaner.Add(ingress)
+
+	t.Log("waiting for routes from Ingress to be operational")
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		resp, err := helpers.DefaultHTTPClient().Get(fmt.Sprintf("%s/test_plugin_reference", proxyHTTPURL))
+		if !assert.NoError(c, err) {
+			return
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode == http.StatusOK {
+			b := new(bytes.Buffer)
+			_, err := b.ReadFrom(resp.Body)
+			assert.NoError(c, err)
+			assert.True(c, strings.Contains(b.String(), "<title>httpbin.org</title>"))
+		}
+	}, ingressWait, waitTick)
+
+	t.Log("creating plugins")
+	kongplugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "teapot",
+		},
+		InstanceName: "example",
+		PluginName:   "request-termination",
+		Config: apiextensionsv1.JSON{
+			Raw: []byte(`{"status_code": 418}`),
+		},
+	}
+	c, err := clientset.NewForConfig(env.Cluster().Config())
+	require.NoError(t, err)
+	kongplugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, kongplugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(kongplugin)
+
+	// this isn't strictly necessary, but it's convenient for diagnosing issues if the test breaks, as it confirms
+	// that a request was indeed recognized as being from the consumer. unlike the other consumer plugin, it's associated
+	// with the consumer alone and works regardless of the grant state or any cross-namespace shenanigans
+	kongslugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "snurch",
+		},
+		InstanceName: "snurch",
+		PluginName:   "correlation-id",
+		Config: apiextensionsv1.JSON{
+			Raw: []byte(`{"header_name": "snurch", "echo_downstream": true}`),
+		},
+	}
+	kongslugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, kongslugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(kongslugin)
+
+	t.Log("creating consumer and credential")
+	credential := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+			Labels: map[string]string{
+				labels.LabelPrefix + labels.CredentialKey: "key-auth",
+			},
+		},
+		StringData: map[string]string{
+			"key": "thirtytangas",
+		},
+	}
+	_, err = env.Cluster().Client().CoreV1().Secrets(remote.Name).Create(ctx, credential, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(credential)
+
+	consumer := &kongv1.KongConsumer{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+			Annotations: map[string]string{
+				annotations.IngressClassKey:                           consts.IngressClass,
+				annotations.AnnotationPrefix + annotations.PluginsKey: strings.Join([]string{kongslugin.Name, fmt.Sprintf("%s:%s", kongplugin.Namespace, kongplugin.Name)}, ","),
+			},
+		},
+		Username:    uuid.NewString(),
+		Credentials: []string{credential.Name},
+	}
+	_, err = c.ConfigurationV1().KongConsumers(remote.Name).Create(ctx, consumer, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(consumer)
+
+	t.Log("creating auth plugin to identify consumer accessing route")
+	authplugin := &kongv1.KongPlugin{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: ns.Name,
+			Name:      "keykey",
+		},
+		PluginName: "key-auth",
+	}
+	authplugin, err = c.ConfigurationV1().KongPlugins(ns.Name).Create(ctx, authplugin, metav1.CreateOptions{})
+	require.NoError(t, err)
+	cleaner.Add(authplugin)
+
+	cleaner.Add(kongplugin)
+	t.Logf("updating Ingress to use auth plugin %s", kongplugin.Name)
+	require.Eventually(t, func() bool {
+		ingress, err := env.Cluster().Client().NetworkingV1().Ingresses(ns.Name).Get(ctx, ingress.Name, metav1.GetOptions{})
+		if err != nil {
+			return false
+		}
+		ingress.ObjectMeta.Annotations[annotations.AnnotationPrefix+annotations.PluginsKey] = strings.Join([]string{kongplugin.Name, authplugin.Name}, ",")
+		_, err = env.Cluster().Client().NetworkingV1().Ingresses(ns.Name).Update(ctx, ingress, metav1.UpdateOptions{})
+		return err == nil
+	}, ingressWait, waitTick)
+
+	t.Logf("validating that plugin %s is not configured without a grant", kongplugin.Name)
+	assert.Never(t, func() bool {
+		req := helpers.MustHTTPRequest(t, "GET", proxyHTTPURL.String(), "/test_plugin_reference?key=thirtytangas", nil)
+		resp, err := helpers.DefaultHTTPClientWithProxy(proxyHTTPURL).Do(req)
+		if err != nil {
+			return false
+		}
+		defer resp.Body.Close()
+		return resp.StatusCode == http.StatusTeapot
+	}, ingressWait, waitTick)
+
+	t.Logf("creating a ReferenceGrant that permits kongconsumer access from %s to kongplugins in %s", remote.Name, ns.Name)
+	grant := &gatewayapi.ReferenceGrant{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: uuid.NewString(),
+		},
+		Spec: gatewayapi.ReferenceGrantSpec{
+			From: []gatewayapi.ReferenceGrantFrom{
+				{
+					Group:     gatewayapi.Group("configuration.konghq.com"),
+					Kind:      gatewayapi.Kind("KongConsumer"),
+					Namespace: gatewayapi.Namespace(remote.Name),
+				},
+			},
+			To: []gatewayapi.ReferenceGrantTo{
+				{
+					Group: gatewayapi.Group("configuration.konghq.com"),
+					Kind:  gatewayapi.Kind("KongPlugin"),
+				},
+			},
+		},
+	}
+	_, err = gatewayClient.GatewayV1beta1().ReferenceGrants(ns.Name).Create(ctx, grant, metav1.CreateOptions{})
+	require.NoError(t, err)
+
+	t.Logf("validating that plugin %s was successfully configured", kongplugin.Name)
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		req := helpers.MustHTTPRequest(t, "GET", proxyHTTPURL.String(), "/test_plugin_reference?apikey=thirtytangas", nil)
+		resp, err := helpers.DefaultHTTPClientWithProxy(proxyHTTPURL).Do(req)
+		if !assert.NoError(c, err) {
+			return
+		}
+		defer resp.Body.Close()
+		assert.True(c, resp.StatusCode == http.StatusTeapot)
+	}, ingressWait, waitTick)
+}
diff --git a/test/kongintegration/containers/kong.go b/test/kongintegration/containers/kong.go
index 12467078f..d2d8b31bb 100644
--- a/test/kongintegration/containers/kong.go
+++ b/test/kongintegration/containers/kong.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"os"
 	"testing"
 	"time"
 
@@ -53,6 +54,7 @@ func NewKong(ctx context.Context, t *testing.T, opts ...KongOpt) Kong {
 			"KONG_ADMIN_LISTEN":  fmt.Sprintf("0.0.0.0:%s", kongAdminPort),
 			"KONG_PROXY_LISTEN":  fmt.Sprintf("0.0.0.0:%s", kongProxyPort),
 			"KONG_ROUTER_FLAVOR": defaultRouterFlavor,
+			"KONG_LICENSE_DATA":  os.Getenv("KONG_LICENSE_DATA"),
 		},
 		WaitingFor: wait.ForAll(
 			wait.ForListeningPort(kongAdminPort),