mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-27 21:47:07 +00:00
aggregator: split availability controller into local and remote part
Signed-off-by: Dr. Stefan Schimanski <stefan.schimanski@gmail.com>
This commit is contained in:
parent
bbdc247406
commit
834cd7ca4a
@ -51,8 +51,9 @@ import (
|
|||||||
openapiaggregator "k8s.io/kube-aggregator/pkg/controllers/openapi/aggregator"
|
openapiaggregator "k8s.io/kube-aggregator/pkg/controllers/openapi/aggregator"
|
||||||
openapiv3controller "k8s.io/kube-aggregator/pkg/controllers/openapiv3"
|
openapiv3controller "k8s.io/kube-aggregator/pkg/controllers/openapiv3"
|
||||||
openapiv3aggregator "k8s.io/kube-aggregator/pkg/controllers/openapiv3/aggregator"
|
openapiv3aggregator "k8s.io/kube-aggregator/pkg/controllers/openapiv3/aggregator"
|
||||||
|
localavailability "k8s.io/kube-aggregator/pkg/controllers/status/local"
|
||||||
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
statuscontrollers "k8s.io/kube-aggregator/pkg/controllers/status/remote"
|
remoteavailability "k8s.io/kube-aggregator/pkg/controllers/status/remote"
|
||||||
apiservicerest "k8s.io/kube-aggregator/pkg/registry/apiservice/rest"
|
apiservicerest "k8s.io/kube-aggregator/pkg/registry/apiservice/rest"
|
||||||
openapicommon "k8s.io/kube-openapi/pkg/common"
|
openapicommon "k8s.io/kube-openapi/pkg/common"
|
||||||
)
|
)
|
||||||
@ -102,12 +103,11 @@ type ExtraConfig struct {
|
|||||||
|
|
||||||
RejectForwardingRedirects bool
|
RejectForwardingRedirects bool
|
||||||
|
|
||||||
// DisableAvailableConditionController disables the controller that updates the Available conditions for
|
// DisableRemoteAvailableConditionController disables the controller that updates the Available conditions for
|
||||||
// APIServices, Endpoints and Services. This controller runs in kube-aggregator and can interfere with
|
// remote APIServices via querying endpoints of the referenced services. In generic controlplane use-cases,
|
||||||
// Generic Control Plane components when certain apis are not available.
|
// the concept of services and endpoints might differ, and might require another implementation of this
|
||||||
// TODO: We should find a better way to handle this. For now it will be for Generic Control Plane authors to
|
// controller. Local APIService are reconciled nevertheless.
|
||||||
// disable this controller if they see issues.
|
DisableRemoteAvailableConditionController bool
|
||||||
DisableAvailableConditionController bool
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Config represents the configuration needed to create an APIAggregator.
|
// Config represents the configuration needed to create an APIAggregator.
|
||||||
@ -320,6 +320,12 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
s.GenericAPIServer.AddPostStartHookOrDie("start-kube-aggregator-informers", func(context genericapiserver.PostStartHookContext) error {
|
||||||
|
informerFactory.Start(context.Done())
|
||||||
|
c.GenericConfig.SharedInformerFactory.Start(context.Done())
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
// create shared (remote and local) availability metrics
|
// create shared (remote and local) availability metrics
|
||||||
// TODO: decouple from legacyregistry
|
// TODO: decouple from legacyregistry
|
||||||
metrics := availabilitymetrics.New()
|
metrics := availabilitymetrics.New()
|
||||||
@ -328,10 +334,25 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// If the AvailableConditionController is disabled, we don't need to start the informers
|
// always run local availability controller
|
||||||
// and the controller.
|
local, err := localavailability.New(
|
||||||
if !c.ExtraConfig.DisableAvailableConditionController {
|
informerFactory.Apiregistration().V1().APIServices(),
|
||||||
availableController, err := statuscontrollers.NewAvailableConditionController(
|
apiregistrationClient.ApiregistrationV1(),
|
||||||
|
metrics,
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
s.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-local-available-controller", func(context genericapiserver.PostStartHookContext) error {
|
||||||
|
// if we end up blocking for long periods of time, we may need to increase workers.
|
||||||
|
go local.Run(5, context.Done())
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
// conditionally run remote availability controller. This could be replaced in certain
|
||||||
|
// generic controlplane use-cases where there is another concept of services and/or endpoints.
|
||||||
|
if !c.ExtraConfig.DisableRemoteAvailableConditionController {
|
||||||
|
remote, err := remoteavailability.New(
|
||||||
informerFactory.Apiregistration().V1().APIServices(),
|
informerFactory.Apiregistration().V1().APIServices(),
|
||||||
c.GenericConfig.SharedInformerFactory.Core().V1().Services(),
|
c.GenericConfig.SharedInformerFactory.Core().V1().Services(),
|
||||||
c.GenericConfig.SharedInformerFactory.Core().V1().Endpoints(),
|
c.GenericConfig.SharedInformerFactory.Core().V1().Endpoints(),
|
||||||
@ -344,16 +365,9 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
s.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-remote-available-controller", func(context genericapiserver.PostStartHookContext) error {
|
||||||
s.GenericAPIServer.AddPostStartHookOrDie("start-kube-aggregator-informers", func(context genericapiserver.PostStartHookContext) error {
|
|
||||||
informerFactory.Start(context.Done())
|
|
||||||
c.GenericConfig.SharedInformerFactory.Start(context.Done())
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
|
|
||||||
s.GenericAPIServer.AddPostStartHookOrDie("apiservice-status-available-controller", func(context genericapiserver.PostStartHookContext) error {
|
|
||||||
// if we end up blocking for long periods of time, we may need to increase workers.
|
// if we end up blocking for long periods of time, we may need to increase workers.
|
||||||
go availableController.Run(5, context.Done())
|
go remote.Run(5, context.Done())
|
||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
@ -0,0 +1,227 @@
|
|||||||
|
/*
|
||||||
|
Copyright 2024 The Kubernetes 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 external
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"k8s.io/apimachinery/pkg/api/equality"
|
||||||
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||||
|
"k8s.io/apimachinery/pkg/util/wait"
|
||||||
|
"k8s.io/client-go/tools/cache"
|
||||||
|
"k8s.io/client-go/util/workqueue"
|
||||||
|
"k8s.io/klog/v2"
|
||||||
|
apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
||||||
|
apiregistrationv1apihelper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
||||||
|
apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/typed/apiregistration/v1"
|
||||||
|
informers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1"
|
||||||
|
listers "k8s.io/kube-aggregator/pkg/client/listers/apiregistration/v1"
|
||||||
|
"k8s.io/kube-aggregator/pkg/controllers"
|
||||||
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
|
)
|
||||||
|
|
||||||
|
// AvailableConditionController handles checking the availability of registered local API services.
|
||||||
|
type AvailableConditionController struct {
|
||||||
|
apiServiceClient apiregistrationclient.APIServicesGetter
|
||||||
|
|
||||||
|
apiServiceLister listers.APIServiceLister
|
||||||
|
apiServiceSynced cache.InformerSynced
|
||||||
|
|
||||||
|
// To allow injection for testing.
|
||||||
|
syncFn func(key string) error
|
||||||
|
|
||||||
|
queue workqueue.TypedRateLimitingInterface[string]
|
||||||
|
|
||||||
|
// metrics registered into legacy registry
|
||||||
|
metrics *availabilitymetrics.Metrics
|
||||||
|
}
|
||||||
|
|
||||||
|
// New returns a new local availability AvailableConditionController.
|
||||||
|
func New(
|
||||||
|
apiServiceInformer informers.APIServiceInformer,
|
||||||
|
apiServiceClient apiregistrationclient.APIServicesGetter,
|
||||||
|
metrics *availabilitymetrics.Metrics,
|
||||||
|
) (*AvailableConditionController, error) {
|
||||||
|
c := &AvailableConditionController{
|
||||||
|
apiServiceClient: apiServiceClient,
|
||||||
|
apiServiceLister: apiServiceInformer.Lister(),
|
||||||
|
queue: workqueue.NewTypedRateLimitingQueueWithConfig(
|
||||||
|
// We want a fairly tight requeue time. The controller listens to the API, but because it relies on the routability of the
|
||||||
|
// service network, it is possible for an external, non-watchable factor to affect availability. This keeps
|
||||||
|
// the maximum disruption time to a minimum, but it does prevent hot loops.
|
||||||
|
workqueue.NewTypedItemExponentialFailureRateLimiter[string](5*time.Millisecond, 30*time.Second),
|
||||||
|
workqueue.TypedRateLimitingQueueConfig[string]{Name: "LocalAvailabilityController"},
|
||||||
|
),
|
||||||
|
metrics: metrics,
|
||||||
|
}
|
||||||
|
|
||||||
|
// resync on this one because it is low cardinality and rechecking the actual discovery
|
||||||
|
// allows us to detect health in a more timely fashion when network connectivity to
|
||||||
|
// nodes is snipped, but the network still attempts to route there. See
|
||||||
|
// https://github.com/openshift/origin/issues/17159#issuecomment-341798063
|
||||||
|
apiServiceHandler, _ := apiServiceInformer.Informer().AddEventHandlerWithResyncPeriod(
|
||||||
|
cache.ResourceEventHandlerFuncs{
|
||||||
|
AddFunc: c.addAPIService,
|
||||||
|
UpdateFunc: c.updateAPIService,
|
||||||
|
DeleteFunc: c.deleteAPIService,
|
||||||
|
},
|
||||||
|
30*time.Second)
|
||||||
|
c.apiServiceSynced = apiServiceHandler.HasSynced
|
||||||
|
|
||||||
|
c.syncFn = c.sync
|
||||||
|
|
||||||
|
return c, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *AvailableConditionController) sync(key string) error {
|
||||||
|
originalAPIService, err := c.apiServiceLister.Get(key)
|
||||||
|
if apierrors.IsNotFound(err) {
|
||||||
|
c.metrics.ForgetAPIService(key)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
if originalAPIService.Spec.Service != nil {
|
||||||
|
// this controller only handles local APIServices
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// local API services are always considered available
|
||||||
|
apiService := originalAPIService.DeepCopy()
|
||||||
|
apiregistrationv1apihelper.SetAPIServiceCondition(apiService, apiregistrationv1apihelper.NewLocalAvailableAPIServiceCondition())
|
||||||
|
_, err = c.updateAPIServiceStatus(originalAPIService, apiService)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// updateAPIServiceStatus only issues an update if a change is detected. We have a tight resync loop to quickly detect dead
|
||||||
|
// apiservices. Doing that means we don't want to quickly issue no-op updates.
|
||||||
|
func (c *AvailableConditionController) updateAPIServiceStatus(originalAPIService, newAPIService *apiregistrationv1.APIService) (*apiregistrationv1.APIService, error) {
|
||||||
|
// update this metric on every sync operation to reflect the actual state
|
||||||
|
c.metrics.SetUnavailableGauge(newAPIService)
|
||||||
|
|
||||||
|
if equality.Semantic.DeepEqual(originalAPIService.Status, newAPIService.Status) {
|
||||||
|
return newAPIService, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
orig := apiregistrationv1apihelper.GetAPIServiceConditionByType(originalAPIService, apiregistrationv1.Available)
|
||||||
|
now := apiregistrationv1apihelper.GetAPIServiceConditionByType(newAPIService, apiregistrationv1.Available)
|
||||||
|
unknown := apiregistrationv1.APIServiceCondition{
|
||||||
|
Type: apiregistrationv1.Available,
|
||||||
|
Status: apiregistrationv1.ConditionUnknown,
|
||||||
|
}
|
||||||
|
if orig == nil {
|
||||||
|
orig = &unknown
|
||||||
|
}
|
||||||
|
if now == nil {
|
||||||
|
now = &unknown
|
||||||
|
}
|
||||||
|
if *orig != *now {
|
||||||
|
klog.V(2).InfoS("changing APIService availability", "name", newAPIService.Name, "oldStatus", orig.Status, "newStatus", now.Status, "message", now.Message, "reason", now.Reason)
|
||||||
|
}
|
||||||
|
|
||||||
|
newAPIService, err := c.apiServiceClient.APIServices().UpdateStatus(context.TODO(), newAPIService, metav1.UpdateOptions{})
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
c.metrics.SetUnavailableCounter(originalAPIService, newAPIService)
|
||||||
|
return newAPIService, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Run starts the AvailableConditionController loop which manages the availability condition of API services.
|
||||||
|
func (c *AvailableConditionController) Run(workers int, stopCh <-chan struct{}) {
|
||||||
|
defer utilruntime.HandleCrash()
|
||||||
|
defer c.queue.ShutDown()
|
||||||
|
|
||||||
|
klog.Info("Starting LocalAvailability controller")
|
||||||
|
defer klog.Info("Shutting down LocalAvailability controller")
|
||||||
|
|
||||||
|
// This waits not just for the informers to sync, but for our handlers
|
||||||
|
// to be called; since the handlers are three different ways of
|
||||||
|
// enqueueing the same thing, waiting for this permits the queue to
|
||||||
|
// maximally de-duplicate the entries.
|
||||||
|
if !controllers.WaitForCacheSync("LocalAvailability", stopCh, c.apiServiceSynced) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
for i := 0; i < workers; i++ {
|
||||||
|
go wait.Until(c.runWorker, time.Second, stopCh)
|
||||||
|
}
|
||||||
|
|
||||||
|
<-stopCh
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *AvailableConditionController) runWorker() {
|
||||||
|
for c.processNextWorkItem() {
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// processNextWorkItem deals with one key off the queue. It returns false when it's time to quit.
|
||||||
|
func (c *AvailableConditionController) processNextWorkItem() bool {
|
||||||
|
key, quit := c.queue.Get()
|
||||||
|
if quit {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
defer c.queue.Done(key)
|
||||||
|
|
||||||
|
err := c.syncFn(key)
|
||||||
|
if err == nil {
|
||||||
|
c.queue.Forget(key)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
utilruntime.HandleError(fmt.Errorf("%v failed with: %w", key, err))
|
||||||
|
c.queue.AddRateLimited(key)
|
||||||
|
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *AvailableConditionController) addAPIService(obj interface{}) {
|
||||||
|
castObj := obj.(*apiregistrationv1.APIService)
|
||||||
|
klog.V(4).Infof("Adding %s", castObj.Name)
|
||||||
|
c.queue.Add(castObj.Name)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *AvailableConditionController) updateAPIService(oldObj, _ interface{}) {
|
||||||
|
oldCastObj := oldObj.(*apiregistrationv1.APIService)
|
||||||
|
klog.V(4).Infof("Updating %s", oldCastObj.Name)
|
||||||
|
c.queue.Add(oldCastObj.Name)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *AvailableConditionController) deleteAPIService(obj interface{}) {
|
||||||
|
castObj, ok := obj.(*apiregistrationv1.APIService)
|
||||||
|
if !ok {
|
||||||
|
tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
|
||||||
|
if !ok {
|
||||||
|
klog.Errorf("Couldn't get object from tombstone %#v", obj)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
castObj, ok = tombstone.Obj.(*apiregistrationv1.APIService)
|
||||||
|
if !ok {
|
||||||
|
klog.Errorf("Tombstone contained object that is not expected %#v", obj)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
klog.V(4).Infof("Deleting %q", castObj.Name)
|
||||||
|
c.queue.Add(castObj.Name)
|
||||||
|
}
|
@ -0,0 +1,168 @@
|
|||||||
|
/*
|
||||||
|
Copyright 2024 The Kubernetes 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 external
|
||||||
|
|
||||||
|
import (
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
|
"k8s.io/apimachinery/pkg/runtime"
|
||||||
|
"k8s.io/apimachinery/pkg/util/dump"
|
||||||
|
clienttesting "k8s.io/client-go/testing"
|
||||||
|
"k8s.io/client-go/tools/cache"
|
||||||
|
apiregistration "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
||||||
|
"k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/fake"
|
||||||
|
apiregistrationclient "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/typed/apiregistration/v1"
|
||||||
|
listers "k8s.io/kube-aggregator/pkg/client/listers/apiregistration/v1"
|
||||||
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
|
"k8s.io/utils/ptr"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
testServicePort int32 = 1234
|
||||||
|
)
|
||||||
|
|
||||||
|
func newLocalAPIService(name string) *apiregistration.APIService {
|
||||||
|
return &apiregistration.APIService{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: name},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func newRemoteAPIService(name string) *apiregistration.APIService {
|
||||||
|
return &apiregistration.APIService{
|
||||||
|
ObjectMeta: metav1.ObjectMeta{Name: name},
|
||||||
|
Spec: apiregistration.APIServiceSpec{
|
||||||
|
Group: strings.SplitN(name, ".", 2)[0],
|
||||||
|
Version: strings.SplitN(name, ".", 2)[1],
|
||||||
|
Service: &apiregistration.ServiceReference{
|
||||||
|
Namespace: "foo",
|
||||||
|
Name: "bar",
|
||||||
|
Port: ptr.To(testServicePort),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestSync(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
|
||||||
|
apiServiceName string
|
||||||
|
apiServices []runtime.Object
|
||||||
|
|
||||||
|
expectedAvailability apiregistration.APIServiceCondition
|
||||||
|
expectedAction bool
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "local",
|
||||||
|
apiServiceName: "local.group",
|
||||||
|
apiServices: []runtime.Object{newLocalAPIService("local.group")},
|
||||||
|
expectedAvailability: apiregistration.APIServiceCondition{
|
||||||
|
Type: apiregistration.Available,
|
||||||
|
Status: apiregistration.ConditionTrue,
|
||||||
|
Reason: "Local",
|
||||||
|
Message: "Local APIServices are always available",
|
||||||
|
},
|
||||||
|
expectedAction: true,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "remote",
|
||||||
|
apiServiceName: "remote.group",
|
||||||
|
apiServices: []runtime.Object{newRemoteAPIService("remote.group")},
|
||||||
|
expectedAction: false,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range tests {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
fakeClient := fake.NewSimpleClientset(tc.apiServices...)
|
||||||
|
apiServiceIndexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc})
|
||||||
|
for _, obj := range tc.apiServices {
|
||||||
|
if err := apiServiceIndexer.Add(obj); err != nil {
|
||||||
|
t.Fatalf("failed to add object to indexer: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
c := AvailableConditionController{
|
||||||
|
apiServiceClient: fakeClient.ApiregistrationV1(),
|
||||||
|
apiServiceLister: listers.NewAPIServiceLister(apiServiceIndexer),
|
||||||
|
metrics: availabilitymetrics.New(),
|
||||||
|
}
|
||||||
|
if err := c.sync(tc.apiServiceName); err != nil {
|
||||||
|
t.Fatalf("unexpect sync error: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
// ought to have one action writing status
|
||||||
|
if e, a := tc.expectedAction, len(fakeClient.Actions()) == 1; e != a {
|
||||||
|
t.Fatalf("%v expected %v, got %v", tc.name, e, fakeClient.Actions())
|
||||||
|
}
|
||||||
|
if tc.expectedAction {
|
||||||
|
action, ok := fakeClient.Actions()[0].(clienttesting.UpdateAction)
|
||||||
|
if !ok {
|
||||||
|
t.Fatalf("%v got %v", tc.name, ok)
|
||||||
|
}
|
||||||
|
|
||||||
|
if e, a := 1, len(action.GetObject().(*apiregistration.APIService).Status.Conditions); e != a {
|
||||||
|
t.Fatalf("%v expected %v, got %v", tc.name, e, action.GetObject())
|
||||||
|
}
|
||||||
|
condition := action.GetObject().(*apiregistration.APIService).Status.Conditions[0]
|
||||||
|
if e, a := tc.expectedAvailability.Type, condition.Type; e != a {
|
||||||
|
t.Errorf("%v expected %v, got %#v", tc.name, e, condition)
|
||||||
|
}
|
||||||
|
if e, a := tc.expectedAvailability.Status, condition.Status; e != a {
|
||||||
|
t.Errorf("%v expected %v, got %#v", tc.name, e, condition)
|
||||||
|
}
|
||||||
|
if e, a := tc.expectedAvailability.Reason, condition.Reason; e != a {
|
||||||
|
t.Errorf("%v expected %v, got %#v", tc.name, e, condition)
|
||||||
|
}
|
||||||
|
if e, a := tc.expectedAvailability.Message, condition.Message; !strings.HasPrefix(a, e) {
|
||||||
|
t.Errorf("%v expected %v, got %#v", tc.name, e, condition)
|
||||||
|
}
|
||||||
|
if condition.LastTransitionTime.IsZero() {
|
||||||
|
t.Error("expected lastTransitionTime to be non-zero")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUpdateAPIServiceStatus(t *testing.T) {
|
||||||
|
foo := &apiregistration.APIService{Status: apiregistration.APIServiceStatus{Conditions: []apiregistration.APIServiceCondition{{Type: "foo"}}}}
|
||||||
|
bar := &apiregistration.APIService{Status: apiregistration.APIServiceStatus{Conditions: []apiregistration.APIServiceCondition{{Type: "bar"}}}}
|
||||||
|
|
||||||
|
fakeClient := fake.NewSimpleClientset(foo)
|
||||||
|
c := AvailableConditionController{
|
||||||
|
apiServiceClient: fakeClient.ApiregistrationV1().(apiregistrationclient.APIServicesGetter),
|
||||||
|
metrics: availabilitymetrics.New(),
|
||||||
|
}
|
||||||
|
|
||||||
|
if _, err := c.updateAPIServiceStatus(foo, foo); err != nil {
|
||||||
|
t.Fatalf("unexpected updateAPIServiceStatus error: %v", err)
|
||||||
|
}
|
||||||
|
if e, a := 0, len(fakeClient.Actions()); e != a {
|
||||||
|
t.Error(dump.Pretty(fakeClient.Actions()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fakeClient.ClearActions()
|
||||||
|
if _, err := c.updateAPIServiceStatus(foo, bar); err != nil {
|
||||||
|
t.Fatalf("unexpected updateAPIServiceStatus error: %v", err)
|
||||||
|
}
|
||||||
|
if e, a := 1, len(fakeClient.Actions()); e != a {
|
||||||
|
t.Error(dump.Pretty(fakeClient.Actions()))
|
||||||
|
}
|
||||||
|
}
|
@ -88,8 +88,8 @@ type AvailableConditionController struct {
|
|||||||
metrics *availabilitymetrics.Metrics
|
metrics *availabilitymetrics.Metrics
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewAvailableConditionController returns a new AvailableConditionController.
|
// New returns a new remote APIService AvailableConditionController.
|
||||||
func NewAvailableConditionController(
|
func New(
|
||||||
apiServiceInformer informers.APIServiceInformer,
|
apiServiceInformer informers.APIServiceInformer,
|
||||||
serviceInformer v1informers.ServiceInformer,
|
serviceInformer v1informers.ServiceInformer,
|
||||||
endpointsInformer v1informers.EndpointsInformer,
|
endpointsInformer v1informers.EndpointsInformer,
|
||||||
@ -110,7 +110,7 @@ func NewAvailableConditionController(
|
|||||||
// service network, it is possible for an external, non-watchable factor to affect availability. This keeps
|
// service network, it is possible for an external, non-watchable factor to affect availability. This keeps
|
||||||
// the maximum disruption time to a minimum, but it does prevent hot loops.
|
// the maximum disruption time to a minimum, but it does prevent hot loops.
|
||||||
workqueue.NewTypedItemExponentialFailureRateLimiter[string](5*time.Millisecond, 30*time.Second),
|
workqueue.NewTypedItemExponentialFailureRateLimiter[string](5*time.Millisecond, 30*time.Second),
|
||||||
workqueue.TypedRateLimitingQueueConfig[string]{Name: "AvailableConditionController"},
|
workqueue.TypedRateLimitingQueueConfig[string]{Name: "RemoteAvailabilityController"},
|
||||||
),
|
),
|
||||||
proxyTransportDial: proxyTransportDial,
|
proxyTransportDial: proxyTransportDial,
|
||||||
proxyCurrentCertKeyContent: proxyCurrentCertKeyContent,
|
proxyCurrentCertKeyContent: proxyCurrentCertKeyContent,
|
||||||
@ -159,6 +159,13 @@ func (c *AvailableConditionController) sync(key string) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if originalAPIService.Spec.Service == nil {
|
||||||
|
// handled by the local APIService controller
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
apiService := originalAPIService.DeepCopy()
|
||||||
|
|
||||||
// if a particular transport was specified, use that otherwise build one
|
// if a particular transport was specified, use that otherwise build one
|
||||||
// construct an http client that will ignore TLS verification (if someone owns the network and messes with your status
|
// construct an http client that will ignore TLS verification (if someone owns the network and messes with your status
|
||||||
// that's not so bad) and sets a very short timeout. This is a best effort GET that provides no additional information
|
// that's not so bad) and sets a very short timeout. This is a best effort GET that provides no additional information
|
||||||
@ -188,21 +195,12 @@ func (c *AvailableConditionController) sync(key string) error {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
apiService := originalAPIService.DeepCopy()
|
|
||||||
|
|
||||||
availableCondition := apiregistrationv1.APIServiceCondition{
|
availableCondition := apiregistrationv1.APIServiceCondition{
|
||||||
Type: apiregistrationv1.Available,
|
Type: apiregistrationv1.Available,
|
||||||
Status: apiregistrationv1.ConditionTrue,
|
Status: apiregistrationv1.ConditionTrue,
|
||||||
LastTransitionTime: metav1.Now(),
|
LastTransitionTime: metav1.Now(),
|
||||||
}
|
}
|
||||||
|
|
||||||
// local API services are always considered available
|
|
||||||
if apiService.Spec.Service == nil {
|
|
||||||
apiregistrationv1apihelper.SetAPIServiceCondition(apiService, apiregistrationv1apihelper.NewLocalAvailableAPIServiceCondition())
|
|
||||||
_, err := c.updateAPIServiceStatus(originalAPIService, apiService)
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
service, err := c.serviceLister.Services(apiService.Spec.Service.Namespace).Get(apiService.Spec.Service.Name)
|
service, err := c.serviceLister.Services(apiService.Spec.Service.Namespace).Get(apiService.Spec.Service.Name)
|
||||||
if apierrors.IsNotFound(err) {
|
if apierrors.IsNotFound(err) {
|
||||||
availableCondition.Status = apiregistrationv1.ConditionFalse
|
availableCondition.Status = apiregistrationv1.ConditionFalse
|
||||||
@ -410,14 +408,14 @@ func (c *AvailableConditionController) Run(workers int, stopCh <-chan struct{})
|
|||||||
defer utilruntime.HandleCrash()
|
defer utilruntime.HandleCrash()
|
||||||
defer c.queue.ShutDown()
|
defer c.queue.ShutDown()
|
||||||
|
|
||||||
klog.Info("Starting AvailableConditionController")
|
klog.Info("Starting RemoteAvailability controller")
|
||||||
defer klog.Info("Shutting down AvailableConditionController")
|
defer klog.Info("Shutting down RemoteAvailability controller")
|
||||||
|
|
||||||
// This waits not just for the informers to sync, but for our handlers
|
// This waits not just for the informers to sync, but for our handlers
|
||||||
// to be called; since the handlers are three different ways of
|
// to be called; since the handlers are three different ways of
|
||||||
// enqueueing the same thing, waiting for this permits the queue to
|
// enqueueing the same thing, waiting for this permits the queue to
|
||||||
// maximally de-duplicate the entries.
|
// maximally de-duplicate the entries.
|
||||||
if !controllers.WaitForCacheSync("AvailableConditionController", stopCh, c.apiServiceSynced, c.servicesSynced, c.endpointsSynced) {
|
if !controllers.WaitForCacheSync("RemoteAvailability", stopCh, c.apiServiceSynced, c.servicesSynced, c.endpointsSynced) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -225,6 +225,7 @@ func TestSync(t *testing.T) {
|
|||||||
|
|
||||||
expectedAvailability apiregistration.APIServiceCondition
|
expectedAvailability apiregistration.APIServiceCondition
|
||||||
expectedSyncError string
|
expectedSyncError string
|
||||||
|
expectedSkipped bool
|
||||||
}{
|
}{
|
||||||
{
|
{
|
||||||
name: "local",
|
name: "local",
|
||||||
@ -237,6 +238,7 @@ func TestSync(t *testing.T) {
|
|||||||
Reason: "Local",
|
Reason: "Local",
|
||||||
Message: "Local APIServices are always available",
|
Message: "Local APIServices are always available",
|
||||||
},
|
},
|
||||||
|
expectedSkipped: true,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
name: "no service",
|
name: "no service",
|
||||||
@ -420,6 +422,13 @@ func TestSync(t *testing.T) {
|
|||||||
t.Fatalf("%v unexpected sync error: %v", tc.name, err)
|
t.Fatalf("%v unexpected sync error: %v", tc.name, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if tc.expectedSkipped {
|
||||||
|
if len(fakeClient.Actions()) > 0 {
|
||||||
|
t.Fatalf("%v expected no actions, got %v", tc.name, fakeClient.Actions())
|
||||||
|
}
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// ought to have one action writing status
|
// ought to have one action writing status
|
||||||
if e, a := 1, len(fakeClient.Actions()); e != a {
|
if e, a := 1, len(fakeClient.Actions()); e != a {
|
||||||
t.Fatalf("%v expected %v, got %v", tc.name, e, fakeClient.Actions())
|
t.Fatalf("%v expected %v, got %v", tc.name, e, fakeClient.Actions())
|
||||||
|
@ -51,7 +51,8 @@ var (
|
|||||||
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
||||||
"[+]poststarthook/start-kube-aggregator-informers ok",
|
"[+]poststarthook/start-kube-aggregator-informers ok",
|
||||||
"[+]poststarthook/apiservice-registration-controller ok",
|
"[+]poststarthook/apiservice-registration-controller ok",
|
||||||
"[+]poststarthook/apiservice-status-available-controller ok",
|
"[+]poststarthook/apiservice-status-local-available-controller ok",
|
||||||
|
"[+]poststarthook/apiservice-status-remote-available-controller ok",
|
||||||
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
||||||
"[+]autoregister-completion ok",
|
"[+]autoregister-completion ok",
|
||||||
"[+]poststarthook/apiservice-openapi-controller ok",
|
"[+]poststarthook/apiservice-openapi-controller ok",
|
||||||
@ -72,7 +73,8 @@ var (
|
|||||||
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
||||||
"[+]poststarthook/start-kube-aggregator-informers ok",
|
"[+]poststarthook/start-kube-aggregator-informers ok",
|
||||||
"[+]poststarthook/apiservice-registration-controller ok",
|
"[+]poststarthook/apiservice-registration-controller ok",
|
||||||
"[+]poststarthook/apiservice-status-available-controller ok",
|
"[+]poststarthook/apiservice-status-local-available-controller ok",
|
||||||
|
"[+]poststarthook/apiservice-status-remote-available-controller ok",
|
||||||
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
||||||
"[+]autoregister-completion ok",
|
"[+]autoregister-completion ok",
|
||||||
"[+]poststarthook/apiservice-openapi-controller ok",
|
"[+]poststarthook/apiservice-openapi-controller ok",
|
||||||
@ -94,7 +96,8 @@ var (
|
|||||||
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
"[+]poststarthook/start-cluster-authentication-info-controller ok",
|
||||||
"[+]poststarthook/start-kube-aggregator-informers ok",
|
"[+]poststarthook/start-kube-aggregator-informers ok",
|
||||||
"[+]poststarthook/apiservice-registration-controller ok",
|
"[+]poststarthook/apiservice-registration-controller ok",
|
||||||
"[+]poststarthook/apiservice-status-available-controller ok",
|
"[+]poststarthook/apiservice-status-local-available-controller ok",
|
||||||
|
"[+]poststarthook/apiservice-status-remote-available-controller ok",
|
||||||
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
"[+]poststarthook/kube-apiserver-autoregistration ok",
|
||||||
"[+]autoregister-completion ok",
|
"[+]autoregister-completion ok",
|
||||||
"[+]poststarthook/apiservice-openapi-controller ok",
|
"[+]poststarthook/apiservice-openapi-controller ok",
|
||||||
|
Loading…
Reference in New Issue
Block a user