From 215cb296994cd1f714ae3ed722d3d01badb9ee23 Mon Sep 17 00:00:00 2001 From: Hugo Shaka Date: Tue, 20 May 2025 20:38:43 -0400 Subject: [PATCH] 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 * fix proto field lookup + address feedback * fix tests + add license --------- Co-authored-by: Edoardo Spadolini --- api/types/autoupdate/report.go | 26 +- api/types/autoupdate/report_test.go | 90 +++++++ lib/auth/agent_version_report.go | 167 +++++++++++++ lib/auth/agent_version_report_test.go | 335 ++++++++++++++++++++++++++ lib/auth/auth.go | 9 + lib/inventory/controller.go | 4 +- lib/inventory/inventory.go | 62 ++++- 7 files changed, 682 insertions(+), 11 deletions(-) create mode 100644 api/types/autoupdate/report_test.go create mode 100644 lib/auth/agent_version_report.go create mode 100644 lib/auth/agent_version_report_test.go 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,