Generate autoupdate agent report periodically (#54865)

* Generate autoupdate agent report periodically

* address edoardo's feedback

* Apply suggestions from code review

Co-authored-by: Edoardo Spadolini <edoardo.spadolini@goteleport.com>

* fix proto field lookup + address feedback

* fix tests + add license

---------

Co-authored-by: Edoardo Spadolini <edoardo.spadolini@goteleport.com>

api/types/autoupdate/report.go

@@ -17,13 +17,22 @@ limitations under the License.
 package autoupdate
 
 import (
+	"time"
+
 	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/gravitational/teleport/api/gen/proto/go/teleport/autoupdate/v1"
 	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
 	"github.com/gravitational/teleport/api/types"
 )
 
+const (
+	autoUpdateAgentReportTTL = time.Hour
+	maxGroups                = 50
+	maxVersions              = 20
+)
+
 // NewAutoUpdateAgentReport creates a new auto update version resource.
 func NewAutoUpdateAgentReport(spec *autoupdate.AutoUpdateAgentReportSpec, authName string) (*autoupdate.AutoUpdateAgentReport, error) {
 	rollout := &autoupdate.AutoUpdateAgentReport{
@@ -31,6 +40,8 @@ func NewAutoUpdateAgentReport(spec *autoupdate.AutoUpdateAgentReportSpec, authNa
 		Version: types.V1,
 		Metadata: &headerv1.Metadata{
 			Name: authName,
+			// Validate will fail later if timestamp is zero
+			Expires: timestamppb.New(spec.GetTimestamp().AsTime().Add(autoUpdateAgentReportTTL)),
 		},
 		Spec: spec,
 	}
@@ -51,6 +62,19 @@ func ValidateAutoUpdateAgentReport(v *autoupdate.AutoUpdateAgentReport) error {
 		return trace.BadParameter("Spec is nil")
 	}
 
-	// TODO: see if we need more validation
+	if ts := v.GetSpec().GetTimestamp(); ts.GetSeconds() == 0 && ts.GetNanos() == 0 {
+		return trace.BadParameter("Spec.Timestamp is empty or zero")
+	}
+
+	if numGroups := len(v.GetSpec().GetGroups()); numGroups > maxGroups {
+		return trace.BadParameter("Spec.Groups is too large (%d while the max is %d)", numGroups, maxGroups)
+	}
+
+	for groupName, group := range v.GetSpec().GetGroups() {
+		if numVersions := len(group.GetVersions()); numVersions > maxVersions {
+			return trace.BadParameter("group %q has too many versions (%d while the max is %d)", groupName, numVersions, maxVersions)
+		}
+	}
+
 	return nil
 }

api/types/autoupdate/report_test.go

@@ -0,0 +1,90 @@
+/*
+ * Teleport
+ * Copyright (C) 2023  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 autoupdate
+
+import (
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/gravitational/teleport/api/gen/proto/go/teleport/autoupdate/v1"
+	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
+	"github.com/gravitational/teleport/api/types"
+)
+
+func TestNewAutoUpdateAgentReport(t *testing.T) {
+	now := timestamppb.Now()
+	expires := timestamppb.New(now.AsTime().Add(autoUpdateAgentReportTTL))
+	tests := []struct {
+		name     string
+		spec     *autoupdate.AutoUpdateAgentReportSpec
+		authName string
+
+		want    *autoupdate.AutoUpdateAgentReport
+		wantErr require.ErrorAssertionFunc
+	}{
+		{
+			name:     "nil spec",
+			authName: "test",
+			wantErr:  require.Error,
+		},
+		{
+			name: "empty name",
+			spec: &autoupdate.AutoUpdateAgentReportSpec{
+				Timestamp: now,
+			},
+			wantErr: require.Error,
+		},
+		{
+			name:     "no timestamp",
+			authName: "test",
+			spec:     &autoupdate.AutoUpdateAgentReportSpec{},
+			wantErr:  require.Error,
+		},
+		{
+			name:     "ok",
+			authName: "test",
+			spec: &autoupdate.AutoUpdateAgentReportSpec{
+				Timestamp: now,
+			},
+			want: &autoupdate.AutoUpdateAgentReport{
+				Kind:    types.KindAutoUpdateAgentReport,
+				Version: types.V1,
+				Metadata: &headerv1.Metadata{
+					Name:    "test",
+					Expires: expires,
+				},
+				Spec: &autoupdate.AutoUpdateAgentReportSpec{
+					Timestamp: now,
+				},
+			},
+			wantErr: require.NoError,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := NewAutoUpdateAgentReport(tt.spec, tt.authName)
+			tt.wantErr(t, err)
+			require.Empty(t, cmp.Diff(tt.want, result, protocmp.Transform()))
+		})
+	}
+}

lib/auth/agent_version_report.go

@@ -0,0 +1,167 @@
+/*
+ * Teleport
+ * Copyright (C) 2025  Gravitational, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package auth
+
+import (
+	"context"
+	"time"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	autoupdatev1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/autoupdate/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/autoupdate"
+	"github.com/gravitational/teleport/lib/inventory"
+)
+
+type instanceReport struct {
+	data      map[string]instanceGroupReport
+	timestamp time.Time
+}
+
+func (ir instanceReport) collectInstance(handle inventory.UpstreamHandle) {
+	// If the instance is being soft-reloaded or shut down, we ignore it.
+	if goodbye := handle.Goodbye(); goodbye.GetSoftReload() || goodbye.GetDeleteResources() {
+		return
+	}
+
+	// We skip servers that joined less than a minute ago as they might have been
+	// connected to another auth instance a few seconds ago, which would lead to double-counting.
+	if ir.timestamp.Sub(handle.RegistrationTime()) < time.Minute {
+		return
+	}
+	// We skip control planes instances because we don't update them.
+	if handle.HasControlPlaneService() {
+		return
+	}
+
+	hello := handle.Hello()
+
+	// If the machine has no updater, we skip it
+	if hello.ExternalUpgrader == "" {
+		return
+	}
+
+	// Reject instance not advertising
+	updaterInfo := hello.GetUpdaterInfo()
+	if updaterInfo == nil {
+		return
+	}
+
+	// Reject instances who are not advertising the group properly.
+	// They might be running too old versions.
+	updateGroup := updaterInfo.UpdateGroup
+	if updateGroup == "" {
+		return
+	}
+
+	// We skip instances whose updater status is not OK.
+	status := updaterInfo.UpdaterStatus
+	if status != types.UpdaterStatus_UPDATER_STATUS_OK {
+		return
+	}
+
+	if _, ok := ir.data[updateGroup]; !ok {
+		ir.data[updateGroup] = instanceGroupReport{}
+	}
+
+	ir.data[updateGroup].collectInstance(handle)
+}
+
+type instanceGroupReport map[string]instanceGroupVersionReport
+
+func (ir instanceGroupReport) collectInstance(handle inventory.UpstreamHandle) {
+	hello := handle.Hello()
+
+	stats, ok := ir[hello.Version]
+	if !ok {
+		stats = instanceGroupVersionReport{}
+	}
+
+	stats.count += 1
+
+	ir[hello.Version] = stats
+}
+
+type instanceGroupVersionReport struct {
+	count int
+	// Leaving room here to add the lowest UUID, as described in RFD 184.
+}
+
+func (a *Server) generateAgentVersionReport(ctx context.Context) (*autoupdatev1pb.AutoUpdateAgentReport, error) {
+	now := a.clock.Now()
+
+	a.logger.DebugContext(ctx, "Collecting agent versions from inventory")
+	rawreport := instanceReport{timestamp: now, data: make(map[string]instanceGroupReport)}
+	a.inventory.AllHandles(rawreport.collectInstance)
+
+	a.logger.DebugContext(ctx, "Building the agent version report")
+	spec := &autoupdatev1pb.AutoUpdateAgentReportSpec{
+		Timestamp: timestamppb.New(a.clock.Now()),
+		Groups:    make(map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup, len(rawreport.data)),
+	}
+
+	// TODO(hugoShaka): gracefully handle too many groups or versions (sort and report only the largest ones).
+	// Currently the agent version report will just fail validation if there are too many groups.
+
+	for groupName, groupData := range rawreport.data {
+		versions := make(map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion, len(groupData))
+		for versionName, groupVersionData := range groupData {
+			versions[versionName] = &autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+				Count: int32(groupVersionData.count),
+			}
+		}
+		spec.Groups[groupName] = &autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+			Versions: versions,
+		}
+	}
+
+	report, err := autoupdate.NewAutoUpdateAgentReport(spec, a.ServerID)
+	if err != nil {
+		return nil, trace.Wrap(err, "failed to generate agent version report")
+	}
+
+	return report, nil
+}
+
+func (a *Server) reportAgentVersions(ctx context.Context) {
+	if _, err := a.GetAutoUpdateAgentRollout(ctx); err != nil {
+		if trace.IsNotFound(err) {
+			a.logger.DebugContext(ctx, "Skipping periodic agent report because the cluster doesn't contain an autoupdate_agent_rollout.")
+			return
+		}
+		a.logger.WarnContext(ctx, "Failed to check if autoupdate_agent_rollout resource exists, aborting periodic agent report", "error", err)
+	}
+
+	a.logger.DebugContext(ctx, "Periodic agent version report routine started")
+	report, err := a.generateAgentVersionReport(ctx)
+	if err != nil {
+		a.logger.WarnContext(ctx, "Failed to report agent versions", "error", err)
+		return
+	}
+
+	a.logger.DebugContext(ctx, "Writing agent version report to the backend", "name", report.GetMetadata().GetName())
+	_, err = a.UpsertAutoUpdateAgentReport(ctx, report)
+	if err != nil {
+		a.logger.ErrorContext(ctx, "Failed to write agent version report", "error", err)
+	}
+	a.logger.DebugContext(ctx, "Finished exporting the agent version report")
+
+}

lib/auth/agent_version_report_test.go

@@ -0,0 +1,335 @@
+/*
+ * Teleport
+ * Copyright (C) 2025  Gravitational, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package auth
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/uuid"
+	"github.com/jonboulle/clockwork"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	"github.com/gravitational/teleport/api/client"
+	"github.com/gravitational/teleport/api/client/proto"
+	autoupdatev1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/autoupdate/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/autoupdate"
+	"github.com/gravitational/teleport/lib/backend/memory"
+	"github.com/gravitational/teleport/lib/inventory"
+	"github.com/gravitational/teleport/lib/services/local"
+	"github.com/gravitational/teleport/lib/utils"
+)
+
+func newFakeControlStream() fakeControlStream {
+	return fakeControlStream{
+		msgChan:  make(chan proto.UpstreamInventoryMessage),
+		doneChan: make(chan struct{}),
+	}
+}
+
+type fakeControlStream struct {
+	client.UpstreamInventoryControlStream
+	msgChan  chan proto.UpstreamInventoryMessage
+	doneChan chan struct{}
+}
+
+func (f fakeControlStream) CloseWithError(err error) error {
+	return nil
+}
+
+func (f fakeControlStream) Close() error {
+	return nil
+}
+
+func (f fakeControlStream) Recv() <-chan proto.UpstreamInventoryMessage {
+	return f.msgChan
+}
+
+func (f fakeControlStream) Done() <-chan struct{} {
+	return f.doneChan
+}
+
+func (f fakeControlStream) fakeMsg(msg proto.UpstreamInventoryMessage) {
+	f.msgChan <- msg
+}
+
+func (f fakeControlStream) close() {
+	close(f.msgChan)
+}
+
+type fakeServer struct {
+	version       string
+	updateGroup   string
+	delay         time.Duration
+	roles         types.SystemRoles
+	updaterStatus types.UpdaterStatus
+	goodbye       *proto.UpstreamInventoryGoodbye
+}
+
+func TestServer_generateAgentVersionReport(t *testing.T) {
+	ctx := context.Background()
+	now := time.Now()
+	twoMinutesAgo := now.Add(-time.Minute * 2)
+	// agentRole are typicial roles an agent can have
+	agentRoles := types.SystemRoles{types.RoleNode, types.RoleApp}
+	updaterOK := types.UpdaterStatus_UPDATER_STATUS_OK
+
+	tests := []struct {
+		name     string
+		fixtures []fakeServer
+		expected *autoupdatev1pb.AutoUpdateAgentReportSpec
+	}{
+		{
+			name: "no servers",
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+			},
+		},
+		{
+			name: "no group, same version",
+			fixtures: []fakeServer{
+				{version: "1.2.3", roles: agentRoles, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updateGroup: "default"},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 4},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "control plane servers are ignored",
+			fixtures: []fakeServer{
+				{version: "1.2.3", roles: types.SystemRoles{types.RoleKube, types.RoleAuth}, updateGroup: "default"},
+				{version: "1.2.3", roles: types.SystemRoles{types.RoleApp, types.RoleProxy}, updateGroup: "default"},
+				{version: "1.2.3", roles: types.SystemRoles{types.RoleApp, types.RoleKube}, updateGroup: "default"},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 1},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "disabled or pinned updaters are ignored",
+			fixtures: []fakeServer{
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: types.UpdaterStatus_UPDATER_STATUS_PINNED, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: types.UpdaterStatus_UPDATER_STATUS_DISABLED, updateGroup: "default"},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 1},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "reloaded and terminating instances are ignored",
+			fixtures: []fakeServer{
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default", goodbye: &proto.UpstreamInventoryGoodbye{SoftReload: true}},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default", goodbye: &proto.UpstreamInventoryGoodbye{DeleteResources: true}},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 1},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "too recent servers are ignored",
+			fixtures: []fakeServer{
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+				{version: "1.2.3", delay: 90 * time.Second, roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+				{version: "1.2.3", roles: agentRoles, updaterStatus: updaterOK, updateGroup: "default"},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 1},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "multiple versions and groups",
+			fixtures: []fakeServer{
+				{version: "1.2.3", updateGroup: "default", roles: agentRoles, updaterStatus: updaterOK},
+				{version: "1.2.4", updateGroup: "default", roles: agentRoles, updaterStatus: updaterOK},
+				{version: "1.2.5", updateGroup: "prod", roles: agentRoles, updaterStatus: updaterOK},
+				{version: "1.2.5", updateGroup: "prod", roles: agentRoles, updaterStatus: updaterOK},
+				{version: "1.2.5", updateGroup: "dev", roles: agentRoles, updaterStatus: updaterOK},
+			},
+			expected: &autoupdatev1pb.AutoUpdateAgentReportSpec{
+				Timestamp: timestamppb.New(now),
+				Groups: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroup{
+					"default": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.3": {Count: 1},
+							"1.2.4": {Count: 1},
+						},
+					},
+					"dev": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.5": {Count: 1},
+						},
+					},
+					"prod": {
+						Versions: map[string]*autoupdatev1pb.AutoUpdateAgentReportSpecGroupVersion{
+							"1.2.5": {Count: 2},
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			clock := clockwork.NewFakeClockAt(twoMinutesAgo)
+			auth := &Server{
+				logger:   utils.NewSlogLoggerForTests(),
+				ServerID: uuid.NewString(),
+			}
+			controller := inventory.NewController(auth, nil, inventory.WithClock(clock))
+			for _, fixture := range tt.fixtures {
+				clock.Advance(fixture.delay)
+				stream := newFakeControlStream()
+				status := fixture.updaterStatus
+				if status == types.UpdaterStatus_UPDATER_STATUS_UNSPECIFIED {
+					status = types.UpdaterStatus_UPDATER_STATUS_OK
+				}
+				controller.RegisterControlStream(stream, proto.UpstreamInventoryHello{
+					Services:         fixture.roles,
+					ServerID:         uuid.New().String(),
+					Version:          fixture.version,
+					ExternalUpgrader: types.UpgraderKindTeleportUpdate,
+					UpdaterInfo:      &types.UpdaterV2Info{UpdaterStatus: status, UpdateGroup: fixture.updateGroup},
+				})
+				if fixture.goodbye != nil {
+					stream.fakeMsg(*fixture.goodbye)
+				}
+				t.Cleanup(stream.close)
+			}
+			auth.inventory = controller
+			auth.clock = clockwork.NewFakeClockAt(now)
+
+			report, err := auth.generateAgentVersionReport(ctx)
+			require.NoError(t, err)
+			require.Empty(t, cmp.Diff(tt.expected, report.GetSpec(), protocmp.Transform()))
+		})
+	}
+}
+
+func TestServer_reportAgentVersions(t *testing.T) {
+	// Test setup: create auth.
+	bk, err := memory.New(memory.Config{})
+	require.NoError(t, err)
+
+	svc, err := local.NewAutoUpdateService(bk)
+	require.NoError(t, err)
+
+	now := time.Now()
+	twoMinutesAgo := now.Add(-time.Minute * 2)
+	clock := clockwork.NewFakeClockAt(twoMinutesAgo)
+
+	// Test setup: load fixtures.
+	const testNodeCount = 10
+	auth := &Server{
+		clock:    clock,
+		ServerID: uuid.NewString(),
+		Services: &Services{AutoUpdateService: svc},
+		logger:   utils.NewSlogLoggerForTests(),
+	}
+	auth.Cache = auth.Services
+	controller := inventory.NewController(auth, nil, inventory.WithClock(clock))
+	auth.inventory = controller
+
+	for range testNodeCount {
+		stream := newFakeControlStream()
+		controller.RegisterControlStream(stream, proto.UpstreamInventoryHello{
+			Services:         types.SystemRoles{types.RoleNode},
+			Version:          "1.2.3",
+			ServerID:         uuid.NewString(),
+			ExternalUpgrader: types.UpgraderKindTeleportUpdate,
+			UpdaterInfo: &types.UpdaterV2Info{
+				UpdaterStatus: types.UpdaterStatus_UPDATER_STATUS_OK,
+				UpdateGroup:   "default",
+			},
+		})
+		t.Cleanup(stream.close)
+	}
+	ctx := context.Background()
+	rollout, err := autoupdate.NewAutoUpdateAgentRollout(&autoupdatev1pb.AutoUpdateAgentRolloutSpec{
+		StartVersion:              "1.2.3",
+		TargetVersion:             "1.2.4",
+		Schedule:                  autoupdate.AgentsScheduleRegular,
+		AutoupdateMode:            autoupdate.AgentsUpdateModeEnabled,
+		Strategy:                  autoupdate.AgentsStrategyHaltOnError,
+		MaintenanceWindowDuration: nil,
+	})
+	require.NoError(t, err)
+	_, err = svc.CreateAutoUpdateAgentRollout(ctx, rollout)
+	require.NoError(t, err)
+
+	// Test execution: compute and write report.
+	clock.Advance(2 * time.Minute)
+	auth.reportAgentVersions(ctx)
+
+	// Test validation
+	report, err := svc.GetAutoUpdateAgentReport(ctx, auth.ServerID)
+	require.NoError(t, err)
+
+	require.NotNil(t, report)
+	require.NotEmpty(t, report.GetSpec().GetGroups())
+	require.NotNil(t, report.GetSpec().GetGroups()["default"])
+	require.NotNil(t, report.GetSpec().GetGroups()["default"].GetVersions())
+	require.NotNil(t, report.GetSpec().GetGroups()["default"].GetVersions()["1.2.3"])
+	require.Equal(t, testNodeCount, int(report.GetSpec().GetGroups()["default"].GetVersions()["1.2.3"].GetCount()))
+}

lib/auth/auth.go

@@ -1461,6 +1461,7 @@ const (
 	upgradeWindowCheckKey
 	roleCountKey
 	accessListReminderNotificationsKey
+	autoUpdateAgentReportKey
 )
 
 // runPeriodicOperations runs some periodic bookkeeping operations
@@ -1550,6 +1551,12 @@ func (a *Server) runPeriodicOperations() {
 			FirstDuration: retryutils.HalfJitter(10 * time.Second),
 			Jitter:        retryutils.HalfJitter,
 		})
+		ticker.Push(interval.SubInterval[periodicIntervalKey]{
+			Key:           autoUpdateAgentReportKey,
+			Duration:      time.Minute,
+			FirstDuration: retryutils.FullJitter(time.Minute),
+			// No jitter here, this is intentional and required for accurate tracking across auths.
+		})
 	}
 
 	if modules.GetModules().IsOSSBuild() {
@@ -1674,6 +1681,8 @@ func (a *Server) runPeriodicOperations() {
 				go a.tallyRoles(a.closeCtx)
 			case accessListReminderNotificationsKey:
 				go a.CreateAccessListReminderNotifications(a.closeCtx)
+			case autoUpdateAgentReportKey:
+				go a.reportAgentVersions(a.closeCtx)
 			}
 		}
 	}

lib/inventory/controller.go

@@ -342,7 +342,7 @@ func NewController(auth Auth, usageReporter usagereporter.UsageReporter, opts ..
 
 // RegisterControlStream registers a new control stream with the controller.
 func (c *Controller) RegisterControlStream(stream client.UpstreamInventoryControlStream, hello proto.UpstreamInventoryHello) {
-	handle := newUpstreamHandle(stream, hello)
+	handle := newUpstreamHandle(stream, hello, c.clock.Now())
 	c.store.Insert(handle)
 
 	// Increment the concurrent connection counter that we use to calculate the
@@ -553,7 +553,7 @@ func (c *Controller) handleControlStream(handle *upstreamHandle) {
 			case proto.UpstreamInventoryPong:
 				c.handlePong(handle, m)
 			case proto.UpstreamInventoryGoodbye:
-				handle.goodbye = m
+				handle.setGoodbye(&m)
 			default:
 				slog.WarnContext(c.closeContext, "Unexpected upstream message type on control stream",
 					"message_type", logutils.TypeAttr(m),

lib/inventory/inventory.go

@@ -483,6 +483,13 @@ type UpstreamHandle interface {
 	// Hello gets the cached upstream hello that was used to initialize the stream.
 	Hello() proto.UpstreamInventoryHello
 
+	// Goodbye gets the cached upstream goodbye. Returns nil if downstream never sent a Goodbye.
+	// This is used to identify if the instance is terminating or being soft-reloaded.
+	Goodbye() *proto.UpstreamInventoryGoodbye
+
+	// RegistrationTime gets the timestamp of the control stream initialization.
+	RegistrationTime() time.Time
+
 	// AgentMetadata is the service's metadata: OS, glibc version, install methods, ...
 	AgentMetadata() proto.UpstreamInventoryAgentMetadata
 
@@ -492,6 +499,10 @@ type UpstreamHandle interface {
 	// stream.
 	HasService(types.SystemRole) bool
 
+	// HasControlPlaneService returns true if at least a control plane service
+	// is associated with this stream.
+	HasControlPlaneService() bool
+
 	// VisitInstanceState runs the provided closure against a representation of the most
 	// recently observed instance state, plus any pending control log entries. The returned
 	// value may optionally include additional control log entries to add to the pending
@@ -665,11 +676,12 @@ func (i *instanceStateTracker) nextHeartbeat(now time.Time, hello proto.Upstream
 
 type upstreamHandle struct {
 	client.UpstreamInventoryControlStream
-	hello   proto.UpstreamInventoryHello
-	goodbye proto.UpstreamInventoryGoodbye
+	hello            proto.UpstreamInventoryHello
+	registrationTime time.Time
 
-	agentMDLock   sync.RWMutex
+	agentInfoLock sync.Mutex
 	agentMetadata proto.UpstreamInventoryAgentMetadata
+	goodbye       *proto.UpstreamInventoryGoodbye
 
 	pingC chan pingRequest
 
@@ -721,12 +733,13 @@ type heartBeatInfo[T any] struct {
 	keepAliveErrs int
 }
 
-func newUpstreamHandle(stream client.UpstreamInventoryControlStream, hello proto.UpstreamInventoryHello) *upstreamHandle {
+func newUpstreamHandle(stream client.UpstreamInventoryControlStream, hello proto.UpstreamInventoryHello, now time.Time) *upstreamHandle {
 	return &upstreamHandle{
 		UpstreamInventoryControlStream: stream,
 		pingC:                          make(chan pingRequest),
 		hello:                          hello,
 		pings:                          make(map[uint64]pendingPing),
+		registrationTime:               now,
 	}
 }
 
@@ -767,24 +780,46 @@ func (h *upstreamHandle) Ping(ctx context.Context, id uint64) (d time.Duration,
 	}
 }
 
+// Goodbye gets the cached upstream goodbye. Returns nil if downstream never sent a Goodbye.
+// This is used to identify if the instance is terminating or being soft-reloaded.
+func (h *upstreamHandle) Goodbye() *proto.UpstreamInventoryGoodbye {
+	h.agentInfoLock.Lock()
+	defer h.agentInfoLock.Unlock()
+	return h.goodbye
+}
+
+// setGoodbye sets the goodbye for the current handler.
+func (h *upstreamHandle) setGoodbye(goodbye *proto.UpstreamInventoryGoodbye) {
+	h.agentInfoLock.Lock()
+	defer h.agentInfoLock.Unlock()
+	h.goodbye = goodbye
+}
+
 func (h *upstreamHandle) Hello() proto.UpstreamInventoryHello {
 	return h.hello
 }
 
+// RegistrationTime implements UpstreamHandle by returning the handle's creation timestamp.
+func (h *upstreamHandle) RegistrationTime() time.Time {
+	return h.registrationTime
+}
+
 // AgentMetadata returns the Agent's metadata (eg os, glibc version, install methods, teleport version).
 func (h *upstreamHandle) AgentMetadata() proto.UpstreamInventoryAgentMetadata {
-	h.agentMDLock.RLock()
-	defer h.agentMDLock.RUnlock()
+	h.agentInfoLock.Lock()
+	defer h.agentInfoLock.Unlock()
 	return h.agentMetadata
 }
 
 // SetAgentMetadata sets the agent metadata for the current handler.
 func (h *upstreamHandle) SetAgentMetadata(agentMD proto.UpstreamInventoryAgentMetadata) {
-	h.agentMDLock.Lock()
-	defer h.agentMDLock.Unlock()
+	h.agentInfoLock.Lock()
+	defer h.agentInfoLock.Unlock()
 	h.agentMetadata = agentMD
 }
 
+// HasService is a helper for checking if a given service is associated with this
+// stream.
 func (h *upstreamHandle) HasService(service types.SystemRole) bool {
 	for _, s := range h.hello.Services {
 		if s == service {
@@ -794,6 +829,17 @@ func (h *upstreamHandle) HasService(service types.SystemRole) bool {
 	return false
 }
 
+// HasControlPlaneService implements UpstreamHandle and returns true if at
+// least a control plane service is associated with this stream.
+func (h *upstreamHandle) HasControlPlaneService() bool {
+	for _, s := range h.hello.Services {
+		if s.IsControlPlane() {
+			return true
+		}
+	}
+	return false
+}
+
 func (h *upstreamHandle) UpdateLabels(ctx context.Context, kind proto.LabelUpdateKind, labels map[string]string) error {
 	req := proto.DownstreamInventoryUpdateLabels{
 		Kind:   kind,