Merge pull request #5473 from telstra/fix/clear_cahces

Use new instance of cache map instead of cleaning

src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/service/ActionService.java

@@ -66,7 +66,7 @@ public class ActionService implements FlowSlaMonitoringCarrier {
     private final int shardCount;
 
     @VisibleForTesting
-    protected Map<FsmKey, FlowLatencyMonitoringFsm> fsms = new HashMap<>();
+    protected Map<FsmKey, FlowLatencyMonitoringFsm> fsms = createNewFsmMapInstance();
 
     public ActionService(FlowOperationsCarrier carrier, PersistenceManager persistenceManager,
                          Clock clock, Duration timeout, float threshold, int shardCount) {
@@ -181,7 +181,7 @@ private FsmKey getFsmKey(String flowId, FlowDirection direction) {
      * Remove all current fsms.
      */
     public void purge() {
-        fsms.clear();
+        fsms = createNewFsmMapInstance();
     }
 
     @Override
@@ -243,4 +243,12 @@ private static class FsmKey {
         String flowId;
         FlowDirection direction;
     }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static Map<FsmKey, FlowLatencyMonitoringFsm> createNewFsmMapInstance() {
+        return new HashMap<>();
+    }
 }

src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/service/FlowCacheService.java

@@ -50,7 +50,7 @@ public class FlowCacheService {
     private final Duration flowRttStatsExpirationTime;
     private final FlowCacheBoltCarrier carrier;
     private boolean active;
-    private final Map<String, FlowState> flowStates;
+    private Map<String, FlowState> flowStates;
     private final FlowRepository flowRepository;
     private final HaSubFlowRepository haSubFlowRepository;
 
@@ -59,7 +59,7 @@ public FlowCacheService(PersistenceManager persistenceManager, Clock clock,
         this.clock = clock;
         this.flowRttStatsExpirationTime = flowRttStatsExpirationTime;
         this.carrier = carrier;
-        flowStates = new HashMap<>();
+        flowStates = createNewFlowStateInstance();
         active = false;
         flowRepository = persistenceManager.getRepositoryFactory().createFlowRepository();
         haSubFlowRepository = persistenceManager.getRepositoryFactory().createHaSubFlowRepository();
@@ -153,7 +153,7 @@ public void activate() {
      */
     public void deactivate() {
         if (active) {
-            flowStates.clear();
+            flowStates = createNewFlowStateInstance();
             log.info("Flow cache cleared.");
             active = false;
         }
@@ -207,4 +207,12 @@ private boolean isCompletedHaFlow(HaSubFlow haSubFlow) {
     protected boolean flowStatesIsEmpty() {
         return flowStates.isEmpty();
     }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static Map<String, FlowState> createNewFlowStateInstance() {
+        return new HashMap<>();
+    }
 }

src-java/flowmonitoring-topology/flowmonitoring-storm-topology/src/main/java/org/openkilda/wfm/topology/flowmonitoring/service/IslCacheService.java

@@ -49,7 +49,7 @@ public IslCacheService(PersistenceManager persistenceManager, Clock clock, Durat
         this.clock = clock;
         this.islRttLatencyExpiration = islRttLatencyExpiration;
         active = false;
-        linkStates = new HashMap<>();
+        linkStates = createNewLinkStateInstance();
         islRepository = persistenceManager.getRepositoryFactory().createIslRepository();
     }
 
@@ -74,7 +74,7 @@ public void activate() {
      */
     public void deactivate() {
         if (active) {
-            linkStates.clear();
+            linkStates = createNewLinkStateInstance();
             log.info("Isl cache cleared.");
             active = false;
         }
@@ -168,4 +168,12 @@ private void cleanUpLinkStatesByEndpoint(SwitchId switchId, int port) {
     protected boolean linkStatesIsEmpty() {
         return linkStates.isEmpty();
     }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private Map<Link, LinkState> createNewLinkStateInstance() {
+        return new HashMap<>();
+    }
 }

src-java/opentsdb-topology/opentsdb-storm-topology/src/main/java/org/openkilda/wfm/topology/opentsdb/models/Storage.java

@@ -36,10 +36,10 @@ public class Storage implements Serializable {
     public static final char TAG_VALUE_DELIMITER = ':';
     public static final String NULL_TAG = "NULL_TAG";
 
-    private final Map<String, DatapointValue> map;
+    private Map<String, DatapointValue> map;
 
     public Storage() {
-        this.map = new HashMap<>();
+        this.map = createNewMapInstance();
     }
 
     public void add(Datapoint datapoint) {
@@ -50,9 +50,15 @@ public DatapointValue get(Datapoint datapoint) {
         return map.get(createKey(datapoint));
     }
 
+    /**
+     * Removes datapoints from the storage if they are older than now() - ttlInMillis.
+     */
     public void removeOutdated(long ttlInMillis) {
         long now = System.currentTimeMillis();
         map.entrySet().removeIf(entry -> now - entry.getValue().getTime() > ttlInMillis);
+        if (map.isEmpty()) {
+            map = createNewMapInstance();
+        }
     }
 
     public int size() {
@@ -95,6 +101,14 @@ private static SortedMap<String, String> getSortedTags(Datapoint datapoint) {
         return sortedTags;
     }
 
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static Map<String, DatapointValue> createNewMapInstance() {
+        return new HashMap<>();
+    }
+
     @Value
     public static class DatapointValue implements Serializable {
         Number value;

src-java/stats-topology/stats-storm-topology/src/main/java/org/openkilda/wfm/topology/stats/service/KildaEntryCacheService.java

@@ -95,12 +95,11 @@ public class KildaEntryCacheService {
     /**
      * Cookie to flow and meter to flow maps.
      */
-    private final HashSetValuedHashMap<CookieCacheKey, KildaEntryDescriptor> cookieToFlow =
-            new HashSetValuedHashMap<>();
-    private final HashSetValuedHashMap<MeterCacheKey, KildaEntryDescriptor> switchAndMeterToFlow =
-            new HashSetValuedHashMap<>();
-    private final HashSetValuedHashMap<GroupCacheKey, KildaEntryDescriptor> switchAndGroupToFlow =
-            new HashSetValuedHashMap<>();
+    private HashSetValuedHashMap<CookieCacheKey, KildaEntryDescriptor> cookieToFlow = createNewCookieCacheInstance();
+    private HashSetValuedHashMap<MeterCacheKey, KildaEntryDescriptor> switchAndMeterToFlow =
+            createNewMeterCacheInstance();
+    private HashSetValuedHashMap<GroupCacheKey, KildaEntryDescriptor> switchAndGroupToFlow =
+            createNewGroupCacheInstance();
 
     public KildaEntryCacheService(PersistenceManager persistenceManager, KildaEntryCacheCarrier carrier) {
         RepositoryFactory repositoryFactory = persistenceManager.getRepositoryFactory();
@@ -548,8 +547,32 @@ private void refreshCache() {
     }
 
     private void clearCache() {
-        cookieToFlow.clear();
-        switchAndMeterToFlow.clear();
-        switchAndGroupToFlow.clear();
+        cookieToFlow = createNewCookieCacheInstance();
+        switchAndMeterToFlow = createNewMeterCacheInstance();
+        switchAndGroupToFlow = createNewGroupCacheInstance();
+    }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static HashSetValuedHashMap<CookieCacheKey, KildaEntryDescriptor> createNewCookieCacheInstance() {
+        return new HashSetValuedHashMap<>();
+    }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static HashSetValuedHashMap<MeterCacheKey, KildaEntryDescriptor>  createNewMeterCacheInstance() {
+        return new HashSetValuedHashMap<>();
+    }
+
+    /**
+     * Instead of map.clear() we are creating a new map here.
+     * We need it because map.clear() doesn't shrink already allocated map capacity, size of which can be significant.
+     */
+    private static HashSetValuedHashMap<GroupCacheKey, KildaEntryDescriptor> createNewGroupCacheInstance() {
+        return new HashSetValuedHashMap<>();
     }
 }