mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Vladyslav SimonenkoandGitHub
1916d05702
## What changed? Instrument gRPC dial calls with metrics monitoring connection establishment. This PR adds the following metrics: 1. ServiceDialLatency 2. ServiceDialSuccessCount 3. ServiceDialErrorCount ## Why? Currently the visibility into gRPC dial calls is limited, there are no metrics for how long establishing the connection took, or how many failures we encountered. Since `grpc.NewClient` is "lazy", in the sense that it doesn't establish connection right away, but rather on the first RPC, we need to instrument it via some hooks. This PR uses `grpc.WithContextDialer` to achieve this. gRPC library itself uses platform-specific keep alive hack to use OS defaults for TCP keepalive, but since we are on Go 1.23+ we can simply use newly introduced KeepAliveConfig. ## How did you test it? - [X] built - [X] run locally and tested manually - [X] covered by existing tests - [X] added new unit test(s) - [ ] added new functional test(s) ## Potential risks
92 lines
2.3 KiB
Go
92 lines
2.3 KiB
Go
package rpc
|
|
|
|
import (
|
|
"context"
|
|
"net/http/httptrace"
|
|
"time"
|
|
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/metrics"
|
|
)
|
|
|
|
type dialTracer struct {
|
|
address string
|
|
metricsHandler metrics.Handler
|
|
logger log.Logger
|
|
}
|
|
|
|
// newDialTracer creates a dial tracer that logs errors during dial and produces metrics for different stages of the dial process (Connect)
|
|
func newDialTracer(
|
|
address string,
|
|
mh metrics.Handler,
|
|
logger log.Logger,
|
|
) *dialTracer {
|
|
l := log.With(
|
|
logger,
|
|
tag.NewStringTag("service", "client"),
|
|
tag.NewStringTag("address", address),
|
|
)
|
|
|
|
return &dialTracer{
|
|
address: address,
|
|
metricsHandler: mh,
|
|
logger: l,
|
|
}
|
|
}
|
|
|
|
func (d *dialTracer) beginNetworkDial(ctx context.Context) (context.Context, *networkDialTrace) {
|
|
ndt := &networkDialTrace{startedAt: time.Now()}
|
|
|
|
// Build a ClientTrace capturing TCP connections.
|
|
trace := &httptrace.ClientTrace{
|
|
ConnectStart: func(_, _ string) {
|
|
ndt.connectStart = time.Now()
|
|
},
|
|
ConnectDone: func(_ string, addr string, err error) {
|
|
// ConnectStart may not have been called if the address is reused
|
|
if !ndt.connectStart.IsZero() {
|
|
ndt.connectDuration = time.Since(ndt.connectStart)
|
|
}
|
|
ndt.connectAddr = addr
|
|
ndt.connectErr = err
|
|
},
|
|
}
|
|
|
|
return httptrace.WithClientTrace(ctx, trace), ndt
|
|
}
|
|
|
|
func (d *dialTracer) endNetworkDial(ndt *networkDialTrace, dialErr error) {
|
|
total := time.Since(ndt.startedAt)
|
|
|
|
if dialErr != nil {
|
|
fields := []tag.Tag{
|
|
tag.NewDurationTag("totalDuration", total),
|
|
tag.Error(dialErr),
|
|
tag.ErrorType(dialErr),
|
|
tag.NewDurationTag("connectDuration", ndt.connectDuration),
|
|
tag.NewStringTag("connectAddr", ndt.connectAddr),
|
|
}
|
|
if ndt.connectErr != nil {
|
|
fields = append(fields, tag.NewStringTag("connectErr", ndt.connectErr.Error()))
|
|
}
|
|
d.logger.Warn("network dial error", fields...)
|
|
metrics.ServiceDialErrorCount.With(d.metricsHandler).Record(1)
|
|
} else {
|
|
metrics.ServiceDialSuccessCount.With(d.metricsHandler).Record(1)
|
|
}
|
|
|
|
if ndt.connectDuration > 0 {
|
|
metrics.ServiceDialLatency.With(d.metricsHandler).Record(ndt.connectDuration)
|
|
}
|
|
}
|
|
|
|
type networkDialTrace struct {
|
|
startedAt time.Time
|
|
|
|
connectStart time.Time
|
|
connectDuration time.Duration
|
|
connectAddr string
|
|
connectErr error
|
|
}
|