chore(cd-service): regularly delete old overview caches

charts/kuberpult/templates/cd-service.yaml

@@ -164,6 +164,8 @@ spec:
         - name: KUBERPULT_GIT_WEB_URL
           value: {{ .Values.git.webUrl | quote }}
 {{- if .Values.datadogTracing.enabled }}
+        - name: KUBERPULT_CACHE_TTL_HOURS
+          value: {{ .Values.cd.cacheTtlHours }}
         - name: DD_AGENT_HOST
           valueFrom:
             fieldRef:

charts/kuberpult/values.yaml

@@ -85,6 +85,7 @@ cd:
   allowLongAppNames: false
 # List of allowed domains that the links provided in releases, release trains and locks must match
   allowedDomains: ""
+  cacheTtlHours: 24
   service:
     annotations: {}
   pod:

pkg/db/overview.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/freiheit-com/kuberpult/pkg/api/v1"
 	"google.golang.org/protobuf/types/known/timestamppb"
+	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/tracer"
 )
 
 func (h *DBHandler) UpdateOverviewTeamLock(ctx context.Context, transaction *sql.Tx, teamLock TeamLock) error {
@@ -318,6 +319,42 @@ func (h *DBHandler) IsOverviewEmpty(overviewResp *api.GetOverviewResponse) bool
 	return false
 }
 
+func (h *DBHandler) DBDeleteOldOverviews(ctx context.Context, tx *sql.Tx, numberOfOverviewsToKeep uint64, timeThreshold time.Time) error {
+	span, _ := tracer.StartSpanFromContext(ctx, "DBDeleteOldOverviews")
+	defer span.Finish()
+
+	if h == nil {
+		return nil
+	}
+
+	if tx == nil {
+		return fmt.Errorf("attempting to delete overview caches without a transaction")
+	}
+
+	deleteQuery := h.AdaptQuery(`
+DELETE FROM overview_cache
+WHERE timestamp < ?
+AND eslversion NOT IN (
+    SELECT eslversion 
+	FROM overview_cache
+	ORDER BY eslversion DESC
+	LIMIT ?
+);
+`)
+	span.SetTag("query", deleteQuery)
+	span.SetTag("numberOfOverviewsToKeep", numberOfOverviewsToKeep)
+	span.SetTag("timeThreshold", timeThreshold)
+	_, err := tx.Exec(
+		deleteQuery,
+		timeThreshold.UTC(),
+		numberOfOverviewsToKeep,
+	)
+	if err != nil {
+		return fmt.Errorf("DBDeleteOldOverviews error executing query: %w", err)
+	}
+	return nil
+}
+
 func getEnvironmentByName(groups []*api.EnvironmentGroup, envNameToReturn string) *api.Environment {
 	for _, currentGroup := range groups {
 		for _, currentEnv := range currentGroup.Environments {

pkg/db/overview_test.go

@@ -1554,3 +1554,213 @@ func TestDeriveUndeploySummary(t *testing.T) {
 		})
 	}
 }
