adds persistent signals

adds retries

Author: maksymlumen

README.md

@@ -76,13 +76,15 @@ make manifests
 ### Create new api
 ```shell
 kubebuilder create api --group operator --version v1alpha1 --kind TinySignal
-
 ```
 
 **NOTE:** Run `make --help` for more information on all potential `make` targets
 
 More information can be found via the [Kubebuilder Documentation](https://book.kubebuilder.io/introduction.html)
 
+### Module development
+tbd
+
 ## License
 
 Copyright 2023.

api/v1alpha1/tinynode_types.go

@@ -26,6 +26,8 @@ const (
 	//ProjectIDLabel project ID k8s label
 	ProjectIDLabel = "tinysystems.io/project-id"
 
+	NodeNameLabel = "tinysystems.io/node-name"
+
 	//ModuleNameMajorLabel module major version label
 	ModuleNameMajorLabel = "tinysystems.io/module-version-major"
 	//ModuleVersionLabel module exact version label
@@ -40,6 +42,7 @@ const (
 	ComponentPosYAnnotation    = "tinysystems.io/component-pos-y"
 	ComponentPosSpinAnnotation = "tinysystems.io/component-pos-spin"
 
+	SignalNonceAnnotation       = "tinysystems.io/signal-nonce"
 	NodeLabelAnnotation         = "tinysystems.io/node-label"
 	NodeCommentAnnotation       = "tinysystems.io/node-comment"
 	SuggestedHttpPortAnnotation = "tinysystems.io/suggested-http-port"

cli/run.go

@@ -3,6 +3,7 @@ package cli
 import (
 	"context"
 	"fmt"
+	"github.com/cenkalti/backoff/v4"
 	"github.com/go-logr/zerologr"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
@@ -182,8 +183,12 @@ var runCmd = &cobra.Command{
 				return scheduler.Handle(ctx, msg)
 			}
 
-			// gRPC call
-			return pool.Handler(ctx, msg)
+			// gRPC call with retries
+
+			return backoff.Retry(func() error {
+				return pool.Handler(ctx, msg)
+			}, backoff.WithContext(backoff.NewExponentialBackOff(), ctx))
+
 		}).
 			SetLogger(l).
 			SetMeter(meter).

internal/client/pool.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"fmt"
 	"github.com/go-logr/logr"
 	cmap "github.com/orcaman/concurrent-map/v2"
 	"github.com/tiny-systems/module/internal/scheduler/runner"
@@ -58,8 +59,7 @@ func (p *AddressPool) Handler(ctx context.Context, msg *runner.Msg) error {
 
 	addr, ok := p.addressTable.Get(moduleName)
 	if !ok {
-		p.log.Error(err, "module address is unknown", "name", moduleName)
-		return err
+		return fmt.Errorf("%s module address is unknown", moduleName)
 	}
 	client, err := p.getClient(ctx, addr)
 	if err != nil {
@@ -73,6 +73,7 @@ func (p *AddressPool) Handler(ctx context.Context, msg *runner.Msg) error {
 		EdgeID:  msg.EdgeID,
 		To:      msg.To,
 	})
+
 	return err
 }
 

internal/controller/tinynode_controller.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	operatorv1alpha1 "github.com/tiny-systems/module/api/v1alpha1"
 	"github.com/tiny-systems/module/internal/scheduler"
 	"github.com/tiny-systems/module/module"
@@ -64,7 +65,6 @@ type TinyNodeReconciler struct {
 func (r *TinyNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	l := log.FromContext(ctx)
 
-	//l.Info("reconcile", "tinynode", req.Name)
 	m, _, err := module.ParseFullName(req.Name)
 	if err != nil {
 		l.Error(err, "node has invalid name", "name", req.Name)
@@ -76,22 +76,49 @@ func (r *TinyNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return reconcile.Result{}, nil
 	}
 
+	l.Info("reconcile", "tinynode", req.Name)
+
 	node := &operatorv1alpha1.TinyNode{}
 
 	if err = r.Get(context.Background(), req.NamespacedName, node); err != nil {
 		l.Error(err, "get tinynode error")
 		if errors.IsNotFound(err) {
 			// Object not found, return.  Created objects are automatically garbage collected.
 			// For additional cleanup logic use finalizers.
+			// delete signals
+
+			_ = r.DeleteAllOf(context.Background(), &operatorv1alpha1.TinySignal{}, client.InNamespace(req.Namespace), client.MatchingLabels{
+				operatorv1alpha1.NodeNameLabel: req.Name,
+			})
+
 			if err = r.Scheduler.Destroy(req.Name); err != nil {
-				l.Error(err, "destroy error")
 				return reconcile.Result{}, err
 			}
 			return reconcile.Result{}, nil
 		}
 		// Error reading the object - requeue the request.
 		return reconcile.Result{}, err
 	}
+	originNode := node.DeepCopy()
+
+	// select all signals for this node
+
+	signalList := &operatorv1alpha1.TinySignalList{}
+
+	selector, err := v1.LabelSelectorAsSelector(&v1.LabelSelector{
+		MatchLabels: map[string]string{
+			operatorv1alpha1.NodeNameLabel: node.Name,
+		},
+	})
+	if err != nil {
+		return reconcile.Result{}, fmt.Errorf("build signal selector error: %s", err)
+	}
+
+	if err = r.List(ctx, signalList, client.MatchingLabelsSelector{
+		Selector: selector,
+	}, client.InNamespace(node.Namespace)); err != nil {
+		return reconcile.Result{}, fmt.Errorf("signal list error: %v", err)
+	}
 
 	status := &node.Status
 
@@ -109,7 +136,7 @@ func (r *TinyNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 	status.LastUpdateTime = &t
 	// upsert in scheduler
 	// todo add app level context
-	err = r.Scheduler.Update(context.Background(), node)
+	err = r.Scheduler.Update(context.Background(), node, signalList)
 	if err != nil {
 		l.Error(err, "scheduler upsert error")
 		status.Error = true
@@ -118,7 +145,8 @@ func (r *TinyNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 
 	node.Status = *status
 
-	err = r.Status().Update(context.Background(), node)
+	err = r.Status().Patch(context.Background(), node, client.MergeFrom(originNode))
+	//err = r.Status().Update(context.Background(), node)
 	if err != nil {
 		l.Error(err, "status update error")
 		return reconcile.Result{}, err
@@ -135,15 +163,43 @@ func (r *TinyNodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&operatorv1alpha1.TinyNode{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
 		Watches(&operatorv1alpha1.TinySignal{}, handler.TypedFuncs[client.Object, reconcile.Request]{
-			DeleteFunc: func(ctx context.Context, e event.TypedDeleteEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+
+			CreateFunc: func(ctx context.Context, e event.TypedCreateEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
 				signal, ok := e.Object.(*operatorv1alpha1.TinySignal)
 				if !ok {
 					return
 				}
-				if signal.Spec.Port != module.ReconcilePort {
-					// do not reconcile if signal was used as a way to send data
+				if signal.Spec.Port == module.ReconcilePort {
+					// reconcile by signal only applies after deletion of the signal
+					return
+				}
+				q.Add(reconcile.Request{
+					NamespacedName: types.NamespacedName{
+						Name:      signal.Spec.Node,
+						Namespace: signal.Namespace,
+					},
+				})
+			},
+			UpdateFunc: func(ctx context.Context, e event.TypedUpdateEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+				signal, ok := e.ObjectNew.(*operatorv1alpha1.TinySignal)
+				if !ok {
+					return
+				}
+				// we do not update reconcile signals, only data ports
+				q.Add(reconcile.Request{
+					NamespacedName: types.NamespacedName{
+						Name:      signal.Spec.Node,
+						Namespace: signal.Namespace,
+					},
+				})
+			},
+			// when tinySignal deleted signal with reconcile port - reconcile
+			DeleteFunc: func(ctx context.Context, e event.TypedDeleteEvent[client.Object], q workqueue.TypedRateLimitingInterface[reconcile.Request]) {
+				signal, ok := e.Object.(*operatorv1alpha1.TinySignal)
+				if !ok {
 					return
 				}
+				// we reconcile if any relates signal deleted
 				q.Add(reconcile.Request{
 					NamespacedName: types.NamespacedName{
 						Name:      signal.Spec.Node,

internal/controller/tinysignal_controller.go

@@ -19,11 +19,10 @@ package controller
 import (
 	"context"
 	"github.com/tiny-systems/module/internal/scheduler"
-	"github.com/tiny-systems/module/internal/scheduler/runner"
 	"github.com/tiny-systems/module/module"
-	"github.com/tiny-systems/module/pkg/utils"
-	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"strings"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -56,7 +55,6 @@ type TinySignalReconciler struct {
 // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.15.0/pkg/reconcile
 func (r *TinySignalReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	l := log.FromContext(ctx)
-	l.Info("reconcile", "tinysignal", req.Name)
 	// tiny signal names after name of node it's signaling to
 
 	// to avoid making many queries to Kubernetes API we check name itself against current module
@@ -70,35 +68,21 @@ func (r *TinySignalReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return reconcile.Result{}, nil
 	}
 
-	signal := &operatorv1alpha1.TinySignal{}
-	err = r.Get(context.Background(), req.NamespacedName, signal)
-
-	if err != nil {
-		if errors.IsNotFound(err) {
-			// delete signal if related node not found
-			return reconcile.Result{}, nil
-		}
-		return reconcile.Result{}, err
-	}
-
-	if signal.Spec.Port == module.ReconcilePort {
-		// we do not use signals to reconcile directly
-		_ = r.Delete(ctx, signal)
-		return ctrl.Result{}, nil
+	// @todo const
+	if !strings.HasSuffix(req.Name, "-reconcile") {
+		return reconcile.Result{}, nil
 	}
 
-	// todo add app level context
-	if err = r.Scheduler.HandleInternal(context.Background(), &runner.Msg{
-		EdgeID: utils.GetPortFullName(signal.Spec.Node, signal.Spec.Port),
-		To:     utils.GetPortFullName(signal.Spec.Node, signal.Spec.Port),
-		From:   "signal",
-		Data:   signal.Spec.Data,
-	}); err != nil {
-		l.Error(err, "invoke error")
+	signal := &operatorv1alpha1.TinySignal{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      req.Name,
+			Namespace: req.Namespace,
+		},
 	}
 
 	_ = r.Delete(ctx, signal)
 	return ctrl.Result{}, nil
+
 }
 
 // SetupWithManager sets up the controller with the Manager.

internal/resource/manager.go

@@ -376,16 +376,43 @@ func (m Manager) DisclosePort(ctx context.Context, port int) error {
 }
 
 func (m Manager) CreateClusterNodeSignal(ctx context.Context, node v1alpha1.TinyNode, port string, data []byte) error {
-	signal := &v1alpha1.TinySignal{
-		Spec: v1alpha1.TinySignalSpec{
-			Node: node.Name,
-			Port: port,
-			Data: data,
-		},
+
+	signal := &v1alpha1.TinySignal{}
+	name := fmt.Sprintf("%s-%s", node.Name, strings.ReplaceAll(port, "_", ""))
+
+	err := m.client.Get(ctx, client.ObjectKey{Namespace: node.Namespace, Name: name}, signal)
+	if err != nil && !errors.IsNotFound(err) {
+		return err
+	}
+
+	newSignal := signal.DeepCopy()
+
+	newSignal.Namespace = node.Namespace
+	newSignal.Name = name
+
+	newSignal.Labels = map[string]string{
+		v1alpha1.NodeNameLabel: node.Name,
+	}
+
+	newSignal.Spec = v1alpha1.TinySignalSpec{
+		Node: node.Name,
+		Port: port,
+		Data: data,
 	}
-	signal.Namespace = node.Namespace
-	signal.GenerateName = fmt.Sprintf("%s-%s-", node.Name, strings.ReplaceAll(port, "_", ""))
-	return m.client.Create(ctx, signal)
+
+	if errors.IsNotFound(err) {
+		if err := m.client.Create(ctx, newSignal); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	err = m.client.Patch(ctx, newSignal, client.MergeFrom(signal))
+	if err != nil && !errors.IsNotFound(err) {
+		return err
+	}
+
+	return nil
 }
 
 func (m Manager) Start(ctx context.Context) error {

internal/scheduler/runner/msg.go

@@ -9,4 +9,6 @@ type Msg struct {
 	// recipient of this message in a format node:port
 	To   string `json:"to"`
 	Data []byte `json:"data"`
+
+	Nonce string `json:"nonce"`
 }