Commit af82d410 authored by Brad Davidson's avatar Brad Davidson Committed by Brad Davidson

Add loadbalancer metrics

Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com> (cherry picked from commit 3d2fabb0) Signed-off-by: 's avatarBrad Davidson <brad.davidson@rancher.com>
parent 160d93f7
...@@ -8,6 +8,7 @@ import ( ...@@ -8,6 +8,7 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"strings" "strings"
"time"
"github.com/inetaf/tcpproxy" "github.com/inetaf/tcpproxy"
"github.com/k3s-io/k3s/pkg/version" "github.com/k3s-io/k3s/pkg/version"
...@@ -95,8 +96,18 @@ func New(ctx context.Context, dataDir, serviceName, defaultServerURL string, lbS ...@@ -95,8 +96,18 @@ func New(ctx context.Context, dataDir, serviceName, defaultServerURL string, lbS
} }
lb.proxy.AddRoute(serviceName, &tcpproxy.DialProxy{ lb.proxy.AddRoute(serviceName, &tcpproxy.DialProxy{
Addr: serviceName, Addr: serviceName,
DialContext: lb.servers.dialContext,
OnDialError: onDialError, OnDialError: onDialError,
DialContext: func(ctx context.Context, network, address string) (net.Conn, error) {
start := time.Now()
status := "success"
conn, err := lb.servers.dialContext(ctx, network, address)
latency := time.Since(start)
if err != nil {
status = "error"
}
loadbalancerDials.WithLabelValues(serviceName, status).Observe(latency.Seconds())
return conn, err
},
}) })
if err := lb.updateConfig(); err != nil { if err := lb.updateConfig(); err != nil {
......
package loadbalancer
import (
"github.com/k3s-io/k3s/pkg/version"
"github.com/prometheus/client_golang/prometheus"
"k8s.io/component-base/metrics"
)
var (
loadbalancerConnections = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: version.Program + "_loadbalancer_server_connections",
Help: "Count of current connections to loadbalancer server",
}, []string{"name", "server"})
loadbalancerState = prometheus.NewGaugeVec(prometheus.GaugeOpts{
Name: version.Program + "_loadbalancer_server_health",
Help: "Current health value of loadbalancer server",
}, []string{"name", "server"})
loadbalancerDials = prometheus.NewHistogramVec(prometheus.HistogramOpts{
Name: version.Program + "_loadbalancer_dial_duration_seconds",
Help: "Time taken to dial a connection to a backend server",
Buckets: metrics.ExponentialBuckets(0.001, 2, 15),
}, []string{"name", "status"})
)
// MustRegister registers loadbalancer metrics
func MustRegister(registerer prometheus.Registerer) {
registerer.MustRegister(loadbalancerConnections, loadbalancerState, loadbalancerDials)
}
...@@ -79,6 +79,9 @@ func (sl *serverList) setAddresses(serviceName string, addresses []string) bool ...@@ -79,6 +79,9 @@ func (sl *serverList) setAddresses(serviceName string, addresses []string) bool
// set state to invalid to prevent server from making additional connections // set state to invalid to prevent server from making additional connections
s.state = stateInvalid s.state = stateInvalid
closeAllFuncs = append(closeAllFuncs, s.closeAll) closeAllFuncs = append(closeAllFuncs, s.closeAll)
// remove metrics
loadbalancerState.DeleteLabelValues(serviceName, s.address)
loadbalancerConnections.DeleteLabelValues(serviceName, s.address)
return true return true
} }
return false return false
...@@ -459,7 +462,7 @@ func (sc *serverConn) Close() error { ...@@ -459,7 +462,7 @@ func (sc *serverConn) Close() error {
return sc.Conn.Close() return sc.Conn.Close()
} }
// runHealthChecks periodically health-checks all servers. // runHealthChecks periodically health-checks all servers and updates metrics
func (sl *serverList) runHealthChecks(ctx context.Context, serviceName string) { func (sl *serverList) runHealthChecks(ctx context.Context, serviceName string) {
wait.Until(func() { wait.Until(func() {
for _, s := range sl.getServers() { for _, s := range sl.getServers() {
...@@ -469,6 +472,10 @@ func (sl *serverList) runHealthChecks(ctx context.Context, serviceName string) { ...@@ -469,6 +472,10 @@ func (sl *serverList) runHealthChecks(ctx context.Context, serviceName string) {
case HealthCheckResultFailed: case HealthCheckResultFailed:
sl.recordFailure(s, reasonHealthCheck) sl.recordFailure(s, reasonHealthCheck)
} }
if s.state != stateInvalid {
loadbalancerState.WithLabelValues(serviceName, s.address).Set(float64(s.state))
loadbalancerConnections.WithLabelValues(serviceName, s.address).Set(float64(len(s.connections)))
}
} }
}, time.Second, ctx.Done()) }, time.Second, ctx.Done())
logrus.Debugf("Stopped health checking for load balancer %s", serviceName) logrus.Debugf("Stopped health checking for load balancer %s", serviceName)
......
...@@ -6,6 +6,7 @@ import ( ...@@ -6,6 +6,7 @@ import (
"github.com/gorilla/mux" "github.com/gorilla/mux"
"github.com/k3s-io/k3s/pkg/agent/https" "github.com/k3s-io/k3s/pkg/agent/https"
"github.com/k3s-io/k3s/pkg/agent/loadbalancer"
"github.com/k3s-io/k3s/pkg/daemons/config" "github.com/k3s-io/k3s/pkg/daemons/config"
"github.com/prometheus/client_golang/prometheus/promhttp" "github.com/prometheus/client_golang/prometheus/promhttp"
lassometrics "github.com/rancher/lasso/pkg/metrics" lassometrics "github.com/rancher/lasso/pkg/metrics"
...@@ -32,6 +33,8 @@ var DefaultMetrics = &Config{ ...@@ -32,6 +33,8 @@ var DefaultMetrics = &Config{
func init() { func init() {
// ensure that lasso exposes metrics through the same registry used by Kubernetes components // ensure that lasso exposes metrics through the same registry used by Kubernetes components
lassometrics.MustRegister(DefaultRegisterer) lassometrics.MustRegister(DefaultRegisterer)
// same for loadbalancer metrics
loadbalancer.MustRegister(DefaultRegisterer)
} }
// Config holds fields for the metrics listener // Config holds fields for the metrics listener
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment