Skip to content

feat(chpool): add healthcheck function #1088

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nmiculinic wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat(chpool): add healthcheck function #1088

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
10694d0
feat(chpool): add healthcheck function
Aug 28, 2025
a0b0d5c
Merge branch 'main' into add_pool_healhchecking
kavirajk Jan 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 10 additions & 9 deletions chpool/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package chpool

import (
"context"
"time"

"github.com/jackc/puddle/v2"

Expand All @@ -21,14 +20,16 @@ func (c *Client) Release() {
return
}

client := c.client()

if client.IsClosed() || time.Since(c.res.CreationTime()) > c.p.options.MaxConnLifetime {
c.res.Destroy()
return
}

c.res.Release()
// calling async since connIsHealthy may block

@kavirajk kavirajk Jan 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm failing to understand why move this to non-async. From the user POV, when I call c.Release() I would expect to release the connection to the pool immediately (not eventually).

go func() {
if c.p.connIsHealthy(c.res) {

@kavirajk kavirajk Jan 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm also curious why is just client.IsClosed() is not good enough here? why introduce connIsHealthy based on ping() that connection?

To me it looks like your use case is not properly handled with just simple check client.IsClosed(). May I know why?

My understanding is, if the health check is passing on the given connection, it should return correct status on client.IsClosed() no? why introduce another layer of additional checks?

c.p.options.ClientOptions.Logger.Debug("chpool: releasing connection")
c.res.Release()
} else {
c.p.options.ClientOptions.Logger.Debug("chpool: destoying connection")
c.res.Destroy()
}
}()
}

func (c *Client) Do(ctx context.Context, q ch.Query) (err error) {
Expand Down
22 changes: 22 additions & 0 deletions chpool/client_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,13 @@ package chpool

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

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"

"github.com/ClickHouse/ch-go"
)

