2016-09-30 02:33:23 -07:00
|
|
|
// 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.
|
|
|
|
|
2016-10-07 05:53:11 -07:00
|
|
|
package kubernetes
|
2016-09-30 01:42:07 -07:00
|
|
|
|
|
|
|
import (
|
|
|
|
"fmt"
|
|
|
|
"net"
|
|
|
|
"strconv"
|
|
|
|
|
|
|
|
"github.com/prometheus/prometheus/config"
|
|
|
|
|
|
|
|
"github.com/prometheus/common/log"
|
|
|
|
"github.com/prometheus/common/model"
|
|
|
|
"golang.org/x/net/context"
|
2016-09-30 02:33:23 -07:00
|
|
|
apiv1 "k8s.io/client-go/1.5/pkg/api/v1"
|
|
|
|
"k8s.io/client-go/1.5/tools/cache"
|
2016-09-30 01:42:07 -07:00
|
|
|
)
|
|
|
|
|
|
|
|
// Endpoints discovers new endpoint targets.
|
|
|
|
type Endpoints struct {
|
|
|
|
logger log.Logger
|
|
|
|
|
|
|
|
endpointsInf cache.SharedInformer
|
2016-09-30 05:18:49 -07:00
|
|
|
serviceInf cache.SharedInformer
|
|
|
|
podInf cache.SharedInformer
|
2016-09-30 01:42:07 -07:00
|
|
|
|
|
|
|
podStore cache.Store
|
|
|
|
endpointsStore cache.Store
|
2016-09-30 05:18:49 -07:00
|
|
|
serviceStore cache.Store
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
// NewEndpoints returns a new endpoints discovery.
|
|
|
|
func NewEndpoints(l log.Logger, svc, eps, pod cache.SharedInformer) *Endpoints {
|
|
|
|
ep := &Endpoints{
|
|
|
|
logger: l,
|
|
|
|
endpointsInf: eps,
|
|
|
|
endpointsStore: eps.GetStore(),
|
2016-09-30 05:18:49 -07:00
|
|
|
serviceInf: svc,
|
|
|
|
serviceStore: svc.GetStore(),
|
|
|
|
podInf: pod,
|
2016-09-30 01:42:07 -07:00
|
|
|
podStore: pod.GetStore(),
|
|
|
|
}
|
|
|
|
|
|
|
|
return ep
|
|
|
|
}
|
|
|
|
|
|
|
|
// Run implements the retrieval.TargetProvider interface.
|
|
|
|
func (e *Endpoints) Run(ctx context.Context, ch chan<- []*config.TargetGroup) {
|
|
|
|
// Send full initial set of endpoint targets.
|
|
|
|
var initial []*config.TargetGroup
|
|
|
|
|
|
|
|
for _, o := range e.endpointsStore.List() {
|
|
|
|
tg := e.buildEndpoints(o.(*apiv1.Endpoints))
|
|
|
|
initial = append(initial, tg)
|
|
|
|
}
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
return
|
|
|
|
case ch <- initial:
|
|
|
|
}
|
|
|
|
// Send target groups for pod updates.
|
|
|
|
send := func(tg *config.TargetGroup) {
|
|
|
|
if tg == nil {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
e.logger.With("tg", fmt.Sprintf("%#v", tg)).Debugln("endpoints update")
|
|
|
|
select {
|
|
|
|
case <-ctx.Done():
|
|
|
|
case ch <- []*config.TargetGroup{tg}:
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
e.endpointsInf.AddEventHandler(cache.ResourceEventHandlerFuncs{
|
|
|
|
AddFunc: func(o interface{}) {
|
2016-11-14 07:21:38 -08:00
|
|
|
eps, err := convertToEndpoints(o)
|
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("converting to Endpoints object failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
send(e.buildEndpoints(eps))
|
2016-09-30 01:42:07 -07:00
|
|
|
},
|
|
|
|
UpdateFunc: func(_, o interface{}) {
|
2016-11-14 07:21:38 -08:00
|
|
|
eps, err := convertToEndpoints(o)
|
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("converting to Endpoints object failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
send(e.buildEndpoints(eps))
|
2016-09-30 01:42:07 -07:00
|
|
|
},
|
|
|
|
DeleteFunc: func(o interface{}) {
|
2016-11-14 07:21:38 -08:00
|
|
|
eps, err := convertToEndpoints(o)
|
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("converting to Endpoints object failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
send(&config.TargetGroup{Source: endpointsSource(eps)})
|
2016-09-30 01:42:07 -07:00
|
|
|
},
|
|
|
|
})
|
|
|
|
|
2016-11-14 07:21:38 -08:00
|
|
|
serviceUpdate := func(o interface{}) {
|
|
|
|
svc, err := convertToService(o)
|
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("converting to Service object failed")
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-09-30 01:42:07 -07:00
|
|
|
ep := &apiv1.Endpoints{}
|
|
|
|
ep.Namespace = svc.Namespace
|
|
|
|
ep.Name = svc.Name
|
|
|
|
obj, exists, err := e.endpointsStore.Get(ep)
|
|
|
|
if exists && err != nil {
|
|
|
|
send(e.buildEndpoints(obj.(*apiv1.Endpoints)))
|
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("retrieving endpoints failed")
|
|
|
|
}
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
e.serviceInf.AddEventHandler(cache.ResourceEventHandlerFuncs{
|
2016-10-17 02:05:13 -07:00
|
|
|
// TODO(fabxc): potentially remove add and delete event handlers. Those should
|
|
|
|
// be triggered via the endpoint handlers already.
|
2016-11-14 07:21:38 -08:00
|
|
|
AddFunc: func(o interface{}) { serviceUpdate(o) },
|
|
|
|
UpdateFunc: func(_, o interface{}) { serviceUpdate(o) },
|
|
|
|
DeleteFunc: func(o interface{}) { serviceUpdate(o) },
|
2016-09-30 01:42:07 -07:00
|
|
|
})
|
|
|
|
|
|
|
|
// Block until the target provider is explicitly canceled.
|
|
|
|
<-ctx.Done()
|
|
|
|
}
|
|
|
|
|
2016-11-14 07:21:38 -08:00
|
|
|
func convertToEndpoints(o interface{}) (*apiv1.Endpoints, error) {
|
|
|
|
endpoints, isEndpoints := o.(*apiv1.Endpoints)
|
|
|
|
if !isEndpoints {
|
|
|
|
deletedState, ok := o.(cache.DeletedFinalStateUnknown)
|
|
|
|
if !ok {
|
|
|
|
return nil, fmt.Errorf("Received unexpected object: %v", o)
|
|
|
|
}
|
|
|
|
endpoints, ok = deletedState.Obj.(*apiv1.Endpoints)
|
|
|
|
if !ok {
|
|
|
|
return nil, fmt.Errorf("DeletedFinalStateUnknown contained non-Endpoints object: %v", deletedState.Obj)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return endpoints, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func endpointsSource(ep *apiv1.Endpoints) string {
|
|
|
|
return "endpoints/" + ep.ObjectMeta.Namespace + "/" + ep.ObjectMeta.Name
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
const (
|
2016-09-30 05:18:49 -07:00
|
|
|
endpointsNameLabel = metaLabelPrefix + "endpoints_name"
|
|
|
|
endpointReadyLabel = metaLabelPrefix + "endpoint_ready"
|
|
|
|
endpointPortNameLabel = metaLabelPrefix + "endpoint_port_name"
|
|
|
|
endpointPortProtocolLabel = metaLabelPrefix + "endpoint_port_protocol"
|
2016-09-30 01:42:07 -07:00
|
|
|
)
|
|
|
|
|
|
|
|
func (e *Endpoints) buildEndpoints(eps *apiv1.Endpoints) *config.TargetGroup {
|
|
|
|
if len(eps.Subsets) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
tg := &config.TargetGroup{
|
2016-11-14 07:21:38 -08:00
|
|
|
Source: endpointsSource(eps),
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
|
|
|
tg.Labels = model.LabelSet{
|
|
|
|
namespaceLabel: lv(eps.Namespace),
|
|
|
|
endpointsNameLabel: lv(eps.Name),
|
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
e.addServiceLabels(eps.Namespace, eps.Name, tg)
|
2016-09-30 01:42:07 -07:00
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
type podEntry struct {
|
|
|
|
pod *apiv1.Pod
|
|
|
|
servicePorts []apiv1.EndpointPort
|
|
|
|
}
|
|
|
|
seenPods := map[string]*podEntry{}
|
2016-09-30 01:42:07 -07:00
|
|
|
|
|
|
|
add := func(addr apiv1.EndpointAddress, port apiv1.EndpointPort, ready string) {
|
2016-10-17 02:05:13 -07:00
|
|
|
a := net.JoinHostPort(addr.IP, strconv.FormatUint(uint64(port.Port), 10))
|
2016-09-30 01:42:07 -07:00
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
target := model.LabelSet{
|
|
|
|
model.AddressLabel: lv(a),
|
|
|
|
endpointPortNameLabel: lv(port.Name),
|
|
|
|
endpointPortProtocolLabel: lv(string(port.Protocol)),
|
|
|
|
endpointReadyLabel: lv(ready),
|
|
|
|
}
|
|
|
|
|
|
|
|
pod := e.resolvePodRef(addr.TargetRef)
|
|
|
|
if pod == nil {
|
2016-10-14 08:16:06 -07:00
|
|
|
// This target is not a Pod, so don't continue with Pod specific logic.
|
2016-09-30 05:18:49 -07:00
|
|
|
tg.Targets = append(tg.Targets, target)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
s := pod.Namespace + "/" + pod.Name
|
|
|
|
|
|
|
|
sp, ok := seenPods[s]
|
|
|
|
if !ok {
|
|
|
|
sp = &podEntry{pod: pod}
|
|
|
|
seenPods[s] = sp
|
|
|
|
}
|
|
|
|
|
|
|
|
// Attach standard pod labels.
|
|
|
|
target = target.Merge(podLabels(pod))
|
|
|
|
|
|
|
|
// Attach potential container port labels matching the endpoint port.
|
|
|
|
for _, c := range pod.Spec.Containers {
|
|
|
|
for _, cport := range c.Ports {
|
|
|
|
if port.Port == cport.ContainerPort {
|
2016-10-17 02:05:13 -07:00
|
|
|
ports := strconv.FormatUint(uint64(port.Port), 10)
|
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
target[podContainerNameLabel] = lv(c.Name)
|
2016-10-17 02:05:13 -07:00
|
|
|
target[podContainerPortNameLabel] = lv(cport.Name)
|
|
|
|
target[podContainerPortNumberLabel] = lv(ports)
|
2016-09-30 05:18:49 -07:00
|
|
|
target[podContainerPortProtocolLabel] = lv(string(port.Protocol))
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Add service port so we know that we have already generated a target
|
|
|
|
// for it.
|
|
|
|
sp.servicePorts = append(sp.servicePorts, port)
|
|
|
|
tg.Targets = append(tg.Targets, target)
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
for _, ss := range eps.Subsets {
|
|
|
|
for _, port := range ss.Ports {
|
|
|
|
for _, addr := range ss.Addresses {
|
|
|
|
add(addr, port, "true")
|
|
|
|
}
|
2016-10-14 08:16:06 -07:00
|
|
|
// Although this generates the same target again, as it was generated in
|
|
|
|
// the loop above, it causes the ready meta label to be overridden.
|
2016-09-30 01:42:07 -07:00
|
|
|
for _, addr := range ss.NotReadyAddresses {
|
|
|
|
add(addr, port, "false")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
// For all seen pods, check all container ports. If they were not covered
|
|
|
|
// by one of the service endpoints, generate targets for them.
|
|
|
|
for _, pe := range seenPods {
|
|
|
|
for _, c := range pe.pod.Spec.Containers {
|
|
|
|
for _, cport := range c.Ports {
|
|
|
|
hasSeenPort := func() bool {
|
|
|
|
for _, eport := range pe.servicePorts {
|
|
|
|
if cport.ContainerPort == eport.Port {
|
|
|
|
return true
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
}
|
|
|
|
if hasSeenPort() {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-10-17 02:05:13 -07:00
|
|
|
a := net.JoinHostPort(pe.pod.Status.PodIP, strconv.FormatUint(uint64(cport.ContainerPort), 10))
|
|
|
|
ports := strconv.FormatUint(uint64(cport.ContainerPort), 10)
|
2016-09-30 05:18:49 -07:00
|
|
|
|
|
|
|
target := model.LabelSet{
|
|
|
|
model.AddressLabel: lv(a),
|
|
|
|
podContainerNameLabel: lv(c.Name),
|
|
|
|
podContainerPortNameLabel: lv(cport.Name),
|
2016-10-17 02:05:13 -07:00
|
|
|
podContainerPortNumberLabel: lv(ports),
|
2016-09-30 05:18:49 -07:00
|
|
|
podContainerPortProtocolLabel: lv(string(cport.Protocol)),
|
|
|
|
}
|
|
|
|
tg.Targets = append(tg.Targets, target.Merge(podLabels(pe.pod)))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-09-30 01:42:07 -07:00
|
|
|
return tg
|
|
|
|
}
|
|
|
|
|
|
|
|
func (e *Endpoints) resolvePodRef(ref *apiv1.ObjectReference) *apiv1.Pod {
|
2016-09-30 05:18:49 -07:00
|
|
|
if ref == nil || ref.Kind != "Pod" {
|
2016-09-30 01:42:07 -07:00
|
|
|
return nil
|
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
p := &apiv1.Pod{}
|
|
|
|
p.Namespace = ref.Namespace
|
|
|
|
p.Name = ref.Name
|
|
|
|
|
|
|
|
obj, exists, err := e.podStore.Get(p)
|
2016-09-30 01:42:07 -07:00
|
|
|
if err != nil || !exists {
|
|
|
|
return nil
|
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("resolving pod ref failed")
|
|
|
|
}
|
|
|
|
return obj.(*apiv1.Pod)
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
func (e *Endpoints) addServiceLabels(ns, name string, tg *config.TargetGroup) {
|
2016-09-30 01:42:07 -07:00
|
|
|
svc := &apiv1.Service{}
|
|
|
|
svc.Namespace = ns
|
|
|
|
svc.Name = name
|
|
|
|
|
2016-09-30 05:18:49 -07:00
|
|
|
obj, exists, err := e.serviceStore.Get(svc)
|
2016-09-30 01:42:07 -07:00
|
|
|
if !exists || err != nil {
|
|
|
|
return
|
|
|
|
}
|
2016-09-30 05:18:49 -07:00
|
|
|
if err != nil {
|
|
|
|
e.logger.With("err", err).Errorln("retrieving service failed")
|
|
|
|
}
|
2016-09-30 01:42:07 -07:00
|
|
|
svc = obj.(*apiv1.Service)
|
|
|
|
|
2016-10-06 07:28:59 -07:00
|
|
|
tg.Labels = tg.Labels.Merge(serviceLabels(svc))
|
2016-09-30 01:42:07 -07:00
|
|
|
}
|