mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2024-12-15 16:30:55 +01:00
de991551f5
This should fix authroization issues described at https://github.com/VictoriaMetrics/VictoriaMetrics/issues/432
222 lines
6.5 KiB
Go
222 lines
6.5 KiB
Go
package kubernetes
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
|
|
)
|
|
|
|
// getEndpointsLabels returns labels for k8s endpoints obtained from the given cfg.
|
|
func getEndpointsLabels(cfg *apiConfig) ([]map[string]string, error) {
|
|
eps, err := getEndpoints(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
pods, err := getPods(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
svcs, err := getServices(cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var ms []map[string]string
|
|
for _, ep := range eps {
|
|
ms = ep.appendTargetLabels(ms, pods, svcs)
|
|
}
|
|
return ms, nil
|
|
}
|
|
|
|
func getEndpoints(cfg *apiConfig) ([]Endpoints, error) {
|
|
if len(cfg.Namespaces) == 0 {
|
|
return getEndpointsByPath(cfg, "/api/v1/endpoints")
|
|
}
|
|
// Query /api/v1/namespaces/* for each namespace.
|
|
// This fixes authorization issue at https://github.com/VictoriaMetrics/VictoriaMetrics/issues/432
|
|
cfgCopy := *cfg
|
|
namespaces := cfgCopy.Namespaces
|
|
cfgCopy.Namespaces = nil
|
|
cfg = &cfgCopy
|
|
var result []Endpoints
|
|
for _, ns := range namespaces {
|
|
path := fmt.Sprintf("/api/v1/namespaces/%s/endpoints", ns)
|
|
eps, err := getEndpointsByPath(cfg, path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, eps...)
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func getEndpointsByPath(cfg *apiConfig, path string) ([]Endpoints, error) {
|
|
data, err := getAPIResponse(cfg, "endpoints", path)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot obtain endpoints data from API server: %s", err)
|
|
}
|
|
epl, err := parseEndpointsList(data)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot parse endpoints response from API server: %s", err)
|
|
}
|
|
return epl.Items, nil
|
|
}
|
|
|
|
// EndpointsList implements k8s endpoints list.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointslist-v1-core
|
|
type EndpointsList struct {
|
|
Items []Endpoints
|
|
}
|
|
|
|
// Endpoints implements k8s endpoints.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpoints-v1-core
|
|
type Endpoints struct {
|
|
Metadata ObjectMeta
|
|
Subsets []EndpointSubset
|
|
}
|
|
|
|
// EndpointSubset implements k8s endpoint subset.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointsubset-v1-core
|
|
type EndpointSubset struct {
|
|
Addresses []EndpointAddress
|
|
NotReadyAddresses []EndpointAddress
|
|
Ports []EndpointPort
|
|
}
|
|
|
|
// EndpointAddress implements k8s endpoint address.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointaddress-v1-core
|
|
type EndpointAddress struct {
|
|
Hostname string
|
|
IP string
|
|
NodeName string
|
|
TargetRef ObjectReference
|
|
}
|
|
|
|
// ObjectReference implements k8s object reference.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#objectreference-v1-core
|
|
type ObjectReference struct {
|
|
Kind string
|
|
Name string
|
|
Namespace string
|
|
}
|
|
|
|
// EndpointPort implements k8s endpoint port.
|
|
//
|
|
// See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointport-v1beta1-discovery-k8s-io
|
|
type EndpointPort struct {
|
|
AppProtocol string
|
|
Name string
|
|
Port int
|
|
Protocol string
|
|
}
|
|
|
|
// parseEndpointsList parses EndpointsList from data.
|
|
func parseEndpointsList(data []byte) (*EndpointsList, error) {
|
|
var esl EndpointsList
|
|
if err := json.Unmarshal(data, &esl); err != nil {
|
|
return nil, fmt.Errorf("cannot unmarshal EndpointsList from %q: %s", data, err)
|
|
}
|
|
return &esl, nil
|
|
}
|
|
|
|
// appendTargetLabels appends labels for each endpoint in eps to ms and returns the result.
|
|
//
|
|
// See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#endpoints
|
|
func (eps *Endpoints) appendTargetLabels(ms []map[string]string, pods []Pod, svcs []Service) []map[string]string {
|
|
svc := getService(svcs, eps.Metadata.Namespace, eps.Metadata.Name)
|
|
podPortsSeen := make(map[*Pod][]int)
|
|
for _, ess := range eps.Subsets {
|
|
for _, epp := range ess.Ports {
|
|
ms = appendEndpointLabelsForAddresses(ms, podPortsSeen, eps, ess.Addresses, epp, pods, svc, "true")
|
|
ms = appendEndpointLabelsForAddresses(ms, podPortsSeen, eps, ess.NotReadyAddresses, epp, pods, svc, "false")
|
|
}
|
|
}
|
|
|
|
// Append labels for skipped ports on seen pods.
|
|
portSeen := func(port int, ports []int) bool {
|
|
for _, p := range ports {
|
|
if p == port {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
for p, ports := range podPortsSeen {
|
|
for _, c := range p.Spec.Containers {
|
|
for _, cp := range c.Ports {
|
|
if portSeen(cp.ContainerPort, ports) {
|
|
continue
|
|
}
|
|
addr := discoveryutils.JoinHostPort(p.Status.PodIP, cp.ContainerPort)
|
|
m := map[string]string{
|
|
"__address__": addr,
|
|
}
|
|
p.appendCommonLabels(m)
|
|
p.appendContainerLabels(m, c, &cp)
|
|
ms = append(ms, m)
|
|
}
|
|
}
|
|
}
|
|
return ms
|
|
}
|
|
|
|
func appendEndpointLabelsForAddresses(ms []map[string]string, podPortsSeen map[*Pod][]int, eps *Endpoints, eas []EndpointAddress, epp EndpointPort,
|
|
pods []Pod, svc *Service, ready string) []map[string]string {
|
|
for _, ea := range eas {
|
|
p := getPod(pods, ea.TargetRef.Namespace, ea.TargetRef.Name)
|
|
m := getEndpointLabelsForAddressAndPort(podPortsSeen, eps, ea, epp, p, svc, ready)
|
|
ms = append(ms, m)
|
|
}
|
|
return ms
|
|
}
|
|
|
|
func getEndpointLabelsForAddressAndPort(podPortsSeen map[*Pod][]int, eps *Endpoints, ea EndpointAddress, epp EndpointPort, p *Pod, svc *Service, ready string) map[string]string {
|
|
m := getEndpointLabels(eps.Metadata, ea, epp, ready)
|
|
if svc != nil {
|
|
svc.appendCommonLabels(m)
|
|
}
|
|
if ea.TargetRef.Kind != "Pod" || p == nil {
|
|
return m
|
|
}
|
|
p.appendCommonLabels(m)
|
|
for _, c := range p.Spec.Containers {
|
|
for _, cp := range c.Ports {
|
|
if cp.ContainerPort == epp.Port {
|
|
p.appendContainerLabels(m, c, &cp)
|
|
podPortsSeen[p] = append(podPortsSeen[p], cp.ContainerPort)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
return m
|
|
}
|
|
|
|
func getEndpointLabels(om ObjectMeta, ea EndpointAddress, epp EndpointPort, ready string) map[string]string {
|
|
addr := discoveryutils.JoinHostPort(ea.IP, epp.Port)
|
|
m := map[string]string{
|
|
"__address__": addr,
|
|
"__meta_kubernetes_namespace": om.Namespace,
|
|
"__meta_kubernetes_endpoints_name": om.Name,
|
|
|
|
"__meta_kubernetes_endpoint_ready": ready,
|
|
"__meta_kubernetes_endpoint_port_name": epp.Name,
|
|
"__meta_kubernetes_endpoint_port_protocol": epp.Protocol,
|
|
}
|
|
if ea.TargetRef.Kind != "" {
|
|
m["__meta_kubernetes_endpoint_address_target_kind"] = ea.TargetRef.Kind
|
|
m["__meta_kubernetes_endpoint_address_target_name"] = ea.TargetRef.Name
|
|
}
|
|
if ea.NodeName != "" {
|
|
m["__meta_kubernetes_endpoint_node_name"] = ea.NodeName
|
|
}
|
|
if ea.Hostname != "" {
|
|
m["__meta_kubernetes_endpoint_hostname"] = ea.Hostname
|
|
}
|
|
return m
|
|
}
|