#8168·keda

External Push Scaler: Partial metric discovery drops activation signals after endpoint recovery

Author: dakota-doordashCreated Sep 10, 2026Updated Sep 18, 2026

Labelsbugscalerscaler-external

Report

A transient metric-discovery failure for an external-push trigger can leave KEDA unable to use that trigger's subsequent positive activation signals, even after the external scaler recovers.

When another trigger returns a metric specification, the controller accepts the combined, incomplete result and writes it to ScaledObject.status.externalMetricNames. The push handler subsequently requires the missing trigger identity from that status field. It drops each positive activation when the name is absent, without initiating a discovery repair.

This is reproducible using upstream KEDA code and synthetic scalers. Kafka is used as the second trigger in the tests; no Kafka broker or application-specific scaler is required.

Expected Behavior

Transient discovery failure should not leave a recovered push scaler unable to activate its target indefinitely. KEDA should reject and retry incomplete discovery, or provide an equivalent recovery mechanism before relying on the incomplete metadata.

A transient failure after successful discovery should preserve the last complete discovery metadata and usable HPA configuration. Recovery should not require a user to change the ScaledObject specification.

Actual Behavior

Discovery returns HPA metric specifications without error even though the push trigger returned no specification.
Persisted external metric names contain only the second trigger's name, including when a previous complete list existed.
This happens both with ordinary HPA metrics and with advanced.scalingModifiers.
A subsequent successful cache discovery returns both metrics but leaves the persisted status incomplete.
The real push handler drops three consecutive positive activations in the reproduction. Updating only the persisted metric names restores activation through the same running handler.

A target at zero replicas can consequently lose its push-based wake-up path. Whether it remains at zero also depends on polling, other triggers, and subsequent controller reconciliation. This reproduction establishes the metadata and activation defects; it does not run a complete Kubernetes scaling loop or establish that reconciliation can never happen.

Steps to Reproduce the Problem

The two tests below isolate the controller and activation portions of the same failure sequence:

Configure trigger 0 as external-push and trigger 1 as kafka. Make trigger 0 return no metric specs while trigger 1 returns s1-broker. This models the external adapter's return value when its unary GetMetricSpec RPC fails.
Invoke the real controller metric-discovery method. Observe that it accepts discovery and persists only s1-broker.
Let the push scaler return s0-push on a later cache discovery. Observe that the persisted names remain incomplete.
Start the real push handler with that incomplete status and deliver repeated true activations. Observe zero calls to the scale executor. Repair only the metric names, deliver another true, and observe a scale-executor call.

For an executable reproduction, use a fresh upstream checkout:

bash

git clone --branch v2.20.2 --depth 1 https://github.com/kedacore/keda.git keda-repro
cd keda-repro

Save the following two code blocks at their indicated repository-relative paths, then run:

bash

go test -mod=vendor ./controllers/keda ./pkg/scaling \
  -run '^Test(ReproducePartialPushDiscovery|ReproduceMissingPushMetricActivation)$' \
  -count=1 -timeout=90s -v

These are diagnostic tests: PASS confirms the defective behavior on the affected revision. They assert partial persistence and dropped activations so that the two stages can be inspected independently. They should be converted to expected-behavior regression assertions when implementing a fix. The controller test covers initial and previously successful discovery, both with and without a composite metric.

controllers/keda/partial_discovery_repro_test.go

package keda

import (
	"context"
	"reflect"
	"testing"

	"github.com/go-logr/logr"
	"go.uber.org/mock/gomock"
	v2 "k8s.io/api/autoscaling/v2"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	"github.com/kedacore/keda/v2/apis/keda/v1alpha1"
	mock_scalers "github.com/kedacore/keda/v2/pkg/mock/mock_scaler"
	"github.com/kedacore/keda/v2/pkg/mock/mock_scaling"
	"github.com/kedacore/keda/v2/pkg/scaling/cache"
)

