Add PreEnforcer Extension

Signed-off-by: David Schwilk <david.schwilk@bosch.io>
eclipse-ditto · May 20, 2022 · f51b321 · f51b321
1 parent 1c1eae9
commit f51b321
Show file tree

Hide file tree

Showing 29 changed files with 229 additions and 99 deletions.
diff --git a/base/service/src/main/java/org/eclipse/ditto/base/service/actors/DittoRootActor.java b/base/service/src/main/java/org/eclipse/ditto/base/service/actors/DittoRootActor.java
@@ -64,36 +64,36 @@ public abstract class DittoRootActor extends AbstractActor {
                 e.printStackTrace(pw);
 
                 log.warning("Illegal Argument in child actor: {}", sw.toString());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(IllegalStateException.class, e -> {
                 log.warning("Illegal State in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(IndexOutOfBoundsException.class, e -> {
                 log.warning("IndexOutOfBounds in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(NoSuchElementException.class, e -> {
                 log.warning("NoSuchElement in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(AskTimeoutException.class, e -> {
                 log.warning("AskTimeoutException in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(ConnectException.class, e -> {
                 log.warning("ConnectException in child actor: {}", e.getMessage());
                 return restartChild();
             }).match(InvalidActorNameException.class, e -> {
                 log.warning("InvalidActorNameException in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(ActorInitializationException.class, e -> {
                 log.error(e, "ActorInitializationException in child actor: {}", e.getMessage());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.stop();
+                return SupervisorStrategy.stop();
             }).match(ActorKilledException.class, e -> {
                 log.error(e, "ActorKilledException in child actor: {}", e.message());
                 return restartChild();
             }).match(DittoRuntimeException.class, e -> {
                 log.error(e,
                         "DittoRuntimeException '{}' should not be escalated to ConnectivityRootActor. Simply resuming Actor.",
                         e.getErrorCode());
-                return (SupervisorStrategy.Directive) SupervisorStrategy.resume();
+                return SupervisorStrategy.resume();
             }).match(Throwable.class, e -> {
                 log.error(e, "Escalating above root actor!");
                 return (SupervisorStrategy.Directive) SupervisorStrategy.escalate();
@@ -151,7 +151,7 @@ protected ActorRef startChildActor(final String actorName, final Props props) {
      */
     protected SupervisorStrategy.Directive restartChild() {
         log.info("Restarting child ...");
-        return (SupervisorStrategy.Directive) SupervisorStrategy.restart();
+        return SupervisorStrategy.restart();
     }
 
     private void startChildActor(final StartChildActor startChildActor) {

diff --git a/...y/service/src/main/java/org/eclipse/ditto/connectivity/service/ConnectivityRootActor.java b/...y/service/src/main/java/org/eclipse/ditto/connectivity/service/ConnectivityRootActor.java
@@ -12,8 +12,6 @@
  */
 package org.eclipse.ditto.connectivity.service;
 
-import java.util.concurrent.CompletableFuture;
-
 import javax.jms.JMSRuntimeException;
 import javax.naming.NamingException;
 
@@ -43,7 +41,6 @@
 import org.eclipse.ditto.internal.utils.persistentactors.PersistencePingActor;
 import org.eclipse.ditto.internal.utils.persistentactors.cleanup.PersistenceCleanupActor;
 import org.eclipse.ditto.internal.utils.pubsub.DittoProtocolSub;
-import org.eclipse.ditto.policies.enforcement.PreEnforcer;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSystem;
@@ -125,12 +122,7 @@ private ConnectivityRootActor(final ConnectivityConfig connectivityConfig,
     private static Props getConnectivitySupervisorActorProps(final ActorRef pubSubMediator,
             final ActorRef proxyActor) {
 
-        return ConnectionSupervisorActor.props(proxyActor, pubSubMediator, providePreEnforcer());
-    }
-
-    private static PreEnforcer providePreEnforcer() {
-        return CompletableFuture::completedStage;
-        // TODO CR-11297 provide extension mechanism here
+        return ConnectionSupervisorActor.props(proxyActor, pubSubMediator);
     }
 
     /**

diff --git a/...g/eclipse/ditto/connectivity/service/messaging/persistence/ConnectionSupervisorActor.java b/...g/eclipse/ditto/connectivity/service/messaging/persistence/ConnectionSupervisorActor.java
@@ -34,7 +34,6 @@
 import org.eclipse.ditto.connectivity.service.enforcement.ConnectivityCommandEnforcement;
 import org.eclipse.ditto.internal.utils.config.DefaultScopedConfig;
 import org.eclipse.ditto.internal.utils.persistentactors.AbstractPersistenceSupervisor;
-import org.eclipse.ditto.policies.enforcement.PreEnforcer;
 
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigFactory;
@@ -76,12 +75,8 @@ public final class ConnectionSupervisorActor
     private boolean isRegisteredForConnectivityConfigChanges = false;
 
     @SuppressWarnings("unused")
-    private ConnectionSupervisorActor(
-            final ActorRef proxyActor,
-            final ActorRef pubSubMediator,
-            final PreEnforcer<ConnectivityCommand<?>> preEnforcer) {
-
-        super(null, preEnforcer);
+    private ConnectionSupervisorActor(final ActorRef proxyActor, final ActorRef pubSubMediator) {
+        super(null);
         this.proxyActor = proxyActor;
         this.pubSubMediator = pubSubMediator;
     }
@@ -95,15 +90,13 @@ private ConnectionSupervisorActor(
      *
      * @param proxyActor the actor used to send signals into the ditto cluster..
      * @param pubSubMediator pub-sub-mediator for the shutdown behavior.
-     * @param preEnforcer the PreEnforcer to apply as extension mechanism of the enforcement.
      * @return the {@link Props} to create this actor.
      */
     public static Props props(final ActorRef proxyActor,
-            final ActorRef pubSubMediator,
-            final PreEnforcer<ConnectivityCommand<?>> preEnforcer) {
+            final ActorRef pubSubMediator) {
 
         return Props.create(ConnectionSupervisorActor.class, proxyActor,
-                 pubSubMediator, preEnforcer);
+                 pubSubMediator);
     }
 
     @Override

diff --git a/connectivity/service/src/main/resources/connectivity.conf b/connectivity/service/src/main/resources/connectivity.conf
@@ -3,6 +3,7 @@ ditto {
   signal-transformer =  "org.eclipse.ditto.edge.api.dispatching.NoOpSignalTransformer"
   root-child-actor-starter = "org.eclipse.ditto.base.service.NoOpRootChildActorStarter"
   root-actor-starter = "org.eclipse.ditto.base.service.NoOpRootActorStarter"
+  pre-enforcer-provider = "org.eclipse.ditto.policies.enforcement.DefaultPreEnforcerProvider"
 
   mongodb {
     database = "connectivity"
@@ -931,7 +932,8 @@ akka {
     }
 
     roles = [
-      "connectivity"
+      "connectivity",
+      "blocked-namespaces-aware"
     ]
   }
   coordinated-shutdown {
@@ -1066,4 +1068,19 @@ kafka-producer-dispatcher {
   executor = "thread-pool-executor"
 }
 
+blocked-namespaces-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    # Min number of threads to cap factor-based parallelism number to
+    parallelism-min = 4
+    # Parallelism (threads) ... ceil(available processors * factor)
+    parallelism-factor = 3.0
+    # Max number of threads to cap factor-based parallelism number to
+    parallelism-max = 32
+    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
+  }
+  throughput = 5
+}
+
 include "connectivity-extension"
diff --git a/...service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/TestConstants.java b/...service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/TestConstants.java
@@ -918,8 +918,7 @@ public static ActorRef createConnectionSupervisorActor(final ConnectionId connec
             final ActorRef pubSubMediator) {
 
         final Props props =
-                ConnectionSupervisorActor.props(proxyActor, pubSubMediator,
-                        CompletableFuture::completedStage);
+                ConnectionSupervisorActor.props(proxyActor, pubSubMediator);
 
         final Props shardRegionMockProps = Props.create(ShardRegionMockActor.class, props, connectionId.toString());
 

diff --git a/...to/connectivity/service/messaging/persistence/ConnectionPersistenceOperationsActorIT.java b/...to/connectivity/service/messaging/persistence/ConnectionPersistenceOperationsActorIT.java
@@ -13,7 +13,6 @@
 package org.eclipse.ditto.connectivity.service.messaging.persistence;
 
 import java.util.Collections;
-import java.util.concurrent.CompletableFuture;
 
 import org.eclipse.ditto.base.model.auth.AuthorizationContext;
 import org.eclipse.ditto.base.model.auth.AuthorizationSubject;
@@ -125,7 +124,7 @@ protected ActorRef startEntityActor(final ActorSystem system, final ActorRef pub
         // essentially never restart
         final TestProbe proxyActorProbe = new TestProbe(system, "proxyActor");
         final Props props =
-                ConnectionSupervisorActor.props(proxyActorProbe.ref(), pubSubMediator, CompletableFuture::completedStage);
+                ConnectionSupervisorActor.props(proxyActorProbe.ref(), pubSubMediator);
 
         return system.actorOf(props, String.valueOf(id));
     }

diff --git a/connectivity/service/src/test/resources/test.conf b/connectivity/service/src/test/resources/test.conf
@@ -20,6 +20,7 @@ akka.cluster.roles = ["thing-event-aware", "live-signal-aware", "acks-aware", "p
 
 ditto {
   mapping-strategy.implementation = "org.eclipse.ditto.connectivity.api.ConnectivityMappingStrategies"
+  pre-enforcer-provider = "org.eclipse.ditto.policies.enforcement.DefaultPreEnforcerProvider"
 
   signal-enrichment = {
     caching-signal-enrichment-facade.provider = org.eclipse.ditto.internal.models.signalenrichment.DittoCachingSignalEnrichmentFacadeProvider
@@ -538,3 +539,18 @@ kafka-producer-dispatcher {
   type = PinnedDispatcher
   executor = "thread-pool-executor"
 }
+
+blocked-namespaces-dispatcher {
+  type = Dispatcher
+  executor = "fork-join-executor"
+  fork-join-executor {
+    # Min number of threads to cap factor-based parallelism number to
+    parallelism-min = 4
+    # Parallelism (threads) ... ceil(available processors * factor)
+    parallelism-factor = 3.0
+    # Max number of threads to cap factor-based parallelism number to
+    parallelism-max = 32
+    parallelism-max = ${?DEFAULT_DISPATCHER_PARALLELISM_MAX}
+  }
+  throughput = 5
+}
diff --git a/...java/org/eclipse/ditto/internal/utils/persistentactors/AbstractPersistenceSupervisor.java b/...java/org/eclipse/ditto/internal/utils/persistentactors/AbstractPersistenceSupervisor.java
@@ -42,6 +42,7 @@
 import org.eclipse.ditto.policies.enforcement.CreationRestrictionEnforcer;
 import org.eclipse.ditto.policies.enforcement.DefaultCreationRestrictionEnforcer;
 import org.eclipse.ditto.policies.enforcement.PreEnforcer;
+import org.eclipse.ditto.policies.enforcement.PreEnforcerProvider;
 import org.eclipse.ditto.policies.enforcement.config.DefaultEntityCreationConfig;
 
 import akka.actor.ActorRef;
@@ -92,20 +93,18 @@ public abstract class AbstractPersistenceSupervisor<E extends EntityId, S extend
     private ExponentialBackOff backOff;
     private boolean waitingForStopBeforeRestart = false;
 
-    protected AbstractPersistenceSupervisor(@Nullable final BlockedNamespaces blockedNamespaces,
-            final PreEnforcer<S> preEnforcer) {
-        this(null, null, blockedNamespaces, preEnforcer);
+    protected AbstractPersistenceSupervisor(@Nullable final BlockedNamespaces blockedNamespaces) {
+        this(null, null, blockedNamespaces);
     }
 
     protected AbstractPersistenceSupervisor(@Nullable final ActorRef persistenceActorChild,
             @Nullable final ActorRef enforcerChild,
-            @Nullable final BlockedNamespaces blockedNamespaces,
-            final PreEnforcer<S> preEnforcer) {
+            @Nullable final BlockedNamespaces blockedNamespaces) {
 
         this.persistenceActorChild = persistenceActorChild;
         this.enforcerChild = enforcerChild;
         this.blockedNamespaces = blockedNamespaces;
-        this.preEnforcer = preEnforcer;
+        preEnforcer = PreEnforcerProvider.get(getContext().getSystem()).getPreEnforcer();
         exponentialBackOffConfig = getExponentialBackOffConfig();
         backOff = ExponentialBackOff.initial(exponentialBackOffConfig);
         creationRestrictionEnforcer = DefaultCreationRestrictionEnforcer.of(

diff --git a/policies/enforcement/pom.xml b/policies/enforcement/pom.xml
@@ -25,6 +25,10 @@
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-base-model</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.ditto</groupId>
+            <artifactId>ditto-base-service</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-policies-model</artifactId>
@@ -119,6 +123,10 @@
             <artifactId>logback-classic</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.ditto</groupId>
+            <artifactId>ditto-internal-utils-namespaces</artifactId>
+        </dependency>
     </dependencies>
 
     <parent>

diff --git a/...ment/src/main/java/org/eclipse/ditto/policies/enforcement/DefaultPreEnforcerProvider.java b/...ment/src/main/java/org/eclipse/ditto/policies/enforcement/DefaultPreEnforcerProvider.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.policies.enforcement;
+
+import org.eclipse.ditto.base.model.headers.DittoHeadersSettable;
+import org.eclipse.ditto.internal.utils.namespaces.BlockNamespaceBehavior;
+import org.eclipse.ditto.internal.utils.namespaces.BlockedNamespaces;
+import org.eclipse.ditto.policies.enforcement.placeholders.PlaceholderSubstitution;
+import org.eclipse.ditto.policies.enforcement.validators.CommandWithOptionalEntityValidator;
+
+import akka.actor.ActorSystem;
+
+public class DefaultPreEnforcerProvider implements PreEnforcerProvider{
+
+    private final ActorSystem actorSystem;
+
+    public DefaultPreEnforcerProvider(final ActorSystem actorSystem) {
+        this.actorSystem = actorSystem;
+    }
+
+    @Override
+    public <T extends DittoHeadersSettable<?>> PreEnforcer<T> getPreEnforcer() {
+        return dittoHeadersSettable ->
+                BlockNamespaceBehavior.of(BlockedNamespaces.of(actorSystem))
+                        .block(dittoHeadersSettable)
+                        .thenApply(CommandWithOptionalEntityValidator.createInstance())
+                        .thenApply(PreEnforcer::setOriginatorHeader)
+                        .thenCompose(PlaceholderSubstitution.newInstance());
+    }
+}
diff --git a/...enforcement/src/main/java/org/eclipse/ditto/policies/enforcement/PreEnforcerProvider.java b/...enforcement/src/main/java/org/eclipse/ditto/policies/enforcement/PreEnforcerProvider.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.policies.enforcement;
+
+import static org.eclipse.ditto.base.model.common.ConditionChecker.checkNotNull;
+
+import java.util.List;
+
+import org.eclipse.ditto.base.model.headers.DittoHeadersSettable;
+import org.eclipse.ditto.base.service.DittoExtensionPoint;
+import org.eclipse.ditto.internal.utils.akka.AkkaClassLoader;
+import org.eclipse.ditto.internal.utils.config.DefaultScopedConfig;
+
+import akka.actor.ActorSystem;
+
+/**
+ * Extension to provide the pre-enforcer for a service.
+ *
+ * @since 3.0.0
+ */
+public interface PreEnforcerProvider extends DittoExtensionPoint {
+
+    static final String CONFIG_PATH = "pre-enforcer-provider";
+
+    /**
+     * Gets the pre-enforcer.
+     */
+    <T extends DittoHeadersSettable<?>> PreEnforcer<T> getPreEnforcer();
+
+    /**
+     * Loads the implementation of {@code PreEnforcerProvider} which is configured for the
+     * {@code ActorSystem}.
+     *
+     * @param actorSystem the actorSystem in which the {@code PreEnforcerProvider} should be loaded.
+     * @return the {@code PreEnforcerProvider} implementation.
+     * @throws NullPointerException if {@code actorSystem} is {@code null}.
+     */
+    static PreEnforcerProvider get(final ActorSystem actorSystem) {
+        checkNotNull(actorSystem, "actorSystem");
+        final DefaultScopedConfig dittoScoped = DefaultScopedConfig.dittoScoped(actorSystem.settings().config());
+        final var implementation = dittoScoped.getString(CONFIG_PATH);
+
+        return AkkaClassLoader.instantiate(actorSystem, PreEnforcerProvider.class,
+                implementation,
+                List.of(ActorSystem.class),
+                List.of(actorSystem));
+    }
+}