karmada-io
/
karmada
mirror of https://github.com/karmada-io/karmada.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #4433 from jwcesign/report-event 

feat: report event when the service doesn't exist/when endpointslice api is disabled
This commit is contained in:
karmada-bot 2023-12-20 09:25:16 +08:00 committed by GitHub
parent ba76508fd3 cfd88926dd
commit 515594592b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 151 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

145
pkg/controllers/multiclusterservice/endpointslice_dispatch_controller.go
View File

@ -52,10 +52,6 @@ import (
// EndpointsliceDispatchControllerName is the controller name that will be used when reporting events.
const EndpointsliceDispatchControllerName = "endpointslice-dispatch-controller"
var (
endpointSliceGVK = discoveryv1.SchemeGroupVersion.WithKind("EndpointSlice")
)
type EndpointsliceDispatchController struct {
client.Client
EventRecorder record.EventRecorder
@ -75,7 +71,7 @@ func (c *EndpointsliceDispatchController) Reconcile(ctx context.Context, req con
return controllerruntime.Result{Requeue: true}, err
}
if !helper.IsWorkContains(work.Spec.Workload.Manifests, endpointSliceGVK) {
if !helper.IsWorkContains(work.Spec.Workload.Manifests, util.EndpointSliceGVK) {
return controllerruntime.Result{}, nil
}
@ -191,7 +187,7 @@ func (c *EndpointsliceDispatchController) newClusterFunc() handler.MapFunc {
var requests []reconcile.Request
for _, mcs := range mcsList.Items {
clusterSet, _, err := helper.GetConsumerClustres(c.Client, mcs.DeepCopy())
clusterSet, err := helper.GetConsumerClusters(c.Client, mcs.DeepCopy())
if err != nil {
klog.Errorf("Failed to get provider clusters, error: %v", err)
continue
@ -283,7 +279,7 @@ func (c *EndpointsliceDispatchController) cleanOrphanDispatchedEndpointSlice(ctx
continue
}
consumerClusters, _, err := helper.GetConsumerClustres(c.Client, mcs)
consumerClusters, err := helper.GetConsumerClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get consumer clusters, error is: %v", err)
return err
@ -315,7 +311,7 @@ func (c *EndpointsliceDispatchController) dispatchEndpointSlice(ctx context.Cont
return err
}
consumerClusters, _, err := helper.GetConsumerClustres(c.Client, mcs)
consumerClusters, err := helper.GetConsumerClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get consumer clusters, error is: %v", err)
return err
@ -324,62 +320,26 @@ func (c *EndpointsliceDispatchController) dispatchEndpointSlice(ctx context.Cont
if clusterName == epsSourceCluster {
continue
}
// It couldn't happen here
if len(work.Spec.Workload.Manifests) == 0 {
clusterObj, err := util.GetCluster(c.Client, clusterName)
if err != nil {
if apierrors.IsNotFound(err) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonClusterNotFound, "Consumer cluster %s is not found", clusterName)
continue
}
klog.Errorf("Failed to get cluster %s, error is: %v", clusterName, err)
return err
}
if !util.IsClusterReady(&clusterObj.Status) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonSyncServiceFailed,
"Consumer cluster %s is not ready, skip to propagate EndpointSlice", clusterName)
continue
}
if !helper.IsAPIEnabled(clusterObj.Status.APIEnablements, util.EndpointSliceGVK.GroupVersion().String(), util.EndpointSliceGVK.Kind) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonAPIIncompatible, "Consumer cluster %s does not support EndpointSlice", clusterName)
continue
}
// There should be only one manifest in the work, let's use the first one.
manifest := work.Spec.Workload.Manifests[0]
unstructuredObj := &unstructured.Unstructured{}
if err := unstructuredObj.UnmarshalJSON(manifest.Raw); err != nil {
klog.Errorf("Failed to unmarshal work manifest, error is: %v", err)
return err
}
endpointSlice := &discoveryv1.EndpointSlice{}
if err := helper.ConvertToTypedObject(unstructuredObj, endpointSlice); err != nil {
klog.Errorf("Failed to convert unstructured object to typed object, error is: %v", err)
return err
}
// Use this name to avoid naming conflicts and locate the EPS source cluster.
endpointSlice.Name = epsSourceCluster + "-" + endpointSlice.Name
clusterNamespace := names.GenerateExecutionSpaceName(clusterName)
endpointSlice.Labels = map[string]string{
discoveryv1.LabelServiceName: mcs.Name,
workv1alpha1.WorkNamespaceLabel: clusterNamespace,
workv1alpha1.WorkNameLabel: work.Name,
util.ManagedByKarmadaLabel: util.ManagedByKarmadaLabelValue,
discoveryv1.LabelManagedBy: util.EndpointSliceDispatchControllerLabelValue,
}
endpointSlice.Annotations = map[string]string{
// This annotation is used to identify the source cluster of EndpointSlice and whether the eps are the newest version
util.EndpointSliceProvisionClusterAnnotation: epsSourceCluster,
}
workMeta := metav1.ObjectMeta{
Name: work.Name,
Namespace: clusterNamespace,
Finalizers: []string{util.ExecutionControllerFinalizer},
Annotations: map[string]string{
util.EndpointSliceProvisionClusterAnnotation: epsSourceCluster,
},
Labels: map[string]string{
util.ManagedByKarmadaLabel: util.ManagedByKarmadaLabelValue,
util.MultiClusterServiceNameLabel: mcs.Name,
util.MultiClusterServiceNamespaceLabel: mcs.Namespace,
},
}
unstructuredEPS, err := helper.ToUnstructured(endpointSlice)
if err != nil {
klog.Errorf("Failed to convert typed object to unstructured object, error is: %v", err)
return err
}
if err := helper.CreateOrUpdateWork(c.Client, workMeta, unstructuredEPS); err != nil {
klog.Errorf("Failed to dispatch EndpointSlice %s/%s from %s to cluster %s:%v",
work.GetNamespace(), work.GetName(), epsSourceCluster, clusterName, err)
if err := c.ensureEndpointSliceWork(mcs, work, epsSourceCluster, clusterName); err != nil {
return err
}
}
@ -394,6 +354,69 @@ func (c *EndpointsliceDispatchController) dispatchEndpointSlice(ctx context.Cont
return nil
}
func (c *EndpointsliceDispatchController) ensureEndpointSliceWork(mcs *networkingv1alpha1.MultiClusterService,
work *workv1alpha1.Work, providerCluster, consumerCluster string) error {
// It couldn't happen here
if len(work.Spec.Workload.Manifests) == 0 {
return nil
}
// There should be only one manifest in the work, let's use the first one.
manifest := work.Spec.Workload.Manifests[0]
unstructuredObj := &unstructured.Unstructured{}
if err := unstructuredObj.UnmarshalJSON(manifest.Raw); err != nil {
klog.Errorf("Failed to unmarshal work manifest, error is: %v", err)
return err
}
endpointSlice := &discoveryv1.EndpointSlice{}
if err := helper.ConvertToTypedObject(unstructuredObj, endpointSlice); err != nil {
klog.Errorf("Failed to convert unstructured object to typed object, error is: %v", err)
return err
}
// Use this name to avoid naming conflicts and locate the EPS source cluster.
endpointSlice.Name = providerCluster + "-" + endpointSlice.Name
clusterNamespace := names.GenerateExecutionSpaceName(consumerCluster)
endpointSlice.Labels = map[string]string{
discoveryv1.LabelServiceName: mcs.Name,
workv1alpha1.WorkNamespaceLabel: clusterNamespace,
workv1alpha1.WorkNameLabel: work.Name,
util.ManagedByKarmadaLabel: util.ManagedByKarmadaLabelValue,
discoveryv1.LabelManagedBy: util.EndpointSliceDispatchControllerLabelValue,
}
endpointSlice.Annotations = map[string]string{
// This annotation is used to identify the source cluster of EndpointSlice and whether the eps are the newest version
util.EndpointSliceProvisionClusterAnnotation: providerCluster,
}
workMeta := metav1.ObjectMeta{
Name: work.Name,
Namespace: clusterNamespace,
Finalizers: []string{util.ExecutionControllerFinalizer},
Annotations: map[string]string{
util.EndpointSliceProvisionClusterAnnotation: providerCluster,
},
Labels: map[string]string{
util.ManagedByKarmadaLabel: util.ManagedByKarmadaLabelValue,
util.MultiClusterServiceNameLabel: mcs.Name,
util.MultiClusterServiceNamespaceLabel: mcs.Namespace,
},
}
unstructuredEPS, err := helper.ToUnstructured(endpointSlice)
if err != nil {
klog.Errorf("Failed to convert typed object to unstructured object, error is: %v", err)
return err
}
if err := helper.CreateOrUpdateWork(c.Client, workMeta, unstructuredEPS); err != nil {
klog.Errorf("Failed to dispatch EndpointSlice %s/%s from %s to cluster %s:%v",
work.GetNamespace(), work.GetName(), providerCluster, consumerCluster, err)
return err
}
return nil
}
func (c *EndpointsliceDispatchController) cleanupEndpointSliceFromConsumerClusters(ctx context.Context, work *workv1alpha1.Work) error {
// TBD: There may be a better way without listing all works.
workList := &workv1alpha1.WorkList{}

68
pkg/controllers/multiclusterservice/mcs_controller.go
View File

@ -19,7 +19,6 @@ package multiclusterservice
import (
"context"
"fmt"
"strings"
"github.com/google/uuid"
corev1 "k8s.io/api/core/v1"
@ -86,23 +85,7 @@ func (c *MCSController) Reconcile(ctx context.Context, req controllerruntime.Req
return c.handleMultiClusterServiceDelete(mcs.DeepCopy())
}
providerClusters, noneExistProviderClusters, err := helper.GetProviderClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get provider clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return controllerruntime.Result{}, err
}
consumerClusters, noneExistConsumerClusters, err := helper.GetConsumerClustres(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get consumer clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return controllerruntime.Result{}, err
}
if len(noneExistProviderClusters) != 0 || len(noneExistConsumerClusters) != 0 {
msgProvider := strings.Join(noneExistProviderClusters.UnsortedList(), ",")
msgConsumer := strings.Join(noneExistConsumerClusters.UnsortedList(), ",")
c.EventRecorder.Eventf(mcs, corev1.EventTypeNormal, events.EventReasonConfigurationRedundant,
fmt.Sprintf("ProviderClusters(%s)/ConsumerClusters(%s) dont's exist, Karmada will ignore them", msgProvider, msgConsumer))
}
var err error
defer func() {
if err != nil {
_ = c.updateMultiClusterServiceStatus(mcs, metav1.ConditionFalse, "ServiceAppliedFailed", err.Error())
@ -113,7 +96,7 @@ func (c *MCSController) Reconcile(ctx context.Context, req controllerruntime.Req
c.EventRecorder.Eventf(mcs, corev1.EventTypeNormal, events.EventReasonSyncServiceSucceed, "Service is propagated to target clusters.")
}()
if err = c.handleMultiClusterServiceCreateOrUpdate(mcs.DeepCopy(), providerClusters, consumerClusters); err != nil {
if err = c.handleMultiClusterServiceCreateOrUpdate(mcs.DeepCopy()); err != nil {
return controllerruntime.Result{}, err
}
return controllerruntime.Result{}, nil
@ -201,7 +184,7 @@ func (c *MCSController) cleanProviderEndpointSliceWork(work *workv1alpha1.Work)
var errs []error
for _, work := range workList.Items {
if !helper.IsWorkContains(work.Spec.Workload.Manifests, endpointSliceGVK) {
if !helper.IsWorkContains(work.Spec.Workload.Manifests, util.EndpointSliceGVK) {
continue
}
// This annotation is only added to the EndpointSlice work in consumer clusters' execution namespace
@ -230,9 +213,20 @@ func (c *MCSController) cleanProviderEndpointSliceWork(work *workv1alpha1.Work)
return nil
}
func (c *MCSController) handleMultiClusterServiceCreateOrUpdate(mcs *networkingv1alpha1.MultiClusterService, providerClusters, consumerClusters sets.Set[string]) error {
func (c *MCSController) handleMultiClusterServiceCreateOrUpdate(mcs *networkingv1alpha1.MultiClusterService) error {
klog.V(4).Infof("Begin to handle MultiClusterService(%s/%s) create or update event", mcs.Namespace, mcs.Name)
providerClusters, err := helper.GetProviderClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get provider clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return err
}
consumerClusters, err := helper.GetConsumerClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get consumer clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return err
}
// 1. if mcs not contain CrossCluster type, delete service work if needed
if !helper.MultiClusterServiceCrossClusterEnabled(mcs) {
if err := c.retrieveService(mcs); err != nil {
@ -274,18 +268,16 @@ func (c *MCSController) handleMultiClusterServiceCreateOrUpdate(mcs *networkingv
// 5. make sure service exist
svc := &corev1.Service{}
err := c.Client.Get(context.Background(), types.NamespacedName{Namespace: mcs.Namespace, Name: mcs.Name}, svc)
err = c.Client.Get(context.Background(), types.NamespacedName{Namespace: mcs.Namespace, Name: mcs.Name}, svc)
// If the Serivice are deleted, the Service's ResourceBinding will be cleaned by GC
if err != nil && !apierrors.IsNotFound(err) {
if err != nil {
klog.Errorf("Failed to get service(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return err
}
// 6. if service exists, create or update corresponding ResourceBinding
if err == nil {
if err := c.propagateService(context.Background(), mcs, svc, providerClusters, consumerClusters); err != nil {
return err
}
if err := c.propagateService(context.Background(), mcs, svc, providerClusters, consumerClusters); err != nil {
return err
}
klog.V(4).Infof("Success to reconcile MultiClusterService(%s/%s)", mcs.Namespace, mcs.Name)
@ -294,9 +286,25 @@ func (c *MCSController) handleMultiClusterServiceCreateOrUpdate(mcs *networkingv
func (c *MCSController) propagateMultiClusterService(mcs *networkingv1alpha1.MultiClusterService, providerClusters sets.Set[string]) error {
for clusterName := range providerClusters {
if !c.IsClusterReady(clusterName) {
clusterObj, err := util.GetCluster(c.Client, clusterName)
if err != nil {
if apierrors.IsNotFound(err) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonClusterNotFound, "Provider cluster %s is not found", clusterName)
continue
}
klog.Errorf("Failed to get cluster %s, error is: %v", clusterName, err)
return err
}
if !util.IsClusterReady(&clusterObj.Status) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonSyncServiceFailed,
"Provider cluster %s is not ready, skip to propagate MultiClusterService", clusterName)
continue
}
if !helper.IsAPIEnabled(clusterObj.Status.APIEnablements, util.EndpointSliceGVK.GroupVersion().String(), util.EndpointSliceGVK.Kind) {
c.EventRecorder.Eventf(mcs, corev1.EventTypeWarning, events.EventReasonAPIIncompatible, "Provider cluster %s does not support EndpointSlice", clusterName)
continue
}
workMeta := metav1.ObjectMeta{
Name: names.GenerateWorkName(mcs.Kind, mcs.Name, mcs.Namespace),
Namespace: names.GenerateExecutionSpaceName(clusterName),
@ -648,7 +656,7 @@ func (c *MCSController) needSyncMultiClusterService(mcs *networkingv1alpha1.Mult
return true, nil
}
providerClusters, _, err := helper.GetProviderClusters(c.Client, mcs)
providerClusters, err := helper.GetProviderClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get provider clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return false, err
@ -657,7 +665,7 @@ func (c *MCSController) needSyncMultiClusterService(mcs *networkingv1alpha1.Mult
return true, nil
}
consumerClusters, _, err := helper.GetConsumerClustres(c.Client, mcs)
consumerClusters, err := helper.GetConsumerClusters(c.Client, mcs)
if err != nil {
klog.Errorf("Failed to get consumer clusters by MultiClusterService(%s/%s):%v", mcs.Namespace, mcs.Name, err)
return false, err

6
pkg/events/events.go
View File

@ -140,6 +140,8 @@ const (
EventReasonDispatchEndpointSliceFailed = "DispatchEndpointSliceFailed"
// EventReasonDispatchEndpointSliceSucceed indicates that dispatch endpointslice succeed.
EventReasonDispatchEndpointSliceSucceed = "DispatchEndpointSliceSucceed"
// EventReasonConfigurationRedundant indicates that MultiClusterService configuration redundant.
EventReasonConfigurationRedundant = "ConfigurationRedundant"
// EventReasonClusterNotFound indicates that the cluster configured in MultiClusterService does not exist.
EventReasonClusterNotFound = "ClusterNotFound"
// EventReasonAPIIncompatible indicates that the MultiClusterService may not function properly as some member clusters do not support EndpointSlice.
EventReasonAPIIncompatible = "APIIncompatible"
)

11
pkg/util/constants.go
View File

@ -16,7 +16,11 @@ limitations under the License.
package util
import "time"
import (
"time"
discoveryv1 "k8s.io/api/discovery/v1"
)
// Define labels used by karmada system.
const (
@ -211,3 +215,8 @@ const (
// CacheSyncTimeout refers to the time limit set on waiting for cache to sync
CacheSyncTimeout = 2 * time.Minute
)
var (
// EndpointSliceGVK is the GroupVersionKind of K8s native EndpointSlice.
EndpointSliceGVK = discoveryv1.SchemeGroupVersion.WithKind("EndpointSlice")
)

27
pkg/util/helper/mcs.go
View File

@ -96,6 +96,7 @@ func DeleteEndpointSlice(c client.Client, selector labels.Set) error {
return errors.NewAggregate(errs)
}
// MultiClusterServiceCrossClusterEnabled will check if it's a CrossCluster MultiClusterService.
func MultiClusterServiceCrossClusterEnabled(mcs *networkingv1alpha1.MultiClusterService) bool {
for _, svcType := range mcs.Spec.Types {
if svcType == networkingv1alpha1.ExposureTypeCrossCluster {
@ -106,34 +107,36 @@ func MultiClusterServiceCrossClusterEnabled(mcs *networkingv1alpha1.MultiCluster
return false
}
func GetProviderClusters(client client.Client, mcs *networkingv1alpha1.MultiClusterService) (existClusters, noneExistClusters sets.Set[string], err error) {
// GetProviderClusters will extract the target provider clusters of the service
func GetProviderClusters(client client.Client, mcs *networkingv1alpha1.MultiClusterService) (sets.Set[string], error) {
providerClusters := sets.New[string]()
for _, p := range mcs.Spec.ProviderClusters {
providerClusters.Insert(p.Name)
}
if len(providerClusters) != 0 {
return providerClusters, nil
}
allClusters, err := util.GetClusterSet(client)
if err != nil {
klog.Errorf("Failed to get cluster set, Error: %v", err)
return nil, nil, err
return nil, err
}
if len(providerClusters) == 0 {
return allClusters, nil, nil
}
return providerClusters.Clone().Intersection(allClusters), providerClusters.Clone().Delete(allClusters.UnsortedList()...), nil
return allClusters, nil
}
func GetConsumerClustres(client client.Client, mcs *networkingv1alpha1.MultiClusterService) (existClusters, noneExistClusters sets.Set[string], err error) {
// GetProviderClusters will extract the target consumer clusters of the service
func GetConsumerClusters(client client.Client, mcs *networkingv1alpha1.MultiClusterService) (sets.Set[string], error) {
consumerClusters := sets.New[string]()
for _, c := range mcs.Spec.ConsumerClusters {
consumerClusters.Insert(c.Name)
}
if len(consumerClusters) != 0 {
return consumerClusters, nil
}
allClusters, err := util.GetClusterSet(client)
if err != nil {
klog.Errorf("Failed to get cluster set, Error: %v", err)
return nil, nil, err
return nil, err
}
if len(consumerClusters) == 0 {
return allClusters, nil, nil
}
return consumerClusters.Clone().Intersection(allClusters), consumerClusters.Clone().Delete(allClusters.UnsortedList()...), nil
return allClusters, nil
}