Make operator metrics MongoDB read preference and concern configurable

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.6.0  # chart version is effectively set by release-job
+version: 4.6.1  # chart version is effectively set by release-job
 appVersion: 3.9.6
 keywords:
   - iot-chart

deployment/helm/ditto/service-config/search-extension.conf.tpl

@@ -146,6 +146,28 @@ ditto {
         }
         {{- end }}
       }
+
+      {{- with .Values.thingsSearch.config.operatorMetrics.customMetricsPersistence }}
+      {{- if .readPreference }}
+      custom-metrics-persistence {
+        readPreference = "{{ .readPreference }}"
+        {{- if .readConcern }}
+        readConcern = "{{ .readConcern }}"
+        {{- end }}
+      }
+      {{- end }}
+      {{- end }}
+
+      {{- with .Values.thingsSearch.config.operatorMetrics.customAggregationMetricsPersistence }}
+      {{- if .readPreference }}
+      custom-aggregation-metrics-persistence {
+        readPreference = "{{ .readPreference }}"
+        {{- if .readConcern }}
+        readConcern = "{{ .readConcern }}"
+        {{- end }}
+      }
+      {{- end }}
+      {{- end }}
     }
   }
 }

deployment/helm/ditto/values.yaml

@@ -1729,6 +1729,19 @@ thingsSearch:
       #   # or as index key spec:
       #   # indexHint:
       #   #   "t.attributes.location": 1
+      # customMetricsPersistence optionally overrides the MongoDB read preference / read concern used for the
+      # count based "customMetrics" queries above. When unset (the default), the general query.persistence
+      # settings are used - so this can offload the periodic operator metric counts to a secondary node without
+      # affecting user-facing search/count requests.
+      # customMetricsPersistence:
+      #   readPreference: secondaryPreferred
+      #   readConcern: default
+      # customAggregationMetricsPersistence optionally overrides the MongoDB read preference / read concern used
+      # for the "customAggregationMetrics" ($group) queries above. When unset (the default), the general
+      # query.persistence settings are used. Can be configured independently from customMetricsPersistence.
+      # customAggregationMetricsPersistence:
+      #   readPreference: secondaryPreferred
+      #   readConcern: default
     # dispatchers contains tuning for the things-search service's Pekko dispatchers.
     # Defaults below match the in-jar HOCON defaults and only need overriding under sustained load.
     dispatchers:

internal/utils/persistence/src/main/java/org/eclipse/ditto/internal/utils/persistence/mongo/config/ReadConcern.java

@@ -42,6 +42,16 @@ public com.mongodb.ReadConcern getMongoReadConcern() {
         return mongoReadConcern;
     }
 
