Merge pull request #2484 from beyonnex-io/feature/cache-read-subject-classification

Memoize per-resource READ subject classification in PolicyEnforcer (per-event enforcement CPU hotspot)

Author: thjaeckle

deployment/helm/ditto/Chart.yaml

@@ -16,7 +16,7 @@ description: |
   A digital twin is a virtual, cloud based, representation of his real world counterpart
   (real world “Things”, e.g. devices like sensors, smart heating, connected cars, smart grids, EV charging stations etc).
 type: application
-version: 4.3.0  # chart version is effectively set by release-job
+version: 4.3.1  # chart version is effectively set by release-job
 appVersion: 3.9.3
 keywords:
   - iot-chart

deployment/helm/ditto/templates/connectivity-deployment.yaml

@@ -312,6 +312,8 @@ spec:
               value: "{{ .Values.connectivity.config.policiesEnforcer.cache.maxSize }}"
             - name: DITTO_POLICIES_ENFORCER_NAMESPACE_FILTERED_MAX_SIZE
               value: "{{ .Values.connectivity.config.policiesEnforcer.cache.namespaceFilteredMaxSize }}"
+            - name: DITTO_POLICIES_ENFORCER_READ_CLASSIFICATION_MAX_SIZE
+              value: "{{ .Values.connectivity.config.policiesEnforcer.cache.readClassificationMaxSize }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_WRITE
               value: "{{ .Values.connectivity.config.policiesEnforcer.cache.expireAfterWrite }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_ACCESS

deployment/helm/ditto/templates/policies-deployment.yaml

@@ -316,6 +316,8 @@ spec:
               value: "{{ .Values.policies.config.policiesEnforcer.cache.maxSize }}"
             - name: DITTO_POLICIES_ENFORCER_NAMESPACE_FILTERED_MAX_SIZE
               value: "{{ .Values.policies.config.policiesEnforcer.cache.namespaceFilteredMaxSize }}"
+            - name: DITTO_POLICIES_ENFORCER_READ_CLASSIFICATION_MAX_SIZE
+              value: "{{ .Values.policies.config.policiesEnforcer.cache.readClassificationMaxSize }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_WRITE
               value: "{{ .Values.policies.config.policiesEnforcer.cache.expireAfterWrite }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_ACCESS

deployment/helm/ditto/templates/things-deployment.yaml

@@ -314,6 +314,8 @@ spec:
               value: "{{ .Values.things.config.policiesEnforcer.cache.maxSize }}"
             - name: DITTO_POLICIES_ENFORCER_NAMESPACE_FILTERED_MAX_SIZE
               value: "{{ .Values.things.config.policiesEnforcer.cache.namespaceFilteredMaxSize }}"
+            - name: DITTO_POLICIES_ENFORCER_READ_CLASSIFICATION_MAX_SIZE
+              value: "{{ .Values.things.config.policiesEnforcer.cache.readClassificationMaxSize }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_WRITE
               value: "{{ .Values.things.config.policiesEnforcer.cache.expireAfterWrite }}"
             - name: DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_ACCESS

deployment/helm/ditto/values.yaml

@@ -847,6 +847,10 @@ policies:
         # namespaceFilteredMaxSize bounds the per-enforcer cache of namespace-filtered enforcers, avoiding
         # rebuilding the enforcer tree per signal for policies with namespace-scoped entries
         namespaceFilteredMaxSize: 100
+        # readClassificationMaxSize bounds the per-enforcer memo of classifySubjects(resource, READ) results,
+        # avoiding re-walking the policy tree per emitted ThingEvent for per-event read-authorization and
+        # read-grant collection; caps distinct resource paths cached per policy
+        readClassificationMaxSize: 1000
         # expireAfterWrite the maximum duration of inconsistency after losing a cache invalidation
         expireAfterWrite: 4h
         # expireAfterAccess prolonged on each cache access by that duration
@@ -1218,6 +1222,10 @@ things:
         # namespaceFilteredMaxSize bounds the per-enforcer cache of namespace-filtered enforcers, avoiding
         # rebuilding the enforcer tree per signal for policies with namespace-scoped entries
         namespaceFilteredMaxSize: 100
