Support request rate limiting (#201)

* Support request rate limiting

Author: Harvey Lowndes

README.md

@@ -173,3 +173,4 @@ See [LICENSE](LICENSE) for more details.
 [7]: https://github.com/oracle/oci-cloud-controller-manager/tree/master/manifests/cloud-provider-example.yaml
 [8]: https://github.com/oracle/oci-cloud-controller-manager/blob/master/docs/tutorial.md
 [9]: https://github.com/oracle/oci-cloud-controller-manager/blob/master/docs/tutorial-ssl.md
+[10]: https://github.com/oracle/oci-cloud-controller-manager/blob/master/docs/rate-limiter-configuration.md

docs/rate-limiter-configuration.md

@@ -0,0 +1,28 @@
+# Rate Limiting Configuration
+
+This file defines a list of Rate Limiting configurations properties,
+supported by the `oci-cloud-controller-manager`.
+
+These properties exist in the `yaml` configuration under `rateLimiter`
+in root. For example:
+
+```yaml
+auth:
+  ...
+loadBalancer:
+  ...
+rateLimiter:
+  rateLimitQPSRead: ...
+  rateLimitBucketRead: ...
+  rateLimitQPSWrite: ...
+  rateLimitBucketWrite: ...
+```
+
+## Request read/write rate limiting properties
+
+| Name | Description | Default |
+| ---- | ----------- | ------- |
+| `rateLimitQPSRead` | The maximum queries allowed per second for read requests. | 20.0 |
+| `rateLimitBucketRead` | The maximum token bucket burst size for read requests. | 5.0 |
+| `rateLimitQPSWrite` | The maximum queries allwoed per second for write requests. | 20.0 |
+| `rateLimitBucketWrite` | The maximim token bucket burst size for write requests. | 5.0 |

pkg/oci/ccm.go

@@ -34,6 +34,7 @@ import (
 	clientset "k8s.io/client-go/kubernetes"
 	listersv1 "k8s.io/client-go/listers/core/v1"
 	cache "k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/flowcontrol"
 	cloudprovider "k8s.io/kubernetes/pkg/cloudprovider"
 	controller "k8s.io/kubernetes/pkg/controller"
 
@@ -42,6 +43,11 @@ import (
 	"github.com/oracle/oci-cloud-controller-manager/pkg/oci/util"
 )
 
+const (
+	rateLimitQPSDefault    = 20.0
+	rateLimitBucketDefault = 5
+)
+
 // ProviderName uniquely identifies the Oracle Bare Metal Cloud Services (OCI)
 // cloud-provider.
 func ProviderName() string {
@@ -78,7 +84,10 @@ func NewCloudProvider(config *Config) (cloudprovider.Interface, error) {
 	if err != nil {
 		return nil, err
 	}
-	c, err := client.New(logger.Sugar(), cp)
+
+	rateLimiter := newRateLimiter(logger.Sugar(), config.RateLimiter)
+
+	c, err := client.New(logger.Sugar(), cp, &rateLimiter)
 	if err != nil {
 		return nil, err
 	}
@@ -225,3 +234,44 @@ func buildConfigurationProvider(logger *zap.Logger, config *Config) (common.Conf
 	)
 	return cp, nil
 }
+
+// newRateLimiter builds and returns a struct containing read and write
+// rate limiters. Defaults are used where no (0) value is provided.
+func newRateLimiter(logger *zap.SugaredLogger, config *RateLimiterConfig) client.RateLimiter {
+	if config == nil {
+		config = &RateLimiterConfig{}
+	}
+
+	// Set to default values if configuration not declared
+	if config.RateLimitQPSRead == 0 {
+		config.RateLimitQPSRead = rateLimitQPSDefault
+	}
+	if config.RateLimitBucketRead == 0 {
+		config.RateLimitBucketRead = rateLimitBucketDefault
+	}
+	if config.RateLimitQPSWrite == 0 {
+		config.RateLimitQPSWrite = rateLimitQPSDefault
+	}
+	if config.RateLimitBucketWrite == 0 {
+		config.RateLimitBucketWrite = rateLimitBucketDefault
+	}
+
+	rateLimiter := client.RateLimiter{
+		Reader: flowcontrol.NewTokenBucketRateLimiter(
+			config.RateLimitQPSRead,
+			config.RateLimitBucketRead),
+		Writer: flowcontrol.NewTokenBucketRateLimiter(
+			config.RateLimitQPSWrite,
+			config.RateLimitBucketWrite),
+	}
+
+	logger.Infof("OCI using read rate limit configuration: QPS=%g, bucket=%d",
+		config.RateLimitQPSRead,
+		config.RateLimitBucketRead)
+
+	logger.Infof("OCI using write rate limit configuration: QPS=%g, bucket=%d",
+		config.RateLimitQPSWrite,
+		config.RateLimitBucketWrite)
+
+	return rateLimiter
+}

pkg/oci/ccm_test.go

@@ -0,0 +1,59 @@
+// Copyright 2017 Oracle and/or its affiliates. All rights reserved.
+//
+// Licensed 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 oci
+
+import (
+	"testing"
+
+	"go.uber.org/zap"
+)
+
+func TestBuildRateLimiterWithConfig(t *testing.T) {
+	qpsRead := float32(6.0)
+	bucketRead := 20
+	qpsWrite := float32(8.0)
+	bucketWrite := 20
+
+	rateLimiterConfig := &RateLimiterConfig{
+		RateLimitQPSRead:     qpsRead,
+		RateLimitBucketRead:  bucketRead,
+		RateLimitQPSWrite:    qpsWrite,
+		RateLimitBucketWrite: bucketWrite,
+	}
+
+	rateLimiter := newRateLimiter(zap.S(), rateLimiterConfig)
+
+	if rateLimiter.Reader.QPS() != qpsRead {
+		t.Errorf("unexpected QPS (read) value: expected %f but found %f", qpsRead, rateLimiter.Reader.QPS())
+	}
+
+	if rateLimiter.Writer.QPS() != qpsWrite {
+		t.Errorf("unexpected QPS (write) value: expected %f but found %f", qpsWrite, rateLimiter.Writer.QPS())
+	}
+}
+
+func TestBuildRateLimiterWithDefaults(t *testing.T) {
+	rateLimiterConfig := &RateLimiterConfig{}
+
+	rateLimiter := newRateLimiter(zap.S(), rateLimiterConfig)
+
+	if rateLimiter.Reader.QPS() != rateLimitQPSDefault {
+		t.Errorf("unexpected QPS (read) value: expected %f but found %f", rateLimitQPSDefault, rateLimiter.Reader.QPS())
+	}
+
+	if rateLimiter.Writer.QPS() != rateLimitQPSDefault {
+		t.Errorf("unexpected QPS (write) value: expected %f but found %f", rateLimitQPSDefault, rateLimiter.Writer.QPS())
+	}
+}

pkg/oci/client/client.go

@@ -15,6 +15,7 @@
 package client
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"io/ioutil"
@@ -26,6 +27,7 @@ import (
 
 	"go.uber.org/zap"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/flowcontrol"
 
 	"github.com/oracle/oci-go-sdk/common"
 	"github.com/oracle/oci-go-sdk/core"
@@ -40,18 +42,54 @@ type Interface interface {
 	Networking() NetworkingInterface
 }
 
+// RateLimiter reader and writer.
+type RateLimiter struct {
+	Reader flowcontrol.RateLimiter
+	Writer flowcontrol.RateLimiter
+}
+
+type computeClient interface {
+	GetInstance(ctx context.Context, request core.GetInstanceRequest) (response core.GetInstanceResponse, err error)
+	ListInstances(ctx context.Context, request core.ListInstancesRequest) (response core.ListInstancesResponse, err error)
+	ListVnicAttachments(ctx context.Context, request core.ListVnicAttachmentsRequest) (response core.ListVnicAttachmentsResponse, err error)
+}
+
+type virtualNetworkClient interface {
+	GetVnic(ctx context.Context, request core.GetVnicRequest) (response core.GetVnicResponse, err error)
+	GetSubnet(ctx context.Context, request core.GetSubnetRequest) (response core.GetSubnetResponse, err error)
+	GetSecurityList(ctx context.Context, request core.GetSecurityListRequest) (response core.GetSecurityListResponse, err error)
+	UpdateSecurityList(ctx context.Context, request core.UpdateSecurityListRequest) (response core.UpdateSecurityListResponse, err error)
+}
+
+type loadBalancerClient interface {
+	GetLoadBalancer(ctx context.Context, request loadbalancer.GetLoadBalancerRequest) (response loadbalancer.GetLoadBalancerResponse, err error)
+	ListLoadBalancers(ctx context.Context, request loadbalancer.ListLoadBalancersRequest) (response loadbalancer.ListLoadBalancersResponse, err error)
+	CreateLoadBalancer(ctx context.Context, request loadbalancer.CreateLoadBalancerRequest) (response loadbalancer.CreateLoadBalancerResponse, err error)
+	DeleteLoadBalancer(ctx context.Context, request loadbalancer.DeleteLoadBalancerRequest) (response loadbalancer.DeleteLoadBalancerResponse, err error)
+	ListCertificates(ctx context.Context, request loadbalancer.ListCertificatesRequest) (response loadbalancer.ListCertificatesResponse, err error)
+	CreateCertificate(ctx context.Context, request loadbalancer.CreateCertificateRequest) (response loadbalancer.CreateCertificateResponse, err error)
+	GetWorkRequest(ctx context.Context, request loadbalancer.GetWorkRequestRequest) (response loadbalancer.GetWorkRequestResponse, err error)
+	CreateBackendSet(ctx context.Context, request loadbalancer.CreateBackendSetRequest) (response loadbalancer.CreateBackendSetResponse, err error)
+	UpdateBackendSet(ctx context.Context, request loadbalancer.UpdateBackendSetRequest) (response loadbalancer.UpdateBackendSetResponse, err error)
+	DeleteBackendSet(ctx context.Context, request loadbalancer.DeleteBackendSetRequest) (response loadbalancer.DeleteBackendSetResponse, err error)
+	CreateListener(ctx context.Context, request loadbalancer.CreateListenerRequest) (response loadbalancer.CreateListenerResponse, err error)
+	UpdateListener(ctx context.Context, request loadbalancer.UpdateListenerRequest) (response loadbalancer.UpdateListenerResponse, err error)
+	DeleteListener(ctx context.Context, request loadbalancer.DeleteListenerRequest) (response loadbalancer.DeleteListenerResponse, err error)
+}
+
 type client struct {
-	compute      *core.ComputeClient
-	network      *core.VirtualNetworkClient
-	loadbalancer *loadbalancer.LoadBalancerClient
+	compute      computeClient
+	network      virtualNetworkClient
+	loadbalancer loadBalancerClient
+
+	rateLimiter RateLimiter
 
 	subnetCache cache.Store
 	logger      *zap.SugaredLogger
 }
 
 // New constructs an OCI API client.
-func New(logger *zap.SugaredLogger, cp common.ConfigurationProvider) (Interface, error) {
-	logger = logger.Named("ociClient")
+func New(logger *zap.SugaredLogger, cp common.ConfigurationProvider, opRateLimiter *RateLimiter) (Interface, error) {
 	compute, err := core.NewComputeClientWithConfigurationProvider(cp)
 	if err != nil {
 		return nil, errors.Wrap(err, "NewComputeClientWithConfigurationProvider")
@@ -86,6 +124,7 @@ func New(logger *zap.SugaredLogger, cp common.ConfigurationProvider) (Interface,
 		compute:      &compute,
 		network:      &network,
 		loadbalancer: &lb,
+		rateLimiter:  *opRateLimiter,
 
 		subnetCache: cache.NewTTLStore(subnetCacheKeyFn, time.Duration(24)*time.Hour),
 		logger:      logger,

pkg/oci/client/client_test.go

@@ -15,9 +15,11 @@
 package client
 
 import (
+	"context"
 	"testing"
 
 	"github.com/oracle/oci-go-sdk/core"
+	"k8s.io/client-go/util/flowcontrol"
 )
 
 func TestInstanceTerminalState(t *testing.T) {
@@ -54,3 +56,90 @@ func TestInstanceTerminalState(t *testing.T) {
 		})
 	}
 }
+
+type mockComputeClient struct{}
+
+type mockVirtualNetworkClient struct{}
+
+type mockLoadBalancerClient struct{}
+
+func TestRateLimiting(t *testing.T) {
+	var qpsRead float32 = 5
+	bucketRead := 5
+	var qpsWrite float32 = 10
+	bucketWrite := 5
+
+	rateLimiter := RateLimiter{
+		Reader: flowcontrol.NewTokenBucketRateLimiter(
+			qpsRead,
+			bucketRead),
+		Writer: flowcontrol.NewTokenBucketRateLimiter(
+			qpsWrite,
+			bucketWrite),
+	}
+
+	client := newClient(rateLimiter)
+
+	// Read requests up to qpsRead should pass and the others fail
+	for i := 0; i < int(qpsRead)*2; i++ {
+		_, err := client.Compute().GetInstance(context.Background(), "123345")
+		p := (err == nil)
+
+		if (i < int(qpsRead) && !p) || (i >= int(qpsRead) && p) {
+			t.Errorf("unexpected result from request %d: %v", i, err)
+		}
+	}
+
+	// Write requests up to qpsWrite should pass and the others fail
+	ids := [2]string{"12334"}
+	for i := 0; i < int(qpsWrite)*2; i++ {
+		req := core.UpdateSecurityListRequest{
+			SecurityListId: &ids[0],
+		}
+
+		_, err := client.Networking().UpdateSecurityList(context.Background(), req)
+		p := (err == nil)
+
+		if (i < int(qpsRead) && !p) || (i >= int(qpsRead) && p) {
+			t.Errorf("unexpected result from request %d: %v", i, err)
+		}
+	}
+}
+
+func newClient(rateLimiter RateLimiter) Interface {
+	return &client{
+		compute:     &mockComputeClient{},
+		network:     &mockVirtualNetworkClient{},
+		rateLimiter: rateLimiter,
+	}
+}
+
+/* Mock ComputeClient interface implementations */
+func (c *mockComputeClient) GetInstance(ctx context.Context, request core.GetInstanceRequest) (response core.GetInstanceResponse, err error) {
+	return core.GetInstanceResponse{}, nil
+}
+
+func (c *mockComputeClient) ListInstances(ctx context.Context, request core.ListInstancesRequest) (response core.ListInstancesResponse, err error) {
+	return core.ListInstancesResponse{}, nil
+}
+
+func (c *mockComputeClient) ListVnicAttachments(ctx context.Context, request core.ListVnicAttachmentsRequest) (response core.ListVnicAttachmentsResponse, err error) {
+	return core.ListVnicAttachmentsResponse{}, nil
+}
+
+/* Mock NetworkClient interface implementations */
+func (c *mockVirtualNetworkClient) GetVnic(ctx context.Context, request core.GetVnicRequest) (response core.GetVnicResponse, err error) {
+	return core.GetVnicResponse{}, nil
+}
+
+func (c *mockVirtualNetworkClient) GetSubnet(ctx context.Context, request core.GetSubnetRequest) (response core.GetSubnetResponse, err error) {
+	return core.GetSubnetResponse{}, nil
+}
+
+func (c *mockVirtualNetworkClient) GetSecurityList(ctx context.Context, request core.GetSecurityListRequest) (response core.GetSecurityListResponse, err error) {
+	return core.GetSecurityListResponse{}, nil
+}
+
+func (c *mockVirtualNetworkClient) UpdateSecurityList(ctx context.Context, request core.UpdateSecurityListRequest) (response core.UpdateSecurityListResponse, err error) {
+	return core.UpdateSecurityListResponse{}, nil
+}