feat:Add SpRateLimiter and SpMemoryManager (#3796)

Author: Yuukadesu

pom.xml

@@ -61,6 +61,7 @@
         <eclipse.milo.version>0.6.16</eclipse.milo.version>
         <error-prone.version>2.10.0</error-prone.version>
         <file-management.version>3.1.0</file-management.version>
+        <findbugs.version>3.0.2</findbugs.version>
         <flink.version>1.13.5</flink.version>
         <fogsy-qudt.version>1.0</fogsy-qudt.version>
         <geojson-jackson.version>1.14</geojson-jackson.version>
@@ -240,6 +241,12 @@
                 <artifactId>gson</artifactId>
                 <version>${gson.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.google.code.findbugs</groupId>
+                <artifactId>jsr305</artifactId>
+                <version>${findbugs.version}</version>
+                <scope>compile</scope>
+            </dependency>
             <dependency>
                 <groupId>com.google.guava</groupId>
                 <artifactId>guava</artifactId>

streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java

@@ -170,10 +170,30 @@ public enum Envs {
   SP_LOGGING_FILE_PREFIX("SP_LOGGING_FILE_PREFIX", "streampipes"),
   SP_LOGGING_FILE_DIR("SP_LOGGING_FILE_DIR", "logs"),
   SP_LOGGING_FILE_PATTERN(
-      "SP_LOGGING_FILE_PATTERN", "%d{yyyy-MM-dd HH:mm:ss} %-5level %logger{36} - %msg%n"
-  );
-
+      "SP_LOGGING_FILE_PATTERN",
+      "%d{yyyy-MM-dd HH:mm:ss} %-5level %logger{36} - %msg%n"
+  ),
 
+  //SpRateLimiter
+  SP_RATE_LIMITER_DEFAULT_WARMUP_PERIOD("SP_RATE_LIMITER_DEFAULT_WARMUP_PERIOD", "1000"),
+  SP_RATE_LIMITER_SCHEDULER_INITIAL_DELAY_SECONDS("SP_RATE_LIMITER_SCHEDULER_INITIAL_DELAY_SECONDS", "0"),
+  SP_RATE_LIMITER_SCHEDULER_PERIOD_SECONDS("SP_RATE_LIMITER_SCHEDULER_PERIOD_SECONDS", "15"),
+  SP_RATE_LIMITER_STATS_RESET_THRESHOLD("SP_RATE_LIMITER_STATS_RESET_THRESHOLD", "1000"),
+  SP_RATE_LIMITER_STATS_RESET_FACTOR("SP_RATE_LIMITER_STATS_RESET_FACTOR", "999"),
+  SP_RATE_LIMITER_STATS_RESET_DIVISOR("SP_RATE_LIMITER_STATS_RESET_DIVISOR", "1000"),
+  SP_RATE_LIMITER_SHUTDOWN_TIMEOUT_SECONDS("SP_RATE_LIMITER_SHUTDOWN_TIMEOUT_SECONDS", "5"),
+  SP_RATE_LIMITER_TIMEOUT_MS("SP_RATE_LIMITER_TIMEOUT_MS", "1000"),
+  SP_RATE_LIMITER_PERMITS_SET_PERCENTAGE("SP_RATE_LIMITER_PERMITS_SET_PERCENTAGE", "0.7"),
+
+  //SpMemoryManager
+  SP_MEMORY_MANAGER_DEFAULT_INITIAL_MEMORY("SP_MEMORY_MANAGER_DEFAULT_INITIAL_MEMORY", "1073741824"),
+  SP_MEMORY_MANAGER_WAIT_TIMEOUT_MS("SP_MEMORY_MANAGER_WAIT_TIMEOUT_MS", "1000"),
+  SP_MEMORY_SCHEDULER_INITIAL_DELAY_SECONDS("SP_MEMORY_SCHEDULER_INITIAL_DELAY_SECONDS", "0"),
+  SP_MEMORY_SCHEDULER_PERIOD_SECONDS("SP_MEMORY_SCHEDULER_PERIOD_SECONDS", "15"),
+  SP_MEMORY_BYTES_TO_MB("SP_MEMORY_BYTES_TO_MB", "1048576"),
+  SP_MEMORY_MANAGER_SHUTDOWN_TIMEOUT_SECONDS("SP_MEMORY_MANAGER_SHUTDOWN_TIMEOUT_SECONDS", "5"),
+  SP_MEMORY_MANAGER_USAGE_THRESHOLD("SP_MEMORY_MANAGER_USAGE_THRESHOLD", "0.9"),
+  SP_MEMORY_WARNING_THRESHOLD("SP_MEMORY_WARNING_THRESHOLD", "0.8");
 
   private final String envVariableName;
   private String defaultValue;

streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java

@@ -22,9 +22,10 @@
 import org.apache.streampipes.commons.environment.parser.OAuthConfigurationParser;
 import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable;
 import org.apache.streampipes.commons.environment.variable.DoubleEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.FloatEnvironmentVariable;
 import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.LongEnvironmentVariable;
 import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
-import org.apache.streampipes.commons.environment.variable.FloatEnvironmentVariable;
 
 import java.util.List;
 
@@ -456,7 +457,6 @@ public StringEnvironmentVariable getRetentionLocalDir() {
     return new StringEnvironmentVariable(Envs.SP_RETENTION_LOCAL_DIR);
   }
 
-
   @Override
   public DoubleEnvironmentVariable getDirMemoryResourceWeight() {
     return new DoubleEnvironmentVariable(Envs.DIR_MEMORY_RESOURCE_WEIGHT);
@@ -517,6 +517,91 @@ public BooleanEnvironmentVariable getLoadManagerEnable() {
     return new BooleanEnvironmentVariable(Envs.LOAD_MANAGER_ENABLE);
   }
 
+  @Override
+  public LongEnvironmentVariable getRateLimiterDefaultWarmupPeriod() {
+    return new LongEnvironmentVariable(Envs.SP_RATE_LIMITER_DEFAULT_WARMUP_PERIOD);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterSchedulerInitialDelaySeconds() {
+    return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_SCHEDULER_INITIAL_DELAY_SECONDS);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterSchedulerPeriodSeconds() {
+        return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_SCHEDULER_PERIOD_SECONDS);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterStatsResetThreshold() {
+    return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_STATS_RESET_THRESHOLD);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterStatsResetFactor() {
+    return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_STATS_RESET_FACTOR);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterStatsResetDivisor() {
+    return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_STATS_RESET_DIVISOR);
+  }
+
+  @Override
+  public IntEnvironmentVariable getRateLimiterShutdownTimeoutSeconds() {
+    return new IntEnvironmentVariable(Envs.SP_RATE_LIMITER_SHUTDOWN_TIMEOUT_SECONDS);
+  }
+
+  @Override
+  public LongEnvironmentVariable getRateLimiterTimeoutMs() {
+    return new LongEnvironmentVariable(Envs.SP_RATE_LIMITER_TIMEOUT_MS);
+  }
+
+  @Override
+  public DoubleEnvironmentVariable getRateLimiterPermitsSetPercentage() {
+      return new DoubleEnvironmentVariable(Envs.SP_RATE_LIMITER_PERMITS_SET_PERCENTAGE);
+  }
+
+  @Override
+  public LongEnvironmentVariable getMemoryManagerDefaultInitialMemory() {
+    return new LongEnvironmentVariable(Envs.SP_MEMORY_MANAGER_DEFAULT_INITIAL_MEMORY);
+  }
+
+  @Override
+  public LongEnvironmentVariable getMemoryManagerWaitTimeoutMs() {
+    return new LongEnvironmentVariable(Envs.SP_MEMORY_MANAGER_WAIT_TIMEOUT_MS);
+  }
+
+  @Override
+  public IntEnvironmentVariable getMemorySchedulerInitialDelaySeconds() {
+    return new IntEnvironmentVariable(Envs.SP_MEMORY_SCHEDULER_INITIAL_DELAY_SECONDS);
+  }
+
+  @Override
+  public IntEnvironmentVariable getMemorySchedulerPeriodSeconds() {
+    return new IntEnvironmentVariable(Envs.SP_MEMORY_SCHEDULER_PERIOD_SECONDS);
+  }
+
+  @Override
+  public LongEnvironmentVariable getMemoryBytesToMb() {
+    return new LongEnvironmentVariable(Envs.SP_MEMORY_BYTES_TO_MB);
+  }
+
+  @Override
+  public IntEnvironmentVariable getMemoryManagerShutdownTimeoutSeconds() {
+    return new IntEnvironmentVariable(Envs.SP_MEMORY_MANAGER_SHUTDOWN_TIMEOUT_SECONDS);
+  }
+
+  @Override
+  public DoubleEnvironmentVariable getMemoryManagerUsageThreshold() {
+    return new DoubleEnvironmentVariable(Envs.SP_MEMORY_MANAGER_USAGE_THRESHOLD);
+  }
+
+  @Override
+  public DoubleEnvironmentVariable getMemoryWarningThreshold() {
+    return new DoubleEnvironmentVariable(Envs.SP_MEMORY_WARNING_THRESHOLD);
+    }
+
   @Override
   public StringEnvironmentVariable getDatalakeSchedulerCron() {
     return new StringEnvironmentVariable(Envs.SP_DATALAKE_SCHEDULER_CRON);

streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java

@@ -18,7 +18,12 @@
 package org.apache.streampipes.commons.environment;
 
 import org.apache.streampipes.commons.environment.model.OAuthConfiguration;
-import org.apache.streampipes.commons.environment.variable.*;
+import org.apache.streampipes.commons.environment.variable.BooleanEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.DoubleEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.FloatEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.IntEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.LongEnvironmentVariable;
+import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
 
 import java.util.List;
 
@@ -234,5 +239,25 @@ public interface Environment {
 
   BooleanEnvironmentVariable getLoadManagerEnable();
 
+  //SpRateLimiter
+  LongEnvironmentVariable getRateLimiterDefaultWarmupPeriod();
+  IntEnvironmentVariable getRateLimiterSchedulerInitialDelaySeconds();
+  IntEnvironmentVariable getRateLimiterSchedulerPeriodSeconds();
+  IntEnvironmentVariable getRateLimiterStatsResetThreshold();
+  IntEnvironmentVariable getRateLimiterStatsResetFactor();
+  IntEnvironmentVariable getRateLimiterStatsResetDivisor();
+  IntEnvironmentVariable getRateLimiterShutdownTimeoutSeconds();
+  LongEnvironmentVariable getRateLimiterTimeoutMs();
+  DoubleEnvironmentVariable getRateLimiterPermitsSetPercentage();
+
+  //SpMemoryManager
+  LongEnvironmentVariable getMemoryManagerDefaultInitialMemory();
+  LongEnvironmentVariable getMemoryManagerWaitTimeoutMs();
+  IntEnvironmentVariable getMemorySchedulerInitialDelaySeconds();
+  IntEnvironmentVariable getMemorySchedulerPeriodSeconds();
+  LongEnvironmentVariable getMemoryBytesToMb();
+  IntEnvironmentVariable getMemoryManagerShutdownTimeoutSeconds();
+  DoubleEnvironmentVariable getMemoryManagerUsageThreshold();
+  DoubleEnvironmentVariable getMemoryWarningThreshold();
   StringEnvironmentVariable getDatalakeSchedulerCron();
 }

streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/variable/LongEnvironmentVariable.java

@@ -0,0 +1,31 @@
+/*
+ * 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.streampipes.commons.environment.variable;
+
+public class LongEnvironmentVariable extends EnvironmentVariable<Long> {
+
+  public LongEnvironmentVariable(org.apache.streampipes.commons.constants.Envs envVariable) {
+    super(envVariable);
+  }
+
+  @Override
+  public Long parse(String value) {
+    return Long.parseLong(value);
+  }
+}

streampipes-commons/src/main/java/org/apache/streampipes/commons/exceptions/SpHttpErrorStatusCode.java

@@ -6,7 +6,7 @@
  * (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
+ *    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,

streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/service/ElementServiceStats.java

@@ -20,8 +20,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Map;
-
 /**
  * Service Statistics
  */

streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/spmemorymanager/SpMemoryManagerMetrics.java

@@ -0,0 +1,44 @@
+/*
+ * 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.streampipes.commons.prometheus.spmemorymanager;
+
+import org.apache.streampipes.commons.prometheus.StreamPipesCollectorRegistry;
+
+import io.prometheus.client.Gauge;
+
+/**
+ * Memory Manager Metrics Manager
+ */
+public class SpMemoryManagerMetrics {
+    
+  public static final Gauge MEMORY_USED_BYTES = StreamPipesCollectorRegistry.registerGauge(
+        "sp_memory_used_bytes",
+        "Amount of memory used in bytes"
+  );
+
+  public static final Gauge MEMORY_ALLOCATION_RATE = StreamPipesCollectorRegistry.registerGauge(
+        "sp_memory_allocation_rate_bytes_per_second",
+        "Memory allocation rate in bytes per second"
+  );
+
+  public static void updateCoreMetrics(double memoryUsedBytes, double allocationRate) {
+    MEMORY_USED_BYTES.set(memoryUsedBytes);
+    MEMORY_ALLOCATION_RATE.set(allocationRate);
+  }
+}

streampipes-commons/src/main/java/org/apache/streampipes/commons/prometheus/spmemorymanager/SpMemoryManagerStats.java

@@ -0,0 +1,66 @@
+/*
+ * 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.streampipes.commons.prometheus.spmemorymanager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Memory Manager Statistics
+ */
+public class SpMemoryManagerStats {
+    
+  private static final Logger log = LoggerFactory.getLogger(SpMemoryManagerStats.class);
+
+  public double memoryUsedBytes = 0.0;
+  public double allocationRate = 0.0;
+
+  private final SpMemoryManagerMetrics metrics;
+
+  public SpMemoryManagerStats() {
+    this.metrics = new SpMemoryManagerMetrics();
+  }
+
+  /**
+   * Update all metrics with custom total memory
+   */
+  public void updateAllMetrics() {
+    SpMemoryManagerMetrics.updateCoreMetrics(memoryUsedBytes, allocationRate);
+  }
+
+  public double getMemoryUsedBytes() {
+    return memoryUsedBytes;
+  }
+
+  public void setMemoryUsedBytes(double memoryUsedBytes) {
+    this.memoryUsedBytes = memoryUsedBytes;
+  }
+
+  public double getAllocationRate() {
+    return allocationRate;
+  }
+
+  public void setAllocationRate(double allocationRate) {
+    this.allocationRate = allocationRate;
+  }
+
+  public SpMemoryManagerMetrics getMetrics() {
+    return metrics;
+  }
+}