#2125 add throttling option for search index update after policy update

hu-ahmed · hu-ahmed · commit 4d501aedcb7c · 2025-02-18T11:35:05.000+01:00
diff --git a/deployment/helm/ditto/Chart.yaml b/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: 3.6.11  # chart version is effectively set by release-job
+version: 3.6.12  # chart version is effectively set by release-job
 appVersion: 3.6.11
 keywords:
   - iot-chart
diff --git a/deployment/helm/ditto/templates/thingssearch-deployment.yaml b/deployment/helm/ditto/templates/thingssearch-deployment.yaml
@@ -201,6 +201,12 @@ spec:
               value: "{{ .Values.thingsSearch.config.mongodb.updaterPersistenceReadConcern }}"
             - name: UPDATER_PERSISTENCE_MONGO_DB_READ_PREFERENCE
               value: "{{ .Values.thingsSearch.config.mongodb.updaterPersistenceReadPreference }}"
+            - name: POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_ENABLED
+              value: "{{ .Values.thingsSearch.config.mongodb.PolicyModificationCausedSearchIndexUpdateThrottling.enabled }}"
+            - name: POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_INTERVAL
+              value: "{{ .Values.thingsSearch.config.mongodb.PolicyModificationCausedSearchIndexUpdateThrottling.interval }}"
+            - name: POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_LIMIT
+              value: "{{ .Values.thingsSearch.config.mongodb.PolicyModificationCausedSearchIndexUpdateThrottling.limit }}"
             - name: THINGS_SEARCH_UPDATER_STREAM_PERSISTENCE_WITH_ACKS_WRITE_CONCERN
               value: "{{ .Values.thingsSearch.config.mongodb.searchWithAcksWriteConcern }}"
             - name: THINGS_SEARCH_UPDATER_STREAM_POLICY_CACHE_SIZE
diff --git a/deployment/helm/ditto/values.yaml b/deployment/helm/ditto/values.yaml
@@ -1376,6 +1376,14 @@ thingsSearch:
       updaterPersistenceReadConcern: "local"
       # updaterPersistenceReadPreference configures the MongoDB read preference for the "ThingUpdater"
       updaterPersistenceReadPreference: "primaryPreferred"
+      # PolicyModificationCausedSearchIndexUpdateThrottling contains throttling configuration for the search Index update after a policy update
+      PolicyModificationCausedSearchIndexUpdateThrottling:
+        # enabled defines whether throttling should be applied for search Index update after a policy update.
+        enabled: false
+        # The time window within which the throttling limit applies.
+        interval: 1s
+        # The maximum number of updates allowed within each throttling interval.
+        limit: 100
     # updater contains configuration for the "Things Updater" of things-search service
     updater:
       # activityCheckInterval configures to keep thing updaters for that amount of time in memory when no update did happen:
diff --git a/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/common/config/DefaultSearchPersistenceConfig.java b/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/common/config/DefaultSearchPersistenceConfig.java
@@ -17,6 +17,7 @@
 
 import javax.annotation.concurrent.Immutable;
 
+import org.eclipse.ditto.base.service.config.ThrottlingConfig;
 import org.eclipse.ditto.internal.utils.config.ConfigWithFallback;
 import org.eclipse.ditto.internal.utils.config.DittoConfigError;
 import org.eclipse.ditto.internal.utils.persistence.mongo.config.MongoDbConfig;
@@ -35,6 +36,7 @@ public final class DefaultSearchPersistenceConfig implements SearchPersistenceCo
 
     private final ReadPreference readPreference;
     private final ReadConcern readConcern;
+    private final ThrottlingConfig policyModificationCausedSearchIndexUpdateThrottling;
 
 
     private DefaultSearchPersistenceConfig(final ConfigWithFallback config) {
@@ -56,6 +58,7 @@ private DefaultSearchPersistenceConfig(final ConfigWithFallback config) {
                                     readConcernString);
                     return new DittoConfigError(msg);
                 });
