Add monitoring of the servers time drifting (#44489)

* Add monitoring of the servers time drifting * Refactor monitor to use inventory stream * Add command for inventory control stream for time reconciliation * Reuse ping/pong logic for time reconciliation * Add integration test to check if global notification is created * Limits notification length, renaming * Return system clock instead of time difference Add parallel requests to ping the nodes Fix issue with locking inventory store while making ping request * Refactor to make time reconciliation part of the inventory control stream * Fix tests * Rewrite tests Return inventory store iterator * Naming adjustments * Fix test * Drop timeReconciliation in order to use direct handler functions * Make time reconciliation with variable duration * Replace with duration type in proto Add test cleanup for possible goroutine leak Add comment about half request duration * Format notification message * Add UpstreamInventoryPong.SystemClock check
2026-09-19 01:58:44 +08:00 · 2024-11-28 21:18:24 +00:00
parent d521af2296
commit af6a6ea734
13 changed files with 4398 additions and 3501 deletions
@@ -2349,16 +2349,21 @@ message DownstreamInventoryOneOf {
  }
 }

-// DownstreamInventoryPing is sent down the inventory control stream for testing/debug
-// purposes.
+// DownstreamInventoryPing is sent down the inventory control stream.
 message DownstreamInventoryPing {
  uint64 ID = 1;
 }

 // UpstreamInventoryPong is sent up the inventory control stream in response to a downstream
-// ping (used for testing/debug purposes).
+// ping including the system clock of the downstream.
 message UpstreamInventoryPong {
  uint64 ID = 1;
+  // SystemClock advertises the system clock of the upstream.
+  google.protobuf.Timestamp SystemClock = 2 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "system_clock,omitempty"
+  ];
 }

 // UpstreamInventoryHello is the hello message sent up the inventory control stream.
@@ -711,6 +711,31 @@ message InstanceSpecV1 {

  // ExternalUpgraderVersion identifies the external upgrader version. Empty if no upgrader is defined.
  string ExternalUpgraderVersion = 8 [(gogoproto.jsontag) = "ext_upgrader_version,omitempty"];
+
+  // LastMeasurement stores information about the latest measurement between services.
+  SystemClockMeasurement LastMeasurement = 9;
+}
+
+// SystemClockMeasurement represents the measurement state of the systems clock difference.
+message SystemClockMeasurement {
+  // ControllerSystemClock is the system clock of the inventory controller.
+  google.protobuf.Timestamp ControllerSystemClock = 1 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "controller_system_clock,omitempty"
+  ];
+  // SystemClock is the system clock of the upstream.
+  google.protobuf.Timestamp SystemClock = 2 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false,
+    (gogoproto.jsontag) = "system_clock,omitempty"
+  ];
+  // RequestDuration stores information about the request duration between auth and remote service.
+  google.protobuf.Duration RequestDuration = 3 [
+    (gogoproto.jsontag) = "request_duration",
+    (gogoproto.nullable) = false,
+    (gogoproto.stdduration) = true
+  ];
 }

 // InstanceControlLogEntry represents an entry in a given instance's control log. The control log of
@@ -179,6 +179,10 @@ type Instance interface {
 	// so appends do not need to be performed in any particular order.
 	AppendControlLog(entries ...InstanceControlLogEntry)

+	// GetLastMeasurement returns information about the system clocks of the auth service and
+	// another instance.
+	GetLastMeasurement() *SystemClockMeasurement
+
 	// Clone performs a deep copy on this instance.
 	Clone() Instance
 }
@@ -299,6 +303,10 @@ func (i *InstanceV1) AppendControlLog(entries ...InstanceControlLogEntry) {
 	})
 }

+func (i *InstanceV1) GetLastMeasurement() *SystemClockMeasurement {
+	return i.Spec.LastMeasurement
+}
+
 // expireControlLog removes expired entries from the control log relative to the supplied
 // "now" value. The supplied ttl is used as the default ttl for entries that do not specify
 // a custom ttl value. The returned timestamp is the observed expiry that was furthest in
@@ -474,6 +474,38 @@ func MakeTestDatabaseServer(t *testing.T, proxyAddr utils.NetAddr, token string,
 	return db
 }