+
+func TestDBDeleteOldOverview(t *testing.T) {
+	upstreamLatest := true
+	dev := "dev"
+	var tcs = []struct {
+		Name                               string
+		inputOverviews                     []*api.GetOverviewResponse
+		timeThresholdDiff                  time.Duration
+		numberOfOverviewsToKeep            uint64
+		expectedNumberOfRemainingOverviews uint64
+	}{
+		{
+			Name: "4 overviews, should keep two",
+			inputOverviews: []*api.GetOverviewResponse{
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{
+					EnvironmentGroups: []*api.EnvironmentGroup{
+						{
+							EnvironmentGroupName: "dev",
+							Environments: []*api.Environment{
+								{
+									Name: "development",
+									Config: &api.EnvironmentConfig{
+										Upstream: &api.EnvironmentConfig_Upstream{
+											Latest: &upstreamLatest,
+										},
+										Argocd:           &api.EnvironmentConfig_ArgoCD{},
+										EnvironmentGroup: &dev,
+									},
+									Applications: map[string]*api.Environment_Application{
+										"test": {
+											Name:    "test",
+											Version: 1,
+											DeploymentMetaData: &api.Environment_Application_DeploymentMetaData{
+												DeployAuthor: "testmail@example.com",
+												DeployTime:   "1",
+											},
+											Team: "team-123",
+										},
+									},
+									Priority: api.Priority_YOLO,
+								},
+							},
+							Priority: api.Priority_YOLO,
+						},
+					},
+					Applications: map[string]*api.Application{
+						"test": {
+							Name: "test",
+							Releases: []*api.Release{
+								{
+									Version:        1,
+									SourceCommitId: "changedcommitId",
+									SourceAuthor:   "changedAuthor",
+									SourceMessage:  "changed changed something (#679)",
+									PrNumber:       "679",
+									CreatedAt:      &timestamppb.Timestamp{Seconds: 1, Nanos: 1},
+								},
+							},
+							Team: "team-123",
+						},
+					},
+					GitRevision: "0",
+				},
+			},
+			timeThresholdDiff:                  150 * time.Second,
+			numberOfOverviewsToKeep:            2,
+			expectedNumberOfRemainingOverviews: 2,
+		},
+		{
+			Name: "4 overviews, early time threshhold, all should remain",
+			inputOverviews: []*api.GetOverviewResponse{
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{
+					EnvironmentGroups: []*api.EnvironmentGroup{
+						{
+							EnvironmentGroupName: "dev",
+							Environments: []*api.Environment{
+								{
+									Name: "development",
+									Config: &api.EnvironmentConfig{
+										Upstream: &api.EnvironmentConfig_Upstream{
+											Latest: &upstreamLatest,
+										},
+										Argocd:           &api.EnvironmentConfig_ArgoCD{},
+										EnvironmentGroup: &dev,
+									},
+									Applications: map[string]*api.Environment_Application{
+										"test": {
+											Name:    "test",
+											Version: 1,
+											DeploymentMetaData: &api.Environment_Application_DeploymentMetaData{
+												DeployAuthor: "testmail@example.com",
+												DeployTime:   "1",
+											},
+											Team: "team-123",
+										},
+									},
+									Priority: api.Priority_YOLO,
+								},
+							},
+							Priority: api.Priority_YOLO,
+						},
+					},
+					Applications: map[string]*api.Application{
+						"test": {
+							Name: "test",
+							Releases: []*api.Release{
+								{
+									Version:        1,
+									SourceCommitId: "changedcommitId",
+									SourceAuthor:   "changedAuthor",
+									SourceMessage:  "changed changed something (#679)",
+									PrNumber:       "679",
+									CreatedAt:      &timestamppb.Timestamp{Seconds: 1, Nanos: 1},
+								},
+							},
+							Team: "team-123",
+						},
+					},
+					GitRevision: "0",
+				},
+			},
+			timeThresholdDiff:                  -300 * time.Second,
+			numberOfOverviewsToKeep:            0,
+			expectedNumberOfRemainingOverviews: 4,
+		},
+		{
+			Name: "4 overviews, late time threshold, zero to remain",
+			inputOverviews: []*api.GetOverviewResponse{
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+				&api.GetOverviewResponse{},
+			},
+			timeThresholdDiff:                  300 * time.Second,
+			numberOfOverviewsToKeep:            0,
+			expectedNumberOfRemainingOverviews: 0,
+		},
+	}
+	for _, tc := range tcs {
+		tc := tc
+		t.Run(tc.Name, func(t *testing.T) {
+			t.Parallel()
+			ctx := testutil.MakeTestContext()
+			dbHandler := setupDB(t)
+
+			err := dbHandler.WithTransaction(ctx, false, func(ctx context.Context, transaction *sql.Tx) error {
+				for _, overview := range tc.inputOverviews {
+					err := dbHandler.WriteOverviewCache(ctx, transaction, overview)
+					if err != nil {
+						return err
+					}
+				}
+				err := dbHandler.DBDeleteOldOverviews(ctx, transaction, tc.numberOfOverviewsToKeep, time.Now().Add(tc.timeThresholdDiff))
+				if err != nil {
+					return err
+				}
+				remainingOverviewsCount, err := calculateNumberOfOverviews(dbHandler, ctx, transaction)
+				if err != nil {
+					return err
+				}
+				if remainingOverviewsCount != tc.expectedNumberOfRemainingOverviews {
+					return fmt.Errorf("Expected number of remaining overviews: %d, got: %d", tc.expectedNumberOfRemainingOverviews, remainingOverviewsCount)
+				}
+				if tc.expectedNumberOfRemainingOverviews > 0 {
+					latestOverview, err := dbHandler.ReadLatestOverviewCache(ctx, transaction)
+					if err != nil {
+						return err
+					}
+					opts := getOverviewIgnoredTypes()
+					if diff := cmp.Diff(tc.inputOverviews[len(tc.inputOverviews)-1], latestOverview, opts); diff != "" {
+						return fmt.Errorf("mismatch latest overview (-want +got):\n%s", diff)
+					}
+				}
+				return nil
+			})
+
+			if err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func calculateNumberOfOverviews(h *DBHandler, ctx context.Context, tx *sql.Tx) (uint64, error) {
+
+	selectQuery := h.AdaptQuery(`SELECT COUNT(*) FROM overview_cache`)
+	rows, err := tx.QueryContext(
+		ctx,
+		selectQuery,
+	)
+	var result int64
+	if err != nil {
+		return 0, fmt.Errorf("error calculating number of overviews: %w", err)
+	}
+	if rows.Next() {
+		err := rows.Scan(&result)
+		if err != nil {
+			return 0, fmt.Errorf("Error scanning overview_cache ,Error: %w\n", err)
+		}
+	} else {
+		result = 0
+	}
+	return uint64(result), nil
+}

services/cd-service/pkg/cmd/server.go

@@ -107,6 +107,7 @@ type Config struct {
 	DbSslMode            string   `default:"verify-full" split_words:"true"`
 	MinorRegexes         string   `default:"" split_words:"true"`
 	AllowedDomains       []string `split_words:"true"`
+	CacheTtlHours        uint     `default:"24" split_words:"true"`
 
 	DisableQueue bool `required:"true" split_words:"true"`
 }
@@ -468,6 +469,15 @@ func RunServer() {
 						return nil
 					},
 				},
+				{
+					Shutdown: nil,
+					Name:     "cache cleanup",
+					Run: func(ctx context.Context, reporter *setup.HealthReporter) error {
+						reporter.ReportReady("Cache cleanup started")
+						repository.RegularlyCleanupOverviewCache(ctx, repo, 3600, c.CacheTtlHours)
+						return nil
+					},
+				},
 				{
 					Shutdown: nil,
 					Name:     "push queue",

services/cd-service/pkg/repository/transformer.go

@@ -281,6 +281,26 @@ func GetRepositoryStateAndUpdateMetrics(ctx context.Context, repo Repository) {
 	}
 }
 
+func RegularlyCleanupOverviewCache(ctx context.Context, repo Repository, interval time.Duration, cacheTtlHours uint) {
+	cleanupEventTimer := time.NewTicker(interval * time.Second)
+	for range cleanupEventTimer.C {
+		logger.FromContext(ctx).Sugar().Warn("Cleaning up old overview caches")
+		s := repo.State()
+		if s.DBHandler.ShouldUseOtherTables() {
+			err := s.DBHandler.WithTransaction(ctx, false, func(ctx context.Context, transaction *sql.Tx) error {
+				err := s.DBHandler.DBDeleteOldOverviews(ctx, transaction, 5, time.Now().Add(-time.Duration(cacheTtlHours)*time.Hour))
+				if err != nil {
+					return err
+				}
+				return nil
+			})
+			if err != nil {
+				panic(err.Error())
+			}
+		}
+	}
+}
+
 // A Transformer updates the files in the worktree
 type Transformer interface {
 	Transform(ctx context.Context, state *State, t TransformerContext, transaction *sql.Tx) (commitMsg string, e error)