+        policyModificationCausedSearchIndexUpdateThrottling = ThrottlingConfig.of(config);
     }
 
     /**
@@ -80,6 +83,11 @@ public ReadConcern readConcern() {
         return readConcern;
     }
 
+    @Override
+    public ThrottlingConfig getPolicyModificationCausedSearchIndexUpdateThrottling() {
+        return policyModificationCausedSearchIndexUpdateThrottling;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -89,19 +97,23 @@ public boolean equals(final Object o) {
             return false;
         }
         final DefaultSearchPersistenceConfig that = (DefaultSearchPersistenceConfig) o;
-        return readPreference == that.readPreference && readConcern == that.readConcern;
+        return readPreference == that.readPreference
+                && readConcern == that.readConcern
+                && Objects.equals(
+                policyModificationCausedSearchIndexUpdateThrottling, that.policyModificationCausedSearchIndexUpdateThrottling);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(readPreference, readConcern);
+        return Objects.hash(readPreference, readConcern, policyModificationCausedSearchIndexUpdateThrottling);
     }
 
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
                 "readPreference=" + readPreference +
                 ", readConcern=" + readConcern +
+                ", policyModificationCausedSearchIndexUpdateThrottling=" + policyModificationCausedSearchIndexUpdateThrottling +
                 "]";
     }
 }
diff --git a/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/common/config/SearchPersistenceConfig.java b/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/common/config/SearchPersistenceConfig.java
@@ -14,6 +14,7 @@
 
 import javax.annotation.concurrent.Immutable;
 
+import org.eclipse.ditto.base.service.config.ThrottlingConfig;
 import org.eclipse.ditto.internal.utils.config.KnownConfigValue;
 import org.eclipse.ditto.internal.utils.persistence.mongo.config.ReadConcern;
 import org.eclipse.ditto.internal.utils.persistence.mongo.config.ReadPreference;
@@ -38,6 +39,14 @@ public interface SearchPersistenceConfig {
      */
     ReadConcern readConcern();
 
+    /**
+     * Returns the {@code ThrottlingConfig} for the policy update after search Index update.
+     *
+     * @return the policy update ThrottlingConfig.
+     * @since 3.7.0
+     */
+    ThrottlingConfig getPolicyModificationCausedSearchIndexUpdateThrottling();
+
     /**
      * An enumeration of known config path expressions and their associated default values for {@code SearchPersistenceConfig}.
      */
diff --git a/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/write/impl/MongoThingsSearchUpdaterPersistence.java b/thingsearch/service/src/main/java/org/eclipse/ditto/thingsearch/service/persistence/write/impl/MongoThingsSearchUpdaterPersistence.java
@@ -31,6 +31,7 @@
 import org.bson.BsonString;
 import org.bson.Document;
 import org.bson.conversions.Bson;
+import org.eclipse.ditto.base.service.config.ThrottlingConfig;
 import org.eclipse.ditto.policies.api.PolicyTag;
 import org.eclipse.ditto.policies.model.PolicyId;
 import org.eclipse.ditto.things.model.ThingId;
@@ -58,8 +59,11 @@ public final class MongoThingsSearchUpdaterPersistence implements ThingsSearchUp
 
     private final MongoCollection<Document> collection;
 