+    /**
+     * Returns the config string value of this read concern (e.g. {@code "local"}).
+     *
+     * @return the read concern config string.
+     * @since 3.9.7
+     */
+    public String getName() {
+        return name;
+    }
+
     /**
      * Tries to create a ReadConcern from the given read concern string.
      *

internal/utils/persistence/src/main/java/org/eclipse/ditto/internal/utils/persistence/mongo/config/ReadPreference.java

@@ -40,6 +40,16 @@ public com.mongodb.ReadPreference getMongoReadPreference() {
         return mongoReadPreference;
     }
 
+    /**
+     * Returns the config string value of this read preference (e.g. {@code "secondaryPreferred"}).
+     *
+     * @return the read preference config string.
+     * @since 3.9.7
+     */
+    public String getName() {
+        return name;
+    }
+
     /**
      * Tries to create a ReadPreference from the given read preference string.
      *

thingsearch/api/src/main/java/org/eclipse/ditto/thingsearch/api/commands/sudo/SudoCountThings.java

@@ -73,6 +73,14 @@ public final class SudoCountThings extends AbstractCommand<SudoCountThings>
             JsonFactory.newJsonValueFieldDefinition("indexHint", FieldType.REGULAR,
                     JsonSchemaVersion.V_2);
 
+    static final JsonFieldDefinition<String> JSON_READ_PREFERENCE =
+            JsonFactory.newStringFieldDefinition("readPreference", FieldType.REGULAR,
+                    JsonSchemaVersion.V_2);
+
+    static final JsonFieldDefinition<String> JSON_READ_CONCERN =
+            JsonFactory.newStringFieldDefinition("readConcern", FieldType.REGULAR,
+                    JsonSchemaVersion.V_2);
+
     @Nullable
     private final String filter;
 
@@ -82,8 +90,15 @@ public final class SudoCountThings extends AbstractCommand<SudoCountThings>
     @Nullable
     private final JsonValue indexHint;
 
+    @Nullable
+    private final String readPreference;
+
+    @Nullable
+    private final String readConcern;
+
     private SudoCountThings(final DittoHeaders dittoHeaders, @Nullable final String filter,
-            @Nullable final Collection<String> namespaces, @Nullable final JsonValue indexHint) {
+            @Nullable final Collection<String> namespaces, @Nullable final JsonValue indexHint,
+            @Nullable final String readPreference, @Nullable final String readConcern) {
         super(TYPE, dittoHeaders);
         this.filter = filter;
         if (namespaces != null) {
@@ -92,6 +107,8 @@ private SudoCountThings(final DittoHeaders dittoHeaders, @Nullable final String
             this.namespaces = null;
         }
         this.indexHint = indexHint;
+        this.readPreference = readPreference;
+        this.readConcern = readConcern;
     }
 
     /**
@@ -103,7 +120,7 @@ private SudoCountThings(final DittoHeaders dittoHeaders, @Nullable final String
      * @throws NullPointerException if {@code dittoHeaders} is {@code null}.
      */
     public static SudoCountThings of(@Nullable final String filter, final DittoHeaders dittoHeaders) {
-        return new SudoCountThings(dittoHeaders, filter, null, null);
+        return new SudoCountThings(dittoHeaders, filter, null, null, null, null);
     }
 
     /**
@@ -117,7 +134,7 @@ public static SudoCountThings of(@Nullable final String filter, final DittoHeade
      */
     public static SudoCountThings of(@Nullable final String filter, @Nullable final Collection<String> namespaces,
             final DittoHeaders dittoHeaders) {
-        return new SudoCountThings(dittoHeaders, filter, namespaces, null);
+        return new SudoCountThings(dittoHeaders, filter, namespaces, null, null, null);
     }
 
     /**
@@ -132,7 +149,47 @@ public static SudoCountThings of(@Nullable final String filter, @Nullable final
      */
     public static SudoCountThings of(@Nullable final String filter, @Nullable final Collection<String> namespaces,
             @Nullable final JsonValue indexHint, final DittoHeaders dittoHeaders) {
-        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint);
+        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint, null, null);
+    }
+
+    /**
+     * Returns a new instance of {@code SudoCountThings}.
+     *
+     * @param filter the optional filter string.
+     * @param namespaces the namespaces to perform the count in.
+     * @param indexHint the optional index hint for the MongoDB query.
+     * @param readPreference the optional MongoDB read preference to use for this count query (e.g.
+     * {@code "secondaryPreferred"}); when {@code null} the persistence default read preference is used.
+     * @param dittoHeaders the headers of the command.
+     * @return a new command for counting Things.
+     * @throws NullPointerException if {@code dittoHeaders} is {@code null}.
+     * @since 3.9.7
+     */
+    public static SudoCountThings of(@Nullable final String filter, @Nullable final Collection<String> namespaces,
+            @Nullable final JsonValue indexHint, @Nullable final String readPreference,
+            final DittoHeaders dittoHeaders) {
+        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint, readPreference, null);
+    }
+
+    /**
+     * Returns a new instance of {@code SudoCountThings}.
+     *
+     * @param filter the optional filter string.
+     * @param namespaces the namespaces to perform the count in.
+     * @param indexHint the optional index hint for the MongoDB query.
+     * @param readPreference the optional MongoDB read preference to use for this count query (e.g.
+     * {@code "secondaryPreferred"}); when {@code null} the persistence default read preference is used.
+     * @param readConcern the optional MongoDB read concern to use for this count query (e.g. {@code "local"});
+     * when {@code null} the persistence default read concern is used.
+     * @param dittoHeaders the headers of the command.
+     * @return a new command for counting Things.
+     * @throws NullPointerException if {@code dittoHeaders} is {@code null}.
+     * @since 3.9.7
+     */
+    public static SudoCountThings of(@Nullable final String filter, @Nullable final Collection<String> namespaces,
+            @Nullable final JsonValue indexHint, @Nullable final String readPreference,
+            @Nullable final String readConcern, final DittoHeaders dittoHeaders) {
+        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint, readPreference, readConcern);
     }
 
     /**
@@ -143,7 +200,7 @@ public static SudoCountThings of(@Nullable final String filter, @Nullable final
      * @throws NullPointerException if any argument is {@code null}.
      */
     public static SudoCountThings of(final DittoHeaders dittoHeaders) {
-        return new SudoCountThings(dittoHeaders, null, null, null);
+        return new SudoCountThings(dittoHeaders, null, null, null, null, null);
     }
 
     /**
@@ -186,7 +243,12 @@ public static SudoCountThings fromJson(final JsonObject jsonObject, final DittoH
 
             final JsonValue extractedIndexHint = jsonObject.getValue(JSON_INDEX_HINT).orElse(null);
 
-            return new SudoCountThings(dittoHeaders, extractedFilter, extractedNamespaces, extractedIndexHint);
+            final String extractedReadPreference = jsonObject.getValue(JSON_READ_PREFERENCE).orElse(null);
+
+            final String extractedReadConcern = jsonObject.getValue(JSON_READ_CONCERN).orElse(null);
+
+            return new SudoCountThings(dittoHeaders, extractedFilter, extractedNamespaces, extractedIndexHint,
+                    extractedReadPreference, extractedReadConcern);
         });
     }
 
@@ -217,6 +279,26 @@ public Optional<JsonValue> getIndexHint() {
         return Optional.ofNullable(indexHint);
     }
 
+    /**
+     * Get the optional MongoDB read preference override for this count query.
+     *
+     * @return the optional read preference (e.g. {@code "secondaryPreferred"}).
+     * @since 3.9.7
+     */
+    public Optional<String> getReadPreference() {
+        return Optional.ofNullable(readPreference);
+    }
+
+    /**
+     * Get the optional MongoDB read concern override for this count query.
+     *
+     * @return the optional read concern (e.g. {@code "local"}).
+     * @since 3.9.7
+     */
+    public Optional<String> getReadConcern() {
+        return Optional.ofNullable(readConcern);
+    }
+
     @Override
     protected void appendPayload(final JsonObjectBuilder jsonObjectBuilder, final JsonSchemaVersion schemaVersion,
             final Predicate<JsonField> thePredicate) {
@@ -227,6 +309,8 @@ protected void appendPayload(final JsonObjectBuilder jsonObjectBuilder, final Js
                 .map(JsonValue::of)
                 .collect(JsonCollectors.valuesToArray()), predicate));
         getIndexHint().ifPresent(hint -> jsonObjectBuilder.set(JSON_INDEX_HINT, hint, predicate));
