k8s: Support discovery of ingresses (#3111)

* k8s: Support discovery of ingresses

* Move additional labels below allocation

This makes it more obvious why the additional elements are allocated.
Also fix allocation for node where we only set a single label.

* k8s: Remove port from ingress discovery

* k8s: Add comment to ingress discovery example
This commit is contained in:
Johannes 'fish' Ziemke 2017-09-04 13:10:44 +02:00 committed by GitHub
parent 29fff1eca4
commit 70f3d1e9f9
9 changed files with 430 additions and 49 deletions

View file

@ -1005,10 +1005,11 @@ type KubernetesRole string
// The valid options for KubernetesRole.
const (
KubernetesRoleNode = "node"
KubernetesRolePod = "pod"
KubernetesRoleService = "service"
KubernetesRoleEndpoint = "endpoints"
KubernetesRoleNode KubernetesRole = "node"
KubernetesRolePod KubernetesRole = "pod"
KubernetesRoleService KubernetesRole = "service"
KubernetesRoleEndpoint KubernetesRole = "endpoints"
KubernetesRoleIngress KubernetesRole = "ingress"
)
// UnmarshalYAML implements the yaml.Unmarshaler interface.
@ -1017,7 +1018,7 @@ func (c *KubernetesRole) UnmarshalYAML(unmarshal func(interface{}) error) error
return err
}
switch *c {
case KubernetesRoleNode, KubernetesRolePod, KubernetesRoleService, KubernetesRoleEndpoint:
case KubernetesRoleNode, KubernetesRolePod, KubernetesRoleService, KubernetesRoleEndpoint, KubernetesRoleIngress:
return nil
default:
return fmt.Errorf("Unknown Kubernetes SD role %q", *c)

View file

@ -154,8 +154,11 @@ func (e *Endpoints) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
}
func convertToEndpoints(o interface{}) (*apiv1.Endpoints, error) {
endpoints, isEndpoints := o.(*apiv1.Endpoints)
if !isEndpoints {
endpoints, ok := o.(*apiv1.Endpoints)
if ok {
return endpoints, nil
}
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
if !ok {
return nil, fmt.Errorf("Received unexpected object: %v", o)
@ -164,8 +167,6 @@ func convertToEndpoints(o interface{}) (*apiv1.Endpoints, error) {
if !ok {
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Endpoints object: %v", deletedState.Obj)
}
}
return endpoints, nil
}

View file

@ -0,0 +1,184 @@
// Copyright 2016 The Prometheus 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 kubernetes
import (
"fmt"
"github.com/prometheus/common/log"
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/config"
"github.com/prometheus/prometheus/util/strutil"
"golang.org/x/net/context"
"k8s.io/client-go/pkg/apis/extensions/v1beta1"
"k8s.io/client-go/tools/cache"
)
// Ingress implements discovery of Kubernetes ingresss.
type Ingress struct {
logger log.Logger
informer cache.SharedInformer
store cache.Store
}
// NewIngress returns a new ingress discovery.
func NewIngress(l log.Logger, inf cache.SharedInformer) *Ingress {
return &Ingress{logger: l, informer: inf, store: inf.GetStore()}
}
// Run implements the TargetProvider interface.
func (s *Ingress) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
// Send full initial set of pod targets.
var initial []*config.TargetGroup
for _, o := range s.store.List() {
tg := s.buildIngress(o.(*v1beta1.Ingress))
initial = append(initial, tg)
}
select {
case <-ctx.Done():
return
case ch <- initial:
}
// Send target groups for ingress updates.
send := func(tg *config.TargetGroup) {
select {
case <-ctx.Done():
case ch <- []*config.TargetGroup{tg}:
}
}
s.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
AddFunc: func(o interface{}) {
eventCount.WithLabelValues("ingress", "add").Inc()
ingress, err := convertToIngress(o)
if err != nil {
s.logger.With("err", err).Errorln("converting to Ingress object failed")
return
}
send(s.buildIngress(ingress))
},
DeleteFunc: func(o interface{}) {
eventCount.WithLabelValues("ingress", "delete").Inc()
ingress, err := convertToIngress(o)
if err != nil {
s.logger.With("err", err).Errorln("converting to Ingress object failed")
return
}
send(&config.TargetGroup{Source: ingressSource(ingress)})
},
UpdateFunc: func(_, o interface{}) {
eventCount.WithLabelValues("ingress", "update").Inc()
ingress, err := convertToIngress(o)
if err != nil {
s.logger.With("err", err).Errorln("converting to Ingress object failed")
return
}
send(s.buildIngress(ingress))
},
})
// Block until the target provider is explicitly canceled.
<-ctx.Done()
}
func convertToIngress(o interface{}) (*v1beta1.Ingress, error) {
ingress, ok := o.(*v1beta1.Ingress)
if ok {
return ingress, nil
}
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
if !ok {
return nil, fmt.Errorf("Received unexpected object: %v", o)
}
ingress, ok = deletedState.Obj.(*v1beta1.Ingress)
if !ok {
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Ingress object: %v", deletedState.Obj)
}
return ingress, nil
}
func ingressSource(s *v1beta1.Ingress) string {
return "ingress/" + s.Namespace + "/" + s.Name
}
const (
ingressNameLabel = metaLabelPrefix + "ingress_name"
ingressLabelPrefix = metaLabelPrefix + "ingress_label_"
ingressAnnotationPrefix = metaLabelPrefix + "ingress_annotation_"
ingressSchemeLabel = metaLabelPrefix + "ingress_scheme"
ingressHostLabel = metaLabelPrefix + "ingress_host"
ingressPathLabel = metaLabelPrefix + "ingress_path"
)
func ingressLabels(ingress *v1beta1.Ingress) model.LabelSet {
ls := make(model.LabelSet, len(ingress.Labels)+len(ingress.Annotations)+2)
ls[ingressNameLabel] = lv(ingress.Name)
ls[namespaceLabel] = lv(ingress.Namespace)
for k, v := range ingress.Labels {
ln := strutil.SanitizeLabelName(ingressLabelPrefix + k)
ls[model.LabelName(ln)] = lv(v)
}
for k, v := range ingress.Annotations {
ln := strutil.SanitizeLabelName(ingressAnnotationPrefix + k)
ls[model.LabelName(ln)] = lv(v)
}
return ls
}
func pathsFromIngressRule(rv *v1beta1.IngressRuleValue) []string {
if rv.HTTP == nil {
return []string{"/"}
}
paths := make([]string, len(rv.HTTP.Paths))
for n, p := range rv.HTTP.Paths {
path := p.Path
if path == "" {
path = "/"
}
paths[n] = path
}
return paths
}
func (s *Ingress) buildIngress(ingress *v1beta1.Ingress) *config.TargetGroup {
tg := &config.TargetGroup{
Source: ingressSource(ingress),
}
tg.Labels = ingressLabels(ingress)
schema := "http"
if ingress.Spec.TLS != nil {
schema = "https"
}
for _, rule := range ingress.Spec.Rules {
paths := pathsFromIngressRule(&rule.IngressRuleValue)
for _, path := range paths {
tg.Targets = append(tg.Targets, model.LabelSet{
model.AddressLabel: lv(rule.Host),
ingressSchemeLabel: lv(schema),
ingressHostLabel: lv(rule.Host),
ingressPathLabel: lv(path),
})
}
}
return tg
}

