From 93f5d6a48876cefd94ed7f5053df2e7c1469bd3d Mon Sep 17 00:00:00 2001 From: emeroad Date: Wed, 24 Jul 2024 16:24:11 +0900 Subject: [PATCH] [#11278] Add gRPC transport and stream monitoring --- .../collector/receiver/grpc/GrpcReceiver.java | 28 +++++++++++ .../receiver/grpc/monitor/BasicMonitor.java | 30 ++++++++++++ .../receiver/grpc/monitor/EmptyMonitor.java | 14 ++++++ .../receiver/grpc/monitor/Monitor.java | 10 ++++ .../ConnectionCountServerTransportFilter.java | 46 +++++++++++++++++++ .../grpc/server/StreamCountInterceptor.java | 39 ++++++++++++++++ 6 files changed, 167 insertions(+) create mode 100644 collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/BasicMonitor.java create mode 100644 collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/EmptyMonitor.java create mode 100644 collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/Monitor.java create mode 100644 grpc/src/main/java/com/navercorp/pinpoint/grpc/server/ConnectionCountServerTransportFilter.java create mode 100644 grpc/src/main/java/com/navercorp/pinpoint/grpc/server/StreamCountInterceptor.java diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/GrpcReceiver.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/GrpcReceiver.java index 70721455c592..4cde7e0604ae 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/GrpcReceiver.java +++ b/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 @@ public class GrpcReceiver implements InitializingBean, DisposableBean, BeanNameA private Server server; private ChannelzRegistry channelzRegistry; + private boolean enableMonitor = true; + private Monitor monitor; + @Override public void afterPropertiesSet() throws Exception { @@ -91,6 +99,11 @@ public void afterPropertiesSet() throws Exception { 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 @@ public void afterPropertiesSet() throws Exception { 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 @@ public void afterPropertiesSet() throws Exception { 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 @@ public void afterPropertiesSet() throws Exception { this.serverFactory.setChannelzRegistry(channelzRegistry); } + this.monitor.register(() -> { + logger.info("{} CurrentTransport:{}, CurrentGrpcStream:{}", beanName, countFilter.getCurrentConnection(), streamCountInterceptor.getCurrentStream()); + }); + // Add service addService(); @@ -171,6 +194,8 @@ public void destroy() throws Exception { logger.info("Destroy {} server {}", this.beanName, this.server); } + monitor.close(); + shutdownServer(); for (Object bindableService : serviceList) { @@ -242,4 +267,7 @@ public void setChannelzRegistry(ChannelzRegistry channelzRegistry) { this.channelzRegistry = Objects.requireNonNull(channelzRegistry, "channelzRegistry"); } + public void setEnableMonitor(boolean enableMonitor) { + this.enableMonitor = enableMonitor; + } } \ No newline at end of file diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/BasicMonitor.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/BasicMonitor.java new file mode 100644 index 000000000000..459f5fb4f3b4 --- /dev/null +++ b/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(); + } +} \ No newline at end of file diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/EmptyMonitor.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/EmptyMonitor.java new file mode 100644 index 000000000000..6f443e35a998 --- /dev/null +++ b/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() { + } +} \ No newline at end of file diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/Monitor.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/grpc/monitor/Monitor.java new file mode 100644 index 000000000000..8470993543a9 --- /dev/null +++ b/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(); +} diff --git a/grpc/src/main/java/com/navercorp/pinpoint/grpc/server/ConnectionCountServerTransportFilter.java b/grpc/src/main/java/com/navercorp/pinpoint/grpc/server/ConnectionCountServerTransportFilter.java new file mode 100644 index 000000000000..ca869c57f57a --- /dev/null +++ b/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(); + } +} \ No newline at end of file diff --git a/grpc/src/main/java/com/navercorp/pinpoint/grpc/server/StreamCountInterceptor.java b/grpc/src/main/java/com/navercorp/pinpoint/grpc/server/StreamCountInterceptor.java new file mode 100644 index 000000000000..54c5a35e5ae7 --- /dev/null +++ b/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 ServerCall.Listener interceptCall(ServerCall call, Metadata headers, ServerCallHandler next) { + currentStream.incrementAndGet(); + ServerCall.Listener listener = next.startCall(call, headers); + return new ForwardingServerCallListener.SimpleForwardingServerCallListener(listener) { + @Override + public void onCancel() { + currentStream.decrementAndGet(); + super.onCancel(); + } + + @Override + public void onComplete() { + currentStream.decrementAndGet(); + super.onComplete(); + } + + }; + } + + public long getCurrentStream() { + return currentStream.get(); + } +}