diff --git a/api/proto/teleport/legacy/types/types.proto b/api/proto/teleport/legacy/types/types.proto index 8dc2b0444a9..bebfe989b1c 100644 --- a/api/proto/teleport/legacy/types/types.proto +++ b/api/proto/teleport/legacy/types/types.proto @@ -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 [ diff --git a/api/types/target_health.go b/api/types/target_health.go new file mode 100644 index 00000000000..193a233b1f2 --- /dev/null +++ b/api/types/target_health.go @@ -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" +) diff --git a/api/types/types.pb.go b/api/types/types.pb.go index e7707f0e73d..d5cb0b41345 100644 --- a/api/types/types.pb.go +++ b/api/types/types.pb.go @@ -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"` diff --git a/lib/healthcheck/config.go b/lib/healthcheck/config.go new file mode 100644 index 00000000000..6f47966631e --- /dev/null +++ b/lib/healthcheck/config.go @@ -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 . + */ + +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 +} diff --git a/lib/healthcheck/config_test.go b/lib/healthcheck/config_test.go new file mode 100644 index 00000000000..8a1357ee061 --- /dev/null +++ b/lib/healthcheck/config_test.go @@ -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 . + */ + +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)) + }) + } +} diff --git a/lib/healthcheck/manager.go b/lib/healthcheck/manager.go new file mode 100644 index 00000000000..3d7a560ca70 --- /dev/null +++ b/lib/healthcheck/manager.go @@ -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 . + */ + +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 +} diff --git a/lib/healthcheck/manager_test.go b/lib/healthcheck/manager_test.go new file mode 100644 index 00000000000..3f593575a65 --- /dev/null +++ b/lib/healthcheck/manager_test.go @@ -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 . + */ + +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, + } +} diff --git a/lib/healthcheck/target.go b/lib/healthcheck/target.go new file mode 100644 index 00000000000..c6e43623f4b --- /dev/null +++ b/lib/healthcheck/target.go @@ -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 . + */ + +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{} +} diff --git a/lib/healthcheck/target_test.go b/lib/healthcheck/target_test.go new file mode 100644 index 00000000000..0d4c3c5ec4b --- /dev/null +++ b/lib/healthcheck/target_test.go @@ -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 . + */ + +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) + } + }) + } +} diff --git a/lib/healthcheck/worker.go b/lib/healthcheck/worker.go new file mode 100644 index 00000000000..65dbaeab534 --- /dev/null +++ b/lib/healthcheck/worker.go @@ -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 . + */ + +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) + } +} diff --git a/lib/healthcheck/worker_test.go b/lib/healthcheck/worker_test.go new file mode 100644 index 00000000000..11b5d7d17c5 --- /dev/null +++ b/lib/healthcheck/worker_test.go @@ -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 . + */ + +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 } diff --git a/rfd/0203-database-healthchecks.md b/rfd/0203-database-healthchecks.md index 23d1c4e55cf..fa0896ddc31 100644 --- a/rfd/0203-database-healthchecks.md +++ b/rfd/0203-database-healthchecks.md @@ -68,7 +68,7 @@ The side panel content may look something like this (or equivalent singular form UUID: Error: -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