mirror of
https://github.com/vee1e/kubeedge.git
synced 2026-09-03 19:27:35 +00:00
374 lines
12 KiB
Go
374 lines
12 KiB
Go
/*
|
|
Copyright 2022 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 testing
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"strconv"
|
|
"sync"
|
|
"time"
|
|
|
|
v1 "k8s.io/api/core/v1"
|
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
|
v12 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/diff"
|
|
core "k8s.io/client-go/testing"
|
|
"k8s.io/klog/v2"
|
|
|
|
"github.com/kubeedge/api/apis/reliablesyncs/v1alpha1"
|
|
"github.com/kubeedge/api/client/clientset/versioned/fake"
|
|
beehivemodel "github.com/kubeedge/beehive/pkg/core/model"
|
|
"github.com/kubeedge/kubeedge/cloud/pkg/cloudhub/common"
|
|
"github.com/kubeedge/kubeedge/cloud/pkg/common/messagelayer"
|
|
"github.com/kubeedge/kubeedge/cloud/pkg/common/modules"
|
|
)
|
|
|
|
var (
|
|
NoErrors []ReactorError
|
|
NoObjectSyncs []*v1alpha1.ObjectSync
|
|
)
|
|
|
|
type TestCase struct {
|
|
// Name of the test, for logging
|
|
Name string
|
|
|
|
// Initial content of ObjectSync cache.
|
|
InitialObjectSyncs []*v1alpha1.ObjectSync
|
|
|
|
// Expected content of ObjectSync cache at the end of the test.
|
|
ExpectedObjectSyncs []*v1alpha1.ObjectSync
|
|
|
|
// Expected message store in node message pool
|
|
ExpectedStoreMessage *beehivemodel.Message
|
|
|
|
// reactorErrors to produce on matching ObjectSync action
|
|
ReactorErrors []ReactorError
|
|
|
|
// injectedConnErrTimes the connection write err times, used for node session test
|
|
InjectedConnErrTimes int
|
|
|
|
// Initial content of message cache.
|
|
InitialMessages []*beehivemodel.Message
|
|
|
|
// CurrentArriveMessage current arrive message
|
|
CurrentArriveMessage *beehivemodel.Message
|
|
|
|
// Simulate downstream message from edgeController or syncController
|
|
SimulateMessageFunc func(pool *common.NodeMessagePool, messages []*beehivemodel.Message)
|
|
}
|
|
|
|
// ErrVersionConflict is the error returned when resource version of requested
|
|
// object conflicts with the object in storage.
|
|
var ErrVersionConflict = errors.New("VersionError")
|
|
|
|
// ObjectSyncReactor is a core.Reactor that simulates etcd and API server. It
|
|
// stores:
|
|
// - Latest version of objectSyncs saved by the session.
|
|
// - Optionally, list of error that should be returned by reactor, simulating
|
|
// etcd / API server failures. These errors are evaluated in order and every
|
|
// error is returned only once. I.e. when the reactor finds matching
|
|
// ReactorError, it return appropriate error and removes the ReactorError from
|
|
// the list.
|
|
type ObjectSyncReactor struct {
|
|
objectSyncs map[string]*v1alpha1.ObjectSync
|
|
lock sync.RWMutex
|
|
errors []ReactorError
|
|
}
|
|
|
|
// ReactorError is an error that is returned by test reactor (=simulated
|
|
// etcd+/API server) when an action performed by the reactor matches given verb
|
|
// ("get", "update", "create", "delete" or "*"") on given resource
|
|
// ("objectsyncs" or "*").
|
|
type ReactorError struct {
|
|
Verb string
|
|
Resource string
|
|
Error error
|
|
}
|
|
|
|
// React is a callback called by fake kubeClient from the controller.
|
|
// In other words, every objectSync change performed by the session ends
|
|
// here.
|
|
// This callback checks versions of the updated objects and refuse those that
|
|
// are too old (simulating real etcd).
|
|
// All updated objects are stored locally to keep track of object versions and
|
|
// to evaluate test results.
|
|
func (r *ObjectSyncReactor) React(action core.Action) (handled bool, ret runtime.Object, err error) {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
|
|
klog.V(4).Infof("reactor got operation %q on %q", action.GetVerb(), action.GetResource())
|
|
|
|
// Inject error when requested
|
|
err = r.injectReactError(action)
|
|
if err != nil {
|
|
return true, nil, err
|
|
}
|
|
|
|
// Test did not request to inject an error, continue simulating API server.
|
|
switch {
|
|
case action.Matches("create", "objectsyncs"):
|
|
obj := action.(core.UpdateAction).GetObject()
|
|
objectSync := obj.(*v1alpha1.ObjectSync)
|
|
|
|
// check the objectSync does not exist
|
|
_, found := r.objectSyncs[objectSync.Name]
|
|
if found {
|
|
return true, nil, fmt.Errorf("cannot create objectSync %s: objectSync already exists", objectSync.Name)
|
|
}
|
|
|
|
// Store the updated object to appropriate places.
|
|
r.objectSyncs[objectSync.Name] = objectSync
|
|
|
|
klog.V(4).Infof("created objectSync %s", objectSync.Name)
|
|
return true, objectSync, nil
|
|
|
|
case action.Matches("update", "objectsyncs"):
|
|
obj := action.(core.UpdateAction).GetObject()
|
|
objectSync := obj.(*v1alpha1.ObjectSync)
|
|
|
|
// Check and bump object version
|
|
storedObjectSync, found := r.objectSyncs[objectSync.Name]
|
|
if found {
|
|
storedVer, _ := strconv.Atoi(storedObjectSync.ResourceVersion)
|
|
requestedVer, _ := strconv.Atoi(objectSync.ResourceVersion)
|
|
if storedVer != requestedVer {
|
|
return true, obj, ErrVersionConflict
|
|
}
|
|
if reflect.DeepEqual(storedObjectSync, objectSync) {
|
|
klog.V(4).Infof("nothing updated objectSync %s", objectSync.Name)
|
|
return true, objectSync, nil
|
|
}
|
|
// Don't modify the existing object
|
|
objectSync = objectSync.DeepCopy()
|
|
objectSync.ResourceVersion = strconv.Itoa(storedVer + 1)
|
|
} else {
|
|
return true, nil, fmt.Errorf("cannot update objectSync %s: objectSync not found", objectSync.Name)
|
|
}
|
|
|
|
r.objectSyncs[objectSync.Name] = objectSync
|
|
klog.V(4).Infof("saved updated objectSync %s", objectSync.Name)
|
|
return true, objectSync, nil
|
|
|
|
case action.Matches("get", "objectsyncs"):
|
|
name := action.(core.GetAction).GetName()
|
|
objectSync, found := r.objectSyncs[name]
|
|
if found {
|
|
klog.V(4).Infof("GetObjectSync: found %s", objectSync.Name)
|
|
return true, objectSync.DeepCopy(), nil
|
|
}
|
|
klog.V(4).Infof("GetObjectSync: objectSync %s not found", name)
|
|
return true, nil, apierrors.NewNotFound(action.GetResource().GroupResource(), name)
|
|
|
|
case action.Matches("delete", "objectsyncs"):
|
|
name := action.(core.DeleteAction).GetName()
|
|
klog.V(4).Infof("deleted objectSync %s", name)
|
|
_, found := r.objectSyncs[name]
|
|
if found {
|
|
delete(r.objectSyncs, name)
|
|
return true, nil, nil
|
|
}
|
|
return true, nil, fmt.Errorf("cannot delete objectSync %s: not found", name)
|
|
}
|
|
|
|
return false, nil, nil
|
|
}
|
|
|
|
// injectReactError returns an error when the test requested given action to
|
|
// fail. nil is returned otherwise.
|
|
func (r *ObjectSyncReactor) injectReactError(action core.Action) error {
|
|
if len(r.errors) == 0 {
|
|
// No more errors to inject, everything should succeed.
|
|
return nil
|
|
}
|
|
|
|
for i, expected := range r.errors {
|
|
klog.V(4).Infof("trying to match %q %q with %q %q", expected.Verb, expected.Resource, action.GetVerb(), action.GetResource())
|
|
if action.Matches(expected.Verb, expected.Resource) {
|
|
// That's the action we're waiting for, remove it from injectedErrors
|
|
r.errors = append(r.errors[:i], r.errors[i+1:]...)
|
|
klog.V(4).Infof("reactor found matching error at index %d: %q %q, returning %v", i, expected.Verb, expected.Resource, expected.Error)
|
|
return expected.Error
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// AddObjectSync adds a objectSync into ObjectSyncReactor.
|
|
func (r *ObjectSyncReactor) AddObjectSync(objectSync *v1alpha1.ObjectSync) {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
r.objectSyncs[objectSync.Name] = objectSync
|
|
}
|
|
|
|
// DeleteObjectSync deletes a objectSync by name.
|
|
func (r *ObjectSyncReactor) DeleteObjectSync(name string) {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
delete(r.objectSyncs, name)
|
|
}
|
|
|
|
// AddObjectSyncs adds objectSyncs into ObjectSyncReactor.
|
|
func (r *ObjectSyncReactor) AddObjectSyncs(objectSyncs []*v1alpha1.ObjectSync) {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
for _, objectSync := range objectSyncs {
|
|
r.objectSyncs[objectSync.Name] = objectSync
|
|
}
|
|
}
|
|
|
|
// CheckObjectSyncs compares all expectedObjectSyncs with set of objectSyncs at the end of the
|
|
// test and reports differences.
|
|
func (r *ObjectSyncReactor) CheckObjectSyncs(expectedObjectSyncs []*v1alpha1.ObjectSync) error {
|
|
r.lock.Lock()
|
|
defer r.lock.Unlock()
|
|
|
|
expectedMap := make(map[string]*v1alpha1.ObjectSync)
|
|
gotMap := make(map[string]*v1alpha1.ObjectSync)
|
|
for _, c := range expectedObjectSyncs {
|
|
// Don't modify the existing object
|
|
c = c.DeepCopy()
|
|
c.ResourceVersion = ""
|
|
expectedMap[c.Name] = c
|
|
}
|
|
for _, c := range r.objectSyncs {
|
|
// We must clone the ObjectSync because of golang race check - it was
|
|
// written by the controller without any locks on it.
|
|
c = c.DeepCopy()
|
|
c.ResourceVersion = ""
|
|
gotMap[c.Name] = c
|
|
}
|
|
if !reflect.DeepEqual(expectedMap, gotMap) {
|
|
// Print ugly but useful diff of expected and received objects for
|
|
// easier debugging.
|
|
return fmt.Errorf("ObjectSync check failed [A-expected, B-got result]: %s", diff.ObjectDiff(expectedMap, gotMap))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
const (
|
|
TestNodeID = "foo-node"
|
|
TestProjectID = "foo-project"
|
|
TestNamespace = "foo-ns"
|
|
TestPodName = "foo-pod"
|
|
TestConfigMapName = "foo-config"
|
|
TestPodUID = "c7399134-f4fb-43e6-8088-c273bfffe7af"
|
|
TestDiffPodUID = "c7399134-f4fb-43e6-8088-c273bfffe999"
|
|
TestConfigMapUID = "22b0074a-8c07-4fc2-adad-1589f7f6f8b1"
|
|
KeepaliveInterval = 2 * time.Second
|
|
NormalSendKeepaliveInterval = 1 * time.Second
|
|
AbnormalSendKeepaliveInterval = 3 * time.Second
|
|
)
|
|
|
|
// NewObjectSyncReactor creates a ObjectSync reactor.
|
|
func NewObjectSyncReactor(client *fake.Clientset, errors []ReactorError) *ObjectSyncReactor {
|
|
reactor := &ObjectSyncReactor{
|
|
objectSyncs: make(map[string]*v1alpha1.ObjectSync),
|
|
errors: errors,
|
|
}
|
|
client.AddReactor("create", "objectsyncs", reactor.React)
|
|
client.AddReactor("update", "objectsyncs", reactor.React)
|
|
client.AddReactor("get", "objectsyncs", reactor.React)
|
|
client.AddReactor("delete", "objectsyncs", reactor.React)
|
|
return reactor
|
|
}
|
|
|
|
func NewTestPodResource(name, UID, resourceVersion string) *v1.Pod {
|
|
return &v1.Pod{
|
|
TypeMeta: v12.TypeMeta{
|
|
Kind: "Pod",
|
|
APIVersion: "v1",
|
|
},
|
|
ObjectMeta: v12.ObjectMeta{
|
|
Name: name,
|
|
Namespace: TestNamespace,
|
|
ResourceVersion: resourceVersion,
|
|
UID: types.UID(UID),
|
|
Labels: map[string]string{
|
|
"version": resourceVersion,
|
|
},
|
|
},
|
|
Spec: v1.PodSpec{
|
|
NodeName: TestNodeID,
|
|
Containers: []v1.Container{{Name: "ctr", Image: "image"}},
|
|
},
|
|
}
|
|
}
|
|
|
|
func NewObjectSync(object v12.Object, kind string) *v1alpha1.ObjectSync {
|
|
return &v1alpha1.ObjectSync{
|
|
ObjectMeta: v12.ObjectMeta{
|
|
Name: fmt.Sprintf("%s.%s", TestNodeID, object.GetUID()),
|
|
Namespace: object.GetNamespace(),
|
|
},
|
|
Spec: v1alpha1.ObjectSyncSpec{
|
|
ObjectAPIVersion: "v1",
|
|
ObjectKind: kind,
|
|
ObjectName: object.GetName(),
|
|
},
|
|
Status: v1alpha1.ObjectSyncStatus{
|
|
ObjectResourceVersion: object.GetResourceVersion(),
|
|
},
|
|
}
|
|
}
|
|
|
|
func NewPodMessage(pod *v1.Pod, operation string) *beehivemodel.Message {
|
|
resource, err := messagelayer.BuildResource(pod.Spec.NodeName, pod.Namespace, "pod", pod.Name)
|
|
if err != nil {
|
|
klog.Warningf("built message resource failed with error: %s", err)
|
|
return nil
|
|
}
|
|
message := beehivemodel.NewMessage("").
|
|
SetResourceVersion(pod.ResourceVersion).
|
|
FillBody(pod)
|
|
|
|
return message.BuildRouter("edgecontroller", "resource", resource, operation)
|
|
}
|
|
|
|
func NewTestConfigMapResource(name, UID, resourceVersion string) *v1.ConfigMap {
|
|
return &v1.ConfigMap{
|
|
TypeMeta: v12.TypeMeta{
|
|
Kind: "ConfigMap",
|
|
APIVersion: "v1",
|
|
},
|
|
ObjectMeta: v12.ObjectMeta{
|
|
Name: name,
|
|
Namespace: TestNamespace,
|
|
ResourceVersion: resourceVersion,
|
|
UID: types.UID(UID),
|
|
Labels: map[string]string{
|
|
"version": resourceVersion,
|
|
},
|
|
},
|
|
Data: map[string]string{"foo": "bar"},
|
|
}
|
|
}
|
|
|
|
func NewConfigMapMessage(configMap *v1.ConfigMap, operation string) *beehivemodel.Message {
|
|
resource, err := messagelayer.BuildResource(TestNodeID, configMap.Namespace, "configmap", configMap.Name)
|
|
if err != nil {
|
|
klog.Warningf("build message resource failed with error: %s", err)
|
|
return nil
|
|
}
|
|
return beehivemodel.NewMessage("").
|
|
SetResourceVersion(configMap.ResourceVersion).
|
|
BuildRouter(modules.EdgeControllerModuleName, "resource", resource, operation).
|
|
FillBody(configMap)
|
|
}
|