TUN-8861: Rename Session Limiter to Flow Limiter

## Summary
Session is the concept used for UDP flows. Therefore, to make
the session limiter ambiguous for both TCP and UDP, this commit
renames it to flow limiter.

Closes TUN-8861

flow/limiter.go

@@ -0,0 +1,77 @@
+package flow
+
+import (
+	"errors"
+	"sync"
+)
+
+const (
+	unlimitedActiveFlows = 0
+)
+
+var (
+	ErrTooManyActiveFlows = errors.New("too many active flows")
+)
+
+type Limiter interface {
+	// Acquire tries to acquire a free slot for a flow, if the value of flows is already above
+	// the maximum it returns ErrTooManyActiveFlows.
+	Acquire(flowType string) error
+	// Release releases a slot for a flow.
+	Release()
+	// SetLimit allows to hot swap the limit value of the limiter.
+	SetLimit(uint64)
+}
+
+type flowLimiter struct {
+	limiterLock        sync.Mutex
+	activeFlowsCounter uint64
+	maxActiveFlows     uint64
+	unlimited          bool
+}
+
+func NewLimiter(maxActiveFlows uint64) Limiter {
+	flowLimiter := &flowLimiter{
+		maxActiveFlows: maxActiveFlows,
+		unlimited:      isUnlimited(maxActiveFlows),
+	}
+
+	return flowLimiter
+}
+
+func (s *flowLimiter) Acquire(flowType string) error {
+	s.limiterLock.Lock()
+	defer s.limiterLock.Unlock()
+
+	if !s.unlimited && s.activeFlowsCounter >= s.maxActiveFlows {
+		flowRegistrationsDropped.WithLabelValues(flowType).Inc()
+		return ErrTooManyActiveFlows
+	}
+
+	s.activeFlowsCounter++
+	return nil
+}
+
+func (s *flowLimiter) Release() {
+	s.limiterLock.Lock()
+	defer s.limiterLock.Unlock()
+
+	if s.activeFlowsCounter <= 0 {
+		return
+	}
+
+	s.activeFlowsCounter--
+}
+
+func (s *flowLimiter) SetLimit(newMaxActiveFlows uint64) {
+	s.limiterLock.Lock()
+	defer s.limiterLock.Unlock()
+
+	s.maxActiveFlows = newMaxActiveFlows
+	s.unlimited = isUnlimited(newMaxActiveFlows)
+}
+
+// isUnlimited checks if the value received matches the configuration for the unlimited flow limiter.
+func isUnlimited(value uint64) bool {
+	return value == unlimitedActiveFlows
+}

flow/limiter_test.go

@@ -0,0 +1,119 @@
+package flow_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/flow"
+)
+
+func TestFlowLimiter_Unlimited(t *testing.T) {
+	unlimitedLimiter := flow.NewLimiter(0)
+
+	for i := 0; i < 1000; i++ {
+		err := unlimitedLimiter.Acquire("test")
+		require.NoError(t, err)
+	}
+}
+
+func TestFlowLimiter_Limited(t *testing.T) {
+	maxFlows := uint64(5)
+	limiter := flow.NewLimiter(maxFlows)
+
+	for i := uint64(0); i < maxFlows; i++ {
+		err := limiter.Acquire("test")
+		require.NoError(t, err)
+	}
+
+	err := limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+}
+
+func TestFlowLimiter_AcquireAndReleaseFlow(t *testing.T) {
+	maxFlows := uint64(5)
+	limiter := flow.NewLimiter(maxFlows)
+
+	// Acquire the maximum number of flows
+	for i := uint64(0); i < maxFlows; i++ {
+		err := limiter.Acquire("test")
+		require.NoError(t, err)
+	}
+
+	// Validate acquire 1 more flows fails
+	err := limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+
+	// Release the maximum number of flows
+	for i := uint64(0); i < maxFlows; i++ {
+		limiter.Release()
+	}
+
+	// Validate acquire 1 more flows works
+	err = limiter.Acquire("shouldn't fail")
+	require.NoError(t, err)
+
+	// Release a 10x the number of max flows
+	for i := uint64(0); i < 10*maxFlows; i++ {
+		limiter.Release()
+	}
+
+	// Validate it still can only acquire a value = number max flows.
+	for i := uint64(0); i < maxFlows; i++ {
+		err := limiter.Acquire("test")
+		require.NoError(t, err)
+	}
+	err = limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+}
+
+func TestFlowLimiter_SetLimit(t *testing.T) {
+	maxFlows := uint64(5)
+	limiter := flow.NewLimiter(maxFlows)
+
+	// Acquire the maximum number of flows
+	for i := uint64(0); i < maxFlows; i++ {
+		err := limiter.Acquire("test")
+		require.NoError(t, err)
+	}
+
+	// Validate acquire 1 more flows fails
+	err := limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+
+	// Set the flow limiter to support one more request
+	limiter.SetLimit(maxFlows + 1)
+
+	// Validate acquire 1 more flows now works
+	err = limiter.Acquire("shouldn't fail")
+	require.NoError(t, err)
+
+	// Validate acquire 1 more flows doesn't work because we already reached the limit
+	err = limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+
+	// Release all flows
+	for i := uint64(0); i < maxFlows+1; i++ {
+		limiter.Release()
+	}
+
+	// Validate 1 flow works again
+	err = limiter.Acquire("shouldn't fail")
+	require.NoError(t, err)
+
+	// Set the flow limit to 1
+	limiter.SetLimit(1)
+
+	// Validate acquire 1 more flows doesn't work
+	err = limiter.Acquire("should fail")
+	require.ErrorIs(t, err, flow.ErrTooManyActiveFlows)
+
+	// Set the flow limit to unlimited
+	limiter.SetLimit(0)
+
+	// Validate it can acquire a lot of flows because it is now unlimited.
+	for i := uint64(0); i < 10*maxFlows; i++ {
+		err := limiter.Acquire("shouldn't fail")
+		require.NoError(t, err)
+	}
+}

flow/metrics.go

@@ -0,0 +1,23 @@
+package flow
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	namespace = "flow"
+)
+
+var (
+	labels = []string{"flow_type"}
+
+	flowRegistrationsDropped = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: namespace,
+		Subsystem: "client",
+		Name:      "registrations_rate_limited_total",
+		Help:      "Count registrations dropped due to high number of concurrent flows being handled",
+	},
+		labels,
+	)
+)