IGNITE-12108 TCP Communication Metrics ported to a new framework.

modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceIoc.java

@@ -25,10 +25,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
+import org.apache.ignite.internal.resources.MetricManagerResource;
 import org.apache.ignite.internal.util.GridLeanIdentitySet;
 import org.apache.ignite.internal.util.IgniteUtils;
 import org.apache.ignite.internal.util.typedef.F;
@@ -48,7 +50,6 @@
 import org.apache.ignite.resources.TaskContinuousMapperResource;
 import org.apache.ignite.resources.TaskSessionResource;
 import org.jetbrains.annotations.Nullable;
-import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * Resource container contains caches for classes used for injection.
@@ -513,6 +514,9 @@ enum ResourceAnnotation {
         /** */
         CACHE_STORE_SESSION(CacheStoreSessionResource.class),
 
+        /** */
+        METRIC_MANAGER(MetricManagerResource.class),
+
         /** */
         FILESYSTEM_RESOURCE(FileSystemResource.class);
 
@@ -537,7 +541,8 @@ public enum AnnotationSet {
             ResourceAnnotation.SPRING,
             ResourceAnnotation.IGNITE_INSTANCE,
             ResourceAnnotation.LOGGER,
-            ResourceAnnotation.SERVICE
+            ResourceAnnotation.SERVICE,
+            ResourceAnnotation.METRIC_MANAGER
         ),
 
         /** */

modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceProcessor.java

@@ -74,6 +74,8 @@ public GridResourceProcessor(GridKernalContext ctx) {
             new GridResourceLoggerInjector(ctx.config().getGridLogger());
         injectorByAnnotation[GridResourceIoc.ResourceAnnotation.IGNITE_INSTANCE.ordinal()] =
             new GridResourceBasicInjector<>(ctx.grid());
+        injectorByAnnotation[GridResourceIoc.ResourceAnnotation.METRIC_MANAGER.ordinal()] =
+            new GridResourceSupplierInjector<>(ctx::metric);
     }
 
     /** {@inheritDoc} */

modules/core/src/main/java/org/apache/ignite/internal/processors/resource/GridResourceSupplierInjector.java

@@ -0,0 +1,64 @@
+/*
+ * 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.internal.processors.resource;
+
+import java.util.function.Supplier;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.managers.deployment.GridDeployment;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Simple injector which wraps resource object supplier.
+ *
+ * @param <T> Type of injected resource.
+ */
+class GridResourceSupplierInjector<T> implements GridResourceInjector {
+    /** Resource to inject. */
+    private final Supplier<T> supplier;
+
+    /**
+     * Creates injector.
+     *
+     * @param supplier Resource supplier.
+     */
+    GridResourceSupplierInjector(Supplier<T> supplier) {
+        this.supplier = supplier;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void inject(GridResourceField field, Object target, Class<?> depCls, GridDeployment dep)
+        throws IgniteCheckedException {
+        GridResourceUtils.inject(field.getField(), target, supplier.get());
+    }
+
+    /** {@inheritDoc} */
+    @Override public void inject(GridResourceMethod mtd, Object target, Class<?> depCls, GridDeployment dep)
+        throws IgniteCheckedException {
+        GridResourceUtils.inject(mtd.getMethod(), target, supplier.get());
+    }
+
+    /** {@inheritDoc} */
+    @Override public void undeploy(GridDeployment dep) {
+        /* No-op. There is no cache. */
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(GridResourceSupplierInjector.class, this);
+    }
+}

modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioMetricsListener.java → modules/core/src/main/java/org/apache/ignite/internal/resources/MetricManagerResource.java

@@ -15,21 +15,18 @@
  * limitations under the License.
  */
 
-package org.apache.ignite.internal.util.nio;
+package org.apache.ignite.internal.resources;
 
-import java.util.EventListener;
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
 
-/**
- * Metrics listener for NIO communication.
- */
-public interface GridNioMetricsListener extends EventListener {
-    /**
-     * @param bytesCnt Number of sent bytes.
-     */
-    public void onBytesSent(int bytesCnt);
-
-    /**
-     * @param bytesCnt Number of received bytes.
-     */
-    public void onBytesReceived(int bytesCnt);
+/** */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD, ElementType.FIELD})
+public @interface MetricManagerResource {
+    // No-op.
 }

modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcToNioAdapter.java

@@ -26,20 +26,26 @@
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.internal.processors.metric.MetricRegistry;
+import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
 import org.apache.ignite.internal.util.nio.GridNioFilter;
 import org.apache.ignite.internal.util.nio.GridNioFilterAdapter;
 import org.apache.ignite.internal.util.nio.GridNioFilterChain;
 import org.apache.ignite.internal.util.nio.GridNioFinishedFuture;
 import org.apache.ignite.internal.util.nio.GridNioFuture;
 import org.apache.ignite.internal.util.nio.GridNioMessageWriterFactory;
-import org.apache.ignite.internal.util.nio.GridNioMetricsListener;
 import org.apache.ignite.internal.util.nio.GridNioServerListener;
 import org.apache.ignite.internal.util.nio.GridNioSession;
 import org.apache.ignite.internal.util.nio.GridNioSessionImpl;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.plugin.extensions.communication.Message;
 
+import static org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_DESC;
+import static org.apache.ignite.internal.util.nio.GridNioServer.RECEIVED_BYTES_METRIC_NAME;
+import static org.apache.ignite.internal.util.nio.GridNioServer.SENT_BYTES_METRIC_DESC;
+import static org.apache.ignite.internal.util.nio.GridNioServer.SENT_BYTES_METRIC_NAME;
+
 /**
  * Allows to re-use existing {@link GridNioFilter}s on IPC (specifically shared memory IPC)
  * communications.
@@ -63,25 +69,32 @@ public class IpcToNioAdapter<T> {
     /** */
     private final ByteBuffer writeBuf;
 
-    /** */
-    private final GridNioMetricsListener metricsLsnr;
+    /** Received bytes count metric. */
+    private final LongAdderMetric rcvdBytesCntMetric;
+
+    /** Sent bytes count metric. */
+    private final LongAdderMetric sentBytesCntMetric;
 
     /** */
     private final GridNioMessageWriterFactory writerFactory;
 
     /**
-     * @param metricsLsnr Metrics listener.
+     * @param mreg Metrics registry.
      * @param log Log.
      * @param endp Endpoint.
      * @param lsnr Listener.
      * @param writerFactory Writer factory.
      * @param filters Filters.
      */
-    public IpcToNioAdapter(GridNioMetricsListener metricsLsnr, IgniteLogger log, IpcEndpoint endp,
-        GridNioServerListener<T> lsnr, GridNioMessageWriterFactory writerFactory, GridNioFilter... filters) {
-        assert metricsLsnr != null;
+    public IpcToNioAdapter(MetricRegistry mreg, IgniteLogger log, IpcEndpoint endp,
+        GridNioServerListener<T> lsnr, GridNioMessageWriterFactory writerFactory, GridNioFilter... filters
+    ) {
+        assert mreg != null;
+
+        rcvdBytesCntMetric = mreg.longAdderMetric(RECEIVED_BYTES_METRIC_NAME, RECEIVED_BYTES_METRIC_DESC);
+
+        sentBytesCntMetric = mreg.longAdderMetric(SENT_BYTES_METRIC_NAME, SENT_BYTES_METRIC_DESC);
 
-        this.metricsLsnr = metricsLsnr;
         this.endp = endp;
         this.writerFactory = writerFactory;
 
@@ -116,7 +129,7 @@ public void serve() throws InterruptedException {
                 int read = in.read(readBuf.array(), pos, readBuf.remaining());
 
                 if (read > 0) {
-                    metricsLsnr.onBytesReceived(read);
+                    rcvdBytesCntMetric.add(read);
 
                     readBuf.position(0);
                     readBuf.limit(pos + read);
@@ -167,7 +180,7 @@ private GridNioFuture<?> send(Message msg) {
         try {
             int cnt = U.writeMessageFully(msg, endp.outputStream(), writeBuf, writerFactory.writer(ses));
 
-            metricsLsnr.onBytesSent(cnt);
+            sentBytesCntMetric.add(cnt);
         }
         catch (IOException | IgniteCheckedException e) {
             return new GridNioFinishedFuture<Object>(e);

modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridAbstractCommunicationClient.java

@@ -20,7 +20,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.Nullable;
 
 /**
  * Implements basic lifecycle for communication clients.
@@ -32,19 +31,14 @@ public abstract class GridAbstractCommunicationClient implements GridCommunicati
     /** Reservations. */
     private final AtomicBoolean closed = new AtomicBoolean();
 
-    /** Metrics listener. */
-    protected final GridNioMetricsListener metricsLsnr;
-
     /** */
     private final int connIdx;
 
     /**
      * @param connIdx Connection index.
-     * @param metricsLsnr Metrics listener.
      */
-    protected GridAbstractCommunicationClient(int connIdx, @Nullable GridNioMetricsListener metricsLsnr) {
+    protected GridAbstractCommunicationClient(int connIdx) {
         this.connIdx = connIdx;
-        this.metricsLsnr = metricsLsnr;
     }
 
     /** {@inheritDoc} */