Move retention at the start of the compactor loop.

Signed-off-by: Olivier Biesmans <[email protected]>

CHANGELOG.md

@@ -17,6 +17,7 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 ### Added
 - [#1852](https://github.com/thanos-io/thanos/pull/1852) Add support for `AWS_CONTAINER_CREDENTIALS_FULL_URI` by upgrading to minio-go v6.0.44
 - [#1854](https://github.com/thanos-io/thanos/pull/1854) Update Rule UI to support alerts count displaying and filtering.
+- [#1735](https://github.com/thanos-io/thanos/pull/1735) Retention policies are now applied at the start of the compactor loop. Retentions policies that break downsampling are now logged as warnings.
 
 ## [v0.9.0](https://github.com/thanos-io/thanos/releases/tag/v0.9.0) - 2019.12.03
 

cmd/thanos/compact.go

@@ -266,17 +266,35 @@ func runCompact(
 		return errors.Wrap(err, "create bucket compactor")
 	}
 
-	if retentionByResolution[compact.ResolutionLevelRaw].Seconds() != 0 {
+	if gretentionByResolution[compact.ResolutionLevelRaw].Seconds() != 0 {
 		level.Info(logger).Log("msg", "retention policy of raw samples is enabled", "duration", retentionByResolution[compact.ResolutionLevelRaw])
+		if retentionByResolution[compact.ResolutionLevelRaw].Seconds() <= downsample.DownsampleRange0 {
+			level.Warn(logger).Log("msg", "retention policy of raw samples is too low to generate 5 min aggregated samples", "duration", retentionByResolution[compact.ResolutionLevelRaw])
+		}
 	}
 	if retentionByResolution[compact.ResolutionLevel5m].Seconds() != 0 {
 		level.Info(logger).Log("msg", "retention policy of 5 min aggregated samples is enabled", "duration", retentionByResolution[compact.ResolutionLevel5m])
+		if retentionByResolution[compact.ResolutionLevel5m].Seconds() <= downsample.DownsampleRange1 {
+			level.Warn(logger).Log("msg", "retention policy of 5 min aggregated samples is too low to generate 1h aggregated samples", "duration", retentionByResolution[compact.ResolutionLevel5m])
+		}
 	}
 	if retentionByResolution[compact.ResolutionLevel1h].Seconds() != 0 {
 		level.Info(logger).Log("msg", "retention policy of 1 hour aggregated samples is enabled", "duration", retentionByResolution[compact.ResolutionLevel1h])
 	}
 
 	f := func() error {
+		// At first, we need at least to be able to genetes 1h resolution metrics
+		// from the lower resolutions, so remove only up to 1h resolution retention.
+		initialRetention := map[compact.ResolutionLevel]time.Duration{
+			compact.ResolutionLevelRaw: initialRetention[compact.ResolutionLevel1h],
+			compact.ResolutionLevel5m:  initialRetention[compact.ResolutionLevel1h],
+			compact.ResolutionLevel1h:  initialRetention[compact.ResolutionLevel1h],
+		}
+
+		if err := compact.ApplyRetentionPolicyByResolution(ctx, logger, bkt, initialRetention); err != nil {
+			return errors.Wrap(err, fmt.Sprintf("retention failed"))
+		}
+
 		if err := compactor.Compact(ctx); err != nil {
 			return errors.Wrap(err, "compaction failed")
 		}
@@ -306,6 +324,7 @@ func runCompact(
 		if err := compact.ApplyRetentionPolicyByResolution(ctx, logger, bkt, retentionByResolution); err != nil {
 			return errors.Wrap(err, fmt.Sprintf("retention failed"))
 		}
+
 		return nil
 	}