Skip to content

Commit 968f43d

Browse files
sean-sean-
committed
Implement connection pool health probes and fix discovery tests
Add health validation during node discovery using GET / endpoint with 5-second timeout. Nodes are health-checked before being added to the connection pool, improving cluster reliability and connection quality. Key changes: - Add isHealthyOpenSearchNode() function with OpenSearch 1.3.0+ compatibility - Integrate health checks into DiscoverNodes() workflow - Enhance resurrection timing with cluster-aware algorithm - Add startup URL fallback with round-robin selection for discovery - Fix TestDiscovery integration tests with dynamic port allocation - Add mock health check support to test transport infrastructure All discovery integration tests now pass with proper health validation. Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>
1 parent 7c85f67 commit 968f43d

11 files changed

Lines changed: 831 additions & 86 deletions

‎.ci/opensearch/Dockerfile.opensearch‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,30 @@ ARG SECURE_INTEGRATION
66
ENV SECURE_INTEGRATION=$SECURE_INTEGRATION
77
ARG OPENSEARCH_INITIAL_ADMIN_PASSWORD
88

9-
# Some opensearch secuirty settings are only present since 2.8.0 and causes older versions to brake if the setting is present
10-
# https://apple.stackexchange.com/a/123408/11374
9+
# Handle plugin dependencies when removing opensearch-security
10+
# OpenSearch 3.x introduced plugin dependencies that prevent direct removal of opensearch-security.
11+
# Plugin dependency evolution:
12+
# 3.0.0: No dependencies - remove opensearch-security directly
13+
# 3.1.0-3.2.0: opensearch-anomaly-detection extends opensearch-security
14+
# 3.3.2: opensearch-skills extends opensearch-ml, which extends opensearch-security
15+
# 3.4.0+: opensearch-flow-framework also extends opensearch-security
16+
# We must remove plugins in dependency order (dependents first, then their dependencies).
17+
ARG OPENSEARCH_VERSION
1118
RUN if [ "$SECURE_INTEGRATION" != "true" ] ; then \
12-
$opensearch_path/bin/opensearch-plugin remove opensearch-security; \
19+
function version { echo "$@" | awk -F. '{ printf("%d%03d%03d%03d\n", $1,$2,$3,$4); }'; }; \
20+
if [ $(version ${OPENSEARCH_VERSION:-0.0.0}) -ge $(version "3.4.0") ] || [ "$OPENSEARCH_VERSION" = "latest" ]; then \
21+
$opensearch_path/bin/opensearch-plugin remove opensearch-flow-framework; \
22+
$opensearch_path/bin/opensearch-plugin remove opensearch-skills; \
23+
$opensearch_path/bin/opensearch-plugin remove opensearch-ml; \
24+
$opensearch_path/bin/opensearch-plugin remove opensearch-anomaly-detection; \
25+
elif [ $(version ${OPENSEARCH_VERSION:-0.0.0}) -ge $(version "3.3.2") ]; then \
26+
$opensearch_path/bin/opensearch-plugin remove opensearch-skills; \
27+
$opensearch_path/bin/opensearch-plugin remove opensearch-ml; \
28+
$opensearch_path/bin/opensearch-plugin remove opensearch-anomaly-detection; \
29+
elif [ $(version ${OPENSEARCH_VERSION:-0.0.0}) -ge $(version "3.1.0") ]; then \
30+
$opensearch_path/bin/opensearch-plugin remove opensearch-anomaly-detection; \
31+
fi; \
32+
$opensearch_path/bin/opensearch-plugin remove opensearch-security; \
1333
else \
1434
$opensearch_path/opensearch-onetime-setup.sh; \
1535
echo "plugins.security.nodes_dn_dynamic_config_enabled: true" | tee -a $opensearch_path/config/opensearch.yml > /dev/null; \

‎.ci/opensearch/docker-compose.yml‎

