Fix throttling

otoroshi/app/controllers/SwaggerController.scala

@@ -1176,9 +1176,9 @@ class SwaggerController(cc: ControllerComponents, assetsBuilder: AssetsBuilder)(
       "description" -> "Quotas state for an api key on a service group",
       "type"        -> "object",
       "required"    -> Json.arr(
-        "authorizedCallsPerSec",
-        "currentCallsPerSec",
-        "remainingCallsPerSec",
+        "authorizedCallsPerWindow",
+        "throttlingCallsPerWindow",
+        "remainingCallsPerWindow",
         "authorizedCallsPerDay",
         "currentCallsPerDay",
         "remainingCallsPerDay",
@@ -1187,9 +1187,9 @@ class SwaggerController(cc: ControllerComponents, assetsBuilder: AssetsBuilder)(
         "remainingCallsPerMonth"
       ),
       "properties"  -> Json.obj(
-        "authorizedCallsPerSec"   -> SimpleLongType ~~> "The number of authorized calls per second",
-        "currentCallsPerSec"      -> SimpleLongType ~~> "The current number of calls per second",
-        "remainingCallsPerSec"    -> SimpleLongType ~~> "The remaining number of calls per second",
+        "authorizedCallsPerWindow"   -> SimpleLongType ~~> "The number of authorized calls per second",
+        "throttlingCallsPerWindow"      -> SimpleLongType ~~> "The current number of calls per second",
+        "remainingCallsPerWindow"    -> SimpleLongType ~~> "The remaining number of calls per second",
         "authorizedCallsPerDay"   -> SimpleLongType ~~> "The number of authorized calls per day",
         "currentCallsPerDay"      -> SimpleLongType ~~> "The current number of calls per day",
         "remainingCallsPerDay"    -> SimpleLongType ~~> "The remaining number of calls per day",

otoroshi/app/gateway/generic.scala

@@ -1279,7 +1279,7 @@ object ReverseProxyHelper {
       val quota                                          = maybeQuota.getOrElse(globalConfig.perIpThrottlingQuota)
       val (restrictionsNotPassing, restrictionsResponse) =
         descriptor.restrictions.handleRestrictions(descriptor.id, descriptor.some, None, req, attrs)
-      if (secCalls > (quota * 10L)) {
+      if (secCalls > quota) {
         errorResult(TooManyRequests, "[IP] You performed too much requests", "errors.too.much.requests")
       } else {
         if (!isSecured && descriptor.forceHttps) {

otoroshi/app/models/apikey.scala

@@ -45,28 +45,52 @@ import scala.jdk.CollectionConverters.asScalaBufferConverter
 import scala.util.{Failure, Success, Try}
 
 case class RemainingQuotas(
-    // secCalls: Long = RemainingQuotas.MaxValue,
-    // secCallsRemaining: Long = RemainingQuotas.MaxValue,
-    // dailyCalls: Long = RemainingQuotas.MaxValue,
-    // dailyCallsRemaining: Long = RemainingQuotas.MaxValue,
-    // monthlyCalls: Long = RemainingQuotas.MaxValue,
-    // monthlyCallsRemaining: Long = RemainingQuotas.MaxValue
-    authorizedCallsPerSec: Long = RemainingQuotas.MaxValue,
-    currentCallsPerSec: Long = RemainingQuotas.MaxValue,
-    remainingCallsPerSec: Long = RemainingQuotas.MaxValue,
+    authorizedCallsPerWindow: Long = RemainingQuotas.MaxValue,
+    throttlingCallsPerWindow: Long = RemainingQuotas.MaxValue,
+    remainingCallsPerWindow: Long = RemainingQuotas.MaxValue,
     authorizedCallsPerDay: Long = RemainingQuotas.MaxValue,
     currentCallsPerDay: Long = RemainingQuotas.MaxValue,
     remainingCallsPerDay: Long = RemainingQuotas.MaxValue,
     authorizedCallsPerMonth: Long = RemainingQuotas.MaxValue,
     currentCallsPerMonth: Long = RemainingQuotas.MaxValue,
     remainingCallsPerMonth: Long = RemainingQuotas.MaxValue
 ) {
-  def toJson: JsObject = RemainingQuotas.fmt.writes(this)
+  def toJson: JsObject = RemainingQuotas.fmt.writes(this).as[JsObject]
 }
 
 object RemainingQuotas {
   val MaxValue: Long = 10000000L
-  implicit val fmt   = Json.format[RemainingQuotas]
+  implicit val fmt   = new Format[RemainingQuotas] {
+
+    override def reads(json: JsValue): JsResult[RemainingQuotas] =  Try {
+      RemainingQuotas(
+        authorizedCallsPerWindow = json.select("authorizedCallsPerWindow").asOpt[Long].getOrElse(json.select("authorizedCallsPerSec").asLong),
+        throttlingCallsPerWindow = json.select("throttlingCallsPerWindow").asOpt[Long].getOrElse(json.select("currentCallsPerSec").asLong),
+        remainingCallsPerWindow = json.select("remainingCallsPerWindow").asOpt[Long].getOrElse(json.select("authorizedCallsPerSec").asLong),
+        authorizedCallsPerDay = json.select("authorizedCallsPerDay").asLong,
+        currentCallsPerDay = json.select("currentCallsPerDay").asLong,
+        remainingCallsPerDay = json.select("remainingCallsPerDay").asLong,
+        authorizedCallsPerMonth = json.select("authorizedCallsPerMonth").asLong,
+        currentCallsPerMonth = json.select("currentCallsPerMonth").asLong,
+        remainingCallsPerMonth = json.select("remainingCallsPerMonth").asLong,
+      )
+    } match {
+      case Failure(e) => JsError(e.getMessage)
+      case Success(e) => JsSuccess(e)
+    }
+
+    override def writes(o: RemainingQuotas): JsValue = Json.obj(
+      "authorizedCallsPerWindow" -> o.authorizedCallsPerWindow,
+      "throttlingCallsPerWindow" -> o.throttlingCallsPerWindow,
+      "remainingCallsPerWindow" -> o.remainingCallsPerWindow,
+      "authorizedCallsPerDay" -> o.authorizedCallsPerDay,
+      "currentCallsPerDay" -> o.currentCallsPerDay,
+      "remainingCallsPerDay" -> o.remainingCallsPerDay,
+      "authorizedCallsPerMonth" -> o.authorizedCallsPerMonth,
+      "currentCallsPerMonth" -> o.currentCallsPerMonth,
+      "remainingCallsPerMonth" -> o.remainingCallsPerMonth,
+    )
+  }
 }
 
 case class ApiKeyRotation(
@@ -244,7 +268,7 @@ case class ApiKey(
       quotas   <- env.datastores.apiKeyDataStore.remainingQuotas(this)
       rotation <- env.datastores.apiKeyDataStore.keyRotation(this)
     } yield {
-      val within = (quotas.currentCallsPerSec <= (throttlingQuota * env.throttlingWindow)) &&
+      val within = quotas.remainingCallsPerWindow > 0 &&
         (quotas.currentCallsPerDay < dailyQuota) &&
         (quotas.currentCallsPerMonth < monthlyQuota)
       (within, rotation, quotas)

otoroshi/app/models/config.scala

@@ -1026,7 +1026,7 @@ object GlobalConfig {
 }
 
 trait GlobalConfigDataStore extends BasicStore[GlobalConfig] {
-  def incrementCallsForIpAddressWithTTL(ip: String, ttl: Int = 10)(implicit ec: ExecutionContext): Future[Long]
+  def incrementCallsForIpAddress(ip: String)(implicit ec: ExecutionContext): Future[Long]
   def quotaForIpAddress(ip: String)(implicit ec: ExecutionContext): Future[Option[Long]]
   def isOtoroshiEmpty()(implicit ec: ExecutionContext): Future[Boolean]
   def withinThrottlingQuota()(implicit ec: ExecutionContext, env: Env): Future[Boolean]

otoroshi/app/next/plugins/context.scala

@@ -113,9 +113,9 @@ class ContextValidation extends NgAccessValidator {
       |    "otoroshi.core.RequestWebsocket" : false,
       |    "otoroshi.core.RequestCounterOut" : 0,
       |    "otoroshi.core.RemainingQuotas" : {
-      |      "authorizedCallsPerSec" : 10000000,
-      |      "currentCallsPerSec" : 0,
-      |      "remainingCallsPerSec" : 10000000,
+      |      "authorizedCallsPerWindow" : 10000000,
+      |      "throttlingCallsPerWindow" : 0,
+      |      "remainingCallsPerWindow" : 10000000,
       |      "authorizedCallsPerDay" : 10000000,
       |      "currentCallsPerDay" : 2,
       |      "remainingCallsPerDay" : 9999998,

otoroshi/app/next/plugins/quotas.scala

@@ -63,7 +63,7 @@ class GlobalPerIpAddressThrottling extends NgAccessValidator {
   ): Future[NgAccess] = {
     val globalConfig = env.datastores.globalConfigDataStore.latest()
     val quota        = quotas.maybeQuota.getOrElse(globalConfig.perIpThrottlingQuota)
-    if (quotas.secCalls > (quota * 10L)) {
+    if (quotas.secCalls > quota) {
       errorResult(ctx, Results.TooManyRequests, "[IP] You performed too much requests", "errors.too.much.requests")
     } else {
       NgAccess.NgAllowed.vfuture
@@ -266,7 +266,7 @@ class NgServiceQuotas extends NgAccessValidator {
   )(implicit ec: ExecutionContext, env: Env): Future[Boolean] =
     env.datastores.rawDataStore
       .get(throttlingKey(route.id))
-      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= (qconf.throttlingQuota * env.throttlingWindow))
+      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= qconf.throttlingQuota)
 
   private def withinDailyQuota(route: NgRoute, qconf: NgServiceQuotasConfig)(implicit
       ec: ExecutionContext,
@@ -559,7 +559,7 @@ object NgCustomThrottling {
     for {
       secCalls <- env.datastores.rawDataStore.incrby(throttlingKey(expr, group), increment)
       secTtl   <- env.datastores.rawDataStore.pttl(throttlingKey(expr, group)).filter(_ > -1).recoverWith { case _ =>
-                    env.datastores.rawDataStore.pexpire(throttlingKey(expr, group), ttl) // env.throttlingWindow * 1000)
+                    env.datastores.rawDataStore.pexpire(throttlingKey(expr, group), ttl) // env.throttlingWindow * 1000
                   }
     } yield ()
   }
@@ -593,7 +593,7 @@ class NgCustomThrottling extends NgAccessValidator {
   )(implicit ec: ExecutionContext, env: Env): Future[Boolean] = {
     env.datastores.rawDataStore
       .get(NgCustomThrottling.throttlingKey(qconf.computeExpression(ctx, env), qconf.computeGroup(ctx, env)))
-      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= (qconf.throttlingQuota * env.throttlingWindow))
+      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= qconf.throttlingQuota)
   }
 
   def forbidden(ctx: NgAccessContext)(implicit env: Env, ec: ExecutionContext): Future[NgAccess] = {

otoroshi/app/next/proxy/engine.scala

@@ -2017,11 +2017,16 @@ class ProxyEngine() extends RequestHandler {
       // quotasValidationFor increments calls for ip address
       FEither(env.datastores.globalConfigDataStore.quotasValidationFor(remoteAddress).flatMap { r =>
         val (within, secCalls, maybeQuota) = r
+//        println(s"maybe quota - $maybeQuota")
         val quota                          = maybeQuota.getOrElse(globalConfig.perIpThrottlingQuota)
-        if (secCalls > (quota * 10L)) {
+        println(s"[engine] secCalls $secCalls - perIpThrottlingQuota $quota")
+        // if (secCalls > (quota * 10L)) {
+        if (secCalls > quota) {
+          println("4")
           errorResult(Results.TooManyRequests, "[IP] You performed too much requests", "errors.too.much.requests")
         } else {
           if (!within) {
+            println("3")
             errorResult(Results.TooManyRequests, "[GLOBAL] You performed too much requests", "errors.too.much.requests")
           } else if (globalConfig.ipFiltering.notMatchesWhitelist(remoteAddress)) {
             errorResult(

otoroshi/app/openapi/openapi-cfg.json

@@ -1957,13 +1957,13 @@
     "otoroshi.models.RegionMatch.region": "Region name",
     "otoroshi.models.RemainingQuotas.authorizedCallsPerDay": "Number of authorized call per day",
     "otoroshi.models.RemainingQuotas.authorizedCallsPerMonth": "Number of authorized call per month",
-    "otoroshi.models.RemainingQuotas.authorizedCallsPerSec": "Number of authorized call per second",
+    "otoroshi.models.RemainingQuotas.authorizedCallsPerWindow": "Number of authorized call per second",
     "otoroshi.models.RemainingQuotas.currentCallsPerDay": "Current number of call per day",
     "otoroshi.models.RemainingQuotas.currentCallsPerMonth": "Current number of call per month",
-    "otoroshi.models.RemainingQuotas.currentCallsPerSec": "Current number of call per second",
+    "otoroshi.models.RemainingQuotas.throttlingCallsPerWindow": "Current number of call per second",
     "otoroshi.models.RemainingQuotas.remainingCallsPerDay": "Remaining number of call per day",
     "otoroshi.models.RemainingQuotas.remainingCallsPerMonth": "Remaining number of call per month",
-    "otoroshi.models.RemainingQuotas.remainingCallsPerSec": "Remaining number of call per second",
+    "otoroshi.models.RemainingQuotas.remainingCallsPerWindow": "Remaining number of call per second",
     "otoroshi.models.RestrictionPath.method": "Method of the http request",
     "otoroshi.models.RestrictionPath.path": "Path of the http request",
     "otoroshi.models.Restrictions.allowLast": "Evalute allowed paths after everything else",

otoroshi/app/plugins/quotas.scala

@@ -195,10 +195,10 @@ class ServiceQuotas extends AccessValidator {
     env.clusterAgent.incrementApi(descriptor.id, increment)
     for {
       _            <- env.datastores.rawDataStore.incrby(totalCallsKey(descriptor.id), increment)
-      secCalls     <- env.datastores.rawDataStore.incrby(throttlingKey(descriptor.id), increment)
       secTtl       <- env.datastores.rawDataStore.pttl(throttlingKey(descriptor.id)).filter(_ > -1).recoverWith { case _ =>
-                        env.datastores.rawDataStore.pexpire(throttlingKey(descriptor.id), env.throttlingWindow * 1000)
-                      }
+                      env.datastores.rawDataStore.pexpire(throttlingKey(descriptor.id), env.throttlingWindow * 1000)
+                    }
+      secCalls     <- env.datastores.rawDataStore.incrby(throttlingKey(descriptor.id), increment)
       dailyCalls   <- env.datastores.rawDataStore.incrby(dailyQuotaKey(descriptor.id), increment)
       dailyTtl     <- env.datastores.rawDataStore.pttl(dailyQuotaKey(descriptor.id)).filter(_ > -1).recoverWith { case _ =>
                         env.datastores.rawDataStore.pexpire(dailyQuotaKey(descriptor.id), toDayEnd.toInt)
@@ -209,9 +209,9 @@ class ServiceQuotas extends AccessValidator {
                       }
     } yield ()
     // RemainingQuotas(
-    //   authorizedCallsPerSec = qconf.throttlingQuota,
-    //   currentCallsPerSec = (secCalls / env.throttlingWindow).toInt,
-    //   remainingCallsPerSec = qconf.throttlingQuota - (secCalls / env.throttlingWindow).toInt,
+    //   authorizedCallsPerWindow = qconf.throttlingQuota,
+    //   throttlingCallsPerWindow = (secCalls / env.throttlingWindow).toInt,
+    //   remainingCallsPerWindow = qconf.throttlingQuota - (secCalls / env.throttlingWindow).toInt,
     //   authorizedCallsPerDay = qconf.dailyQuota,
     //   currentCallsPerDay = dailyCalls,
     //   remainingCallsPerDay = qconf.dailyQuota - dailyCalls,
@@ -238,7 +238,7 @@ class ServiceQuotas extends AccessValidator {
     env.datastores.rawDataStore
       .get(throttlingKey(descriptor.id))
       .fast
-      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= (qconf.throttlingQuota * env.throttlingWindow))
+      .map(_.map(_.utf8String.toLong).getOrElse(0L) <= qconf.throttlingQuota)
 
   private def withinDailyQuota(descriptor: ServiceDescriptor, qconf: ServiceQuotasConfig)(implicit
       ec: ExecutionContext,

otoroshi/app/storage/drivers/inmemory/InMemoryDataStores.scala

@@ -49,14 +49,13 @@ class InMemoryDataStores(
     )
   val materializer     = Materializer(actorSystem)
   val _optimized       = configuration.getOptionalWithFileSupport[Boolean]("app.inmemory.optimized").getOrElse(false)
-  val _modern          = configuration.getOptionalWithFileSupport[Boolean]("app.inmemory.modern").getOrElse(false)
+  val _modern          = configuration.getOptionalWithFileSupport[Boolean]("app.inmemory.modern").getOrElse(true)
   // lazy val redis       = new SwappableInMemoryRedis(_optimized, env, actorSystem)
-  lazy val _redis      = if (_modern) {
+  lazy val swredis      = if (_modern) {
     new ModernSwappableInMemoryRedis(_optimized, env, actorSystem)
   } else {
     new SwappableInMemoryRedis(_optimized, env, actorSystem)
   }
-  lazy val swredis     = if (env.isDev) new SwappableRedisLikeMetricsWrapper(_redis, env) else _redis
 
   def redis(): otoroshi.storage.RedisLike = swredis
 

otoroshi/app/storage/drivers/inmemory/SwappableInMemoryRedis.scala

@@ -12,7 +12,7 @@ import otoroshi.utils.syntax.implicits.BetterSyntax
 import play.api.Logger
 import play.api.libs.json.{JsValue, Json}
 
-import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.atomic.{AtomicLong, AtomicReference}
 import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 import java.util.regex.Pattern
 import scala.collection.concurrent.TrieMap
@@ -513,7 +513,15 @@ class ModernSwappableInMemoryRedis(_optimized: Boolean, env: Env, actorSystem: A
 
   def rawGet(key: String): Future[Option[Any]] = memory.get(key).future
 
-  override def get(key: String): Future[Option[ByteString]] = memory.getTyped[ByteString](key).future
+  override def get(key: String): Future[Option[ByteString]] = {
+    //memory.getTyped[ByteString](key).future
+    memory.get(key) match {
+      case Some(bs: ByteString) => bs.some.vfuture
+      case Some(counter: AtomicLong) => ByteString(counter.get().toString).some.vfuture
+      case _ => None.vfuture
+    }
+  }
+
 
   override def set(
       key: String,
@@ -546,10 +554,32 @@ class ModernSwappableInMemoryRedis(_optimized: Boolean, env: Env, actorSystem: A
   override def incr(key: String): Future[Long] = incrby(key, 1L)
 
   override def incrby(key: String, increment: Long): Future[Long] = {
-    val value: Long    = memory.getTyped[ByteString](key).map(_.utf8String.toLong).getOrElse(0L)
-    val newValue: Long = value + increment
-    memory.put(key, ByteString(newValue.toString))
-    newValue.future
+    (memory.get(key) match {
+      case Some(bs: ByteString) => {
+        val asLng = bs.utf8String.toLong
+        val cnt = new AtomicLong(asLng)
+        val fcnt = cnt.addAndGet(increment)
+        memory.put(key, cnt)
+        fcnt
+      }
+      case Some(cnt: AtomicLong) => {
+        val fcnt = cnt.addAndGet(increment)
+        memory.put(key, cnt)
+        fcnt
+      }
+      case _ => {
+        val cnt = new AtomicLong(increment)
+        memory.put(key, cnt) match {
+          case Some(bs: ByteString) => cnt.addAndGet(bs.utf8String.toLong)
+          case Some(c: AtomicLong) => cnt.addAndGet(c.get())
+          case _ => increment
+        }
+      }
+    }).vfuture
+    // val value: Long    = memory.getTyped[ByteString](key).map(_.utf8String.toLong).getOrElse(0L)
+    // val newValue: Long = value + increment
+    // memory.put(key, ByteString(newValue.toString))
+    // newValue.future
   }
 
   override def exists(key: String): Future[Boolean] = memory.containsKey(key).future
@@ -693,4 +723,4 @@ class ModernSwappableInMemoryRedis(_optimized: Boolean, env: Env, actorSystem: A
   }
 
   override def health()(implicit ec: ExecutionContext): Future[DataStoreHealth] = Healthy.future
-}
+}