Skip to content

Commit

Permalink
Bump admiral and adjust to Federator API changes
Browse files Browse the repository at this point in the history 
This also required further propagating ctx params to satisfy the
contextcheck linter.

Signed-off-by: Tom Pantelis <[email protected]>
  • Loading branch information
@tpantelis
tpantelis committed Oct 30, 2023
1 parent 630d976 commit 2010640
Show file tree
Hide file tree
Showing 12 changed files with 106 additions and 89 deletions.
2 changes: 1 addition & 1 deletion go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ require (
github.com/onsi/gomega v1.27.10
github.com/pkg/errors v0.9.1
github.com/prometheus/client_golang v1.16.0
github.com/submariner-io/admiral v0.16.1-0.20231025063702-858d0984799c
github.com/submariner-io/admiral v0.16.1-0.20231030143920-15adb86d8eca
github.com/submariner-io/shipyard v0.16.0
github.com/uw-labs/lichen v0.1.7
k8s.io/api v0.27.6
Expand Down
4 changes: 2 additions & 2 deletions go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -408,8 +408,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
github.com/submariner-io/admiral v0.16.1-0.20231025063702-858d0984799c h1:zy5mZZrB885JAuLPqpb/RoGhtd9N9tUCFE5OGAZEzWw=
github.com/submariner-io/admiral v0.16.1-0.20231025063702-858d0984799c/go.mod h1:GP0TCJkt444r2ONKVHKBbSPaKjJb0S5Qj0MyNUl2keQ=
github.com/submariner-io/admiral v0.16.1-0.20231030143920-15adb86d8eca h1:O3gUAdYldm4LQJD/IpqaFcOlJVf9ikM+uBWsfVADmjg=
github.com/submariner-io/admiral v0.16.1-0.20231030143920-15adb86d8eca/go.mod h1:GP0TCJkt444r2ONKVHKBbSPaKjJb0S5Qj0MyNUl2keQ=
github.com/submariner-io/shipyard v0.16.0 h1:PTvp2aKNBoCkfC8nS38k+DW5ZaXNMq/wzzjGOvsiAQM=
github.com/submariner-io/shipyard v0.16.0/go.mod h1:aKCotVktXJO3azjBOmhu/0KbRcYLY3eUcSNSDDJNbxs=
github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
Expand Down
31 changes: 19 additions & 12 deletions pkg/agent/controller/agent.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ limitations under the License.
package controller

import (
"context"
"encoding/json"
"fmt"
"reflect"
Expand Down Expand Up @@ -179,6 +180,8 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
func (a *Controller) serviceExportToServiceImport(obj runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
svcExport := obj.(*mcsv1a1.ServiceExport)

ctx := context.Background()

logger.V(log.DEBUG).Infof("ServiceExport %s/%s %sd", svcExport.Namespace, svcExport.Name, op)

if op == syncer.Delete {
Expand All @@ -188,17 +191,19 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, _ int, op
obj, found, err := a.serviceSyncer.GetResource(svcExport.Name, svcExport.Namespace)
if err != nil {
// some other error. Log and requeue
a.serviceExportClient.updateStatusConditions(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
corev1.ConditionUnknown, "ServiceRetrievalFailed", fmt.Sprintf("Error retrieving the Service: %v", err)))
a.serviceExportClient.updateStatusConditions(ctx, svcExport.Name, svcExport.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionUnknown, "ServiceRetrievalFailed",
fmt.Sprintf("Error retrieving the Service: %v", err)))
logger.Errorf(err, "Error retrieving Service %s/%s", svcExport.Namespace, svcExport.Name)

return nil, true
}

if !found {
logger.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't exist", svcExport.Namespace, svcExport.Name)
a.serviceExportClient.updateStatusConditions(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist"))
a.serviceExportClient.updateStatusConditions(ctx, svcExport.Name, svcExport.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionFalse, serviceUnavailable,
"Service to be exported doesn't exist"))

return nil, false
}
Expand All @@ -208,11 +213,12 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, _ int, op
svcType, ok := getServiceImportType(svc)

if !ok {
a.serviceExportClient.updateStatusConditions(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
corev1.ConditionFalse, invalidServiceType, fmt.Sprintf("Service of type %v not supported", svc.Spec.Type)))
a.serviceExportClient.updateStatusConditions(ctx, svcExport.Name, svcExport.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionFalse, invalidServiceType,
fmt.Sprintf("Service of type %v not supported", svc.Spec.Type)))
logger.Errorf(nil, "Service type %q not supported for Service (%s/%s)", svc.Spec.Type, svcExport.Namespace, svcExport.Name)

err = a.localServiceImportFederator.Delete(a.newServiceImport(svcExport.Name, svcExport.Namespace))
err = a.localServiceImportFederator.Delete(ctx, a.newServiceImport(svcExport.Name, svcExport.Namespace))
if err == nil || apierrors.IsNotFound(err) {
return nil, false
}
Expand Down Expand Up @@ -246,7 +252,7 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, _ int, op
logger.V(log.DEBUG).Infof("Service to be exported (%s/%s) doesn't have a global IP yet",
svcExport.Namespace, svcExport.Name)
// Globalnet enabled but service doesn't have globalIp yet, Update the status and requeue
a.serviceExportClient.updateStatusConditions(svcExport.Name, svcExport.Namespace,
a.serviceExportClient.updateStatusConditions(ctx, svcExport.Name, svcExport.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionFalse, reason, msg))

return nil, true
Expand All @@ -260,8 +266,8 @@ func (a *Controller) serviceExportToServiceImport(obj runtime.Object, _ int, op
serviceImport.Spec.Ports = a.getPortsForService(svc)
}

a.serviceExportClient.updateStatusConditions(svcExport.Name, svcExport.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
corev1.ConditionTrue, "", ""))
a.serviceExportClient.updateStatusConditions(ctx, svcExport.Name, svcExport.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionTrue, "", ""))

