[edgefunc] Implement Js source deletion

Author: dilyevsky

cmd/backplane/main.go

@@ -302,6 +302,7 @@ func main() {
 }
 if err := bpctrl.NewEdgeFunctionRevisionReconciler(
 mgr.GetClient(),
+*replicaName,
 net.JoinHostPort(apiServerHost, strconv.Itoa(*wasmStorePort)),
 ms,
 *goPluginDir,

pkg/backplane/controllers/edgefunction.go

@@ -17,6 +17,7 @@ import (
 "sigs.k8s.io/controller-runtime/pkg/builder"
 "sigs.k8s.io/controller-runtime/pkg/client"
 "sigs.k8s.io/controller-runtime/pkg/controller"
+"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 clog "sigs.k8s.io/controller-runtime/pkg/log"
 "sigs.k8s.io/controller-runtime/pkg/predicate"
 "sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -29,6 +30,8 @@ import (
 
 const (
 edgeFuncRevsOwnerIndex = "edgeFuncRevsOwner"
+
+backplaneFinalizerPrefix = "controllers.apoxy.dev/backplane-"
 )
 
 var _ reconcile.Reconciler = &EdgeFunctionRevisionReconciler{}
@@ -38,6 +41,7 @@ var _ reconcile.Reconciler = &EdgeFunctionRevisionReconciler{}
 type EdgeFunctionRevisionReconciler struct {
 client.Client
 
+replicaName string
 apiserverHost string
 wasmStore manifest.Store
 goStoreDir string
@@ -48,6 +52,7 @@ type EdgeFunctionRevisionReconciler struct {
 // NewEdgeFunctionRevisionReconciler returns a new reconcile.Reconciler.
 func NewEdgeFunctionRevisionReconciler(
 c client.Client,
+replicaName string,
 apiserverHost string,
 wasmStore manifest.Store,
 goStoreDir string,
@@ -56,6 +61,7 @@ func NewEdgeFunctionRevisionReconciler(
 ) *EdgeFunctionRevisionReconciler {
 return &EdgeFunctionRevisionReconciler{
 Client: c,
+replicaName: replicaName,
 apiserverHost: apiserverHost,
 wasmStore: wasmStore,
 goStoreDir: goStoreDir,
@@ -152,6 +158,48 @@ func (r *EdgeFunctionRevisionReconciler) reconileEdgeRuntime(
 return nil
 }
 
+func (r *EdgeFunctionRevisionReconciler) maybeCleanupEdgeRuntime(ctx context.Context, ref string) (bool, error) {
+log := clog.FromContext(ctx, "Ref", ref)
+
+log.Info("Cleaning up Edge Runtime")
+
+if err := r.edgeRuntime.StopExec(ctx, ref); err != nil {
+log.Error(err, "Failed to stop Edge Runtime")
+return false, err
+}
+s, err := r.edgeRuntime.ExecStatus(ctx, ref)
+if err != nil {
+log.Error(err, "Failed to get Edge Runtime status")
+return false, err
+}
+
+switch s.State {
+case edgefunc.StateRunning, edgefunc.StateCreated:
+log.Info("Edge Runtime is still waiting for termination")
+return false, nil
+case edgefunc.StateStopped:
+log.Info("Edge Runtime is stopped, cleaning up")
+}
+
+if err := r.edgeRuntime.DeleteExec(ctx, ref); err != nil && !errors.Is(err, edgefunc.ErrNotFound) {
+log.Error(err, "Failed to delete Edge Runtime")
+} else if errors.Is(err, edgefunc.ErrNotFound) {
+log.Info("Edge Runtime not found, nothing to delete")
+return true, nil
+}
+
+log.Info("Edge Runtime deleted, cleaning up data")
+
+jsBundlePath := filepath.Join(r.jsStoreDir, ref, "data")
+if err := os.RemoveAll(jsBundlePath); err != nil {
+log.Error(err, "Failed to delete Js bundle")
+}
+
+log.Info("Edge Runtime cleaned up")
+
+return true, nil
+}
+
 // Reconcile implements reconcile.Reconciler.
 func (r *EdgeFunctionRevisionReconciler) Reconcile(ctx context.Context, request reconcile.Request) (ctrl.Result, error) {
 log := clog.FromContext(ctx)
@@ -179,32 +227,57 @@ func (r *EdgeFunctionRevisionReconciler) Reconcile(ctx context.Context, request
 }
 
 if !rev.DeletionTimestamp.IsZero() {
-if ef.Status.LiveRevision == rev.Name { // Do not delete the live revision.
-log.Info("EdgeFunctionRevision is the live revision, Skipping deletion.")
-return ctrl.Result{RequeueAfter: time.Second * 10}, nil
-}
-
-log.Info("Deleting EdgeFunctionRevision")
-
-if rev.Spec.Code.WasmSource != nil {
-log.Info("Deleting Wasm data", "ref", rev.Status.Ref)
-if err := r.wasmStore.Delete(ctx, rev.Status.Ref); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to delete Wasm data: %w", err)
+finalizer := backplaneFinalizerPrefix + r.replicaName
+if controllerutil.ContainsFinalizer(rev, finalizer) {
+if ef.Status.LiveRevision == rev.Name { // Do not delete the live revision.
+log.Info("EdgeFunctionRevision is the live revision, Skipping deletion.")
+return ctrl.Result{RequeueAfter: time.Second * 10}, nil
 }
-} else if rev.Spec.Code.GoPluginSource != nil {
-log.Info("Deleting Go plugin data", "ref", rev.Status.Ref)
-if err := os.RemoveAll(filepath.Join(r.goStoreDir, rev.Status.Ref)); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to delete Go plugin data: %w", err)
+
+log.Info("Deleting EdgeFunctionRevision")
+
+if rev.Spec.Code.WasmSource != nil {
+log.Info("Deleting Wasm data", "ref", rev.Status.Ref)
+if err := r.wasmStore.Delete(ctx, rev.Status.Ref); err != nil {
+return ctrl.Result{}, fmt.Errorf("failed to delete Wasm data: %w", err)
+}
+} else if rev.Spec.Code.GoPluginSource != nil {
+log.Info("Deleting Go plugin data", "ref", rev.Status.Ref)
+if err := os.RemoveAll(filepath.Join(r.goStoreDir, rev.Status.Ref)); err != nil {
+return ctrl.Result{}, fmt.Errorf("failed to delete Go plugin data: %w", err)
+}
+} else if rev.Spec.Code.JsSource != nil {
+log.Info("Deleting Js data", "ref", rev.Status.Ref)
+if err := os.RemoveAll(filepath.Join(r.jsStoreDir, rev.Status.Ref)); err != nil {
+return ctrl.Result{}, fmt.Errorf("failed to delete Js data: %w", err)
+}
+
+if done, err := r.maybeCleanupEdgeRuntime(ctx, rev.Status.Ref); err != nil {
+log.Error(err, "Failed to cleanup Edge Runtime")
+return ctrl.Result{}, err
+} else if !done {
+log.Info("Edge Runtime is still running, waiting for termination")
+return ctrl.Result{RequeueAfter: time.Second * 10}, nil
+}
 }
-} else if rev.Spec.Code.JsSource != nil {
-log.Info("Deleting Js data", "ref", rev.Status.Ref)
-if err := os.RemoveAll(filepath.Join(r.jsStoreDir, rev.Status.Ref)); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to delete Js data: %w", err)
+
+controllerutil.RemoveFinalizer(rev, finalizer)
+if err := r.Update(ctx, rev); err != nil {
+return ctrl.Result{}, err
 }
 }
 return ctrl.Result{}, nil
 }
 
+finalizer := backplaneFinalizerPrefix + r.replicaName
+if !controllerutil.ContainsFinalizer(rev, finalizer) {
+log.Info("Adding finalizer", "Finalizer", finalizer)
+controllerutil.AddFinalizer(rev, finalizer)
+if err := r.Update(ctx, rev); err != nil {
+return ctrl.Result{}, err
+}
+}
+
 // Stage the EdgeFunctionRevision that was successfully ingested (has a Ready condition)
 // even if it's not Live yet.
 if !hasReadyCondition(rev.Status.Conditions) {

pkg/edgefunc/runc/network/net.go

@@ -442,29 +442,22 @@ func (n *Network) Up(ctx context.Context, cid string) error {
 
 func tearDownVeth(info *SandboxInfo) error {
 log.Infof("Removing veth pair for %s", info.Veth)
+
 veth, err := netlink.LinkByName(info.Veth)
-if err == nil {
-// Deleting one end of the veth pair automatically deletes the other end.
-if err := netlink.LinkDel(veth); err != nil {
-return fmt.Errorf("failed to delete veth pair: %w", err)
-}
+if err != nil && !errors.Is(err, netlink.LinkNotFoundError{}) {
+return fmt.Errorf("failed to get veth: %w", err)
+} else if errors.Is(err, netlink.LinkNotFoundError{}) {
+log.Debugf("veth not found, nothing to delete")
+return nil
 }
 
-log.Infof("Removing routes for %s", info.Veth)
-for _, r := range info.Routes {
-if err := netlink.RouteDel(&netlink.Route{
-LinkIndex: veth.Attrs().Index,
-Scope: netlink.SCOPE_UNIVERSE,
-Dst: &net.IPNet{
-IP: net.ParseIP(r.Dst),
-Mask: net.CIDRMask(32, 32),
-},
-Table: r.Table,
-}); err != nil && !errors.As(err, &netlink.LinkNotFoundError{}) {
-return fmt.Errorf("failed to delete route: %w", err)
-}
+// Deleting one end of the veth pair automatically deletes the other end.
+if err := netlink.LinkDel(veth); err != nil {
+return fmt.Errorf("failed to delete veth pair: %w", err)
 }
 
+// No need to remove routes - they are delete when the device is removed above.
+
 return nil
 }
 
@@ -500,6 +493,7 @@ func (n *Network) Down(ctx context.Context, cid string) error {
 }
 
 // Release prefix from the ipam.
+log.Infof("Releasing prefix %s", info.Cidr)
 prefix, err := n.ipamer.PrefixFrom(ctx, info.Cidr.String())
 if err != nil {
 return fmt.Errorf("failed to get prefix: %v", err)
@@ -511,6 +505,9 @@ func (n *Network) Down(ctx context.Context, cid string) error {
 if err := ns.Close(); err != nil {
 return fmt.Errorf("failed to close netns: %w", err)
 }
+if err := netns.DeleteNamed(cid); err != nil {
+return fmt.Errorf("failed to delete netns: %w", err)
+}
 
 // Delete cid network info from the db
 if err := n.db.Update(func(txn *badger.Txn) error {