diff --git a/api/types/autoupdate/report.go b/api/types/autoupdate/report.go
index 278b77e9cd0..0a2bb9d1642 100644
--- a/api/types/autoupdate/report.go
+++ b/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
}
diff --git a/api/types/autoupdate/report_test.go b/api/types/autoupdate/report_test.go
new file mode 100644
index 00000000000..b93d9719c20
--- /dev/null
+++ b/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 .
+ */
+
+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()))
+ })
+ }
+}
diff --git a/lib/auth/agent_version_report.go b/lib/auth/agent_version_report.go
new file mode 100644
index 00000000000..383c525697a
--- /dev/null
+++ b/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 .
+ */
+
+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")
+
+}
diff --git a/lib/auth/agent_version_report_test.go b/lib/auth/agent_version_report_test.go
new file mode 100644
index 00000000000..afd39794d6e
--- /dev/null
+++ b/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 .
+ */
+
+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()))
+}
diff --git a/lib/auth/auth.go b/lib/auth/auth.go
index 7fffd1096bb..34b5f61ae98 100644
--- a/lib/auth/auth.go
+++ b/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)
}
}
}
diff --git a/lib/inventory/controller.go b/lib/inventory/controller.go
index bbd23ecc8b7..b6a592d9f38 100644
--- a/lib/inventory/controller.go
+++ b/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),
diff --git a/lib/inventory/inventory.go b/lib/inventory/inventory.go
index 3c97bebcbee..1eac7234ba2 100644
--- a/lib/inventory/inventory.go
+++ b/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,