Merge branch 'v0.9.x' into develop

Author: pskrbasu

CHANGELOG.md

@@ -1,3 +1,20 @@
+## v0.9.0 [2025-07-02]
+_Whats new_
+
+- Refactor collection state to support time ranges, enabling `--to` flag support. ([#241](https://github.com/turbot/tailpipe-plugin-sdk/issues/241)) 
+  - TimeRangeCollectionState supports array of time ranges
+  - ShouldCollect ensures that the gaps between the ranges are filled but we do not collect for times we have already collected
+  - The time ranges, of type DirectionalTimeRange, are direction aware and work for collection in forwards or backwards direction
+  - A Clear function allows clearing the state for a specified time range
+  - Move all persistence logic into SaveableCollectionState
+  - Added `overwrite` parameter to CollectRequest - if set, clear collection state becore collecting
+  - Added migration support for legacy collection states
+
+_Bug fixes_
+* Fix issue where collection state end-objects are cleared when collection is complete,
+  meaning no further data will be collected for that day. ([#250](https://github.com/turbot/tailpipe-plugin-sdk/issues/250))
+
+
 ## v0.8.0 [2025-06-23]
 _Whats new_
 * Remove row validation and rely entirely on CLI to execute validation. ([#202](https://github.com/turbot/tailpipe-plugin-sdk/issues/202))

artifact_source/artifact_source_impl.go

@@ -538,8 +538,10 @@ func (a *ArtifactSourceImpl[S, T]) getMetadataAndApplyFilters(targetPath string,
 	satisfied := match && metadataSatisfiesFilters(metadata, filterMap)
 
 	// if we have a from time, check whether that excludes this directory
-	if satisfied && isDir && !a.FromTime.IsZero() {
-		satisfied = dirSatisfiesFromTime(a.FromTime, metadata)
+	// NOTE: we only support forward collection for artifacts, so we only check the LowerBoundary
+	// - we do not need the directional logic here
+	if satisfied && isDir {
+		satisfied = dirSatisfiesFromTime(a.CollectionTimeRange.LowerBoundary, metadata)
 	}
 
 	return metadata, satisfied, nil
@@ -570,12 +572,14 @@ func (a *ArtifactSourceImpl[S, T]) walkFileNode(ctx context.Context, targetPath
 
 	// if the artifact has a timestamp, check the from and to time
 	if !artifactInfo.Timestamp.IsZero() {
+		from := a.CollectionTimeRange.LowerBoundary
+		to := a.CollectionTimeRange.UpperBoundary
 		// if we have a from time, check if the artifact is newer than the from time
-		if !a.FromTime.IsZero() && artifactInfo.Timestamp.Compare(a.FromTime) < 0 {
+		if !from.IsZero() && artifactInfo.Timestamp.Compare(from) < 0 {
 			return nil
 		}
 		// if we have a to time, check if the artifact is older than the to time
-		if !a.ToTime.IsZero() && artifactInfo.Timestamp.Compare(a.ToTime) > 0 {
+		if !to.IsZero() && artifactInfo.Timestamp.Compare(to) > 0 {
 			return nil
 		}
 	}

artifact_source/nil_artifact_collection_state.go

@@ -21,7 +21,7 @@ func (s *NilArtifactCollectionState) GetToTime() time.Time {
 func (s *NilArtifactCollectionState) SetEndTime(_ time.Time) {
 }
 
-func (*NilArtifactCollectionState) Init(collection_state.CollectionTimeRange, time.Duration) {
+func (*NilArtifactCollectionState) Init(collection_state.DirectionalTimeRange, time.Duration) {
 }
 
 func (s *NilArtifactCollectionState) RegisterPath(_ string, _ map[string]string) {
@@ -53,5 +53,5 @@ func (*NilArtifactCollectionState) MigrateFromLegacyState(_ []byte) error {
 func (*NilArtifactCollectionState) Validate() error {
 	return nil
 }
-func (*NilArtifactCollectionState) Clear(_ collection_state.CollectionTimeRange) {
+func (*NilArtifactCollectionState) Clear(_ collection_state.DirectionalTimeRange) {
 }

artifact_source/plugin_source_wrapper.go

@@ -87,10 +87,15 @@ func (w *PluginSourceWrapper) Init(ctx context.Context, params *row_source.RowSo
 	if err != nil {
 		return err
 	}
+	// the init call returns the resolved from time - use to store on the wrapper
 	// set the from time
 	fromTime := row_source.ResolvedFromTimeFromProto(resp.FromTime)
 
-	w.FromTime = fromTime.Time
+	// store the collection time range
+	w.CollectionTimeRange = collection_state.DirectionalTimeRange{
+		LowerBoundary: fromTime.Time,
+		UpperBoundary: params.To,
+	}
 	w.FromTimeSource = fromTime.Source
 
 	return nil

collection_state/artifact_collection_state.go

@@ -26,7 +26,7 @@ type ArtifactCollectionState struct {
 	granularity time.Duration
 
 	// the time range for the underway collection - populated by Init
-	currentCollectionTimeRange *CollectionTimeRange
+	currentCollectionTimeRange *DirectionalTimeRange
 	// map of the trunk state for each object which has been passed to ShouldCollect
 	// this is to avoid recomputing the trunk state for each object on every OnCollected call
 	objectTrunkMap map[string]*TimeRangeCollectionState
@@ -40,7 +40,7 @@ func NewArtifactCollectionState() CollectionState {
 }
 
 // Init sets the filepath of the collection state and loads the state from the file if it exists
-func (s *ArtifactCollectionState) Init(collectionTimeRange CollectionTimeRange, granularity time.Duration) {
+func (s *ArtifactCollectionState) Init(collectionTimeRange DirectionalTimeRange, granularity time.Duration) {
 	// ensure the granularity is no smaller than the minimum
 	if granularity < MinArtifactGranularity && granularity != 0 {
 		granularity = MinArtifactGranularity
@@ -79,7 +79,6 @@ func (s *ArtifactCollectionState) GetFromTime() time.Time {
 // (we may have collected some data after this - within the granularity period)
 // return the earliest end time of all the trunk states
 func (s *ArtifactCollectionState) GetToTime() time.Time {
-	// TODO #CS KAI think about continuation for reverse order
 	// find the earliest end time of all the trunk states
 	var endTime time.Time
 	for _, trunkState := range s.TrunkStates {
@@ -223,7 +222,7 @@ func (s *ArtifactCollectionState) IsEmpty() bool {
 	return true
 }
 
-func (s *ArtifactCollectionState) Clear(timeRange CollectionTimeRange) {
+func (s *ArtifactCollectionState) Clear(timeRange DirectionalTimeRange) {
 	for _, trunkState := range s.TrunkStates {
 		if trunkState == nil {
 			continue

collection_state/artifact_collection_state_test.go

@@ -12,21 +12,6 @@ func TestArtifactCollectionState_MigrateFromLegacyState(t *testing.T) {
 		legacy   *ArtifactCollectionStateLegacy
 		expected *ArtifactCollectionState
 	}{
-		{
-			name: "migrate two trunks with different orders",
-			legacy: buildArtifactCollectionStateLegacy(map[string]*TimeRangeCollectionStateLegacy{
-				"/trunk1": buildTimeRangeCollectionStateLegacy("2023-10-01 00:00:00", "2023-12-01 01:00:00", time.Hour*24, CollectionOrderChronological, "object1", "object2"),
-				"/trunk2": buildTimeRangeCollectionStateLegacy("2023-11-01 00:00:00", "2023-11-30 00:00:00", time.Hour*24, CollectionOrderReverse, "object3"),
-			}, timeString("2023-12-01 12:00:00")),
-			expected: buildArtifactCollectionState(map[string]*TimeRangeCollectionState{
-				"/trunk1": buildTimeRangeCollectionState(CollectionOrderChronological, time.Hour*24,
-					buildTimeRangeState("2023-10-01 00:00:00", "2023-11-30 01:00:00", time.Hour*24, CollectionOrderChronological, "object1", "object2"),
-				),
-				"/trunk2": buildTimeRangeCollectionState(CollectionOrderReverse, time.Hour*24,
-					buildTimeRangeState("2023-11-30 00:00:00", "2023-11-01 00:00:00", time.Hour*24, CollectionOrderReverse, "object3"),
-				),
-			}, time.Hour*24),
-		},
 		{
 			name:     "empty trunks",
 			legacy:   buildArtifactCollectionStateLegacy(map[string]*TimeRangeCollectionStateLegacy{}, timeString("2023-12-01 12:00:00")),
@@ -46,7 +31,8 @@ func TestArtifactCollectionState_MigrateFromLegacyState(t *testing.T) {
 			}, timeString("2023-12-01 12:00:00")),
 			expected: buildArtifactCollectionState(map[string]*TimeRangeCollectionState{
 				"/trunk1": buildTimeRangeCollectionState(CollectionOrderChronological, time.Hour*24,
-					buildTimeRangeState("2023-10-01 00:00:00", "2023-11-30 01:00:00", time.Hour*24, CollectionOrderChronological),
+
+					buildTimeRangeState("2023-10-01 00:00:00", "2023-12-01 01:00:00", time.Hour*24, CollectionOrderChronological),
 				),
 			}, time.Hour*24),
 		},
@@ -57,7 +43,7 @@ func TestArtifactCollectionState_MigrateFromLegacyState(t *testing.T) {
 			}, timeString("2023-12-01 12:00:00")),
 			expected: buildArtifactCollectionState(map[string]*TimeRangeCollectionState{
 				"/trunk1": buildTimeRangeCollectionState(CollectionOrderReverse, time.Hour*24,
-					buildTimeRangeState("2023-12-01 01:00:00", "2023-10-01 00:00:00", time.Hour*24, CollectionOrderReverse, "object1", "object2"),
+					buildTimeRangeState("2023-10-01 00:00:00", "2023-12-01 01:00:00", time.Hour*24, CollectionOrderReverse, "object1", "object2"),
 				),
 			}, time.Hour*24),
 		},
@@ -68,7 +54,7 @@ func TestArtifactCollectionState_MigrateFromLegacyState(t *testing.T) {
 			}, timeString("2023-12-01 12:00:00")),
 			expected: buildArtifactCollectionState(map[string]*TimeRangeCollectionState{
 				"/trunk1": buildTimeRangeCollectionState(CollectionOrderChronological, time.Hour*24,
-					buildTimeRangeState("2023-10-01 00:00:00", "2023-10-01 00:00:00", time.Hour*24, CollectionOrderChronological, "object1"),
+					buildTimeRangeState("2023-10-01 00:00:00", "2023-10-02 00:00:00", time.Hour*24, CollectionOrderChronological, "object1"),
 				),
 			}, time.Hour*24),
 		},
@@ -102,6 +88,7 @@ func buildArtifactCollectionStateLegacy(trunks map[string]*TimeRangeCollectionSt
 	}
 }
 
+// TODO update this to set end time on or after last entry time to me more realistic
 // buildTimeRangeCollectionStateLegacy constructs a legacy time range collection state for tests
 func buildTimeRangeCollectionStateLegacy(fromStr, toStr string, granularity time.Duration, order CollectionOrder, endObjects ...string) *TimeRangeCollectionStateLegacy {
 	from, err := time.Parse("2006-01-02 15:04:05", fromStr)