FEATURE/MEDIUM: transaction: rate limiting on max amount of open transactions

Author: prometherion

configuration/configuration.go

@@ -58,6 +58,7 @@ type HAProxyConfiguration struct {
 	UpdateMapFilesPeriod int64  `long:"update-map-files-period" description:"Elapsed time in seconds between two maps syncing operations" default:"10"`
 	ClusterTLSCertDir    string `long:"cluster-tls-dir" description:"Path where cluster tls certificates will be stored. Defaults to same directory as dataplane configuration file"`
 	MasterWorkerMode     bool   `long:"master-worker-mode" description:"Flag to enable helpers when running within HAProxy"`
+	MaxOpenTransactions  int64  `long:"max-open-transactions" description:"Limit for active transaction in pending state" default:"20"`
 }
 
 type APIConfiguration struct {

configure_data_plane.go

@@ -54,6 +54,7 @@ import (
 	dataplaneapi_config "github.com/haproxytech/dataplaneapi/configuration"
 	"github.com/haproxytech/dataplaneapi/handlers"
 	"github.com/haproxytech/dataplaneapi/haproxy"
+	"github.com/haproxytech/dataplaneapi/rate"
 
 	"github.com/go-openapi/errors"
 	"github.com/go-openapi/runtime"
@@ -102,6 +103,15 @@ func configureFlags(api *operations.DataPlaneAPI) {
 	api.CommandLineOptionsGroups = append(api.CommandLineOptionsGroups, apiOptionsGroup)
 }
 
+func currentOpenTransactions(client *client_native.HAProxyClient) int {
+	ts, err := client.Configuration.GetTransactions("in_progress")
+	if err != nil {
+		log.Errorf("Cannot retrieve current open transactions for rate limit, default to zero (%s)", err.Error())
+		return 0
+	}
+	return len(*ts)
+}
+
 func configureAPI(api *operations.DataPlaneAPI) http.Handler {
 	cfg := dataplaneapi_config.Get()
 
@@ -261,6 +271,23 @@ func configureAPI(api *operations.DataPlaneAPI) http.Handler {
 	api.TransactionsGetTransactionHandler = &handlers.GetTransactionHandlerImpl{Client: client}
 	api.TransactionsGetTransactionsHandler = &handlers.GetTransactionsHandlerImpl{Client: client}
 	api.TransactionsCommitTransactionHandler = &handlers.CommitTransactionHandlerImpl{Client: client, ReloadAgent: ra}
+	if cfg.HAProxy.MaxOpenTransactions > 0 {
+		// creating the threshold limit using the CLI flag as hard quota and current open transactions as starting point
+		transactionLimiter := rate.NewThresholdLimit(uint64(cfg.HAProxy.MaxOpenTransactions), uint64(currentOpenTransactions(client)))
+
+		api.TransactionsStartTransactionHandler = &handlers.RateLimitedStartTransactionHandlerImpl{
+			TransactionCounter: transactionLimiter,
+			Handler:            api.TransactionsStartTransactionHandler,
+		}
+		api.TransactionsDeleteTransactionHandler = &handlers.RateLimitedDeleteTransactionHandlerImpl{
+			TransactionCounter: transactionLimiter,
+			Handler:            api.TransactionsDeleteTransactionHandler,
+		}
+		api.TransactionsCommitTransactionHandler = &handlers.RateLimitedCommitTransactionHandlerImpl{
+			TransactionCounter: transactionLimiter,
+			Handler:            api.TransactionsCommitTransactionHandler,
+		}
+	}
 
 	// setup sites handlers
 	api.SitesCreateSiteHandler = &handlers.CreateSiteHandlerImpl{Client: client, ReloadAgent: ra}

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/rs/cors v1.7.0
 	github.com/shirou/gopsutil v2.20.3+incompatible
 	github.com/sirupsen/logrus v1.5.0
+	github.com/stretchr/testify v1.6.1
 	golang.org/x/net v0.0.0-20200425230154-ff2c4b7c35a0
 	golang.org/x/sys v0.0.0-20200420163511-1957bb5e6d1f
 	gopkg.in/yaml.v2 v2.3.0

handlers/transaction.go

@@ -21,13 +21,26 @@ import (
 	"github.com/haproxytech/dataplaneapi/haproxy"
 	"github.com/haproxytech/dataplaneapi/misc"
 	"github.com/haproxytech/dataplaneapi/operations/transactions"
+	"github.com/haproxytech/dataplaneapi/rate"
 )
 
+// RateLimitedStartTransactionHandlerImpl decorates StartTransactionHandlerImpl with the rate limiting logic
+type RateLimitedStartTransactionHandlerImpl struct {
+	TransactionCounter rate.Threshold
+	Handler            transactions.StartTransactionHandler
+}
+
 //StartTransactionHandlerImpl implementation of the StartTransactionHandler interface using client-native client
 type StartTransactionHandlerImpl struct {
 	Client *client_native.HAProxyClient
 }
 
+// RateLimitedDeleteTransactionHandlerImpl decorates the DeleteTransactionHandlerImpl with the rate limiting logic
+type RateLimitedDeleteTransactionHandlerImpl struct {
+	TransactionCounter rate.Threshold
+	Handler            transactions.DeleteTransactionHandler
+}
+
 //DeleteTransactionHandlerImpl implementation of the DeleteTransactionHandler interface using client-native client
 type DeleteTransactionHandlerImpl struct {
 	Client *client_native.HAProxyClient
@@ -43,6 +56,12 @@ type GetTransactionsHandlerImpl struct {
 	Client *client_native.HAProxyClient
 }
 
+// RateLimitedCommitTransactionHandlerImpl decorates the CommitTransactionHandlerImpl with the rate limiting logic
+type RateLimitedCommitTransactionHandlerImpl struct {
+	TransactionCounter rate.Threshold
+	Handler            transactions.CommitTransactionHandler
+}
+
 //CommitTransactionHandlerImpl implementation of the CommitTransactionHandlerImpl interface using client-native client
 type CommitTransactionHandlerImpl struct {
 	Client      *client_native.HAProxyClient
@@ -111,3 +130,36 @@ func (th *CommitTransactionHandlerImpl) Handle(params transactions.CommitTransac
 	rID := th.ReloadAgent.Reload()
 	return transactions.NewCommitTransactionAccepted().WithReloadID(rID).WithPayload(t)
 }
+
+// Handle executes the decorated Handler and, in case of successful deletion, decrease the counter
+func (r RateLimitedDeleteTransactionHandlerImpl) Handle(params transactions.DeleteTransactionParams, principal interface{}) middleware.Responder {
+	res := r.Handler.Handle(params, principal)
+	if _, ok := res.(*transactions.DeleteTransactionNoContent); ok {
+		r.TransactionCounter.Decrease()
+	}
+	return res
+}
+
+// Handle executes the decorated Handler and, in case of successful creation, increase the counter if this is
+func (r RateLimitedStartTransactionHandlerImpl) Handle(params transactions.StartTransactionParams, principal interface{}) middleware.Responder {
+	if err := r.TransactionCounter.LimitReached(); err != nil {
+		e := misc.HandleError(err)
+		return transactions.NewStartTransactionDefault(int(*e.Code)).WithPayload(e)
+	}
+	res := r.Handler.Handle(params, principal)
+	if _, ok := res.(*transactions.StartTransactionCreated); ok {
+		r.TransactionCounter.Increase()
+	}
+	return res
+}
+
+func (r RateLimitedCommitTransactionHandlerImpl) Handle(params transactions.CommitTransactionParams, principal interface{}) middleware.Responder {
+	res := r.Handler.Handle(params, principal)
+	switch res.(type) {
+	case *transactions.CommitTransactionOK:
+		r.TransactionCounter.Decrease()
+	case *transactions.CommitTransactionAccepted:
+		r.TransactionCounter.Decrease()
+	}
+	return res
+}

misc/misc.go

@@ -22,11 +22,12 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/haproxytech/dataplaneapi/haproxy"
-
 	"github.com/haproxytech/client-native/v2/configuration"
 	client_errors "github.com/haproxytech/client-native/v2/errors"
 	"github.com/haproxytech/models/v2"
+
+	"github.com/haproxytech/dataplaneapi/haproxy"
+	"github.com/haproxytech/dataplaneapi/rate"
 )
 
 const (
@@ -38,6 +39,8 @@ const (
 	ErrHTTPInternalServerError = int64(500)
 	// ErrHTTPBadRequest HTTP status code 400
 	ErrHTTPBadRequest = int64(400)
+	// ErrHTTPRateLimit HTTP status code 429
+	ErrHTTPRateLimit = int64(429)
 )
 
 // HandleError translates error codes from client native into models.Error with appropriate http status code
@@ -61,6 +64,10 @@ func HandleError(err error) *models.Error {
 		httpCode := ErrHTTPBadRequest
 		msg := t.Error()
 		return &models.Error{Code: &httpCode, Message: &msg}
+	case *rate.TransactionLimitReachedErr:
+		httpCode := ErrHTTPRateLimit
+		msg := t.Error()
+		return &models.Error{Code: &httpCode, Message: &msg}
 	default:
 		msg := t.Error()
 		code := ErrHTTPInternalServerError

rate/error.go

@@ -0,0 +1,29 @@
+// Copyright 2019 HAProxy Technologies
+//
+// 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 rate
+
+import "fmt"
+
+func NewTransactionLimitReachedError(limit uint64) error {
+	return &TransactionLimitReachedErr{limit: limit}
+}
+
+type TransactionLimitReachedErr struct {
+	limit uint64
+}
+
+func (l TransactionLimitReachedErr) Error() string {
+	return fmt.Sprintf("cannot start a new transaction, reached the maximum amount of %d active transactions available", l.limit)
+}

rate/limiter.go

@@ -0,0 +1,21 @@
+// Copyright 2019 HAProxy Technologies
+//
+// 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 rate
+
+type Threshold interface {
+	LimitReached() error
+	Increase()
+	Decrease()
+}

rate/threshold_limit.go

@@ -0,0 +1,46 @@
+// Copyright 2019 HAProxy Technologies
+//
+// 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 rate
+
+import (
+	"sync/atomic"
+)
+
+type thresholdLimit struct {
+	limit  *uint64
+	actual *uint64
+}
+
+func (t *thresholdLimit) LimitReached() (err error) {
+	if *t.actual >= *t.limit {
+		err = NewTransactionLimitReachedError(*t.limit)
+	}
+	return
+}
+
+func NewThresholdLimit(limit uint64, startingFrom uint64) Threshold {
+	return &thresholdLimit{
+		actual: &startingFrom,
+		limit:  &limit,
+	}
+}
+
+func (t *thresholdLimit) Increase() {
+	atomic.AddUint64(t.actual, 1)
+}
+
+func (t thresholdLimit) Decrease() {
+	atomic.AddUint64(t.actual, ^uint64(0))
+}

rate/threshold_limit_test.go

@@ -0,0 +1,89 @@
+// Copyright 2019 HAProxy Technologies
+//
+// 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 rate
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewThresholdLimit(t *testing.T) {
+	type tc struct {
+		limit  uint64
+		actual uint64
+	}
+	for name, tc := range map[string]tc{
+		"from zero": {10, 0},
+		"reached":   {5, 5},
+	} {
+		t.Run(name, func(t *testing.T) {
+			l := NewThresholdLimit(tc.limit, tc.actual).(*thresholdLimit)
+			assert.Equal(t, *l.limit, tc.limit)
+			assert.Equal(t, *l.actual, tc.actual)
+		})
+	}
+}
+
+func Test_thresholdLimit_Decrease(t *testing.T) {
+	for _, tc := range []uint64{5, 10} {
+		t.Run(fmt.Sprintf("%d", tc), func(t *testing.T) {
+			l := thresholdLimit{
+				limit: func(v uint64) *uint64 {
+					return &v
+				}(tc),
+				actual: func(v uint64) *uint64 {
+					return &v
+				}(tc),
+			}
+			var counter uint64
+			for *l.actual > 0 {
+				l.Decrease()
+				counter++
+			}
+			assert.Equal(t, tc, counter)
+		})
+	}
+}
+
+func Test_thresholdLimit_Increase(t *testing.T) {
+	t.Run("success", func(t *testing.T) {
+		l := thresholdLimit{
+			limit: func(v uint64) *uint64 {
+				return &v
+			}(10),
+			actual: func(v uint64) *uint64 {
+				return &v
+			}(10),
+		}
+		var counter int
+		for *l.actual < *l.limit {
+			l.Increase()
+			counter++
+		}
+	})
+	t.Run("failure", func(t *testing.T) {
+		l := thresholdLimit{
+			limit: func(v uint64) *uint64 {
+				return &v
+			}(10),
+			actual: func(v uint64) *uint64 {
+				return &v
+			}(10),
+		}
+		assert.NotNil(t, l.LimitReached())
+	})
+}