// Diagnostic reproduction: PASS means the incomplete-discovery bug is present.
func TestReproducePartialPushDiscovery(t *testing.T) {
	for _, composite := range []bool{false, true} {
		for _, previouslyHealthy := range []bool{false, true} {
			name := "plain/initial"
			if composite {
				name = "composite/initial"
			}
			if previouslyHealthy {
				name += "/previously-healthy"
			}
			t.Run(name, func(t *testing.T) {
				ctx := context.Background()
				ctrl := gomock.NewController(t)
				push := mock_scalers.NewMockScaler(ctrl)
				broker := mock_scalers.NewMockScaler(ctrl)
				handler := mock_scaling.NewMockScaleHandler(ctrl)
				so := &v1alpha1.ScaledObject{
					ObjectMeta: metav1.ObjectMeta{Name: "push-discovery-repro", Namespace: "default", Generation: 1},
					Spec: v1alpha1.ScaledObjectSpec{Triggers: []v1alpha1.ScaleTriggers{
						{Type: "external-push", Name: "push"},
						{Type: "kafka", Name: "broker"},
					}},
				}
				if composite {
					so.Spec.Advanced = &v1alpha1.AdvancedConfig{ScalingModifiers: v1alpha1.ScalingModifiers{
						Formula: "push + broker", Target: "1", MetricType: v2.AverageValueMetricType,
					}}
				}
				if previouslyHealthy {
					so.Status.ExternalMetricNames = []string{"s0-push", "s1-broker"}
				}
				scheme := runtime.NewScheme()
				if err := v1alpha1.AddToScheme(scheme); err != nil {
					t.Fatal(err)
				}
				client := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(so).WithObjects(so).Build()
				scalerCache := &cache.ScalersCache{Scalers: []cache.ScalerBuilder{{Scaler: push}, {Scaler: broker}}}
				metric := func(name string) v2.MetricSpec {
					return v2.MetricSpec{Type: v2.ExternalMetricSourceType, External: &v2.ExternalMetricSource{
						Metric: v2.MetricIdentifier{Name: name},
						Target: v2.MetricTarget{Type: v2.AverageValueMetricType, AverageValue: resource.NewQuantity(1, resource.DecimalSI)},
					}}
				}
				handler.EXPECT().GetScalersCache(ctx, gomock.Any()).Return(scalerCache, nil)
				push.EXPECT().GetMetricSpecForScaling(ctx).Return(nil)
				broker.EXPECT().GetMetricSpecForScaling(ctx).Return([]v2.MetricSpec{metric("s1-broker")}).AnyTimes()
				reconciler := &ScaledObjectReconciler{Client: client, ScaleHandler: handler}
				specs, err := reconciler.getScaledObjectMetricSpecs(ctx, logr.Discard(), so)
				if err != nil {
					t.Fatalf("partial discovery was rejected; defect not reproduced: %v", err)
				}
				expectedHPAName := "s1-broker"
				if composite {
					expectedHPAName = v1alpha1.CompositeMetricName
				}
				if len(specs) != 1 || specs[0].External == nil || specs[0].External.Metric.Name != expectedHPAName {
					t.Fatalf("unexpected HPA metric specs: %+v", specs)
				}
				assertIncompleteStatus := func() {
					t.Helper()
					persisted := &v1alpha1.ScaledObject{}
					if err := client.Get(ctx, types.NamespacedName{Name: so.Name, Namespace: so.Namespace}, persisted); err != nil {
						t.Fatal(err)
					}
					if !reflect.DeepEqual(persisted.Status.ExternalMetricNames, []string{"s1-broker"}) {
						t.Fatalf("unexpected persisted names: %v", persisted.Status.ExternalMetricNames)
					}
				}
				assertIncompleteStatus()
				push.EXPECT().GetMetricSpecForScaling(ctx).Return([]v2.MetricSpec{metric("s0-push")})
				if len(scalerCache.GetMetricSpecForScaling(ctx)) != 2 {
					t.Fatal("expected both specs after simulated endpoint recovery")
				}
				assertIncompleteStatus()
				t.Log("REPRODUCED: partial discovery persisted; successful cache discovery did not repair status")
			})
		}
	}
}

pkg/scaling/push_activation_repro_test.go

package scaling

import (
	"context"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"go.uber.org/mock/gomock"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
	"github.com/kedacore/keda/v2/pkg/mock/mock_scaling/mock_executor"
	"github.com/kedacore/keda/v2/pkg/scalers"
	"github.com/kedacore/keda/v2/pkg/scalers/scalersconfig"
	"github.com/kedacore/keda/v2/pkg/scaling/cache"
	"github.com/kedacore/keda/v2/pkg/scaling/executor"
)

type reproActivation struct {
	value bool
	acked chan struct{}
}

type reproPushProbe struct {
	scalers.Scaler
	commands chan reproActivation
	stopped  chan struct{}
}

func (p *reproPushProbe) Run(ctx context.Context, active chan<- bool) {
	defer close(active)
	defer close(p.stopped)
	for {
		select {
		case <-ctx.Done():
			return
		case command := <-p.commands:
			select {
			case <-ctx.Done():
				return
			case active <- command.value:
				close(command.acked)
			}
		}
	}
}

