[pinpoint-apm#8762] apply throttled logger, add collector properties …

…tests, add throttled logger tests
donghun-cho · Apr 12, 2022 · ab9a874 · ab9a874
1 parent 876d852
commit ab9a874
Show file tree

Hide file tree

Showing 14 changed files with 122 additions and 11 deletions.
diff --git a/...r/src/main/java/com/navercorp/pinpoint/collector/grpc/config/GrpcStreamConfiguration.java b/...r/src/main/java/com/navercorp/pinpoint/collector/grpc/config/GrpcStreamConfiguration.java
@@ -23,15 +23,17 @@ public class GrpcStreamConfiguration {
     private final int schedulerRecoveryMessageCount;
 
     private final long idleTimeout;
+    private final long throttledLoggerRatio;
 
     GrpcStreamConfiguration(int schedulerThreadSize, int callInitRequestCount,
-                                   int schedulerPeriodMillis, int schedulerRecoveryMessageCount,
-                                   long idleTimeout) {
+                            int schedulerPeriodMillis, int schedulerRecoveryMessageCount,
+                            long idleTimeout, long throttledLoggerRatio) {
         this.schedulerThreadSize = schedulerThreadSize;
         this.callInitRequestCount = callInitRequestCount;
         this.schedulerPeriodMillis = schedulerPeriodMillis;
         this.schedulerRecoveryMessageCount = schedulerRecoveryMessageCount;
         this.idleTimeout = idleTimeout;
+        this.throttledLoggerRatio = throttledLoggerRatio;
     }
 
     public int getSchedulerThreadSize() {
@@ -54,6 +56,10 @@ public long getIdleTimeout() {
         return idleTimeout;
     }
 
+    public long getThrottledLoggerRatio() {
+        return throttledLoggerRatio;
+    }
+
     public static Builder newBuilder() {
         return new Builder();
     }
@@ -64,6 +70,7 @@ public static class Builder {
         private int schedulerPeriodMillis =  64;
         private int schedulerRecoveryMessageCount = 10;
         private long idleTimeout = -1;
+        private long throttledLoggerRatio = 1;
 
 
         public int getSchedulerThreadSize() {
@@ -106,9 +113,17 @@ public void setIdleTimeout(long idleTimeout) {
             this.idleTimeout = idleTimeout;
         }
 
+        public long getThrottledLoggerRatio() {
+            return throttledLoggerRatio;
+        }
+
+        public void setThrottledLoggerRatio(long throttledLoggerRatio) {
+            this.throttledLoggerRatio = throttledLoggerRatio;
+        }
+
         public GrpcStreamConfiguration build() {
             return new GrpcStreamConfiguration(this.schedulerThreadSize, this.callInitRequestCount,
-                    this.schedulerPeriodMillis, this.schedulerRecoveryMessageCount, this.idleTimeout);
+                    this.schedulerPeriodMillis, this.schedulerRecoveryMessageCount, this.idleTimeout, this.throttledLoggerRatio);
         }
     }
 
@@ -120,6 +135,7 @@ public String toString() {
                 ", schedulerPeriodMillis=" + schedulerPeriodMillis +
                 ", schedulerRecoveryMessageCount=" + schedulerRecoveryMessageCount +
                 ", idleTimeout=" + idleTimeout +
+                ", throttledLoggerRatio=" + throttledLoggerRatio +
                 '}';
     }
 }
diff --git a/...corp/pinpoint/collector/receiver/grpc/service/StreamExecutorServerInterceptorFactory.java b/...corp/pinpoint/collector/receiver/grpc/service/StreamExecutorServerInterceptorFactory.java
@@ -43,6 +43,7 @@ public class StreamExecutorServerInterceptorFactory implements FactoryBean<Serve
     private final int periodMillis;
     private final int recoveryMessagesCount;
     private final long idleTimeout;
+    private final long throttledLoggerRatio;
 
     public StreamExecutorServerInterceptorFactory(Executor executor,
                                                   ScheduledExecutorService scheduledExecutorService,
@@ -57,6 +58,7 @@ public StreamExecutorServerInterceptorFactory(Executor executor,
         Assert.isTrue(periodMillis > 0, "periodMillis must be positive");
         this.recoveryMessagesCount = streamConfiguration.getSchedulerRecoveryMessageCount();
         this.idleTimeout = streamConfiguration.getIdleTimeout();
+        this.throttledLoggerRatio = streamConfiguration.getThrottledLoggerRatio();
     }
 
     @Override
@@ -71,7 +73,7 @@ public Future<?> schedule(Runnable command) {
         RejectedExecutionListenerFactory listenerFactory = new RejectedExecutionListenerFactory(this.beanName, recoveryMessagesCount, idleTimeoutFactory);
 
         return new StreamExecutorServerInterceptor(this.beanName, this.executor, initRequestCount,
-                scheduledExecutor, listenerFactory);
+                scheduledExecutor, listenerFactory, throttledLoggerRatio);
     }
 
     @Override

diff --git a/collector/src/main/resources/profiles/local/pinpoint-collector-grpc.properties b/collector/src/main/resources/profiles/local/pinpoint-collector-grpc.properties
@@ -30,6 +30,7 @@ collector.receiver.grpc.stat.stream.scheduler_thread_size=1
 collector.receiver.grpc.stat.stream.scheduler_period_millis=1000
 collector.receiver.grpc.stat.stream.call_init_request_count=100
 collector.receiver.grpc.stat.stream.scheduler_recovery_message_count=100
+collector.receiver.grpc.stat.stream.throttled_logger_ratio=100
 
 
 # Span
@@ -51,6 +52,7 @@ collector.receiver.grpc.span.stream.scheduler_period_millis=1000
 collector.receiver.grpc.span.stream.call_init_request_count=100
 collector.receiver.grpc.span.stream.scheduler_recovery_message_count=100
 collector.receiver.grpc.span.stream.idletimeout=-1
+collector.receiver.grpc.span.stream.throttled_logger_ratio=100
 
 
 ### For ssl config

diff --git a/collector/src/main/resources/profiles/release/pinpoint-collector-grpc.properties b/collector/src/main/resources/profiles/release/pinpoint-collector-grpc.properties
@@ -35,6 +35,7 @@ collector.receiver.grpc.stat.stream.scheduler_thread_size=1
 collector.receiver.grpc.stat.stream.scheduler_period_millis=1000
 collector.receiver.grpc.stat.stream.call_init_request_count=100
 collector.receiver.grpc.stat.stream.scheduler_recovery_message_count=100
+collector.receiver.grpc.stat.stream.throttled_logger_ratio=100
 
 
 # Span
@@ -58,6 +59,7 @@ collector.receiver.grpc.span.stream.scheduler_period_millis=1000
 collector.receiver.grpc.span.stream.call_init_request_count=100
 collector.receiver.grpc.span.stream.scheduler_recovery_message_count=100
 collector.receiver.grpc.span.stream.idletimeout=-1
+collector.receiver.grpc.span.stream.throttled_logger_ratio=100
 
 
 ### For ssl config

diff --git a/.../java/com/navercorp/pinpoint/collector/grpc/config/GrpcSpanReceiverConfigurationTest.java b/.../java/com/navercorp/pinpoint/collector/grpc/config/GrpcSpanReceiverConfigurationTest.java
@@ -55,6 +55,7 @@ public void properties() {
         assertEquals(3, configuration.getStreamConfiguration().getSchedulerThreadSize());
         assertEquals(3, configuration.getStreamConfiguration().getSchedulerPeriodMillis());
         assertEquals(3, configuration.getStreamConfiguration().getCallInitRequestCount());
+        assertEquals(3, configuration.getStreamConfiguration().getThrottledLoggerRatio());
 
     }
 

diff --git a/.../java/com/navercorp/pinpoint/collector/grpc/config/GrpcStatReceiverConfigurationTest.java b/.../java/com/navercorp/pinpoint/collector/grpc/config/GrpcStatReceiverConfigurationTest.java
@@ -48,6 +48,7 @@ public void properties() {
         assertEquals(2, configuration.getStreamConfiguration().getSchedulerThreadSize());
         assertEquals(2, configuration.getStreamConfiguration().getSchedulerPeriodMillis());
         assertEquals(2, configuration.getStreamConfiguration().getCallInitRequestCount());
+        assertEquals(2, configuration.getStreamConfiguration().getThrottledLoggerRatio());
 
     }
 

diff --git a/collector/src/test/resources/test-pinpoint-collector.properties b/collector/src/test/resources/test-pinpoint-collector.properties
@@ -152,6 +152,7 @@ collector.receiver.grpc.stat.worker.executor.monitor_enable=false
 collector.receiver.grpc.stat.stream.scheduler_thread_size=2
 collector.receiver.grpc.stat.stream.scheduler_period_millis=2
 collector.receiver.grpc.stat.stream.call_init_request_count=2
+collector.receiver.grpc.stat.stream.throttled_logger_ratio=2
 
 # Server Option
 collector.receiver.grpc.stat.keepalive_time_millis=2
@@ -181,6 +182,7 @@ collector.receiver.grpc.span.worker.executor.monitor_enable=false
 collector.receiver.grpc.span.stream.scheduler_thread_size=3
 collector.receiver.grpc.span.stream.scheduler_period_millis=3
 collector.receiver.grpc.span.stream.call_init_request_count=3
+collector.receiver.grpc.span.stream.throttled_logger_ratio=3
 
 # Server Option
 collector.receiver.grpc.span.keepalive_time_millis=3

diff --git a/commons-profiler/pom.xml b/commons-profiler/pom.xml
@@ -42,7 +42,6 @@
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>
             <artifactId>log4j-api</artifactId>
-            <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.logging.log4j</groupId>

diff --git a/...inpoint/grpc/logging/ThrottledLogger.java → ...mon/profiler/logging/ThrottledLogger.java b/...inpoint/grpc/logging/ThrottledLogger.java → ...mon/profiler/logging/ThrottledLogger.java
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package com.navercorp.pinpoint.grpc.logging;
+package com.navercorp.pinpoint.common.profiler.logging;
 
 import org.apache.logging.log4j.Logger;
 
@@ -36,6 +36,9 @@ public static ThrottledLogger getLogger(Logger logger, long ratio) {
 
     private ThrottledLogger(Logger logger, long ratio) {
         this.logger = Objects.requireNonNull(logger, "logger");
+        if (ratio < 1) {
+            ratio = 1;
+        }
         this.ratio = ratio;
     }
 

diff --git a/...ler/src/test/java/com/navercorp/pinpoint/common/profiler/logging/ThrottledLoggerTest.java b/...ler/src/test/java/com/navercorp/pinpoint/common/profiler/logging/ThrottledLoggerTest.java
@@ -0,0 +1,57 @@
+package com.navercorp.pinpoint.common.profiler.logging;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ThrottledLoggerTest {
+
+    @Test
+    public void LoggerTest() {
+        final Logger logger = LogManager.getLogger(this.getClass());
+
+        final ThrottledLogger throttledLogger = ThrottledLogger.getLogger(logger, 1);
+
+        throttledLogger.info("test logger info: logger");
+        throttledLogger.debug("test logger debug: logger");
+        throttledLogger.warn("test logger warn: logger");
+    }
+
+    @Test
+    public void throttledLoggerTest() {
+        final Logger logger = LogManager.getLogger(this.getClass());
+        final ThrottledLogger throttledLogger = ThrottledLogger.getLogger(logger, 2);
+
+        //only odd number logs should be shown
+        throttledLogger.debug("test tlogger debug1: throttled");
+        throttledLogger.debug("test tLogger debug2: throttled ERROR");
+        throttledLogger.debug("test tlogger debug3: throttled");
+        throttledLogger.debug("test tLogger debug4: throttled ERROR");
+    }
+
+    @Test
+    public void throttledLoggerZeroRatioTest() {
+        final Logger logger = LogManager.getLogger(this.getClass());
+        final ThrottledLogger throttledLogger = ThrottledLogger.getLogger(logger, 0);
+
+        throttledLogger.info("test tLogger info: zero ratio");
+        throttledLogger.debug("test tLogger debug: zero ratio");
+        throttledLogger.warn("test tLogger warn: zero ratio");
+    }
+
+    @Test
+    public void throttledLoggerNegativeRatioTest() {
+        final Logger logger = LogManager.getLogger(this.getClass());
+        final ThrottledLogger throttledLogger = ThrottledLogger.getLogger(logger, -1);
+
+        throttledLogger.info("test tLogger info: negative ratio");
+        throttledLogger.debug("test tLogger debug: negative ratio");
+        throttledLogger.warn("test tLogger warn: negative ratio");
+    }
+}
diff --git a/commons-profiler/src/test/resources/log4j2-test.xml b/commons-profiler/src/test/resources/log4j2-test.xml
@@ -0,0 +1,22 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Configuration status="INFO">
+    <Properties>
+        <Property name="console_message_pattern">%d{MM-dd HH:mm:ss.sss} [%15.15t] %clr{%-5level} %clr{%-40.40logger{1.}}{cyan}:%3L -- %msg{nolookups}%n</Property>
+        <Property name="file_message_pattern">%d{MM-dd HH:mm:ss.sss} [%15.15t] %-5level %-40.40logger{1.}:%3L -- %msg{nolookups}%n</Property>
+    </Properties>
+
+    <Appenders>
+        <Console name="console" target="system_out">
+            <PatternLayout pattern="${file_message_pattern}"/>
+        </Console>
+    </Appenders>
+
+    <Loggers>
+        <Logger name="com.navercorp.pinpoint" level="DEBUG" additivity="false">
+            <AppenderRef ref="console"/>
+        </Logger>
+        <Root level="DEBUG">
+            <AppenderRef ref="console"/>
+        </Root>
+    </Loggers>
+</Configuration>
diff --git a/...main/java/com/navercorp/pinpoint/grpc/client/interceptor/LoggingDiscardEventListener.java b/...main/java/com/navercorp/pinpoint/grpc/client/interceptor/LoggingDiscardEventListener.java
@@ -16,7 +16,7 @@
 
 package com.navercorp.pinpoint.grpc.client.interceptor;
 
-import com.navercorp.pinpoint.grpc.logging.ThrottledLogger;
+import com.navercorp.pinpoint.common.profiler.logging.ThrottledLogger;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
 

diff --git a/.../java/com/navercorp/pinpoint/grpc/server/flowcontrol/StreamExecutorServerInterceptor.java b/.../java/com/navercorp/pinpoint/grpc/server/flowcontrol/StreamExecutorServerInterceptor.java
@@ -16,6 +16,8 @@
 
 package com.navercorp.pinpoint.grpc.server.flowcontrol;
 
+import com.navercorp.pinpoint.bootstrap.config.Value;
+import com.navercorp.pinpoint.common.profiler.logging.ThrottledLogger;
 import com.navercorp.pinpoint.common.util.Assert;
 import com.navercorp.pinpoint.grpc.Header;
 import io.grpc.Context;
@@ -36,13 +38,14 @@
  */
 public class StreamExecutorServerInterceptor implements ServerInterceptor {
     private final Logger logger = LogManager.getLogger(this.getClass());
+    private final ThrottledLogger throttledLogger;
     private final String name;
     private final Executor executor;
     private final int initNumMessages;
     private final StreamExecutorRejectedExecutionRequestScheduler scheduler;
 
-    public StreamExecutorServerInterceptor(String name, final Executor executor, final int initNumMessages, final ScheduledExecutor scheduledExecutor,
-                                           RejectedExecutionListenerFactory listenerFactory) {
+    public StreamExecutorServerInterceptor(String name, final Executor executor, final int initNumMessages,
+                                           final ScheduledExecutor scheduledExecutor, RejectedExecutionListenerFactory listenerFactory, final long throttledLoggerRatio) {
         this.name = Objects.requireNonNull(name, "name");
 
         Objects.requireNonNull(executor, "executor");
@@ -55,6 +58,7 @@ public StreamExecutorServerInterceptor(String name, final Executor executor, fin
         Objects.requireNonNull(listenerFactory, "listenerFactory");
 
         this.scheduler = new StreamExecutorRejectedExecutionRequestScheduler(scheduledExecutor, listenerFactory);
+        throttledLogger = ThrottledLogger.getLogger(logger, throttledLoggerRatio);
     }
 
     @Override
@@ -86,7 +90,7 @@ public void run() {
                 } catch (RejectedExecutionException ree) {
                     // Defense code, need log ?
                     scheduleListener.onRejectedExecution();
-                    logger.info("Failed to request. Rejected execution, count={}", scheduleListener.getRejectedExecutionCount());
+                    throttledLogger.info("Failed to request. Rejected execution, count={}", scheduleListener.getRejectedExecutionCount());
                 }
             }
 

diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/grpc/GrpcDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/grpc/GrpcDataSender.java
@@ -19,10 +19,10 @@
 import java.util.Objects;
 import com.navercorp.pinpoint.common.profiler.concurrent.ExecutorFactory;
 import com.navercorp.pinpoint.common.profiler.concurrent.PinpointThreadFactory;
+import com.navercorp.pinpoint.common.profiler.logging.ThrottledLogger;
 import com.navercorp.pinpoint.grpc.ExecutorUtils;
 import com.navercorp.pinpoint.grpc.ManagedChannelUtils;
 import com.navercorp.pinpoint.grpc.client.ChannelFactory;
-import com.navercorp.pinpoint.grpc.logging.ThrottledLogger;
 import com.navercorp.pinpoint.profiler.context.thrift.MessageConverter;
 import com.navercorp.pinpoint.profiler.sender.DataSender;