+// MakeAgentServer creates SSH agent Service
+// It receives the Proxy Address, a Token (to join the cluster).
+func MakeAgentServer(t *testing.T, cfg *servicecfg.Config, proxyAddr utils.NetAddr, token string) *service.TeleportProcess {
+	// Proxy uses self-signed certificates in tests.
+	lib.SetInsecureDevMode(true)
+
+	cfg.Hostname = "localhost"
+	cfg.DataDir = t.TempDir()
+	cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig()
+	cfg.InstanceMetadataClient = imds.NewDisabledIMDSClient()
+	cfg.SetAuthServerAddress(proxyAddr)
+	cfg.SetToken(token)
+	cfg.SSH.Enabled = true
+	cfg.Auth.Enabled = false
+	cfg.Proxy.Enabled = false
+	cfg.Databases.Enabled = false
+	cfg.Log = utils.NewLoggerForTests()
+
+	agent, err := service.NewTeleport(cfg)
+	require.NoError(t, err)
+	require.NoError(t, agent.Start())
+
+	t.Cleanup(func() {
+		assert.NoError(t, agent.Close())
+	})
+
+	_, err = agent.WaitForEventTimeout(30*time.Second, service.NodeSSHReady)
+	require.NoError(t, err, "agent server didn't start after 10s")
+
+	return agent
+}
+
 // MustCreateListener creates a tcp listener at 127.0.0.1 with random port.
 func MustCreateListener(t *testing.T) net.Listener {
 	t.Helper()
@@ -0,0 +1,75 @@
+/*
+ * Teleport
+ * Copyright (C) 2024  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 integration
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+
+	"github.com/gravitational/teleport/api/utils/retryutils"
+	"github.com/gravitational/teleport/integration/helpers"
+	"github.com/gravitational/teleport/lib"
+	"github.com/gravitational/teleport/lib/service/servicecfg"
+)
+
+// TestTimeReconciliation launches two instances with clock differences in system clock,
+// to verify that global notification is created about time drifting.
+func TestTimeReconciliation(t *testing.T) {
+	lib.SetInsecureDevMode(true)
+	defer lib.SetInsecureDevMode(false)
+	helpers.SetTestTimeouts(2 * time.Second)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Start Teleport Auth and Proxy services
+	authProcess, proxyProcess, provisionToken := helpers.MakeTestServers(t)
+	authService := authProcess.GetAuthServer()
+	proxyAddr, err := proxyProcess.ProxyWebAddr()
+	require.NoError(t, err)
+
+	cfg := servicecfg.MakeDefaultConfig()
+	agentClock := clockwork.NewFakeClockAt(time.Now().Add(24 * time.Hour))
+	cfg.Clock = agentClock
+	agent := helpers.MakeAgentServer(t, cfg, *proxyAddr, provisionToken)
+	require.NotNil(t, agent)
+
+	err = retryutils.RetryStaticFor(30*time.Second, time.Second, func() error {
+		agentClock.Advance(time.Minute)
+		notifications, _, err := authService.ListGlobalNotifications(ctx, 100, "")
+		if err != nil {
+			return trace.Wrap(err)
+		}
+
+		var found bool
+		for _, notification := range notifications {
+			found = found || notification.GetMetadata().GetName() == "cluster-monitor-system-clock-warning"
+		}
+		if !found {
+			return trace.BadParameter("expected notification is not found")
+		}
+		return nil
+	})
+	require.NoError(t, err)
+}
@@ -525,6 +525,7 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 	}
 	as.inventory = inventory.NewController(&as, services,
 		inventory.WithAuthServerID(cfg.HostUUID),
+		inventory.WithClock(cfg.Clock),
 		inventory.WithOnConnect(func(s string) {
 			if g, ok := connectedResourceGauges[s]; ok {
 				g.Inc()
@@ -0,0 +1,175 @@
+/*
+ * Teleport
+ * Copyright (C) 2024  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 auth
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"strings"
+	"time"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
+	notificationsv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/notifications/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/inventory"
+	"github.com/gravitational/teleport/lib/services"
+	"github.com/gravitational/teleport/lib/utils/interval"
+)
+
+const (
+	// systemClockCheckCycle is the period when system clock comparison is launched
+	// across all inventories, to be gathered for global notifications, if any.
+	systemClockCheckCycle = 10 * time.Minute
+	// systemClockThreshold is the duration threshold for triggering a warning
+	// if the time difference exceeds this threshold.
+	systemClockThreshold = time.Minute
+	// systemClockNotificationWarningName is the ID for adding the global notification.
+	systemClockNotificationWarningName = "cluster-monitor-system-clock-warning"
+	// systemClockNotificationExpiration is the expiration time for global notification
+	// warning about time difference.
+	systemClockNotificationExpiration = time.Hour * 24 * 30
+	// systemClockMessagesLimit is limit for showing the list of affected inventories.
+	systemClockMessagesLimit = 10
+)
+
+// MonitorSystemTime runs the periodic check for iterating through all inventories
+// to ping them and receive the system clock difference.
+func (a *Server) MonitorSystemTime(ctx context.Context) error {
+	checkInterval := interval.New(interval.Config{
+		FirstDuration: time.Second * 10,
+		Duration:      systemClockCheckCycle,
+		Clock:         a.GetClock(),
+	})
+	defer checkInterval.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-checkInterval.Next():
+			a.checkInventorySystemClocks(ctx)
+		}
+	}
+}
+
+// checkInventoryClocks iterates through inventory store instance state to gather
+// information about the system clock differences.
+func (a *Server) checkInventorySystemClocks(ctx context.Context) {
+	var counter int
+	var messages []string
+	a.inventory.Iter(func(handle inventory.UpstreamHandle) {
+		counter++
+		if counter >= systemClockMessagesLimit {
+			return
+		}
+
+		hello := handle.Hello()
+		handle.VisitInstanceState(func(ref inventory.InstanceStateRef) (update inventory.InstanceStateUpdate) {
+			if ref.LastHeartbeat != nil && ref.LastHeartbeat.GetLastMeasurement() != nil {
+				m := ref.LastHeartbeat.GetLastMeasurement()
+				// RequestDuration is request and response duration between upstream and downstream,
+				// since we capture system clock on downstream we have to ignore response duration
+				// and only count request duration.
+				diff := m.ControllerSystemClock.Sub(m.SystemClock) - m.RequestDuration/2
+				if diff > systemClockThreshold || -diff > systemClockThreshold {
+					slog.WarnContext(ctx, "server time difference detected",
+						"server", hello.GetServerID(),
+						"services", hello.GetServices(),
+						"difference", durationText(diff),
+					)
+					messages = append(messages, fmt.Sprintf(
+						"%s[%s] is %s",
+						hello.GetServerID(),
+						types.SystemRoles(hello.GetServices()).String(),
+						durationText(diff),
+					))
+				}
+			}
+			return
+		})
+	})
+
+	if len(messages) > 0 {
+		title, text := generateClockWarningNotificationMessage(messages, counter)
+		err := upsertClockWarningGlobalNotification(ctx, a.Services, title, text)
+		if err != nil {
+			slog.ErrorContext(ctx, "can't set notification about system clock issue", "error", err)
+		}
+	}
+}
+
+// upsertClockWarningGlobalNotification sets predefined global notification for notifying the issues with the cluster
+// servers related to the system clock difference in nodes.
+func upsertClockWarningGlobalNotification(ctx context.Context, notification services.Notifications, title, text string) error {
+	now := time.Now()
+	_, err := notification.UpsertGlobalNotification(ctx, &notificationsv1.GlobalNotification{
+		Kind:     types.KindGlobalNotification,
+		Version:  types.V1,
+		Metadata: &headerv1.Metadata{Name: systemClockNotificationWarningName},
+		Spec: &notificationsv1.GlobalNotificationSpec{
+			Matcher: &notificationsv1.GlobalNotificationSpec_All{
+				All: true,
+			},
+			Notification: &notificationsv1.Notification{
+				SubKind: types.NotificationDefaultWarningSubKind,
+				Spec: &notificationsv1.NotificationSpec{
+					Created: timestamppb.New(now),
+				},
+				Metadata: &headerv1.Metadata{
+					Expires: timestamppb.New(now.Add(systemClockNotificationExpiration)),
+					Name:    systemClockNotificationWarningName,
+					Labels: map[string]string{
+						types.NotificationTitleLabel:       title,
+						types.NotificationTextContentLabel: text,
+					},
+				},
+			},
+		},
+	})
+	return trace.Wrap(err)
+}
+
+// generateClockWarningNotificationMessage formats the notification message with the inventory list.
+func generateClockWarningNotificationMessage(messages []string, total int) (string, string) {
+	title := "Incorrect system clock detected in the cluster"
+	text := "Incorrect system clock may lead to certificate validation issues.\n" +
+		"Ensure that the clock is accurate on all nodes to avoid potential access problems.\n" +
+		"All comparisons are made with the Auth service system clock.\n" +
+		"List of servers with a time drift: \n" + strings.Join(messages, ", ")
+
+	if total > len(messages) {
+		text += fmt.Sprintf("(%d in total)", total)
+	}
+
+	return title, text
+}
+
+// durationText formats the specified duration to text by adding the suffix "ahead" or "behind"
+// and converts nanoseconds to a formatted text with hours, minutes and seconds.
+func durationText(duration time.Duration) string {
+	if duration > 0 {
+		return fmt.Sprintf("%s ahead", duration.String())
+	} else {
+		return fmt.Sprintf("%s behind", (-duration).String())
+	}
+}
@@ -20,11 +20,13 @@ package inventory

 import (
 	"context"
+	"math/rand/v2"
 	"os"
 	"strings"
 	"time"

 	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 	log "github.com/sirupsen/logrus"

 	"github.com/gravitational/teleport/api/client"
@@ -102,6 +104,8 @@ const (
 	instanceHeartbeatOk  testEvent = "instance-heartbeat-ok"
 	instanceHeartbeatErr testEvent = "instance-heartbeat-err"

+	timeReconciliationOk testEvent = "time-reconciliation-ok"
+
 	instanceCompareFailed testEvent = "instance-compare-failed"

 	handlerStart = "handler-start"
@@ -113,7 +117,7 @@ const (
 	keepAliveKubeTick     = "keep-alive-kube-tick"
 )

-// instanceHBStepSize is the step size used for the variable instance hearbteat duration. This value is
+// instanceHBStepSize is the step size used for the variable instance heartbeat duration. This value is
 // basically arbitrary. It was selected because it produces a scaling curve that makes a fairly reasonable
 // tradeoff between heartbeat availability and load scaling. See test coverage in the 'interval' package
 // for a demonstration of the relationship between step sizes and interval/duration scaling.
@@ -127,6 +131,7 @@ type controllerOptions struct {
 	authID             string
 	onConnectFunc      func(string)
 	onDisconnectFunc   func(string, int)
+	clock              clockwork.Clock
 }

 func (options *controllerOptions) SetDefaults() {
@@ -154,6 +159,10 @@ func (options *controllerOptions) SetDefaults() {
 	if options.onDisconnectFunc == nil {
 		options.onDisconnectFunc = func(string, int) {}
 	}
+
+	if options.clock == nil {
+		options.clock = clockwork.NewRealClock()
+	}
 }

 type ControllerOption func(c *controllerOptions)
@@ -204,6 +213,13 @@ func withTestEventsChannel(ch chan testEvent) ControllerOption {
 	}
 }

+// WithClock sets the clock for the controller to have a general clock configuration.
+func WithClock(clock clockwork.Clock) ControllerOption {
+	return func(opts *controllerOptions) {
+		opts.clock = clock
+	}
+}
+
 // Controller manages the inventory control streams registered with a given auth instance. Incoming
 // messages are processed by invoking the appropriate methods on the Auth interface.
 type Controller struct {
@@ -221,6 +237,7 @@ type Controller struct {
 	testEvents                 chan testEvent
 	onConnectFunc              func(string)
 	onDisconnectFunc           func(string, int)
+	clock                      clockwork.Clock
 	closeContext               context.Context
 	cancel                     context.CancelFunc
 }
@@ -255,6 +272,7 @@ func NewController(auth Auth, usageReporter usagereporter.UsageReporter, opts ..
 		usageReporter:              usageReporter,
 		onConnectFunc:              options.onConnectFunc,
 		onDisconnectFunc:           options.onDisconnectFunc,
+		clock:                      options.clock,
 		closeContext:               ctx,
 		cancel:                     cancel,
 	}
@@ -322,12 +340,19 @@ func (c *Controller) handleControlStream(handle *upstreamHandle) {
 	// mitigate load spikes on auth restart, and is reasonably safe to do since
 	// the instance resource is not directly relied upon for use of any
 	// particular Teleport service.
+	firstDuration := retryutils.FullJitter(c.instanceHBVariableDuration.Duration())
 	instanceHeartbeatDelay := delay.New(delay.Params{
-		FirstInterval:    retryutils.FullJitter(c.instanceHBVariableDuration.Duration()),
+		FirstInterval:    firstDuration,
 		VariableInterval: c.instanceHBVariableDuration,
 		Jitter:           retryutils.SeventhJitter,
 	})
 	defer instanceHeartbeatDelay.Stop()
+	timeReconciliationDelay := delay.New(delay.Params{
+		FirstInterval:    firstDuration / 2,
+		VariableInterval: c.instanceHBVariableDuration,
+		Jitter:           retryutils.SeventhJitter,
+	})
+	defer timeReconciliationDelay.Stop()

 	// these delays are lazily initialized upon receipt of the first heartbeat
 	// since not all servers send all heartbeats
@@ -483,6 +508,18 @@ func (c *Controller) handleControlStream(handle *upstreamHandle) {
 			}
 			c.testEvent(keepAliveAppTick)

+		case now := <-timeReconciliationDelay.Elapsed():
+			timeReconciliationDelay.Advance(now)
+
+			if err := c.handlePingRequest(handle, pingRequest{
+				id:   rand.Uint64(),
+				rspC: make(chan pingResponse, 1),
+			}); err != nil {
+				handle.CloseWithError(err)
+				return
+			}
+			c.testEvent(timeReconciliationOk)
+
 		case now := <-dbKeepAliveDelay.Elapsed():
 			dbKeepAliveDelay.Advance(now)

@@ -582,9 +619,18 @@ func (c *Controller) handlePong(handle *upstreamHandle, msg proto.UpstreamInvent
 		log.Warnf("Unexpected upstream pong from server %q (id=%d).", handle.Hello().ServerID, msg.ID)
 		return
 	}
-	pending.rspC <- pingResponse{
-		d: time.Since(pending.start),
+	now := c.clock.Now()
+	pong := pingResponse{
+		reqDuration:     now.Sub(pending.start),
+		systemClock:     msg.SystemClock,
+		controllerClock: now,
 	}
+
+	handle.stateTracker.mu.Lock()
+	handle.stateTracker.pingResponse = pong
+	handle.stateTracker.mu.Unlock()
+
+	pending.rspC <- pong
 	delete(handle.pings, msg.ID)
 }

@@ -592,10 +638,11 @@ func (c *Controller) handlePingRequest(handle *upstreamHandle, req pingRequest)
 	ping := proto.DownstreamInventoryPing{
 		ID: req.id,
 	}
-	start := time.Now()
+	start := c.clock.Now()
 	if err := handle.Send(c.closeContext, ping); err != nil {
 		req.rspC <- pingResponse{
-			err: err,
+			controllerClock: start,
+			err:             err,
 		}
 		return trace.Wrap(err)
 	}
@@ -660,7 +707,7 @@ func (c *Controller) handleSSHServerHB(handle *upstreamHandle, sshServer *types.
 		handle.sshServer = &heartBeatInfo[*types.ServerV2]{}
 	}

-	now := time.Now()
+	now := c.clock.Now()

 	sshServer.SetExpiry(now.Add(c.serverTTL).UTC())

@@ -705,7 +752,7 @@ func (c *Controller) handleAppServerHB(handle *upstreamHandle, appServer *types.
 		handle.appServers[appKey] = &heartBeatInfo[*types.AppServerV3]{}
 	}

-	now := time.Now()
+	now := c.clock.Now()

 	appServer.SetExpiry(now.Add(c.serverTTL).UTC())

@@ -874,7 +921,7 @@ func (c *Controller) keepAliveAppServer(handle *upstreamHandle, now time.Time) e
 				c.testEvent(appKeepAliveOk)
 			}
 		} else if srv.retryUpsert {
-			srv.resource.SetExpiry(time.Now().Add(c.serverTTL).UTC())
+			srv.resource.SetExpiry(c.clock.Now().Add(c.serverTTL).UTC())
 			lease, err := c.auth.UpsertApplicationServer(c.closeContext, srv.resource)
 			if err != nil {
 				c.testEvent(appUpsertRetryErr)
@@ -1003,7 +1050,7 @@ func (c *Controller) keepAliveSSHServer(handle *upstreamHandle, now time.Time) e
 			c.testEvent(sshKeepAliveOk)
 		}
 	} else if handle.sshServer.retryUpsert {
-		handle.sshServer.resource.SetExpiry(time.Now().Add(c.serverTTL).UTC())
+		handle.sshServer.resource.SetExpiry(c.clock.Now().Add(c.serverTTL).UTC())
 		lease, err := c.auth.UpsertNode(c.closeContext, handle.sshServer.resource)
 		if err != nil {
 			c.testEvent(sshUpsertRetryErr)
@@ -28,6 +28,7 @@ import (
 	"time"

 	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

@@ -828,10 +829,25 @@ func TestInstanceHeartbeat(t *testing.T) {
 		withInstanceHBInterval(time.Millisecond*200),
 		withTestEventsChannel(events),
 	)
-	defer controller.Close()

 	// set up fake in-memory control stream
 	upstream, downstream := client.InventoryControlStreamPipe(client.ICSPipePeerAddr(peerAddr))
+	t.Cleanup(func() {
+		controller.Close()
+		downstream.Close()
+		upstream.Close()
+	})
+
+	// Launch goroutine to consume downstream request and don't block control steam handler.
+	go func() {
+		for {
+			select {
+			case <-downstream.Recv():
+			case <-downstream.Done():
+				return
+			}
+		}
+	}()

 	controller.RegisterControlStream(upstream, proto.UpstreamInventoryHello{
 		ServerID: serverID,
@@ -1426,6 +1442,77 @@ func TestGetSender(t *testing.T) {
 	}, 10*time.Second, 100*time.Millisecond)
 }

+// TestTimeReconciliation verifies basic behavior of the time reconciliation check.
+func TestTimeReconciliation(t *testing.T) {
+	const serverID = "test-server"
+	const peerAddr = "1.2.3.4:456"
+	const wantAddr = "1.2.3.4:123"
+
+	ctx, cancel := context.WithCancel(context.Background())
+	events := make(chan testEvent, 1024)
+	auth := &fakeAuth{
+		expectAddr: wantAddr,
+	}
+
+	clock := clockwork.NewRealClock()
+	controller := NewController(
+		auth,
+		usagereporter.DiscardUsageReporter{},
+		withInstanceHBInterval(time.Millisecond*200),
+		withTestEventsChannel(events),
+		WithClock(clock),
+	)
+
+	// Set up fake in-memory control stream.
+	upstream, downstream := client.InventoryControlStreamPipe(client.ICSPipePeerAddr(peerAddr))
+
+	t.Cleanup(func() {
+		require.NoError(t, downstream.Close())
+		require.NoError(t, upstream.Close())
+		require.NoError(t, controller.Close())
+		cancel()
+	})
+
+	controller.RegisterControlStream(upstream, proto.UpstreamInventoryHello{
+		ServerID: serverID,
+		Version:  teleport.Version,
+		Services: []types.SystemRole{types.RoleNode},
+	})
+
+	// Launch goroutine to respond to clock request.
+	go func() {
+		for {
+			select {
+			case msg := <-downstream.Recv():
+				downstream.Send(ctx, proto.UpstreamInventoryPong{
+					ID:          msg.(proto.DownstreamInventoryPing).ID,
+					SystemClock: clock.Now().Add(-time.Minute).UTC(),
+				})
+				return
+			case <-downstream.Done():
+				return
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	_, ok := controller.GetControlStream(serverID)
+	require.True(t, ok)
+
+	awaitEvents(t, events,
+		expect(timeReconciliationOk),
+	)
+	awaitEvents(t, events,
+		expect(instanceHeartbeatOk),
+		deny(instanceHeartbeatErr, instanceCompareFailed, handlerClose),
+	)
+	auth.mu.Lock()
+	m := auth.lastInstance.GetLastMeasurement()
+	auth.mu.Unlock()
+	require.InDelta(t, time.Minute, m.ControllerSystemClock.Sub(m.SystemClock)-m.RequestDuration/2, float64(time.Second))
+}
+
 type eventOpts struct {
 	expect map[testEvent]int
 	deny   map[testEvent]struct{}
@@ -27,6 +27,7 @@ import (
 	"time"

 	"github.com/gravitational/trace"
+	"github.com/jonboulle/clockwork"
 	log "github.com/sirupsen/logrus"

 	"github.com/gravitational/teleport/api/client"
@@ -89,12 +90,16 @@ type DownstreamSender interface {

 type downstreamHandleOptions struct {
 	metadataGetter func(ctx context.Context) (*metadata.Metadata, error)
+	clock          clockwork.Clock
 }

 func (options *downstreamHandleOptions) SetDefaults() {
 	if options.metadataGetter == nil {
 		options.metadataGetter = metadata.Get
 	}
+	if options.clock == nil {
+		options.clock = clockwork.NewRealClock()
+	}
 }

 type DownstreamHandleOption func(c *downstreamHandleOptions)
@@ -105,6 +110,13 @@ func withMetadataGetter(getter func(ctx context.Context) (*metadata.Metadata, er
 	}
 }

+// WithDownstreamClock overrides existing clock for downstream handle.
+func WithDownstreamClock(clock clockwork.Clock) DownstreamHandleOption {
+	return func(opts *downstreamHandleOptions) {
+		opts.clock = clock
+	}
+}
+
 // NewDownstreamHandle creates a new downstream inventory control handle which will create control streams via the
 // supplied create func and manage hello exchange with the supplied upstream hello.
 func NewDownstreamHandle(fn DownstreamCreateFunc, hello proto.UpstreamInventoryHello, opts ...DownstreamHandleOption) DownstreamHandle {
@@ -121,6 +133,7 @@ func NewDownstreamHandle(fn DownstreamCreateFunc, hello proto.UpstreamInventoryH
 		closeContext:   ctx,
 		cancel:         cancel,
 		metadataGetter: options.metadataGetter,
+		clock:          options.clock,
 	}
 	go handle.run(fn, hello)
 	go handle.autoEmitMetadata()
@@ -137,6 +150,7 @@ type downstreamHandle struct {
 	cancel            context.CancelFunc
 	upstreamSSHLabels map[string]string
 	metadataGetter    func(ctx context.Context) (*metadata.Metadata, error)
+	clock             clockwork.Clock
 }

 func (h *downstreamHandle) closing() bool {
@@ -389,6 +403,10 @@ type UpstreamHandle interface {
 	AgentMetadata() proto.UpstreamInventoryAgentMetadata

 	Ping(ctx context.Context, id uint64) (d time.Duration, err error)
+
+	// SystemClock makes ping request to fetch the system clock of the node.
+	SystemClock(ctx context.Context, id uint64) (time.Time, time.Duration, error)
+
 	// HasService is a helper for checking if a given service is associated with this
 	// stream.
 	HasService(types.SystemRole) bool
@@ -454,6 +472,10 @@ type instanceStateTracker struct {
 	// observe the committed state of the instance control log should skip instances for which this field is nil.
 	lastHeartbeat types.Instance

+	// pingResponse stores information about last system clock request to propagate this data in the
+	// next heartbeat request.
+	pingResponse pingResponse
+
 	// retryHeartbeat is set to true if an unexpected error is hit. We retry exactly once, closing
 	// the stream if the retry does not succeede.
 	retryHeartbeat bool
@@ -520,6 +542,15 @@ func (i *instanceStateTracker) WithLock(fn func()) {

 // nextHeartbeat calculates the next heartbeat value. *Must* be called only while lock is held.
 func (i *instanceStateTracker) nextHeartbeat(now time.Time, hello proto.UpstreamInventoryHello, authID string) (types.Instance, error) {
+	var lastMeasurement *types.SystemClockMeasurement
+	if !i.pingResponse.systemClock.IsZero() {
+		lastMeasurement = &types.SystemClockMeasurement{
+			ControllerSystemClock: i.pingResponse.controllerClock,
+			SystemClock:           i.pingResponse.systemClock,
+			RequestDuration:       i.pingResponse.reqDuration,
+		}
+	}
+
 	instance, err := types.NewInstance(hello.ServerID, types.InstanceSpecV1{
 		Version:                 vc.Normalize(hello.Version),
 		Services:                hello.Services,
@@ -528,6 +559,7 @@ func (i *instanceStateTracker) nextHeartbeat(now time.Time, hello proto.Upstream
 		LastSeen:                now.UTC(),
 		ExternalUpgrader:        hello.GetExternalUpgrader(),
 		ExternalUpgraderVersion: vc.Normalize(hello.GetExternalUpgraderVersion()),
+		LastMeasurement:         lastMeasurement,
 	})
 	if err != nil {
 		return nil, trace.Wrap(err)
@@ -615,8 +647,10 @@ type pingRequest struct {
 }

 type pingResponse struct {
-	d   time.Duration
-	err error
+	reqDuration     time.Duration
+	systemClock     time.Time
+	controllerClock time.Time
+	err             error
 }

 func (h *upstreamHandle) Ping(ctx context.Context, id uint64) (d time.Duration, err error) {
@@ -631,7 +665,7 @@ func (h *upstreamHandle) Ping(ctx context.Context, id uint64) (d time.Duration,

 	select {
 	case rsp := <-rspC:
-		return rsp.d, rsp.err
+		return rsp.reqDuration, rsp.err
 	case <-h.Done():
 		return 0, trace.Errorf("failed to recv upstream pong (stream closed)")
 	case <-ctx.Done():
@@ -639,6 +673,27 @@ func (h *upstreamHandle) Ping(ctx context.Context, id uint64) (d time.Duration,
 	}
 }

+// SystemClock makes ping request to fetch the system clock of the downstream.
+func (h *upstreamHandle) SystemClock(ctx context.Context, id uint64) (time.Time, time.Duration, error) {
+	rspC := make(chan pingResponse, 1)
+	select {
+	case h.pingC <- pingRequest{rspC: rspC, id: id}:
+	case <-h.Done():
+		return time.Time{}, 0, trace.Errorf("failed to send downstream ping (stream closed)")
+	case <-ctx.Done():
+		return time.Time{}, 0, trace.Errorf("failed to send downstream ping: %v", ctx.Err())
+	}
+
+	select {
+	case rsp := <-rspC:
+		return rsp.systemClock, rsp.reqDuration, rsp.err
+	case <-h.Done():
+		return time.Time{}, 0, trace.Errorf("failed to recv upstream pong (stream closed)")
+	case <-ctx.Done():
+		return time.Time{}, 0, trace.Errorf("failed to recv upstream ping: %v", ctx.Err())
+	}
+}
+
 func (h *upstreamHandle) Hello() proto.UpstreamInventoryHello {
 	return h.hello
 }
@@ -1221,12 +1221,16 @@ func NewTeleport(cfg *servicecfg.Config) (*TeleportProcess, error) {
 		Hostname:                cfg.Hostname,
 		ExternalUpgrader:        externalUpgrader,
 		ExternalUpgraderVersion: vc.Normalize(upgraderVersion),
-	})
+	}, inventory.WithDownstreamClock(process.Clock))

 	process.inventoryHandle.RegisterPingHandler(func(sender inventory.DownstreamSender, ping proto.DownstreamInventoryPing) {
-		process.logger.InfoContext(process.ExitContext(), "Handling incoming inventory ping.", "id", ping.ID)
+		systemClock := process.Clock.Now().UTC()
+		process.logger.InfoContext(process.ExitContext(), "Handling incoming inventory ping.",
+			"id", ping.ID,
+			"clock", systemClock)
 		err := sender.Send(process.ExitContext(), proto.UpstreamInventoryPong{
-			ID: ping.ID,
+			ID:          ping.ID,
+			SystemClock: systemClock,
 		})
 		if err != nil {
 			process.logger.WarnContext(process.ExitContext(), "Failed to respond to inventory ping.", "id", ping.ID, "error", err)
@@ -2442,6 +2446,10 @@ func (process *TeleportProcess) initAuthService() error {
 	process.RegisterFunc("auth.server_info", func() error {
 		return trace.Wrap(auth.ReconcileServerInfos(process.GracefulExitContext(), authServer))
 	})
+	process.RegisterFunc("auth.server.system-clock-monitor", func() error {
+		return trace.Wrap(authServer.MonitorSystemTime(process.GracefulExitContext()))
+	})
+
 	// execute this when process is asked to exit:
 	process.OnExit("auth.shutdown", func(payload any) {
 		// The listeners have to be closed here, because if shutdown