+        getReadPreference().ifPresent(rp -> jsonObjectBuilder.set(JSON_READ_PREFERENCE, rp, predicate));
+        getReadConcern().ifPresent(rc -> jsonObjectBuilder.set(JSON_READ_CONCERN, rc, predicate));
     }
 
     @Override
@@ -236,7 +320,7 @@ public Category getCategory() {
 
     @Override
     public SudoCountThings setDittoHeaders(final DittoHeaders dittoHeaders) {
-        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint);
+        return new SudoCountThings(dittoHeaders, filter, namespaces, indexHint, readPreference, readConcern);
     }
 
     @Override
@@ -250,12 +334,14 @@ public boolean equals(@Nullable final Object o) {
         final SudoCountThings that = (SudoCountThings) o;
         return Objects.equals(filter, that.filter) &&
                 Objects.equals(namespaces, that.namespaces) &&
-                Objects.equals(indexHint, that.indexHint);
+                Objects.equals(indexHint, that.indexHint) &&
+                Objects.equals(readPreference, that.readPreference) &&
+                Objects.equals(readConcern, that.readConcern);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(super.hashCode(), filter, namespaces, indexHint);
+        return Objects.hash(super.hashCode(), filter, namespaces, indexHint, readPreference, readConcern);
     }
 
     @Override
@@ -264,6 +350,8 @@ public String toString() {
                 "filter='" + filter + "'" +
                 ", namespaces=" + namespaces +
                 ", indexHint=" + indexHint +
+                ", readPreference=" + readPreference +
+                ", readConcern=" + readConcern +
                 "]";
     }
 

thingsearch/api/src/test/java/org/eclipse/ditto/thingsearch/api/commands/sudo/SudoCountThingsTest.java

@@ -135,4 +135,42 @@ public void setDittoHeadersPreservesIndexHint() {
         assertThat(withNewHeaders.getIndexHint()).contains(JsonValue.of("my_index"));
     }
 
+    @Test
+    public void toJsonWithReadPreferenceAndReadConcern() {
+        final SudoCountThings command = SudoCountThings.of(
+                KNOWN_FILTER_STR, List.of("ns1"), JsonValue.of("my_index"), "secondaryPreferred", "local",
+                DittoHeaders.empty());
+
+        final String json = command.toJsonString();
+        final SudoCountThings deserialized = SudoCountThings.fromJson(json, DittoHeaders.empty());
+
+        assertThat(deserialized.getFilter()).contains(KNOWN_FILTER_STR);
+        assertThat(deserialized.getIndexHint()).contains(JsonValue.of("my_index"));
+        assertThat(deserialized.getReadPreference()).contains("secondaryPreferred");
+        assertThat(deserialized.getReadConcern()).contains("local");
+    }
+
+    @Test
+    public void toJsonWithoutReadPreferenceOrReadConcern() {
+        final SudoCountThings command = SudoCountThings.of(KNOWN_FILTER_STR, DittoHeaders.empty());
+
+        final String json = command.toJsonString();
+        final SudoCountThings deserialized = SudoCountThings.fromJson(json, DittoHeaders.empty());
+
+        assertThat(deserialized.getReadPreference()).isEmpty();
+        assertThat(deserialized.getReadConcern()).isEmpty();
+    }
+
+    @Test
+    public void setDittoHeadersPreservesReadPreferenceAndReadConcern() {
+        final SudoCountThings command = SudoCountThings.of(
+                KNOWN_FILTER_STR, null, null, "secondaryPreferred", "local", DittoHeaders.empty());
+
+        final SudoCountThings withNewHeaders = command.setDittoHeaders(
+                DittoHeaders.newBuilder().correlationId("test").build());
+
+        assertThat(withNewHeaders.getReadPreference()).contains("secondaryPreferred");
+        assertThat(withNewHeaders.getReadConcern()).contains("local");
+    }
+
 }