[apiserver/edgefunction] Cleanup on GC

Author: dilyevsky

pkg/apiserver/extensions/edgefunctionrevision.go

@@ -5,6 +5,7 @@ import (
 	"sort"
 	"time"
 
+	tclient "go.temporal.io/sdk/client"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -14,6 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	extensionsv1alpha2 "github.com/apoxy-dev/apoxy-cli/api/extensions/v1alpha2"
+	"github.com/apoxy-dev/apoxy-cli/pkg/apiserver/ingest"
 )
 
 const (
@@ -25,21 +27,22 @@ var _ reconcile.Reconciler = &EdgeFunctionRevisionGCReconciler{}
 // EdgeFunctionRevisionGCReconciler garbage collects EdgeFunctionRevisions.
 type EdgeFunctionRevisionGCReconciler struct {
 	client.Client
-	Scheme *runtime.Scheme
-
+	scheme     *runtime.Scheme
+	tc         tclient.Client
 	gcInterval time.Duration
 }
 
 // NewEdgeFunctionRevisionGCReconciler returns a new EdgeFunctionRevisionGCReconciler.
 func NewEdgeFunctionRevisionGCReconciler(
 	c client.Client,
 	s *runtime.Scheme,
+	tc tclient.Client,
 	gcInterval time.Duration,
 ) *EdgeFunctionRevisionGCReconciler {
 	return &EdgeFunctionRevisionGCReconciler{
-		Client: c,
-		Scheme: s,
-
+		Client:     c,
+		scheme:     s,
+		tc:         tc,
 		gcInterval: gcInterval,
 	}
 }
@@ -81,6 +84,18 @@ func (r *EdgeFunctionRevisionGCReconciler) Reconcile(ctx context.Context, req re
 					log.Error(err, "Failed to delete old revision", "revision", revision.Name)
 					continue
 				}
+				// Execute cleanup workflow.
+				wOpts := tclient.StartWorkflowOptions{
+					ID:                       revision.Name + "-cleanup",
+					TaskQueue:                ingest.EdgeFunctionCleanupQueue,
+					WorkflowExecutionTimeout: 10 * time.Minute,
+				}
+				in := &ingest.EdgeFunctionIngestParams{
+					Obj: revision.DeepCopy(),
+				}
+				if _, err := r.tc.ExecuteWorkflow(ctx, wOpts, ingest.EdgeFunctionCleanupWorkflow, in); err != nil {
+					log.Error(err, "Failed to start cleanup workflow", "revision", revision.Name)
+				}
 				log.Info("Deleted old revision", "revision", revision.Name)
 			}
 		}

pkg/apiserver/ingest/edgefunction.go

@@ -41,7 +41,8 @@ import (
 )
 
 const (
-	EdgeFunctionIngestQueue = "EDGE_FUNC_INGEST_TASK_QUEUE"
+	EdgeFunctionIngestQueue  = "EDGE_FUNC_INGEST_TASK_QUEUE"
+	EdgeFunctionCleanupQueue = "EDGE_FUNC_CLEANUP_TASK_QUEUE"
 )
 
 const (
@@ -61,6 +62,7 @@ type IngestResult struct {
 // RegisterWorkflows registers Edge Function workflows with the Temporal worker.
 func RegisterWorkflows(w tworker.Worker) {
 	w.RegisterWorkflow(EdgeFunctionIngestWorkflow)
+	w.RegisterWorkflow(EdgeFunctionCleanupWorkflow)
 }
 
 // EdgeFunctionIngestWorkflow is a Temporal workflow that ingests Edge Functions.
@@ -191,6 +193,35 @@ Finalize:
 	return nil
 }
 
+func EdgeFunctionCleanupWorkflow(ctx workflow.Context, in *EdgeFunctionIngestParams) error {
+	opts := workflow.ActivityOptions{
+		StartToCloseTimeout: 10 * time.Minute,
+		RetryPolicy: &temporal.RetryPolicy{
+			InitialInterval:    time.Second,
+			BackoffCoefficient: 2.0,
+			MaximumInterval:    10 * time.Second,
+			MaximumAttempts:    3,
+		},
+	}
+	ctx = workflow.WithActivityOptions(ctx, opts)
+	log := tlog.With(workflow.GetLogger(ctx), "Name", in.Obj.Name, "ResourceVersion", in.Obj.ResourceVersion)
+
+	var w *worker
+
+	if err := workflow.ExecuteActivity(ctx, w.CleanupStagedData, in).Get(ctx, nil); err != nil {
+		log.Error("Failed to cleanup staged data", "Error", err)
+		return err
+	}
+
+	if err := workflow.ExecuteActivity(ctx, w.CleanupStoredData, in).Get(ctx, nil); err != nil {
+		log.Error("Failed to cleanup stored data", "Error", err)
+		return err
+	}
+
+	log.Info("Edge Function cleanup completed successfully")
+	return nil
+}
+
 // Worker implements Temporal Activities for Edge Functions Ingest queue.
 type worker struct {
 	k8s     kubernetes.Interface
@@ -223,6 +254,7 @@ func (w *worker) RegisterActivities(tw tworker.Worker) {
 	tw.RegisterActivity(w.StoreGoPluginActivity)
 	tw.RegisterActivity(w.FinalizeActivity)
 	tw.RegisterActivity(w.CleanupStagedData)
+	tw.RegisterActivity(w.CleanupStoredData)
 }
 
 func (w *worker) AddIngestConditionActivity(
@@ -872,6 +904,25 @@ func (w *worker) CleanupStagedData(
 	return nil
 }
 
+func (w *worker) CleanupStoredData(
+	ctx context.Context,
+	in *EdgeFunctionIngestParams,
+) error {
+	log := tlog.With(activity.GetLogger(ctx), "Name", in.Obj.Name, "ResourceVersion", in.Obj.ResourceVersion)
+	log.Info("Cleaning up stored data")
+
+	wid := activity.GetInfo(ctx).WorkflowExecution.ID
+	storeDir := w.storeDir(wid)
+
+	if err := os.RemoveAll(storeDir); err != nil {
+		log.Error("Failed to remove store directory", "Error", err)
+		return err
+	}
+
+	log.Info("Store directory cleaned up successfully")
+	return nil
+}
+
 func hasCondition(conditions []metav1.Condition, condType string, status metav1.ConditionStatus) bool {
 	for _, c := range conditions {
 		if c.Type == condType && c.Status == status {

pkg/apiserver/manager.go

@@ -426,6 +426,7 @@ func (m *Manager) Start(
 	if err := extensionscontroller.NewEdgeFunctionRevisionGCReconciler(
 		m.manager.GetClient(),
 		m.manager.GetScheme(),
+		tc,
 		dOpts.gcInterval,
 	).SetupWithManager(ctx, m.manager); err != nil {
 		return fmt.Errorf("failed to set up EdgeFunctionRevision GC controller: %v", err)