logger.V(log.DEBUG).Infof("Returning ServiceImport %s/%s: %s", svcExport.Namespace, svcExport.Name,
serviceImportStringer{serviceImport})
Expand Down Expand Up @@ -329,8 +335,9 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, _ int, op
serviceImport := a.newServiceImport(svc.Name, svc.Namespace)

// Update the status and requeue
a.serviceExportClient.updateStatusConditions(svc.Name, svc.Namespace, newServiceExportCondition(mcsv1a1.ServiceExportValid,
corev1.ConditionFalse, serviceUnavailable, "Service to be exported doesn't exist"))
a.serviceExportClient.updateStatusConditions(context.Background(), svc.Name, svc.Namespace,
newServiceExportCondition(mcsv1a1.ServiceExportValid, corev1.ConditionFalse, serviceUnavailable,
"Service to be exported doesn't exist"))

return serviceImport, false
}
Expand Down
22 changes: 11 additions & 11 deletions pkg/agent/controller/cleanup.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,12 +48,12 @@ var (
}
)

func (a *Controller) Cleanup() error {
func (a *Controller) Cleanup(ctx context.Context) error {
// Delete all ServiceImports from the local cluster skipping those in the broker namespace if the broker is on the
// local cluster.
siClient := a.serviceImportController.localClient.Resource(serviceImportGVR)

list, err := listResources(siClient, metav1.NamespaceAll,
list, err := listResources(ctx, siClient, metav1.NamespaceAll,
&metav1.ListOptions{
FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace",
a.serviceImportController.serviceImportAggregator.brokerNamespace).String(),
Expand All @@ -65,13 +65,13 @@ func (a *Controller) Cleanup() error {
for i := range list {
_, ok := list[i].GetLabels()[mcsv1a1.LabelServiceName]
if ok {
err = a.serviceImportController.Delete(&list[i])
err = a.serviceImportController.Delete(ctx, &list[i])
if err != nil && !apierrors.IsNotFound(err) {
return err
}
}

err = siClient.Namespace(list[i].GetNamespace()).Delete(context.TODO(), list[i].GetName(), metav1.DeleteOptions{})
err = siClient.Namespace(list[i].GetNamespace()).Delete(ctx, list[i].GetName(), metav1.DeleteOptions{})

if err != nil && !apierrors.IsNotFound(err) {
return err //nolint:wrapcheck // Let the caller wrap
Expand All @@ -80,7 +80,7 @@ func (a *Controller) Cleanup() error {

// Delete all EndpointSlices from the local cluster skipping those in the broker namespace if the broker is on the
// local cluster.
err = deleteResources(a.endpointSliceController.syncer.GetLocalClient().Resource(endpointSliceGVR), metav1.NamespaceAll,
err = deleteResources(ctx, a.endpointSliceController.syncer.GetLocalClient().Resource(endpointSliceGVR), metav1.NamespaceAll,
&metav1.ListOptions{
FieldSelector: fields.OneTermNotEqualSelector("metadata.namespace",
a.serviceImportController.serviceImportAggregator.brokerNamespace).String(),
Expand All @@ -91,7 +91,7 @@ func (a *Controller) Cleanup() error {
}

// Delete all local EndpointSlices from the broker.
err = deleteResources(a.endpointSliceController.syncer.GetBrokerClient().Resource(endpointSliceGVR),
err = deleteResources(ctx, a.endpointSliceController.syncer.GetBrokerClient().Resource(endpointSliceGVR),
a.endpointSliceController.syncer.GetBrokerNamespace(),
&metav1.ListOptions{
LabelSelector: labels.Set(map[string]string{constants.MCSLabelSourceCluster: a.clusterID}).String(),
Expand All @@ -100,14 +100,14 @@ func (a *Controller) Cleanup() error {
return errors.Wrap(err, "error deleting remote EndpointSlices")
}

func deleteResources(client dynamic.NamespaceableResourceInterface, ns string, options *metav1.ListOptions) error {
list, err := listResources(client, ns, options)
func deleteResources(ctx context.Context, client dynamic.NamespaceableResourceInterface, ns string, options *metav1.ListOptions) error {
list, err := listResources(ctx, client, ns, options)
if err != nil {
return err
}

for i := range list {
err = client.Namespace(list[i].GetNamespace()).Delete(context.TODO(), list[i].GetName(), metav1.DeleteOptions{})
err = client.Namespace(list[i].GetNamespace()).Delete(ctx, list[i].GetName(), metav1.DeleteOptions{})
if err != nil && !apierrors.IsNotFound(err) {
return err //nolint:wrapcheck // Let the caller wrap
}
Expand All @@ -116,10 +116,10 @@ func deleteResources(client dynamic.NamespaceableResourceInterface, ns string, o
return nil
}

func listResources(client dynamic.NamespaceableResourceInterface, ns string,
func listResources(ctx context.Context, client dynamic.NamespaceableResourceInterface, ns string,
options *metav1.ListOptions,
) ([]unstructured.Unstructured, error) {
list, err := client.Namespace(ns).List(context.TODO(), *options)
list, err := client.Namespace(ns).List(ctx, *options)
if err != nil && !apierrors.IsNotFound(err) {
return nil, err //nolint:wrapcheck // Let the caller wrap
}
Expand Down
2 changes: 1 addition & 1 deletion pkg/agent/controller/cleanup_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -228,7 +228,7 @@ var _ = Describe("Cleanup", func() {
})

It("should correctly remove local and remote ServiceImports and EndpointSlices", func() {
Expect(t.cluster1.agentController.Cleanup()).To(Succeed())
Expect(t.cluster1.agentController.Cleanup(context.Background())).To(Succeed())

test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(test.LocalNamespace), localServiceImport1.Name)
test.AwaitNoResource(t.cluster1.localServiceImportClient.Namespace(serviceNamespace), localAggregatedServiceImport1.Name)
Expand Down
32 changes: 19 additions & 13 deletions pkg/agent/controller/endpoint_slice.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ func newEndpointSliceController(spec *AgentSpecification, syncerConfig broker.Sy
BrokerResourceType: &discovery.EndpointSlice{},
TransformBrokerToLocal: c.onRemoteEndpointSlice,
OnSuccessfulSyncFromBroker: func(obj runtime.Object, op syncer.Operation) bool {
c.enqueueForConflictCheck(obj.(*discovery.EndpointSlice), op)
c.enqueueForConflictCheck(context.TODO(), obj.(*discovery.EndpointSlice), op)
return false
},
BrokerResyncPeriod: BrokerResyncPeriod,
Expand Down Expand Up @@ -102,12 +102,13 @@ func (c *EndpointSliceController) start(stopCh <-chan struct{}) error {

func (c *EndpointSliceController) onLocalEndpointSlice(obj runtime.Object, _ int, op syncer.Operation) (runtime.Object, bool) {
endpointSlice := obj.(*discovery.EndpointSlice)
ctx := context.TODO()

if op != syncer.Delete && isLegacyEndpointSlice(endpointSlice) {
logger.Infof("Found legacy EndpointSlice %s/%s - deleting it",
endpointSlice.Namespace, endpointSlice.Name)

err := c.syncer.GetLocalFederator().Delete(endpointSlice)
err := c.syncer.GetLocalFederator().Delete(ctx, endpointSlice)
if err != nil {
logger.Errorf(err, "Error deleting legacy EndpointSlice %s/%s", endpointSlice.Namespace, endpointSlice.Name)
}
Expand Down Expand Up @@ -135,6 +136,7 @@ func (c *EndpointSliceController) onRemoteEndpointSlice(obj runtime.Object, _ in

func (c *EndpointSliceController) onLocalEndpointSliceSynced(obj runtime.Object, op syncer.Operation) bool {
endpointSlice := obj.(*discovery.EndpointSlice)
ctx := context.TODO()

serviceName := endpointSlice.Labels[mcsv1a1.LabelServiceName]
serviceNamespace := endpointSlice.Labels[constants.LabelSourceNamespace]
Expand All @@ -153,18 +155,20 @@ func (c *EndpointSliceController) onLocalEndpointSliceSynced(obj runtime.Object,

if op == syncer.Delete {
if c.hasNoRemainingEndpointSlices(endpointSlice) {
err = c.serviceImportAggregator.updateOnDelete(serviceName, serviceNamespace)
err = c.serviceImportAggregator.updateOnDelete(ctx, serviceName, serviceNamespace)
}
} else {
err = c.serviceImportAggregator.updateOnCreateOrUpdate(serviceName, serviceNamespace)
err = c.serviceImportAggregator.updateOnCreateOrUpdate(ctx, serviceName, serviceNamespace)
if err != nil {
c.serviceExportClient.updateStatusConditions(serviceName, serviceNamespace, newServiceExportCondition(constants.ServiceExportReady,
corev1.ConditionFalse, exportFailedReason, fmt.Sprintf("Unable to export: %v", err)))
c.serviceExportClient.updateStatusConditions(ctx, serviceName, serviceNamespace,
newServiceExportCondition(constants.ServiceExportReady, corev1.ConditionFalse, exportFailedReason,
fmt.Sprintf("Unable to export: %v", err)))
} else {
c.serviceExportClient.updateStatusConditions(serviceName, serviceNamespace, newServiceExportCondition(constants.ServiceExportReady,
corev1.ConditionTrue, "", "Service was successfully exported to the broker"))
c.serviceExportClient.updateStatusConditions(ctx, serviceName, serviceNamespace,
newServiceExportCondition(constants.ServiceExportReady, corev1.ConditionTrue, "",
"Service was successfully exported to the broker"))

c.enqueueForConflictCheck(endpointSlice, op)
c.enqueueForConflictCheck(ctx, endpointSlice, op)
}
}

Expand Down Expand Up @@ -201,6 +205,8 @@ func (c *EndpointSliceController) hasNoRemainingEndpointSlices(endpointSlice *di
}

func (c *EndpointSliceController) checkForConflicts(_, name, namespace string) (bool, error) {
ctx := context.TODO()

localServiceExport := c.serviceExportClient.getLocalInstance(name, namespace)
if localServiceExport == nil {
return false, nil
Expand Down Expand Up @@ -230,26 +236,26 @@ func (c *EndpointSliceController) checkForConflicts(_, name, namespace string) (
}

if conflict {
c.serviceExportClient.updateStatusConditions(name, namespace, newServiceExportCondition(
c.serviceExportClient.updateStatusConditions(ctx, name, namespace, newServiceExportCondition(
mcsv1a1.ServiceExportConflict, corev1.ConditionTrue, portConflictReason,
fmt.Sprintf("The service ports conflict between the constituent clusters %s. "+
"The service will expose the intersection of all the ports: %s",
fmt.Sprintf("[%s]", strings.Join(clusterNames, ", ")), servicePortsToString(intersectedServicePorts))))
} else if FindServiceExportStatusCondition(localServiceExport.Status.Conditions, mcsv1a1.ServiceExportConflict) != nil {
c.serviceExportClient.removeStatusCondition(name, namespace, mcsv1a1.ServiceExportConflict, portConflictReason)
c.serviceExportClient.removeStatusCondition(ctx, name, namespace, mcsv1a1.ServiceExportConflict, portConflictReason)
}

return false, nil
}

func (c *EndpointSliceController) enqueueForConflictCheck(eps *discovery.EndpointSlice, op syncer.Operation) {
func (c *EndpointSliceController) enqueueForConflictCheck(ctx context.Context, eps *discovery.EndpointSlice, op syncer.Operation) {
if eps.Labels[constants.LabelIsHeadless] != "false" {
return
}

// Since the conflict checking works off of the local cache for efficiency, wait a little bit here for the local cache to be updated
// with the latest state of the EndpointSlice.
_ = wait.PollUntilContextTimeout(context.Background(), 10*time.Millisecond, 100*time.Millisecond, true,
_ = wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, 100*time.Millisecond, true,
func(_ context.Context) (bool, error) {
_, found, _ := c.syncer.GetLocalResource(eps.Name, eps.Namespace, eps)
return (op == syncer.Delete && !found) || (op != syncer.Delete && found), nil
Expand Down
18 changes: 9 additions & 9 deletions pkg/agent/controller/service_endpoint_slices.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ func (c *ServiceEndpointSliceController) stop() {
})
}

func (c *ServiceEndpointSliceController) cleanup() (bool, error) {
func (c *ServiceEndpointSliceController) cleanup(ctx context.Context) (bool, error) {
listOptions := metav1.ListOptions{
LabelSelector: k8slabels.SelectorFromSet(map[string]string{
discovery.LabelManagedBy: constants.LabelValueManagedBy,
Expand All @@ -109,7 +109,7 @@ func (c *ServiceEndpointSliceController) cleanup() (bool, error) {
}).String(),
}

list, err := c.localClient.List(context.Background(), listOptions)
list, err := c.localClient.List(ctx, listOptions)
if err != nil {
return false, errors.Wrapf(err, "error listing the EndpointSlices associated with service %s/%s",
c.serviceNamespace, c.serviceName)
Expand All @@ -119,7 +119,7 @@ func (c *ServiceEndpointSliceController) cleanup() (bool, error) {
return false, nil
}

err = c.localClient.DeleteCollection(context.Background(), metav1.DeleteOptions{}, listOptions)
err = c.localClient.DeleteCollection(ctx, metav1.DeleteOptions{}, listOptions)

if err != nil && !apierrors.IsNotFound(err) {
return false, errors.Wrapf(err, "error deleting the EndpointSlices associated with service %s/%s",
Expand Down Expand Up @@ -155,7 +155,7 @@ func (c *ServiceEndpointSliceController) onServiceEndpointSlice(obj runtime.Obje
}

if op == syncer.Delete {
list, err := c.localClient.List(context.Background(), metav1.ListOptions{
list, err := c.localClient.List(context.TODO(), metav1.ListOptions{
LabelSelector: k8slabels.SelectorFromSet(returnEPS.Labels).String(),
})
if err != nil {
Expand Down Expand Up @@ -320,18 +320,18 @@ func (c *ServiceEndpointSliceController) isHeadless() bool {
return c.serviceImportSpec.Type == mcsv1a1.Headless
}

func (c *ServiceEndpointSliceController) Distribute(obj runtime.Object) error {
return c.federator.Distribute(obj) //nolint:wrapcheck // No need to wrap here
func (c *ServiceEndpointSliceController) Distribute(ctx context.Context, obj runtime.Object) error {
return c.federator.Distribute(ctx, obj) //nolint:wrapcheck // No need to wrap here
}

func (c *ServiceEndpointSliceController) Delete(obj runtime.Object) error {
func (c *ServiceEndpointSliceController) Delete(ctx context.Context, obj runtime.Object) error {
if c.isHeadless() {
return c.federator.Delete(obj) //nolint:wrapcheck // No need to wrap here
return c.federator.Delete(ctx, obj) //nolint:wrapcheck // No need to wrap here
}

// For a non-headless service, we never delete the single exported EPS - we update its endpoint condition based on
// the backend service EPS's as they are created/updated/deleted.
return c.Distribute(obj)
return c.Distribute(ctx, obj)
}

type endpointSliceStringer struct {
Expand Down
Loading

0 comments on commit 2010640

Please sign in to comment.