mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-08-05 18:24:07 +00:00
aggregator: separate out status controller metrics
Signed-off-by: Dr. Stefan Schimanski <stefan.schimanski@gmail.com>
This commit is contained in:
parent
10496b35a8
commit
c5095069a8
@ -20,6 +20,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
@ -37,6 +38,7 @@ import (
|
|||||||
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
||||||
"k8s.io/client-go/kubernetes"
|
"k8s.io/client-go/kubernetes"
|
||||||
"k8s.io/client-go/transport"
|
"k8s.io/client-go/transport"
|
||||||
|
"k8s.io/component-base/metrics/legacyregistry"
|
||||||
"k8s.io/component-base/tracing"
|
"k8s.io/component-base/tracing"
|
||||||
v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
v1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
||||||
v1helper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
v1helper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
||||||
@ -50,10 +52,14 @@ import (
|
|||||||
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"
|
||||||
statuscontrollers "k8s.io/kube-aggregator/pkg/controllers/status"
|
statuscontrollers "k8s.io/kube-aggregator/pkg/controllers/status"
|
||||||
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
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"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// making sure we only register metrics once into legacy registry
|
||||||
|
var registerIntoLegacyRegistryOnce sync.Once
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
// we need to add the options (like ListOptions) to empty v1
|
// we need to add the options (like ListOptions) to empty v1
|
||||||
metav1.AddToGroupVersion(aggregatorscheme.Scheme, schema.GroupVersion{Group: "", Version: "v1"})
|
metav1.AddToGroupVersion(aggregatorscheme.Scheme, schema.GroupVersion{Group: "", Version: "v1"})
|
||||||
@ -314,6 +320,14 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// create shared (remote and local) availability metrics
|
||||||
|
// TODO: decouple from legacyregistry
|
||||||
|
metrics := availabilitymetrics.New()
|
||||||
|
registerIntoLegacyRegistryOnce.Do(func() { err = metrics.Register(legacyregistry.Register, legacyregistry.CustomRegister) })
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
// If the AvailableConditionController is disabled, we don't need to start the informers
|
// If the AvailableConditionController is disabled, we don't need to start the informers
|
||||||
// and the controller.
|
// and the controller.
|
||||||
if !c.ExtraConfig.DisableAvailableConditionController {
|
if !c.ExtraConfig.DisableAvailableConditionController {
|
||||||
@ -325,6 +339,7 @@ func (c completedConfig) NewWithDelegate(delegationTarget genericapiserver.Deleg
|
|||||||
proxyTransportDial,
|
proxyTransportDial,
|
||||||
(func() ([]byte, []byte))(s.proxyCurrentCertKeyContent),
|
(func() ([]byte, []byte))(s.proxyCurrentCertKeyContent),
|
||||||
s.serviceResolver,
|
s.serviceResolver,
|
||||||
|
metrics,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -39,7 +39,6 @@ import (
|
|||||||
"k8s.io/client-go/tools/cache"
|
"k8s.io/client-go/tools/cache"
|
||||||
"k8s.io/client-go/transport"
|
"k8s.io/client-go/transport"
|
||||||
"k8s.io/client-go/util/workqueue"
|
"k8s.io/client-go/util/workqueue"
|
||||||
"k8s.io/component-base/metrics/legacyregistry"
|
|
||||||
"k8s.io/klog/v2"
|
"k8s.io/klog/v2"
|
||||||
apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
||||||
apiregistrationv1apihelper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
apiregistrationv1apihelper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
||||||
@ -47,11 +46,9 @@ import (
|
|||||||
informers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1"
|
informers "k8s.io/kube-aggregator/pkg/client/informers/externalversions/apiregistration/v1"
|
||||||
listers "k8s.io/kube-aggregator/pkg/client/listers/apiregistration/v1"
|
listers "k8s.io/kube-aggregator/pkg/client/listers/apiregistration/v1"
|
||||||
"k8s.io/kube-aggregator/pkg/controllers"
|
"k8s.io/kube-aggregator/pkg/controllers"
|
||||||
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
)
|
)
|
||||||
|
|
||||||
// making sure we only register metrics once into legacy registry
|
|
||||||
var registerIntoLegacyRegistryOnce sync.Once
|
|
||||||
|
|
||||||
type certKeyFunc func() ([]byte, []byte)
|
type certKeyFunc func() ([]byte, []byte)
|
||||||
|
|
||||||
// ServiceResolver knows how to convert a service reference into an actual location.
|
// ServiceResolver knows how to convert a service reference into an actual location.
|
||||||
@ -88,7 +85,7 @@ type AvailableConditionController struct {
|
|||||||
cacheLock sync.RWMutex
|
cacheLock sync.RWMutex
|
||||||
|
|
||||||
// metrics registered into legacy registry
|
// metrics registered into legacy registry
|
||||||
metrics *availabilityMetrics
|
metrics *availabilitymetrics.Metrics
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewAvailableConditionController returns a new AvailableConditionController.
|
// NewAvailableConditionController returns a new AvailableConditionController.
|
||||||
@ -100,6 +97,7 @@ func NewAvailableConditionController(
|
|||||||
proxyTransportDial *transport.DialHolder,
|
proxyTransportDial *transport.DialHolder,
|
||||||
proxyCurrentCertKeyContent certKeyFunc,
|
proxyCurrentCertKeyContent certKeyFunc,
|
||||||
serviceResolver ServiceResolver,
|
serviceResolver ServiceResolver,
|
||||||
|
metrics *availabilitymetrics.Metrics,
|
||||||
) (*AvailableConditionController, error) {
|
) (*AvailableConditionController, error) {
|
||||||
c := &AvailableConditionController{
|
c := &AvailableConditionController{
|
||||||
apiServiceClient: apiServiceClient,
|
apiServiceClient: apiServiceClient,
|
||||||
@ -116,7 +114,7 @@ func NewAvailableConditionController(
|
|||||||
),
|
),
|
||||||
proxyTransportDial: proxyTransportDial,
|
proxyTransportDial: proxyTransportDial,
|
||||||
proxyCurrentCertKeyContent: proxyCurrentCertKeyContent,
|
proxyCurrentCertKeyContent: proxyCurrentCertKeyContent,
|
||||||
metrics: newAvailabilityMetrics(),
|
metrics: metrics,
|
||||||
}
|
}
|
||||||
|
|
||||||
// resync on this one because it is low cardinality and rechecking the actual discovery
|
// resync on this one because it is low cardinality and rechecking the actual discovery
|
||||||
@ -148,15 +146,6 @@ func NewAvailableConditionController(
|
|||||||
|
|
||||||
c.syncFn = c.sync
|
c.syncFn = c.sync
|
||||||
|
|
||||||
// TODO: decouple from legacyregistry
|
|
||||||
var err error
|
|
||||||
registerIntoLegacyRegistryOnce.Do(func() {
|
|
||||||
err = c.metrics.Register(legacyregistry.Register, legacyregistry.CustomRegister)
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
return c, nil
|
return c, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -385,7 +374,7 @@ func (c *AvailableConditionController) sync(key string) error {
|
|||||||
// apiservices. Doing that means we don't want to quickly issue no-op updates.
|
// 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) {
|
func (c *AvailableConditionController) updateAPIServiceStatus(originalAPIService, newAPIService *apiregistrationv1.APIService) (*apiregistrationv1.APIService, error) {
|
||||||
// update this metric on every sync operation to reflect the actual state
|
// update this metric on every sync operation to reflect the actual state
|
||||||
c.setUnavailableGauge(newAPIService)
|
c.metrics.SetUnavailableGauge(newAPIService)
|
||||||
|
|
||||||
if equality.Semantic.DeepEqual(originalAPIService.Status, newAPIService.Status) {
|
if equality.Semantic.DeepEqual(originalAPIService.Status, newAPIService.Status) {
|
||||||
return newAPIService, nil
|
return newAPIService, nil
|
||||||
@ -412,7 +401,7 @@ func (c *AvailableConditionController) updateAPIServiceStatus(originalAPIService
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
c.setUnavailableCounter(originalAPIService, newAPIService)
|
c.metrics.SetUnavailableCounter(originalAPIService, newAPIService)
|
||||||
return newAPIService, nil
|
return newAPIService, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -599,28 +588,3 @@ func (c *AvailableConditionController) deleteEndpoints(obj interface{}) {
|
|||||||
c.queue.Add(apiService)
|
c.queue.Add(apiService)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// setUnavailableGauge set the metrics so that it reflect the current state base on availability of the given service
|
|
||||||
func (c *AvailableConditionController) setUnavailableGauge(newAPIService *apiregistrationv1.APIService) {
|
|
||||||
if apiregistrationv1apihelper.IsAPIServiceConditionTrue(newAPIService, apiregistrationv1.Available) {
|
|
||||||
c.metrics.SetAPIServiceAvailable(newAPIService.Name)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
c.metrics.SetAPIServiceUnavailable(newAPIService.Name)
|
|
||||||
}
|
|
||||||
|
|
||||||
// setUnavailableCounter increases the metrics only if the given service is unavailable and its APIServiceCondition has changed
|
|
||||||
func (c *AvailableConditionController) setUnavailableCounter(originalAPIService, newAPIService *apiregistrationv1.APIService) {
|
|
||||||
wasAvailable := apiregistrationv1apihelper.IsAPIServiceConditionTrue(originalAPIService, apiregistrationv1.Available)
|
|
||||||
isAvailable := apiregistrationv1apihelper.IsAPIServiceConditionTrue(newAPIService, apiregistrationv1.Available)
|
|
||||||
statusChanged := isAvailable != wasAvailable
|
|
||||||
|
|
||||||
if statusChanged && !isAvailable {
|
|
||||||
reason := "UnknownReason"
|
|
||||||
if newCondition := apiregistrationv1apihelper.GetAPIServiceConditionByType(newAPIService, apiregistrationv1.Available); newCondition != nil {
|
|
||||||
reason = newCondition.Reason
|
|
||||||
}
|
|
||||||
c.metrics.UnavailableCounter(newAPIService.Name, reason).Inc()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
@ -25,6 +25,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
availabilitymetrics "k8s.io/kube-aggregator/pkg/controllers/status/metrics"
|
||||||
"k8s.io/utils/pointer"
|
"k8s.io/utils/pointer"
|
||||||
|
|
||||||
v1 "k8s.io/api/core/v1"
|
v1 "k8s.io/api/core/v1"
|
||||||
@ -133,7 +134,7 @@ func setupAPIServices(apiServices []*apiregistration.APIService) (*AvailableCond
|
|||||||
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: "AvailableConditionController"},
|
||||||
),
|
),
|
||||||
metrics: newAvailabilityMetrics(),
|
metrics: availabilitymetrics.New(),
|
||||||
}
|
}
|
||||||
for _, svc := range apiServices {
|
for _, svc := range apiServices {
|
||||||
c.addAPIService(svc)
|
c.addAPIService(svc)
|
||||||
@ -395,7 +396,7 @@ func TestSync(t *testing.T) {
|
|||||||
endpointsLister: v1listers.NewEndpointsLister(endpointsIndexer),
|
endpointsLister: v1listers.NewEndpointsLister(endpointsIndexer),
|
||||||
serviceResolver: &fakeServiceResolver{url: testServer.URL},
|
serviceResolver: &fakeServiceResolver{url: testServer.URL},
|
||||||
proxyCurrentCertKeyContent: func() ([]byte, []byte) { return emptyCert(), emptyCert() },
|
proxyCurrentCertKeyContent: func() ([]byte, []byte) { return emptyCert(), emptyCert() },
|
||||||
metrics: newAvailabilityMetrics(),
|
metrics: availabilitymetrics.New(),
|
||||||
}
|
}
|
||||||
c.sync(tc.apiServiceName)
|
c.sync(tc.apiServiceName)
|
||||||
|
|
||||||
@ -447,7 +448,7 @@ func TestUpdateAPIServiceStatus(t *testing.T) {
|
|||||||
fakeClient := fake.NewSimpleClientset()
|
fakeClient := fake.NewSimpleClientset()
|
||||||
c := AvailableConditionController{
|
c := AvailableConditionController{
|
||||||
apiServiceClient: fakeClient.ApiregistrationV1().(apiregistrationclient.APIServicesGetter),
|
apiServiceClient: fakeClient.ApiregistrationV1().(apiregistrationclient.APIServicesGetter),
|
||||||
metrics: newAvailabilityMetrics(),
|
metrics: availabilitymetrics.New(),
|
||||||
}
|
}
|
||||||
|
|
||||||
c.updateAPIServiceStatus(foo, foo)
|
c.updateAPIServiceStatus(foo, foo)
|
||||||
|
@ -14,12 +14,14 @@ See the License for the specific language governing permissions and
|
|||||||
limitations under the License.
|
limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package apiserver
|
package metrics
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"k8s.io/component-base/metrics"
|
"k8s.io/component-base/metrics"
|
||||||
|
apiregistrationv1 "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1"
|
||||||
|
apiregistrationv1apihelper "k8s.io/kube-aggregator/pkg/apis/apiregistration/v1/helper"
|
||||||
)
|
)
|
||||||
|
|
||||||
/*
|
/*
|
||||||
@ -41,14 +43,14 @@ var (
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
type availabilityMetrics struct {
|
type Metrics struct {
|
||||||
unavailableCounter *metrics.CounterVec
|
unavailableCounter *metrics.CounterVec
|
||||||
|
|
||||||
*availabilityCollector
|
*availabilityCollector
|
||||||
}
|
}
|
||||||
|
|
||||||
func newAvailabilityMetrics() *availabilityMetrics {
|
func New() *Metrics {
|
||||||
return &availabilityMetrics{
|
return &Metrics{
|
||||||
unavailableCounter: metrics.NewCounterVec(
|
unavailableCounter: metrics.NewCounterVec(
|
||||||
&metrics.CounterOpts{
|
&metrics.CounterOpts{
|
||||||
Name: "aggregator_unavailable_apiservice_total",
|
Name: "aggregator_unavailable_apiservice_total",
|
||||||
@ -62,7 +64,7 @@ func newAvailabilityMetrics() *availabilityMetrics {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Register registers apiservice availability metrics.
|
// Register registers apiservice availability metrics.
|
||||||
func (m *availabilityMetrics) Register(
|
func (m *Metrics) Register(
|
||||||
registrationFunc func(metrics.Registerable) error,
|
registrationFunc func(metrics.Registerable) error,
|
||||||
customRegistrationFunc func(metrics.StableCollector) error,
|
customRegistrationFunc func(metrics.StableCollector) error,
|
||||||
) error {
|
) error {
|
||||||
@ -80,7 +82,7 @@ func (m *availabilityMetrics) Register(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// UnavailableCounter returns a counter to track apiservices marked as unavailable.
|
// UnavailableCounter returns a counter to track apiservices marked as unavailable.
|
||||||
func (m *availabilityMetrics) UnavailableCounter(apiServiceName, reason string) metrics.CounterMetric {
|
func (m *Metrics) UnavailableCounter(apiServiceName, reason string) metrics.CounterMetric {
|
||||||
return m.unavailableCounter.WithLabelValues(apiServiceName, reason)
|
return m.unavailableCounter.WithLabelValues(apiServiceName, reason)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -91,6 +93,31 @@ type availabilityCollector struct {
|
|||||||
availabilities map[string]bool
|
availabilities map[string]bool
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SetUnavailableGauge set the metrics so that it reflect the current state base on availability of the given service
|
||||||
|
func (m *Metrics) SetUnavailableGauge(newAPIService *apiregistrationv1.APIService) {
|
||||||
|
if apiregistrationv1apihelper.IsAPIServiceConditionTrue(newAPIService, apiregistrationv1.Available) {
|
||||||
|
m.SetAPIServiceAvailable(newAPIService.Name)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
m.SetAPIServiceUnavailable(newAPIService.Name)
|
||||||
|
}
|
||||||
|
|
||||||
|
// SetUnavailableCounter increases the metrics only if the given service is unavailable and its APIServiceCondition has changed
|
||||||
|
func (m *Metrics) SetUnavailableCounter(originalAPIService, newAPIService *apiregistrationv1.APIService) {
|
||||||
|
wasAvailable := apiregistrationv1apihelper.IsAPIServiceConditionTrue(originalAPIService, apiregistrationv1.Available)
|
||||||
|
isAvailable := apiregistrationv1apihelper.IsAPIServiceConditionTrue(newAPIService, apiregistrationv1.Available)
|
||||||
|
statusChanged := isAvailable != wasAvailable
|
||||||
|
|
||||||
|
if statusChanged && !isAvailable {
|
||||||
|
reason := "UnknownReason"
|
||||||
|
if newCondition := apiregistrationv1apihelper.GetAPIServiceConditionByType(newAPIService, apiregistrationv1.Available); newCondition != nil {
|
||||||
|
reason = newCondition.Reason
|
||||||
|
}
|
||||||
|
m.UnavailableCounter(newAPIService.Name, reason).Inc()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Check if apiServiceStatusCollector implements necessary interface.
|
// Check if apiServiceStatusCollector implements necessary interface.
|
||||||
var _ metrics.StableCollector = &availabilityCollector{}
|
var _ metrics.StableCollector = &availabilityCollector{}
|
||||||
|
|
@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
|
|||||||
limitations under the License.
|
limitations under the License.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
package apiserver
|
package metrics
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"strings"
|
"strings"
|
Loading…
Reference in New Issue
Block a user