mirror of
https://github.com/gravitational/teleport.git
synced 2026-10-11 22:49:54 +00:00
Add resource endpoint health check manager (#54335)
This commit is contained in:
@@ -8374,7 +8374,7 @@ message TargetHealth {
|
||||
string Address = 1 [(gogoproto.jsontag) = "address,omitempty"];
|
||||
// Protocol is the health check protocol such as "tcp".
|
||||
string Protocol = 2 [(gogoproto.jsontag) = "protocol,omitempty"];
|
||||
// Status is the health status, one of "", "disabled", "initialized", "healthy", "unhealthy".
|
||||
// Status is the health status, one of "", "unknown", "healthy", "unhealthy".
|
||||
string Status = 3 [(gogoproto.jsontag) = "status,omitempty"];
|
||||
// TransitionTimestamp is the time that the last status transition occurred.
|
||||
google.protobuf.Timestamp TransitionTimestamp = 4 [
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
/*
|
||||
Copyright 2025 Gravitational, Inc.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package types
|
||||
|
||||
// TargetHealthProtocol is the network protocol for a health checker.
|
||||
type TargetHealthProtocol string
|
||||
|
||||
const (
|
||||
// TargetHealthProtocolTCP is a target health check protocol.
|
||||
TargetHealthProtocolTCP TargetHealthProtocol = "TCP"
|
||||
)
|
||||
|
||||
// TargetHealthStatus is a target resource's health status.
|
||||
type TargetHealthStatus string
|
||||
|
||||
const (
|
||||
// TargetHealthStatusHealthy indicates that a health check target is healthy.
|
||||
TargetHealthStatusHealthy TargetHealthStatus = "healthy"
|
||||
// TargetHealthStatusUnhealthy indicates that a health check target is unhealthy.
|
||||
TargetHealthStatusUnhealthy TargetHealthStatus = "unhealthy"
|
||||
// TargetHealthStatusUnknown indicates that an unknown health check target health status.
|
||||
TargetHealthStatusUnknown TargetHealthStatus = "unknown"
|
||||
)
|
||||
|
||||
// TargetHealthTransitionReason is the reason for the target health status
|
||||
// transition.
|
||||
type TargetHealthTransitionReason string
|
||||
|
||||
const (
|
||||
// TargetHealthTransitionReasonInit means that initial health checks are in
|
||||
// progress.
|
||||
TargetHealthTransitionReasonInit TargetHealthTransitionReason = "initialized"
|
||||
// TargetHealthStatusDisabled indicates that health checks are disabled.
|
||||
TargetHealthTransitionReasonDisabled TargetHealthTransitionReason = "disabled"
|
||||
// TargetHealthTransitionReasonThreshold means that the health status
|
||||
// changed because the healthy or unhealthy threshold was reached.
|
||||
TargetHealthTransitionReasonThreshold TargetHealthTransitionReason = "threshold_reached"
|
||||
// TargetHealthTransitionReasonInternalError indicates that health checks
|
||||
// encountered an internal error (this is a bug).
|
||||
TargetHealthTransitionReasonInternalError TargetHealthTransitionReason = "internal_error"
|
||||
)
|
||||
@@ -22477,7 +22477,7 @@ type TargetHealth struct {
|
||||
Address string `protobuf:"bytes,1,opt,name=Address,proto3" json:"address,omitempty"`
|
||||
// Protocol is the health check protocol such as "tcp".
|
||||
Protocol string `protobuf:"bytes,2,opt,name=Protocol,proto3" json:"protocol,omitempty"`
|
||||
// Status is the health status, one of "", "disabled", "initialized", "healthy", "unhealthy".
|
||||
// Status is the health status, one of "", "unknown", "healthy", "unhealthy".
|
||||
Status string `protobuf:"bytes,3,opt,name=Status,proto3" json:"status,omitempty"`
|
||||
// TransitionTimestamp is the time that the last status transition occurred.
|
||||
TransitionTimestamp *time.Time `protobuf:"bytes,4,opt,name=TransitionTimestamp,proto3,stdtime" json:"transition_timestamp,omitempty"`
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"time"
|
||||
|
||||
"github.com/gravitational/teleport/api/defaults"
|
||||
healthcheckconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/healthcheckconfig/v1"
|
||||
labelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/label/v1"
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/api/types/label"
|
||||
apiutils "github.com/gravitational/teleport/api/utils"
|
||||
)
|
||||
|
||||
// healthCheckConfig is an internal health check config type converted from a
|
||||
// [*healthcheckconfigv1.HealthCheckConfig] with defaults set.
|
||||
type healthCheckConfig struct {
|
||||
name string
|
||||
protocol types.TargetHealthProtocol
|
||||
interval time.Duration
|
||||
timeout time.Duration
|
||||
healthyThreshold uint32
|
||||
unhealthyThreshold uint32
|
||||
databaseLabelMatchers types.LabelMatchers
|
||||
}
|
||||
|
||||
// newHealthCheckConfig converts a health check config protobuf message into a
|
||||
// [healthCheckConfig] and sets defaults.
|
||||
func newHealthCheckConfig(cfg *healthcheckconfigv1.HealthCheckConfig) *healthCheckConfig {
|
||||
spec := cfg.GetSpec()
|
||||
match := spec.GetMatch()
|
||||
return &healthCheckConfig{
|
||||
name: cfg.GetMetadata().GetName(),
|
||||
timeout: cmp.Or(spec.GetTimeout().AsDuration(), defaults.HealthCheckTimeout),
|
||||
interval: cmp.Or(spec.GetInterval().AsDuration(), defaults.HealthCheckInterval),
|
||||
healthyThreshold: cmp.Or(spec.GetHealthyThreshold(), defaults.HealthCheckHealthyThreshold),
|
||||
unhealthyThreshold: cmp.Or(spec.GetUnhealthyThreshold(), defaults.HealthCheckUnhealthyThreshold),
|
||||
// we only support plain TCP health checks currently, but eventually we
|
||||
// may add support for other protocols such as TLS or HTTP
|
||||
protocol: types.TargetHealthProtocolTCP,
|
||||
databaseLabelMatchers: newLabelMatchers(match.GetDbLabelsExpression(), match.GetDbLabels()),
|
||||
}
|
||||
}
|
||||
|
||||
// equivalent returns whether the config is equivalent to another, where
|
||||
// equivalence is defined as equality in all fields except the matchers.
|
||||
func (h *healthCheckConfig) equivalent(other *healthCheckConfig) bool {
|
||||
return (h == nil && other == nil) ||
|
||||
h != nil && other != nil &&
|
||||
h.name == other.name &&
|
||||
h.protocol == other.protocol &&
|
||||
h.interval == other.interval &&
|
||||
h.healthyThreshold == other.healthyThreshold &&
|
||||
h.unhealthyThreshold == other.unhealthyThreshold
|
||||
}
|
||||
|
||||
// getLabelMatchers returns the label matchers to use for the given resource
|
||||
// kind.
|
||||
func (h *healthCheckConfig) getLabelMatchers(kind string) types.LabelMatchers {
|
||||
switch kind {
|
||||
case types.KindDatabase:
|
||||
return h.databaseLabelMatchers
|
||||
}
|
||||
// unreachable since we enforce a list of supported target resource kinds,
|
||||
// but empty matchers do the right thing anyway: don't match anything.
|
||||
return types.LabelMatchers{}
|
||||
}
|
||||
|
||||
// newLabelMatchers creates a new [types.LabelMatchers] from an expression and
|
||||
// r153 labels.
|
||||
func newLabelMatchers(expr string, labels []*labelv1.Label) types.LabelMatchers {
|
||||
out := types.LabelMatchers{
|
||||
Expression: expr,
|
||||
Labels: make(types.Labels),
|
||||
}
|
||||
for k, vs := range label.ToMap(labels) {
|
||||
out.Labels[k] = apiutils.Strings(vs)
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,226 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
|
||||
"github.com/gravitational/teleport/api/defaults"
|
||||
healthcheckconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/healthcheckconfig/v1"
|
||||
labelv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/label/v1"
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/api/types/healthcheckconfig"
|
||||
"github.com/gravitational/teleport/api/utils"
|
||||
)
|
||||
|
||||
func Test_newHealthCheckConfig(t *testing.T) {
|
||||
t.Parallel()
|
||||
fullCfg, err := healthcheckconfig.NewHealthCheckConfig(
|
||||
"full",
|
||||
&healthcheckconfigv1.HealthCheckConfigSpec{
|
||||
Match: &healthcheckconfigv1.Matcher{
|
||||
DbLabels: []*labelv1.Label{{
|
||||
Name: "foo",
|
||||
Values: []string{"bar", "baz"},
|
||||
}},
|
||||
DbLabelsExpression: `labels["qux"] == "*"`,
|
||||
},
|
||||
Timeout: durationpb.New(time.Second * 42),
|
||||
Interval: durationpb.New(time.Second * 43),
|
||||
HealthyThreshold: 7,
|
||||
UnhealthyThreshold: 8,
|
||||
},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
minimalCfg, err := healthcheckconfig.NewHealthCheckConfig(
|
||||
"minimal",
|
||||
&healthcheckconfigv1.HealthCheckConfigSpec{
|
||||
Match: &healthcheckconfigv1.Matcher{
|
||||
DbLabelsExpression: `labels["*"] == "*"`,
|
||||
},
|
||||
},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
|
||||
tests := []struct {
|
||||
desc string
|
||||
cfg *healthcheckconfigv1.HealthCheckConfig
|
||||
want *healthCheckConfig
|
||||
}{
|
||||
{
|
||||
desc: "copies all settings",
|
||||
cfg: fullCfg,
|
||||
want: &healthCheckConfig{
|
||||
name: "full",
|
||||
timeout: time.Second * 42,
|
||||
interval: time.Second * 43,
|
||||
healthyThreshold: 7,
|
||||
unhealthyThreshold: 8,
|
||||
protocol: types.TargetHealthProtocolTCP,
|
||||
databaseLabelMatchers: types.LabelMatchers{
|
||||
Labels: types.Labels{
|
||||
"foo": utils.Strings{"bar", "baz"},
|
||||
},
|
||||
Expression: `labels["qux"] == "*"`,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "applies defaults",
|
||||
cfg: minimalCfg,
|
||||
want: &healthCheckConfig{
|
||||
name: "minimal",
|
||||
timeout: defaults.HealthCheckTimeout,
|
||||
interval: defaults.HealthCheckInterval,
|
||||
healthyThreshold: defaults.HealthCheckHealthyThreshold,
|
||||
unhealthyThreshold: defaults.HealthCheckUnhealthyThreshold,
|
||||
protocol: types.TargetHealthProtocolTCP,
|
||||
databaseLabelMatchers: types.LabelMatchers{
|
||||
Labels: types.Labels{},
|
||||
Expression: `labels["*"] == "*"`,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
got := newHealthCheckConfig(test.cfg)
|
||||
require.Equal(t, test.want, got)
|
||||
})
|
||||
}
|
||||
}
|
||||
func TestHealthCheckConfig_equivalent(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
desc string
|
||||
a, b *healthCheckConfig
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
desc: "both nil",
|
||||
a: nil,
|
||||
b: nil,
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
desc: "one nil, one non-nil",
|
||||
a: &healthCheckConfig{},
|
||||
b: nil,
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
desc: "all fields equal",
|
||||
a: &healthCheckConfig{
|
||||
name: "test",
|
||||
protocol: "http",
|
||||
interval: time.Second,
|
||||
healthyThreshold: 3,
|
||||
unhealthyThreshold: 5,
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
name: "test",
|
||||
protocol: "http",
|
||||
interval: time.Second,
|
||||
healthyThreshold: 3,
|
||||
unhealthyThreshold: 5,
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
desc: "all fields equal ignoring labels",
|
||||
a: &healthCheckConfig{
|
||||
name: "test",
|
||||
protocol: "http",
|
||||
interval: time.Second,
|
||||
healthyThreshold: 3,
|
||||
unhealthyThreshold: 5,
|
||||
databaseLabelMatchers: types.LabelMatchers{Expression: "a", Labels: types.Labels{"a": {"a"}}},
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
name: "test",
|
||||
protocol: "http",
|
||||
interval: time.Second,
|
||||
healthyThreshold: 3,
|
||||
unhealthyThreshold: 5,
|
||||
databaseLabelMatchers: types.LabelMatchers{Expression: "b", Labels: types.Labels{"b": {"b"}}},
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
desc: "different name",
|
||||
a: &healthCheckConfig{
|
||||
name: "test1",
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
name: "test2",
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
desc: "different protocol",
|
||||
a: &healthCheckConfig{
|
||||
protocol: "http",
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
protocol: "tcp",
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
desc: "different interval",
|
||||
a: &healthCheckConfig{
|
||||
interval: time.Second,
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
interval: 2 * time.Second,
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
desc: "different healthyThreshold",
|
||||
a: &healthCheckConfig{
|
||||
healthyThreshold: 2,
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
healthyThreshold: 3,
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
desc: "different unhealthyThreshold",
|
||||
a: &healthCheckConfig{
|
||||
unhealthyThreshold: 2,
|
||||
},
|
||||
b: &healthCheckConfig{
|
||||
unhealthyThreshold: 3,
|
||||
},
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.desc, func(t *testing.T) {
|
||||
require.Equal(t, tt.want, tt.a.equivalent(tt.b))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,324 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/jonboulle/clockwork"
|
||||
|
||||
"github.com/gravitational/teleport"
|
||||
healthcheckconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/healthcheckconfig/v1"
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/api/utils/retryutils"
|
||||
"github.com/gravitational/teleport/lib/services"
|
||||
"github.com/gravitational/teleport/lib/utils/interval"
|
||||
)
|
||||
|
||||
// Manager manages health checks for registered resource targets.
|
||||
type Manager interface {
|
||||
// Start starts the health check manager.
|
||||
Start(ctx context.Context) error
|
||||
// AddTarget adds a new target health checker and starts the health checker.
|
||||
AddTarget(ctx context.Context, target Target) error
|
||||
// RemoveTarget stops a given resource's health checker and removes it from
|
||||
// the manager's monitored resources.
|
||||
RemoveTarget(r types.ResourceWithLabels) error
|
||||
// GetTargetHealth returns the health of a given resource.
|
||||
GetTargetHealth(r types.ResourceWithLabels) (*types.TargetHealth, error)
|
||||
}
|
||||
|
||||
// ManagerConfig is the configuration options for [Manager].
|
||||
type ManagerConfig struct {
|
||||
// Clock is optional and can be used to control time in tests.
|
||||
Clock clockwork.Clock
|
||||
// Component is a component used in logs.
|
||||
Component string
|
||||
// Events is used to create new event watchers.
|
||||
Events types.Events
|
||||
// HealthCheckConfigReader can list or get health check config resources.
|
||||
HealthCheckConfigReader services.HealthCheckConfigReader
|
||||
}
|
||||
|
||||
// checkAndSetDefaults checks the manager config for errors and applies defaults.
|
||||
func (c *ManagerConfig) checkAndSetDefaults() error {
|
||||
if c.Component == "" {
|
||||
return trace.BadParameter("missing Component")
|
||||
}
|
||||
if c.Events == nil {
|
||||
return trace.BadParameter("missing Events")
|
||||
}
|
||||
if c.HealthCheckConfigReader == nil {
|
||||
return trace.BadParameter("missing HealthCheckConfigReader")
|
||||
}
|
||||
if c.Clock == nil {
|
||||
c.Clock = clockwork.NewRealClock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// NewManager creates a new unstarted health check [Manager].
|
||||
func NewManager(cfg ManagerConfig) (Manager, error) {
|
||||
mgr, err := newManager(cfg)
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
return mgr, nil
|
||||
}
|
||||
|
||||
func newManager(cfg ManagerConfig) (*manager, error) {
|
||||
if err := cfg.checkAndSetDefaults(); err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
mgr := &manager{
|
||||
cfg: cfg,
|
||||
logger: slog.With(teleport.ComponentKey, cfg.Component),
|
||||
workers: make(map[resourceKey]*worker),
|
||||
}
|
||||
return mgr, nil
|
||||
}
|
||||
|
||||
// manager implements [Manager].
|
||||
type manager struct {
|
||||
cfg ManagerConfig
|
||||
configWatcher *services.HealthCheckConfigWatcher
|
||||
logger *slog.Logger
|
||||
|
||||
// mu guards concurrent access to the monitored health check configs and
|
||||
// target resource workers.
|
||||
mu sync.RWMutex
|
||||
configs []*healthCheckConfig
|
||||
workers map[resourceKey]*worker
|
||||
}
|
||||
|
||||
// newResourceKey creates a new key for a given resource.
|
||||
func newResourceKey(r types.ResourceWithLabels) resourceKey {
|
||||
return resourceKey{
|
||||
name: r.GetName(),
|
||||
kind: r.GetKind(),
|
||||
}
|
||||
}
|
||||
|
||||
// resourceKey is a map key for target resource workers.
|
||||
type resourceKey struct {
|
||||
// name is the target resource name.
|
||||
name string
|
||||
// kind is the target resource kind.
|
||||
kind string
|
||||
}
|
||||
|
||||
func (r resourceKey) String() string {
|
||||
return fmt.Sprintf("name=%s, kind=%s", r.name, r.kind)
|
||||
}
|
||||
|
||||
// Start starts the health check manager.
|
||||
func (m *manager) Start(ctx context.Context) error {
|
||||
if err := m.startConfigWatcher(ctx); err != nil {
|
||||
return trace.Wrap(err, "failed to start health check config watcher")
|
||||
}
|
||||
m.startWorkerUpdater(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
// supportedTargetKinds is a list of resource kinds that support health checks.
|
||||
var supportedTargetKinds = []string{
|
||||
types.KindDatabase,
|
||||
}
|
||||
|
||||
// AddTarget adds a new target health checker and starts the health checker.
|
||||
func (m *manager) AddTarget(ctx context.Context, target Target) error {
|
||||
if err := target.checkAndSetDefaults(); err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
resource := target.GetResource()
|
||||
if !slices.Contains(supportedTargetKinds, resource.GetKind()) {
|
||||
return trace.BadParameter("health check target resource kind %q is not supported", resource.GetKind())
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
key := newResourceKey(resource)
|
||||
if _, ok := m.workers[key]; ok {
|
||||
return trace.AlreadyExists("target health checker %q already exists", key)
|
||||
}
|
||||
worker, err := newWorker(ctx, workerConfig{
|
||||
Clock: m.cfg.Clock,
|
||||
HealthCheckCfg: m.getConfigLocked(ctx, resource),
|
||||
Log: m.logger.With(
|
||||
"target_name", resource.GetName(),
|
||||
"target_kind", resource.GetKind(),
|
||||
"target_origin", resource.Origin(),
|
||||
),
|
||||
Target: target,
|
||||
})
|
||||
if err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
m.workers[key] = worker
|
||||
return nil
|
||||
}
|
||||
|
||||
// RemoveTarget stops a given resource's health checker and removes it from
|
||||
// the manager's monitored resources.
|
||||
func (m *manager) RemoveTarget(r types.ResourceWithLabels) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
key := newResourceKey(r)
|
||||
worker, ok := m.workers[key]
|
||||
if !ok {
|
||||
return trace.NotFound("health checker %q not found", key)
|
||||
}
|
||||
delete(m.workers, key)
|
||||
if err := worker.Close(); err != nil {
|
||||
m.logger.DebugContext(context.Background(),
|
||||
"Health checker failed to close",
|
||||
"error", err,
|
||||
)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetTargetHealth returns the health of a given resource.
|
||||
func (m *manager) GetTargetHealth(r types.ResourceWithLabels) (*types.TargetHealth, error) {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
key := newResourceKey(r)
|
||||
worker, ok := m.workers[key]
|
||||
if !ok {
|
||||
return nil, trace.NotFound("health checker %q not found", key)
|
||||
}
|
||||
return worker.GetTargetHealth(), nil
|
||||
}
|
||||
|
||||
// startConfigWatcher starts a watcher for health check config resources.
|
||||
func (m *manager) startConfigWatcher(ctx context.Context) error {
|
||||
watcher, err := services.NewHealthCheckConfigWatcher(ctx,
|
||||
services.HealthCheckConfigWatcherConfig{
|
||||
Reader: m.cfg.HealthCheckConfigReader,
|
||||
ResourceWatcherConfig: services.ResourceWatcherConfig{
|
||||
Client: m.cfg.Events,
|
||||
Clock: m.cfg.Clock,
|
||||
Component: m.cfg.Component,
|
||||
Logger: m.logger,
|
||||
},
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
m.configWatcher = watcher
|
||||
m.logger.DebugContext(ctx, "Started health check config resource watcher")
|
||||
go func() {
|
||||
defer watcher.Close()
|
||||
defer m.logger.DebugContext(ctx, "Stopped health check config resource watcher")
|
||||
for {
|
||||
select {
|
||||
case configs := <-watcher.ResourcesC:
|
||||
m.updateConfigs(ctx, configs)
|
||||
case <-watcher.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateConfigs updates the manager's known health check configs and
|
||||
// recalculates the matching config for each registered target resource.
|
||||
func (m *manager) updateConfigs(ctx context.Context, cs []*healthcheckconfigv1.HealthCheckConfig) {
|
||||
// Config priority is by ascending order of name - the first config to match, wins.
|
||||
slices.SortFunc(cs, func(a, b *healthcheckconfigv1.HealthCheckConfig) int {
|
||||
return cmp.Compare(a.GetMetadata().GetName(), b.GetMetadata().GetName())
|
||||
})
|
||||
configs := make([]*healthCheckConfig, 0, len(cs))
|
||||
for _, c := range cs {
|
||||
configs = append(configs, newHealthCheckConfig(c))
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.configs = configs
|
||||
m.updateWorkersLocked(ctx)
|
||||
}
|
||||
|
||||
// startWorkerUpdater periodically evaluates and updates the matching health
|
||||
// check config for each worker.
|
||||
// The config for a worker may change if its target resource labels change
|
||||
// dynamically.
|
||||
func (m *manager) startWorkerUpdater(ctx context.Context) {
|
||||
updateInterval := interval.New(interval.Config{
|
||||
Duration: time.Minute,
|
||||
Jitter: retryutils.SeventhJitter,
|
||||
Clock: m.cfg.Clock,
|
||||
})
|
||||
go func() {
|
||||
defer updateInterval.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-updateInterval.Next():
|
||||
m.mu.Lock()
|
||||
m.updateWorkersLocked(ctx)
|
||||
m.mu.Unlock()
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// updateWorkersLocked evaluates and updates the matching health check config
|
||||
// for each worker.
|
||||
func (m *manager) updateWorkersLocked(ctx context.Context) {
|
||||
for _, w := range m.workers {
|
||||
newCfg := m.getConfigLocked(ctx, w.GetTargetResource())
|
||||
w.UpdateHealthCheckConfig(newCfg)
|
||||
}
|
||||
}
|
||||
|
||||
// getConfigLocked gets a matching config for the the given resource or returns
|
||||
// nil if no config matches.
|
||||
func (m *manager) getConfigLocked(ctx context.Context, r types.ResourceWithLabels) *healthCheckConfig {
|
||||
for _, cfg := range m.configs {
|
||||
matched, _, err := services.CheckLabelsMatch(
|
||||
types.Allow,
|
||||
cfg.getLabelMatchers(r.GetKind()),
|
||||
nil, // userTraits
|
||||
r,
|
||||
false, // debug
|
||||
)
|
||||
if err != nil {
|
||||
m.logger.WarnContext(ctx, "Health check config failed to match",
|
||||
"health_check_config", cfg.name,
|
||||
"error", err,
|
||||
)
|
||||
continue
|
||||
}
|
||||
if matched {
|
||||
return cfg
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,577 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/jonboulle/clockwork"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
|
||||
apidefaults "github.com/gravitational/teleport/api/defaults"
|
||||
healthcheckconfigv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/healthcheckconfig/v1"
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/api/types/healthcheckconfig"
|
||||
"github.com/gravitational/teleport/lib/backend/memory"
|
||||
"github.com/gravitational/teleport/lib/defaults"
|
||||
"github.com/gravitational/teleport/lib/services"
|
||||
"github.com/gravitational/teleport/lib/services/local"
|
||||
"github.com/gravitational/teleport/lib/utils"
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
utils.InitLoggerForTests()
|
||||
os.Exit(m.Run())
|
||||
}
|
||||
|
||||
type mockEvents struct {
|
||||
types.Events
|
||||
}
|
||||
|
||||
type mockHealthCheckConfigReader struct {
|
||||
services.HealthCheckConfigReader
|
||||
}
|
||||
|
||||
func TestNewManager(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
desc string
|
||||
wantErr string
|
||||
cfg ManagerConfig
|
||||
}{
|
||||
{
|
||||
desc: "missing Component",
|
||||
wantErr: "missing Component",
|
||||
cfg: ManagerConfig{
|
||||
Events: &mockEvents{},
|
||||
HealthCheckConfigReader: &mockHealthCheckConfigReader{},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "missing Events",
|
||||
wantErr: "missing Events",
|
||||
cfg: ManagerConfig{
|
||||
Component: "test-component",
|
||||
HealthCheckConfigReader: &mockHealthCheckConfigReader{},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "missing HealthCheckConfigReader",
|
||||
wantErr: "missing HealthCheckConfigReader",
|
||||
cfg: ManagerConfig{
|
||||
Component: "test-component",
|
||||
Events: &mockEvents{},
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "success with missing clock applies default",
|
||||
cfg: ManagerConfig{
|
||||
Component: "test-component",
|
||||
Events: &mockEvents{},
|
||||
HealthCheckConfigReader: &mockHealthCheckConfigReader{},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
mgr, err := NewManager(test.cfg)
|
||||
if test.wantErr != "" {
|
||||
require.ErrorContains(t, err, test.wantErr)
|
||||
require.Nil(t, mgr)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, mgr)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Minute*2)
|
||||
t.Cleanup(cancel)
|
||||
bk, err := memory.New(memory.Config{
|
||||
Context: ctx,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
healthConfigSvc, err := local.NewHealthCheckConfigService(bk)
|
||||
require.NoError(t, err)
|
||||
|
||||
// create a health check config that only matches prod databases
|
||||
prodHCC := healthCheckConfigFixture(t, "prod")
|
||||
prodHCC.Spec.Match.DbLabelsExpression = `labels.env == "prod"`
|
||||
_, err = healthConfigSvc.CreateHealthCheckConfig(ctx, prodHCC)
|
||||
require.NoError(t, err)
|
||||
|
||||
clock := clockwork.NewFakeClock()
|
||||
eventsCh := make(chan testEvent, 1024)
|
||||
mgr, err := NewManager(ManagerConfig{
|
||||
Component: "test",
|
||||
Events: local.NewEventsService(bk),
|
||||
HealthCheckConfigReader: healthConfigSvc,
|
||||
Clock: clock,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.NoError(t, mgr.Start(ctx))
|
||||
|
||||
require.EventuallyWithT(t, func(t *assert.CollectT) {
|
||||
mgr := mgr.(*manager)
|
||||
mgr.mu.RLock()
|
||||
defer mgr.mu.RUnlock()
|
||||
assert.Len(t, mgr.configs, 1)
|
||||
}, 5*time.Second, 100*time.Millisecond, "waiting for manager config watcher to observe the health check config")
|
||||
|
||||
listener, err := net.Listen("tcp", "localhost:0")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = listener.Close() })
|
||||
prodDB, err := types.NewDatabaseV3(types.Metadata{
|
||||
Name: "prodDB",
|
||||
Labels: map[string]string{
|
||||
"env": "prod",
|
||||
},
|
||||
}, types.DatabaseSpecV3{
|
||||
Protocol: defaults.ProtocolPostgres,
|
||||
URI: listener.Addr().String(),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
devDB, err := types.NewDatabaseV3(types.Metadata{
|
||||
Name: "devDB",
|
||||
Labels: map[string]string{
|
||||
"env": "dev",
|
||||
},
|
||||
}, types.DatabaseSpecV3{
|
||||
Protocol: defaults.ProtocolPostgres,
|
||||
URI: listener.Addr().String(),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
var endpointMu sync.Mutex
|
||||
prodDialer := fakeDialer{}
|
||||
err = mgr.AddTarget(ctx, Target{
|
||||
GetResource: func() types.ResourceWithLabels {
|
||||
endpointMu.Lock()
|
||||
defer endpointMu.Unlock()
|
||||
return prodDB
|
||||
},
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) {
|
||||
endpointMu.Lock()
|
||||
defer endpointMu.Unlock()
|
||||
return []string{prodDB.GetURI()}, nil
|
||||
},
|
||||
dialFn: prodDialer.DialContext,
|
||||
onHealthCheck: func(lastResultErr error) {
|
||||
eventsCh <- lastResultTestEvent(prodDB.GetName(), lastResultErr)
|
||||
},
|
||||
onConfigUpdate: func() {
|
||||
eventsCh <- configUpdateTestEvent(prodDB.GetName())
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
devDialer := fakeDialer{}
|
||||
err = mgr.AddTarget(ctx, Target{
|
||||
GetResource: func() types.ResourceWithLabels {
|
||||
endpointMu.Lock()
|
||||
defer endpointMu.Unlock()
|
||||
return devDB
|
||||
},
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) {
|
||||
endpointMu.Lock()
|
||||
defer endpointMu.Unlock()
|
||||
return []string{devDB.GetURI()}, nil
|
||||
},
|
||||
dialFn: devDialer.DialContext,
|
||||
onHealthCheck: func(lastResultErr error) {
|
||||
eventsCh <- lastResultTestEvent(devDB.GetName(), lastResultErr)
|
||||
},
|
||||
onConfigUpdate: func() {
|
||||
eventsCh <- configUpdateTestEvent(devDB.GetName())
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Run("duplicate target is an error", func(t *testing.T) {
|
||||
err = mgr.AddTarget(ctx, Target{
|
||||
GetResource: func() types.ResourceWithLabels { return devDB },
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) { return nil, nil },
|
||||
})
|
||||
require.Error(t, err)
|
||||
require.IsType(t, trace.AlreadyExists(""), err)
|
||||
})
|
||||
t.Run("unsupported target resource is an error", func(t *testing.T) {
|
||||
err = mgr.AddTarget(ctx, Target{
|
||||
GetResource: func() types.ResourceWithLabels { return &fakeResource{kind: "node"} },
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) { return nil, nil },
|
||||
})
|
||||
require.Error(t, err)
|
||||
require.IsType(t, trace.BadParameter(""), err)
|
||||
})
|
||||
|
||||
requireTargetHealth := func(t *testing.T, r types.ResourceWithLabels, status types.TargetHealthStatus, reason types.TargetHealthTransitionReason) {
|
||||
t.Helper()
|
||||
health, err := mgr.GetTargetHealth(r)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, string(status), health.Status)
|
||||
require.Equal(t, string(reason), health.TransitionReason)
|
||||
}
|
||||
|
||||
prodPass := lastResultPassTestEvent(prodDB.GetName())
|
||||
prodFail := lastResultFailTestEvent(prodDB.GetName())
|
||||
// initially checks should be disabled for dev but enabled for prod
|
||||
awaitTestEvents(t, eventsCh, nil,
|
||||
expect(prodPass),
|
||||
deny(prodFail),
|
||||
denyAll(devDB.GetName()),
|
||||
)
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusUnknown, types.TargetHealthTransitionReasonDisabled)
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonInit)
|
||||
|
||||
// another check should reach the prodHCC configured threshold.
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC),
|
||||
expect(prodPass),
|
||||
deny(prodFail),
|
||||
denyAll(devDB.GetName()),
|
||||
)
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// now reject health check connections to simulate an unhealthy endpoint
|
||||
prodDialer.deny()
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC),
|
||||
advanceCount(2),
|
||||
expect(prodFail, prodFail),
|
||||
deny(prodPass),
|
||||
denyAll(devDB.GetName()))
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusUnhealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// enable dev health checks for dev db and simulate an unhealthy endpoint on init
|
||||
devDialer.deny()
|
||||
devHCC := healthCheckConfigFixture(t, "dev")
|
||||
devHCC.Spec.Match.DbLabelsExpression = `labels.env == "dev"`
|
||||
devHCC, err = healthConfigSvc.CreateHealthCheckConfig(ctx, devHCC)
|
||||
require.NoError(t, err)
|
||||
|
||||
devConfigUpdate := configUpdateTestEvent(devDB.GetName())
|
||||
awaitTestEvents(t, eventsCh, nil,
|
||||
expect(devConfigUpdate),
|
||||
denyAll(prodDB.GetName()),
|
||||
)
|
||||
// the first failing check after dev checks are enabled will transition to unhealthy regardless of unhealthy threshold
|
||||
devPass := lastResultPassTestEvent(devDB.GetName())
|
||||
devFail := lastResultFailTestEvent(devDB.GetName())
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
expect(devFail, prodFail),
|
||||
deny(prodPass, devPass),
|
||||
)
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusUnhealthy, types.TargetHealthTransitionReasonInit)
|
||||
|
||||
// the next dev check should update health status because the unhealthy threshold was met
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
expect(devFail, prodFail),
|
||||
deny(prodPass, devPass),
|
||||
)
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusUnhealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// set the unhealthy threshold high for dev, so we can simulate several
|
||||
// failing checks after dev becomes healthy without making it become unhealthy
|
||||
devHCC.Spec.UnhealthyThreshold = 10
|
||||
devHCC, err = healthConfigSvc.UpsertHealthCheckConfig(ctx, devHCC)
|
||||
require.NoError(t, err)
|
||||
awaitTestEvents(t, eventsCh, nil,
|
||||
expect(devConfigUpdate),
|
||||
denyAll(prodDB.GetName()),
|
||||
)
|
||||
|
||||
// now simulate the endpoints becoming healthy again
|
||||
prodDialer.allow()
|
||||
devDialer.allow()
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
expect(devPass, prodPass),
|
||||
deny(devFail, prodFail),
|
||||
)
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
expect(devPass, prodPass),
|
||||
deny(devFail, prodFail),
|
||||
)
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// now disable the prod health checks
|
||||
err = healthConfigSvc.DeleteHealthCheckConfig(ctx, "prod")
|
||||
require.NoError(t, err)
|
||||
|
||||
awaitTestEvents(t, eventsCh, nil,
|
||||
expect(configUpdateTestEvent(prodDB.GetName())),
|
||||
denyAll(devDB.GetName()),
|
||||
)
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
advanceCount(2),
|
||||
expect(devPass, devPass),
|
||||
deny(devFail, prodFail),
|
||||
)
|
||||
// prod db should be disabled eventually
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusUnknown, types.TargetHealthTransitionReasonDisabled)
|
||||
// but dev db should still be healthy
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// fail some checks, then update config to lower the unhealthy threshold
|
||||
devDialer.deny()
|
||||
awaitTestEvents(t, eventsCh, clock,
|
||||
advanceByHCC(prodHCC, devHCC),
|
||||
advanceCount(3),
|
||||
expect(devFail, devFail, devFail),
|
||||
denyAll(prodDB.GetName()),
|
||||
)
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusHealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
devHCC.Spec.UnhealthyThreshold = 1
|
||||
devHCC.Spec.Interval = durationpb.New(time.Second * 100)
|
||||
_, err = healthConfigSvc.UpdateHealthCheckConfig(ctx, devHCC)
|
||||
require.NoError(t, err)
|
||||
awaitTestEvents(t, eventsCh, nil,
|
||||
expect(devConfigUpdate),
|
||||
denyAll(prodDB.GetName()),
|
||||
// dev shouldn't need to run any checks to observe this behavior
|
||||
deny(devPass, devFail),
|
||||
)
|
||||
// config update should set unhealthy status since the new threshold is already met
|
||||
requireTargetHealth(t, devDB, types.TargetHealthStatusUnhealthy, types.TargetHealthTransitionReasonThreshold)
|
||||
|
||||
// remove a target
|
||||
err = mgr.RemoveTarget(devDB)
|
||||
require.NoError(t, err)
|
||||
// shouldn't be any target health after the target is removed
|
||||
_, err = mgr.GetTargetHealth(devDB)
|
||||
require.Error(t, err)
|
||||
require.IsType(t, trace.NotFound(""), err)
|
||||
|
||||
err = mgr.RemoveTarget(devDB)
|
||||
require.Error(t, err)
|
||||
require.IsType(t, trace.NotFound(""), err)
|
||||
|
||||
// prodDB should still be disabled
|
||||
requireTargetHealth(t, prodDB, types.TargetHealthStatusUnknown, types.TargetHealthTransitionReasonDisabled)
|
||||
}
|
||||
|
||||
func healthCheckConfigFixture(t *testing.T, name string) *healthcheckconfigv1.HealthCheckConfig {
|
||||
t.Helper()
|
||||
out, err := healthcheckconfig.NewHealthCheckConfig(name,
|
||||
&healthcheckconfigv1.HealthCheckConfigSpec{
|
||||
Match: &healthcheckconfigv1.Matcher{
|
||||
DbLabelsExpression: `labels["*"] == "*"`,
|
||||
},
|
||||
Interval: durationpb.New(apidefaults.HealthCheckInterval),
|
||||
Timeout: durationpb.New(apidefaults.HealthCheckTimeout),
|
||||
HealthyThreshold: 2,
|
||||
UnhealthyThreshold: 2,
|
||||
},
|
||||
)
|
||||
require.NoError(t, err)
|
||||
return out
|
||||
}
|
||||
|
||||
type testEventOptions struct {
|
||||
expect map[testEvent]int
|
||||
assertions []func(*testing.T, testEvent) bool
|
||||
advance time.Duration
|
||||
advanceCount int
|
||||
}
|
||||
|
||||
type eventOption func(*testEventOptions)
|
||||
|
||||
func expect(events ...testEvent) eventOption {
|
||||
return func(opts *testEventOptions) {
|
||||
for _, evt := range events {
|
||||
opts.expect[evt] = opts.expect[evt] + 1
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func deny(events ...testEvent) eventOption {
|
||||
denied := make(map[testEvent]struct{})
|
||||
for _, event := range events {
|
||||
denied[event] = struct{}{}
|
||||
}
|
||||
return func(opts *testEventOptions) {
|
||||
opts.assertions = append(opts.assertions, func(t *testing.T, event testEvent) bool {
|
||||
t.Helper()
|
||||
return assert.NotContains(t, denied, event, "observed a denied event")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// denyAll denies all events associated with the target
|
||||
func denyAll(deniedTarget string) eventOption {
|
||||
return func(opts *testEventOptions) {
|
||||
opts.assertions = append(opts.assertions, func(t *testing.T, event testEvent) bool {
|
||||
t.Helper()
|
||||
return assert.NotEqual(t, deniedTarget, event.target, "unexpected event target")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// advanceByHCC advances the clock by the largest interval to trigger all matching targets to run a check
|
||||
func advanceByHCC(first *healthcheckconfigv1.HealthCheckConfig, rest ...*healthcheckconfigv1.HealthCheckConfig) eventOption {
|
||||
max := first.GetSpec().GetInterval().AsDuration()
|
||||
for _, hcc := range rest {
|
||||
d := hcc.GetSpec().GetInterval().AsDuration()
|
||||
if d > max {
|
||||
max = d
|
||||
}
|
||||
}
|
||||
return advanceBy(max)
|
||||
}
|
||||
|
||||
func advanceBy(d time.Duration) eventOption {
|
||||
return func(opts *testEventOptions) {
|
||||
opts.advance = d
|
||||
}
|
||||
}
|
||||
|
||||
// advanceCount advance the clock clock times.
|
||||
// Probably a bad idea to advance more than once when you expect more than one type of event in the await loop.
|
||||
func advanceCount(count int) eventOption {
|
||||
return func(opts *testEventOptions) {
|
||||
opts.advanceCount = count
|
||||
}
|
||||
}
|
||||
|
||||
func awaitTestEvents(t *testing.T, ch <-chan testEvent, clock *clockwork.FakeClock, optFns ...eventOption) {
|
||||
t.Helper()
|
||||
options := testEventOptions{
|
||||
expect: make(map[testEvent]int),
|
||||
advance: 30 * time.Second,
|
||||
advanceCount: 1,
|
||||
}
|
||||
for _, o := range optFns {
|
||||
o(&options)
|
||||
}
|
||||
for {
|
||||
if len(options.expect) == 0 {
|
||||
return
|
||||
}
|
||||
if clock != nil && options.advanceCount > 0 {
|
||||
options.advanceCount--
|
||||
clock.Advance(options.advance)
|
||||
}
|
||||
|
||||
select {
|
||||
case event := <-ch:
|
||||
for _, denyFn := range options.assertions {
|
||||
if !denyFn(t, event) {
|
||||
require.Failf(t, "unexpected event", "event=%v", event)
|
||||
}
|
||||
}
|
||||
|
||||
options.expect[event] = options.expect[event] - 1
|
||||
if options.expect[event] < 1 {
|
||||
delete(options.expect, event)
|
||||
}
|
||||
case <-time.After(time.Second * 5):
|
||||
require.Failf(t, "timeout waiting for events", "expect=%+v", options.expect)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type fakeDialer struct {
|
||||
net.Dialer
|
||||
mu sync.RWMutex
|
||||
fakeErr error
|
||||
}
|
||||
|
||||
func (f *fakeDialer) allow() {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.fakeErr = nil
|
||||
}
|
||||
|
||||
func (f *fakeDialer) deny() {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.fakeErr = errors.New("fake error in test")
|
||||
}
|
||||
|
||||
func (f *fakeDialer) getFakeErr() error {
|
||||
f.mu.RLock()
|
||||
defer f.mu.RUnlock()
|
||||
return f.fakeErr
|
||||
}
|
||||
|
||||
func (f *fakeDialer) DialContext(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
if err := f.getFakeErr(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return f.Dialer.DialContext(ctx, network, addr)
|
||||
}
|
||||
|
||||
type testEvent struct {
|
||||
name testEventName
|
||||
target string
|
||||
}
|
||||
|
||||
type testEventName string
|
||||
|
||||
const (
|
||||
configUpdate testEventName = "configUpdate"
|
||||
lastResultPass testEventName = "lastResultPass"
|
||||
lastResultFail testEventName = "lastResultFail"
|
||||
)
|
||||
|
||||
func configUpdateTestEvent(targetName string) testEvent {
|
||||
return testEvent{
|
||||
name: configUpdate,
|
||||
target: targetName,
|
||||
}
|
||||
}
|
||||
|
||||
func lastResultTestEvent(targetName string, err error) testEvent {
|
||||
if err != nil {
|
||||
return lastResultFailTestEvent(targetName)
|
||||
}
|
||||
return lastResultPassTestEvent(targetName)
|
||||
}
|
||||
|
||||
func lastResultPassTestEvent(targetName string) testEvent {
|
||||
return testEvent{
|
||||
name: lastResultPass,
|
||||
target: targetName,
|
||||
}
|
||||
}
|
||||
|
||||
func lastResultFailTestEvent(targetName string) testEvent {
|
||||
return testEvent{
|
||||
name: lastResultFail,
|
||||
target: targetName,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
)
|
||||
|
||||
// EndpointsResolverFunc is callback func that returns endpoints for a target.
|
||||
type EndpointsResolverFunc func(ctx context.Context) ([]string, error)
|
||||
|
||||
// OnHealthChangeFunc is a func called on each health change.
|
||||
type OnHealthChangeFunc func(oldHealth, newHealth types.TargetHealth)
|
||||
|
||||
// Target is a health check target.
|
||||
type Target struct {
|
||||
// GetResource gets a copy of the target resource with updated labels.
|
||||
GetResource func() types.ResourceWithLabels
|
||||
// ResolverFn resolves the target endpoint(s).
|
||||
ResolverFn EndpointsResolverFunc
|
||||
|
||||
// -- test fields below --
|
||||
|
||||
// dialFn used to mock dialing in tests
|
||||
dialFn dialFunc
|
||||
// onHealthCheck is called after each health check.
|
||||
onHealthCheck func(lastResultErr error)
|
||||
// onConfigUpdate is called after each config update.
|
||||
onConfigUpdate func()
|
||||
}
|
||||
|
||||
func (t *Target) checkAndSetDefaults() error {
|
||||
if t.GetResource == nil {
|
||||
return trace.BadParameter("missing target resource getter")
|
||||
}
|
||||
if t.ResolverFn == nil {
|
||||
return trace.BadParameter("missing target endpoint resolver")
|
||||
}
|
||||
if t.dialFn == nil {
|
||||
t.dialFn = defaultDialer().DialContext
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func defaultDialer() *net.Dialer {
|
||||
return &net.Dialer{}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
)
|
||||
|
||||
type fakeResource struct {
|
||||
kind string
|
||||
types.ResourceWithLabels
|
||||
}
|
||||
|
||||
func (r *fakeResource) GetKind() string {
|
||||
return r.kind
|
||||
}
|
||||
|
||||
func TestTarget_checkAndSetDefaults(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
target *Target
|
||||
wantErr string
|
||||
}{
|
||||
{
|
||||
name: "valid target",
|
||||
target: &Target{
|
||||
GetResource: func() types.ResourceWithLabels { return &fakeResource{kind: types.KindDatabase} },
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) { return []string{"127.0.0.1"}, nil },
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "missing resource getter",
|
||||
target: &Target{
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) { return nil, nil },
|
||||
},
|
||||
wantErr: "missing target resource getter",
|
||||
},
|
||||
{
|
||||
name: "missing resolver",
|
||||
target: &Target{
|
||||
GetResource: func() types.ResourceWithLabels { return nil },
|
||||
},
|
||||
wantErr: "missing target endpoint resolver",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
err := tt.target.checkAndSetDefaults()
|
||||
if tt.wantErr != "" {
|
||||
require.Error(t, err)
|
||||
require.True(t, trace.IsBadParameter(err))
|
||||
require.Contains(t, err.Error(), tt.wantErr)
|
||||
} else {
|
||||
require.NoError(t, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,422 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/jonboulle/clockwork"
|
||||
"golang.org/x/sync/errgroup"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/api/utils"
|
||||
"github.com/gravitational/teleport/api/utils/retryutils"
|
||||
"github.com/gravitational/teleport/lib/utils/interval"
|
||||
)
|
||||
|
||||
// workerConfig is the configuration for a [workerI].
|
||||
type workerConfig struct {
|
||||
// Clock is optional and is used to control time in tests.
|
||||
Clock clockwork.Clock
|
||||
// HealthCheckCfg is the config for performing health checks.
|
||||
// If not specified, then health checks are disabled until the worker is
|
||||
// given a non-nil health check config to use.
|
||||
HealthCheckCfg *healthCheckConfig
|
||||
// Log is an optional logger.
|
||||
Log *slog.Logger
|
||||
// Target is the health check target.
|
||||
Target Target
|
||||
}
|
||||
|
||||
// checkAndSetDefaults checks the worker config and sets defaults.
|
||||
func (cfg *workerConfig) checkAndSetDefaults() error {
|
||||
if cfg.Clock == nil {
|
||||
cfg.Clock = clockwork.NewRealClock()
|
||||
}
|
||||
if cfg.Log == nil {
|
||||
cfg.Log = slog.Default()
|
||||
}
|
||||
if err := cfg.Target.checkAndSetDefaults(); err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// newWorker creates and starts a new worker.
|
||||
func newWorker(ctx context.Context, cfg workerConfig) (*worker, error) {
|
||||
if err := cfg.checkAndSetDefaults(); err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
w := &worker{
|
||||
cancel: cancel,
|
||||
clock: cfg.Clock,
|
||||
healthCheckCfg: cfg.HealthCheckCfg,
|
||||
healthCheckConfigUpdateCh: make(chan *healthCheckConfig, 1),
|
||||
log: cfg.Log,
|
||||
target: cfg.Target,
|
||||
}
|
||||
if w.healthCheckCfg != nil {
|
||||
const reason = types.TargetHealthTransitionReasonInit
|
||||
const message = "Health checker initialized"
|
||||
w.setTargetHealthStatus(types.TargetHealthStatusUnknown, reason, message)
|
||||
} else {
|
||||
w.setTargetDisabled(ctx)
|
||||
}
|
||||
go w.run(ctx)
|
||||
return w, nil
|
||||
}
|
||||
|
||||
// worker perform health checks against a target resource and keeps track of
|
||||
// the target resource's health.
|
||||
type worker struct {
|
||||
// cancel stops the worker permanently when called.
|
||||
cancel context.CancelFunc
|
||||
// clock is used to control time in tests.
|
||||
clock clockwork.Clock
|
||||
// healthCheckCfg is the current config used for performing health checks.
|
||||
// Nil if no health check config currently matches the target resource.
|
||||
healthCheckCfg *healthCheckConfig
|
||||
// healthCheckConfigUpdateCh is used to notify the worker of a new health
|
||||
// check config.
|
||||
healthCheckConfigUpdateCh chan *healthCheckConfig
|
||||
// healthCheckInterval is the current interval between health checks. Nil
|
||||
// if there isn't currently a matching health check config for this worker.
|
||||
healthCheckInterval *interval.Interval
|
||||
// log emits logs.
|
||||
log *slog.Logger
|
||||
// target is the health check target.
|
||||
target Target
|
||||
|
||||
// lastResultErr is the last health check error or nil if there was no error.
|
||||
lastResultErr error
|
||||
// lastResultCount is the count of consecutive passing or failing health
|
||||
// check results.
|
||||
lastResultCount uint32
|
||||
// lastResolvedEndpoints are the endpoints last resolved for a health check.
|
||||
lastResolvedEndpoints []string
|
||||
|
||||
// mu guards concurrent access to the target health.
|
||||
mu sync.RWMutex
|
||||
// targetHealth is the latest target health. Initialized to "unknown" status
|
||||
// before the worker starts.
|
||||
targetHealth types.TargetHealth
|
||||
}
|
||||
|
||||
// dialFunc dials an address on the given network.
|
||||
type dialFunc func(ctx context.Context, network, addr string) (net.Conn, error)
|
||||
|
||||
// GetTargetHealth returns the worker's target health.
|
||||
func (w *worker) GetTargetHealth() *types.TargetHealth {
|
||||
w.mu.RLock()
|
||||
defer w.mu.RUnlock()
|
||||
return utils.CloneProtoMsg(&w.targetHealth)
|
||||
}
|
||||
|
||||
// GetTargetResource returns the target resource.
|
||||
func (w *worker) GetTargetResource() types.ResourceWithLabels {
|
||||
return w.target.GetResource()
|
||||
}
|
||||
|
||||
// UpdateHealthCheckConfig updates the worker to use a new health check
|
||||
// config.
|
||||
func (w *worker) UpdateHealthCheckConfig(newCfg *healthCheckConfig) {
|
||||
// drain pending config update, if any
|
||||
select {
|
||||
case <-w.healthCheckConfigUpdateCh:
|
||||
default:
|
||||
}
|
||||
w.healthCheckConfigUpdateCh <- newCfg
|
||||
}
|
||||
|
||||
// Close closes the health checker.
|
||||
func (w *worker) Close() error {
|
||||
w.cancel()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *worker) run(ctx context.Context) {
|
||||
defer func() {
|
||||
if w.healthCheckInterval != nil {
|
||||
w.healthCheckInterval.Stop()
|
||||
}
|
||||
}()
|
||||
|
||||
if w.healthCheckCfg != nil {
|
||||
w.startHealthCheckInterval(ctx)
|
||||
// no delay for the first health check after a target is registered
|
||||
w.healthCheckInterval.FireNow()
|
||||
}
|
||||
|
||||
// for simplicity, the worker runs a single-threaded loop and everything the
|
||||
// worker does is synchronized through channels, so it only needs to use its
|
||||
// mutex to guard target health and resource exposed in its getter methods
|
||||
for {
|
||||
select {
|
||||
case <-w.nextHealthCheck():
|
||||
w.checkHealth(ctx)
|
||||
if w.target.onHealthCheck != nil {
|
||||
w.target.onHealthCheck(w.lastResultErr)
|
||||
}
|
||||
case newCfg := <-w.healthCheckConfigUpdateCh:
|
||||
w.updateHealthCheckConfig(ctx, newCfg)
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// startHealthCheckInterval starts the health check interval.
|
||||
func (w *worker) startHealthCheckInterval(ctx context.Context) {
|
||||
w.log.InfoContext(ctx, "Health checker started",
|
||||
"health_check_config", w.healthCheckCfg.name,
|
||||
"protocol", w.healthCheckCfg.protocol,
|
||||
"interval", w.healthCheckCfg.interval,
|
||||
"timeout", w.healthCheckCfg.timeout,
|
||||
"healthy_threshold", w.healthCheckCfg.healthyThreshold,
|
||||
"unhealthy_threshold", w.healthCheckCfg.unhealthyThreshold,
|
||||
)
|
||||
w.healthCheckInterval = interval.New(interval.Config{
|
||||
Duration: w.healthCheckCfg.interval,
|
||||
Jitter: retryutils.SeventhJitter,
|
||||
FirstDuration: retryutils.FullJitter(w.healthCheckCfg.interval),
|
||||
Clock: w.clock,
|
||||
})
|
||||
}
|
||||
|
||||
// stopHealthCheckInterval stops the health check interval.
|
||||
func (w *worker) stopHealthCheckInterval(ctx context.Context) {
|
||||
w.log.InfoContext(ctx, "Health checker stopped")
|
||||
w.lastResultErr = nil
|
||||
w.lastResultCount = 0
|
||||
w.lastResolvedEndpoints = nil
|
||||
w.healthCheckInterval.Stop()
|
||||
w.healthCheckInterval = nil
|
||||
}
|
||||
|
||||
// nextHealthCheck returns a receiver channel that fires for the next health
|
||||
// check. If health checks are currently disabled, then it returns a nil channel
|
||||
// that blocks forever.
|
||||
func (w *worker) nextHealthCheck() <-chan time.Time {
|
||||
if w.healthCheckInterval != nil {
|
||||
return w.healthCheckInterval.Next()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// checkHealth performs a single health check against resolved endpoints,
|
||||
// updates the worker's health check result history, and possibly updates the
|
||||
// target health.
|
||||
func (w *worker) checkHealth(ctx context.Context) {
|
||||
initializing := w.lastResultCount == 0
|
||||
dialErr := w.dialEndpoints(ctx)
|
||||
if ctx.Err() == context.Canceled {
|
||||
return
|
||||
}
|
||||
if (dialErr == nil) == (w.lastResultErr == nil) {
|
||||
w.lastResultCount++
|
||||
} else {
|
||||
// the passing/failing result streak has ended, so reset the count
|
||||
w.lastResultCount = 1
|
||||
}
|
||||
w.lastResultErr = dialErr
|
||||
|
||||
if w.lastResultErr != nil {
|
||||
w.log.DebugContext(ctx, "Failed health check",
|
||||
"error", w.lastResultErr,
|
||||
)
|
||||
}
|
||||
threshold := w.getThreshold(w.healthCheckCfg)
|
||||
switch {
|
||||
// when initializing the effective threshold is 1 regardless of configured
|
||||
// threshold, but if the configured threshold is 1 then there is no need for
|
||||
// that special behavior
|
||||
case !initializing || threshold == 1:
|
||||
// we don't want every result above the threshold to trigger a target
|
||||
// health update, so only update target health when we exactly reach the
|
||||
// threshold
|
||||
if threshold == w.lastResultCount {
|
||||
w.setThresholdReached(ctx)
|
||||
}
|
||||
case w.lastResultErr == nil:
|
||||
w.setTargetHealthy(ctx, types.TargetHealthTransitionReasonInit,
|
||||
"Passed the initial health check",
|
||||
)
|
||||
default:
|
||||
w.setTargetUnhealthy(ctx, types.TargetHealthTransitionReasonInit,
|
||||
"Failed the initial health check",
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// updateHealthCheckConfig updates the current health check config.
|
||||
func (w *worker) updateHealthCheckConfig(ctx context.Context, newCfg *healthCheckConfig) {
|
||||
oldCfg := w.healthCheckCfg
|
||||
w.healthCheckCfg = newCfg
|
||||
if newCfg.equivalent(oldCfg) {
|
||||
return
|
||||
}
|
||||
if w.target.onConfigUpdate != nil {
|
||||
defer w.target.onConfigUpdate()
|
||||
}
|
||||
switch {
|
||||
case newCfg == nil:
|
||||
w.setTargetDisabled(ctx)
|
||||
w.stopHealthCheckInterval(ctx)
|
||||
return
|
||||
case oldCfg == nil:
|
||||
w.startHealthCheckInterval(ctx)
|
||||
return
|
||||
}
|
||||
w.log.DebugContext(ctx, "Updated health check config",
|
||||
"health_check_config", w.healthCheckCfg.name,
|
||||
"protocol", w.healthCheckCfg.protocol,
|
||||
"interval", w.healthCheckCfg.interval,
|
||||
"timeout", w.healthCheckCfg.timeout,
|
||||
"healthy_threshold", w.healthCheckCfg.healthyThreshold,
|
||||
"unhealthy_threshold", w.healthCheckCfg.unhealthyThreshold,
|
||||
)
|
||||
if newCfg.interval != oldCfg.interval {
|
||||
// we don't want config updates that match multiple targets to align all
|
||||
// the interval timers too closely, so create a new interval with full
|
||||
// jitter rather than trying to account for elapsed time since last tick
|
||||
w.healthCheckInterval.Stop()
|
||||
w.healthCheckInterval = interval.New(interval.Config{
|
||||
Duration: w.healthCheckCfg.interval,
|
||||
Jitter: retryutils.SeventhJitter,
|
||||
FirstDuration: retryutils.FullJitter(w.healthCheckCfg.interval),
|
||||
Clock: w.clock,
|
||||
})
|
||||
}
|
||||
oldThreshold := w.getThreshold(oldCfg)
|
||||
newThreshold := w.getThreshold(newCfg)
|
||||
if newThreshold < oldThreshold && w.lastResultCount >= newThreshold {
|
||||
w.setThresholdReached(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *worker) dialEndpoints(ctx context.Context) error {
|
||||
ctx, cancel := context.WithTimeout(ctx, w.healthCheckCfg.timeout)
|
||||
defer cancel()
|
||||
endpoints, err := w.target.ResolverFn(ctx)
|
||||
if err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
w.lastResolvedEndpoints = endpoints
|
||||
switch len(endpoints) {
|
||||
case 0:
|
||||
return trace.NotFound("resolved zero target endpoints")
|
||||
case 1:
|
||||
return w.dialEndpoint(ctx, endpoints[0])
|
||||
default:
|
||||
group, ctx := errgroup.WithContext(ctx)
|
||||
group.SetLimit(10)
|
||||
for _, ep := range endpoints {
|
||||
group.Go(func() error {
|
||||
return trace.Wrap(w.dialEndpoint(ctx, ep))
|
||||
})
|
||||
}
|
||||
return group.Wait()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *worker) dialEndpoint(ctx context.Context, endpoint string) error {
|
||||
conn, err := w.target.dialFn(ctx, "tcp", endpoint)
|
||||
if err != nil {
|
||||
return trace.Wrap(err)
|
||||
}
|
||||
// an error while closing the connection could indicate an RST packet from
|
||||
// the endpoint - that's a health check failure.
|
||||
return trace.Wrap(conn.Close())
|
||||
}
|
||||
|
||||
// getThreshold returns the appropriate threshold to compare against the last
|
||||
// result.
|
||||
func (w *worker) getThreshold(cfg *healthCheckConfig) uint32 {
|
||||
if w.lastResultErr == nil {
|
||||
return cfg.healthyThreshold
|
||||
}
|
||||
return cfg.unhealthyThreshold
|
||||
}
|
||||
|
||||
func (w *worker) setThresholdReached(ctx context.Context) {
|
||||
const transitionReason = types.TargetHealthTransitionReasonThreshold
|
||||
if w.lastResultErr == nil {
|
||||
msg := fmt.Sprintf("%d consecutive health checks passed", w.lastResultCount)
|
||||
w.setTargetHealthy(ctx, transitionReason, msg)
|
||||
} else {
|
||||
msg := fmt.Sprintf("%d consecutive health checks failed", w.lastResultCount)
|
||||
w.setTargetUnhealthy(ctx, transitionReason, msg)
|
||||
}
|
||||
}
|
||||
|
||||
func (w *worker) setTargetHealthy(ctx context.Context, reason types.TargetHealthTransitionReason, message string) {
|
||||
w.log.InfoContext(ctx, "Target became healthy",
|
||||
"reason", reason,
|
||||
"message", message,
|
||||
)
|
||||
w.setTargetHealthStatus(types.TargetHealthStatusHealthy, reason, message)
|
||||
}
|
||||
|
||||
func (w *worker) setTargetUnhealthy(ctx context.Context, reason types.TargetHealthTransitionReason, message string) {
|
||||
w.log.WarnContext(ctx, "Target became unhealthy",
|
||||
"reason", reason,
|
||||
"message", message,
|
||||
)
|
||||
w.setTargetHealthStatus(types.TargetHealthStatusUnhealthy, reason, message)
|
||||
}
|
||||
|
||||
func (w *worker) setTargetDisabled(ctx context.Context) {
|
||||
const reason = types.TargetHealthTransitionReasonDisabled
|
||||
const message = "No health check config matches this resource"
|
||||
w.log.InfoContext(ctx, "Target health checks are disabled",
|
||||
"message", message,
|
||||
)
|
||||
w.setTargetHealthStatus(types.TargetHealthStatusUnknown, reason, message)
|
||||
}
|
||||
|
||||
func (w *worker) setTargetHealthStatus(newStatus types.TargetHealthStatus, reason types.TargetHealthTransitionReason, message string) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
oldHealth := w.targetHealth
|
||||
if oldHealth.Status == string(newStatus) && oldHealth.TransitionReason == string(reason) {
|
||||
return
|
||||
}
|
||||
now := w.clock.Now()
|
||||
w.targetHealth = types.TargetHealth{
|
||||
Address: strings.Join(w.lastResolvedEndpoints, ","),
|
||||
Status: string(newStatus),
|
||||
TransitionTimestamp: &now,
|
||||
TransitionReason: string(reason),
|
||||
Message: message,
|
||||
}
|
||||
if w.lastResultErr != nil {
|
||||
w.targetHealth.TransitionError = w.lastResultErr.Error()
|
||||
}
|
||||
if w.healthCheckCfg != nil {
|
||||
w.targetHealth.Protocol = string(w.healthCheckCfg.protocol)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2025 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package healthcheck
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"net"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/gravitational/teleport/api/types"
|
||||
"github.com/gravitational/teleport/lib/defaults"
|
||||
)
|
||||
|
||||
func Test_newWorker(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
listener, err := net.Listen("tcp", "localhost:0")
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { _ = listener.Close() })
|
||||
|
||||
db, err := types.NewDatabaseV3(types.Metadata{
|
||||
Name: "example-pg",
|
||||
Labels: map[string]string{
|
||||
"target": "true",
|
||||
},
|
||||
}, types.DatabaseSpecV3{
|
||||
Protocol: defaults.ProtocolPostgres,
|
||||
URI: listener.Addr().String(),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
tests := []struct {
|
||||
desc string
|
||||
cfg workerConfig
|
||||
wantHealth types.TargetHealth
|
||||
}{
|
||||
{
|
||||
desc: "disabled",
|
||||
cfg: workerConfig{
|
||||
HealthCheckCfg: nil,
|
||||
Target: Target{
|
||||
GetResource: func() types.ResourceWithLabels { return db },
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{db.GetURI()}, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
wantHealth: types.TargetHealth{
|
||||
Address: "",
|
||||
Protocol: "",
|
||||
Status: string(types.TargetHealthStatusUnknown),
|
||||
TransitionReason: string(types.TargetHealthTransitionReasonDisabled),
|
||||
Message: "No health check config matches this resource",
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "enabled",
|
||||
cfg: workerConfig{
|
||||
HealthCheckCfg: &healthCheckConfig{
|
||||
interval: time.Minute,
|
||||
timeout: time.Minute,
|
||||
healthyThreshold: 10,
|
||||
unhealthyThreshold: 10,
|
||||
},
|
||||
Target: Target{
|
||||
GetResource: func() types.ResourceWithLabels { return db },
|
||||
ResolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{db.GetURI()}, nil
|
||||
},
|
||||
},
|
||||
},
|
||||
wantHealth: types.TargetHealth{
|
||||
Address: "",
|
||||
Protocol: "",
|
||||
Status: string(types.TargetHealthStatusUnknown),
|
||||
TransitionReason: string(types.TargetHealthTransitionReasonInit),
|
||||
Message: "Health checker initialized",
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
w, err := newWorker(ctx, test.cfg)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, w.Close()) })
|
||||
require.Empty(t, cmp.Diff(test.wantHealth, *w.GetTargetHealth(),
|
||||
cmpopts.IgnoreFields(types.TargetHealth{}, "TransitionTimestamp")),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_dialEndpoints(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(cancel)
|
||||
|
||||
const healthyAddr = "healthy.com:123"
|
||||
const unhealthyAddr = "unhealthy.com:123"
|
||||
tests := []struct {
|
||||
desc string
|
||||
resolverFn EndpointsResolverFunc
|
||||
wantErrContains string
|
||||
}{
|
||||
{
|
||||
desc: "resolver error",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return nil, trace.Errorf("resolver error")
|
||||
},
|
||||
wantErrContains: "resolver error",
|
||||
},
|
||||
{
|
||||
desc: "resolved zero addrs",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return nil, nil
|
||||
},
|
||||
wantErrContains: "resolved zero target endpoints",
|
||||
},
|
||||
{
|
||||
desc: "resolved one healthy addr",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{healthyAddr}, nil
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "resolved one unhealthy addr",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{unhealthyAddr}, nil
|
||||
},
|
||||
wantErrContains: "unhealthy addr",
|
||||
},
|
||||
{
|
||||
desc: "resolved multiple healthy addrs",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{healthyAddr, healthyAddr, healthyAddr}, nil
|
||||
},
|
||||
},
|
||||
{
|
||||
desc: "resolved a mix of healthy and unhealthy addrs",
|
||||
resolverFn: func(ctx context.Context) ([]string, error) {
|
||||
return []string{healthyAddr, unhealthyAddr, healthyAddr}, nil
|
||||
},
|
||||
wantErrContains: "unhealthy addr",
|
||||
},
|
||||
}
|
||||
for _, test := range tests {
|
||||
t.Run(test.desc, func(t *testing.T) {
|
||||
w := &worker{
|
||||
healthCheckCfg: &healthCheckConfig{},
|
||||
log: slog.Default(),
|
||||
target: Target{
|
||||
ResolverFn: test.resolverFn,
|
||||
dialFn: func(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
if addr == healthyAddr {
|
||||
return fakeConn{}, nil
|
||||
}
|
||||
return nil, trace.Errorf("unhealthy addr")
|
||||
},
|
||||
},
|
||||
}
|
||||
err := w.dialEndpoints(ctx)
|
||||
if test.wantErrContains != "" {
|
||||
require.ErrorContains(t, err, test.wantErrContains)
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type fakeConn struct {
|
||||
net.Conn
|
||||
}
|
||||
|
||||
func (fakeConn) Close() error { return nil }
|
||||
@@ -68,7 +68,7 @@ The side panel content may look something like this (or equivalent singular form
|
||||
UUID: <uuid>
|
||||
Error: <last health check error message>
|
||||
|
||||
Databases with a mix of `healthy` and `""` or `disabled` status will also show a warning icon.
|
||||
Databases with a mix of `healthy` and `""` or `unknown` status will also show a warning icon.
|
||||
The side panel will encourage the user to enable health checks or update to a supported version on all of the agents that proxy the db.
|
||||
The side panel content may look something like this:
|
||||
|
||||
@@ -245,30 +245,28 @@ The `target_health` will include, among other things, a `status` field.
|
||||
|
||||
These are the possible values for the `db_server.status.target_health.status` field:
|
||||
- `""`
|
||||
- `disabled`
|
||||
- `initialized`
|
||||
- `unknown`
|
||||
- `healthy`
|
||||
- `unhealthy`
|
||||
|
||||
An empty status `""` indicates an unknown status.
|
||||
If an older agent is proxying the DB, then the health status will be empty.
|
||||
|
||||
The `disabled` status indicates that health checks are disabled for this database.
|
||||
|
||||
The `initialized` status is the initial health status and means that the health checking has started but the status has yet to be determined as `healthy` or `unhealthy`.
|
||||
The `unknown` status indicates that health checks are disabled, still initializing, or unsupported for this database type.
|
||||
|
||||
The `healthy` status is reported after the number of consecutive passing health checks has reached a healthy threshold.
|
||||
|
||||
The `unhealthy` status is reported after the number of consecutive failing health checks has reached an unhealthy threshold.
|
||||
|
||||
As a special case when the status is `initialized`, the first health check will change the status to `healthy` or `unhealthy` regardless of the configured thresholds.
|
||||
This special case is to bound the amount of time spent in the `initialized` status if health checks flap between pass/fail without reaching either threshold.
|
||||
As a special case the first health check will change the status to `healthy` or `unhealthy` regardless of the configured thresholds.
|
||||
This special case is to bound the amount of time spent in the `unknown` status if health checks flap between pass/fail without reaching either threshold.
|
||||
|
||||
### Types of health checks
|
||||
|
||||
For the initial implementation we will only support TCP health checks and only for databases.
|
||||
|
||||
A TCP health check is generic and simple, so the initial implementation will support health checks for all databases that Teleport supports: Postgres, MySQL, MongoDB, etc.
|
||||
The initial implementation will support health checks for Postgres, MySQL, and MongoDB databases.
|
||||
We will gradually expand support to other databases.
|
||||
|
||||
We can extend this feature later to include additional health check protocols such as HTTP and TLS checks.
|
||||
|
||||
@@ -375,7 +373,7 @@ As a brief summary of the Teleport health checker behavior:
|
||||
3. it repeats the check after interval time
|
||||
4. each check blocks until pass or fail
|
||||
|
||||
For Teleport the minimum time initialized->healthy or initialized->unhealthy is roughly 0s.
|
||||
For Teleport the minimum time unknown->healthy or unknown->unhealthy is roughly 0s.
|
||||
|
||||
The maximum time is bounded by the timeout: 5s.
|
||||
|
||||
@@ -492,29 +490,9 @@ Each health check should lookup the SRV record and check each endpoint returned.
|
||||
|
||||
### Health checker behavior
|
||||
|
||||
A health checker manager will be added that manages a mapping from DB name to a single health checker:
|
||||
A health checker manager will be added that manages registered resource health check targets.
|
||||
|
||||
// Target is a health check target.
|
||||
type Target struct {
|
||||
// Resource is the target resource.
|
||||
Resource types.ResourceWithLabels
|
||||
// ResolveEndpointsFn is callback func that returns the target endpoints.
|
||||
ResolveEndpointsFn TargetEndpointsResolverFunc
|
||||
}
|
||||
|
||||
// Manager manages health checkers.
|
||||
type Manager interface {
|
||||
// AddTarget adds a new target health checker and starts the health checker.
|
||||
AddTarget(ctx context.Context, target Target) error
|
||||
// GetTargetHealth returns the health of a given target.
|
||||
GetTargetHealth(name string) (types.TargetHealth, error)
|
||||
// RemoveTarget removes a given target and stops its health checker.
|
||||
RemoveTarget(name string) error
|
||||
// Close stops all health checkers.
|
||||
Close()
|
||||
}
|
||||
|
||||
When a health checker starts, it will set its health status to `initialized` and immediately run its first health check.
|
||||
When a health check target is added, it will set its initial health status to `unknown` with an `initialized` transition reason, then health checker starts it will immediately run its first health check.
|
||||
It will then wait for an interval of time before checking again.
|
||||
|
||||
The health check interval will be jittered to avoid correlated health check times.
|
||||
@@ -538,11 +516,11 @@ Likewise, if the last check failed but the current check passes, then the counte
|
||||
|
||||
If the number of consecutive passing or failing checks reaches the healthy or unhealthy threshold, respectively, then the health status is transitioned to the corresponding status: `healthy` or `unhealthy`.
|
||||
|
||||
When the status is `initialized`, a single pass or failure with transition the status to `healthy` or `unhealthy`, regardless of configured thresholds.
|
||||
When the status is `unknown`, a single pass or failure with transition the status to `healthy` or `unhealthy`, regardless of configured thresholds.
|
||||
|
||||
Example behavior with the default settings (Interval=30s, Timeout=5s, HealthyThreshold=2, UnhealthyThreshold=1):
|
||||
|
||||
1. health checker starts - status=`initialized`, startTime=0s, endTime=0s, count=0, lastErr=nil
|
||||
1. health checker starts - status=`unknown`, startTime=0s, endTime=0s, count=0, lastErr=nil
|
||||
2. health check passes - status=`healthy`, startTime=0s, endTime=0s, count=1, lastErr=nil
|
||||
3. health check fails - status=`unhealthy`, startTime=30s, endTime=35s, count=1, lastErr="connection timeout"
|
||||
4. health check passes - status=`unhealthy`, startTime=60s, endTime=60s, count=1, lastErr=nil
|
||||
@@ -556,7 +534,7 @@ That will be changed to group the agents by health status, shuffle each group, a
|
||||
|
||||
The priority of health statuses will be:
|
||||
1. `healthy`
|
||||
2. `initialized`, `""` (unknown), and `disabled`
|
||||
2. `unknown`, `""` (also unknown, older agent)
|
||||
3. `unhealthy`
|
||||
|
||||
The proxy routing prioritization strategy is a fail-open system: if all databases are `unhealthy`, users will still be able to attempt a connection through them, even though their attempt will likely fail.
|
||||
@@ -631,7 +609,7 @@ TargetHealth will be added as a field to the database heartbeat: `db_server.stat
|
||||
string Address = 1 [(gogoproto.jsontag) = "address,omitempty"];
|
||||
// Protocol is the health check protocol such as "tcp".
|
||||
string Protocol = 2 [(gogoproto.jsontag) = "protocol,omitempty"];
|
||||
// Status is the health status, one of "", "init", "healthy", "unhealthy".
|
||||
// Status is the health status, one of "", "unknown", "healthy", "unhealthy".
|
||||
string Status = 3 [(gogoproto.jsontag) = "status,omitempty"];
|
||||
// TransitionTimestamp is the time that the last status transition occurred.
|
||||
google.protobuf.Timestamp TransitionTimestamp = 4 [
|
||||
@@ -652,7 +630,7 @@ TargetHealth will be added as a field to the database heartbeat: `db_server.stat
|
||||
|
||||
Older database agents will not perform health checks, so they will always report health status `""` (unknown), which is equivalent to agents that have disabled health checks.
|
||||
|
||||
An `""` (unknown) or `disabled` health status will not prevent connections to that database through that agent, but it will prioritize `healthy` agents first.
|
||||
An `""` or `unknown` health status will not prevent connections to that database through that agent, but it will prioritize `healthy` agents first.
|
||||
|
||||
### Audit Events
|
||||
|
||||
|
||||
Reference in New Issue
Block a user