[vnet] feat: SSH usage reporting

lib/teleterm/vnet/service.go

@@ -530,6 +530,17 @@ func (p *clientApplication) GetDialOptions(ctx context.Context, profileName stri
 	return dialOpts, nil
 }
 
+// OnNewSSHSession submits a usage event for a new SSH session.
+func (p *clientApplication) OnNewSSHSession(ctx context.Context, profileName, targetClusterName string) {
+	// Enqueue the event from a separate goroutine since we don't care about errors anyway and we also
+	// don't want to slow down VNet connections.
+	go func() {
+		if err := p.usageReporter.ReportSSHSession(ctx, profileName, targetClusterName); err != nil {
+			log.ErrorContext(ctx, "Failed to submit SSH usage event")
+		}
+	}()
+}
+
 // OnNewConnection submits a usage event once per clientApplication lifetime.
 // That is, if a user makes multiple connections to a single app, OnNewConnection submits a single
 // event. This is to mimic how Connect submits events for its app gateways. This lets us compare
@@ -542,9 +553,8 @@ func (p *clientApplication) OnNewConnection(ctx context.Context, appKey *vnetv1.
 
 		// Not passing ctx to ReportApp since ctx is tied to the lifetime of the connection.
 		// If it's a short-lived connection, inheriting its context would interrupt reporting.
-		err := p.usageReporter.ReportApp(uri)
-		if err != nil {
-			log.ErrorContext(ctx, "Failed to submit usage event", "app", uri, "error", err)
+		if err := p.usageReporter.ReportApp(uri); err != nil {
+			log.ErrorContext(ctx, "Failed to submit app usage event", "app", uri, "error", err)
 		}
 	}()
 
@@ -606,6 +616,7 @@ func (p *clientApplication) OnInvalidLocalPort(ctx context.Context, appInfo *vne
 
 type usageReporter interface {
 	ReportApp(uri.ResourceURI) error
+	ReportSSHSession(ctx context.Context, profileName, targetClusterName string) error
 	Stop()
 }
 
@@ -675,6 +686,48 @@ func newDaemonUsageReporter(cfg daemonUsageReporterConfig) (*daemonUsageReporter
 	}, nil
 }
 
+// ReportSSHSession adds an event for a new SSH session to the events queue.
+func (r *daemonUsageReporter) ReportSSHSession(ctx context.Context, profileName, targetClusterName string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.closed.Load() {
+		return trace.CompareFailed("usage reported has been stopped")
+	}
+
+	rootClusterURI := uri.NewClusterURI(profileName)
+	_, tc, err := r.cfg.ClientCache.ResolveClusterURI(rootClusterURI)
+	if err != nil {
+		return trace.Wrap(err)
+	}
+
+	clusterID, ok := r.cfg.ClusterIDCache.Load(rootClusterURI)
+	if !ok {
+		return trace.NotFound("cluster ID for %q not found", rootClusterURI)
+	}
+
+	log.DebugContext(ctx, "Reporting SSH usage event", "profile", profileName, "cluster", targetClusterName)
+	if err := r.cfg.EventConsumer.ReportUsageEvent(&apiteleterm.ReportUsageEventRequest{
+		AuthClusterId: clusterID,
+		PrehogReq: &prehogv1alpha.SubmitConnectEventRequest{
+			DistinctId: r.cfg.InstallationID,
+			Timestamp:  timestamppb.Now(),
+			Event: &prehogv1alpha.SubmitConnectEventRequest_ProtocolUse{
+				ProtocolUse: &prehogv1alpha.ConnectProtocolUseEvent{
+					ClusterName:   targetClusterName,
+					UserName:      tc.Username,
+					Protocol:      "ssh",
+					Origin:        "vnet",
+					AccessThrough: "vnet",
+				},
+			},
+		},
+	}); err != nil {
+		return trace.Wrap(err, "adding SSH usage event to queue")
+	}
+	return nil
+}
+
 // ReportApp adds an event related to the given app to the events queue, if the app wasn't reported
 // already. Only one invocation of ReportApp can be in flight at a time.
 func (r *daemonUsageReporter) ReportApp(appURI uri.ResourceURI) error {
@@ -716,9 +769,9 @@ func (r *daemonUsageReporter) ReportApp(appURI uri.ResourceURI) error {
 		return trace.NotFound("cluster ID for %q not found", rootClusterURI)
 	}
 
-	log.DebugContext(ctx, "Reporting usage event", "app", appURI.String())
+	log.DebugContext(ctx, "Reporting app usage event", "app", appURI.String())
 
-	err = r.cfg.EventConsumer.ReportUsageEvent(&apiteleterm.ReportUsageEventRequest{
+	if err := r.cfg.EventConsumer.ReportUsageEvent(&apiteleterm.ReportUsageEventRequest{
 		AuthClusterId: clusterID,
 		PrehogReq: &prehogv1alpha.SubmitConnectEventRequest{
 			DistinctId: r.cfg.InstallationID,
@@ -733,9 +786,8 @@ func (r *daemonUsageReporter) ReportApp(appURI uri.ResourceURI) error {
 				},
 			},
 		},
-	})
-	if err != nil {
-		return trace.Wrap(err, "adding usage event to queue")
+	}); err != nil {
+		return trace.Wrap(err, "adding app usage event to queue")
 	}
 
 	r.reportedApps[appURI.String()] = struct{}{}
@@ -762,7 +814,12 @@ func (r *daemonUsageReporter) Stop() {
 type disabledTelemetryUsageReporter struct{}
 
 func (r *disabledTelemetryUsageReporter) ReportApp(appURI uri.ResourceURI) error {
-	log.DebugContext(context.Background(), "Skipping usage event, usage reporting is turned off", "app", appURI.String())
+	log.DebugContext(context.Background(), "Skipping app usage event, usage reporting is turned off", "app", appURI.String())
+	return nil
+}
+
+func (r *disabledTelemetryUsageReporter) ReportSSHSession(ctx context.Context, profileName, targetClusterName string) error {
+	log.DebugContext(ctx, "Skipping SSH usage event, usage reporting is turned off", "profile", profileName, "cluster", targetClusterName)
 	return nil
 }
 

lib/teleterm/vnet/service_test.go

@@ -77,6 +77,11 @@ func TestDaemonUsageReporter(t *testing.T) {
 	err = usageReporter.ReportApp(clusterWithoutClusterID.AppendApp("bar"))
 	require.ErrorIs(t, err, trace.NotFound("cluster ID for \"/clusters/no-cluster-id\" not found"))
 	require.Equal(t, 1, eventConsumer.EventCount())
+
+	// Verify that reporting an SSH session works.
+	err = usageReporter.ReportSSHSession(context.Background(), validCluster.GetProfileName(), "foo")
+	require.NoError(t, err)
+	require.Equal(t, 2, eventConsumer.EventCount())
 }
 
 func TestDaemonUsageReporter_Stop(t *testing.T) {

lib/vnet/client_application_service.go

@@ -373,6 +373,10 @@ func (s *clientApplicationService) SessionSSHConfig(ctx context.Context, req *vn
 		return nil, trace.Errorf("user KeyRing host no trusted SSH CAs for cluster %s", targetCluster)
 	}
 	sessionID := s.setSignerForSSHSession(keyRing.SSHPrivateKey)
+
+	// Submit usage event.
+	s.cfg.clientApplication.OnNewSSHSession(ctx, req.GetProfile(), targetCluster)
+
 	return &vnetv1.SessionSSHConfigResponse{
 		SessionId:  sessionID,
 		Cert:       sshCert.Marshal(),

lib/vnet/user_process.go

@@ -52,6 +52,10 @@ type ClientApplication interface {
 	// GetDialOptions returns ALPN dial options for the profile.
 	GetDialOptions(ctx context.Context, profileName string) (*vnetv1.DialOptions, error)
 
+	// OnNewSSHSession should be called whenever a new SSH session is about to be
+	// started, after getting the user SSH certificate for the session.
+	OnNewSSHSession(ctx context.Context, profileName, targetClusterName string)
+
 	// OnNewConnection gets called whenever a new connection is about to be established through VNet.
 	// By the time OnNewConnection, VNet has already verified that the user holds a valid cert for the
 	// app.

tool/tsh/common/vnet_client_application.go

@@ -134,6 +134,11 @@ func (p *vnetClientApplication) GetDialOptions(ctx context.Context, profileName
 	return dialOpts, nil
 }
 
+// OnNewSSHSession gets called before each VNet SSH connection. It's a noop as
+// tsh doesn't need to do anything extra here.
+func (p *vnetClientApplication) OnNewSSHSession(ctx context.Context, profileName, targetClusterName string) {
+}
+
 // OnNewConnection gets called before each VNet connection. It's a noop as tsh doesn't need to do
 // anything extra here.
 func (p *vnetClientApplication) OnNewConnection(_ context.Context, _ *vnetv1.AppKey) error {