Lines changed: 120 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
services:
2-
opensearch:
2+
opensearch-node1:
33
deploy:
44
restart_policy:
55
condition: any
@@ -11,11 +11,128 @@ services:
1111
- OPENSEARCH_VERSION=${OPENSEARCH_VERSION:-latest}
1212
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
1313
environment:
14-
- discovery.type=single-node
15-
- bootstrap.memory_lock=true
14+
- cluster.name=opensearch-cluster
15+
- node.name=opensearch-node1
16+
- node.roles=cluster_manager,data,ingest
17+
- discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
18+
- cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
19+
- bootstrap.memory_lock=false # Disable memory locking for development
1620
- path.repo=/usr/share/opensearch/mnt
1721
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
1822
- plugins.index_state_management.job_interval=1
23+
# Network settings for proper node discovery
24+
- network.host=0.0.0.0
25+
- transport.host=0.0.0.0
26+
- http.host=0.0.0.0
27+
# Publish HTTP address for external clients only
28+
- http.publish_host=localhost
29+
- http.publish_port=9200
30+
# Memory settings
31+
- OPENSEARCH_JAVA_OPTS=-Xms1g -Xmx1g
1932
ports:
2033
- "9200:9200"
34+
- "9300:9300"
35+
networks:
36+
- opensearch-net
37+
volumes:
38+
- opensearch-data1:/usr/share/opensearch/data
2139
user: opensearch
40+
ulimits:
41+
memlock:
42+
soft: -1
43+
hard: -1
44+
45+
opensearch-node2:
46+
deploy:
47+
restart_policy:
48+
condition: any
49+
build:
50+
context: .
51+
dockerfile: Dockerfile.opensearch
52+
args:
53+
- SECURE_INTEGRATION=${SECURE_INTEGRATION:-false}
54+
- OPENSEARCH_VERSION=${OPENSEARCH_VERSION:-latest}
55+
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
56+
environment:
57+
- cluster.name=opensearch-cluster
58+
- node.name=opensearch-node2
59+
- node.roles=cluster_manager,data,ingest
60+
- discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
61+
- cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
62+
- bootstrap.memory_lock=false # Disable memory locking for development
63+
- path.repo=/usr/share/opensearch/mnt
64+
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
65+
- plugins.index_state_management.job_interval=1
66+
# Network settings for proper node discovery
67+
- network.host=0.0.0.0
68+
- transport.host=0.0.0.0
69+
- http.host=0.0.0.0
70+
# Publish HTTP address for external clients only
71+
- http.publish_host=localhost
72+
- http.publish_port=9201
73+
# Memory settings
74+
- OPENSEARCH_JAVA_OPTS=-Xms1g -Xmx1g
75+
ports:
76+
- "9201:9200"
77+
- "9301:9300"
78+
networks:
79+
- opensearch-net
80+
volumes:
81+
- opensearch-data2:/usr/share/opensearch/data
82+
user: opensearch
83+
ulimits:
84+
memlock:
85+
soft: -1
86+
hard: -1
87+
88+
opensearch-node3:
89+
deploy:
90+
restart_policy:
91+
condition: any
92+
build:
93+
context: .
94+
dockerfile: Dockerfile.opensearch
95+
args:
96+
- SECURE_INTEGRATION=${SECURE_INTEGRATION:-false}
97+
- OPENSEARCH_VERSION=${OPENSEARCH_VERSION:-latest}
98+
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
99+
environment:
100+
- cluster.name=opensearch-cluster
101+
- node.name=opensearch-node3
102+
- node.roles=cluster_manager,data,ingest
103+
- discovery.seed_hosts=opensearch-node1,opensearch-node2,opensearch-node3
104+
- cluster.initial_cluster_manager_nodes=opensearch-node1,opensearch-node2,opensearch-node3
105+
- bootstrap.memory_lock=false # Disable memory locking for development
106+
- path.repo=/usr/share/opensearch/mnt
107+
- OPENSEARCH_INITIAL_ADMIN_PASSWORD=myStrongPassword123!
108+
- plugins.index_state_management.job_interval=1
109+
# Network settings for proper node discovery
110+
- network.host=0.0.0.0
111+
- transport.host=0.0.0.0
112+
- http.host=0.0.0.0
113+
# Publish HTTP address for external clients only
114+
- http.publish_host=localhost
115+
- http.publish_port=9202
116+
# Memory settings
117+
- OPENSEARCH_JAVA_OPTS=-Xms1g -Xmx1g
118+
ports:
119+
- "9202:9200"
120+
- "9302:9300"
121+
networks:
122+
- opensearch-net
123+
volumes:
124+
- opensearch-data3:/usr/share/opensearch/data
125+
user: opensearch
126+
ulimits:
127+
memlock:
128+
soft: -1
129+
hard: -1
130+
131+
networks:
132+
opensearch-net:
133+
driver: bridge
134+
135+
volumes:
136+
opensearch-data1:
137+
opensearch-data2:
138+
opensearch-data3:

