IGNITE-14355 CDC Metrics (apache#9398)

RapidTransit · Sep 28, 2021 · 2f39591 · 2f39591
1 parent fd67579
commit 2f39591
Show file tree

Hide file tree

Showing 12 changed files with 542 additions and 163 deletions.
diff --git a/modules/core/src/main/java/org/apache/ignite/cdc/CdcConfiguration.java b/modules/core/src/main/java/org/apache/ignite/cdc/CdcConfiguration.java
@@ -20,6 +20,7 @@
 import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.internal.cdc.CdcMain;
 import org.apache.ignite.lang.IgniteExperimental;
+import org.apache.ignite.spi.metric.MetricExporterSpi;
 
 /**
  * This class defines {@link CdcMain} runtime configuration.
@@ -39,6 +40,9 @@ public class CdcConfiguration {
     /** Change Data Capture consumer. */
     private CdcConsumer consumer;
 
+    /** Metric exporter SPI. */
+    private MetricExporterSpi[] metricExporterSpi;
+
     /** Keep binary flag.<br>Default value {@code true}. */
     private boolean keepBinary = DFLT_KEEP_BINARY;
 
@@ -66,6 +70,25 @@ public void setConsumer(CdcConsumer consumer) {
         this.consumer = consumer;
     }
 
+    /**
+     * Sets fully configured instances of {@link MetricExporterSpi}.
+     *
+     * @param metricExporterSpi Fully configured instances of {@link MetricExporterSpi}.
+     * @see CdcConfiguration#getMetricExporterSpi()
+     */
+    public void setMetricExporterSpi(MetricExporterSpi... metricExporterSpi) {
+        this.metricExporterSpi = metricExporterSpi;
+    }
+
+    /**
+     * Gets fully configured metric SPI implementations.
+     *
+     * @return Metric exporter SPI implementations.
+     */
+    public MetricExporterSpi[] getMetricExporterSpi() {
+        return metricExporterSpi;
+    }
+
     /** @return keep binary value. */
     public boolean isKeepBinary() {
         return keepBinary;

diff --git a/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java b/modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java
@@ -21,6 +21,7 @@
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cache.CacheEntryVersion;
 import org.apache.ignite.internal.cdc.CdcMain;
+import org.apache.ignite.internal.processors.metric.MetricRegistry;
 import org.apache.ignite.lang.IgniteExperimental;
 import org.apache.ignite.resources.LoggerResource;
 
@@ -29,7 +30,7 @@
  * This consumer will receive data change events during {@link CdcMain} application invocation.
  * The lifecycle of the consumer is the following:
  * <ul>
- *     <li>Start of the consumer {@link #start()}.</li>
+ *     <li>Start of the consumer {@link #start(MetricRegistry)}.</li>
  *     <li>Notification of the consumer by the {@link #onEvents(Iterator)} call.</li>
  *     <li>Stop of the consumer {@link #stop()}.</li>
  * </ul>
@@ -55,8 +56,9 @@
 public interface CdcConsumer {
     /**
      * Starts the consumer.
+     * @param mreg Metric registry for consumer specific metrics.
      */
-    public void start();
+    public void start(MetricRegistry mreg);
 
     /**
      * Handles entry changes events.
@@ -71,7 +73,7 @@ public interface CdcConsumer {
 
     /**
      * Stops the consumer.
-     * This methods can be invoked only after {@link #start()}.
+     * This methods can be invoked only after {@link #start(MetricRegistry)}.
      */
     public void stop();
 }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java
@@ -1991,8 +1991,7 @@ private IgniteConfiguration initializeConfiguration(IgniteConfiguration cfg)
             if (myCfg.getUserAttributes() == null)
                 myCfg.setUserAttributes(Collections.<String, Object>emptyMap());
 
-            if (myCfg.getMBeanServer() == null && !U.IGNITE_MBEANS_DISABLED)
-                myCfg.setMBeanServer(ManagementFactory.getPlatformMBeanServer());
+            initializeDefaultMBeanServer(myCfg);
 
             Marshaller marsh = myCfg.getMarshaller();
 
@@ -2674,6 +2673,12 @@ public boolean hasStartLatchCompleted() {
         }
     }
 
+    /** Initialize default mbean server. */
+    public static void initializeDefaultMBeanServer(IgniteConfiguration myCfg) {
+        if (myCfg.getMBeanServer() == null && !U.IGNITE_MBEANS_DISABLED)
+            myCfg.setMBeanServer(ManagementFactory.getPlatformMBeanServer());
+    }
+
     /**
      * @param cfg Ignite Configuration with legacy data storage configuration.
      */