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; + 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: Removed unneccessary `#close()` methods. Replaced it with the `#stop()` for manager and metric spi. -- 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]