func TestClient_Do(t *testing.T) {
Expand All @@ -17,6 +21,24 @@ func TestClient_Do(t *testing.T) {
testDo(t, conn)
}

func TestClient_ReleaseHealthCheck(t *testing.T) {
t.Parallel()
var healthCheckCnt int64
p := PoolConnOpt(t, Options{
HealthCheckFunc: func(ctx context.Context, client *ch.Client) error {
atomic.AddInt64(&healthCheckCnt, 1)
return nil
},
})
conn, err := p.Acquire(context.Background())
require.NoError(t, err)
assert.Equal(t, int64(0), atomic.LoadInt64(&healthCheckCnt))

conn.Release()
waitForReleaseToComplete()

@kavirajk kavirajk Jan 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This would just make the test more fragile IMHO. Again I understand this is the result of making the release async. See my above comment.

assert.Equal(t, int64(1), atomic.LoadInt64(&healthCheckCnt))
}

func TestClient_Ping(t *testing.T) {
t.Parallel()
p := PoolConn(t)
Expand Down
7 changes: 5 additions & 2 deletions chpool/conn.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,14 +1,17 @@
package chpool

import (
"time"

"github.com/jackc/puddle/v2"

"github.com/ClickHouse/ch-go"
)

type connResource struct {
client *ch.Client
clients []Client
lastHealthCheckTimestamp time.Time
client *ch.Client
clients []Client
}

func (cr *connResource) getConn(p *Pool, res *puddle.Resource[*connResource]) *Client {
Expand Down
84 changes: 66 additions & 18 deletions chpool/pool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@ import (
"sync"
"time"

"github.com/ClickHouse/ch-go"

"github.com/jackc/puddle/v2"
"go.uber.org/zap"

"github.com/ClickHouse/ch-go"
)

// Pool of connections to ClickHouse.
Expand All @@ -23,19 +24,26 @@ type Pool struct {

// Options for Pool.
type Options struct {
ClientOptions ch.Options
MaxConnLifetime time.Duration
MaxConnIdleTime time.Duration
MaxConns int32
MinConns int32
HealthCheckPeriod time.Duration
ClientOptions ch.Options
MaxConnLifetime time.Duration
MaxConnIdleTime time.Duration
MaxConns int32
MinConns int32
HealthCheckPeriod time.Duration
HealthCheckFunc func(ctx context.Context, client *ch.Client) error
HealthCheckTimeout time.Duration
}

func DefaultHealthCheckFunc(ctx context.Context, client *ch.Client) error {
return client.Ping(ctx)
}

// Defaults for pool.
const (
DefaultMaxConnLifetime = time.Hour
DefaultMaxConnIdleTime = time.Minute * 30
DefaultHealthCheckPeriod = time.Minute
DefaultMaxConnLifetime = time.Hour
DefaultMaxConnIdleTime = time.Minute * 30
DefaultHealthCheckPeriod = time.Minute
DefaultHealthCheckTimeout = time.Second
)

func (o *Options) setDefaults() {
Expand All @@ -51,6 +59,15 @@ func (o *Options) setDefaults() {
if o.HealthCheckPeriod == 0 {
o.HealthCheckPeriod = DefaultHealthCheckPeriod
}
if o.HealthCheckFunc == nil {
o.HealthCheckFunc = DefaultHealthCheckFunc
}
if o.HealthCheckTimeout == 0 {
o.HealthCheckTimeout = DefaultHealthCheckTimeout
}
if o.ClientOptions.Logger == nil {
o.ClientOptions.Logger = zap.NewNop()

@nmiculinic nmiculinic Aug 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To prevent any nil-ptr exception later

}
}

// Dial returns a pool of connections to ClickHouse.
Expand Down Expand Up @@ -162,16 +179,47 @@ func (p *Pool) backgroundHealthCheck() {
func (p *Pool) checkIdleConnsHealth() {
resources := p.pool.AcquireAllIdle()

now := time.Now()
wg := sync.WaitGroup{}
for _, res := range resources {
if now.Sub(res.CreationTime()) > p.options.MaxConnLifetime {
res.Destroy()
} else if res.IdleDuration() > p.options.MaxConnIdleTime {
res.Destroy()
} else {
res.ReleaseUnused()
res := res
wg.Add(1)
go func() {
defer wg.Done()
if res.IdleDuration() > p.options.MaxConnIdleTime || !p.connIsHealthy(res) {
res.Destroy()
} else {
res.ReleaseUnused()
}
}()
wg.Wait()

@kavirajk kavirajk Jan 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also I don't think calling wg.Wait() here inside the loop will do what you expected to do. Usually you start all the goroutines (inside the for loop here). Then call wg.Wait() once all the go-routines are spawned (basically outside the loop here)

Calling inside the loop just forces each go-routine to finish before starting the another (sequential not concurrent).

May be I'm missing something here?

}
}

func (p *Pool) connIsHealthy(res *puddle.Resource[*connResource]) bool {

@nmiculinic nmiculinic Aug 28, 2025

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Happy to change this name to something better

logger := p.options.ClientOptions.Logger
if res.Value().client.IsClosed() {
logger.Debug("chpool: connection is closed")
return false
}

if time.Since(res.CreationTime()) > p.options.MaxConnLifetime {
logger.Debug("chpool: connection over max lifetime")
return false
}

if p.options.HealthCheckFunc != nil {
ctx, cancel := context.WithTimeout(context.Background(), p.options.HealthCheckTimeout)
defer cancel()
if err := p.options.HealthCheckFunc(ctx, res.Value().client); err != nil {
if logger := p.options.ClientOptions.Logger; logger != nil {
logger.Warn("chpool: health check failed", zap.Error(err))
}
return false
}
}

res.Value().lastHealthCheckTimestamp = time.Now()
return true
}

func (p *Pool) checkMinConns() {
Expand Down
24 changes: 24 additions & 0 deletions chpool/pool_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,12 @@ package chpool

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

"github.com/ClickHouse/ch-go"

"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
Expand Down Expand Up @@ -61,6 +64,7 @@ func TestPool_Ping(t *testing.T) {
p := PoolConn(t)

require.NoError(t, p.Ping(context.Background()))
waitForReleaseToComplete()

stats := p.Stat()
assert.EqualValues(t, 0, stats.AcquiredResources())
Expand All @@ -78,3 +82,23 @@ func TestPool_Acquire(t *testing.T) {
waitForReleaseToComplete()
require.EqualValues(t, 2, p.Stat().AcquireCount())
}

func TestPool_backgroundHealthCheck(t *testing.T) {
t.Parallel()
var healthCheckCnt int64
p := PoolConnOpt(t, Options{
MinConns: 1,
HealthCheckFunc: func(ctx context.Context, client *ch.Client) error {
atomic.AddInt64(&healthCheckCnt, 1)
return nil
},
HealthCheckPeriod: 500 * time.Millisecond,
})
p.checkMinConns()
p.checkIdleConnsHealth()
assert.GreaterOrEqual(t, int64(1), atomic.LoadInt64(&healthCheckCnt))

hc := atomic.LoadInt64(&healthCheckCnt)
time.Sleep(750 * time.Millisecond)

@kavirajk kavirajk Jan 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This random sleep going to make the tests only more fragile IMHO

assert.Equal(t, hc+1, atomic.LoadInt64(&healthCheckCnt))
}
Loading