2021-03-05 11:36:05 +01:00
package kubernetes
import (
"encoding/json"
"errors"
2021-03-11 15:41:09 +01:00
"flag"
2021-03-05 11:36:05 +01:00
"fmt"
"io"
"io/ioutil"
"net/http"
"net/url"
"reflect"
"strconv"
"strings"
"sync"
2021-08-29 10:16:40 +02:00
"sync/atomic"
2021-03-05 11:36:05 +01:00
"time"
2022-04-20 15:09:40 +02:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
2021-03-05 11:36:05 +01:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
2021-03-11 15:41:09 +01:00
"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
2021-03-05 11:36:05 +01:00
"github.com/VictoriaMetrics/metrics"
)
2022-04-20 09:50:36 +02:00
var apiServerTimeout = flag . Duration ( "promscrape.kubernetes.apiServerTimeout" , 30 * time . Minute , "How frequently to reload the full state from Kubernetes API server" )
2021-03-11 15:41:09 +01:00
2021-03-05 11:36:05 +01:00
// WatchEvent is a watch event returned from API server endpoints if `watch=1` query arg is set.
//
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
type WatchEvent struct {
Type string
Object json . RawMessage
}
// object is any Kubernetes object.
type object interface {
2021-04-02 13:45:08 +02:00
key ( ) string
2021-04-29 09:14:24 +02:00
// getTargetLabels must be called under gw.mu lock.
2021-03-14 20:10:35 +01:00
getTargetLabels ( gw * groupWatcher ) [ ] map [ string ] string
2021-03-05 11:36:05 +01:00
}
// parseObjectFunc must parse object from the given data.
type parseObjectFunc func ( data [ ] byte ) ( object , error )
2021-03-11 15:41:09 +01:00
// parseObjectListFunc must parse objectList from the given r.
type parseObjectListFunc func ( r io . Reader ) ( map [ string ] object , ListMeta , error )
2021-03-05 11:36:05 +01:00
2022-04-20 09:50:36 +02:00
// apiWatcher is used for watching for Kubernetes object changes and caching their latest states.
2021-03-05 11:36:05 +01:00
type apiWatcher struct {
2021-03-14 20:10:35 +01:00
role string
2021-03-05 11:36:05 +01:00
2021-03-11 15:41:09 +01:00
// Constructor for creating ScrapeWork objects from labels
2021-03-05 11:36:05 +01:00
swcFunc ScrapeWorkConstructorFunc
2021-03-14 20:10:35 +01:00
gw * groupWatcher
2021-05-17 22:45:20 +02:00
// swos contains per-urlWatcher maps of ScrapeWork objects for the given apiWatcher
swosByURLWatcher map [ * urlWatcher ] map [ string ] [ ] interface { }
swosByURLWatcherLock sync . Mutex
2021-03-05 11:36:05 +01:00
2021-03-14 20:10:35 +01:00
swosCount * metrics . Counter
2021-03-05 11:36:05 +01:00
}
2021-03-11 15:41:09 +01:00
func newAPIWatcher ( apiServer string , ac * promauth . Config , sdc * SDConfig , swcFunc ScrapeWorkConstructorFunc ) * apiWatcher {
2021-03-14 20:10:35 +01:00
namespaces := sdc . Namespaces . Names
2022-01-13 21:44:14 +01:00
if len ( namespaces ) == 0 {
if sdc . Namespaces . OwnNamespace {
namespace , err := ioutil . ReadFile ( "/var/run/secrets/kubernetes.io/serviceaccount/namespace" )
if err != nil {
logger . Fatalf ( "cannot determine namespace for the current pod according to `own_namespace: true` option in kubernetes_sd_config: %s" , err )
}
namespaces = [ ] string { string ( namespace ) }
}
}
2021-03-14 20:10:35 +01:00
selectors := sdc . Selectors
proxyURL := sdc . ProxyURL . URL ( )
gw := getGroupWatcher ( apiServer , ac , namespaces , selectors , proxyURL )
2021-08-29 11:34:55 +02:00
role := sdc . role ( )
2021-03-05 11:36:05 +01:00
return & apiWatcher {
2021-08-29 11:34:55 +02:00
role : role ,
2021-05-17 22:45:20 +02:00
swcFunc : swcFunc ,
gw : gw ,
swosByURLWatcher : make ( map [ * urlWatcher ] map [ string ] [ ] interface { } ) ,
2021-08-29 11:34:55 +02:00
swosCount : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_scrape_works { role=%q} ` , role ) ) ,
2021-03-05 11:36:05 +01:00
}
}
2021-04-05 21:02:09 +02:00
func ( aw * apiWatcher ) mustStart ( ) {
2021-05-05 11:21:39 +02:00
aw . gw . startWatchersForRole ( aw . role , aw )
2021-04-05 21:02:09 +02:00
}
2021-03-11 15:41:09 +01:00
func ( aw * apiWatcher ) mustStop ( ) {
2021-03-14 20:10:35 +01:00
aw . gw . unsubscribeAPIWatcher ( aw )
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcherLock . Lock ( )
2021-05-18 14:29:32 +02:00
for _ , swosByKey := range aw . swosByURLWatcher {
aw . swosCount . Add ( - len ( swosByKey ) )
}
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcher = make ( map [ * urlWatcher ] map [ string ] [ ] interface { } )
aw . swosByURLWatcherLock . Unlock ( )
2021-03-11 15:41:09 +01:00
}
2021-05-17 22:45:20 +02:00
func ( aw * apiWatcher ) reloadScrapeWorks ( uw * urlWatcher , swosByKey map [ string ] [ ] interface { } ) {
aw . swosByURLWatcherLock . Lock ( )
aw . swosCount . Add ( len ( swosByKey ) - len ( aw . swosByURLWatcher [ uw ] ) )
aw . swosByURLWatcher [ uw ] = swosByKey
aw . swosByURLWatcherLock . Unlock ( )
2021-03-11 15:41:09 +01:00
}
2021-05-17 22:45:20 +02:00
func ( aw * apiWatcher ) setScrapeWorks ( uw * urlWatcher , key string , labels [ ] map [ string ] string ) {
2022-04-20 15:11:37 +02:00
swos := getScrapeWorkObjectsForLabels ( aw . swcFunc , labels )
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcherLock . Lock ( )
swosByKey := aw . swosByURLWatcher [ uw ]
2021-04-02 13:45:08 +02:00
if swosByKey == nil {
swosByKey = make ( map [ string ] [ ] interface { } )
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcher [ uw ] = swosByKey
2021-04-02 13:17:53 +02:00
}
2021-05-18 14:29:32 +02:00
aw . swosCount . Add ( len ( swos ) - len ( swosByKey [ key ] ) )
2021-03-11 15:41:09 +01:00
if len ( swos ) > 0 {
2021-04-02 13:45:08 +02:00
swosByKey [ key ] = swos
2021-03-11 15:41:09 +01:00
} else {
2021-04-02 13:45:08 +02:00
delete ( swosByKey , key )
2021-03-11 15:41:09 +01:00
}
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcherLock . Unlock ( )
2021-03-11 15:41:09 +01:00
}
2021-05-17 22:45:20 +02:00
func ( aw * apiWatcher ) removeScrapeWorks ( uw * urlWatcher , key string ) {
aw . swosByURLWatcherLock . Lock ( )
swosByKey := aw . swosByURLWatcher [ uw ]
2021-04-02 13:45:08 +02:00
if len ( swosByKey ) > 0 {
aw . swosCount . Add ( - len ( swosByKey [ key ] ) )
delete ( swosByKey , key )
2021-04-02 13:17:53 +02:00
}
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcherLock . Unlock ( )
2021-03-11 15:41:09 +01:00
}
2022-04-20 15:11:37 +02:00
func getScrapeWorkObjectsForLabels ( swcFunc ScrapeWorkConstructorFunc , labelss [ ] map [ string ] string ) [ ] interface { } {
2021-03-11 15:41:09 +01:00
swos := make ( [ ] interface { } , 0 , len ( labelss ) )
for _ , labels := range labelss {
2022-04-20 15:11:37 +02:00
swo := swcFunc ( labels )
2021-03-11 15:41:09 +01:00
// The reflect check is needed because of https://mangatmodi.medium.com/go-check-nil-interface-the-right-way-d142776edef1
if swo != nil && ! reflect . ValueOf ( swo ) . IsNil ( ) {
swos = append ( swos , swo )
2021-03-05 11:36:05 +01:00
}
}
2021-03-11 15:41:09 +01:00
return swos
}
// getScrapeWorkObjects returns all the ScrapeWork objects for the given aw.
func ( aw * apiWatcher ) getScrapeWorkObjects ( ) [ ] interface { } {
2021-04-05 21:02:09 +02:00
aw . gw . registerPendingAPIWatchers ( )
2022-04-20 15:11:37 +02:00
swos := make ( [ ] interface { } , 0 , aw . swosCount . Get ( ) )
2021-05-17 22:45:20 +02:00
aw . swosByURLWatcherLock . Lock ( )
for _ , swosByKey := range aw . swosByURLWatcher {
2021-04-02 13:45:08 +02:00
for _ , swosLocal := range swosByKey {
2021-04-02 13:17:53 +02:00
swos = append ( swos , swosLocal ... )
}
2021-03-11 15:41:09 +01:00
}
2022-04-20 15:11:37 +02:00
aw . swosByURLWatcherLock . Unlock ( )
2021-03-05 11:36:05 +01:00
return swos
}
2021-03-14 20:10:35 +01:00
// groupWatcher watches for Kubernetes objects on the given apiServer with the given namespaces,
2021-05-14 19:00:05 +02:00
// selectors using the given client.
2021-03-14 20:10:35 +01:00
type groupWatcher struct {
2021-08-29 10:16:40 +02:00
// Old Kubernetes doesn't support /apis/networking.k8s.io/v1/, so /apis/networking.k8s.io/v1beta1/ must be used instead.
// This flag is used for automatic substitution of v1 API path with v1beta1 API path during requests to apiServer.
useNetworkingV1Beta1 uint32
// Old Kubernetes doesn't support /apis/discovery.k8s.io/v1/, so discovery.k8s.io/v1beta1/ must be used instead.
// This flag is used for automatic substitution of v1 API path with v1beta1 API path during requests to apiServer.
useDiscoveryV1Beta1 uint32
2021-03-14 20:10:35 +01:00
apiServer string
namespaces [ ] string
selectors [ ] Selector
2021-05-14 19:00:05 +02:00
getAuthHeader func ( ) string
2021-03-14 20:10:35 +01:00
client * http . Client
mu sync . Mutex
m map [ string ] * urlWatcher
}
func newGroupWatcher ( apiServer string , ac * promauth . Config , namespaces [ ] string , selectors [ ] Selector , proxyURL * url . URL ) * groupWatcher {
var proxy func ( * http . Request ) ( * url . URL , error )
if proxyURL != nil {
proxy = http . ProxyURL ( proxyURL )
}
client := & http . Client {
Transport : & http . Transport {
TLSClientConfig : ac . NewTLSConfig ( ) ,
Proxy : proxy ,
TLSHandshakeTimeout : 10 * time . Second ,
IdleConnTimeout : * apiServerTimeout ,
2021-05-14 17:10:19 +02:00
MaxIdleConnsPerHost : 100 ,
2021-03-14 20:10:35 +01:00
} ,
Timeout : * apiServerTimeout ,
}
return & groupWatcher {
apiServer : apiServer ,
2021-05-14 19:00:05 +02:00
getAuthHeader : ac . GetAuthHeader ,
2021-03-14 20:10:35 +01:00
namespaces : namespaces ,
selectors : selectors ,
client : client ,
m : make ( map [ string ] * urlWatcher ) ,
}
}
func getGroupWatcher ( apiServer string , ac * promauth . Config , namespaces [ ] string , selectors [ ] Selector , proxyURL * url . URL ) * groupWatcher {
2021-04-05 21:25:31 +02:00
proxyURLStr := "<nil>"
if proxyURL != nil {
proxyURLStr = proxyURL . String ( )
}
key := fmt . Sprintf ( "apiServer=%s, namespaces=%s, selectors=%s, proxyURL=%s, authConfig=%s" ,
apiServer , namespaces , selectorsKey ( selectors ) , proxyURLStr , ac . String ( ) )
2021-03-14 20:10:35 +01:00
groupWatchersLock . Lock ( )
gw := groupWatchers [ key ]
if gw == nil {
gw = newGroupWatcher ( apiServer , ac , namespaces , selectors , proxyURL )
groupWatchers [ key ] = gw
}
groupWatchersLock . Unlock ( )
return gw
}
func selectorsKey ( selectors [ ] Selector ) string {
var sb strings . Builder
for _ , s := range selectors {
fmt . Fprintf ( & sb , "{role=%q, label=%q, field=%q}" , s . Role , s . Label , s . Field )
}
return sb . String ( )
}
var (
groupWatchersLock sync . Mutex
groupWatchers = make ( map [ string ] * groupWatcher )
_ = metrics . NewGauge ( ` vm_promscrape_discovery_kubernetes_group_watchers ` , func ( ) float64 {
groupWatchersLock . Lock ( )
n := len ( groupWatchers )
groupWatchersLock . Unlock ( )
return float64 ( n )
} )
)
2021-04-29 09:14:24 +02:00
func ( gw * groupWatcher ) getObjectByRoleLocked ( role , namespace , name string ) object {
2021-03-14 20:10:35 +01:00
if gw == nil {
2021-03-11 15:41:09 +01:00
// this is needed for testing
2021-03-05 11:36:05 +01:00
return nil
}
2021-04-02 13:45:08 +02:00
key := namespace + "/" + name
2021-04-29 09:14:24 +02:00
for _ , uw := range gw . m {
2021-03-05 11:36:05 +01:00
if uw . role != role {
2021-04-02 13:17:53 +02:00
// Role mismatch
continue
}
if uw . namespace != "" && uw . namespace != namespace {
// Namespace mismatch
2021-03-05 11:36:05 +01:00
continue
}
2021-04-29 09:14:24 +02:00
if o := uw . objectsByKey [ key ] ; o != nil {
2021-03-11 15:41:09 +01:00
return o
2021-03-05 11:36:05 +01:00
}
}
2021-03-11 15:41:09 +01:00
return nil
2021-03-05 11:36:05 +01:00
}
2021-03-14 20:10:35 +01:00
func ( gw * groupWatcher ) startWatchersForRole ( role string , aw * apiWatcher ) {
2021-08-29 10:23:06 +02:00
if role == "endpoints" || role == "endpointslice" {
// endpoints and endpointslice watchers query pod and service objects. So start watchers for these roles as well.
2021-05-12 13:10:10 +02:00
gw . startWatchersForRole ( "pod" , nil )
gw . startWatchersForRole ( "service" , nil )
}
2021-05-17 22:45:20 +02:00
paths := getAPIPathsWithNamespaces ( role , gw . namespaces , gw . selectors )
for _ , path := range paths {
2021-03-14 20:10:35 +01:00
apiURL := gw . apiServer + path
gw . mu . Lock ( )
uw := gw . m [ apiURL ]
2021-04-05 12:53:26 +02:00
needStart := uw == nil
if needStart {
2021-05-17 22:45:20 +02:00
uw = newURLWatcher ( role , apiURL , gw )
2021-03-14 20:10:35 +01:00
gw . m [ apiURL ] = uw
}
2021-04-29 09:14:24 +02:00
if aw != nil {
uw . subscribeAPIWatcherLocked ( aw )
}
2021-03-14 20:10:35 +01:00
gw . mu . Unlock ( )
2021-04-05 12:53:26 +02:00
if needStart {
uw . reloadObjects ( )
go uw . watchForUpdates ( )
2021-03-13 14:18:47 +01:00
}
2021-03-14 20:10:35 +01:00
}
2021-03-05 11:36:05 +01:00
}
2021-03-14 20:10:35 +01:00
// doRequest performs http request to the given requestURL.
func ( gw * groupWatcher ) doRequest ( requestURL string ) ( * http . Response , error ) {
2021-08-29 10:16:40 +02:00
if strings . Contains ( requestURL , "/apis/networking.k8s.io/v1/" ) && atomic . LoadUint32 ( & gw . useNetworkingV1Beta1 ) == 1 {
// Update networking URL for old Kubernetes API, which supports only v1beta1 path.
requestURL = strings . Replace ( requestURL , "/apis/networking.k8s.io/v1/" , "/apis/networking.k8s.io/v1beta1/" , 1 )
}
if strings . Contains ( requestURL , "/apis/discovery.k8s.io/v1/" ) && atomic . LoadUint32 ( & gw . useDiscoveryV1Beta1 ) == 1 {
// Update discovery URL for old Kuberentes API, which supports only v1beta1 path.
requestURL = strings . Replace ( requestURL , "/apis/discovery.k8s.io/v1/" , "/apis/discovery.k8s.io/v1beta1/" , 1 )
}
2021-03-14 20:10:35 +01:00
req , err := http . NewRequest ( "GET" , requestURL , nil )
if err != nil {
logger . Fatalf ( "cannot create a request for %q: %s" , requestURL , err )
2021-03-05 11:36:05 +01:00
}
2021-05-14 19:00:05 +02:00
if ah := gw . getAuthHeader ( ) ; ah != "" {
req . Header . Set ( "Authorization" , ah )
2021-03-14 20:10:35 +01:00
}
2021-08-29 10:16:40 +02:00
resp , err := gw . client . Do ( req )
if err != nil {
return nil , err
}
if resp . StatusCode == http . StatusNotFound {
if strings . Contains ( requestURL , "/apis/networking.k8s.io/v1/" ) && atomic . LoadUint32 ( & gw . useNetworkingV1Beta1 ) == 0 {
atomic . StoreUint32 ( & gw . useNetworkingV1Beta1 , 1 )
return gw . doRequest ( requestURL )
}
if strings . Contains ( requestURL , "/apis/discovery.k8s.io/v1/" ) && atomic . LoadUint32 ( & gw . useDiscoveryV1Beta1 ) == 0 {
atomic . StoreUint32 ( & gw . useDiscoveryV1Beta1 , 1 )
return gw . doRequest ( requestURL )
}
}
return resp , nil
2021-03-05 11:36:05 +01:00
}
2021-04-05 21:02:09 +02:00
func ( gw * groupWatcher ) registerPendingAPIWatchers ( ) {
2021-03-14 20:10:35 +01:00
gw . mu . Lock ( )
2021-04-29 09:14:24 +02:00
defer gw . mu . Unlock ( )
2021-03-14 20:10:35 +01:00
for _ , uw := range gw . m {
2021-04-29 09:14:24 +02:00
uw . registerPendingAPIWatchersLocked ( )
2021-03-14 20:10:35 +01:00
}
2021-04-05 21:02:09 +02:00
}
func ( gw * groupWatcher ) unsubscribeAPIWatcher ( aw * apiWatcher ) {
2021-04-29 09:14:24 +02:00
gw . mu . Lock ( )
defer gw . mu . Unlock ( )
for _ , uw := range gw . m {
uw . unsubscribeAPIWatcherLocked ( aw )
2021-04-05 21:02:09 +02:00
}
2021-03-14 20:10:35 +01:00
}
2021-03-05 11:36:05 +01:00
2021-04-02 13:45:08 +02:00
// urlWatcher watches for an apiURL and updates object states in objectsByKey.
2021-04-29 09:14:24 +02:00
//
// urlWatcher fields must be accessed under gw.mu lock.
2021-03-05 11:36:05 +01:00
type urlWatcher struct {
2021-04-02 13:17:53 +02:00
role string
namespace string
apiURL string
gw * groupWatcher
2021-03-05 11:36:05 +01:00
parseObject parseObjectFunc
parseObjectList parseObjectListFunc
2021-04-05 21:02:09 +02:00
// awsPending contains pending apiWatcher objects, which are registered in a batch.
// Batch registering saves CPU time needed for registering big number of Kubernetes objects
// shared among big number of scrape jobs, since per-object labels are generated only once
// for all the scrape jobs (each scrape job is associated with a single apiWatcher).
2022-04-20 15:11:37 +02:00
// See registerPendingAPIWatchersLocked for details.
2021-04-05 21:02:09 +02:00
awsPending map [ * apiWatcher ] struct { }
2021-03-11 15:41:09 +01:00
// aws contains registered apiWatcher objects
aws map [ * apiWatcher ] struct { }
2021-04-02 13:45:08 +02:00
// objectsByKey contains the latest state for objects obtained from apiURL
objectsByKey map [ string ] object
2021-03-05 16:29:54 +01:00
2021-03-07 18:50:01 +01:00
resourceVersion string
2021-03-05 11:36:05 +01:00
2021-03-26 11:28:10 +01:00
objectsCount * metrics . Counter
objectsAdded * metrics . Counter
objectsRemoved * metrics . Counter
objectsUpdated * metrics . Counter
staleResourceVersions * metrics . Counter
2021-03-05 11:36:05 +01:00
}
2021-05-17 22:45:20 +02:00
func newURLWatcher ( role , apiURL string , gw * groupWatcher ) * urlWatcher {
2021-03-11 15:41:09 +01:00
parseObject , parseObjectList := getObjectParsersForRole ( role )
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_url_watchers { role=%q} ` , role ) ) . Inc ( )
uw := & urlWatcher {
2021-05-17 22:45:20 +02:00
role : role ,
apiURL : apiURL ,
gw : gw ,
2021-03-05 11:36:05 +01:00
parseObject : parseObject ,
parseObjectList : parseObjectList ,
2021-04-05 21:02:09 +02:00
awsPending : make ( map [ * apiWatcher ] struct { } ) ,
2021-04-02 13:45:08 +02:00
aws : make ( map [ * apiWatcher ] struct { } ) ,
objectsByKey : make ( map [ string ] object ) ,
2021-03-05 11:36:05 +01:00
2021-03-26 11:28:10 +01:00
objectsCount : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_objects { role=%q} ` , role ) ) ,
objectsAdded : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_objects_added_total { role=%q} ` , role ) ) ,
objectsRemoved : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_objects_removed_total { role=%q} ` , role ) ) ,
objectsUpdated : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_objects_updated_total { role=%q} ` , role ) ) ,
staleResourceVersions : metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_stale_resource_versions_total { role=%q} ` , role ) ) ,
2021-03-11 15:41:09 +01:00
}
2021-03-14 20:10:35 +01:00
logger . Infof ( "started %s watcher for %q" , uw . role , uw . apiURL )
2021-03-11 15:41:09 +01:00
return uw
}
2021-03-05 11:36:05 +01:00
2021-04-29 09:14:24 +02:00
func ( uw * urlWatcher ) subscribeAPIWatcherLocked ( aw * apiWatcher ) {
2021-03-14 20:10:35 +01:00
if _ , ok := uw . aws [ aw ] ; ! ok {
2021-04-05 21:02:09 +02:00
if _ , ok := uw . awsPending [ aw ] ; ! ok {
uw . awsPending [ aw ] = struct { } { }
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_subscribers { role=%q,status="pending"} ` , uw . role ) ) . Inc ( )
2021-03-14 20:10:35 +01:00
}
2021-03-05 11:36:05 +01:00
}
}
2021-04-29 09:14:24 +02:00
func ( uw * urlWatcher ) registerPendingAPIWatchersLocked ( ) {
2021-04-27 13:57:48 +02:00
if len ( uw . awsPending ) == 0 {
return
}
2022-04-20 15:11:37 +02:00
aws := make ( [ ] * apiWatcher , 0 , len ( uw . awsPending ) )
2021-04-05 21:02:09 +02:00
for aw := range uw . awsPending {
2021-04-07 12:07:36 +02:00
uw . aws [ aw ] = struct { } { }
2022-04-20 15:11:37 +02:00
aws = append ( aws , aw )
}
swosByKey := make ( [ ] map [ string ] [ ] interface { } , len ( aws ) )
for i := range aws {
swosByKey [ i ] = make ( map [ string ] [ ] interface { } )
}
// Generate ScrapeWork objects in parallel on available CPU cores.
// This should reduce the time needed for their generation on systems with many CPU cores.
var swosByKeyLock sync . Mutex
var wg sync . WaitGroup
limiterCh := make ( chan struct { } , cgroup . AvailableCPUs ( ) )
for key , o := range uw . objectsByKey {
labels := o . getTargetLabels ( uw . gw )
wg . Add ( 1 )
limiterCh <- struct { } { }
go func ( key string , labels [ ] map [ string ] string ) {
for i , aw := range aws {
swos := getScrapeWorkObjectsForLabels ( aw . swcFunc , labels )
if len ( swos ) > 0 {
swosByKeyLock . Lock ( )
swosByKey [ i ] [ key ] = swos
swosByKeyLock . Unlock ( )
}
}
wg . Done ( )
<- limiterCh
} ( key , labels )
}
wg . Wait ( )
for i , aw := range aws {
aw . reloadScrapeWorks ( uw , swosByKey [ i ] )
2021-04-05 21:02:09 +02:00
}
2021-04-29 09:14:24 +02:00
uw . awsPending = make ( map [ * apiWatcher ] struct { } )
2022-04-20 15:11:37 +02:00
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_subscribers { role=%q,status="working"} ` , uw . role ) ) . Add ( len ( aws ) )
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_subscribers { role=%q,status="pending"} ` , uw . role ) ) . Add ( - len ( aws ) )
2021-04-05 21:02:09 +02:00
}
2021-04-29 09:14:24 +02:00
func ( uw * urlWatcher ) unsubscribeAPIWatcherLocked ( aw * apiWatcher ) {
2021-04-05 21:02:09 +02:00
if _ , ok := uw . awsPending [ aw ] ; ok {
delete ( uw . awsPending , aw )
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_subscribers { role=%q,status="pending"} ` , uw . role ) ) . Dec ( )
}
2021-03-14 20:10:35 +01:00
if _ , ok := uw . aws [ aw ] ; ok {
delete ( uw . aws , aw )
2021-04-05 21:02:09 +02:00
metrics . GetOrCreateCounter ( fmt . Sprintf ( ` vm_promscrape_discovery_kubernetes_subscribers { role=%q,status="working"} ` , uw . role ) ) . Dec ( )
2021-03-11 15:41:09 +01:00
}
}
2021-03-05 11:36:05 +01:00
// reloadObjects reloads objects to the latest state and returns resourceVersion for the latest state.
func ( uw * urlWatcher ) reloadObjects ( ) string {
2021-05-18 14:10:45 +02:00
if uw . resourceVersion != "" {
2021-03-11 15:41:09 +01:00
// Fast path - there is no need in reloading the objects.
2021-05-18 14:10:45 +02:00
return uw . resourceVersion
2021-03-07 18:50:01 +01:00
}
2021-03-05 11:36:05 +01:00
requestURL := uw . apiURL
2021-03-14 20:10:35 +01:00
resp , err := uw . gw . doRequest ( requestURL )
2021-03-05 11:36:05 +01:00
if err != nil {
2021-03-11 15:41:09 +01:00
logger . Errorf ( "cannot perform request to %q: %s" , requestURL , err )
2021-03-05 11:36:05 +01:00
return ""
}
if resp . StatusCode != http . StatusOK {
2021-03-11 15:41:09 +01:00
body , _ := ioutil . ReadAll ( resp . Body )
_ = resp . Body . Close ( )
2021-03-05 11:36:05 +01:00
logger . Errorf ( "unexpected status code for request to %q: %d; want %d; response: %q" , requestURL , resp . StatusCode , http . StatusOK , body )
return ""
}
2021-04-02 13:45:08 +02:00
objectsByKey , metadata , err := uw . parseObjectList ( resp . Body )
2021-03-11 15:41:09 +01:00
_ = resp . Body . Close ( )
2021-03-05 11:36:05 +01:00
if err != nil {
2021-03-11 15:41:09 +01:00
logger . Errorf ( "cannot parse objects from %q: %s" , requestURL , err )
2021-03-05 11:36:05 +01:00
return ""
}
2021-03-11 15:41:09 +01:00
2021-04-29 09:14:24 +02:00
uw . gw . mu . Lock ( )
2021-03-11 15:41:09 +01:00
var updated , removed , added int
2021-04-02 13:45:08 +02:00
for key := range uw . objectsByKey {
2022-04-20 15:11:37 +02:00
o , ok := objectsByKey [ key ]
if ok {
uw . updateObjectLocked ( key , o )
2021-03-11 15:41:09 +01:00
updated ++
} else {
2022-04-20 15:11:37 +02:00
uw . removeObjectLocked ( key )
2021-03-11 15:41:09 +01:00
removed ++
2021-03-05 11:36:05 +01:00
}
}
2022-04-20 15:11:37 +02:00
for key , o := range objectsByKey {
2021-04-02 13:45:08 +02:00
if _ , ok := uw . objectsByKey [ key ] ; ! ok {
2022-04-20 15:11:37 +02:00
uw . updateObjectLocked ( key , o )
2021-03-11 15:41:09 +01:00
added ++
}
}
2021-05-18 14:10:45 +02:00
uw . gw . mu . Unlock ( )
2021-03-11 15:41:09 +01:00
uw . objectsUpdated . Add ( updated )
uw . objectsRemoved . Add ( removed )
uw . objectsAdded . Add ( added )
uw . objectsCount . Add ( added - removed )
2021-03-07 18:50:01 +01:00
uw . resourceVersion = metadata . ResourceVersion
2021-04-29 09:14:24 +02:00
2021-05-18 14:29:32 +02:00
logger . Infof ( "reloaded %d objects from %q; updated=%d, removed=%d, added=%d, resourceVersion=%q" ,
len ( objectsByKey ) , requestURL , updated , removed , added , uw . resourceVersion )
2021-04-05 12:53:26 +02:00
return uw . resourceVersion
}
2021-03-07 18:50:01 +01:00
// watchForUpdates watches for object updates starting from uw.resourceVersion and updates the corresponding objects to the latest state.
2021-03-05 11:36:05 +01:00
//
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes
2021-03-07 18:50:01 +01:00
func ( uw * urlWatcher ) watchForUpdates ( ) {
2021-03-05 11:36:05 +01:00
backoffDelay := time . Second
maxBackoffDelay := 30 * time . Second
backoffSleep := func ( ) {
time . Sleep ( backoffDelay )
backoffDelay *= 2
if backoffDelay > maxBackoffDelay {
backoffDelay = maxBackoffDelay
}
}
apiURL := uw . apiURL
delimiter := "?"
if strings . Contains ( apiURL , "?" ) {
delimiter = "&"
}
2021-03-14 20:10:35 +01:00
timeoutSeconds := time . Duration ( 0.9 * float64 ( uw . gw . client . Timeout ) ) . Seconds ( )
2021-03-10 14:06:33 +01:00
apiURL += delimiter + "watch=1&allowWatchBookmarks=true&timeoutSeconds=" + strconv . Itoa ( int ( timeoutSeconds ) )
2021-03-05 11:36:05 +01:00
for {
2021-03-07 18:50:01 +01:00
resourceVersion := uw . reloadObjects ( )
2021-03-14 20:55:00 +01:00
if resourceVersion == "" {
backoffSleep ( )
continue
2021-03-05 11:36:05 +01:00
}
2021-03-14 20:55:00 +01:00
requestURL := apiURL + "&resourceVersion=" + url . QueryEscape ( resourceVersion )
2021-03-14 20:10:35 +01:00
resp , err := uw . gw . doRequest ( requestURL )
2021-03-05 11:36:05 +01:00
if err != nil {
2021-03-13 14:18:47 +01:00
logger . Errorf ( "cannot perform request to %q: %s" , requestURL , err )
2021-03-05 11:36:05 +01:00
backoffSleep ( )
continue
}
if resp . StatusCode != http . StatusOK {
if resp . StatusCode == 410 {
// There is no need for sleep on 410 error. See https://kubernetes.io/docs/reference/using-api/api-concepts/#410-gone-responses
backoffDelay = time . Second
2021-03-26 11:28:10 +01:00
uw . staleResourceVersions . Inc ( )
2021-05-18 14:10:45 +02:00
uw . resourceVersion = ""
2021-03-05 11:36:05 +01:00
} else {
2021-03-26 11:28:10 +01:00
body , _ := ioutil . ReadAll ( resp . Body )
_ = resp . Body . Close ( )
logger . Errorf ( "unexpected status code for request to %q: %d; want %d; response: %q" , requestURL , resp . StatusCode , http . StatusOK , body )
2021-03-05 11:36:05 +01:00
backoffSleep ( )
}
continue
}
backoffDelay = time . Second
err = uw . readObjectUpdateStream ( resp . Body )
_ = resp . Body . Close ( )
if err != nil {
if ! errors . Is ( err , io . EOF ) {
logger . Errorf ( "error when reading WatchEvent stream from %q: %s" , requestURL , err )
2021-05-18 22:25:42 +02:00
uw . resourceVersion = ""
2021-03-05 11:36:05 +01:00
}
backoffSleep ( )
continue
}
}
}
2022-04-20 09:50:36 +02:00
// readObjectUpdateStream reads Kubernetes watch events from r and updates locally cached objects according to the received events.
2021-03-05 11:36:05 +01:00
func ( uw * urlWatcher ) readObjectUpdateStream ( r io . Reader ) error {
d := json . NewDecoder ( r )
var we WatchEvent
for {
if err := d . Decode ( & we ) ; err != nil {
return err
}
switch we . Type {
case "ADDED" , "MODIFIED" :
2021-03-11 12:06:40 +01:00
o , err := uw . parseObject ( we . Object )
if err != nil {
2021-05-18 22:25:42 +02:00
return fmt . Errorf ( "cannot parse %s object: %w" , we . Type , err )
2021-03-11 12:06:40 +01:00
}
2021-04-02 13:45:08 +02:00
key := o . key ( )
2021-04-29 09:14:24 +02:00
uw . gw . mu . Lock ( )
2021-04-02 13:45:08 +02:00
if _ , ok := uw . objectsByKey [ key ] ; ! ok {
2021-03-11 15:41:09 +01:00
uw . objectsCount . Inc ( )
uw . objectsAdded . Inc ( )
2021-03-05 11:36:05 +01:00
} else {
2021-03-11 15:41:09 +01:00
uw . objectsUpdated . Inc ( )
2021-03-05 11:36:05 +01:00
}
2022-04-20 15:11:37 +02:00
uw . updateObjectLocked ( key , o )
2021-04-29 09:14:24 +02:00
uw . gw . mu . Unlock ( )
2021-03-05 11:36:05 +01:00
case "DELETED" :
2021-03-11 12:06:40 +01:00
o , err := uw . parseObject ( we . Object )
if err != nil {
2021-05-18 22:25:42 +02:00
return fmt . Errorf ( "cannot parse %s object: %w" , we . Type , err )
2021-03-11 12:06:40 +01:00
}
2021-04-02 13:45:08 +02:00
key := o . key ( )
2021-04-29 09:14:24 +02:00
uw . gw . mu . Lock ( )
2021-05-20 11:00:07 +02:00
if _ , ok := uw . objectsByKey [ key ] ; ok {
2021-05-18 22:25:42 +02:00
uw . objectsCount . Dec ( )
uw . objectsRemoved . Inc ( )
2021-03-11 15:41:09 +01:00
}
2022-04-20 15:11:37 +02:00
uw . removeObjectLocked ( key )
2021-04-29 09:14:24 +02:00
uw . gw . mu . Unlock ( )
2021-03-10 14:06:33 +01:00
case "BOOKMARK" :
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
2021-03-11 12:06:40 +01:00
bm , err := parseBookmark ( we . Object )
if err != nil {
return fmt . Errorf ( "cannot parse bookmark from %q: %w" , we . Object , err )
}
2021-05-18 14:10:45 +02:00
uw . resourceVersion = bm . Metadata . ResourceVersion
2021-03-26 11:28:10 +01:00
case "ERROR" :
em , err := parseError ( we . Object )
if err != nil {
2021-03-26 11:45:59 +01:00
return fmt . Errorf ( "cannot parse error message from %q: %w" , we . Object , err )
2021-03-26 11:28:10 +01:00
}
if em . Code == 410 {
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#410-gone-responses
uw . staleResourceVersions . Inc ( )
2021-05-18 14:10:45 +02:00
uw . resourceVersion = ""
2021-03-26 11:28:10 +01:00
return nil
}
return fmt . Errorf ( "unexpected error message: %q" , we . Object )
2021-03-05 11:36:05 +01:00
default :
2021-03-26 11:28:10 +01:00
return fmt . Errorf ( "unexpected WatchEvent type %q: %q" , we . Type , we . Object )
2021-03-05 11:36:05 +01:00
}
}
}
2022-04-20 15:11:37 +02:00
func ( uw * urlWatcher ) updateObjectLocked ( key string , o object ) {
oPrev , ok := uw . objectsByKey [ key ]
if ok && reflect . DeepEqual ( oPrev , o ) {
// Nothing to do, since the new object is equal to the previous one.
return
}
uw . objectsByKey [ key ] = o
if len ( uw . aws ) == 0 {
return
}
labels := o . getTargetLabels ( uw . gw )
for aw := range uw . aws {
aw . setScrapeWorks ( uw , key , labels )
}
}
func ( uw * urlWatcher ) removeObjectLocked ( key string ) {
delete ( uw . objectsByKey , key )
for aw := range uw . aws {
aw . removeScrapeWorks ( uw , key )
}
}
2021-03-26 11:28:10 +01:00
// Bookmark is a bookmark message from Kubernetes Watch API.
2021-03-11 12:06:40 +01:00
// See https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
type Bookmark struct {
Metadata struct {
ResourceVersion string
}
}
func parseBookmark ( data [ ] byte ) ( * Bookmark , error ) {
var bm Bookmark
if err := json . Unmarshal ( data , & bm ) ; err != nil {
return nil , err
}
return & bm , nil
}
2021-03-26 11:28:10 +01:00
// Error is an error message from Kubernetes Watch API.
type Error struct {
Code int
}
func parseError ( data [ ] byte ) ( * Error , error ) {
var em Error
if err := json . Unmarshal ( data , & em ) ; err != nil {
return nil , err
}
return & em , nil
}
2021-05-17 22:45:20 +02:00
func getAPIPathsWithNamespaces ( role string , namespaces [ ] string , selectors [ ] Selector ) [ ] string {
2021-04-05 19:27:23 +02:00
objectType := getObjectTypeByRole ( role )
if objectType == "nodes" || len ( namespaces ) == 0 {
2021-03-05 11:36:05 +01:00
query := joinSelectors ( role , selectors )
2021-04-05 19:27:23 +02:00
path := getAPIPath ( objectType , "" , query )
2021-05-17 22:45:20 +02:00
return [ ] string { path }
2021-03-05 11:36:05 +01:00
}
query := joinSelectors ( role , selectors )
paths := make ( [ ] string , len ( namespaces ) )
for i , namespace := range namespaces {
2021-04-05 19:27:23 +02:00
paths [ i ] = getAPIPath ( objectType , namespace , query )
2021-03-05 11:36:05 +01:00
}
2021-05-17 22:45:20 +02:00
return paths
2021-03-05 11:36:05 +01:00
}
2021-04-05 19:27:23 +02:00
func getAPIPath ( objectType , namespace , query string ) string {
suffix := objectType
2021-03-05 11:36:05 +01:00
if namespace != "" {
2021-04-05 19:27:23 +02:00
suffix = "namespaces/" + namespace + "/" + objectType
2021-03-05 11:36:05 +01:00
}
if len ( query ) > 0 {
suffix += "?" + query
}
2021-04-05 19:27:23 +02:00
if objectType == "ingresses" {
2021-08-29 10:16:40 +02:00
return "/apis/networking.k8s.io/v1/" + suffix
2021-03-05 11:36:05 +01:00
}
2021-04-05 19:27:23 +02:00
if objectType == "endpointslices" {
2021-08-29 10:16:40 +02:00
return "/apis/discovery.k8s.io/v1/" + suffix
2021-03-05 11:36:05 +01:00
}
return "/api/v1/" + suffix
}
func joinSelectors ( role string , selectors [ ] Selector ) string {
var labelSelectors , fieldSelectors [ ] string
for _ , s := range selectors {
if s . Role != role {
continue
}
if s . Label != "" {
labelSelectors = append ( labelSelectors , s . Label )
}
if s . Field != "" {
fieldSelectors = append ( fieldSelectors , s . Field )
}
}
var args [ ] string
if len ( labelSelectors ) > 0 {
args = append ( args , "labelSelector=" + url . QueryEscape ( strings . Join ( labelSelectors , "," ) ) )
}
if len ( fieldSelectors ) > 0 {
args = append ( args , "fieldSelector=" + url . QueryEscape ( strings . Join ( fieldSelectors , "," ) ) )
}
return strings . Join ( args , "&" )
}
2021-04-05 19:27:23 +02:00
func getObjectTypeByRole ( role string ) string {
2021-03-05 11:36:05 +01:00
switch role {
case "node" :
return "nodes"
case "pod" :
return "pods"
case "service" :
return "services"
case "endpoints" :
return "endpoints"
2021-08-29 10:23:06 +02:00
case "endpointslice" :
2021-03-05 11:36:05 +01:00
return "endpointslices"
case "ingress" :
return "ingresses"
default :
logger . Panicf ( "BUG: unknonw role=%q" , role )
return ""
}
}
func getObjectParsersForRole ( role string ) ( parseObjectFunc , parseObjectListFunc ) {
switch role {
case "node" :
return parseNode , parseNodeList
case "pod" :
return parsePod , parsePodList
case "service" :
return parseService , parseServiceList
case "endpoints" :
return parseEndpoints , parseEndpointsList
2021-08-29 10:23:06 +02:00
case "endpointslice" :
2021-03-05 11:36:05 +01:00
return parseEndpointSlice , parseEndpointSliceList
case "ingress" :
return parseIngress , parseIngressList
default :
logger . Panicf ( "BUG: unsupported role=%q" , role )
return nil , nil
}
}