604 api rate limit

api/api.go

@@ -6,11 +6,13 @@ import (
 	"net"
 	"net/http"
 
+	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/grpc-ecosystem/grpc-gateway/runtime"
 	log "github.com/sirupsen/logrus"
 	"github.com/zoobc/zoobc-core/api/handler"
 	"github.com/zoobc/zoobc-core/api/service"
 	"github.com/zoobc/zoobc-core/common/chaintype"
+	"github.com/zoobc/zoobc-core/common/constant"
 	"github.com/zoobc/zoobc-core/common/crypto"
 	"github.com/zoobc/zoobc-core/common/interceptor"
 	"github.com/zoobc/zoobc-core/common/kvdb"
@@ -54,15 +56,19 @@ func startGrpcServer(
 	}
 	grpcServer := grpc.NewServer(
 		grpc.Creds(creds),
-		grpc.UnaryInterceptor(interceptor.NewServerInterceptor(
-			logger,
-			ownerAccountAddress,
-			map[codes.Code]string{
-				codes.Unavailable:     "indicates the destination service is currently unavailable",
-				codes.InvalidArgument: "indicates the argument request is invalid",
-				codes.Unauthenticated: "indicates the request is unauthenticated",
-			},
-		)),
+		grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(
+			interceptor.NewServerRateLimiterInterceptor(constant.MaxAPIRequestPerSecond),
+			interceptor.NewServerInterceptor(
+				logger,
+				ownerAccountAddress,
+				map[codes.Code]string{
+					codes.Unavailable:     "indicates the destination service is currently unavailable",
+					codes.InvalidArgument: "indicates the argument request is invalid",
+					codes.Unauthenticated: "indicates the request is unauthenticated",
+				},
+			),
+		),
+		),
 		grpc.StreamInterceptor(interceptor.NewStreamInterceptor(ownerAccountAddress)),
 	)
 	actionTypeSwitcher := &transaction.TypeSwitcher{

common/constant/api.go

@@ -0,0 +1,5 @@
+package constant
+
+var (
+	MaxAPIRequestPerSecond uint32 = 10
+)

common/interceptor/interceptor.go

@@ -3,19 +3,90 @@ package interceptor
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/signal"
 	"runtime"
+	"sync"
+	"syscall"
 	"time"
 
 	"github.com/sirupsen/logrus"
 	"github.com/zoobc/zoobc-core/common/blocker"
 	"github.com/zoobc/zoobc-core/common/crypto"
 	"github.com/zoobc/zoobc-core/common/model"
+	"github.com/zoobc/zoobc-core/common/monitoring"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
 
+type simpleRateLimiter struct {
+	numberOfAllowedRequest uint32
+	numberOfRequest        uint32
+	sync.Mutex
+}
+
+func (rl *simpleRateLimiter) isAllowed() bool {
+	rl.Lock()
+	defer rl.Unlock()
+	if rl.numberOfRequest >= rl.numberOfAllowedRequest {
+		return false
+	}
+	rl.numberOfRequest++
+	return true
+}
+
+func (rl *simpleRateLimiter) requestFinished() {
+	rl.Lock()
+	defer rl.Unlock()
+	if rl.numberOfRequest > 0 {
+		rl.numberOfRequest--
+	}
+}
+
+func (rl *simpleRateLimiter) start() {
+	ticker := time.NewTicker(time.Second)
+	sigs := make(chan os.Signal, 1)
+	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
+	for {
+		select {
+		case <-ticker.C:
+			func() {
+				rl.Lock()
+				defer rl.Unlock()
+				rl.numberOfRequest = 0
+			}()
+		case <-sigs:
+			ticker.Stop()
+			return
+		}
+	}
+}
+
+/*
+NewServerRateLimiterInterceptor function can used to add rate limit to the server call
+*/
+
+func NewServerRateLimiterInterceptor(requestLimitPerSecond uint32) grpc.UnaryServerInterceptor {
+	rateLimiter := &simpleRateLimiter{
+		numberOfAllowedRequest: requestLimitPerSecond,
+	}
+	go rateLimiter.start()
+	return func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (interface{}, error) {
+		if !rateLimiter.isAllowed() {
+			return nil, status.Error(codes.ResourceExhausted, "requests are limited")
+		}
+		defer rateLimiter.requestFinished()
+		return handler(ctx, req)
+	}
+}
+
 /*
 NewServerInterceptor function can use to inject middleware like:
 	- `recover`
@@ -34,6 +105,7 @@ func NewServerInterceptor(
 		info *grpc.UnaryServerInfo,
 		handler grpc.UnaryHandler,
 	) (interface{}, error) {
+		monitoring.IncrementRunningAPIHandling(info.FullMethod)
 		var (
 			authorizedErr, errHandler error
 			resp                      interface{}
@@ -49,10 +121,13 @@ func NewServerInterceptor(
 
 		defer func() {
 			var (
-				err = recover()
+				err     = recover()
+				latency = time.Since(start)
 			)
 
-			fields["latency"] = fmt.Sprintf("%d ns", time.Since(start).Nanoseconds())
+			fields["latency"] = fmt.Sprintf("%d ns", latency.Nanoseconds())
+			monitoring.SetAPIResponseTime(info.FullMethod, latency.Seconds())
+			monitoring.DecrementRunningAPIHandling(info.FullMethod)
 			if err != nil {
 				// get stack after panic called and perhaps its first error
 				_, file, line, _ := runtime.Caller(4)