Leverage the inventory control stream to report updater metrics

lib/auth/auth.go

@@ -1265,7 +1265,6 @@ func (a *Server) doInstancePeriodics(ctx context.Context) {
 	skipControlPlane := modules.GetModules().Features().Cloud
 
 	// set up aggregators for our periodics
-	imp := newInstanceMetricsPeriodic()
 	uep := newUpgradeEnrollPeriodic()
 
 	// stream all instances to all aggregators
@@ -1278,7 +1277,6 @@ func (a *Server) doInstancePeriodics(ctx context.Context) {
 			}
 		}
 
-		imp.VisitInstance(instances.Item())
 		uep.VisitInstance(instances.Item())
 	}
 
@@ -1287,21 +1285,7 @@ func (a *Server) doInstancePeriodics(ctx context.Context) {
 		return
 	}
 
-	// set instance metric values
-	totalInstancesMetric.Set(float64(imp.TotalInstances()))
-	enrolledInUpgradesMetric.Set(float64(imp.TotalEnrolledInUpgrades()))
-
-	// reset upgrader counts
-	upgraderCountsMetric.Reset()
-
-	for upgraderType, upgraderVersions := range imp.upgraderCounts {
-		for version, count := range upgraderVersions {
-			upgraderCountsMetric.With(prometheus.Labels{
-				teleport.TagUpgrader: upgraderType,
-				teleport.TagVersion:  version,
-			}).Set(float64(count))
-		}
-	}
+	a.updateUpdaterVersionMetrics()
 
 	// create/delete upgrade enroll prompt as appropriate
 	enrollMsg, shouldPrompt := uep.GenerateEnrollPrompt()
@@ -1488,6 +1472,32 @@ func (a *Server) doReleaseAlertSync(ctx context.Context, current vc.Target, visi
 	}
 }
 
+// updateUpdaterVersionMetrics leverages the inventory control stream to report the
+// number of teleport updaters installed and their versions. To get an accurate representation
+// of versions in an entire cluster the metric must be aggregated with all auth instances.
+func (a *Server) updateUpdaterVersionMetrics() {
+	imp := newInstanceMetricsPeriodic()
+
+	// record versions for all connected resources
+	a.inventory.Iter(func(handle inventory.UpstreamHandle) {
+		imp.VisitInstance(handle.Hello())
+	})
+
+	totalInstancesMetric.Set(float64(imp.TotalInstances()))
+	enrolledInUpgradesMetric.Set(float64(imp.TotalEnrolledInUpgrades()))
+
+	// reset the gauges so that any versions that fall off are removed from exported metrics
+	upgraderCountsMetric.Reset()
+	for upgraderType, upgraderVersions := range imp.upgraderCounts {
+		for version, count := range upgraderVersions {
+			upgraderCountsMetric.With(prometheus.Labels{
+				teleport.TagUpgrader: upgraderType,
+				teleport.TagVersion:  version,
+			}).Set(float64(count))
+		}
+	}
+}
+
 // updateVersionMetrics leverages the inventory control stream to report the versions of
 // all instances that are connected to a single auth server via prometheus metrics. To
 // get an accurate representation of versions in an entire cluster the metric must be aggregated

lib/auth/periodic.go

@@ -23,6 +23,7 @@ import (
 
 	"golang.org/x/mod/semver"
 
+	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 	vc "github.com/gravitational/teleport/lib/versioncontrol"
 )
@@ -114,7 +115,7 @@ func newInstanceMetricsPeriodic() *instanceMetricsPeriodic {
 }
 
 // VisitInstance adds an instance to ongoing aggregations.
-func (i *instanceMetricsPeriodic) VisitInstance(instance types.Instance) {
+func (i *instanceMetricsPeriodic) VisitInstance(instance proto.UpstreamInventoryHello) {
 	i.totalInstances++
 	if upgrader := instance.GetExternalUpgrader(); upgrader != "" {
 		if _, exists := i.upgraderCounts[upgrader]; !exists {

lib/auth/periodic_test.go

@@ -25,20 +25,21 @@ import (
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 
+	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/types"
 )
 
 func TestInstanceMetricsPeriodic(t *testing.T) {
 	tts := []struct {
 		desc           string
-		specs          []types.InstanceSpecV1
+		instances      []proto.UpstreamInventoryHello
 		expectedCounts map[string]map[string]int
 		upgraders      []string
 		expectEnrolled int
 	}{
 		{
 			desc: "mixed",
-			specs: []types.InstanceSpecV1{
+			instances: []proto.UpstreamInventoryHello{
 				{ExternalUpgrader: "kube", ExternalUpgraderVersion: "13.0.0"},
 				{ExternalUpgrader: "kube", ExternalUpgraderVersion: "14.0.0"},
 				{ExternalUpgrader: "unit", ExternalUpgraderVersion: "13.0.0"},
@@ -68,7 +69,7 @@ func TestInstanceMetricsPeriodic(t *testing.T) {
 		},
 		{
 			desc: "all-unenrolled",
-			specs: []types.InstanceSpecV1{
+			instances: []proto.UpstreamInventoryHello{
 				{},
 				{},
 			},
@@ -80,7 +81,7 @@ func TestInstanceMetricsPeriodic(t *testing.T) {
 		},
 		{
 			desc: "all-enrolled",
-			specs: []types.InstanceSpecV1{
+			instances: []proto.UpstreamInventoryHello{
 				{ExternalUpgrader: "kube", ExternalUpgraderVersion: "13.0.0"},
 				{ExternalUpgrader: "kube", ExternalUpgraderVersion: "13.0.0"},
 				{ExternalUpgrader: "unit", ExternalUpgraderVersion: "13.0.0"},
@@ -104,7 +105,7 @@ func TestInstanceMetricsPeriodic(t *testing.T) {
 		},
 		{
 			desc: "nil version",
-			specs: []types.InstanceSpecV1{
+			instances: []proto.UpstreamInventoryHello{
 				{ExternalUpgrader: "kube"},
 				{ExternalUpgrader: "unit"},
 			},
@@ -132,10 +133,7 @@ func TestInstanceMetricsPeriodic(t *testing.T) {
 		t.Run(tt.desc, func(t *testing.T) {
 			periodic := newInstanceMetricsPeriodic()
 
-			for _, upgrader := range tt.specs {
-				instance, err := types.NewInstance(uuid.New().String(), upgrader)
-				require.NoError(t, err)
-
+			for _, instance := range tt.instances {
 				periodic.VisitInstance(instance)
 			}