[#11278] Add gRPC transport and stream monitoring

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/GrpcReceiver.java

@@ -17,13 +17,18 @@
 package com.navercorp.pinpoint.collector.receiver.grpc;
 
 import com.navercorp.pinpoint.collector.receiver.BindAddress;
+import com.navercorp.pinpoint.collector.receiver.grpc.monitor.BasicMonitor;
+import com.navercorp.pinpoint.collector.receiver.grpc.monitor.EmptyMonitor;
+import com.navercorp.pinpoint.collector.receiver.grpc.monitor.Monitor;
 import com.navercorp.pinpoint.common.server.util.AddressFilter;
 import com.navercorp.pinpoint.common.util.Assert;
 import com.navercorp.pinpoint.common.util.CollectionUtils;
 import com.navercorp.pinpoint.grpc.channelz.ChannelzRegistry;
+import com.navercorp.pinpoint.grpc.server.ConnectionCountServerTransportFilter;
 import com.navercorp.pinpoint.grpc.server.MetadataServerTransportFilter;
 import com.navercorp.pinpoint.grpc.server.ServerFactory;
 import com.navercorp.pinpoint.grpc.server.ServerOption;
+import com.navercorp.pinpoint.grpc.server.StreamCountInterceptor;
 import com.navercorp.pinpoint.grpc.server.TransportMetadataFactory;
 import com.navercorp.pinpoint.grpc.server.TransportMetadataServerInterceptor;
 import com.navercorp.pinpoint.grpc.util.ServerUtils;
@@ -78,6 +83,9 @@
     private Server server;
     private ChannelzRegistry channelzRegistry;
 
+    private boolean enableMonitor = true;
+    private Monitor monitor;
+
 
     @Override
     public void afterPropertiesSet() throws Exception {
@@ -91,6 +99,11 @@
         Objects.requireNonNull(this.addressFilter, "addressFilter");
         Assert.isTrue(CollectionUtils.hasLength(this.serviceList), "serviceList must not be empty");
         Objects.requireNonNull(this.serverOption, "serverOption");
+        if (enableMonitor) {
+            this.monitor = new BasicMonitor(beanName + "-Monitor");
+        } else {
+            this.monitor = new EmptyMonitor();
+        }
 
         if (sslContext != null) {
             this.serverFactory = new ServerFactory(beanName, this.bindAddress.getIp(), this.bindAddress.getPort(), this.executor, this.serverCallExecutorSupplier, serverOption, sslContext);
@@ -101,6 +114,9 @@
         ServerTransportFilter permissionServerTransportFilter = new PermissionServerTransportFilter(this.beanName, addressFilter);
         this.serverFactory.addTransportFilter(permissionServerTransportFilter);
 
+        ConnectionCountServerTransportFilter countFilter = new ConnectionCountServerTransportFilter();
+        this.serverFactory.addTransportFilter(countFilter);
+
         TransportMetadataFactory transportMetadataFactory = new TransportMetadataFactory(beanName);
         // Mandatory interceptor
         final ServerTransportFilter metadataTransportFilter = new MetadataServerTransportFilter(transportMetadataFactory);
@@ -116,6 +132,9 @@
         ServerInterceptor transportMetadataServerInterceptor = new TransportMetadataServerInterceptor();
         this.serverFactory.addInterceptor(transportMetadataServerInterceptor);
 
+        StreamCountInterceptor streamCountInterceptor = new StreamCountInterceptor();
+        this.serverFactory.addInterceptor(streamCountInterceptor);
+
         if (CollectionUtils.hasLength(serverInterceptorList)) {
             for (ServerInterceptor serverInterceptor : serverInterceptorList) {
                 this.serverFactory.addInterceptor(serverInterceptor);
@@ -125,6 +144,10 @@
             this.serverFactory.setChannelzRegistry(channelzRegistry);
         }
 
+        this.monitor.register(() -> {
+            logger.info("{} CurrentTransport:{}, CurrentGrpcStream:{}", beanName, countFilter.getCurrentConnection(), streamCountInterceptor.getCurrentStream());
+        });
+
         // Add service
         addService();
 
@@ -171,6 +194,8 @@
             logger.info("Destroy {} server {}", this.beanName, this.server);
         }
 
+        monitor.close();
+
         shutdownServer();
 
         for (Object bindableService : serviceList) {
@@ -242,4 +267,7 @@
         this.channelzRegistry = Objects.requireNonNull(channelzRegistry, "channelzRegistry");
     }
 
+    public void setEnableMonitor(boolean enableMonitor) {
+        this.enableMonitor = enableMonitor;
+    }
 }

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/BasicMonitor.java

@@ -0,0 +1,30 @@
+package com.navercorp.pinpoint.collector.receiver.grpc.monitor;
+
+import java.util.Objects;
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class BasicMonitor implements Monitor {
+    private final Timer timer;
+
+    public BasicMonitor(String name) {
+        this.timer = new Timer(name, true);
+    }
+
+    public void register(Runnable job) {
+        Objects.requireNonNull(job, "job");
+        this.timer.schedule(new TimerTask() {
+            @Override
+            public void run() {
+                job.run();
+            }
+        }, 60_000, 60_000);
+    }
+
+
+    @Override
+    public void close() {
+        this.timer.purge();
+        this.timer.cancel();
+    }
+}

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/EmptyMonitor.java

@@ -0,0 +1,14 @@
+package com.navercorp.pinpoint.collector.receiver.grpc.monitor;
+
+public class EmptyMonitor implements Monitor {
+
+    public EmptyMonitor() {
+    }
+
+    public void register(Runnable job) {
+    }
+
+    @Override
+    public void close() {
+    }
+}

collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/Monitor.java

@@ -0,0 +1,10 @@
+package com.navercorp.pinpoint.collector.receiver.grpc.monitor;
+
+import java.io.Closeable;
+
+public interface Monitor extends Closeable {
+
+    void register(Runnable job);
+
+    void close();
+}

grpc/src/main/java/com/navercorp/pinpoint/grpc/server/ConnectionCountServerTransportFilter.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 NAVER Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.grpc.server;
+
+import io.grpc.Attributes;
+import io.grpc.ServerTransportFilter;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class ConnectionCountServerTransportFilter extends ServerTransportFilter {
+
+    private final AtomicLong currentConnection = new AtomicLong();
+
+    public ConnectionCountServerTransportFilter() {
+    }
+
+
+    @Override
+    public Attributes transportReady(Attributes transportAttrs) {
+        currentConnection.incrementAndGet();
+        return transportAttrs;
+    }
+
+    @Override
+    public void transportTerminated(Attributes transportAttrs) {
+        currentConnection.decrementAndGet();
+    }
+
+    public long getCurrentConnection() {
+        return currentConnection.get();
+    }
+}

grpc/src/main/java/com/navercorp/pinpoint/grpc/server/StreamCountInterceptor.java

@@ -0,0 +1,39 @@
+package com.navercorp.pinpoint.grpc.server;
+
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class StreamCountInterceptor implements ServerInterceptor {
+    private final AtomicLong currentStream = new AtomicLong();
+    public StreamCountInterceptor() {
+    }
+
+    @Override
+    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
+        currentStream.incrementAndGet();
+        ServerCall.Listener<ReqT> listener = next.startCall(call, headers);
+        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listener) {
+            @Override
+            public void onCancel() {
+                currentStream.decrementAndGet();
+                super.onCancel();
+            }
+
+            @Override
+            public void onComplete() {
+                currentStream.decrementAndGet();
+                super.onComplete();
+            }
+
+        };
+    }
+
+    public long getCurrentStream() {
+        return currentStream.get();
+    }
+}