alex-plekhanov commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665242578
##########
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/AbstractIgniteCdcStreamer.java:
##########
@@ -44,28 +44,28 @@
*/
public abstract class AbstractIgniteCdcStreamer implements CdcConsumer {
/** */
- public static final String I2I_EVTS_SNT_CNT = "EventsCount";
+ public static final String EVTS_SNT_CNT = "EventsCount";
Review Comment:
`snt` - as an abbreviation is not commonly used in Ignite, let's use the
full word.
Also `rsvd` -> `rcvd`, since `received` doesn't contain 'S' char.
##########
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetrics.java:
##########
@@ -138,7 +137,7 @@ private void initStandaloneMetricsKernal() throws
IgniteCheckedException {
@Override protected IgniteConfiguration
prepareIgniteConfiguration() {
IgniteConfiguration cfg = super.prepareIgniteConfiguration();
- cfg.setIgniteInstanceName("kafka-ignite-streamer-" +
igniteInstanceName);
+
cfg.setIgniteInstanceName(cdcInstanceName(streamerCfg.getMetricDirectoryName()));
Review Comment:
If we allow to configure it, lets use `streamerCfg.getMetricRegistryName()`
as is, without cdcInstanceName
##########
modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java:
##########
@@ -139,11 +155,85 @@ public class CdcKafkaReplicationTest extends
AbstractReplicationTest {
return futs;
}
+ /** {@inheritDoc} */
+ @Override protected void checkMetrics() throws
IgniteInterruptedCheckedException {
+ super.checkMetrics();
+
+ for (AbstractKafkaToIgniteCdcStreamer k2i : k2is) {
+ KafkaToIgniteMetrics metrics = getFieldValue(k2i, "metrics");
+ MetricRegistryImpl mreg = getFieldValue(metrics, "mreg");
+ checkK2IMetrics(m -> mreg.<AtomicLongMetric>findMetric(m).value());
+ }
+ }
+
/** {@inheritDoc} */
@Override protected void checkConsumerMetrics(Function<String, Long>
longMetric) {
-
assertNotNull(longMetric.apply(IgniteToIgniteCdcStreamer.LAST_EVT_TIME));
- assertNotNull(longMetric.apply(IgniteToIgniteCdcStreamer.EVTS_CNT));
- assertNotNull(longMetric.apply(IgniteToKafkaCdcStreamer.BYTES_SENT));
+ assertNotNull(longMetric.apply(LAST_EVT_SNT_TIME));
+ assertNotNull(longMetric.apply(EVTS_SNT_CNT));
+ assertNotNull(longMetric.apply(BYTES_SNT));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void checkMetricsCount(int putCnt, int rmvCnt) {
Review Comment:
Looks like we can union `putCnt` and `rmvCnt` to `evtsCnt` (and I think we
can pass `KEYS_CNT` and `KEYS_CNT * 2` here)
##########
modules/cdc-ext/src/test/java/org/apache/ignite/cdc/AbstractReplicationTest.java:
##########
@@ -633,6 +637,40 @@ protected String[] hostAddresses(IgniteEx[] dest) {
/** */
protected abstract void checkConsumerMetrics(Function<String, Long>
longMetric);
+ /** Checks the count for CDC metrics */
+ protected abstract void checkMetricsCount(int putCnt, int rmvCnt);
+
+ /** Checks the events count for CDC metrics */
+ protected void checkMetricsEventsCount(int putCnt, int rmvCnt,
Supplier<Long> supplier) {
+ try {
+ waitForCondition(() -> supplier.get() == (long)(putCnt + rmvCnt) *
KEYS_CNT * (backups + 1) *
+ (mode == CacheMode.PARTITIONED ? 1 : srcCluster.length),
getTestTimeout());
Review Comment:
`evtsCnt * (mode == CacheMode.PARTITIONED ? (backups + 1) :
srcCluster.length)`
##########
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteCdcStreamerConfiguration.java:
##########
@@ -48,6 +48,9 @@ public class KafkaToIgniteCdcStreamerConfiguration {
/** {@link KafkaToIgniteCdcStreamerApplier} thread count. */
private int threadCnt = DFLT_THREAD_CNT;
+ /** Default name for metrics registry directory for Kafka to Ignite CDC. */
+ public static final String DFLT_METRICS_DIR_NAME = "K2I";
Review Comment:
It's not a directory name. It's metric registry name.
Never used in production code, only in tests.
Let's use 'cdc-KafkaToIgnite', or 'cdc-kafka-to-ignite' (full words).
##########
modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java:
##########
@@ -61,6 +74,9 @@ public class CdcKafkaReplicationTest extends
AbstractReplicationTest {
/** */
private static EmbeddedKafkaCluster KAFKA = null;
+ /** */
+ protected final List<AbstractKafkaToIgniteCdcStreamer> k2is =
Collections.synchronizedList(new ArrayList<>());
Review Comment:
Let's avoid abbreviation, except commonly used ('k2i' is not commonly used)
Why do we need synchronizedList here? Looks like this list is always used
sequantially in one thread.
##########
modules/cdc-ext/src/test/java/org/apache/ignite/cdc/kafka/CdcKafkaReplicationTest.java:
##########
@@ -139,11 +155,85 @@ public class CdcKafkaReplicationTest extends
AbstractReplicationTest {
return futs;
}
+ /** {@inheritDoc} */
+ @Override protected void checkMetrics() throws
IgniteInterruptedCheckedException {
+ super.checkMetrics();
+
+ for (AbstractKafkaToIgniteCdcStreamer k2i : k2is) {
+ KafkaToIgniteMetrics metrics = getFieldValue(k2i, "metrics");
+ MetricRegistryImpl mreg = getFieldValue(metrics, "mreg");
+ checkK2IMetrics(m -> mreg.<AtomicLongMetric>findMetric(m).value());
+ }
+ }
+
/** {@inheritDoc} */
@Override protected void checkConsumerMetrics(Function<String, Long>
longMetric) {
-
assertNotNull(longMetric.apply(IgniteToIgniteCdcStreamer.LAST_EVT_TIME));
- assertNotNull(longMetric.apply(IgniteToIgniteCdcStreamer.EVTS_CNT));
- assertNotNull(longMetric.apply(IgniteToKafkaCdcStreamer.BYTES_SENT));
+ assertNotNull(longMetric.apply(LAST_EVT_SNT_TIME));
+ assertNotNull(longMetric.apply(EVTS_SNT_CNT));
+ assertNotNull(longMetric.apply(BYTES_SNT));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void checkMetricsCount(int putCnt, int rmvCnt) {
+ checkMetricsEventsCount(putCnt, rmvCnt,
getConsumerEventsCount(EVTS_SNT_CNT));
+
+ checkMetricsEventsCount(putCnt, rmvCnt,
getKafkaConsumerEventsCount(EVTS_RSVD_CNT));
Review Comment:
`getKafkaConsumerEventsCount` - > `getKafkaStreamerEventsCount` or
`getKafkaApplierEventsCount`
The same for JMX
##########
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetrics.java:
##########
@@ -0,0 +1,250 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cdc.kafka;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Consumer;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import
org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext;
+import
org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneSpiContext;
+import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
+import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.metric.MetricExporterSpi;
+import org.apache.ignite.spi.metric.ReadOnlyMetricManager;
+import org.apache.ignite.spi.metric.ReadOnlyMetricRegistry;
+import org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi;
+import org.apache.ignite.spi.metric.noop.NoopMetricExporterSpi;
+
+import static
org.apache.ignite.internal.IgnitionEx.initializeDefaultMBeanServer;
+import static org.apache.ignite.internal.cdc.CdcMain.cdcInstanceName;
+import static
org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext.closeAllComponents;
+import static
org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
+
+/** CDC kafka to ignite metrics. */
+public class KafkaToIgniteMetrics {
+ /** Count of events received name. */
+ public static final String EVTS_RSVD_CNT = "EventsReceivedCount";
+
+ /** Count of events received description. */
+ public static final String EVTS_RSVD_CNT_DESC = "Count of events received
from Kafka";
+
+ /** Timestamp of last received event name. */
+ public static final String LAST_EVT_RSVD_TIME = "LastEventReceivedTime";
+
+ /** Timestamp of last received event description. */
+ public static final String LAST_EVT_RSVD_TIME_DESC = "Timestamp of last
received event from Kafka";
+
+ /** Count of metadata markers received name. */
+ public static final String MARKERS_RSVD_CNT = "MarkersCount";
+
+ /** Count of metadata markers received description. */
+ public static final String MARKERS_RSVD_CNT_DESC = "Count of metadata
markers received from Kafka";
+
+ /** Count of events sent name. */
+ public static final String MSGS_SNT_CNT = "EventsSentCount";
+
+ /** Count of events sent description. */
+ public static final String MSGS_SNT_CNT_DESC = "Count of events sent to
destination cluster";
+
+ /** Timestamp of last sent batch name. */
+ public static final String LAST_MSG_SNT_TIME = "LastBatchSentTime";
+
+ /** Timestamp of last sent batch description. */
+ public static final String LAST_MSG_SNT_TIME_DESC = "Timestamp of last
sent batch to the destination cluster";
+
+ /** Timestamp of last received message. */
+ private AtomicLongMetric lastRcvdEvtTs;
+
+ /** Count of received events. */
+ private AtomicLongMetric evtsRcvdCnt;
+
+ /** Timestamp of last sent message. */
+ private AtomicLongMetric lastSntMsgTs;
+
+ /** Count of sent events. */
+ private AtomicLongMetric evtsSntCnt;
+
+ /** Count of received markers. */
+ private AtomicLongMetric markersCnt;
+
+ /** Standalone kernal context. */
+ private StandaloneGridKernalContext kctx;
+
+ /** CDC metrics registry. */
+ private MetricRegistryImpl mreg;
+
+ /** */
+ private final IgniteLogger log;
+
+ /** Streamer configuration. */
+ private final KafkaToIgniteCdcStreamerConfiguration streamerCfg;
+
+ /** */
+ private KafkaToIgniteMetrics(
+ IgniteLogger log,
+ KafkaToIgniteCdcStreamerConfiguration streamerCfg
+ ) throws IgniteCheckedException {
+ this.log = log;
+ this.streamerCfg = streamerCfg;
+
+ initStandaloneMetricsKernal();
+ initMetrics();
+ }
+
+ /**
+ * Creates an instance of {@link KafkaToIgniteMetrics}.
+ * @param log Logger.
+ * @param streamerCfg Streamer config.
+ * @return {@link KafkaToIgniteMetrics} instance.
+ */
+ public static KafkaToIgniteMetrics startMetrics(
+ IgniteLogger log,
+ KafkaToIgniteCdcStreamerConfiguration streamerCfg
+ ) {
+ try {
+ return new KafkaToIgniteMetrics(log, streamerCfg);
+ }
+ catch (IgniteCheckedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** @throws IgniteCheckedException If failed. */
+ private void initStandaloneMetricsKernal() throws IgniteCheckedException {
+ kctx = new StandaloneGridKernalContext(log, null, null) {
+ @Override protected IgniteConfiguration
prepareIgniteConfiguration() {
+ IgniteConfiguration cfg = super.prepareIgniteConfiguration();
+
+
cfg.setIgniteInstanceName(cdcInstanceName(streamerCfg.getMetricDirectoryName()));
+
+ if (!F.isEmpty(streamerCfg.getMetricExporterSpi()))
+
cfg.setMetricExporterSpi(streamerCfg.getMetricExporterSpi());
+ else {
+ cfg.setMetricExporterSpi(U.IGNITE_MBEANS_DISABLED
+ ? new NoopMetricExporterSpi()
+ : new JmxMetricExporterSpi());
+ }
+
+ initializeDefaultMBeanServer(cfg);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String igniteInstanceName() {
+ return config().getIgniteInstanceName();
+ }
+ };
+
+ mreg = new MetricRegistryImpl(metricName("cdc", "applier"), null,
null, log);
+
+ ReadOnlyMetricManager mregMgr = new SingleMetricRegistryManager(mreg);
+
+ for (MetricExporterSpi exporterSpi :
kctx.config().getMetricExporterSpi()) {
+ kctx.resource().injectGeneric(exporterSpi);
+ exporterSpi.setMetricRegistry(mregMgr);
+ exporterSpi.onContextInitialized(new StandaloneSpiContext());
+ exporterSpi.spiStart(null);
+ }
+ }
+
+ /** Initialize metrics. */
+ private void initMetrics() {
+ this.evtsRcvdCnt = mreg.longMetric(EVTS_RSVD_CNT, EVTS_RSVD_CNT_DESC);
+ this.lastRcvdEvtTs = mreg.longMetric(LAST_EVT_RSVD_TIME,
LAST_EVT_RSVD_TIME_DESC);
+ this.evtsSntCnt = mreg.longMetric(MSGS_SNT_CNT, MSGS_SNT_CNT_DESC);
+ this.lastSntMsgTs = mreg.longMetric(LAST_MSG_SNT_TIME,
LAST_MSG_SNT_TIME_DESC);
+ this.markersCnt = mreg.longMetric(MARKERS_RSVD_CNT,
MARKERS_RSVD_CNT_DESC);
+ }
+
+ /**
+ * Stops metric manager and metrics SPI.
+ */
+ public void stopMetrics() throws IgniteCheckedException {
+ if (kctx != null)
+ closeAllComponents(kctx);
Review Comment:
IMO no need to stop it, since we don't start it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
