[edgefunc] Many small fixes. Finally starts.

Author: dilyevsky

cmd/backplane/main.go

@@ -38,6 +38,7 @@ import (
 "github.com/apoxy-dev/apoxy-cli/pkg/backplane/wasm/ext_proc"
 "github.com/apoxy-dev/apoxy-cli/pkg/backplane/wasm/manifest"
 "github.com/apoxy-dev/apoxy-cli/pkg/cmd/utils"
+"github.com/apoxy-dev/apoxy-cli/pkg/edgefunc/runc"
 "github.com/apoxy-dev/apoxy-cli/pkg/log"
 
 ctrlv1alpha1 "github.com/apoxy-dev/apoxy-cli/api/controllers/v1alpha1"
@@ -252,7 +253,7 @@ func main() {
 log.Fatalf("Failed to add readyz check: %v", err)
 }
 
-log.Infof("Setting up controllers")
+log.Infof("Setting up controllers...")
 proxyOpts := []bpctrl.Option{
 bpctrl.WithGoPluginDir(*goPluginDir),
 }
@@ -277,6 +278,8 @@ func main() {
 apiServerHost = *apiServerAddr
 }
 
+log.Infof("Starting Backplane controller")
+
 pctrl := bpctrl.NewProxyReconciler(
 mgr.GetClient(),
 *proxyName,
@@ -285,19 +288,24 @@ func main() {
 proxyOpts...,
 )
 if err := pctrl.SetupWithManager(ctx, mgr); err != nil {
-log.Errorf("failed to set up Backplane controller: %v", err)
-return
+log.Fatalf("failed to set up Backplane controller: %v", err)
 }
 
+log.Infof("Starting EdgeFunction controller")
+
+edgeRuntime, err := runc.NewRuntime(ctx)
+if err != nil {
+log.Fatalf("failed to set up EdgeFunction controller: %v", err)
+}
 if err := bpctrl.NewEdgeFunctionRevisionReconciler(
 mgr.GetClient(),
 net.JoinHostPort(apiServerHost, strconv.Itoa(*wasmStorePort)),
 ms,
 *goPluginDir,
 *esZipDir,
+edgeRuntime,
 ).SetupWithManager(ctx, mgr, *proxyName); err != nil {
-log.Errorf("failed to set up EdgeFunction controller: %v", err)
-return
+log.Fatalf("failed to set up EdgeFunction controller: %v", err)
 }
 
 // Setup SIGTERM handler.

go.mod

@@ -121,6 +121,7 @@ require (
 github.com/census-instrumentation/opencensus-proto v0.4.1 // indirect
 github.com/cespare/xxhash/v2 v2.3.0 // indirect
 github.com/checkpoint-restore/go-criu/v6 v6.3.0 // indirect
+github.com/cilium/ebpf v0.16.0 // indirect
 github.com/containerd/console v1.0.4 // indirect
 github.com/containerd/stargz-snapshotter/estargz v0.14.3 // indirect
 github.com/coreos/go-oidc/v3 v3.1.0 // indirect

go.sum

@@ -774,6 +774,8 @@ github.com/checkpoint-restore/go-criu/v6 v6.3.0/go.mod h1:rrRTN/uSwY2X+BPRl/gkul
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
+github.com/cilium/ebpf v0.16.0 h1:+BiEnHL6Z7lXnlGUsXQPPAE7+kenAd4ES8MQ5min0Ok=
+github.com/cilium/ebpf v0.16.0/go.mod h1:L7u2Blt2jMM/vLAVgjxluxtBKlz3/GWjB0dMOEngfwE=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58 h1:F1EaeKL/ta07PY/k9Os/UFtwERei2/XzGemhpGnBKNg=
 github.com/cloudflare/golz4 v0.0.0-20150217214814-ef862a3cdc58/go.mod h1:EOBUe0h4xcZ5GoxqC5SDxFQ8gwyZPKQoEzownBlhI80=

pkg/apiserver/ingest/edgefunction.go

@@ -199,7 +199,9 @@ func NewWorker(kc *rest.Config, c versioned.Interface, baseDir string) *worker {
 a3y: c,
 baseDir: baseDir,
 }
-w.k8s = kubernetes.NewForConfigOrDie(kc)
+if kc != nil {
+w.k8s = kubernetes.NewForConfigOrDie(kc)
+}
 return w
 }
 
@@ -841,6 +843,7 @@ func (w *worker) ListenAndServeEdgeFuncs(host string, port int) error {
 mux := http.NewServeMux()
 mux.Handle("/wasm/", w)
 mux.Handle("/go/", w)
+mux.Handle("/js/", w)
 addr := fmt.Sprintf("%s:%d", host, port)
 log.Infof("Listening on %s", addr)
 return http.ListenAndServe(addr, mux)

pkg/backplane/controllers/edgefunction.go

@@ -2,19 +2,21 @@ package controllers
 
 import (
 "context"
+"errors"
 "fmt"
 "io"
 "net/http"
 "os"
 "path/filepath"
-"sync"
 "time"
 
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/apimachinery/pkg/types"
+"k8s.io/utils/ptr"
 ctrl "sigs.k8s.io/controller-runtime"
 "sigs.k8s.io/controller-runtime/pkg/builder"
 "sigs.k8s.io/controller-runtime/pkg/client"
+"sigs.k8s.io/controller-runtime/pkg/controller"
 clog "sigs.k8s.io/controller-runtime/pkg/log"
 "sigs.k8s.io/controller-runtime/pkg/predicate"
 "sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -41,9 +43,6 @@ type EdgeFunctionRevisionReconciler struct {
 goStoreDir string
 jsStoreDir string
 edgeRuntime edgefunc.Runtime
-
-mu sync.Mutex
-edgeFuncs map[string]*v1alpha1.EdgeFunctionRevision
 }
 
 // NewEdgeFunctionRevisionReconciler returns a new reconcile.Reconciler.
@@ -53,15 +52,15 @@ func NewEdgeFunctionRevisionReconciler(
 wasmStore manifest.Store,
 goStoreDir string,
 jsStoreDir string,
+edgeRuntime edgefunc.Runtime,
 ) *EdgeFunctionRevisionReconciler {
 return &EdgeFunctionRevisionReconciler{
 Client: c,
 apiserverHost: apiserverHost,
 wasmStore: wasmStore,
 goStoreDir: goStoreDir,
 jsStoreDir: jsStoreDir,
-
-edgeFuncs: make(map[string]*v1alpha1.EdgeFunctionRevision),
+edgeRuntime: edgeRuntime,
 }
 }
 
@@ -74,17 +73,16 @@ func (r *EdgeFunctionRevisionReconciler) downloadFuncData(
 
 resp, err := http.Get(fmt.Sprintf("http://%s/%s/%s", r.apiserverHost, fnType, ref))
 if err != nil {
-return nil, fmt.Errorf("failed to download Wasm module: %w", err)
+return nil, fmt.Errorf("failed to download edge function: %w", err)
 }
-
 if resp.StatusCode != http.StatusOK {
-return nil, fmt.Errorf("failed to download Wasm module: %s", resp.Status)
+return nil, fmt.Errorf("failed to download edge function: %s", resp.Status)
 }
 defer resp.Body.Close()
 
 data, err := io.ReadAll(resp.Body)
 if err != nil {
-return nil, fmt.Errorf("failed to read Wasm module: %w", err)
+return nil, fmt.Errorf("failed to read edge function data: %w", err)
 }
 
 log.Info("Downloaded EdgeFunction module", "size", len(data), "type", fnType, "ref", ref)
@@ -101,6 +99,55 @@ func hasReadyCondition(conditions []metav1.Condition) bool {
 return false
 }
 
+func (r *EdgeFunctionRevisionReconciler) reconileEdgeRuntime(ctx context.Context, ref string) error {
+log := clog.FromContext(ctx)
+
+esZipPath := filepath.Join(r.jsStoreDir, ref, "bin.eszip")
+if _, err := os.Stat(esZipPath); err != nil && !os.IsNotExist(err) {
+return fmt.Errorf("Js bundle already exists for ref")
+} else if os.IsNotExist(err) {
+jsBundle, err := r.downloadFuncData(clog.IntoContext(ctx, log), "js", ref)
+if err != nil {
+return fmt.Errorf("failed to download Js data: %w", err)
+}
+
+if err := os.MkdirAll(filepath.Join(r.jsStoreDir, ref), 0755); err != nil {
+return fmt.Errorf("failed to create Js directory: %w", err)
+}
+
+if err := os.WriteFile(filepath.Join(r.jsStoreDir, ref, "data"), jsBundle, 0644); err != nil {
+return fmt.Errorf("failed to write Js data: %w", err)
+}
+// Use symlink to prevent Envoy from loading the plugin while it's being written to.
+if err := os.Symlink("data", esZipPath); err != nil {
+return fmt.Errorf("failed to create Js symlink: %w", err)
+}
+}
+
+s, err := r.edgeRuntime.Status(ctx, ref)
+if err != nil && !errors.Is(err, edgefunc.ErrNotFound) {
+return fmt.Errorf("failed to get Edge Runtime status: %w", err)
+}
+
+log.Info("Edge Runtime status", "state", s.State)
+
+switch s.State {
+case edgefunc.StateRunning, edgefunc.StateCreated:
+log.Info("Edge Runtime is already running or created", "state", s.State)
+return nil
+case edgefunc.StateStopped:
+log.Info("Edge Runtime is stopped, starting it")
+}
+
+log.Info("Starting Edge Runtime")
+
+if err := r.edgeRuntime.Start(ctx, ref, esZipPath); err != nil {
+return fmt.Errorf("failed to start Edge Runtime: %w", err)
+}
+
+return nil
+}
+
 // Reconcile implements reconcile.Reconciler.
 func (r *EdgeFunctionRevisionReconciler) Reconcile(ctx context.Context, request reconcile.Request) (ctrl.Result, error) {
 log := clog.FromContext(ctx)
@@ -208,27 +255,9 @@ func (r *EdgeFunctionRevisionReconciler) Reconcile(ctx context.Context, request
 } else if rev.Spec.Code.JsSource != nil {
 log.Info("Js source detected")
 
-jsBundle, err := r.downloadFuncData(clog.IntoContext(ctx, log), "js", ref)
-if err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to download Js data: %w", err)
-}
-
-if err := os.MkdirAll(filepath.Join(r.jsStoreDir, ref), 0755); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to create Js directory: %w", err)
-}
-
-if err := os.WriteFile(filepath.Join(r.jsStoreDir, ref, "data"), jsBundle, 0644); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to write Js data: %w", err)
-}
-// Use symlink to prevent Envoy from loading the plugin while it's being written to.
-esZipPath := filepath.Join(r.jsStoreDir, ref, "bin.eszip")
-if err := os.Symlink("data", esZipPath); err != nil {
-return ctrl.Result{}, fmt.Errorf("failed to create Js symlink: %w", err)
+if err := r.reconileEdgeRuntime(ctx, ref); err != nil {
+return ctrl.Result{}, fmt.Errorf("failed to reconile Edge Runtime: %w", err)
 }
-
-r.mu.Lock()
-r.edgeFuncs[ref] = rev
-r.mu.Unlock()
 } else {
 log.Info("No source detected")
 return ctrl.Result{}, nil
@@ -237,33 +266,6 @@ func (r *EdgeFunctionRevisionReconciler) Reconcile(ctx context.Context, request
 return ctrl.Result{}, nil
 }
 
-func (r *EdgeFunctionRevisionReconciler) reconileEdgeRuntime(ctx context.Context, ref string) error {
-log := clog.FromContext(ctx)
-
-s, err := r.edgeRuntime.Status(ctx, ref)
-if err != nil {
-return fmt.Errorf("failed to get Edge Runtime status: %w", err)
-}
-
-switch s.State {
-case edgefunc.StateRunning:
-log.Info("Edge Runtime is already running")
-return nil
-case edgefunc.StateCreated:
-log.Info("Edge Runtime is already created")
-return nil
-case edgefunc.StateStopped:
-log.Info("Edge Runtime is stopped, starting it")
-default:
-return fmt.Errorf("Edge Runtime is in an unknown state: %s", s.State)
-}
-
-if err := r.edgeRuntime.Start(ctx, ref, filepath.Join(r.jsStoreDir, ref, "bin.eszip")); err != nil {
-return fmt.Errorf("failed to start Edge Runtime: %w", err)
-}
-return nil
-}
-
 func targetRefPredicate(proxyName string) predicate.Funcs {
 return predicate.NewPredicateFuncs(func(obj client.Object) bool {
 if obj == nil {
@@ -309,5 +311,9 @@ func (r *EdgeFunctionRevisionReconciler) SetupWithManager(
 targetRefPredicate(proxyName),
 ),
 ).
+WithOptions(controller.Options{
+MaxConcurrentReconciles: 1,
+RecoverPanic: ptr.To(true),
+}).
 Complete(r)
 }