External Network: connection controller

.github/workflows/integration.yml

@@ -8,11 +8,11 @@ on:
       - network-general
       - network-external
       - network-internal
+      - frc/connectioncontroller
   repository_dispatch:
     types:
       - test-command
       - build-command
-
 jobs:
   configure:
     name: Preliminary configuration
@@ -91,6 +91,7 @@ jobs:
         - telemetry
         - proxy
         - gateway/tunnel/wireguard
+        - gateway/main
     steps:
       - name: Set up QEMU
         uses: docker/[email protected]

build/gateway/main/Dockerfile

@@ -0,0 +1,20 @@
+FROM golang:1.21 as goBuilder
+WORKDIR /tmp/builder
+
+COPY go.mod ./go.mod
+COPY go.sum ./go.sum
+RUN  go mod download
+
+COPY . ./
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=$(go env GOARCH) go build -ldflags="-s -w" ./cmd/gateway/main
+
+
+FROM alpine:3.18
+
+RUN apk update && \
+    apk add iptables bash tcpdump conntrack-tools curl iputils && \
+    rm -rf /var/cache/apk/*
+
+COPY --from=goBuilder /tmp/builder/main /usr/bin/liqo-gateway
+
+ENTRYPOINT [ "/usr/bin/liqo-gateway" ]

cmd/gateway/main/main.go

@@ -0,0 +1,128 @@
+// Copyright 2019-2023 The Liqo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package wireguard contains the logic to configure the Wireguard interface.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+
+	networkingv1alpha1 "github.com/liqotech/liqo/apis/networking/v1alpha1"
+	"github.com/liqotech/liqo/pkg/gateway/connection"
+	flagsutils "github.com/liqotech/liqo/pkg/utils/flags"
+	"github.com/liqotech/liqo/pkg/utils/mapper"
+	"github.com/liqotech/liqo/pkg/utils/restcfg"
+)
+
+var (
+	addToSchemeFunctions = []func(*runtime.Scheme) error{
+		networkingv1alpha1.AddToScheme,
+	}
+	options = connection.NewOptions()
+)
+
+func main() {
+	var cmd = cobra.Command{
+		Use:  "liqo-gateway",
+		RunE: run,
+	}
+
+	legacyflags := flag.NewFlagSet("legacy", flag.ExitOnError)
+	restcfg.InitFlags(legacyflags)
+	klog.InitFlags(legacyflags)
+	flagsutils.FromFlagToPflag(legacyflags, cmd.Flags())
+
+	connection.InitFlags(cmd.Flags(), options)
+	if err := connection.MarkFlagsRequired(&cmd); err != nil {
+		klog.Error(err)
+		os.Exit(1)
+	}
+
+	if err := cmd.Execute(); err != nil {
+		klog.Error(err)
+		os.Exit(1)
+	}
+}
+
+func run(_ *cobra.Command, _ []string) error {
+	var err error
+	ctx := ctrl.SetupSignalHandler()
+	scheme := runtime.NewScheme()
+
+	// Adds the APIs to the scheme.
+	for _, addToScheme := range addToSchemeFunctions {
+		if err = addToScheme(scheme); err != nil {
+			return fmt.Errorf("unable to add scheme: %w", err)
+		}
+	}
+
+	// Set controller-runtime logger.
+	log.SetLogger(klog.NewKlogr())
+
+	// Get the rest config.
+	cfg := config.GetConfigOrDie()
+
+	// Create the manager.
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+		MapperProvider:         mapper.LiqoMapperProvider(scheme),
+		Scheme:                 scheme,
+		Namespace:              options.Namespace,
+		MetricsBindAddress:     "0", // Metrics are exposed by "connection" container.
+		HealthProbeBindAddress: options.ProbeAddr,
+		LeaderElection:         options.LeaderElection,
+		LeaderElectionID: fmt.Sprintf(
+			"%s.%s.%s.connections.liqo.io",
+			options.Name, options.Namespace, options.Mode,
+		),
+		LeaderElectionNamespace:       options.Namespace,
+		LeaderElectionReleaseOnCancel: true,
+		LeaderElectionResourceLock:    resourcelock.LeasesResourceLock,
+		LeaseDuration:                 &options.LeaderElectionLeaseDuration,
+		RenewDeadline:                 &options.LeaderElectionRenewDeadline,
+		RetryPeriod:                   &options.LeaderElectionRetryPeriod,
+	})
+	if err != nil {
+		return fmt.Errorf("unable to create manager: %w", err)
+	}
+
+	// Setup the controller.
+	connr, err := connection.NewConnectionsReconciler(
+		ctx,
+		mgr.GetClient(),
+		mgr.GetScheme(),
+		mgr.GetEventRecorderFor("connections-controller"),
+		options,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to create connectioons reconciler: %w", err)
+	}
+
+	// Setup the controller.
+	if err = connr.SetupWithManager(mgr); err != nil {
+		return fmt.Errorf("unable to setup connections reconciler: %w", err)
+	}
+
+	// Start the manager.
+	return mgr.Start(ctx)
+}

cmd/liqonet/gateway-operator.go

@@ -34,7 +34,7 @@ import (
 
 	tunneloperator "github.com/liqotech/liqo/internal/liqonet/tunnel-operator"
 	liqoconst "github.com/liqotech/liqo/pkg/consts"
-	"github.com/liqotech/liqo/pkg/liqonet/conncheck"
+	"github.com/liqotech/liqo/pkg/gateway/connection/conncheck"
 	liqonetns "github.com/liqotech/liqo/pkg/liqonet/netns"
 	liqonetutils "github.com/liqotech/liqo/pkg/liqonet/utils"
 	"github.com/liqotech/liqo/pkg/liqonet/utils/links"

deployments/liqo/files/liqo-newgateway-ClusterRole.yaml

@@ -9,6 +9,16 @@ rules:
   - get
   - list
   - update
+- apiGroups:
+  - networking.liqo.io
+  resources:
+  - connections
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - update
 - apiGroups:
   - networking.liqo.io
   resources:

internal/liqonet/tunnel-operator/tunnel-operator.go

@@ -41,7 +41,7 @@ import (
 
 	netv1alpha1 "github.com/liqotech/liqo/apis/net/v1alpha1"
 	liqoconst "github.com/liqotech/liqo/pkg/consts"
-	"github.com/liqotech/liqo/pkg/liqonet/conncheck"
+	"github.com/liqotech/liqo/pkg/gateway/connection/conncheck"
 	"github.com/liqotech/liqo/pkg/liqonet/iptables"
 	liqonetns "github.com/liqotech/liqo/pkg/liqonet/netns"
 	liqorouting "github.com/liqotech/liqo/pkg/liqonet/routing"

pkg/liqonet/conncheck/conncheck.go → pkg/gateway/connection/conncheck/conncheck.go

@@ -139,3 +139,11 @@ func (c *ConnChecker) GetConnected(clusterID string) (bool, error) {
 	}
 	return false, fmt.Errorf("sender %s not found", clusterID)
 }
+
+// IsSenderRunning returns true if the sender is running.
+func (c *ConnChecker) IsSenderRunning(clusterID string) bool {
+	c.sm.RLock()
+	defer c.sm.RUnlock()
+	_, ok := c.senders[clusterID]
+	return ok
+}

pkg/gateway/connection/connections-controller.go

@@ -0,0 +1,121 @@
+// Copyright 2019-2023 The Liqo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package connection
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+	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/predicate"
+
+	networkingv1alpha1 "github.com/liqotech/liqo/apis/networking/v1alpha1"
+	"github.com/liqotech/liqo/pkg/gateway/connection/conncheck"
+	"github.com/liqotech/liqo/pkg/gateway/tunnel/common"
+)
+
+// cluster-role
+// +kubebuilder:rbac:groups=networking.liqo.io,resources=connections,verbs=get;list;create;delete;update
+
+// ConnectionsReconciler updates the PublicKey resource used to establish the Wireguard connection.
+type ConnectionsReconciler struct {
+	ConnChecker    *conncheck.ConnChecker
+	Client         client.Client
+	Scheme         *runtime.Scheme
+	EventsRecorder record.EventRecorder
+	Options        *Options
+}
+
+// NewConnectionsReconciler returns a new PublicKeysReconciler.
+func NewConnectionsReconciler(ctx context.Context, cl client.Client,
+	s *runtime.Scheme, er record.EventRecorder, options *Options) (*ConnectionsReconciler, error) {
+	connchecker, err := conncheck.NewConnChecker()
+	if err != nil {
+		return nil, fmt.Errorf("unable to create the connection checker: %w", err)
+	}
+	go connchecker.RunReceiver(ctx)
+	go connchecker.RunReceiverDisconnectObserver(ctx)
+	return &ConnectionsReconciler{
+		ConnChecker:    connchecker,
+		Client:         cl,
+		Scheme:         s,
+		EventsRecorder: er,
+		Options:        options,
+	}, nil
+}
+
+// Reconcile manage PublicKey resources.
+func (r *ConnectionsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	connection := &networkingv1alpha1.Connection{}
+	if err := r.Client.Get(ctx, req.NamespacedName, connection); err != nil {
+		if apierrors.IsNotFound(err) {
+			klog.Infof("There is no connection %s", req.String())
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, fmt.Errorf("unable to get the connection %q: %w", req.NamespacedName, err)
+	}
+
+	if r.ConnChecker.IsSenderRunning(r.Options.RemoteClusterID) {
+		return ctrl.Result{}, nil
+	}
+
+	go r.ConnChecker.AddAndRunSender(
+		ctx, r.Options.RemoteClusterID,
+		common.GetRemoteInterfaceIP(r.Options.Mode),
+		ForgeUpdateConnectionCallback(ctx, r.Client, req),
+	)
+
+	return ctrl.Result{}, nil
+}
+
+// SetupWithManager register the ConfigurationReconciler to the manager.
+func (r *ConnectionsReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&networkingv1alpha1.Connection{}, r.Predicates()).
+		Complete(r)
+}
+
+// Predicates returns the predicates required for the PublicKey controller.
+func (r *ConnectionsReconciler) Predicates() builder.Predicates {
+	return builder.WithPredicates(
+		predicate.NewPredicateFuncs(func(object client.Object) bool {
+			return true
+		}))
+}
+
+// ForgeUpdateConnectionCallback forges the UpdateConnectionStatus function.
+func ForgeUpdateConnectionCallback(ctx context.Context, cl client.Client, req ctrl.Request) conncheck.UpdateFunc {
+	return func(connected bool, latency time.Duration, timestamp time.Time) error {
+		connection := &networkingv1alpha1.Connection{}
+		if err := cl.Get(ctx, req.NamespacedName, connection); err != nil {
+			return err
+		}
+		var connStatusValue networkingv1alpha1.ConnectionStatusValue
+		switch connected {
+		case true:
+			connStatusValue = networkingv1alpha1.Connected
+		case false:
+			connStatusValue = networkingv1alpha1.ConnectionError
+		}
+		return UpdateConnectionStatus(ctx, cl, connection, connStatusValue, latency, timestamp)
+	}
+}

pkg/gateway/connection/doc.go

@@ -0,0 +1,16 @@
+// Copyright 2019-2023 The Liqo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package connection menages the connection resource.
+package connection