Delete stale session trackers

lib/services/local/sessiontracker.go

@@ -26,7 +26,9 @@ import (
 	"github.com/gravitational/teleport/lib/modules"
 	"github.com/gravitational/teleport/lib/services"
 	"github.com/gravitational/teleport/lib/utils"
+
 	"github.com/gravitational/trace"
+	"github.com/sirupsen/logrus"
 )
 
 const (
@@ -116,16 +118,43 @@ func (s *sessionTracker) GetActiveSessionTrackers(ctx context.Context) ([]types.
 		return nil, trace.Wrap(err)
 	}
 
-	sessions := make([]types.SessionTracker, len(result.Items))
-	for i, item := range result.Items {
+	sessions := make([]types.SessionTracker, 0, len(result.Items))
+
+	// We don't overallocate expired since cleaning up sessions here should be rare.
+	noExpiry := make([]backend.Item, 0)
+	now := time.Now().UTC()
+	for _, item := range result.Items {
 		session, err := unmarshalSession(item.Value)
 		if err != nil {
 			return nil, trace.Wrap(err)
 		}
 
-		sessions[i] = session
+		// NOTE: This is the session expiry timestamp, not the backend timestamp stored in `item.Expires`.
+		after := session.GetExpires().After(now)
+
+		switch {
+		case after:
+			// Keep any items that aren't expired.
+			sessions = append(sessions, session)
+		case !after && item.Expires.IsZero():
+			// Clear item if expiry is not set on the backend.
+			// We currently don't set the expiry here but we will when #11551 is merged.
+			noExpiry = append(noExpiry, item)
+		default:
+			// If we take this branch, the expiry is set and the backend is responsible for cleaning up the item.
+		}
 	}
 
+	go func() {
+		for _, item := range noExpiry {
+			if err := s.bk.Delete(ctx, item.Key); err != nil {
+				if !trace.IsNotFound(err) {
+					logrus.WithError(err).Error("Failed to remove stale session tracker")
+				}
+			}
+		}
+	}()
+
 	return sessions, nil
 }
 

lib/services/local/sessiontracker_test.go

@@ -16,6 +16,7 @@ package local
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/gravitational/teleport/api/client/proto"
@@ -47,6 +48,7 @@ func TestSessionTrackerStorage(t *testing.T) {
 			User: "eve",
 			Mode: string(types.SessionPeerMode),
 		},
+		Expires: time.Now().UTC().Add(24 * time.Hour),
 	})
 	require.NoError(t, err)
 
@@ -87,3 +89,52 @@ func TestSessionTrackerStorage(t *testing.T) {
 	require.Error(t, err)
 	require.Nil(t, session)
 }
+
+func TestSessionTrackerImplicitExpiry(t *testing.T) {
+	ctx := context.Background()
+	bk, err := memory.New(memory.Config{})
+	require.NoError(t, err)
+
+	id := uuid.New().String()
+	id2 := uuid.New().String()
+	srv, err := NewSessionTrackerService(bk)
+	require.NoError(t, err)
+
+	_, err = srv.CreateSessionTracker(ctx, &proto.CreateSessionTrackerRequest{
+		Namespace:   defaults.Namespace,
+		ID:          id,
+		Type:        types.KindSSHSession,
+		Hostname:    "hostname",
+		ClusterName: "cluster",
+		Login:       "root",
+		Initiator: &types.Participant{
+			ID:   uuid.New().String(),
+			User: "eve",
+			Mode: string(types.SessionPeerMode),
+		},
+		Expires: time.Now().UTC().Add(time.Second),
+	})
+	require.NoError(t, err)
+
+	_, err = srv.CreateSessionTracker(ctx, &proto.CreateSessionTrackerRequest{
+		Namespace:   defaults.Namespace,
+		ID:          id2,
+		Type:        types.KindSSHSession,
+		Hostname:    "hostname",
+		ClusterName: "cluster",
+		Login:       "root",
+		Initiator: &types.Participant{
+			ID:   uuid.New().String(),
+			User: "eve",
+			Mode: string(types.SessionPeerMode),
+		},
+		Expires: time.Now().UTC().Add(24 * time.Hour),
+	})
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		sessions, err := srv.GetActiveSessionTrackers(ctx)
+		require.NoError(t, err)
+		return len(sessions) == 1
+	}, time.Minute, time.Second)
+}