amassivus/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-07-03 11:03:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Kubernetes SD: Support discovery.k8s.io/v1 EndpointSlice

Browse source 
Fixes #9498

Signed-off-by: Furkan <furkan.turkal@trendyol.com>
Signed-off-by: Erkan Zileli <erkan.zileli@trendyol.com>
Co-authored-by: Batuhan Apaydin <batuhan.apaydin@trendyol.com>
This commit is contained in:
Furkan 2021-10-22 22:26:31 +03:00
parent 9ed56c6065
commit 2939966634
No known key found for this signature in database
GPG key ID: C26B5EA6BB8709D8
5 changed files with 775 additions and 174 deletions
Download patch file Download diff file Expand all files Collapse all files

78
discovery/kubernetes/kubernetes.go
View file

@ -30,6 +30,7 @@ import (
"github.com/prometheus/common/model"
"github.com/prometheus/common/version"
apiv1 "k8s.io/api/core/v1"
disv1 "k8s.io/api/discovery/v1"
disv1beta1 "k8s.io/api/discovery/v1beta1"
networkv1 "k8s.io/api/networking/v1"
"k8s.io/api/networking/v1beta1"
@ -385,20 +386,56 @@ func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
switch d.role {
case RoleEndpointSlice:
// Check "networking.k8s.io/v1" availability with retries.
// If "v1" is not available, use "networking.k8s.io/v1beta1" for backward compatibility
var v1Supported bool
if retryOnError(ctx, 10*time.Second,
func() (err error) {
v1Supported, err = checkDiscoveryV1Supported(d.client)
if err != nil {
level.Error(d.logger).Log("msg", "Failed to check networking.k8s.io/v1 availability", "err", err)
}
return err
},
) {
d.Unlock()
return
}
for _, namespace := range namespaces {
e := d.client.DiscoveryV1beta1().EndpointSlices(namespace)
elw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.List(ctx, options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.Watch(ctx, options)
},
var informer cache.SharedInformer
if v1Supported {
e := d.client.DiscoveryV1().EndpointSlices(namespace)
elw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.List(ctx, options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.Watch(ctx, options)
},
}
informer = cache.NewSharedInformer(elw, &disv1.EndpointSlice{}, resyncPeriod)
} else {
e := d.client.DiscoveryV1beta1().EndpointSlices(namespace)
elw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.List(ctx, options)
},
WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
options.FieldSelector = d.selectors.endpointslice.field
options.LabelSelector = d.selectors.endpointslice.label
return e.Watch(ctx, options)
},
}
informer = cache.NewSharedInformer(elw, &disv1beta1.EndpointSlice{}, resyncPeriod)
}
s := d.client.CoreV1().Services(namespace)
slw := &cache.ListWatch{
ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
@ -428,7 +465,7 @@ func (d *Discovery) Run(ctx context.Context, ch chan<- []*targetgroup.Group) {
eps := NewEndpointSlice(
log.With(d.logger, "role", "endpointslice"),
cache.NewSharedInformer(slw, &apiv1.Service{}, resyncPeriod),
cache.NewSharedInformer(elw, &disv1beta1.EndpointSlice{}, resyncPeriod),
informer,
cache.NewSharedInformer(plw, &apiv1.Pod{}, resyncPeriod),
)
d.discoverers = append(d.discoverers, eps)
@ -694,3 +731,18 @@ func (d *Discovery) newPodsByNodeInformer(plw *cache.ListWatch) cache.SharedInde
return cache.NewSharedIndexInformer(plw, &apiv1.Pod{}, resyncPeriod, indexers)
}
func checkDiscoveryV1Supported(client kubernetes.Interface) (bool, error) {
k8sVer, err := client.Discovery().ServerVersion()
if err != nil {
return false, err
}
semVer, err := utilversion.ParseSemantic(k8sVer.String())
if err != nil {
return false, err
}
// The discovery.k8s.io/v1beta1 API version of EndpointSlice will no longer be served in v1.25.
// discovery.k8s.io/v1 is available since Kubernetes v1.21
// https://kubernetes.io/docs/reference/using-api/deprecation-guide/#v1-25
return semVer.Major() >= 1 && semVer.Minor() >= 21, nil
}