Skip to content

Commit 26030c3

Browse files
davidlin20devdavidlin20dev
committed
feat(executor): add garbage collector Prometheus metrics
Emit objects_deleted, deletion_errors, and sweep_duration under a dedicated executor:gc: sub-scope, created once in the constructor and updated during each collect() sweep. Unit tests assert the counters move and the sweep is timed. Part of #7455. Signed-off-by: davidlin20dev <davidlin20.dev@gmail.com>
1 parent c629831 commit 26030c3

5 files changed

Lines changed: 133 additions & 7 deletions

File tree

‎executor/pkg/controller/garbage_collector.go‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,26 +4,48 @@ import (
44
"context"
55
"time"
66

7+
"github.com/prometheus/client_golang/prometheus"
78
"sigs.k8s.io/controller-runtime/pkg/client"
89
"sigs.k8s.io/controller-runtime/pkg/log"
910

1011
flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1"
12+
"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
1113
)
1214

15+
// gcMetrics holds the Prometheus instruments for the garbage collector. They are
16+
// created once (in newGCMetrics) and only updated thereafter, never re-registered.
17+
type gcMetrics struct {
18+
deleted prometheus.Counter
19+
errors prometheus.Counter
20+
sweepTime promutils.StopWatch
21+
}
22+
23+
// newGCMetrics builds the garbage collector instruments under the given scope.
24+
// Call exactly once per scope to avoid duplicate registration panics.
25+
func newGCMetrics(scope promutils.Scope) gcMetrics {
26+
return gcMetrics{
27+
deleted: scope.MustNewCounter("objects_deleted", "Total TaskActions deleted by the garbage collector"),
28+
errors: scope.MustNewCounter("deletion_errors", "Total errors encountered while deleting expired TaskActions"),
29+
sweepTime: scope.MustNewStopWatch("sweep_duration", "Duration of a full garbage collection sweep", time.Millisecond),
30+
}
31+
}
32+
1333
// GarbageCollector periodically deletes terminal TaskActions that have exceeded their TTL.
1434
// It implements the controller-runtime manager.Runnable interface.
1535
type GarbageCollector struct {
1636
client client.Client
1737
interval time.Duration
1838
maxTTL time.Duration
39+
metrics gcMetrics
1940
}
2041

