Added version checks to streaming functions

peteralfonsi · Oct 20, 2023 · 9d8d433 · 9d8d433
1 parent c300d8b
commit 9d8d433
Show file tree

Hide file tree

Showing 3 changed files with 44 additions and 18 deletions.
diff --git a/server/src/main/java/org/opensearch/index/cache/request/RequestCacheStats.java b/server/src/main/java/org/opensearch/index/cache/request/RequestCacheStats.java
@@ -32,6 +32,7 @@
 
 package org.opensearch.index.cache.request;
 
+import org.opensearch.Version;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 import org.opensearch.core.common.io.stream.Writeable;
@@ -61,14 +62,25 @@ public RequestCacheStats() {
     }
 
     public RequestCacheStats(StreamInput in) throws IOException {
-        // Any RequestCacheStats written to a stream should already have a counter for each possible tier type
-        this.map = in.readMap(StreamInput::readString, StatsHolder::new); // does it know to use the right constructor? does it rly need to be registered?
-    }
-
-    public RequestCacheStats(TierType tierType, StatsHolder statsHolder) {
-        // Create a RequestCacheStats object with only one tier's statistics populated
         this();
-        map.put(tierType.getStringValue(), statsHolder);
+        if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+            this.map = in.readMap(StreamInput::readString, StatsHolder::new); // does it know to use the right constructor? does it rly need to be registered?
+        } else {
+            // objects from earlier versions only contain on-heap info, and do not have entries info
+            long memorySize = in.readVLong();
+            long evictions = in.readVLong();
+            long hitCount = in.readVLong();
+            long missCount = in.readVLong();
+            this.map.put(
+                TierType.ON_HEAP.getStringValue(),
+                new StatsHolder(
+                    memorySize,
+                    evictions,
+                    hitCount,
+                    missCount,
+                    0
+                ));
+        }
     }
 
     public RequestCacheStats(Map<TierType, StatsHolder> inputMap) {
@@ -91,7 +103,6 @@ private StatsHolder getTierStats(TierType tierType) {
         return map.get(tierType.getStringValue());
     }
 
-    // should these take in strings bc thats whats done in the xcontent builder? seems wasteful
     public long getMemorySizeInBytes(TierType tierType) {
         return getTierStats(tierType).totalMetric.count();
     }
@@ -144,7 +155,16 @@ public long getEntries() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        out.writeMap(this.map, StreamOutput::writeString, (o, v) -> v.writeTo(o)); // ?
+        if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+            out.writeMap(this.map, StreamOutput::writeString, (o, v) -> v.writeTo(o)); // ?
+        } else {
+            // Write only on-heap values, and don't write entries metric
+            StatsHolder heapStats = map.get(TierType.ON_HEAP.getStringValue());
+            out.writeVLong(heapStats.getMemorySize());
+            out.writeVLong(heapStats.getEvictions());
+            out.writeVLong(heapStats.getHitCount());
+            out.writeVLong(heapStats.getMissCount());
+        }
     }
 
     @Override

diff --git a/server/src/main/java/org/opensearch/index/cache/request/StatsHolder.java b/server/src/main/java/org/opensearch/index/cache/request/StatsHolder.java
@@ -35,18 +35,25 @@ public StatsHolder() {
         this.entries = new CounterMetric();
     }
 
-    public StatsHolder(StreamInput in) throws IOException  {
-        // Read and write the values of the counter metrics. They should always be positive
+    public StatsHolder(long evictions, long memorySize, long hitCount, long missCount, long entries) {
         this.evictionsMetric = new CounterMetric();
-        this.evictionsMetric.inc(in.readVLong());
+        this.evictionsMetric.inc(evictions);
         this.totalMetric = new CounterMetric();
-        this.totalMetric.inc(in.readVLong());
+        this.totalMetric.inc(memorySize);
         this.hitCount = new CounterMetric();
-        this.hitCount.inc(in.readVLong());
+        this.hitCount.inc(hitCount);
         this.missCount = new CounterMetric();
-        this.missCount.inc(in.readVLong());
+        this.missCount.inc(missCount);
         this.entries = new CounterMetric();
-        this.entries.inc(in.readVLong());
+        this.entries.inc(entries);
+    }
+
+    public StatsHolder(StreamInput in) throws IOException  {
+        // Read and write the values of the counter metrics. They should always be positive
+        // This object is new, so we shouldn't need version checks for different behavior
+        this(in.readVLong(), in.readVLong(), in.readVLong(), in.readVLong(), in.readVLong());
+        // java forces us to do this in one line
+        // guaranteed to be evaluated in correct order (https://docs.oracle.com/javase/specs/jls/se7/html/jls-15.html#jls-15.7.4)
     }
 
     @Override

diff --git a/server/src/main/java/org/opensearch/indices/TieredCacheSpilloverStrategyHandler.java b/server/src/main/java/org/opensearch/indices/TieredCacheSpilloverStrategyHandler.java
@@ -54,8 +54,6 @@ public V computeIfAbsent(K key, TieredCacheLoader<K, V> loader) throws Exception
             V value = onHeapCachingTier.compute(key, loader);
             tieredCacheEventListener.onCached(key, value, TierType.ON_HEAP);
             return value;
-        } else {
-            //tieredCacheEventListener.onHit(key, cacheValue.value, cacheValue.source); // this double counts, see line 122
         }
         return cacheValue.value;
     }
@@ -105,6 +103,7 @@ public void onRemoval(RemovalNotification<K, V> notification) {
             switch (notification.getTierType()) {
                 case ON_HEAP:
                     diskCachingTier.put(notification.getKey(), notification.getValue());
+
                     break;
                 default:
                     break;