NMS-16486: Extracted caches from DocumentEnricherImpl

container/features/src/main/resources/features-sentinel.xml

@@ -139,6 +139,7 @@
         <bundle>wrap:mvn:org.freemarker/freemarker/${freemarkerVersion}$Bundle-SymbolicName=org.freemarker&amp;Bundle-Version=${freemarkerVersion}</bundle>
         <bundle>mvn:org.opennms.core/org.opennms.core.cache/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols.netflow/org.opennms.features.telemetry.protocols.netflow.transport/${project.version}</bundle>
+        <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.cache/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.common/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.flows/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols.netflow/org.opennms.features.telemetry.protocols.netflow.adapter/${project.version}</bundle>

container/features/src/main/resources/features.xml

@@ -1151,6 +1151,7 @@
         -->
         <bundle>mvn:com.google.protobuf/protobuf-java/${protobufVersion}</bundle>
         <bundle>wrap:mvn:com.google.protobuf/protobuf-java-util/${protobufVersion}$overwrite=merge&amp;Import-Package=javax.annotation;version=!,*</bundle>
+        <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.cache/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.common/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols/org.opennms.features.telemetry.protocols.flows/${project.version}</bundle>
         <bundle>mvn:org.opennms.features.telemetry.protocols.netflow/org.opennms.features.telemetry.protocols.netflow.parser/${project.version}</bundle>

features/flows/elastic/src/main/java/org/opennms/netmgt/flows/elastic/FlowDocument.java

@@ -29,13 +29,10 @@
 package org.opennms.netmgt.flows.elastic;
 
 import java.util.LinkedHashSet;
-import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 
 import org.opennms.integration.api.v1.flows.Flow;
-import org.opennms.netmgt.flows.processing.enrichment.EnrichedFlow;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
 
 import com.google.gson.annotations.SerializedName;
 

features/flows/elastic/src/test/java/org/opennms/netmgt/flows/elastic/FlowDocumentTest.java

@@ -37,15 +37,14 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.time.Instant;
 import java.util.List;
 import java.util.Optional;
 
 import org.junit.Test;
 import org.opennms.core.test.xml.JsonTest;
 import org.opennms.netmgt.flows.api.Flow;
 import org.opennms.netmgt.flows.processing.enrichment.EnrichedFlow;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeInfo;
 
 import com.google.common.io.Resources;
 import com.google.gson.Gson;

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/AggregatedFlowQueryIT.java

@@ -93,13 +93,14 @@
 import org.opennms.netmgt.flows.classification.persistence.api.RuleBuilder;
 import org.opennms.netmgt.flows.elastic.agg.AggregatedFlowQueryService;
 import org.opennms.netmgt.flows.processing.impl.DocumentEnricherImpl;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeInfo;
 import org.opennms.netmgt.flows.filter.api.Filter;
 import org.opennms.netmgt.flows.filter.api.SnmpInterfaceIdFilter;
 import org.opennms.netmgt.flows.filter.api.TimeRangeFilter;
 import org.opennms.netmgt.flows.persistence.FlowDocumentBuilder;
 import org.opennms.netmgt.flows.processing.FlowBuilder;
 import org.opennms.netmgt.flows.processing.impl.DocumentMangler;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeMetadataCacheImpl;
 
 import com.codahale.metrics.MetricRegistry;
 import com.google.common.collect.ImmutableSet;
@@ -168,18 +169,27 @@ public void setUp() throws MalformedURLException, ExecutionException, Interrupte
                 new RuleBuilder().withName("https").withSrcPort("443").withProtocol("tcp,udp").build()),
                                                                          FilterService.NOOP);
 
