fix(splunk-observability): prevent indefinite block on stuck stream

CHANGELOG.md

@@ -141,6 +141,7 @@ To learn more about active deprecations, we recommend checking [GitHub Discussio
 - **Redis Scaler**: Use literal command names in Lua script to fix compatibility with Alibaba Cloud Redis Cluster ([#7758](https://github.com/kedacore/keda/issues/7758))
 - **Solace Scaler**: Fix URL escaping for Message VPN and Queue names ([#7481](https://github.com/kedacore/keda/pull/7481))
 - **Solr Scaler**: Use net/url to safely encode query parameters ([#7467](https://github.com/kedacore/keda/pull/7467))
+- **Splunk Observability Scaler**: Add MTS stream handling with context timeout ([#7799](https://github.com/kedacore/keda/pull/7799))
 
 ### Deprecations
 

pkg/scalers/splunk_observability_scaler.go

@@ -15,6 +15,12 @@ import (
 	kedautil "github.com/kedacore/keda/v2/pkg/util"
 )
 
+// splunkO11yStreamMargin bounds the stream read beyond the configured Duration.
+const splunkO11yStreamMargin = 10 * time.Second
+
+// splunkO11yDrainTimeout is a short best-effort budget for draining after Stop.
+const splunkO11yDrainTimeout = 2 * time.Second
+
 type splunkObservabilityMetadata struct {
 	TriggerIndex int
 
@@ -78,6 +84,30 @@ func NewSplunkObservabilityScaler(config *scalersconfig.ScalerConfig) (Scaler, e
 	}, nil
 }
 
+// stopAndDrain stops the computation and drains the data channel so the client's goroutines can exit.
+func (s *splunkObservabilityScaler) stopAndDrain(comp *signalflow.Computation) {
+	// Fresh, short context: the caller's may be done and some backends never close on Stop.
+	stopCtx, cancel := context.WithTimeout(context.Background(), splunkO11yDrainTimeout)
+	defer cancel()
+
+	if err := comp.Stop(stopCtx); err != nil {
+		s.logger.V(1).Info("Failed to stop SignalFlow computation", "error", err)
+	}
+
+	dataCh := comp.Data()
+	for {
+		select {
+		case _, ok := <-dataCh:
+			if !ok {
+				return
+			}
+		case <-stopCtx.Done():
+			s.logger.V(1).Info("Gave up draining SignalFlow data channel after stop")
+			return
+		}
+	}
+}
+
 func (s *splunkObservabilityScaler) getQueryResult(ctx context.Context) (float64, error) {
 	comp, err := s.apiClient.Execute(ctx, &signalflow.ExecuteRequest{
 		Program: s.metadata.Query,
@@ -88,35 +118,50 @@ func (s *splunkObservabilityScaler) getQueryResult(ctx context.Context) (float64
 
 	s.logger.V(1).Info("Started MTS stream.")
 
-	stopTimer := time.After(time.Duration(s.metadata.Duration) * time.Second)
-	go func() {
-		<-stopTimer
-		s.logger.V(1).Info("Stopping MTS stream after duration.")
-		if err := comp.Stop(ctx); err != nil {
-			s.logger.Error(err, "Failed to stop SignalFlow computation")
-		}
-	}()
+	// Hard deadline beyond the Duration window so a non-responsive backend cannot block forever.
+	streamDuration := time.Duration(s.metadata.Duration) * time.Second
+	streamCtx, cancel := context.WithTimeout(ctx, streamDuration+splunkO11yStreamMargin)
+	defer cancel()
+
+	stopTimer := time.After(streamDuration)
 
 	maxValue := math.Inf(-1)
 	minValue := math.Inf(1)
 	valueSum := 0.0
 	valueCount := 0
 	s.logger.V(1).Info("Now iterating over results.")
-	for msg := range comp.Data() {
-		if len(msg.Payloads) == 0 {
-			s.logger.V(1).Info("No data retrieved.")
-			continue
-		}
-		for _, pl := range msg.Payloads {
-			value, ok := pl.Value().(float64)
+
+	dataCh := comp.Data()
+loop:
+	for {
+		select {
+		case <-streamCtx.Done():
+			s.logger.V(1).Info("Context done before stream completed; stopping computation.")
+			s.stopAndDrain(comp)
+			return -1, fmt.Errorf("splunk observability query did not complete in time: %w", streamCtx.Err())
+		case <-stopTimer:
+			s.logger.V(1).Info("Stopping MTS stream after duration.")
+			s.stopAndDrain(comp)
+			break loop
+		case msg, ok := <-dataCh:
 			if !ok {
-				return -1, fmt.Errorf("could not convert Splunk Observability metric value to float64")
+				break loop
+			}
+			if len(msg.Payloads) == 0 {
+				s.logger.V(1).Info("No data retrieved.")
+				continue
+			}
+			for _, pl := range msg.Payloads {
+				value, ok := pl.Value().(float64)
+				if !ok {
+					return -1, fmt.Errorf("could not convert Splunk Observability metric value to float64")
+				}
+				s.logger.V(1).Info(fmt.Sprintf("Encountering value %.4f\n", value))
+				maxValue = math.Max(maxValue, value)
+				minValue = math.Min(minValue, value)
+				valueSum += value
+				valueCount++
 			}
-			s.logger.V(1).Info(fmt.Sprintf("Encountering value %.4f\n", value))
-			maxValue = math.Max(maxValue, value)
-			minValue = math.Min(minValue, value)
-			valueSum += value
-			valueCount++
 		}
 	}
 

pkg/scalers/splunk_observability_scaler_test.go

@@ -3,6 +3,11 @@ package scalers
 import (
 	"context"
 	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	"github.com/signalfx/signalflow-client-go/v2/signalflow"
+	"github.com/signalfx/signalfx-go/idtool"
 
 	"github.com/kedacore/keda/v2/pkg/scalers/scalersconfig"
 )
@@ -94,3 +99,58 @@ func TestSplunkObservabilityGetMetricSpecForScaling(t *testing.T) {
 		}
 	}
 }
+
+// newFakeSplunkO11yScaler wires a scaler to a fake backend that streams indefinitely without closing.
+func newFakeSplunkO11yScaler(t *testing.T, program string, duration int) (*splunkObservabilityScaler, func()) {
+	t.Helper()
+
+	fake := signalflow.NewRunningFakeBackend()
+	client, err := fake.Client()
+	if err != nil {
+		fake.Stop()
+		t.Fatal("could not create fake backend client:", err)
+	}
+
+	tsid := idtool.ID(1)
+	fake.AddProgramTSIDs(program, []idtool.ID{tsid})
+	fake.SetTSIDFloatData(tsid, 42.0)
+
+	scaler := &splunkObservabilityScaler{
+		metadata: &splunkObservabilityMetadata{
+			Query:           program,
+			Duration:        duration,
+			QueryAggregator: "max",
+		},
+		apiClient: client,
+		logger:    logr.Discard(),
+	}
+
+	return scaler, fake.Stop
+}
+
+// Regression guard: a stuck stream must not block getQueryResult past the parent context deadline.
+func TestSplunkObservabilityGetQueryResultReturnsOnParentContextCancel(t *testing.T) {
+	const program = "data('demo.trans.latency').max().publish()"
+	// Large duration so the stopTimer never fires; the parent deadline must bound the call.
+	scaler, stop := newFakeSplunkO11yScaler(t, program, 3600)
+	defer stop()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	done := make(chan struct{})
+	start := time.Now()
+	go func() {
+		defer close(done)
+		_, _ = scaler.getQueryResult(ctx)
+	}()
+
+	select {
+	case <-done:
+		if elapsed := time.Since(start); elapsed > 5*time.Second {
+			t.Fatalf("getQueryResult returned after %v, far longer than the context deadline", elapsed)
+		}
+	case <-time.After(10 * time.Second):
+		t.Fatal("getQueryResult did not return after parent context was cancelled; it is hanging")
+	}
+}