Merge pull request #185 from ing-bank/fix/check-bucket-location-rate-limiting

Fix/check bucket location rate limiting

Author: kr7ysztof

docker-compose-extra.yaml

@@ -0,0 +1,17 @@
+version: "3"
+services:
+
+  mockServer:
+    image: mockserver/mockserver:5.14.0
+    ports:
+      - 1080:1080
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    environment:
+      MOCKSERVER_WATCH_INITIALIZATION_JSON: "true"
+      MOCKSERVER_INITIALIZATION_JSON_PATH: /config/simulateBucketNotFoundDuringMultipartUpload.jsonc
+      genericJVMOptions: "-Xmx1024m -Xms512m"
+    volumes:
+      - type: bind
+        source: ./mockServer
+        target: /config

mockServer/simulateBucketNotFoundDuringMultipartUpload.jsonc

@@ -0,0 +1,39 @@
+// To use this configuration make sure it is selected as MOCKSERVER_INITIALIZATION_JSON_PATH and switch rokku to go through proxy port: 1080
+// Remember to comment out `updateBucketCredentials` in `NamespacesHandler.scala:52` so rokku will try to refetch bucket location before each request and to enable rokku multiple namespace feature
+[
+    {
+        // Allow first few requests to start multipart upload
+        "httpRequest": {
+            "method": "GET"
+        },
+        "httpForward": {
+            "host": "host.docker.internal",
+            "port": 8010,
+            "scheme": "HTTP"
+        },
+        "times": {
+            "remainingTimes": 4,
+            "unlimited": false
+        },
+        "priority": 20,
+    },
+    {
+        // Simulate rate limiting
+        "httpRequest": {
+            "method": "GET"
+        },
+        "httpResponse": {
+            "statusCode": 503,
+        },
+        "priority": 10,
+    },
+    {
+        // allow all other requests
+        "httpForward": {
+            "host": "host.docker.internal",
+            "port": 8010,
+            "scheme": "HTTP"
+        },
+        "priority": 0,
+    },
+]

src/main/resources/application.conf

@@ -24,7 +24,8 @@ rokku {
         }
         region = ${?ROKKU_STORAGE_S3_AWS_REGION}
         v2SignatureEnabled = ${?ROKKU_STORAGE_S3_V2_ENABLED}
-
+        # To add more then one code, seperate them with comma
+        slowdownCodes = ${?ROKKU_STORAGE_S3_SLOWDOWN_CODES}
         healthCheck {
            # can be one of:
            #   s3ListBucket  - uses AWS S3 client to list single bucket

src/main/resources/reference.conf

@@ -31,7 +31,8 @@ rokku {
         }
         region = "us-east-1"
         v2SignatureEnabled = false
-
+        # To add more then one code, seperate them with comma
+        slowdownCodes = "502, 503"
         healthCheck {
            # can be one of:
            #   s3ListBucket  - uses AWS S3 client to list single bucket

src/main/scala/com/ing/wbaa/rokku/proxy/config/StorageS3Settings.scala

@@ -14,6 +14,7 @@ class StorageS3Settings(config: Config) extends Extension {
   val storageS3AdminSecretkey: String = config.getString("rokku.storage.s3.admin.secretkey")
   val awsRegion: String = config.getString("rokku.storage.s3.region")
   val v2SignatureEnabled: Boolean = config.getBoolean("rokku.storage.s3.v2SignatureEnabled")
+  val slowdownCodes: Array[Int] = config.getString("rokku.storage.s3.slowdownCodes").split(",").map(o => o.trim.toInt)
   val isRequestUserQueueEnabled: Boolean = config.getBoolean("rokku.storage.s3.request.queue.enable")
   private val hcMethodString = config.getString("rokku.storage.s3.healthCheck.method")
   val hcMethod: HCMethod = hcMethodString match {

src/main/scala/com/ing/wbaa/rokku/proxy/handler/namespace/NamespacesHandler.scala

@@ -4,8 +4,10 @@ import akka.http.scaladsl.model.HttpRequest
 import com.amazonaws.auth.BasicAWSCredentials
 import com.amazonaws.services.s3.model.AmazonS3Exception
 import com.ing.wbaa.rokku.proxy.config.NamespaceSettings
+import com.ing.wbaa.rokku.proxy.config.StorageS3Settings
 import com.ing.wbaa.rokku.proxy.data.RequestId
 import com.ing.wbaa.rokku.proxy.handler.LoggerHandlerWithId
+import com.ing.wbaa.rokku.proxy.handler.exception.RokkuThrottlingException
 import com.ing.wbaa.rokku.proxy.metrics.MetricsFactory.{ incrementBucketNamespaceCacheHit, incrementBucketNamespacesNotFound, incrementBucketNamespacesSearch }
 import com.ing.wbaa.rokku.proxy.util.S3Utils
 
@@ -21,6 +23,7 @@ trait NamespacesHandler {
   private val bucketCredentials: scala.collection.concurrent.Map[BucketName, BasicAWSCredentials] = scala.collection.concurrent.TrieMap[BucketName, BasicAWSCredentials]()
 
   protected[this] val namespaceSettings: NamespaceSettings
+  protected[this] def storageS3Settings: StorageS3Settings
 
   private def namespaceCredentials: ListMap[NamespaceName, BasicAWSCredentials] = namespaceSettings.namespaceCredentialsMap
 
@@ -72,6 +75,10 @@ trait NamespacesHandler {
               logger.info("bucket {} in namespace {} return 403 for credentials {} so bucket exist but the credentials cannot see location ", bucketName.name, namespaceName, ex, credentials.getAWSAccessKeyId)
               return true
             }
+            if (storageS3Settings.slowdownCodes contains ex.asInstanceOf[AmazonS3Exception].getStatusCode) {
+              logger.info("throttling, cannot check bucket location")
+              throw new RokkuThrottlingException("cannot check bucket location")
+            }
             if (ex.asInstanceOf[AmazonS3Exception].getStatusCode != 404) {
               logger.error("namespace {} returned exception {} for credentials {} but should only status code 404", namespaceName, ex, credentials.getAWSAccessKeyId)
             }