-        documentEnricher = new DocumentEnricherImpl(metricRegistry,
-                                                    new MockNodeDao(),
-                                                    new MockIpInterfaceDao(),
-                                                    new MockInterfaceToNodeCache(),
-                                                    new MockSessionUtils(),
+        final NodeMetadataCacheImpl nodeMetadataCache = new NodeMetadataCacheImpl(
+                new CacheConfigBuilder()
+                        .withName("nodeInfoCache")
+                        .withMaximumSize(1000)
+                        .withExpireAfterWrite(300)
+                        .build(),
+                new CacheConfigBuilder()
+                        .withName("nodeMetadataCache")
+                        .withMaximumSize(1000)
+                        .withExpireAfterWrite(300)
+                        .build(),
+                new MetricRegistry(),
+                new MockNodeDao(),
+                new MockIpInterfaceDao(),
+                new MockInterfaceToNodeCache()
+        );
+        documentEnricher = new DocumentEnricherImpl(new MockSessionUtils(),
                                                     classificationEngine,
-                                                    new CacheConfigBuilder()
-                                                        .withName("flows.node")
-                                                        .withMaximumSize(1000)
-                                                        .withExpireAfterWrite(300)
-                                                        .build(), 0,
-                                                    new DocumentMangler(new ScriptEngineManager()));
+                                                    0,
+                                                    new DocumentMangler(new ScriptEngineManager()),
+                                                    nodeMetadataCache);
 
         // The repository should be empty
         assertThat(smartQueryService.getFlowCount(Collections.singletonList(new TimeRangeFilter(0, System.currentTimeMillis()))).get(), equalTo(0L));

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/FlowQueryIT.java

@@ -69,6 +69,7 @@
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
+import org.opennms.core.cache.CacheConfig;
 import org.opennms.core.cache.CacheConfigBuilder;
 import org.opennms.core.test.elastic.ElasticSearchRule;
 import org.opennms.core.test.elastic.ElasticSearchServerConfig;
@@ -101,6 +102,7 @@
 import org.opennms.netmgt.flows.filter.api.SnmpInterfaceIdFilter;
 import org.opennms.netmgt.flows.filter.api.TimeRangeFilter;
 import org.opennms.netmgt.flows.processing.impl.DocumentMangler;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeMetadataCacheImpl;
 
 import com.codahale.metrics.MetricRegistry;
 import com.google.common.collect.ImmutableSet;
@@ -143,19 +145,28 @@ public void setUp() throws MalformedURLException, ExecutionException, Interrupte
                 new RuleBuilder().withName("http").withSrcPort("80").withProtocol("tcp,udp").build(),
                 new RuleBuilder().withName("https").withSrcPort("443").withProtocol("tcp,udp").build()),
                                                                          FilterService.NOOP);
+        final NodeMetadataCacheImpl nodeMetadataCache = new NodeMetadataCacheImpl(
+                new CacheConfigBuilder()
+                        .withName("nodeInfoCache")
+                        .withMaximumSize(1000)
+                        .withExpireAfterWrite(300)
+                        .build(),
+                new CacheConfigBuilder()
+                        .withName("nodeMetadataCache")
+                        .withMaximumSize(1000)
+                        .withExpireAfterWrite(300)
+                        .build(),
+                new MetricRegistry(),
+                new MockNodeDao(),
+                new MockIpInterfaceDao(),
+                new MockInterfaceToNodeCache()
+        );
 
-        documentEnricher = new DocumentEnricherImpl(metricRegistry,
-                                                    new MockNodeDao(),
-                                                    new MockIpInterfaceDao(),
-                                                    new MockInterfaceToNodeCache(),
-                                                    new MockSessionUtils(),
+        documentEnricher = new DocumentEnricherImpl(new MockSessionUtils(),
                                                     classificationEngine,
-                                                    new CacheConfigBuilder()
-                                                                  .withName("flows.node")
-                                                                  .withMaximumSize(1000)
-                                                                  .withExpireAfterWrite(300)
-                                                                  .build(), 0,
-                                                    new DocumentMangler(new ScriptEngineManager()));
+                                                    0,
+                                                    new DocumentMangler(new ScriptEngineManager()),
+                                                    nodeMetadataCache);
 
         final RawIndexInitializer initializer = new RawIndexInitializer(client, settings);
 

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/KafkaFlowForwarderIT.java