+        # readClassificationMaxSize bounds the per-enforcer memo of classifySubjects(resource, READ) results,
+        # avoiding re-walking the policy tree per emitted ThingEvent for per-event read-authorization and
+        # read-grant collection; caps distinct resource paths cached per policy
+        readClassificationMaxSize: 1000
         # expireAfterWrite the maximum duration of inconsistency after losing a cache invalidation
         expireAfterWrite: 4h
         # expireAfterAccess prolonged on each cache access by that duration
@@ -1929,6 +1937,10 @@ connectivity:
         # namespaceFilteredMaxSize bounds the per-enforcer cache of namespace-filtered enforcers, avoiding
         # rebuilding the enforcer tree per signal for policies with namespace-scoped entries
         namespaceFilteredMaxSize: 100
+        # readClassificationMaxSize bounds the per-enforcer memo of classifySubjects(resource, READ) results,
+        # avoiding re-walking the policy tree per emitted ThingEvent for per-event read-authorization and
+        # read-grant collection; caps distinct resource paths cached per policy
+        readClassificationMaxSize: 1000
         # expireAfterWrite the maximum duration of inconsistency after losing a cache invalidation
         expireAfterWrite: 4h
         # expireAfterAccess prolonged on each cache access by that duration

policies/enforcement/src/main/java/org/eclipse/ditto/policies/enforcement/PolicyEnforcer.java

@@ -40,10 +40,15 @@ public final class PolicyEnforcerCacheLoader implements AsyncCacheLoader<PolicyI
      * namespace-filtered-enforcer cache size; default provided by reference.conf. */
     private static final String NAMESPACE_FILTERED_ENFORCER_MAX_SIZE_KEY = "namespace-filtered-enforcer-max-size";
 
+    /** Config key (relative to {@link PolicyEnforcerProvider#ENFORCER_CACHE_CONFIG_KEY}) for the per-enforcer
+     * read-classification cache size; default provided by reference.conf. */
+    private static final String READ_CLASSIFICATION_MAX_SIZE_KEY = "read-classification-cache-max-size";
+
     private final PolicyCacheLoader delegate;
     private final Executor enforcementCacheExecutor;
     private final NamespacePoliciesConfig namespacePoliciesConfig;
     private final long namespaceFilteredEnforcerCacheMaxSize;
+    private final long readClassificationCacheMaxSize;
     @Nullable
     private final CompletableFuture<Cache<PolicyId, Entry<PolicyEnforcer>>> cacheFuture;
 
@@ -83,6 +88,9 @@ public PolicyEnforcerCacheLoader(final PolicyCacheLoader policyCacheLoader, fina
         this.namespaceFilteredEnforcerCacheMaxSize = actorSystem.settings().config()
                 .getLong(PolicyEnforcerProvider.ENFORCER_CACHE_CONFIG_KEY + "." +
                         NAMESPACE_FILTERED_ENFORCER_MAX_SIZE_KEY);
+        this.readClassificationCacheMaxSize = actorSystem.settings().config()
+                .getLong(PolicyEnforcerProvider.ENFORCER_CACHE_CONFIG_KEY + "." +
+                        READ_CLASSIFICATION_MAX_SIZE_KEY);
         this.cacheFuture = cacheFuture;
     }
 
