Skip to content

Commit 3e1e620

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 3deff37 commit 3e1e620

11 files changed

Lines changed: 844 additions & 88 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: 105 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"
@@ -40,6 +41,8 @@ import (
4041
const (
4142
defaultResurrectTimeoutInitial = 60 * time.Second
4243
defaultResurrectTimeoutFactorCutoff = 5
44+
defaultMinimumResurrectTimeout = 10 * time.Millisecond
45+
defaultJitterScale = 0.1
4346
)
4447

4548
// Selector defines the interface for selecting connections from the pool.
@@ -96,6 +99,11 @@ type statusConnectionPool struct {
9699
selector Selector
97100
resurrectTimeoutInitial time.Duration
98101
resurrectTimeoutFactorCutoff int
102+
minimumResurrectTimeout time.Duration
103+
jitterScale float64
104+
105+
// Health check function - returns version string on success, error on failure
106+
healthCheck func(*url.URL) (string, error)
99107

100108
metrics *metrics
101109
}
@@ -128,6 +136,8 @@ func NewConnectionPool(conns []*Connection, selector Selector) ConnectionPool {
128136
selector: selector,
129137
resurrectTimeoutInitial: defaultResurrectTimeoutInitial,
130138
resurrectTimeoutFactorCutoff: defaultResurrectTimeoutFactorCutoff,
139+
minimumResurrectTimeout: defaultMinimumResurrectTimeout,
140+
jitterScale: defaultJitterScale,
131141
}
132142
pool.mu.live = conns
133143
pool.mu.dead = []*Connection{}
@@ -306,7 +316,7 @@ func (cp *statusConnectionPool) Unlock() {
306316
cp.mu.Unlock()
307317
}
308318

309-
// resurrect adds the connection to the list of available connections.
319+
// resurrect adds the connection to the list of available connections after health validation.
310320
// When removeDead is true, it also removes it from the dead list.
311321
//
312322
// CALLER RESPONSIBILITIES:
@@ -315,7 +325,27 @@ func (cp *statusConnectionPool) Unlock() {
315325
// - Caller must handle any errors from subsequent connection attempts
316326
func (cp *statusConnectionPool) resurrectWithLock(c *Connection, removeDead bool) {
317327
if debugLogger != nil {
318-
debugLogger.Logf("Resurrecting %s\n", c.URL)
328+
debugLogger.Logf("Attempting to resurrect %q\n", c.URL)
329+
}
330+
331+
// Perform health check if available
332+
if cp.healthCheck != nil {
333+
version, err := cp.healthCheck(c.URL)
334+
if err != nil {
335+
if debugLogger != nil {
336+
debugLogger.Logf("Health check failed for %q: %s; will retry later\n", c.URL, err)
337+
}
338+
// Health check failed - schedule another resurrection attempt
339+
cp.scheduleResurrect(c)
340+
return
341+
}
342+
if debugLogger != nil {
343+
debugLogger.Logf("Health check passed for %q (version: %q)\n", c.URL, version)
344+
}
345+
}
346+
347+
if debugLogger != nil {
348+
debugLogger.Logf("Resurrecting %q\n", c.URL)
319349
}
320350

321351
c.markAsLiveWithLock()
@@ -338,24 +368,61 @@ func (cp *statusConnectionPool) resurrectWithLock(c *Connection, removeDead bool
338368
}
339369
}
340370

341-
// scheduleResurrect schedules the connection to be resurrected.
371+
// scheduleResurrect schedules the connection to be resurrected using cluster-aware timing.
372+
// Formula: ((1 - ((total - live) / total)) * total) * jitterScale
373+
// - All dead: immediate resurrection
374+
// - Healthy clusters: longer waits with more jitter
375+
// - Incident scenarios: faster recovery
342376
func (cp *statusConnectionPool) scheduleResurrect(c *Connection, deadSince time.Time) {
377+
// Calculate basic exponential backoff factor
343378
failures := c.failures.Load()
344-
factor := min(failures-1, int64(cp.resurrectTimeoutFactorCutoff))
345-
timeout := time.Duration(cp.resurrectTimeoutInitial.Seconds() * math.Exp2(float64(factor)) * float64(time.Second))
379+
factor := math.Min(float64(failures-1), float64(cp.resurrectTimeoutFactorCutoff))
380+
baseTimeout := time.Duration(cp.resurrectTimeoutInitial.Seconds() * math.Exp2(factor) * float64(time.Second))
381+
382+
// Get cluster health metrics
383+
totalNodes := len(cp.mu.live) + len(cp.mu.dead)
384+
liveNodes := len(cp.mu.live)
385+
386+
var finalTimeout time.Duration
387+
388+
if totalNodes == 0 || liveNodes == 0 {
389+
// All dead or no nodes: immediate resurrection
390+
finalTimeout = cp.minimumResurrectTimeout
391+
} else {
392+
// Cluster-aware formula: ((1 - ((total - live) / total)) * total) * jitterScale
393+
deadNodes := totalNodes - liveNodes
394+
healthRatio := 1.0 - (float64(deadNodes) / float64(totalNodes))
395+
clusterFactor := healthRatio * float64(totalNodes) * cp.jitterScale
396+
397+
// Apply base timeout and cluster factor
398+
clusterTimeout := time.Duration(float64(baseTimeout) * clusterFactor)
399+
400+
// Add random jitter (0 to clusterTimeout range)
401+
jitter := time.Duration(rand.Float64() * float64(clusterTimeout))
402+
finalTimeout = jitter
403+
404+
// Ensure minimum timeout
405+
if finalTimeout < cp.minimumResurrectTimeout {
406+
finalTimeout = cp.minimumResurrectTimeout
407+
}
408+
}
346409

347410
if debugLogger != nil {
348411
debugLogger.Logf(
349-
"Resurrect %s (failures=%d, factor=%d, timeout=%s) in %s\n",
412+
"Resurrect %q (failures=%d, factor=%1.1f, live=%d, dead=%d, total=%d, base=%s, final=%s) in %s\n",
350413
c.URL,
351414
failures,
352415
factor,
353-
timeout,
354-
deadSince.Add(timeout).Sub(time.Now().UTC()).Truncate(time.Second),
416+
liveNodes,
417+
len(cp.mu.dead),
418+
totalNodes,
419+
baseTimeout,
420+
finalTimeout,
421+
deadSince.Add(finalTimeout).Sub(time.Now().UTC()).Truncate(time.Millisecond),
355422
)
356423
}
357424

358-
time.AfterFunc(timeout, func() {
425+
time.AfterFunc(finalTimeout, func() {
359426
cp.mu.Lock()
360427
defer cp.mu.Unlock()
361428

@@ -364,7 +431,35 @@ func (cp *statusConnectionPool) scheduleResurrect(c *Connection, deadSince time.
364431

365432
if !c.mu.isDead {
366433
if debugLogger != nil {
367-
debugLogger.Logf("Already resurrected %s\n", c.URL)
434+
debugLogger.Logf("Already resurrected %q\n", c.URL)
435+
}
436+
return
437+
}
438+
439+
cp.resurrectWithLock(c, true)
440+
})
441+
c.URL,
442+
failures,
443+
factor,
444+
liveNodes,
445+
len(cp.dead),
446+
totalNodes,
447+
baseTimeout,
448+
finalTimeout,
449+
c.DeadSince.Add(finalTimeout).Sub(time.Now().UTC()).Truncate(time.Millisecond),
450+
)
451+
}
452+
453+
time.AfterFunc(finalTimeout, func() {
454+
cp.Lock()
455+
defer cp.Unlock()
456+
457+
c.mu.Lock()
458+
defer c.mu.Unlock()
459+
460+
if !c.mu.isDead {
461+
if debugLogger != nil {
462+
debugLogger.Logf("Already resurrected %q\n", c.URL)
368463
}
369464
return
370465
}

0 commit comments

Comments
 (0)