Add beanstalkd scaler

CHANGELOG.md

@@ -57,6 +57,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 
 ### New
 
+- **Beanstalkd**: Introduce new beanstalkd scaler ([#5901](https://github.com/kedacore/keda/issues/5901))
 - **CloudEventSource**: Introduce ClusterCloudEventSource ([#3533](https://github.com/kedacore/keda/issues/3533))
 - **CloudEventSource**: Provide CloudEvents around the management of ScaledJobs resources ([#3523](https://github.com/kedacore/keda/issues/3523))
 

go.mod

@@ -117,6 +117,7 @@ require (
 	sigs.k8s.io/controller-tools v0.15.0
 	sigs.k8s.io/custom-metrics-apiserver v1.29.0
 	sigs.k8s.io/kustomize/kustomize/v5 v5.4.3
+	github.com/beanstalkd/go-beanstalk v0.2.0
 )
 
 // Remove this when they merge the PR and cut a release https://github.com/open-policy-agent/cert-controller/pull/202

go.sum

@@ -947,6 +947,8 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.30.3/go.mod h1:zwySh8fpFyXp9yOr/KVzx
 github.com/aws/smithy-go v1.13.0/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
 github.com/aws/smithy-go v1.20.3 h1:ryHwveWzPV5BIof6fyDvor6V3iUL7nTfiTKXHiW05nE=
 github.com/aws/smithy-go v1.20.3/go.mod h1:krry+ya/rV9RDcV/Q16kpu6ypI4K2czasz0NC3qS14E=
+github.com/beanstalkd/go-beanstalk v0.2.0 h1:6UOJugnu47uNB2jJO/lxyDgeD1Yds7owYi1USELqexA=
+github.com/beanstalkd/go-beanstalk v0.2.0/go.mod h1:/G8YTyChOtpOArwLTQPY1CHB+i212+av35bkPXXj56Y=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=

pkg/scalers/beanstalkd_scaler.go

@@ -0,0 +1,179 @@
+package scalers
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"time"
+
+	beanstalk "github.com/beanstalkd/go-beanstalk"
+	"github.com/go-logr/logr"
+	"github.com/mitchellh/mapstructure"
+	v2 "k8s.io/api/autoscaling/v2"
+	"k8s.io/metrics/pkg/apis/external_metrics"
+
+	"github.com/kedacore/keda/v2/pkg/scalers/scalersconfig"
+	"github.com/kedacore/keda/v2/pkg/util"
+)
+
+const (
+	beanstalkdJobsMetricName                   = "jobs"
+	beanstalkdValueConfigName                  = "value"
+	beanstalkdActivationValueTriggerConfigName = "activationValue"
+	beanstalkdMetricType                       = "External"
+	beanstalkdNetworkProtocol                  = "tcp"
+)
+
+type BeanstalkdScaler struct {
+	metricType v2.MetricTargetType
+	metadata   *BeanstalkdMetadata
+	connection *beanstalk.Conn
+	tube       *beanstalk.Tube
+	logger     logr.Logger
+}
+
+type BeanstalkdMetadata struct {
+	Server          string  `keda:"name=server, order=triggerMetadata"`
+	Tube            string  `keda:"name=tube, order=triggerMetadata"`
+	Value           float64 `keda:"name=value, order=triggerMetadata"`
+	ActivationValue float64 `keda:"name=activationValue, order=triggerMetadata, optional"`
+	IncludeDelayed  bool    `keda:"name=includeDelayed, order=triggerMetadata, optional"`
+	Timeout         uint    `keda:"name=timeout, order=triggerMetadata, optional, default=30"`
+	TriggerIndex    int
+}
+
+// TubeStats represents a set of tube statistics.
+type tubeStats struct {
+	TotalJobs    int64 `mapstructure:"total-jobs"`
+	JobsReady    int64 `mapstructure:"current-jobs-ready"`
+	JobsReserved int64 `mapstructure:"current-jobs-reserved"`
+	JobsUrgent   int64 `mapstructure:"current-jobs-urgent"`
+	JobsBuried   int64 `mapstructure:"current-jobs-buried"`
+	JobsDelayed  int64 `mapstructure:"current-jobs-delayed"`
+}
+
+func NewBeanstalkdScaler(config *scalersconfig.ScalerConfig) (Scaler, error) {
+	s := &BeanstalkdScaler{}
+
+	metricType, err := GetMetricTargetType(config)
+	if err != nil {
+		return nil, fmt.Errorf("error getting scaler metric type: %w", err)
+	}
+	s.metricType = metricType
+
+	s.logger = InitializeLogger(config, "beanstalkd_scaler")
+
+	meta, err := parseBeanstalkdMetadata(config)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing beanstalkd metadata: %w", err)
+	}
+	s.metadata = meta
+
+	timeout := time.Duration(s.metadata.Timeout) * time.Second
+
+	conn, err := beanstalk.DialTimeout(beanstalkdNetworkProtocol, s.metadata.Server, timeout)
+	if err != nil {
+		return nil, fmt.Errorf("error connecting to beanstalkd: %w", err)
+	}
+
+	s.connection = conn
+
+	s.tube = beanstalk.NewTube(s.connection, meta.Tube)
+
+	return s, nil
+}
+
+func parseBeanstalkdMetadata(config *scalersconfig.ScalerConfig) (*BeanstalkdMetadata, error) {
+	meta := &BeanstalkdMetadata{}
+
+	meta.TriggerIndex = config.TriggerIndex
+	if err := config.TypedConfig(meta); err != nil {
+		return nil, fmt.Errorf("error parsing beanstalkd metadata: %w", err)
+	}
+
+	return meta, nil
+}
+
+func (s *BeanstalkdScaler) getTubeStats(ctx context.Context) (*tubeStats, error) {
+	errCh := make(chan error)
+	statsCh := make(chan *tubeStats)
+
+	go func() {
+		rawStats, err := s.tube.Stats()
+		if err != nil {
+			errCh <- fmt.Errorf("error retrieving stats from beanstalkd: %w", err)
+		}
+
+		var stats tubeStats
+		err = mapstructure.WeakDecode(rawStats, &stats)
+		if err != nil {
+			errCh <- fmt.Errorf("error decoding stats from beanstalkd: %w", err)
+		}
+
+		statsCh <- &stats
+	}()
+
+	select {
+	case err := <-errCh:
+		if errors.Is(err, beanstalk.ErrNotFound) {
+			s.logger.Info("tube not found, setting stats to 0")
+			return &tubeStats{
+				TotalJobs:    0,
+				JobsReady:    0,
+				JobsDelayed:  0,
+				JobsReserved: 0,
+				JobsUrgent:   0,
+				JobsBuried:   0,
+			}, nil
+		}
+		return nil, err
+	case tubeStats := <-statsCh:
+		return tubeStats, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (s *BeanstalkdScaler) GetMetricsAndActivity(ctx context.Context, metricName string) ([]external_metrics.ExternalMetricValue, bool, error) {
+	stats, err := s.getTubeStats(ctx)
+	if err != nil {
+		return []external_metrics.ExternalMetricValue{}, false, fmt.Errorf("error interacting with beanstalkd: %w", err)
+	}
+
+	totalJobs := stats.JobsReady + stats.JobsReserved
+
+	if s.metadata.IncludeDelayed {
+		totalJobs += stats.JobsDelayed
+	}
+
+	metric := GenerateMetricInMili(metricName, float64(totalJobs))
+	isActive := float64(totalJobs) > s.metadata.ActivationValue
+
+	return []external_metrics.ExternalMetricValue{metric}, isActive, nil
+}
+
+func (s *BeanstalkdScaler) GetMetricSpecForScaling(context.Context) []v2.MetricSpec {
+	externalMetric := &v2.ExternalMetricSource{
+		Metric: v2.MetricIdentifier{
+			Name: GenerateMetricNameWithIndex(s.metadata.TriggerIndex, util.NormalizeString(fmt.Sprintf("beanstalkd-%s", url.QueryEscape(s.metadata.Tube)))),
+		},
+		Target: GetMetricTargetMili(s.metricType, s.metadata.Value),
+	}
+	metricSpec := v2.MetricSpec{
+		External: externalMetric, Type: beanstalkdMetricType,
+	}
+
+	return []v2.MetricSpec{metricSpec}
+}
+
+func (s *BeanstalkdScaler) Close(context.Context) error {
+	if s.connection != nil {
+		err := s.connection.Close()
+		if err != nil {
+			s.logger.Error(err, "Error closing beanstalkd connection")
+			return err
+		}
+	}
+	return nil
+}