@@ -114,7 +122,8 @@ private CompletionStage<Entry<PolicyEnforcer>> evaluatePolicy(final Entry<Policy
             final var revision = entry.getRevision();
             final var policy = entry.getValueOrThrow();
             return PolicyEnforcer.withResolvedImportsAndNamespacePolicies(policy, policyResolver,
-                            namespacePoliciesConfig, namespaceFilteredEnforcerCacheMaxSize)
+                            namespacePoliciesConfig, namespaceFilteredEnforcerCacheMaxSize,
+                            readClassificationCacheMaxSize)
                     .thenApply(enforcer -> Entry.of(revision, enforcer));
         } else {
             return CompletableFuture.completedFuture(Entry.nonexistent());

policies/enforcement/src/main/java/org/eclipse/ditto/policies/enforcement/PolicyEnforcerCacheLoader.java

@@ -53,6 +53,12 @@ ditto.policies-enforcer-cache {
   namespace-filtered-enforcer-max-size = 100
   namespace-filtered-enforcer-max-size = ${?DITTO_POLICIES_ENFORCER_NAMESPACE_FILTERED_MAX_SIZE}
 
+  # per-enforcer memo of classifySubjects(resource, READ) results: reused by the things-service per-event
+  # read-authorization and read-grant collection instead of re-walking the policy tree per emitted ThingEvent;
+  # bounds the number of distinct resource paths cached per policy
+  read-classification-cache-max-size = 1000
+  read-classification-cache-max-size = ${?DITTO_POLICIES_ENFORCER_READ_CLASSIFICATION_MAX_SIZE}
+
   # maximum duration of inconsistency after losing a cache invalidation
   expire-after-write = 1h
   expire-after-write = ${?DITTO_POLICIES_ENFORCER_CACHE_EXPIRE_AFTER_WRITE}

policies/enforcement/src/main/resources/reference.conf

@@ -29,8 +29,10 @@
 import org.eclipse.ditto.policies.model.PolicyEntry;
 import org.eclipse.ditto.policies.model.PolicyId;
 import org.eclipse.ditto.policies.model.PoliciesResourceType;
+import org.eclipse.ditto.policies.model.ResourceKey;
 import org.eclipse.ditto.policies.model.Subject;
 import org.eclipse.ditto.policies.model.SubjectId;
+import org.eclipse.ditto.policies.model.enforcers.SubjectClassification;
 import org.junit.Test;
 
 public final class PolicyEnforcerTest {
@@ -131,6 +133,67 @@ public void forNamespaceWithNullPolicyReturnsSameInstance() {
         assertThat(enforcer.forNamespace("any.ns")).isSameAs(enforcer);
     }
 
+    @Test
+    public void classifyReadSubjectsMatchesDirectEnforcerClassification() {
+        final Policy policy = PoliciesModelFactory.newPolicyBuilder(PolicyId.of("test:policy"))
+                .set(newScopedEntry("global", "google:global-user", Collections.emptyList()))
+                .build();
+        final PolicyEnforcer policyEnforcer = PolicyEnforcer.of(policy);
+        final ResourceKey root = PoliciesResourceType.thingResource("/");
+
+        final SubjectClassification memoized = policyEnforcer.classifyReadSubjects(root);
+        final SubjectClassification direct = policyEnforcer.getEnforcer()
+                .classifySubjects(root, PoliciesModelFactory.newPermissions(Permission.READ));
+
+        assertThat(memoized.getUnrestricted()).isEqualTo(direct.getUnrestricted());
+        assertThat(memoized.getPartialOnly()).isEqualTo(direct.getPartialOnly());
+        assertThat(memoized.getEffectedGranted()).isEqualTo(direct.getEffectedGranted());
+    }
+
+    @Test
+    public void classifyReadSubjectsMemoizesRepeatedCallsForSameResource() {
+        final Policy policy = PoliciesModelFactory.newPolicyBuilder(PolicyId.of("test:policy"))
+                .set(newScopedEntry("global", "google:global-user", Collections.emptyList()))
+                .build();
+        final PolicyEnforcer policyEnforcer = PolicyEnforcer.of(policy);
+        final ResourceKey root = PoliciesResourceType.thingResource("/");
+
+        // A second call for the same resource must return the memoized instance, not re-walk the policy tree.
+        final SubjectClassification first = policyEnforcer.classifyReadSubjects(root);
+        final SubjectClassification second = policyEnforcer.classifyReadSubjects(root);
+
+        assertThat(second).isSameAs(first);
+    }
+
+    @Test
+    public void getRootResourceReadClassificationDelegatesToClassifyReadSubjects() {
+        final Policy policy = PoliciesModelFactory.newPolicyBuilder(PolicyId.of("test:policy"))
+                .set(newScopedEntry("global", "google:global-user", Collections.emptyList()))
+                .build();
+        final PolicyEnforcer policyEnforcer = PolicyEnforcer.of(policy);
+
+        assertThat(policyEnforcer.getRootResourceReadClassification())
+                .isSameAs(policyEnforcer.classifyReadSubjects(PoliciesResourceType.thingResource("/")));
+    }
+
+    @Test
+    public void classifyReadSubjectsStaysCorrectUnderBoundedCache() {
+        final Policy policy = PoliciesModelFactory.newPolicyBuilder(PolicyId.of("test:policy"))
+                .set(newScopedEntry("global", "google:global-user", Collections.emptyList()))
+                .build();
+        // Tiny bound: cache correctness must not depend on capacity — a miss simply recomputes an equal result.
+        final PolicyEnforcer bounded = PolicyEnforcer.of(policy, 100, 1);
+        final ResourceKey a = PoliciesResourceType.thingResource("/features/a");
+        final ResourceKey b = PoliciesResourceType.thingResource("/features/b");
+
+        final SubjectClassification classificationA = bounded.classifyReadSubjects(a);
+        bounded.classifyReadSubjects(b);
+        final SubjectClassification classificationAgain = bounded.classifyReadSubjects(a);
+
+        assertThat(classificationAgain.getUnrestricted()).isEqualTo(classificationA.getUnrestricted());
+        assertThat(classificationAgain.getPartialOnly()).isEqualTo(classificationA.getPartialOnly());
+    }
+
     private static PolicyEntry newScopedEntry(final String label, final String subjectId, final List<String> namespaces) {
         return PoliciesModelFactory.newPolicyEntry(label,
                 Collections.singletonList(Subject.newInstance(SubjectId.newInstance(subjectId))),

policies/enforcement/src/test/java/org/eclipse/ditto/policies/enforcement/PolicyEnforcerTest.java

@@ -167,10 +167,10 @@ private Signal<?> enforceLiveSignal(final StreamingType streamingType, final Sig
             case MESSAGES:
                 return enforceMessageCommand((MessageCommand<?, ?>) liveSignal, enforcer.getEnforcer());
             case LIVE_EVENTS:
-                return enforceLiveEvent(liveSignal, enforcer.getEnforcer());
+                return enforceLiveEvent(liveSignal, enforcer);
             case LIVE_COMMANDS:
                 ThingCommandEnforcement.authorizeByPolicyOrThrow(
-                        enforcer.getEnforcer(),
+                        enforcer,
                         (ThingCommand<?>) liveSignal,
                         partialAccessEventsEnabled);
                 final var withReadSubjects = addEffectedReadSubjectsToThingLiveSignal((ThingCommand<?>) liveSignal,
@@ -212,13 +212,14 @@ <T extends Signal<T>> T addEffectedReadSubjectsToThingLiveSignal(final Signal<T>
         return signal.setDittoHeaders(newHeaders);
     }
 
-    private Signal<?> enforceLiveEvent(final Signal<?> liveSignal, final Enforcer enforcer) {
-        if (enforcer.hasUnrestrictedPermissions(PoliciesResourceType.thingResource(liveSignal.getResourcePath()),
-                liveSignal.getDittoHeaders().getAuthorizationContext(), WRITE)) {
+    private Signal<?> enforceLiveEvent(final Signal<?> liveSignal, final PolicyEnforcer policyEnforcer) {
+        if (policyEnforcer.getEnforcer()
+                .hasUnrestrictedPermissions(PoliciesResourceType.thingResource(liveSignal.getResourcePath()),
+                        liveSignal.getDittoHeaders().getAuthorizationContext(), WRITE)) {
             LOGGER.withCorrelationId(liveSignal)
                     .info("Live Event was authorized: <{}>", liveSignal);
             return ThingCommandEnforcement.addEffectedReadSubjectsToThingSignal((ThingEvent<?>) liveSignal,
-                    enforcer,
+                    policyEnforcer,
                     partialAccessEventsEnabled);
         } else {
             LOGGER.withCorrelationId(liveSignal)