Merge branch 'main' of https://cloud.o-forge.io/core/oc-datacenter

Dockerfile

@@ -30,7 +30,7 @@ RUN export CGO_ENABLED=0 && \
 COPY . .
 
 RUN sed -i '/replace/d' go.mod
-RUN if [ ! -f swagger/index.html ]; then timeout 15 bee run --gendoc=true --downdoc=true; fi
+RUN if [ ! -f swagger/index.html ]; then timeout 15 bee run -gendoc=true -downdoc=true; fi
 RUN bee generate routers
 RUN bee generate docs
 RUN bee pack

controllers/admiralty.go

@@ -10,10 +10,13 @@ import (
 	"slices"
 	"time"
 
+	oclib "cloud.o-forge.io/core/oc-lib"
+
 	beego "github.com/beego/beego/v2/server/web"
 	jwt "github.com/golang-jwt/jwt/v5"
 	"gopkg.in/yaml.v2"
 	v1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
 type KubeInfo struct {
@@ -139,6 +142,12 @@ func (c *AdmiraltyController) CreateSource() {
 
 	res, err := serv.CreateAdmiraltySource(c.Ctx.Request.Context(),execution)
 	if err != nil {
+		if apierrors.IsAlreadyExists(err) {
+			c.Ctx.Output.SetStatus(409)
+			c.Data["json"] = map[string]string{"info" : "A source already exists for this namespace : " + execution}
+			c.ServeJSON()
+			return
+		}
 		// change code to 500
 		c.Ctx.Output.SetStatus(500)
 		c.Data["json"] = map[string]string{"error": err.Error()}
@@ -315,7 +324,6 @@ func (c *AdmiraltyController) CreateKubeSecret() {
 // @router /node/:execution [get]
 func (c *AdmiraltyController) GetNodeReady(){
 	var secret v1.Secret
-	
 	execution := c.Ctx.Input.Param(":execution")
 
 
@@ -339,12 +347,14 @@ func (c *AdmiraltyController) GetNodeReady(){
 	if node == nil {
 		c.Ctx.Output.SetStatus(404)
 		c.Data["json"] = map[string]string{
-			"error" : "the node for " + execution + " can't be found, make sure both target and source resources are set up on local and remote hosts",
+			"node" : "the node for " + execution + " can't be found, make sure both target and source resources are set up on local and remote hosts",
 		}
 		c.ServeJSON()
 		return
 	}
 
+
+
 	resp, err := serv.GetKubeconfigSecret(c.Ctx.Request.Context(),execution)
 	if err != nil {
 		// change code to 500
@@ -360,7 +370,6 @@ func (c *AdmiraltyController) GetNodeReady(){
 		return
 	}
 
-
 	// Extract JWT token RS265 encoded
 	var editedKubeconfig map[string]interface{}
 	json.Unmarshal(resp,&secret)
@@ -393,14 +402,15 @@ func (c *AdmiraltyController) GetNodeReady(){
 	}
 
 	if *isExpired {
-		c.Data["json"] = map[string]string{
+		c.Data["json"] = map[string]interface{}{
 			"token" : "token in the secret is expired and must be regenerated",
+			"node": node,
 		}
 		c.Ctx.Output.SetStatus(410)
 		c.ServeJSON()
 	}
 
-	c.Data["json"] = map[string]bool{"ok": true}
+	c.Data["json"] = map[string]interface{}{"node": node,"token": true}
 	c.ServeJSON()
 	
 }
@@ -426,6 +436,8 @@ func retrieveTokenFromKonfig(editedKubeconfig map[string]interface{}) (string,er
 }
 
 func isTokenExpired(token string) (*bool, error){
+	logger := oclib.GetLogger()
+
 	t, _, err := new(jwt.Parser).ParseUnverified(token, jwt.MapClaims{})
 	if err != nil {
 		fmt.Println("couldn't decode token")
@@ -437,7 +449,11 @@ func isTokenExpired(token string) (*bool, error){
 		fmt.Println("Error while checking token's expiration time")
 		return nil, err
 	}
-	fmt.Println("Expiration date : " + expiration.UTC().Format("2006-01-02T15:04:05"))
+
+	logger.Debug().Msg("Expiration date : " + expiration.UTC().Format("2006-01-02T15:04:05"))
+	logger.Debug().Msg(fmt.Sprint("Now : ", time.Now().Unix()))
+	logger.Debug().Msg(fmt.Sprint("Token : ", expiration.Unix()))
+
 
 	expired := expiration.Unix() < time.Now().Unix()
 

infrastructure/kubernetes.go

@@ -1,13 +1,11 @@
 package infrastructure
 
 import (
-	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"html/template"
 	"oc-datacenter/conf"
 	"strings"
 
@@ -16,14 +14,42 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	apply "k8s.io/client-go/applyconfigurations/core/v1"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 )
 
+var gvrSources = schema.GroupVersionResource{Group: "multicluster.admiralty.io", Version: "v1alpha1", Resource: "sources"}
+var gvrTargets = schema.GroupVersionResource{Group: "multicluster.admiralty.io", Version: "v1alpha1", Resource: "targets"}
+
 type KubernetesService struct {
 	Set 			*kubernetes.Clientset
 }
 
+func NewDynamicClient() (*dynamic.DynamicClient, error) {
+	config := &rest.Config{
+		Host: conf.GetConfig().KubeHost + ":" + conf.GetConfig().KubePort,
+		TLSClientConfig: rest.TLSClientConfig{
+			CAData:   []byte(conf.GetConfig().KubeCA),
+			CertData: []byte(conf.GetConfig().KubeCert),
+			KeyData:  []byte(conf.GetConfig().KubeData),
+		},
+	}
+
+	dynamicClient, err := dynamic.NewForConfig(config)
+	if err != nil {
+		return nil, errors.New("Error creating Dynamic client: " + err.Error())
+	}
+	if dynamicClient == nil {
+		return nil, errors.New("Error creating Dynamic client: dynamicClient is nil")
+	}
+
+	return dynamicClient, nil
+}
+
 func NewKubernetesService() (Infrastructure, error) {
 	config := &rest.Config{
 		Host: conf.GetConfig().KubeHost + ":" + conf.GetConfig().KubePort,
@@ -33,6 +59,7 @@ func NewKubernetesService() (Infrastructure, error) {
 			KeyData:  []byte(conf.GetConfig().KubeData),
 		},
 	}
+	
 	// Create clientset
 	clientset, err := kubernetes.NewForConfig(config)
 	fmt.Println("NewForConfig", clientset, err)
@@ -43,6 +70,7 @@ func NewKubernetesService() (Infrastructure, error) {
 		return nil, errors.New("Error creating Kubernetes client: clientset is nil")
 	}
 
+
 	return &KubernetesService{
 		Set: clientset,
 	}, nil
@@ -82,6 +110,9 @@ func (k *KubernetesService) CreateNamespace(ctx context.Context, ns string) erro
 	namespace := &v1.Namespace{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: ns,
+			Labels: map[string]string{
+				"multicluster-scheduler":"enabled",
+			},
 		},
 	}
 	// Create the namespace
@@ -261,32 +292,28 @@ func (k *KubernetesService) CreateAdmiraltyTarget(context context.Context, execu
 		return nil, nil // Maybe we could create a wrapper for errors and add more info to have
 	}
 	
-	var targetManifest string
+
-	var tpl bytes.Buffer
+	target := map[string]interface{}{
-	tmpl, err := template.New("target").
+        "apiVersion": "multicluster.admiralty.io/v1alpha1",
-		Parse("{\"apiVersion\": \"multicluster.admiralty.io/v1alpha1\", \"kind\": \"Target\", \"metadata\": {\"name\": \"target-{{.ExecutionId}}\"}, \"spec\": { \"kubeconfigSecret\" :{\"name\": \"kube-secret-{{.ExecutionId}}\"}} }")
+        "kind":       "Target",
-	if err != nil {
+        "metadata": map[string]interface{}{
-		fmt.Println("Error creating the template for the target Manifest")
+            "name":      "target-"+executionId,
-		return nil, err
+            "namespace": executionId,
+        },
+        "spec": map[string]interface{}{
+            "kubeconfigSecret": map[string]string{
+				"name" : "kube-secret-"+executionId,
+			},
+        },
     }
 
-	err = tmpl.Execute(&tpl, map[string]string{"ExecutionId": executionId})
+	res, err := dynamicClientApply(executionId, "target", gvrTargets, context, target)
-	targetManifest = tpl.String()
-	resp, err := postCDRapiKube(
-		*k.Set,
-		context,
-		"/apis/multicluster.admiralty.io/v1alpha1/namespaces/"+executionId+"/targets",
-		[]byte(targetManifest),
-		map[string]string{"fieldManager": "kubectl-client-side-apply"},
-		map[string]string{"fieldValidation": "Strict"},
-	)
-
 	if err != nil {
-		fmt.Println("Error trying to create a Source on remote cluster : ", err, " : ", resp)
+		return nil, errors.New("Error when trying to apply Source definition :" + err.Error())
-		return nil, err
 	}
 
-	return resp, nil
+	return res, nil
+
 }
 
 // Admiralty Source allows a cluster to receive pods from a remote cluster
@@ -297,35 +324,27 @@ func (k *KubernetesService) CreateAdmiraltyTarget(context context.Context, execu
 // This method is temporary to implement the use of Admiralty, but must be edited
 // to rather contact the oc-datacenter from the remote cluster to create the source
 // locally and retrieve the token for the serviceAccount
-func (k *KubernetesService) CreateAdmiraltySource(context context.Context, executionId string) ([]byte, error) {
+func (k *KubernetesService) CreateAdmiraltySource(context context.Context,executionId string) ([]byte, error) {
-	var sourceManifest string
+
-	var tpl bytes.Buffer
+	source := map[string]interface{}{
-	tmpl, err := template.New("source").
+        "apiVersion": "multicluster.admiralty.io/v1alpha1",
-		Parse("{\"apiVersion\": \"multicluster.admiralty.io/v1alpha1\", \"kind\": \"Source\", \"metadata\": {\"name\": \"source-{{.ExecutionId}}\"}, \"spec\": {\"serviceAccountName\": \"sa-{{.ExecutionId}}\"} }")
+        "kind":       "Source",
-	if err != nil {
+        "metadata": map[string]interface{}{
-		fmt.Println("Error creating the template for the source Manifest")
+            "name":      "source-"+executionId,
-		return nil, err
+            "namespace": executionId,
+        },
+        "spec": map[string]interface{}{
+            "serviceAccountName": "sa-"+executionId,
+        },
     }
 
-	err = tmpl.Execute(&tpl, map[string]string{"ExecutionId": executionId})
-	sourceManifest = tpl.String()
 
-	resp, err := postCDRapiKube(
+	res, err := dynamicClientApply(executionId, "source",gvrSources, context, source)
-		*k.Set,
-		context,
-		"/apis/multicluster.admiralty.io/v1alpha1/namespaces/"+executionId+"/sources",
-		[]byte(sourceManifest),
-		map[string]string{"fieldManager": "kubectl-client-side-apply"},
-		map[string]string{"fieldValidation": "Strict"},
-	)
-
-	// We can add more info to the log with the content of resp if not nil
 	if err != nil {
-		fmt.Println("Error trying to create a Source on remote cluster : ", err, " : ", resp)
+		return nil, errors.New("Error when trying to apply Source definition :" + err.Error())
-		return nil, err
 	}
 
-	return resp, nil
+	return res, nil
 }
 
 // Create a secret from a kubeconfing. Use it to create the secret binded to an Admiralty
@@ -339,32 +358,37 @@ func (k *KubernetesService) CreateKubeconfigSecret(context context.Context, kube
 		return nil, err
 	}
 
-	secretManifest := &v1.Secret{
+	secretApplyConfig := apply.Secret("kube-secret-" + executionId,
-		ObjectMeta: metav1.ObjectMeta{
+									executionId).
-			Name:      "kube-secret-" + executionId,
+								WithData(map[string][]byte{
-			Namespace: executionId,
-		},
-		Data: map[string][]byte{
 										"config": config,
 										},
-	}
+									)
+
+
+
+	// exists, err := k.GetKubeconfigSecret(context,executionId)
+	// if err != nil {
+	// 	fmt.Println("Error verifying if kube secret exists in namespace ", executionId)
+	// 	return nil, err
+	// }
+	// if exists != nil {
+	// 	fmt.Println("kube-secret already exists in namespace", executionId)
+	// 	fmt.Println("Overriding existing kube-secret with a newer resource")
+	// 	// TODO : implement DeleteKubeConfigSecret(executionID)
+	// 	deleted, err := k.DeleteKubeConfigSecret(executionId)
+	// 	_ = deleted
+	// 	_ = err
+	// }
+
 
-	exists, err := k.GetKubeconfigSecret(context, executionId)
-	if err != nil {
-		fmt.Println("Error verifying if kube secret exists in namespace ", executionId)
-		return nil, err
-	}
-	if exists != nil {
-		fmt.Println("kube-secret already exists in namespace", executionId)
-		fmt.Println("Overriding existing kube-secret with a newer resource")
-		// TODO : implement DeleteKubeConfigSecret(executionID)
-		deleted, err := k.DeleteKubeConfigSecret(executionId)
-		_ = deleted
-		_ = err
-	}
 	resp, err := k.Set.CoreV1().
 				Secrets(executionId).
-		Create(context, secretManifest, metav1.CreateOptions{})
+				Apply(context,
+					secretApplyConfig,
+					metav1.ApplyOptions{
+						FieldManager: "admiralty-manager",
+				})
 				
 	if err != nil {
 		fmt.Println("Error while trying to contact API to get secret kube-secret-" + executionId)
@@ -425,7 +449,39 @@ func getCDRapiKube(client kubernetes.Clientset, ctx context.Context, path string
 	return resp, nil
 }
 
-func postCDRapiKube(client kubernetes.Clientset, ctx context.Context, path string, body []byte, params ...map[string]string) ([]byte, error) {
+func dynamicClientApply(executionId string, typeResource string, resourceDefinition schema.GroupVersionResource, ctx context.Context, object map[string]interface{}) ([]byte, error) {
+	cli, err := NewDynamicClient()
+	if err != nil {
+		return nil, errors.New("Could not retrieve dynamic client when creating Admiralty Source : " + err.Error())
+	}
+
+	res, err := cli.Resource(resourceDefinition).
+				Namespace(executionId).
+				Apply(ctx, 
+					typeResource + "-" + executionId,
+					&unstructured.Unstructured{Object: object},
+					metav1.ApplyOptions{
+						FieldManager: "kubectl-client-side-apply", 
+					},
+				)
+	if err != nil {
+		fmt.Println("Error from k8s API when applying " + fmt.Sprint(object) + " to " + gvrSources.String() + " : " , err)
+		return nil,err
+	}
+
+
+	// We can add more info to the log with the content of resp if not nil
+	resByte, err := json.Marshal(res) 
+	if err != nil {
+		// fmt.Println("Error trying to create a Source on remote cluster : ", err , " : ", res)
+		return nil, err
+	}
+
+	return resByte, nil
+
+}
+
+func putCDRapiKube(client kubernetes.Clientset, ctx context.Context, path string, body []byte, params ...map[string]string) ([]byte, error){
 	req := client.RESTClient().
 		Post().
 		AbsPath(path).