summaryrefslogtreecommitdiff
path: root/tests/stubs/cloud/controllerstub
diff options
context:
space:
mode:
Diffstat (limited to 'tests/stubs/cloud/controllerstub')
-rw-r--r--tests/stubs/cloud/controllerstub/downstream.go58
-rw-r--r--tests/stubs/cloud/controllerstub/module.go92
-rw-r--r--tests/stubs/cloud/controllerstub/podmanager.go200
-rw-r--r--tests/stubs/cloud/controllerstub/upstream.go126
4 files changed, 0 insertions, 476 deletions
diff --git a/tests/stubs/cloud/controllerstub/downstream.go b/tests/stubs/cloud/controllerstub/downstream.go
deleted file mode 100644
index 788fce2d3..000000000
--- a/tests/stubs/cloud/controllerstub/downstream.go
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
-Copyright 2019 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 controllerstub
-
-import (
- "k8s.io/klog/v2"
-
- beehiveContext "github.com/kubeedge/beehive/pkg/core/context"
- "github.com/kubeedge/kubeedge/tests/stubs/common/constants"
-)
-
-// NewDownstreamController creates a downstream controller
-func NewDownstreamController(pm *PodManager) (*DownstreamController, error) {
- // New downstream controller
- dc := &DownstreamController{podManager: pm}
- return dc, nil
-}
-
-// DownstreamController receives http request and send to cloudhub
-type DownstreamController struct {
- podManager *PodManager
-}
-
-// Start DownstreamController
-func (dc *DownstreamController) Start() error {
- klog.Infof("Start downstream controller")
- go dc.SyncPods()
- return nil
-}
-
-// SyncPods is used to send message to cloudhub
-func (dc *DownstreamController) SyncPods() {
- for {
- select {
- case <-beehiveContext.Done():
- klog.Info("Stop sync pod")
- return
- case msg := <-dc.podManager.GetEvent():
- klog.Infof("Send message to cloudhub: %v", *msg)
- beehiveContext.Send(constants.CloudHub, *msg)
- klog.Info("Finish send message to cloudhub")
- }
- }
-}
diff --git a/tests/stubs/cloud/controllerstub/module.go b/tests/stubs/cloud/controllerstub/module.go
deleted file mode 100644
index 89077addd..000000000
--- a/tests/stubs/cloud/controllerstub/module.go
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
-Copyright 2019 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 controllerstub
-
-import (
- "net/http"
-
- "k8s.io/klog/v2"
-
- "github.com/kubeedge/beehive/pkg/core"
- "github.com/kubeedge/kubeedge/tests/stubs/common/constants"
-)
-
-// Init module
-func init() {
- core.Register(&ControllerStub{})
-}
-
-// HandlerStub definition
-type ControllerStub struct {
-}
-
-var _ core.Module = (*ControllerStub)(nil)
-
-func (*ControllerStub) Enable() bool {
- return true
-}
-
-// Return module name
-func (*ControllerStub) Name() string {
- return constants.ControllerStub
-}
-
-// Return module group
-func (*ControllerStub) Group() string {
- return constants.ControllerGroup
-}
-
-// Start controller hub
-func (cs *ControllerStub) Start() {
- // New pod manager
- pm, err := NewPodManager()
- if err != nil {
- klog.Errorf("Failed to create pod manager with error: %v", err)
- return
- }
-
- // Start downstream controller
- downstream, err := NewDownstreamController(pm)
- if err != nil {
- klog.Errorf("New downstream controller failed with error: %v", err)
- return
- }
- if err := downstream.Start(); err != nil {
- klog.Errorf("Start downstream controller failed with error: %v", err)
- return
- }
-
- // Start upstream controller
- upstream, err := NewUpstreamController(pm)
- if err != nil {
- klog.Errorf("New upstream controller failed with error: %v", err)
- return
- }
- if err := upstream.Start(); err != nil {
- klog.Errorf("Start upstream controller failed with error: %v", err)
- return
- }
-
- // Start http server
- http.HandleFunc(constants.PodResource, pm.PodHandlerFunc)
- klog.Info("Start http service")
- go func() {
- if err := http.ListenAndServe(":54321", nil); err != nil {
- klog.Errorf("Start http service failed with error: %v", err)
- }
- }()
-}
diff --git a/tests/stubs/cloud/controllerstub/podmanager.go b/tests/stubs/cloud/controllerstub/podmanager.go
deleted file mode 100644
index 6d3123c22..000000000
--- a/tests/stubs/cloud/controllerstub/podmanager.go
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
-Copyright 2019 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 controllerstub
-
-import (
- "bytes"
- "encoding/json"
- "io"
- "net/http"
- "sync"
- "time"
-
- "k8s.io/klog/v2"
-
- "github.com/kubeedge/beehive/pkg/core/model"
- "github.com/kubeedge/kubeedge/tests/stubs/common/constants"
- "github.com/kubeedge/kubeedge/tests/stubs/common/types"
- "github.com/kubeedge/kubeedge/tests/stubs/common/utils"
-)
-
-// NewPodManager creates pod manger
-func NewPodManager() (*PodManager, error) {
- event := make(chan *model.Message, 1024)
- pm := &PodManager{event: event}
- return pm, nil
-}
-
-// PodManager is a manager watch pod change event
-type PodManager struct {
- // event
- event chan *model.Message
- // pods map
- pods sync.Map
-}
-
-// GetEvent return a channel which receives event
-func (pm *PodManager) GetEvent() chan *model.Message {
- return pm.event
-}
-
-// AddPod adds pod in cache
-func (pm *PodManager) AddPod(k string, v types.FakePod) {
- pm.pods.Store(k, v)
-}
-
-// DeletePod deletes pod in cache
-func (pm *PodManager) DeletePod(k string) {
- pm.pods.Delete(k)
-}
-
-// UpdatePodStatus update pod status in cache
-func (pm *PodManager) UpdatePodStatus(k string, s string) {
- v, ok := pm.pods.Load(k)
- if ok {
- pod := v.(types.FakePod)
- // Status becomes running in the first time
- if pod.Status != s && s == constants.PodRunning {
- pod.RunningTime = time.Now().UnixNano()
- }
- pod.Status = s
- pm.pods.Store(k, pod)
- }
-}
-
-// GetPod gets pod from cache
-func (pm *PodManager) GetPod(key string) types.FakePod {
- v, ok := pm.pods.Load(key)
- if ok {
- return v.(types.FakePod)
- }
- return types.FakePod{}
-}
-
-// ListPods lists all pods in cache
-func (pm *PodManager) ListPods() []types.FakePod {
- pods := make([]types.FakePod, 0)
- pm.pods.Range(func(k, v interface{}) bool {
- pods = append(pods, v.(types.FakePod))
- return true
- })
- return pods
-}
-
-// PodHandlerFunc is used to receive and process message
-func (pm *PodManager) PodHandlerFunc(w http.ResponseWriter, req *http.Request) {
- switch req.Method {
- case http.MethodGet:
- // List Pod
- klog.V(4).Infof("Receive list pod request")
- pods := pm.ListPods()
- klog.V(4).Infof("Current pods number: %v", len(pods))
- rspBodyBytes := new(bytes.Buffer)
- if err := json.NewEncoder(rspBodyBytes).Encode(pods); err != nil {
- klog.Errorf("Encode to json file with error: %v", err)
- return
- }
- if _, err := w.Write(rspBodyBytes.Bytes()); err != nil {
- klog.Errorf("Write error: %v", err)
- return
- }
- klog.V(4).Infof("Finish list pod request")
- case http.MethodPost:
- klog.V(4).Infof("Receive add pod request")
- var p types.FakePod
- // Get request body
- if req.Body != nil {
- body, err := io.ReadAll(req.Body)
- if err != nil {
- klog.Errorf("Read body error %v", err)
- if _, err := w.Write([]byte("Read request body error")); err != nil {
- klog.Errorf("Write error: %v", err)
- }
- return
- }
- klog.V(4).Infof("Request body is %s", string(body))
- if err = json.Unmarshal(body, &p); err != nil {
- klog.Errorf("Unmarshal request body error %v", err)
- if _, err := w.Write([]byte("Unmarshal request body error")); err != nil {
- klog.Errorf("Wrire body error %v", err)
- }
- return
- }
- }
- // Add Pod
- ns := constants.NamespaceDefault
- if p.Namespace != "" {
- ns = p.Namespace
- }
-
- // Build Add message
- msg := model.NewMessage("")
- resource, err := utils.BuildResource(p.NodeName, p.Namespace, model.ResourceTypePod, p.Name)
- if err != nil {
- klog.Errorf("Build message resource failed with error: %s", err)
- if _, err := w.Write([]byte("Build message resource failed with error")); err != nil {
- klog.Errorf("Write body error %v", err)
- }
- return
- }
- msg.Content = p
- msg.BuildRouter(constants.ControllerStub, constants.GroupResource, resource, model.InsertOperation)
-
- // Add pod in cache
- p.CreateTime = time.Now().UnixNano()
- pm.AddPod(ns+"/"+p.Name, p)
-
- // Send msg
- pm.event <- msg
- klog.V(4).Infof("Finish add pod request")
-
- case http.MethodDelete:
- // Delete Pod
- klog.V(4).Infof("Receive delete pod request")
- params := req.URL.Query()
- ns := params.Get("namespace")
- if ns == "" {
- ns = constants.NamespaceDefault
- }
- nodename := params.Get("nodename")
- name := params.Get("name")
- klog.V(4).Infof("Pod Namespace: %s NodeName: %s Name: %s", ns, nodename, name)
-
- // Build delete message
- msg := model.NewMessage("")
- resource, err := utils.BuildResource(nodename, ns, model.ResourceTypePod, name)
- if err != nil {
- klog.Errorf("Build message resource failed with error: %s", err)
- if _, err := w.Write([]byte("Build message resource failed with error")); err != nil {
- klog.Errorf("write error: %v", err)
- }
- return
- }
- msg.Content = pm.GetPod(ns + "/" + name)
- msg.BuildRouter(constants.ControllerStub, constants.GroupResource, resource, model.DeleteOperation)
-
- // Delete pod in cache
- pm.DeletePod(ns + "/" + name)
-
- // Send msg
- pm.event <- msg
- klog.V(4).Infof("Finish delete pod request")
-
- default:
- klog.Errorf("Http type: %s unsupported", req.Method)
- }
-}
diff --git a/tests/stubs/cloud/controllerstub/upstream.go b/tests/stubs/cloud/controllerstub/upstream.go
deleted file mode 100644
index fd3c82191..000000000
--- a/tests/stubs/cloud/controllerstub/upstream.go
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
-Copyright 2019 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 controllerstub
-
-import (
- "encoding/json"
-
- "k8s.io/klog/v2"
-
- beehiveContext "github.com/kubeedge/beehive/pkg/core/context"
- "github.com/kubeedge/beehive/pkg/core/model"
- "github.com/kubeedge/kubeedge/tests/stubs/common/constants"
- "github.com/kubeedge/kubeedge/tests/stubs/common/types"
- "github.com/kubeedge/kubeedge/tests/stubs/common/utils"
-)
-
-// NewUpstreamController creates a upstream controller
-func NewUpstreamController(pm *PodManager) (*UpstreamController, error) {
- // New upstream controller
- uc := &UpstreamController{podManager: pm}
- return uc, nil
-}
-
-// UpstreamController subscribe messages from edge
-type UpstreamController struct {
- podManager *PodManager
- podStatusChan chan model.Message
-}
-
-// Start UpstreamController
-func (uc *UpstreamController) Start() error {
- klog.Infof("Start upstream controller")
- uc.podStatusChan = make(chan model.Message, 1024)
-
- go uc.WaitforMessage()
- go uc.UpdatePodStatus()
-
- return nil
-}
-
-// WaitforMessage from cloudhub
-func (uc *UpstreamController) WaitforMessage() {
- for {
- select {
- case <-beehiveContext.Done():
- klog.Infof("Stop waiting for message")
- return
- default:
- }
- // Receive message from cloudhub
- msg, err := beehiveContext.Receive(constants.ControllerStub)
- if err != nil {
- klog.Errorf("Receive message failed: %v", err)
- continue
- }
- klog.V(4).Infof("Receive message: %v", msg)
-
- // Get resource type in message
- resourceType, err := utils.GetResourceType(msg)
- if err != nil {
- klog.Errorf("Get message: %s resource type with error: %v", msg.GetID(), err)
- continue
- }
- klog.Infof("Message: %s resource type: %s", msg.GetID(), resourceType)
-
- switch resourceType {
- case model.ResourceTypePodStatus:
- uc.podStatusChan <- msg
- default:
- klog.V(4).Infof("Message: %s, resource type: %s unsupported", msg.GetID(), resourceType)
- }
- }
-}
-
-// UpdatePodStatus is used to update pod status in cache map
-func (uc *UpstreamController) UpdatePodStatus() {
- for {
- select {
- case <-beehiveContext.Done():
- klog.Infof("Stop updatePodStatus")
- return
- case msg := <-uc.podStatusChan:
- klog.Infof("Message: %s operation: %s resource: %s",
- msg.GetID(), msg.GetOperation(), msg.GetResource())
- switch msg.GetOperation() {
- case model.UpdateOperation:
- // get message content data
- data, err := msg.GetContentData()
- if err != nil {
- klog.Warningf("message: %s process failure, get content data failed: %s", msg.GetID(), err)
- continue
- }
-
- // Get pod
- var pod types.FakePod
- if err := json.Unmarshal(data, &pod); err != nil {
- klog.Errorf("Unmarshal content failed with error: %s, %v", msg.GetID(), err)
- continue
- }
-
- // Update pod status in cache
- uc.podManager.UpdatePodStatus(pod.Namespace+"/"+pod.Name, pod.Status)
-
- klog.Infof("Pod namespace: %s name: %s status: %s",
- pod.Namespace, pod.Name, pod.Status)
- default:
- klog.V(4).Infof("Pod operation: %s unsupported", msg.GetOperation())
- }
- klog.V(4).Infof("Message: %s process successfully", msg.GetID())
- }
- }
-}
generated by cgit v1.2.3 (git 2.52.0) at 2026-08-20 01:57:53 +0000