FEAT: Producer side rate limiting

.run/Dev, start pulsar and zk.run.xml

@@ -5,7 +5,7 @@
         <option name="envFilePath" value="" />
         <option name="profiles">
           <list>
-            <option value="dev" />
+            <option value="dev-metric" />
           </list>
         </option>
         <option name="removeOrphansOnComposeDown" value="false" />

rateLimiter/build.gradle

@@ -0,0 +1,16 @@
+plugins {
+    id 'com.flipkart.varadhi.java-library-conventions'
+//    id 'java-test-fixtures'
+}
+
+dependencies {
+//    api(project(':common'))
+//    api(project(':spi'))
+//    implementation("io.vertx:vertx-core:${vertx_version}")
+    implementation('com.fasterxml.jackson.core:jackson-databind')
+    implementation("org.apache.commons:commons-lang3")
+//    implementation("com.google.guava:guava")
+
+//    testFixturesImplementation(project(":common"))
+//    testImplementation(project(':pulsar'))
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/RateLimiter.java

@@ -0,0 +1,11 @@
+package com.flipkart.varadhi.qos;
+
+import com.flipkart.varadhi.qos.entity.RateLimiterType;
+
+public interface RateLimiter {
+    RateLimiterType getType();
+
+    // assumes 1 QPS
+    // for batch request (to be considered as multi QPS), consider creating a new API
+    boolean isAllowed(long value);
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/TopicRateLimiter.java

@@ -0,0 +1,45 @@
+package com.flipkart.varadhi.qos;
+
+import com.flipkart.varadhi.qos.entity.RateLimiterType;
+
+import java.util.concurrent.atomic.LongAdder;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class TopicRateLimiter implements RateLimiter {
+
+    @Getter
+    private final String topic;
+    @Getter
+    private final RateLimiterType type;
+    @Getter
+    private volatile double suppressionFactor;
+    private long lastObserved;
+    private final LongAdder currentObserved;
+
+    public TopicRateLimiter(String topic, RateLimiterType type) {
+        this.topic = topic;
+        this.type = type;
+        this.suppressionFactor = 0;
+        this.lastObserved = 0L;
+        this.currentObserved = new LongAdder();
+    }
+
+    @Override
+    public boolean isAllowed(long value) {
+        this.currentObserved.add(value);
+        if (this.suppressionFactor == 0) {
+            return true;
+        }
+        return this.currentObserved.longValue() <= this.lastObserved * (1 - this.suppressionFactor);
+    }
+
+    public void setSuppressionFactor(double suppressionFactor) {
+        this.lastObserved = this.currentObserved.longValue();
+        // remove last recorded value
+        this.currentObserved.add(-lastObserved);
+        this.suppressionFactor = suppressionFactor;
+    }
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/ClientHistory.java

@@ -0,0 +1,54 @@
+package com.flipkart.varadhi.qos.entity;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Slf4j
+public class ClientHistory<T> {
+    Map<String, Deque<T>> clientHistoryMap; // clientId to history records
+    private final int slots;
+
+    public ClientHistory(int historySlots) {
+        this.clientHistoryMap = new ConcurrentHashMap<>();
+        this.slots = historySlots;
+    }
+
+    public T getRecentRecord(String clientId) {
+        var load = clientHistoryMap.get(clientId);
+        if(ObjectUtils.isEmpty(load)) {
+            return null;
+        }
+        return load.peekLast();
+    }
+
+    public void add(String clientId, T load) {
+        // first time for client, create a new history queue
+        if(!clientHistoryMap.containsKey(clientId)) {
+            addClient(clientId);
+//            // other option to throw exception and handle client addition somewhere else
+//            throw new IllegalArgumentException("Client not found");
+        }
+
+        // if history queue is full, remove the oldest data
+        if(clientHistoryMap.get(clientId).size() == slots)
+            clientHistoryMap.get(clientId).poll();
+
+        // add new data
+        clientHistoryMap.get(clientId).add(load);
+    }
+
+    public void addClient(String clientId) {
+        clientHistoryMap.put(clientId, new ArrayDeque<>(slots));
+    }
+
+    // TODO(rl): remove not active clients
+    public void removeClient(String clientId) {
+        clientHistoryMap.remove(clientId);
+    }
+
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/ClientLoadInfo.java

@@ -0,0 +1,30 @@
+package com.flipkart.varadhi.qos.entity;
+
+
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@AllArgsConstructor
+public class ClientLoadInfo {
+    // current millis
+    private String clientId;
+    private long from;
+    private long to;
+    private List<TrafficData> topicUsageList; // topic to incoming traffic map
+
+    @Override
+    public String toString() {
+        return "\nLoadInfo{" +
+                "clientId='" + clientId + '\'' +
+                ", from=" + from +
+                ", to=" + to +
+                ", topicUsageMap=" + topicUsageList +
+                "}\n";
+    }
+
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/RateLimiterType.java

@@ -0,0 +1,6 @@
+package com.flipkart.varadhi.qos.entity;
+
+public enum RateLimiterType {
+    THROUGHPUT_CHECK,
+    QPS_CHECK
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/SuppressionData.java

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.qos.entity;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Suppression data for rate limiting. Stores suppression factor for each topic.
+ */
+@Getter
+@Setter
+public class SuppressionData implements Serializable {
+    private Map<String, SuppressionFactor> suppressionFactor = new HashMap<>();
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/SuppressionFactor.java

@@ -0,0 +1,19 @@
+package com.flipkart.varadhi.qos.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * Suppression factor for a topic. Factor must be between 0 and 1. 0 means no suppression, 1 means full suppression.
+ */
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class SuppressionFactor {
+    private double throughputFactor;
+    // can be extended to support suppression factor for QPS, custom measurements etc.
+    // double qpsFactor;
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/TopicLoadInfo.java

@@ -0,0 +1,29 @@
+package com.flipkart.varadhi.qos.entity;
+
+
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter
+@Setter
+@AllArgsConstructor
+public class TopicLoadInfo {
+    // current millis
+    private String clientId; // todo(rl): maybe optional?
+    private long from;
+    private long to;
+    private TrafficData topicLoad; // topic to incoming traffic map
+
+    @Override
+    public String toString() {
+        return "\nTopicLoadInfo{" +
+                "clientId='" + clientId + '\'' +
+                ", from=" + from +
+                ", to=" + to +
+                ", topicLoad=" + topicLoad +
+                "}\n";
+    }
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/entity/TrafficData.java

@@ -0,0 +1,27 @@
+package com.flipkart.varadhi.qos.entity;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Single topic's traffic data
+ */
+@Getter
+@Setter
+@Builder
+// TODO(rl): not in RL module, move to entity module
+public class TrafficData {
+    private String topic;
+    private long bytesIn;
+    private long rateIn;
+
+    @Override
+    public String toString() {
+        return "\nTrafficData{" +
+                "throughputIn=" + bytesIn +
+                ", rateIn=" + rateIn +
+                "}";
+    }
+
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/weights/ConstantWeightFunction.java

@@ -0,0 +1,11 @@
+package com.flipkart.varadhi.qos.weights;
+
+public class ConstantWeightFunction implements WeightFunction {
+    @Override
+    public float applyWeight(long time, long currentTime, long windowSize) {
+        if(time < (currentTime - windowSize)) {
+            return 0.0f;
+        }
+        return 1.0f;
+    }
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/weights/ExponentialWeightFunction.java

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.qos.weights;
+
+public class ExponentialWeightFunction implements WeightFunction {
+    private final float k;
+
+    public ExponentialWeightFunction(float k) {
+        this.k = k;
+    }
+
+    @Override
+    public float applyWeight(long time, long currentTime, long windowSize) {
+        if(time < (currentTime - windowSize)) {
+            return 0.0f;
+        }
+        return 1.0f - (float) Math.exp(-k * (time - (currentTime - windowSize)) / windowSize);
+    }
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/weights/LinearWeightFunction.java

@@ -0,0 +1,11 @@
+package com.flipkart.varadhi.qos.weights;
+
+public class LinearWeightFunction implements WeightFunction {
+    @Override
+    public float applyWeight(long time, long currentTime, long windowSize) {
+        if(time < (currentTime - windowSize)) {
+            return 0.0f;
+        }
+        return (float) (time - (currentTime - windowSize)) / windowSize;
+    }
+}

rateLimiter/src/main/java/com/flipkart/varadhi/qos/weights/WeightFunction.java

@@ -0,0 +1,5 @@
+package com.flipkart.varadhi.qos.weights;
+
+public interface WeightFunction {
+    float applyWeight(long time, long currentTime, long windowSize);
+}

rateLimiter/src/main/java/module-info.java

@@ -0,0 +1,14 @@
+module com.flipkart.varadhi.rateLimiter {
+    requires static lombok;
+    requires org.slf4j;
+//    requires com.google.common;
+    requires org.apache.commons.lang3;
+    requires jakarta.validation;
+    requires com.fasterxml.jackson.databind;
+
+//    requires com.flipkart.varadhi.common;
+//    requires com.flipkart.varadhi.spi;
+//    requires com.flipkart.varadhi.entities;
+
+    exports com.flipkart.varadhi.qos.entity;
+}

server/build.gradle

@@ -16,6 +16,7 @@ dependencies {
     implementation(project(":authz"))
     implementation(project(":controller"))
     implementation(project(":consumer"))
+    implementation(project(':rateLimiter'))
 
     implementation("org.apache.logging.log4j:log4j-slf4j2-impl")
     implementation("org.apache.logging.log4j:log4j-core")

server/src/main/java/com/flipkart/varadhi/cluster/MessageExchange.java

@@ -2,7 +2,6 @@
 
 import com.flipkart.varadhi.cluster.messages.ClusterMessage;
 import com.flipkart.varadhi.cluster.messages.ResponseMessage;
-import com.flipkart.varadhi.exceptions.NotImplementedException;
 import com.flipkart.varadhi.utils.JsonMapper;
 import io.vertx.core.eventbus.DeliveryOptions;
 import io.vertx.core.eventbus.EventBus;
@@ -48,7 +47,13 @@ public CompletableFuture<Void> send(String routeName, String apiName, ClusterMes
     }
 
     public void publish(String routeName, String apiName, ClusterMessage msg) {
-        throw new NotImplementedException("publish not implemented");
+        String apiPath = getPath(routeName, apiName, RouteMethod.PUBLISH);
+        try {
+            vertxEventBus.publish(apiPath, JsonMapper.jsonSerialize(msg), deliveryOptions);
+            log.debug("publish({}, {}) delivered.", apiPath, msg.getId());
+        } catch (Exception e) {
+            log.error("publish({}, {}) Unexpected failure:{}", apiPath, msg.getId(), e.getMessage());
+        }
     }
 
     public CompletableFuture<ResponseMessage> request(String routeName, String apiName, ClusterMessage msg) {
@@ -57,7 +62,7 @@ public CompletableFuture<ResponseMessage> request(String routeName, String apiNa
         try {
             vertxEventBus.request(apiPath, JsonMapper.jsonSerialize(msg), deliveryOptions, ar -> {
                 if (ar.succeeded()) {
-                    log.debug("request({}, {}) delivered. {}.", apiPath, msg.getId(), ar.result().body());
+                    log.info("request({}, {}) delivered. {}.", apiPath, msg.getId(), ar.result().body());
                     ResponseMessage response =
                             JsonMapper.jsonDeserialize((String) ar.result().body(), ResponseMessage.class);
                     if (response.getException() != null) {