homelab/traefik
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix open connections metric

Browse source 
Co-authored-by: Romain <rtribotte@users.noreply.github.com>
This commit is contained in:
mpl 2023-03-20 16:02:06 +01:00 • committed by GitHub
parent 598a257ae1
commit 7c2af10bbd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 126 additions and 236 deletions
Download patch file Download diff file Expand all files Collapse all files

49
pkg/server/server_entrypoint_tcp.go
View file

@ -15,12 +15,14 @@ import (
"time"
"github.com/containous/alice"
gokitmetrics "github.com/go-kit/kit/metrics"
"github.com/pires/go-proxyproto"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"github.com/traefik/traefik/v3/pkg/config/static"
"github.com/traefik/traefik/v3/pkg/ip"
"github.com/traefik/traefik/v3/pkg/logs"
"github.com/traefik/traefik/v3/pkg/metrics"
"github.com/traefik/traefik/v3/pkg/middlewares"
"github.com/traefik/traefik/v3/pkg/middlewares/contenttype"
"github.com/traefik/traefik/v3/pkg/middlewares/forwardedheaders"
@ -67,7 +69,7 @@ func (h *httpForwarder) Accept() (net.Conn, error) {
type TCPEntryPoints map[string]*TCPEntryPoint
// NewTCPEntryPoints creates a new TCPEntryPoints.
func NewTCPEntryPoints(entryPointsConfig static.EntryPoints, hostResolverConfig *types.HostResolverConfig) (TCPEntryPoints, error) {
func NewTCPEntryPoints(entryPointsConfig static.EntryPoints, hostResolverConfig *types.HostResolverConfig, metricsRegistry metrics.Registry) (TCPEntryPoints, error) {
serverEntryPointsTCP := make(TCPEntryPoints)
for entryPointName, config := range entryPointsConfig {
protocol, err := config.GetProtocol()
@ -81,7 +83,11 @@ func NewTCPEntryPoints(entryPointsConfig static.EntryPoints, hostResolverConfig
ctx := log.With().Str(logs.EntryPointName, entryPointName).Logger().WithContext(context.Background())
serverEntryPointsTCP[entryPointName], err = NewTCPEntryPoint(ctx, config, hostResolverConfig)
openConnectionsGauge := metricsRegistry.
OpenConnectionsGauge().
With("entrypoint", entryPointName, "protocol", "TCP")
serverEntryPointsTCP[entryPointName], err = NewTCPEntryPoint(ctx, config, hostResolverConfig, openConnectionsGauge)
if err != nil {
return nil, fmt.Errorf("error while building entryPoint %s: %w", entryPointName, err)
}
@ -137,8 +143,8 @@ type TCPEntryPoint struct {
}
// NewTCPEntryPoint creates a new TCPEntryPoint.
func NewTCPEntryPoint(ctx context.Context, configuration *static.EntryPoint, hostResolverConfig *types.HostResolverConfig) (*TCPEntryPoint, error) {
tracker := newConnectionTracker()
func NewTCPEntryPoint(ctx context.Context, configuration *static.EntryPoint, hostResolverConfig *types.HostResolverConfig, openConnectionsGauge gokitmetrics.Gauge) (*TCPEntryPoint, error) {
tracker := newConnectionTracker(openConnectionsGauge)
listener, err := buildListener(ctx, configuration)
if err != nil {
@ -440,34 +446,45 @@ func buildListener(ctx context.Context, entryPoint *static.EntryPoint) (net.List
return listener, nil
}
func newConnectionTracker() *connectionTracker {
func newConnectionTracker(openConnectionsGauge gokitmetrics.Gauge) *connectionTracker {
return &connectionTracker{
conns: make(map[net.Conn]struct{}),
conns: make(map[net.Conn]struct{}),
openConnectionsGauge: openConnectionsGauge,
}
}
type connectionTracker struct {
conns map[net.Conn]struct{}
lock sync.RWMutex
connsMu sync.RWMutex
conns map[net.Conn]struct{}
openConnectionsGauge gokitmetrics.Gauge
}
// AddConnection add a connection in the tracked connections list.
func (c *connectionTracker) AddConnection(conn net.Conn) {
c.lock.Lock()
defer c.lock.Unlock()
c.connsMu.Lock()
c.conns[conn] = struct{}{}
c.connsMu.Unlock()
if c.openConnectionsGauge != nil {
c.openConnectionsGauge.Add(1)
}
}
// RemoveConnection remove a connection from the tracked connections list.
func (c *connectionTracker) RemoveConnection(conn net.Conn) {
c.lock.Lock()
defer c.lock.Unlock()
c.connsMu.Lock()
delete(c.conns, conn)
c.connsMu.Unlock()
if c.openConnectionsGauge != nil {
c.openConnectionsGauge.Add(-1)
}
}
func (c *connectionTracker) isEmpty() bool {
c.lock.RLock()
defer c.lock.RUnlock()
c.connsMu.RLock()
defer c.connsMu.RUnlock()
return len(c.conns) == 0
}
@ -489,8 +506,8 @@ func (c *connectionTracker) Shutdown(ctx context.Context) error {
// Close close all the connections in the tracked connections list.
func (c *connectionTracker) Close() {
c.lock.Lock()
defer c.lock.Unlock()
c.connsMu.Lock()
defer c.connsMu.Unlock()
for conn := range c.conns {
if err := conn.Close(); err != nil {
log.Error().Err(err).Msg("Error while closing connection")