2142
// NewGarbageCollector creates a new GarbageCollector.
22-
func NewGarbageCollector(c client.Client, interval, maxTTL time.Duration) *GarbageCollector {
43+
func NewGarbageCollector(c client.Client, interval, maxTTL time.Duration, scope promutils.Scope) *GarbageCollector {
2344
return &GarbageCollector{
2445
client: c,
2546
interval: interval,
2647
maxTTL: maxTTL,
48+
metrics: newGCMetrics(scope),
2749
}
2850
}
2951

@@ -55,6 +77,10 @@ const gcPageSize = 500
5577
func (gc *GarbageCollector) collect(ctx context.Context) error {
5678
logger := log.FromContext(ctx).WithName("gc")
5779

80+
// Time the full sweep, defer guarantees it records on every return path.
81+
timer := gc.metrics.sweepTime.Start()
82+
defer timer.Stop()
83+
5884
cutoff := time.Now().UTC().Add(-gc.maxTTL).Format(labelTimeFormat)
5985
deleted := 0
6086
total := 0
@@ -87,11 +113,13 @@ func (gc *GarbageCollector) collect(ctx context.Context) error {
87113
// The minute-precision format is lexicographically ordered, so string comparison works.
88114
if completedTime < cutoff {
89115
if err := gc.client.Delete(ctx, ta); err != nil {
116+
gc.metrics.errors.Inc()
90117
logger.Error(err, "failed to delete expired TaskAction",
91118
"name", ta.Name, "namespace", ta.Namespace, "completedTime", completedTime)
92119
continue
93120
}
94121
deleted++
122+
gc.metrics.deleted.Inc()
95123
}
96124
}
97125

‎executor/pkg/controller/garbage_collector_metrics_test.go‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
package controller
2+
3+
import (
4+
"context"
5+
"errors"
6+
"testing"
7+
"time"
8+
9+
"github.com/prometheus/client_golang/prometheus"
10+
"github.com/prometheus/client_golang/prometheus/testutil"
11+
dto "github.com/prometheus/client_model/go"
12+
"github.com/stretchr/testify/require"
13+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
14+
"k8s.io/apimachinery/pkg/runtime"
15+
"sigs.k8s.io/controller-runtime/pkg/client"
16+
"sigs.k8s.io/controller-runtime/pkg/client/fake"
17+
"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
18+
19+
flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1"
20+
"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
21+
)
22+
23+
// newExpiredTaskAction builds a terminated TaskAction whose completed time is well past
24+
// any sane maxTTL, so the GC will treat it as expired and delete it.
25+
func newExpiredTaskAction() *flyteorgv1.TaskAction {
26+
expired := time.Now().UTC().Add(-2 * time.Hour).Format(labelTimeFormat)
27+
return &flyteorgv1.TaskAction{
28+
ObjectMeta: metav1.ObjectMeta{
29+
Name: "gc-expired",
30+
Namespace: "default",
31+
Labels: map[string]string{
32+
LabelTerminationStatus: LabelValueTerminated,
33+
LabelCompletedTime: expired,
34+
},
35+
},
36+
}
37+
}
38+
39+
// newGCTestClient returns an in-memory fake client seeded with one expired TaskAction.
40+
// If deleteErr is non-nil, every Delete fails with it, so we can exercise the error path.
41+
func newGCTestClient(t *testing.T, deleteErr error) client.Client {
42+
scheme := runtime.NewScheme()
43+
require.NoError(t, flyteorgv1.AddToScheme(scheme))
44+
builder := fake.NewClientBuilder().WithScheme(scheme).WithObjects(newExpiredTaskAction())
45+
if deleteErr != nil {
46+
builder = builder.WithInterceptorFuncs(interceptor.Funcs{
47+
Delete: func(context.Context, client.WithWatch, client.Object, ...client.DeleteOption) error {
48+
return deleteErr
49+
},
50+
})
51+
}
52+
return builder.Build()
53+
}
54+
55+
// sweepObservations reports how many times the sweep_duration stopwatch has recorded.
56+
// sweep_duration is a Summary, so testutil.ToFloat64 can not read it.
57+
// Instead, we collect the metric and read its sample count directly.
58+
func sweepObservations(t *testing.T, gc *GarbageCollector) uint64 {
59+
t.Helper()
60+
ch := make(chan prometheus.Metric, 1)
61+
gc.metrics.sweepTime.Observer.(prometheus.Collector).Collect(ch)
62+
close(ch)
63+
var m dto.Metric
64+
require.NoError(t, (<-ch).Write(&m))
65+
return m.GetSummary().GetSampleCount()
66+
}
67+
68+
// TestGarbageCollectorDeletedMetric: a successful delete moves objects_deleted
69+
// from 0 to 1, records one sweep, and leaves deletion_errors at 0.
70+
func TestGarbageCollectorDeletedMetric(t *testing.T) {
71+
gc := NewGarbageCollector(newGCTestClient(t, nil), time.Minute, time.Hour, promutils.NewTestScope())
72+
73+
require.Equal(t, 0.0, testutil.ToFloat64(gc.metrics.deleted))
74+
require.Equal(t, 0.0, testutil.ToFloat64(gc.metrics.errors))
75+
76+
require.NoError(t, gc.collect(context.Background()))
77+
78+
require.Equal(t, 1.0, testutil.ToFloat64(gc.metrics.deleted))
79+
require.Equal(t, 0.0, testutil.ToFloat64(gc.metrics.errors))
80+
require.GreaterOrEqual(t, sweepObservations(t, gc), uint64(1))
81+
}
82+
83+
// TestGarbageCollectorDeletionErrorMetric: when a delete fails, deletion_errors moves
84+
// 0 to 1 and objects_deleted stays at 0.
85+
func TestGarbageCollectorDeletionErrorMetric(t *testing.T) {
86+
gc := NewGarbageCollector(
87+
newGCTestClient(t, errors.New("simulated delete failure")),
88+
time.Minute, time.Hour, promutils.NewTestScope(),
89+
)
90+
91+
require.Equal(t, 0.0, testutil.ToFloat64(gc.metrics.errors))
92+
93+
require.NoError(t, gc.collect(context.Background()))
94+
95+
require.Equal(t, 1.0, testutil.ToFloat64(gc.metrics.errors))
96+
require.Equal(t, 0.0, testutil.ToFloat64(gc.metrics.deleted))
97+
}

‎executor/pkg/controller/garbage_collector_test.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"sigs.k8s.io/controller-runtime/pkg/client"
1212

1313
flyteorgv1 "github.com/flyteorg/flyte/v2/executor/api/v1"
14+
"github.com/flyteorg/flyte/v2/flytestdlib/promutils"
1415
)
1516

1617
func createTaskAction(ctx context.Context, name string, labels map[string]string) *flyteorgv1.TaskAction {
@@ -62,7 +63,7 @@ var _ = Describe("GarbageCollector", func() {
6263
LabelCompletedTime: expiredTime,
6364
})
6465

65-
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour)
66+
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour, promutils.NewTestScope())
6667
Expect(gc.collect(ctx)).To(Succeed())
6768