// Diagnostic reproduction: PASS means the missing-identity bug is present.
// Keep the actual push handler running, prove it drops repeated
// true signals with incomplete discovery metadata, then repair only that status.
func TestReproduceMissingPushMetricActivation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	const pushMetric = "s0-push"
	so := &kedav1alpha1.ScaledObject{
		TypeMeta:   metav1.TypeMeta{APIVersion: "keda.sh/v1alpha1", Kind: "ScaledObject"},
		ObjectMeta: metav1.ObjectMeta{Name: "push-discovery", Namespace: "default", Generation: 1},
		Spec: kedav1alpha1.ScaledObjectSpec{
			Triggers: []kedav1alpha1.ScaleTriggers{{Type: "external-push", Name: "push"}, {Type: "kafka", Name: "broker"}},
		},
		Status: kedav1alpha1.ScaledObjectStatus{ExternalMetricNames: []string{"s1-broker"}},
	}
	scheme := runtime.NewScheme()
	if err := kedav1alpha1.AddToScheme(scheme); err != nil {
		t.Fatal(err)
	}
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithStatusSubresource(so).WithObjects(so).Build()
	probe := &reproPushProbe{commands: make(chan reproActivation), stopped: make(chan struct{})}
	ctrl := gomock.NewController(t)
	actuator := mock_executor.NewMockScaleExecutor(ctrl)
	var scaleCalls atomic.Int32
	actuator.EXPECT().RequestScale(gomock.Any(), gomock.Any(), true, false, gomock.Any()).DoAndReturn(
		func(_ context.Context, _ *kedav1alpha1.ScaledObject, _, _ bool, opts executor.ScaleExecutorOptions) executor.ScaleResult {
			if opts.ForPushScalerMetric != pushMetric || len(opts.ActiveTriggers) != 1 || opts.ActiveTriggers[0] != pushMetric {
				t.Errorf("wrong trigger identity: %+v", opts)
			}
			scaleCalls.Add(1)
			return executor.ScaleResult{}
		},
	).AnyTimes()
	h := &scaleHandler{
		client: kubeClient, scaleExecutor: actuator, scalerCachesLock: &sync.RWMutex{},
		scalerCaches: map[string]*cache.ScalersCache{so.GenerateIdentifier(): {
			ScalableObjectGeneration: so.Generation,
			Scalers:                  []cache.ScalerBuilder{{Scaler: probe, ScalerConfig: scalersconfig.ScalerConfig{TriggerIndex: 0}}},
		}},
	}
	duck, err := kedav1alpha1.AsDuckWithTriggers(so)
	if err != nil {
		t.Fatal(err)
	}
	h.startPushScalers(ctx, duck, so.DeepCopy(), &sync.Mutex{})
	publish := func(value bool) {
		t.Helper()
		command := reproActivation{value: value, acked: make(chan struct{})}
		select {
		case probe.commands <- command:
		case <-time.After(time.Second):
			t.Fatal("push stream did not accept test command")
		}
		select {
		case <-command.acked:
		case <-time.After(time.Second):
			t.Fatal("push handler did not receive activation")
		}
	}
	for range 3 {
		publish(true)
	}
	// Delivery of a subsequent signal is a barrier: the preceding activation
	// has completed processing in the handler's serial receive loop.
	publish(false)
	if got := scaleCalls.Load(); got != 0 {
		t.Fatalf("expected missing-name defect, got %d scale calls", got)
	}
	so.Status.ExternalMetricNames = []string{pushMetric, "s1-broker"}
	if err := kubeClient.Status().Update(ctx, so); err != nil {
		t.Fatal(err)
	}
	publish(true)
	publish(false)
	if got := scaleCalls.Load(); got != 1 {
		t.Fatalf("status repair did not restore activation: %d calls", got)
	}
	cancel()
	select {
	case <-probe.stopped:
	case <-time.After(time.Second):
		t.Fatal("test push stream did not stop")
	}
	t.Log("REPRODUCED: three true activations dropped; repairing only persisted metric names restored the existing push handler")
}

Logs from KEDA operator

The relevant message in the affected push handler is logged at verbosity 1:

Could not resolve metric name for push scaler, will retry on next activation

This is a source-level diagnostic string, not an excerpt from a deployed environment. The unit reproduction emits:

REPRODUCED: partial discovery persisted; successful cache discovery did not repair status
REPRODUCED: three true activations dropped; repairing only persisted metric names restored the existing push handler

KEDA Version

2.20.2

Kubernetes Version

Other

Platform

Other

Scaler Details

external-push plus a second trigger that successfully returns metric specifications. The synthetic example uses kafka for the second trigger and generic metric names s0-push and s1-broker.

Would you be open to contributing a fix?

Maybe

Anything else?

Relevant upstream code:

A failed unary GetMetricSpec RPC produces an empty result in the external adapter.
The cache concatenates discovery results, and the controller rejects only an entirely empty aggregate. It later persists the resulting status.
The push handler's missing-name branch skips RequestScale and repeats the same lookup on the next activation.

The aggregate-only guard and missing-name branch were also inspected at upstream main commit aef28cb2a8a91980529dec2a5ee8dbb0ea753cfe: controller, push handler. The runnable diagnostic above is pinned to v2.20.2; no claim is made about all earlier releases.

Source: kedacore/keda

View original on GitHub View discussion on GitHub