+    private final ThrottlingConfig PolicyModificationCausedSearchIndexUpdateThrottling;
+
     private MongoThingsSearchUpdaterPersistence(final MongoDatabase database,
-            final SearchPersistenceConfig updaterPersistenceConfig) {
+                                                final SearchPersistenceConfig updaterPersistenceConfig) {
+        this.PolicyModificationCausedSearchIndexUpdateThrottling = updaterPersistenceConfig.getPolicyModificationCausedSearchIndexUpdateThrottling();
 
         collection = database.getCollection(PersistenceConstants.THINGS_COLLECTION_NAME)
                 .withReadConcern(updaterPersistenceConfig.readConcern().getMongoReadConcern())
@@ -106,18 +110,25 @@ public Source<PolicyReferenceTag, NotUsed> getPolicyReferenceTags(final Map<Poli
                         .append(PersistenceConstants.FIELD_POLICY_ID, new BsonInt32(1))
                         .append(PersistenceConstants.FIELD_REFERENCED_POLICIES, new BsonInt32(1)));
 
-        return Source.fromPublisher(publisher)
-                .mapConcat(doc -> {
-                    final ThingId thingId = ThingId.of(doc.getString(PersistenceConstants.FIELD_ID));
-                    final Collection<PolicyId> referencedPolicyIds = referencedPolicyIds(doc);
-                    return referencedPolicyIds.stream()
-                            .map(referencedPolicyId -> Optional.ofNullable(policyRevisions.get(referencedPolicyId))
-                                    .map(revision -> PolicyTag.of(referencedPolicyId, revision))
-                                    .map(policyTag -> PolicyReferenceTag.of(thingId, policyTag))
-                                    .orElse(null))
-                            .filter(Objects::nonNull)
-                            .toList();
-                });
+
+        final Source<Document, NotUsed> throttledSource = PolicyModificationCausedSearchIndexUpdateThrottling.isEnabled()
+                ? Source.fromPublisher(publisher).throttle(
+                    PolicyModificationCausedSearchIndexUpdateThrottling.getLimit(),
+                    PolicyModificationCausedSearchIndexUpdateThrottling.getInterval()
+                )
+                : Source.fromPublisher(publisher);
+
+        return throttledSource.mapConcat(doc -> {
+            final ThingId thingId = ThingId.of(doc.getString(PersistenceConstants.FIELD_ID));
+            final Collection<PolicyId> referencedPolicyIds = referencedPolicyIds(doc);
+            return referencedPolicyIds.stream()
+                    .map(referencedPolicyId -> Optional.ofNullable(policyRevisions.get(referencedPolicyId))
+                            .map(revision -> PolicyTag.of(referencedPolicyId, revision))
+                            .map(policyTag -> PolicyReferenceTag.of(thingId, policyTag))
+                            .orElse(null))
+                    .filter(Objects::nonNull)
+                    .toList();
+        });
     }
 
     private Collection<PolicyId> referencedPolicyIds(final Document doc) {
diff --git a/thingsearch/service/src/main/resources/search.conf b/thingsearch/service/src/main/resources/search.conf
@@ -299,6 +299,19 @@ ditto {
         # read concern is one of: default, local, majority, linearizable, snapshot, available
         readConcern = ${ditto.mongodb.options.readConcern}
         readConcern = ${?UPDATER_PERSISTENCE_MONGO_DB_READ_CONCERN}
+        # PolicyModificationCausedSearchIndexUpdateThrottling contains throttling configuration for the search Index update after a policy update
+        policyModificationCausedSearchIndexUpdateThrottling {
+        # enabled defines whether throttling should be applied for search Index update after a policy update.
+          enabled = false
+          enabled = ${?POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_ENABLED}
+          # The interval at which updates are throttled (e.g., every 1 second)
+          interval = 1s
+          interval = ${?POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_INTERVAL}
+
+          # Maximum number of updates allowed per interval (e.g., 100 updates per second)
+          limit = 100
+          limit = ${?POLICY_MODIFICATION_CAUSED_SEARCH_INDEX_UPDATE_THROTTLING_LIMIT}
+        }
       }
     }
 
diff --git a/thingsearch/service/src/test/java/org/eclipse/ditto/thingsearch/service/common/config/DefaultSearchPersistenceConfigTest.java b/thingsearch/service/src/test/java/org/eclipse/ditto/thingsearch/service/common/config/DefaultSearchPersistenceConfigTest.java
@@ -13,6 +13,7 @@
 package org.eclipse.ditto.thingsearch.service.common.config;
 
 import org.assertj.core.api.JUnitSoftAssertions;
+import org.eclipse.ditto.base.service.config.ThrottlingConfig;
 import org.eclipse.ditto.internal.utils.persistence.mongo.config.ReadConcern;
 import org.eclipse.ditto.internal.utils.persistence.mongo.config.ReadPreference;
 import org.junit.BeforeClass;
@@ -62,6 +63,10 @@ public void underTestReturnsDefaultValuesIfBaseConfigWasEmpty() {
                 .isEqualTo(ReadPreference.ofReadPreference(
                         (String) SearchPersistenceConfig.ConfigValue.READ_PREFERENCE.getDefaultValue())
                         .orElseThrow());
+
+        softly.assertThat(underTest.getPolicyModificationCausedSearchIndexUpdateThrottling())
+                .as("Throttling Config should have default values")
+                .isEqualTo(ThrottlingConfig.of(ConfigFactory.empty()));
     }
 
     @Test
@@ -75,6 +80,10 @@ public void underTestReturnsValuesOfConfigFile() {
         softly.assertThat(underTest.readPreference())
                 .as(SearchPersistenceConfig.ConfigValue.READ_PREFERENCE.getConfigPath())
                 .isEqualTo(ReadPreference.SECONDARY_PREFERRED);
+
+        softly.assertThat(underTest.getPolicyModificationCausedSearchIndexUpdateThrottling())
+                .as("Throttling Config should match the configuration file")
+                .isNotNull();
     }
 
 }
diff --git a/thingsearch/service/src/test/resources/updater-persistence-test.conf b/thingsearch/service/src/test/resources/updater-persistence-test.conf
@@ -3,4 +3,10 @@ persistence {
   readPreference = secondaryPreferred
   # read concern is one of: default, local, majority, linearizable, snapshot, available
   readConcern = available
+
+  policyModificationCausedSearchIndexUpdateThrottling {
+    enabled = false
+    interval = 1s
+    limit = 100
+  }
 }
