Skip to content

chore: add logs to pq notification dialer #12020

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

Merged
merged 3 commits into from
Feb 6, 2024
Merged

chore: add logs to pq notification dialer #12020

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ab040dc
chore: add logs to pq notification dialer
deansheather Feb 5, 2024
0047c70
fixup! chore: add logs to pq notification dialer
deansheather Feb 5, 2024
87cea55
PR comments
deansheather Feb 6, 2024
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
57 changes: 55 additions & 2 deletions coderd/database/pubsub/pubsub.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"errors"
"net"
"sync"
"time"

Expand Down Expand Up @@ -352,11 +353,63 @@ func (p *PGPubsub) recordReconnect() {
}
}

// logDialer is a pq.Dialer and pq.DialerContext that logs when it starts
// connecting and when the TCP connection is established.
type logDialer struct {
logger slog.Logger
d net.Dialer
}

var (
_ pq.Dialer = logDialer{}
_ pq.DialerContext = logDialer{}
)

func (d logDialer) Dial(network, address string) (net.Conn, error) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
return d.DialContext(ctx, network, address)
}

func (d logDialer) DialTimeout(network, address string, timeout time.Duration) (net.Conn, error) {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
return d.DialContext(ctx, network, address)
}

func (d logDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
deadline, hasDeadline := ctx.Deadline()
timeoutMS := 0
if hasDeadline {
timeoutMS = int(time.Until(deadline) / time.Millisecond)
}

logger := d.logger.With(slog.F("network", network), slog.F("address", address), slog.F("timeout_ms", timeoutMS))

logger.Info(ctx, "pubsub dialing postgres")
start := time.Now()
conn, err := d.d.DialContext(ctx, network, address)
if err != nil {
logger.Error(ctx, "pubsub failed to dial postgres")
return nil, err
}
elapsed := time.Since(start)
logger.Info(ctx, "pubsub postgres TCP connection established", slog.F("elapsed_ms", elapsed.Milliseconds()))
return conn, nil
}

func (p *PGPubsub) startListener(ctx context.Context, connectURL string) error {
p.connected.Set(0)
// Creates a new listener using pq.
errCh := make(chan error)
p.pgListener = pq.NewListener(connectURL, time.Second, time.Minute, func(t pq.ListenerEventType, err error) {
var (
errCh = make(chan error)
dialer = logDialer{
logger: p.logger,
// pq.defaultDialer uses a zero net.Dialer as well.
d: net.Dialer{},
}
)
p.pgListener = pq.NewDialListener(dialer, connectURL, time.Second, time.Minute, func(t pq.ListenerEventType, err error) {
switch t {
case pq.ListenerEventConnected:
p.logger.Info(ctx, "pubsub connected to postgres")
Expand Down