View file

@ -0,0 +1,137 @@
// Copyright 2016 The Prometheus 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 kubernetes
import (
"testing"
"github.com/prometheus/common/log"
"github.com/prometheus/common/model"
"github.com/prometheus/prometheus/config"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/pkg/apis/extensions/v1beta1"
)
func ingressStoreKeyFunc(obj interface{}) (string, error) {
return obj.(*v1beta1.Ingress).ObjectMeta.Name, nil
}
func newFakeIngressInformer() *fakeInformer {
return newFakeInformer(ingressStoreKeyFunc)
}
func makeTestIngressDiscovery() (*Ingress, *fakeInformer) {
i := newFakeIngressInformer()
return NewIngress(log.Base(), i), i
}
func makeIngress(tls []v1beta1.IngressTLS) *v1beta1.Ingress {
return &v1beta1.Ingress{
ObjectMeta: metav1.ObjectMeta{
Name: "testingress",
Namespace: "default",
Labels: map[string]string{"testlabel": "testvalue"},
Annotations: map[string]string{"testannotation": "testannotationvalue"},
},
Spec: v1beta1.IngressSpec{
TLS: tls,
Rules: []v1beta1.IngressRule{
{
Host: "example.com",
IngressRuleValue: v1beta1.IngressRuleValue{
HTTP: &v1beta1.HTTPIngressRuleValue{
Paths: []v1beta1.HTTPIngressPath{
{Path: "/"},
{Path: "/foo"},
},
},
},
},
{
// No backend config, ignored
Host: "nobackend.example.com",
IngressRuleValue: v1beta1.IngressRuleValue{
HTTP: &v1beta1.HTTPIngressRuleValue{},
},
},
{
Host: "test.example.com",
IngressRuleValue: v1beta1.IngressRuleValue{
HTTP: &v1beta1.HTTPIngressRuleValue{
Paths: []v1beta1.HTTPIngressPath{{}},
},
},
},
},
},
}
}
func expectedTargetGroups(tls bool) []*config.TargetGroup {
scheme := "http"
if tls {
scheme = "https"
}
return []*config.TargetGroup{
{
Targets: []model.LabelSet{
{
"__meta_kubernetes_ingress_scheme": lv(scheme),
"__meta_kubernetes_ingress_host": "example.com",
"__meta_kubernetes_ingress_path": "/",
"__address__": "example.com",
},
{
"__meta_kubernetes_ingress_scheme": lv(scheme),
"__meta_kubernetes_ingress_host": "example.com",
"__meta_kubernetes_ingress_path": "/foo",
"__address__": "example.com",
},
{
"__meta_kubernetes_ingress_scheme": lv(scheme),
"__meta_kubernetes_ingress_host": "test.example.com",
"__address__": "test.example.com",
"__meta_kubernetes_ingress_path": "/",
},
},
Labels: model.LabelSet{
"__meta_kubernetes_ingress_name": "testingress",
"__meta_kubernetes_namespace": "default",
"__meta_kubernetes_ingress_label_testlabel": "testvalue",
"__meta_kubernetes_ingress_annotation_testannotation": "testannotationvalue",
},
Source: "ingress/default/testingress",
},
}
}
func TestIngressDiscoveryInitial(t *testing.T) {
n, i := makeTestIngressDiscovery()
i.GetStore().Add(makeIngress(nil))
k8sDiscoveryTest{
discovery: n,
expectedInitial: expectedTargetGroups(false),
}.Run(t)
}
func TestIngressDiscoveryInitialTLS(t *testing.T) {
n, i := makeTestIngressDiscovery()
i.GetStore().Add(makeIngress([]v1beta1.IngressTLS{{}}))
k8sDiscoveryTest{
discovery: n,
expectedInitial: expectedTargetGroups(true),
}.Run(t)
}

