Skip to content

Commit 4213128

Browse files
harshrajdebugharshrajdebug
authored
External Scaler: strengthen the connection pool tests and pin the TLS error path (#8213)
* External Scaler: strengthen the connection pool tests and pin the TLS error path Follow-up to #8193, addressing the review points left open at merge. Look the pooled connection up by its key. The tests found their entry by scanning connectionPool for a connectionGroup with an expected refCount, which matches any entry in a pool shared by the whole package, so a test could pass on a connection another test created. getConnectionPoolKey gives each test the key it built with, and poolEntryFor loads that entry directly. Release each scaler on test cleanup. Every test relied on its own Close calls to leave the pool empty, so a test failing part way through left an entry behind for the tests that follow. Close is idempotent, so a cleanup Close after an explicit one is a no-op. Cover the push scaler. NewExternalPushScaler acquires a share of the same pool and externalPushScaler takes Close from the externalScaler it embeds, neither of which was exercised. The new test also shares one connection between a push scaler and a plain scaler on the same address. Pin where a TLS error is reported. The pooled connection is now built from the scaler constructor, so an unparseable client certificate is reported there rather than on the first poll. buildScalers turns a constructor error into a failure for the whole ScaledObject, so this moves a malformed certificate from degrading one trigger at poll time to failing every trigger on the object at build time. The test records that behaviour so a change to it is deliberate, and checks the failed construction leaves nothing in the pool. Signed-off-by: Harsh Raj <harshrajdebug@gmail.com> * External Scaler: clean up TCP test scalers and drop a lookup that cannot match Register cleanup for the scalers the TCP test builds, as the pool tests already do, so a test that fails part way through does not leave its pooled connection behind for the tests that follow. Drop the poolEntryFor check from the TLS test. The pool key includes the TLS fields, and poolEntryFor builds its key from the address alone, so it can never find an entry left by the failed construction. The poolEntries count next to it is what catches that case. Signed-off-by: Harsh Raj <218219077+harshrajdebug@users.noreply.github.com> * External Scaler: close the scaler if the TLS test's constructor succeeds Keep the scaler the TLS test builds and register its cleanup before failing, so a constructor that wrongly succeeds does not leave a live scaler and its pooled connection behind for the tests that follow. Signed-off-by: Harsh Raj <harshrajdebug@gmail.com> --------- Signed-off-by: Harsh Raj <harshrajdebug@gmail.com> Signed-off-by: Harsh Raj <218219077+harshrajdebug@users.noreply.github.com>
1 parent 81c6b83 commit 4213128

2 files changed

Lines changed: 174 additions & 46 deletions

File tree

‎pkg/scalers/external_scaler_pool_test.go‎

Lines changed: 173 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,49 @@ func poolEntries() int {
1818
return n
1919
}
2020

21+
// poolEntryFor returns the pooled connection for an address, or nil when the
22+
// pool holds none. Looking the entry up by its key keeps each test pinned to
23+
// the connection it created, so a test cannot pass by finding an entry another
24+
// test left in the shared pool.
25+
func poolEntryFor(t *testing.T, address string) *connectionGroup {
26+
t.Helper()
27+
key, err := getConnectionPoolKey(externalScalerMetadata{ScalerAddress: address})
28+
if err != nil {
29+
t.Fatalf("getConnectionPoolKey: %v", err)
30+
}
31+
i, ok := connectionPool.Load(key)
32+
if !ok {
33+
return nil
34+
}
35+
connGroup, ok := i.(*connectionGroup)
36+
if !ok {
37+
t.Fatalf("pool entry for %s is a %T, want *connectionGroup", address, i)
38+
}
39+
return connGroup
40+
}
41+
42+
// refCountFor returns the number of scalers sharing the connection for an
43+
// address, and 0 when the pool holds no connection for it.
44+
func refCountFor(t *testing.T, address string) int {
45+
t.Helper()
46+
connGroup := poolEntryFor(t, address)
47+
if connGroup == nil {
48+
return 0
49+
}
50+
return connGroup.refCount
51+
}
52+
53+
// closeOnCleanup releases the scaler's share when the test ends, so a test that
54+
// fails part way through does not leave an entry in the pool for the tests that
55+
// follow. Close is idempotent, so this is a no-op once a test has closed the
56+
// scaler itself.
57+
func closeOnCleanup(t *testing.T, s Scaler) {
58+
t.Helper()
59+
t.Cleanup(func() {
60+
_ = s.Close(context.Background())
61+
})
62+
}
63+
2164
func newTestExternalScaler(t *testing.T, address string) Scaler {
2265
t.Helper()
2366
s, err := NewExternalScaler(&scalersconfig.ScalerConfig{
@@ -29,6 +72,22 @@ func newTestExternalScaler(t *testing.T, address string) Scaler {
2972
if err != nil {
3073
t.Fatalf("NewExternalScaler: %v", err)
3174
}
75+
closeOnCleanup(t, s)
76+
return s
77+
}
78+
79+
func newTestExternalPushScaler(t *testing.T, address string) PushScaler {
80+
t.Helper()
81+
s, err := NewExternalPushScaler(&scalersconfig.ScalerConfig{
82+
ScalableObjectName: "app",
83+
ScalableObjectNamespace: "namespace",
84+
TriggerMetadata: map[string]string{"scalerAddress": address},
85+
ResolvedEnv: map[string]string{},
86+
})
87+
if err != nil {
88+
t.Fatalf("NewExternalPushScaler: %v", err)
89+
}
90+
closeOnCleanup(t, s)
3291
return s
3392
}
3493

@@ -39,33 +98,23 @@ func newTestExternalScaler(t *testing.T, address string) Scaler {
3998
func TestExternalScalerConnectionPoolReleasedOnClose(t *testing.T) {
4099
const address = "pool-release-test.default.svc.cluster.local:9090"
41100

42-
before := poolEntries()
43-
44101
first := newTestExternalScaler(t, address)
45102
second := newTestExternalScaler(t, address)
46103

47-
if got := poolEntries(); got != before+1 {
48-
t.Fatalf("pool entries = %d, want %d, two scalers should share one connection", got, before+1)
49-
}
50-
51-
var connGroup *connectionGroup
52-
connectionPool.Range(func(_, v any) bool {
53-
if cg, ok := v.(*connectionGroup); ok && cg.refCount == 2 {
54-
connGroup = cg
55-
return false
56-
}
57-
return true
58-
})
104+
connGroup := poolEntryFor(t, address)
59105
if connGroup == nil {
60-
t.Fatal("expected a pooled connection with two users")
106+
t.Fatal("no pooled connection for the address the scalers were built with")
107+
}
108+
if got := connGroup.refCount; got != 2 {
109+
t.Fatalf("refCount = %d, want 2, two scalers should share one connection", got)
61110
}
62111

63112
// The first scaler goes away. The connection is still in use.
64113
if err := first.Close(context.Background()); err != nil {
65114
t.Fatalf("closing the first scaler: %v", err)
66115
}
67-
if got := poolEntries(); got != before+1 {
68-
t.Errorf("pool entries = %d, want %d, the connection is still in use", got, before+1)
116+
if got := refCountFor(t, address); got != 1 {
117+
t.Errorf("refCount = %d, want 1, the connection is still in use", got)
69118
}
70119
if got := connGroup.grpcConnection.GetState(); got == connectivity.Shutdown {
71120
t.Error("the connection was closed while another scaler was still using it")
@@ -75,8 +124,53 @@ func TestExternalScalerConnectionPoolReleasedOnClose(t *testing.T) {
75124
if err := second.Close(context.Background()); err != nil {
76125
t.Fatalf("closing the second scaler: %v", err)
77126
}
78-
if got := poolEntries(); got != before {
79-
t.Errorf("pool entries = %d, want %d, the connection was not dropped from the pool", got, before)
127+
if entry := poolEntryFor(t, address); entry != nil {
128+
t.Error("the connection was not dropped from the pool")
129+
}
130+
if got := connGroup.grpcConnection.GetState(); got != connectivity.Shutdown {
131+
t.Errorf("connection state = %v, want %v", got, connectivity.Shutdown)
132+
}
133+
}
134+
135+
// NewExternalPushScaler acquires a share of the same pool, and externalPushScaler
136+
// takes Close from the externalScaler it embeds. A push scaler must therefore
137+
// release its share like any other, including when it shares the connection
138+
// with a plain scaler on the same address.
139+
func TestExternalPushScalerConnectionPoolReleasedOnClose(t *testing.T) {
140+
const address = "pool-push-release.default.svc.cluster.local:9090"
141+
142+
push := newTestExternalPushScaler(t, address)
143+
144+
connGroup := poolEntryFor(t, address)
145+
if connGroup == nil {
146+
t.Fatal("no pooled connection for the address the push scaler was built with")
147+
}
148+
if got := connGroup.refCount; got != 1 {
149+
t.Fatalf("refCount = %d, want 1", got)
150+
}
151+
152+
// A plain scaler on the same address shares the push scaler's connection.
153+
plain := newTestExternalScaler(t, address)
154+
if got := refCountFor(t, address); got != 2 {
155+
t.Fatalf("refCount = %d, want 2, the two scalers should share one connection", got)
156+
}
157+
158+
// Closing the push scaler releases only its own share.
159+
if err := push.Close(context.Background()); err != nil {
160+
t.Fatalf("closing the push scaler: %v", err)
161+
}
162+
if got := refCountFor(t, address); got != 1 {
163+
t.Errorf("refCount = %d, want 1, the plain scaler is still using the connection", got)
164+
}
165+
if got := connGroup.grpcConnection.GetState(); got == connectivity.Shutdown {
166+
t.Error("the connection was closed while the plain scaler was still using it")
167+
}
168+
169+
if err := plain.Close(context.Background()); err != nil {
170+
t.Fatalf("closing the plain scaler: %v", err)
171+
}
172+
if entry := poolEntryFor(t, address); entry != nil {
173+
t.Error("the connection was not dropped from the pool")
80174
}
81175
if got := connGroup.grpcConnection.GetState(); got != connectivity.Shutdown {
82176
t.Errorf("connection state = %v, want %v", got, connectivity.Shutdown)
@@ -88,8 +182,6 @@ func TestExternalScalerConnectionPoolReleasedOnClose(t *testing.T) {
88182
func TestExternalScalerCloseIsIdempotent(t *testing.T) {
89183
const address = "pool-double-close.default.svc.cluster.local:9090"
90184

91-
before := poolEntries()
92-
93185
first := newTestExternalScaler(t, address)
94186
second := newTestExternalScaler(t, address)
95187

@@ -100,21 +192,12 @@ func TestExternalScalerCloseIsIdempotent(t *testing.T) {
100192
}
101193
}
102194

103-
if got := poolEntries(); got != before+1 {
104-
t.Fatalf("pool entries = %d, want %d, a repeated Close released a share it did not hold", got, before+1)
105-
}
106-
107-
// The second scaler still has a working connection.
108-
var connGroup *connectionGroup
109-
connectionPool.Range(func(_, v any) bool {
110-
if cg, ok := v.(*connectionGroup); ok && cg.refCount == 1 {
111-
connGroup = cg
112-
return false
113-
}
114-
return true
115-
})
195+
connGroup := poolEntryFor(t, address)
116196
if connGroup == nil {
117-
t.Fatal("expected the connection to still be held by one scaler")
197+
t.Fatal("a repeated Close released a share it did not hold")
198+
}
199+
if got := connGroup.refCount; got != 1 {
200+
t.Fatalf("refCount = %d, want 1, a repeated Close released a share it did not hold", got)
118201
}
119202
if got := connGroup.grpcConnection.GetState(); got == connectivity.Shutdown {
120203
t.Error("the connection was closed while a scaler was still using it")
@@ -123,8 +206,8 @@ func TestExternalScalerCloseIsIdempotent(t *testing.T) {
123206
if err := second.Close(context.Background()); err != nil {
124207
t.Fatalf("closing the second scaler: %v", err)
125208
}
126-
if got := poolEntries(); got != before {
127-
t.Errorf("pool entries = %d, want %d", got, before)
209+
if entry := poolEntryFor(t, address); entry != nil {
210+
t.Error("the connection was not dropped from the pool")
128211
}
129212
}
130213

@@ -136,8 +219,6 @@ func TestExternalScalerCloseIsIdempotent(t *testing.T) {
136219
func TestGetClientForConnectionPoolDoesNotRecreateAfterClose(t *testing.T) {
137220
const address = "pool-no-recreate.default.svc.cluster.local:9090"
138221

139-
before := poolEntries()
140-
141222
s := newTestExternalScaler(t, address)
142223
md := externalScalerMetadata{ScalerAddress: address}
143224

@@ -152,34 +233,80 @@ func TestGetClientForConnectionPoolDoesNotRecreateAfterClose(t *testing.T) {
152233
if _, err := getClientForConnectionPool(md); err == nil {
153234
t.Error("expected an error looking up the connection after the scaler closed")
154235
}
155-
if got := poolEntries(); got != before {
156-
t.Errorf("pool entries = %d, want %d, the lookup recreated an entry nothing owns", got, before)
236+
if entry := poolEntryFor(t, address); entry != nil {
237+
t.Error("the lookup recreated an entry nothing owns")
157238
}
158239
}
159240

160241
// Scalers pointing at different addresses do not share a connection, and
161242
// releasing one leaves the other alone.
162243
func TestExternalScalerConnectionPoolPerAddress(t *testing.T) {
244+
const (
245+
addressA = "pool-a.default.svc.cluster.local:9090"
246+
addressB = "pool-b.default.svc.cluster.local:9090"
247+
)
248+
163249
before := poolEntries()
164250

165-
one := newTestExternalScaler(t, "pool-a.default.svc.cluster.local:9090")
166-
two := newTestExternalScaler(t, "pool-b.default.svc.cluster.local:9090")
251+
one := newTestExternalScaler(t, addressA)
252+
two := newTestExternalScaler(t, addressB)
167253

168254
if got := poolEntries(); got != before+2 {
169-
t.Fatalf("pool entries = %d, want %d", got, before+2)
255+
t.Fatalf("pool entries = %d, want %d, the two addresses should not share a connection", got, before+2)
170256
}
171257

172258
if err := one.Close(context.Background()); err != nil {
173259
t.Fatalf("closing the first scaler: %v", err)
174260
}
175-
if got := poolEntries(); got != before+1 {
176-
t.Errorf("pool entries = %d, want %d", got, before+1)
261+
if entry := poolEntryFor(t, addressA); entry != nil {
262+
t.Error("the first connection was not dropped from the pool")
263+
}
264+
if got := refCountFor(t, addressB); got != 1 {
265+
t.Errorf("refCount for the second address = %d, want 1, releasing one address must leave the other alone", got)
177266
}
178267

179268
if err := two.Close(context.Background()); err != nil {
180269
t.Fatalf("closing the second scaler: %v", err)
181270
}
271+
if entry := poolEntryFor(t, addressB); entry != nil {
272+
t.Error("the second connection was not dropped from the pool")
273+
}
182274
if got := poolEntries(); got != before {
183275
t.Errorf("pool entries = %d, want %d", got, before)
184276
}
185277
}
278+
279+
// The pooled connection is now built from the scaler constructor, so a client
280+
// certificate that cannot be parsed is reported there rather than on the first
281+
// poll. buildScalers turns any constructor error into a failure for the whole
282+
// ScaledObject, so this moves a malformed certificate from degrading one
283+
// trigger at poll time to failing every trigger on the object at build time.
284+
// This test pins that behaviour so a change to it is deliberate. Whichever way
285+
// the error is reported, the failed construction must leave nothing in the
286+
// pool.
287+
func TestExternalScalerReportsTLSErrorFromTheConstructor(t *testing.T) {
288+
const address = "pool-bad-tls.default.svc.cluster.local:9090"
289+
290+
before := poolEntries()
291+
292+
s, err := NewExternalScaler(&scalersconfig.ScalerConfig{
293+
ScalableObjectName: "app",
294+
ScalableObjectNamespace: "namespace",
295+
TriggerMetadata: map[string]string{"scalerAddress": address, "enableTLS": "true"},
296+
AuthParams: map[string]string{
297+
"tlsClientCert": "not a certificate",
298+
"tlsClientKey": "not a key",
299+
},
300+
ResolvedEnv: map[string]string{},
301+
})
302+
if err == nil {
303+
closeOnCleanup(t, s)
304+
t.Fatal("expected an error for an unparseable client certificate")
305+
}
306+
307+
// The pool key includes the TLS fields, so the entry a failed construction
308+
// could leave behind is found by counting rather than by looking it up.
309+
if got := poolEntries(); got != before {
310+
t.Errorf("pool entries = %d, want %d, the failed construction left a connection in the pool", got, before)
311+
}
312+
}

‎pkg/scalers/external_scaler_tcp_test.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,7 @@ func TestExternalScalerReleasesTheTCPConnection(t *testing.T) {
7777
if err != nil {
7878
t.Fatalf("NewExternalScaler: %v", err)
7979
}
80+
closeOnCleanup(t, s)
8081
// grpc.NewClient dials lazily, so issue one call to bring the transport up.
8182
// The metric name carries the trigger index prefix the scaler strips before
8283
// the request, without which it returns early and never reaches the wire.

0 commit comments

Comments
 (0)