Merge pull request #134 from llhuii/refactor-gm

gm: decouple all features into independent package

cmd/sedna-gm/app/controller.go

@@ -18,6 +18,7 @@ package app
 
 import (
 	"fmt"
+	"os"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
@@ -27,7 +28,7 @@ import (
 	"k8s.io/klog/v2"
 
 	"github.com/kubeedge/sedna/cmd/sedna-gm/app/options"
-	controller "github.com/kubeedge/sedna/pkg/globalmanager"
+	controller "github.com/kubeedge/sedna/pkg/globalmanager/controllers"
 	"github.com/kubeedge/sedna/pkg/util"
 	"github.com/kubeedge/sedna/pkg/version/verflag"
 )
@@ -61,8 +62,12 @@ func NewControllerCommand() *cobra.Command {
 			if errs := config.Validate(); len(errs) > 0 {
 				klog.Fatal(util.SpliceErrors(errs.ToAggregate().Errors()))
 			}
-			c := controller.NewController(config)
-			c.Start()
+			c := controller.New(config)
+			err = c.Start()
+			if err != nil {
+				klog.Errorf("failed to start controller: %v", err)
+				os.Exit(1)
+			}
 		},
 	}
 	fs := cmd.Flags()

cmd/sedna-gm/sedna-gm.go