View file

@ -28,6 +28,7 @@ import (
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/pkg/api"
apiv1 "k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/extensions"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
)
@ -159,6 +160,7 @@ const resyncPeriod = 10 * time.Minute
// Run implements the TargetProvider interface.
func (d *Discovery) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
rclient := d.client.Core().RESTClient()
reclient := d.client.Extensions().RESTClient()
namespaces := d.getNamespaces()
@ -236,6 +238,26 @@ func (d *Discovery) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
}()
}
wg.Wait()
case "ingress":
var wg sync.WaitGroup
for _, namespace := range namespaces {
ilw := cache.NewListWatchFromClient(reclient, "ingresses", namespace, nil)
ingress := NewIngress(
d.logger.With("kubernetes_sd", "ingress"),
cache.NewSharedInformer(ilw, &extensions.Ingress{}, resyncPeriod),
)
go ingress.informer.Run(ctx.Done())
for !ingress.informer.HasSynced() {
time.Sleep(100 * time.Millisecond)
}
wg.Add(1)
go func() {
defer wg.Done()
ingress.Run(ctx, ch)
}()
}
wg.Wait()
case "node":
nlw := cache.NewListWatchFromClient(rclient, "nodes", api.NamespaceAll, nil)
node := NewNode(

View file

@ -99,8 +99,11 @@ func (n *Node) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
}
func convertToNode(o interface{}) (*apiv1.Node, error) {
node, isNode := o.(*apiv1.Node)
if !isNode {
node, ok := o.(*apiv1.Node)
if ok {
return node, nil
}
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
if !ok {
return nil, fmt.Errorf("Received unexpected object: %v", o)
@ -109,8 +112,6 @@ func convertToNode(o interface{}) (*apiv1.Node, error) {
if !ok {
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Node object: %v", deletedState.Obj)
}
}
return node, nil
}
@ -126,7 +127,7 @@ const (
)
func nodeLabels(n *apiv1.Node) model.LabelSet {
ls := make(model.LabelSet, len(n.Labels)+len(n.Annotations)+2)
ls := make(model.LabelSet, len(n.Labels)+len(n.Annotations)+1)
ls[nodeNameLabel] = lv(n.Name)

View file

@ -107,8 +107,11 @@ func (p *Pod) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
}
func convertToPod(o interface{}) (*apiv1.Pod, error) {
pod, isPod := o.(*apiv1.Pod)
if !isPod {
pod, ok := o.(*apiv1.Pod)
if ok {
return pod, nil
}
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
if !ok {
return nil, fmt.Errorf("Received unexpected object: %v", o)
@ -117,8 +120,6 @@ func convertToPod(o interface{}) (*apiv1.Pod, error) {
if !ok {
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Pod object: %v", deletedState.Obj)
}
}
return pod, nil
}

View file

@ -98,8 +98,10 @@ func (s *Service) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
}
func convertToService(o interface{}) (*apiv1.Service, error) {
service, isService := o.(*apiv1.Service)
if !isService {
service, ok := o.(*apiv1.Service)
if ok {
return service, nil
}
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
if !ok {
return nil, fmt.Errorf("Received unexpected object: %v", o)
@ -108,8 +110,6 @@ func convertToService(o interface{}) (*apiv1.Service, error) {
if !ok {
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Service object: %v", deletedState.Obj)
}
}
return service, nil
}
@ -129,6 +129,7 @@ func serviceLabels(svc *apiv1.Service) model.LabelSet {
ls := make(model.LabelSet, len(svc.Labels)+len(svc.Annotations)+2)
ls[serviceNameLabel] = lv(svc.Name)
ls[namespaceLabel] = lv(svc.Namespace)
for k, v := range svc.Labels {
ln := strutil.SanitizeLabelName(serviceLabelPrefix + k)
@ -147,7 +148,6 @@ func (s *Service) buildService(svc *apiv1.Service) *config.TargetGroup {
Source: serviceSource(svc),
}
tg.Labels = serviceLabels(svc)
tg.Labels[namespaceLabel] = lv(svc.Namespace)
for _, port := range svc.Spec.Ports {
addr := net.JoinHostPort(svc.Name+"."+svc.Namespace+".svc", strconv.FormatInt(int64(port.Port), 10))

View file

@ -192,7 +192,7 @@ scrape_configs:
- source_labels: [__address__]
target_label: __param_target
- target_label: __address__
replacement: blackbox
replacement: blackbox-exporter.example.com:9115
- source_labels: [__param_target]
target_label: instance
- action: labelmap
@ -202,6 +202,40 @@ scrape_configs:
- source_labels: [__meta_kubernetes_service_name]
target_label: kubernetes_name
# Example scrape config for probing ingresses via the Blackbox Exporter.
#
# The relabeling allows the actual ingress scrape endpoint to be configured
# via the following annotations:
#
# * `prometheus.io/probe`: Only probe services that have a value of `true`
- job_name: 'kubernetes-ingresses'
metrics_path: /probe
params:
module: [http_2xx]
kubernetes_sd_configs:
- role: ingress
relabel_configs:
- source_labels: [__meta_kubernetes_ingress_annotation_prometheus_io_probe]
action: keep
regex: true
- source_labels: [__meta_kubernetes_ingress_scheme,__address__,__meta_kubernetes_ingress_path]
regex: (.+);(.+);(.+)
replacement: ${1}://${2}${3}
target_label: __param_target
- target_label: __address__
replacement: blackbox-exporter.example.com:9115
- source_labels: [__param_target]
target_label: instance
- action: labelmap
regex: __meta_kubernetes_ingress_label_(.+)
- source_labels: [__meta_kubernetes_namespace]
target_label: kubernetes_namespace
- source_labels: [__meta_kubernetes_ingress_name]
target_label: kubernetes_name
# Example scrape config for pods
#
# The relabeling allows the actual pod scrape endpoint to be configured via the