@@ -61,7 +61,7 @@
 import org.opennms.netmgt.flows.persistence.KafkaFlowForwarder;
 import org.opennms.netmgt.flows.persistence.model.FlowDocument;
 import org.opennms.netmgt.flows.processing.enrichment.EnrichedFlow;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeInfo;
 import org.osgi.service.cm.ConfigurationAdmin;
 
 import com.codahale.metrics.MetricRegistry;

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/MarkerCacheIT.java

@@ -59,7 +59,7 @@
 import org.opennms.netmgt.dao.api.SessionUtils;
 import org.opennms.netmgt.dao.api.SnmpInterfaceDao;
 import org.opennms.netmgt.flows.processing.enrichment.EnrichedFlow;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeInfo;
 import org.opennms.netmgt.flows.processing.impl.InterfaceMarkerImpl;
 import org.opennms.netmgt.model.OnmsSnmpInterface;
 import org.opennms.test.JUnitConfigurationEnvironment;

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/NodeIdentificationIT.java

@@ -62,6 +62,7 @@
 import org.opennms.netmgt.model.NetworkBuilder;
 import org.opennms.netmgt.model.OnmsMetaData;
 import org.opennms.netmgt.model.OnmsNode;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeMetadataCacheImpl;
 import org.opennms.test.JUnitConfigurationEnvironment;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.test.context.ContextConfiguration;
@@ -115,16 +116,29 @@ public void before() {
     @Test
     public void testSomething() throws InterruptedException {
         final ClassificationEngine classificationEngine = new DefaultClassificationEngine(() -> Collections.emptyList(), FilterService.NOOP);
-        final DocumentEnricherImpl documentEnricher = new DocumentEnricherImpl(
-                new MetricRegistry(),
-                databasePopulator.getNodeDao(), databasePopulator.getIpInterfaceDao(),
-                interfaceToNodeCache, sessionUtils, classificationEngine,
+        final NodeMetadataCacheImpl nodeMetadataCache = new NodeMetadataCacheImpl(
+                new CacheConfigBuilder()
+                        .withName("nodeInfoCache")
+                        .withMaximumSize(1000)
+                        .withExpireAfterWrite(300)
+                        .build(),
                 new CacheConfigBuilder()
-                        .withName("flows.node")
+                        .withName("nodeMetadataCache")
                         .withMaximumSize(1000)
                         .withExpireAfterWrite(300)
-                        .build(), 0,
-                new DocumentMangler(new ScriptEngineManager()));
+                        .build(),
+                new MetricRegistry(),
+                databasePopulator.getNodeDao(),
+                databasePopulator.getIpInterfaceDao(),
+                interfaceToNodeCache
+        );
+
+        final DocumentEnricherImpl documentEnricher = new DocumentEnricherImpl(
+                sessionUtils,
+                classificationEngine,
+                 0,
+                new DocumentMangler(new ScriptEngineManager()),
+                nodeMetadataCache);
 
         final TestFlow testFlow = new TestFlow();
         testFlow.setSrcAddr("1.1.1.1");

features/flows/itests/src/test/java/org/opennms/netmgt/flows/elastic/ThresholdingIT.java

@@ -65,7 +65,7 @@
 import org.opennms.netmgt.filter.api.FilterDao;
 import org.opennms.integration.api.v1.flows.Flow;
 import org.opennms.netmgt.flows.processing.enrichment.EnrichedFlow;
-import org.opennms.netmgt.flows.processing.enrichment.NodeInfo;
+import org.opennms.netmgt.telemetry.protocols.common.cache.NodeInfo;
 import org.opennms.netmgt.flows.api.FlowSource;
 import org.opennms.netmgt.flows.classification.ClassificationEngine;
 import org.opennms.netmgt.flows.classification.ClassificationRuleProvider;

features/flows/processing/pom.xml

@@ -72,6 +72,16 @@
       <artifactId>org.opennms.features.telemetry.config.api</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.opennms.features.telemetry.protocols</groupId>
+      <artifactId>org.opennms.features.telemetry.protocols.common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.opennms.features.telemetry.protocols</groupId>
+      <artifactId>org.opennms.features.telemetry.protocols.cache</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>org.opennms.dependencies</groupId>
       <artifactId>oia-dependencies</artifactId>