fix dashboard version cleanup on large datasets

pkg/services/sqlstore/dashboard_version.go

 package sqlstore
 
 import (
-	"strings"
+	"fmt"
 
 	"github.com/grafana/grafana/pkg/bus"
 	m "github.com/grafana/grafana/pkg/models"
@@ -69,36 +69,31 @@ func GetDashboardVersions(query *m.GetDashboardVersionsQuery) error {
 
 func DeleteExpiredVersions(cmd *m.DeleteExpiredVersionsCommand) error {
 	return inTransaction(func(sess *DBSession) error {
-		versions := []DashboardVersionExp{}
 		versionsToKeep := setting.DashboardVersionsToKeep
-
 		if versionsToKeep < 1 {
 			versionsToKeep = 1
 		}
 
-		err := sess.Table("dashboard_version").
-			Select("dashboard_version.id, dashboard_version.version, dashboard_version.dashboard_id").
-			Where(`dashboard_id IN (
-			SELECT dashboard_id FROM dashboard_version
-			GROUP BY dashboard_id HAVING COUNT(dashboard_version.id) > ?
-		)`, versionsToKeep).
-			Desc("dashboard_version.dashboard_id", "dashboard_version.version").
-			Find(&versions)
-
-		if err != nil {
-			return err
-		}
-
-		// Keep last versionsToKeep versions and delete other
-		versionIdsToDelete := getVersionIDsToDelete(versions, versionsToKeep)
-		if len(versionIdsToDelete) > 0 {
-			deleteExpiredSql := `DELETE FROM dashboard_version WHERE id IN (?` + strings.Repeat(",?", len(versionIdsToDelete)-1) + `)`
-			expiredResponse, err := sess.Exec(deleteExpiredSql, versionIdsToDelete...)
+		// Idea of this query is finding version IDs to delete based on formula:
+		// min_version_to_keep = min_version + (versions_count - versions_to_keep)
+		// where version stats is processed for each dashboard. This guarantees that we keep at least versions_to_keep
+		// versions, but in some cases (when versions are sparse) this number may be more.
+		versionIdsToDeleteSybqueryTemplate := `SELECT id
+			FROM dashboard_version, (
+				SELECT dashboard_id, count(version) as count, min(version) as min
+				FROM dashboard_version
+				GROUP BY dashboard_id
+			) AS vtd
+			WHERE dashboard_version.dashboard_id=vtd.dashboard_id
+				AND version < vtd.min + vtd.count - %v`
+
+		versionIdsToDeleteSubquery := fmt.Sprintf(versionIdsToDeleteSybqueryTemplate, versionsToKeep)
+		deleteExpiredSql := fmt.Sprintf(`DELETE FROM dashboard_version WHERE id IN (%s)`, versionIdsToDeleteSubquery)
+		expiredResponse, err := sess.Exec(deleteExpiredSql)
 		if err != nil {
 			return err
 		}
 		cmd.DeletedRows, _ = expiredResponse.RowsAffected()
-		}
 
 		return nil
 	})

pkg/services/sqlstore/dashboard_version_test.go

@@ -136,7 +136,7 @@ func TestDeleteExpiredVersions(t *testing.T) {
 			err := DeleteExpiredVersions(&m.DeleteExpiredVersionsCommand{})
 			So(err, ShouldBeNil)
 
-			query := m.GetDashboardVersionsQuery{DashboardId: savedDash.Id, OrgId: 1}
+			query := m.GetDashboardVersionsQuery{DashboardId: savedDash.Id, OrgId: 1, Limit: versionsToWrite}
 			GetDashboardVersions(&query)
 
 			So(len(query.Result), ShouldEqual, versionsToWrite)