Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-16 Thread via GitHub 


asfgit closed pull request #277: IGNITE-18372 metrics for k2i consumer
URL: https://github.com/apache/ignite-extensions/pull/277


-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665911055


##
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetrics.java:
##
@@ -0,0 +1,253 @@
+/*
+ * 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.processors.metric.impl.MetricUtils.metricName;
+
+/** CDC kafka to ignite metrics. */
+public class KafkaToIgniteMetrics {
+/** Count of events received name. */
+public static final String EVTS_RCVD_CNT = "EventsReceivedCount";
+
+/** Count of events received description. */
+public static final String EVTS_RCVD_CNT_DESC = "Count of events received 
from Kafka";
+
+/** Timestamp of last received event name. */
+public static final String LAST_EVT_RCVD_TIME = "LastEventReceivedTime";
+
+/** Timestamp of last received event description. */
+public static final String LAST_EVT_RCVD_TIME_DESC = "Timestamp of last 
received event from Kafka";
+
+/** Count of metadata markers received name. */
+public static final String MARKERS_RCVD_CNT = "MarkersCount";
+
+/** Count of metadata markers received description. */
+public static final String MARKERS_RCVD_CNT_DESC = "Count of metadata 
markers received from Kafka";
+
+/** Count of events sent name. */
+public static final String MSGS_SENT_CNT = "EventsSentCount";
+
+/** Count of events sent description. */
+public static final String MSGS_SENT_CNT_DESC = "Count of events sent to 
destination cluster";
+
+/** Timestamp of last sent batch name. */
+public static final String LAST_MSG_SENT_TIME = "LastBatchSentTime";
+
+/** Timestamp of last sent batch description. */
+public static final String LAST_MSG_SENT_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;
+
+/** Metric registry manager. */
+private SingleMetricRegistryManager mregMgr;
+
+/** */
+private KafkaToIgniteMetrics(
+IgniteLogger log,
+KafkaToIgniteCdcStreamerConfiguration streamerCfg
+) throws IgniteCheckedException {
+this.log = log;
+this.streamerCfg = streamerCfg;
+
+initStandaloneMetricsKernal();
+initMetrics();
+}
+
+/**
+

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


alex-plekhanov commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665894752


##
modules/cdc-ext/src/main/java/org/apache/ignite/cdc/kafka/KafkaToIgniteMetrics.java:
##
@@ -0,0 +1,253 @@
+/*
+ * 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.processors.metric.impl.MetricUtils.metricName;
+
+/** CDC kafka to ignite metrics. */
+public class KafkaToIgniteMetrics {
+/** Count of events received name. */
+public static final String EVTS_RCVD_CNT = "EventsReceivedCount";
+
+/** Count of events received description. */
+public static final String EVTS_RCVD_CNT_DESC = "Count of events received 
from Kafka";
+
+/** Timestamp of last received event name. */
+public static final String LAST_EVT_RCVD_TIME = "LastEventReceivedTime";
+
+/** Timestamp of last received event description. */
+public static final String LAST_EVT_RCVD_TIME_DESC = "Timestamp of last 
received event from Kafka";
+
+/** Count of metadata markers received name. */
+public static final String MARKERS_RCVD_CNT = "MarkersCount";
+
+/** Count of metadata markers received description. */
+public static final String MARKERS_RCVD_CNT_DESC = "Count of metadata 
markers received from Kafka";
+
+/** Count of events sent name. */
+public static final String MSGS_SENT_CNT = "EventsSentCount";
+
+/** Count of events sent description. */
+public static final String MSGS_SENT_CNT_DESC = "Count of events sent to 
destination cluster";
+
+/** Timestamp of last sent batch name. */
+public static final String LAST_MSG_SENT_TIME = "LastBatchSentTime";
+
+/** Timestamp of last sent batch description. */
+public static final String LAST_MSG_SENT_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;
+
+/** Metric registry manager. */
+private SingleMetricRegistryManager mregMgr;
+
+/** */
+private KafkaToIgniteMetrics(
+IgniteLogger log,
+KafkaToIgniteCdcStreamerConfiguration streamerCfg
+) throws IgniteCheckedException {
+this.log = log;
+this.streamerCfg = streamerCfg;
+
+initStandaloneMetricsKernal();
+initMetrics();
+}
+
+/**

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665889591


##
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 
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 supplier) {
+try {
+waitForCondition(() -> supplier.get() == (long)(putCnt + rmvCnt) * 
KEYS_CNT * (backups + 1) *
+(mode == CacheMode.PARTITIONED ? 1 : srcCluster.length), 
getTestTimeout());

Review Comment:
   Fixed this mistake



-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665889414


##
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.findMetric(m).value());
+}
+}
+
 /** {@inheritDoc} */
 @Override protected void checkConsumerMetrics(Function 
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:
   Yeah, I agree. I refactored this part of the tests



-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665889101


##
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.findMetric(m).value());
+}
+}
+
 /** {@inheritDoc} */
 @Override protected void checkConsumerMetrics(Function 
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:
   Renamed 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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665888673


##
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 k2is = 
Collections.synchronizedList(new ArrayList<>());

Review Comment:
   Renamed the variable to `kafkaStreamers`. As for sync collection, I've lost 
myself in appliers concurrency and used this collection by mistake.



-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665881808


##
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;
+

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665880629


##
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:
   Yeah, it's reasonable. Fixed 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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665880334


##
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:
   I renamed the variables and switch the description to regustry name



-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


maksaska commented on code in PR #277:
URL: https://github.com/apache/ignite-extensions/pull/277#discussion_r1665879887


##
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:
   I renamed the variables



-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org

Re: [PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-07-04 Thread via GitHub 


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.findMetric(m).value());
+}
+}
+
 /** {@inheritDoc} */
 @Override protected void checkConsumerMetrics(Function 
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 
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 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 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

[PR] IGNITE-18372 metrics for k2i consumer [ignite-extensions]

2024-06-11 Thread via GitHub 


maksaska opened a new pull request, #277:
URL: https://github.com/apache/ignite-extensions/pull/277

   ### Metrics for CDC with Kafka 
   **CDC with Kafka** doesn't have metrics for 'Kafka To Ignite' consumer. The 
improvement in this pull request contains the neccessary updates to fix this 
issue.
   
   ***What is new***:
   - Count of messages applied to destination cluster
   - Timestamp of last applied event
   
   To achieve the proposed goals, I've used `StandaloneGridKernalContext` dummy 
context to get the manager for the metrics SPI. With 
`KafkaToIgniteCdcStreamerConfiguration` configured to supply the CDC toolkit 
with Metrics SPIs, we are now able to register metrics exporter. 


-- 
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: notifications-unsubscr...@ignite.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org