‎Makefile‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,15 @@ cluster.start:
213213
cluster.stop:
214214
docker compose --project-directory .ci/opensearch down;
215215

216+
cluster.scale.1: ## Start single-node cluster
217+
docker compose --project-directory .ci/opensearch up -d --scale opensearch-node2=0 --scale opensearch-node3=0;
218+
219+
cluster.scale.2: ## Start 2-node cluster
220+
docker compose --project-directory .ci/opensearch up -d --scale opensearch-node1=1 --scale opensearch-node2=1 --scale opensearch-node3=0;
221+
222+
cluster.scale.3: ## Start full 3-node cluster
223+
docker compose --project-directory .ci/opensearch up -d --scale opensearch-node1=1 --scale opensearch-node2=1 --scale opensearch-node3=1;
224+
216225
cluster.get-cert:
217226
@if [[ -v SECURE_INTEGRATION ]] && [[ $$SECURE_INTEGRATION == "true" ]]; then \
218227
docker cp $$(docker compose --project-directory .ci/opensearch ps --format '{{.Name}}'):/usr/share/opensearch/config/kirk.pem admin.pem && \

‎opensearchtransport/connection.go‎

Lines changed: 94 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ import (
3030
"errors"
3131
"fmt"
3232
"math"
33+
"math/rand/v2"
3334
"net/url"
3435
"sort"
3536
"sync"
@@ -39,6 +40,8 @@ import (
3940
const (
4041
defaultResurrectTimeoutInitial = 60 * time.Second
4142
defaultResurrectTimeoutFactorCutoff = 5
43+
defaultMinimumResurrectTimeout = 10 * time.Millisecond
44+
defaultJitterScale = 0.1
4245
)
4346

4447
// Selector defines the interface for selecting connections from the pool.
@@ -83,6 +86,11 @@ type statusConnectionPool struct {
8386
selector Selector
8487
resurrectTimeoutInitial time.Duration
8588
resurrectTimeoutFactorCutoff int
89+
minimumResurrectTimeout time.Duration
90+
jitterScale float64
91+
92+
// Health check function - returns version string on success, error on failure
93+
healthCheck func(*url.URL) (string, error)
8694

8795
metrics *metrics
8896
}
@@ -108,6 +116,8 @@ func NewConnectionPool(conns []*Connection, selector Selector) ConnectionPool {
108116
selector: selector,
109117
resurrectTimeoutInitial: defaultResurrectTimeoutInitial,
110118
resurrectTimeoutFactorCutoff: defaultResurrectTimeoutFactorCutoff,
119+
minimumResurrectTimeout: defaultMinimumResurrectTimeout,
120+
jitterScale: defaultJitterScale,
111121
}
112122
}
113123

@@ -162,7 +172,24 @@ func (cp *statusConnectionPool) OnSuccess(c *Connection) {
162172

163173
cp.Lock()
164174
defer cp.Unlock()
165-
cp.resurrect(c, true)
175+
176+
// Connection just successfully handled a request - no health check needed
177+
if debugLogger != nil {
178+
debugLogger.Logf("Resurrecting %q (successful request)\n", c.URL)
179+
}
180+
181+
c.markAsLive()
182+
cp.live = append(cp.live, c)
183+
184+
// Remove from dead list
185+
for i, conn := range cp.dead {
186+
if conn == c {
187+
// Remove item; https://github.com/golang/go/wiki/SliceTricks
188+
copy(cp.dead[i:], cp.dead[i+1:])
189+
cp.dead = cp.dead[:len(cp.dead)-1]
190+
break
191+
}
192+
}
166193
}
167194

168195
// OnFailure marks the connection as failed.
@@ -246,12 +273,32 @@ func (cp *statusConnectionPool) connections() []*Connection {
246273
return conns
247274
}
248275

249-
// resurrect adds the connection to the list of available connections.
276+
// resurrect adds the connection to the list of available connections after health validation.
250277
// When removeDead is true, it also removes it from the dead list.
251278
// The calling code is responsible for locking.
252279
func (cp *statusConnectionPool) resurrect(c *Connection, removeDead bool) {
253280
if debugLogger != nil {
254-
debugLogger.Logf("Resurrecting %s\n", c.URL)
281+
debugLogger.Logf("Attempting to resurrect %q\n", c.URL)
282+
}
283+
284+
// Perform health check if available
285+
if cp.healthCheck != nil {
286+
version, err := cp.healthCheck(c.URL)
287+
if err != nil {
288+
if debugLogger != nil {
289+
debugLogger.Logf("Health check failed for %q: %s; will retry later\n", c.URL, err)
290+
}
291+
// Health check failed - schedule another resurrection attempt
292+
cp.scheduleResurrect(c)
293+
return
294+
}
295+
if debugLogger != nil {
296+
debugLogger.Logf("Health check passed for %q (version: %q)\n", c.URL, version)
297+
}
298+
}
299+
300+
if debugLogger != nil {
301+
debugLogger.Logf("Resurrecting %q\n", c.URL)
255302
}
256303

257304
c.markAsLive()
@@ -274,23 +321,60 @@ func (cp *statusConnectionPool) resurrect(c *Connection, removeDead bool) {
274321
}
275322
}
276323

277-
// scheduleResurrect schedules the connection to be resurrected.
324+
// scheduleResurrect schedules the connection to be resurrected using cluster-aware timing.
325+
// Formula: ((1 - ((total - live) / total)) * total) * jitterScale
326+
// - All dead: immediate resurrection
327+
// - Healthy clusters: longer waits with more jitter
328+
// - Incident scenarios: faster recovery
278329
func (cp *statusConnectionPool) scheduleResurrect(c *Connection) {
330+
// Calculate basic exponential backoff factor
279331
factor := math.Min(float64(c.Failures-1), float64(cp.resurrectTimeoutFactorCutoff))
280-
timeout := time.Duration(cp.resurrectTimeoutInitial.Seconds() * math.Exp2(factor) * float64(time.Second))
332+
baseTimeout := time.Duration(cp.resurrectTimeoutInitial.Seconds() * math.Exp2(factor) * float64(time.Second))
333+
334+
// Get cluster health metrics
335+
totalNodes := len(cp.live) + len(cp.dead)
336+
liveNodes := len(cp.live)
337+
338+
var finalTimeout time.Duration
339+
340+
if totalNodes == 0 || liveNodes == 0 {
341+
// All dead or no nodes: immediate resurrection
342+
finalTimeout = cp.minimumResurrectTimeout
343+
} else {
344+
// Cluster-aware formula: ((1 - ((total - live) / total)) * total) * jitterScale
345+
deadNodes := totalNodes - liveNodes
346+
healthRatio := 1.0 - (float64(deadNodes) / float64(totalNodes))
347+
clusterFactor := healthRatio * float64(totalNodes) * cp.jitterScale
348+
349+
// Apply base timeout and cluster factor
350+
clusterTimeout := time.Duration(float64(baseTimeout) * clusterFactor)
351+
352+
// Add random jitter (0 to clusterTimeout range)
353+
jitter := time.Duration(rand.Float64() * float64(clusterTimeout))
354+
finalTimeout = jitter
355+
356+
// Ensure minimum timeout
357+
if finalTimeout < cp.minimumResurrectTimeout {
358+
finalTimeout = cp.minimumResurrectTimeout
359+
}
360+
}
281361

282362
if debugLogger != nil {
283363
debugLogger.Logf(
284-
"Resurrect %s (failures=%d, factor=%1.1f, timeout=%s) in %s\n",
364+
"Resurrect %q (failures=%d, factor=%1.1f, live=%d, dead=%d, total=%d, base=%s, final=%s) in %s\n",
285365
c.URL,
286366
c.Failures,
287367
factor,
288-
timeout,
289-
c.DeadSince.Add(timeout).Sub(time.Now().UTC()).Truncate(time.Second),
368+
liveNodes,
369+
len(cp.dead),
370+
totalNodes,
371+
baseTimeout,
372+
finalTimeout,
373+
c.DeadSince.Add(finalTimeout).Sub(time.Now().UTC()).Truncate(time.Millisecond),
290374
)
291375
}
292376

293-
time.AfterFunc(timeout, func() {
377+
time.AfterFunc(finalTimeout, func() {
294378
cp.Lock()
295379
defer cp.Unlock()
296380

@@ -299,7 +383,7 @@ func (cp *statusConnectionPool) scheduleResurrect(c *Connection) {
299383

300384
if !c.IsDead {
301385
if debugLogger != nil {
302-
debugLogger.Logf("Already resurrected %s\n", c.URL)
386+
debugLogger.Logf("Already resurrected %q\n", c.URL)
303387
}
304388
return
305389
}

0 commit comments

Comments
 (0)