@@ -17,14 +17,18 @@ limitations under the License.
 package main
 
 import (
+	"math/rand"
 	"os"
+	"time"
 
 	"k8s.io/component-base/logs"
 
 	"github.com/kubeedge/sedna/cmd/sedna-gm/app"
 )
 
 func main() {
+	rand.Seed(time.Now().UnixNano())
+
 	command := app.NewControllerCommand()
 	logs.InitLogs()
 	defer logs.FlushLogs()

pkg/globalmanager/controllers/dataset/dataset.go

@@ -0,0 +1,74 @@
+/*
+Copyright 2021 The KubeEdge 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 dataset
+
+import (
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	sednaclientset "github.com/kubeedge/sedna/pkg/client/clientset/versioned/typed/sedna/v1alpha1"
+	"github.com/kubeedge/sedna/pkg/globalmanager/config"
+	"github.com/kubeedge/sedna/pkg/globalmanager/runtime"
+)
+
+const (
+	// KindName is the kind name of CR this controller controls
+	KindName = "Dataset"
+
+	// Name is this controller name
+	Name = "Dataset"
+)
+
+// Controller handles all dataset objects including: syncing to edge and update from edge.
+type Controller struct {
+	kubeClient kubernetes.Interface
+	client     sednaclientset.SednaV1alpha1Interface
+
+	cfg *config.ControllerConfig
+
+	sendToEdgeFunc runtime.DownstreamSendFunc
+}
+
+func (c *Controller) Run(stopCh <-chan struct{}) {
+	// noop now
+}
+
+// New creates a dataset controller
+func New(cc *runtime.ControllerContext) (runtime.FeatureControllerI, error) {
+	c := &Controller{
+		client:     cc.SednaClient.SednaV1alpha1(),
+		kubeClient: cc.KubeClient,
+	}
+	informer := cc.SednaInformerFactory.Sedna().V1alpha1().Datasets().Informer()
+	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+
+		AddFunc: func(obj interface{}) {
+			c.syncToEdge(watch.Added, obj)
+		},
+
+		UpdateFunc: func(old, cur interface{}) {
+			c.syncToEdge(watch.Added, cur)
+		},
+
+		DeleteFunc: func(obj interface{}) {
+			c.syncToEdge(watch.Deleted, obj)
+		},
+	})
+
+	return c, nil
+}

pkg/globalmanager/controllers/dataset/downstream.go

@@ -0,0 +1,54 @@
+/*
+Copyright 2021 The KubeEdge 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 dataset
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/watch"
+
+	sednav1 "github.com/kubeedge/sedna/pkg/apis/sedna/v1alpha1"
+	"github.com/kubeedge/sedna/pkg/globalmanager/runtime"
+)
+
+// syncToEdge syncs the dataset resources
+func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) error {
+	dataset, ok := obj.(*sednav1.Dataset)
+	if !ok {
+		return nil
+	}
+
+	// Since t.Kind may be empty,
+	// we need to fix the kind here if missing.
+	// more details at https://github.com/kubernetes/kubernetes/issues/3030
+	dataset.Kind = KindName
+
+	// Here only propagate to the nodes with non empty name
+	nodeName := dataset.Spec.NodeName
+	if len(nodeName) == 0 {
+		return fmt.Errorf("empty node name")
+	}
+
+	runtime.InjectSecretAnnotations(c.kubeClient, dataset, dataset.Spec.CredentialName)
+
+	return c.sendToEdgeFunc(nodeName, eventType, dataset)
+}
+
+func (c *Controller) SetDownstreamSendFunc(f runtime.DownstreamSendFunc) error {
+	c.sendToEdgeFunc = f
+	return nil
+}

pkg/globalmanager/controllers/dataset/upstream.go

@@ -0,0 +1,62 @@
+/*
+Copyright 2021 The KubeEdge 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 dataset
+
+import (
+	"context"
+	"encoding/json"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	sednav1 "github.com/kubeedge/sedna/pkg/apis/sedna/v1alpha1"
+	"github.com/kubeedge/sedna/pkg/globalmanager/runtime"
+)
+
+// updateFromEdge syncs update from edge
+func (c *Controller) updateFromEdge(name, namespace, operation string, content []byte) error {
+	status := sednav1.DatasetStatus{}
+	err := json.Unmarshal(content, &status)
+	if err != nil {
+		return err
+	}
+
+	return c.updateStatus(name, namespace, status)
+}
+
+// updateStatus updates the dataset status
+func (c *Controller) updateStatus(name, namespace string, status sednav1.DatasetStatus) error {
+	client := c.client.Datasets(namespace)
+
+	if status.UpdateTime == nil {
+		now := metav1.Now()
+		status.UpdateTime = &now
+	}
+
+	return runtime.RetryUpdateStatus(name, namespace, func() error {
+		dataset, err := client.Get(context.TODO(), name, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		dataset.Status = status
+		_, err = client.UpdateStatus(context.TODO(), dataset, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+func (c *Controller) SetUpstreamHandler(addFunc runtime.UpstreamHandlerAddFunc) error {
+	return addFunc(KindName, c.updateFromEdge)
+}

pkg/globalmanager/controllers/federatedlearning/downstream.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2021 The KubeEdge 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 federatedlearning
+
+import (
+	"k8s.io/apimachinery/pkg/watch"
+
+	sednav1 "github.com/kubeedge/sedna/pkg/apis/sedna/v1alpha1"
+	"github.com/kubeedge/sedna/pkg/globalmanager/runtime"
+)
+
+func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) error {
+	job, ok := obj.(*sednav1.FederatedLearningJob)
+	if !ok {
+		return nil
+	}
+
+	// Since Kind may be empty,
+	// we need to fix the kind here if missing.
+	// more details at https://github.com/kubernetes/kubernetes/issues/3030
+	job.Kind = KindName
+
+	// broadcast to all nodes specified in spec
+	nodeset := make(map[string]bool)
+	for _, trainingWorker := range job.Spec.TrainingWorkers {
+		// Here only propagate to the nodes with non empty name
+		if len(trainingWorker.Template.Spec.NodeName) > 0 {
+			nodeset[trainingWorker.Template.Spec.NodeName] = true
+		}
+	}
+
+	for nodeName := range nodeset {
+		c.sendToEdgeFunc(nodeName, eventType, job)
+	}
+	return nil
+}
+
+func (c *Controller) SetDownstreamSendFunc(f runtime.DownstreamSendFunc) error {
+	c.sendToEdgeFunc = f
+
+	return nil
+}