6869
ta := &flyteorgv1.TaskAction{}
@@ -78,7 +79,7 @@ var _ = Describe("GarbageCollector", func() {
7879
LabelCompletedTime: recentTime,
7980
})
8081

81-
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour)
82+
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour, promutils.NewTestScope())
8283
Expect(gc.collect(ctx)).To(Succeed())
8384

8485
ta := &flyteorgv1.TaskAction{}
@@ -89,7 +90,7 @@ var _ = Describe("GarbageCollector", func() {
8990
It("should retain non-terminated TaskActions", func() {
9091
createTaskAction(ctx, "gc-active", nil)
9192

92-
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour)
93+
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour, promutils.NewTestScope())
9394
Expect(gc.collect(ctx)).To(Succeed())
9495

9596
ta := &flyteorgv1.TaskAction{}
@@ -98,7 +99,7 @@ var _ = Describe("GarbageCollector", func() {
9899
})
99100

100101
It("should handle empty list gracefully", func() {
101-
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour)
102+
gc := NewGarbageCollector(k8sClient, 1*time.Minute, 1*time.Hour, promutils.NewTestScope())
102103
Expect(gc.collect(ctx)).To(Succeed())
103104
})
104105
})

‎executor/setup.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,7 +189,7 @@ func Setup(ctx context.Context, sc *app.SetupContext) error {
189189
if cfg.GC.MaxTTL.Duration <= 0 {
190190
return fmt.Errorf("executor: gc.maxTTL must be positive when gc is enabled, got %v", cfg.GC.MaxTTL.Duration)
191191
}
192-
gc := controller.NewGarbageCollector(mgr.GetClient(), cfg.GC.Interval.Duration, cfg.GC.MaxTTL.Duration)
192+
gc := controller.NewGarbageCollector(mgr.GetClient(), cfg.GC.Interval.Duration, cfg.GC.MaxTTL.Duration, executorScope.NewSubScope("gc"))
193193
if err := mgr.Add(gc); err != nil {
194194
return fmt.Errorf("executor: failed to add garbage collector: %w", err)
195195
}

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,7 @@ require (
9696
require (
9797
github.com/Masterminds/semver/v3 v3.5.0
9898
github.com/alicebob/miniredis/v2 v2.38.0
99+
github.com/prometheus/client_model v0.6.2
99100
)
100101

101102
require (
@@ -189,7 +190,6 @@ require (
189190
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
190191
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
191192
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
192-
github.com/prometheus/client_model v0.6.2 // indirect
193193
github.com/prometheus/procfs v0.20.1 // indirect
194194
github.com/spf13/afero v1.15.0 // indirect
195195
github.com/spf13/cast v1.7.1 // indirect

0 commit comments

Comments
 (0)