Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1082 (2021-02-22 20:25:25)

vmagent kubernetes watch stream discovery.


Patch:
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index b41bcc463..ee2f4b2d9 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -48,6 +48,8 @@ type Config struct {
 
 	// This is set to the directory from where the config has been loaded.
 	baseDir string
+	// used for data sync with kubernetes.
+	kwh *kubernetesWatchHandler
 }
 
 // GlobalConfig represents essential parts for `global` section of Prometheus config.
@@ -139,6 +141,7 @@ func loadConfig(path string) (cfg *Config, data []byte, err error) {
 	if err := cfgObj.parse(data, path); err != nil {
 		return nil, nil, fmt.Errorf("cannot parse Prometheus config from %q: %w", path, err)
 	}
+	cfgObj.kwh = newKubernetesWatchHandler()
 	return &cfgObj, data, nil
 }
 
@@ -187,30 +190,41 @@ func getSWSByJob(sws []*ScrapeWork) map[string][]*ScrapeWork {
 }
 
 // getKubernetesSDScrapeWork returns `kubernetes_sd_configs` ScrapeWork from cfg.
-func (cfg *Config) getKubernetesSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
-	swsPrevByJob := getSWSByJob(prev)
+func getKubernetesSDScrapeWorkStream(cfg *Config, prev []*ScrapeWork) []*ScrapeWork {
+	cfg.kwh.startOnce.Do(func() {
+		go processKubernetesSyncEvents(cfg)
+	})
 	dst := make([]*ScrapeWork, 0, len(prev))
+	// updated access time.
+	cfg.kwh.mu.Lock()
+	cfg.kwh.lastAccessTime = time.Now()
+	cfg.kwh.mu.Unlock()
 	for i := range cfg.ScrapeConfigs {
 		sc := &cfg.ScrapeConfigs[i]
-		dstLen := len(dst)
-		ok := true
 		for j := range sc.KubernetesSDConfigs {
+			// generate set name
+			setKey := fmt.Sprintf("%d/%d/%s", i, j, sc.JobName)
+			cfg.kwh.mu.Lock()
+			cfg.kwh.sdcSet[setKey] = sc.swc
+			cfg.kwh.mu.Unlock()
 			sdc := &sc.KubernetesSDConfigs[j]
-			var okLocal bool
-			dst, okLocal = appendSDScrapeWork(dst, sdc, cfg.baseDir, sc.swc, "kubernetes_sd_config")
-			if ok {
-				ok = okLocal
+			ms, err := kubernetes.StartWatchOnce(cfg.kwh.watchCfg, setKey, sdc, cfg.baseDir)
+			if err != nil {
+				logger.Errorf("got unexpected error: %v", err)
 			}
+			dst = appendScrapeWorkForTargetLabels(dst, sc.swc, ms, "kubernetes_sd_config")
 		}
-		if ok {
-			continue
-		}
-		swsPrev := swsPrevByJob[sc.swc.jobName]
-		if len(swsPrev) > 0 {
-			logger.Errorf("there were errors when discovering kubernetes targets for job %q, so preserving the previous targets", sc.swc.jobName)
-			dst = append(dst[:dstLen], swsPrev...)
-		}
 	}
+	// dst will
+	if len(dst) > 0 {
+		return dst
+	}
+	// result from cache
+	cfg.kwh.mu.Lock()
+	for _, v := range cfg.kwh.swCache {
+		dst = append(dst, v...)
+	}
+	cfg.kwh.mu.Unlock()
 	return dst
 }
 
diff --git a/lib/promscrape/discovery/kubernetes/api.go b/lib/promscrape/discovery/kubernetes/api.go
index 2c0463214..a2b3834d9 100644
--- a/lib/promscrape/discovery/kubernetes/api.go
+++ b/lib/promscrape/discovery/kubernetes/api.go
@@ -1,77 +1,42 @@
 package kubernetes
 
 import (
-	"fmt"
-	"net"
-	"os"
+	"sync"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
 // apiConfig contains config for API server
 type apiConfig struct {
-	client     *discoveryutils.Client
+	setName    string
 	namespaces []string
 	selectors  []Selector
+	wc         *watchClient
+	targetChan chan SyncEvent
+	watchOnce  sync.Once
 }
 
 var configMap = discoveryutils.NewConfigMap()
 
-func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
-	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(sdc, baseDir) })
+func getAPIConfig(watchCfg *WatchConfig, setName string, sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(watchCfg, setName, sdc, baseDir) })
 	if err != nil {
 		return nil, err
 	}
 	return v.(*apiConfig), nil
 }
 
-func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
-	ac, err := promauth.NewConfig(baseDir, sdc.BasicAuth, sdc.BearerToken, sdc.BearerTokenFile, sdc.TLSConfig)
+func newAPIConfig(watchCfg *WatchConfig, setName string, sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	wc, err := newWatchClient(watchCfg.WG, sdc, baseDir)
 	if err != nil {
-		return nil, fmt.Errorf("cannot parse auth config: %w", err)
-	}
-	apiServer := sdc.APIServer
-	if len(apiServer) == 0 {
-		// Assume we run at k8s pod.
-		// Discover apiServer and auth config according to k8s docs.
-		// See https://kubernetes.io/docs/reference/access-authn-authz/service-accounts-admin/#service-account-admission-controller
-		host := os.Getenv("KUBERNETES_SERVICE_HOST")
-		port := os.Getenv("KUBERNETES_SERVICE_PORT")
-		if len(host) == 0 {
-			return nil, fmt.Errorf("cannot find KUBERNETES_SERVICE_HOST env var; it must be defined when running in k8s; " +
-				"probably, `kubernetes_sd_config->api_server` is missing in Prometheus configs?")
-		}
-		if len(port) == 0 {
-			return nil, fmt.Errorf("cannot find KUBERNETES_SERVICE_PORT env var; it must be defined when running in k8s; "+
-				"KUBERNETES_SERVICE_HOST=%q", host)
-		}
-		apiServer = "https://" + net.JoinHostPort(host, port)
-		tlsConfig := promauth.TLSConfig{
-			CAFile: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
-		}
-		acNew, err := promauth.NewConfig(".", nil, "", "/var/run/secrets/kubernetes.io/serviceaccount/token", &tlsConfig)
-		if err != nil {
-			return nil, fmt.Errorf("cannot initialize service account auth: %w; probably, `kubernetes_sd_config->api_server` is missing in Prometheus configs?", err)
-		}
-		ac = acNew
-	}
-	client, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL)
-	if err != nil {
-		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
+		return nil, err
 	}
 	cfg := &apiConfig{
-		client:     client,
+		setName:    setName,
+		targetChan: watchCfg.WatchChan,
+		wc:         wc,
 		namespaces: sdc.Namespaces.Names,
 		selectors:  sdc.Selectors,
 	}
 	return cfg, nil
 }
-
-func getAPIResponse(cfg *apiConfig, role, path string) ([]byte, error) {
-	query := joinSelectors(role, cfg.namespaces, cfg.selectors)
-	if len(query) > 0 {
-		path += "?" + query
-	}
-	return cfg.client.GetAPIResponse(path)
-}
diff --git a/lib/promscrape/discovery/kubernetes/common_types.go b/lib/promscrape/discovery/kubernetes/common_types.go
index d1bc21203..f1b8dcce9 100644
--- a/lib/promscrape/discovery/kubernetes/common_types.go
+++ b/lib/promscrape/discovery/kubernetes/common_types.go
@@ -19,6 +19,12 @@ type ObjectMeta struct {
 	OwnerReferences []OwnerReference
 }
 
+// listMetadata kubernetes list metadata
+// https://v1-17.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#listmeta-v1-meta
+type listMetadata struct {
+	ResourceVersion string `json:"resourceVersion"`
+}
+
 func (om *ObjectMeta) registerLabelsAndAnnotations(prefix string, m map[string]string) {
 	for _, lb := range om.Labels {
 		ln := discoveryutils.SanitizeLabelName(lb.Name)
diff --git a/lib/promscrape/discovery/kubernetes/endpoints.go b/lib/promscrape/discovery/kubernetes/endpoints.go
index 6f95a2a9d..2839202d0 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints.go
@@ -3,70 +3,18 @@ package kubernetes
 import (
 	"encoding/json"
 	"fmt"
+	"sync"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"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: %w", err)
-	}
-	epl, err := parseEndpointsList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse endpoints response from API server: %w", 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
+	Items    []Endpoints
+	Metadata listMetadata `json:"metadata"`
 }
 
 // Endpoints implements k8s endpoints.
@@ -77,6 +25,10 @@ type Endpoints struct {
 	Subsets  []EndpointSubset
 }
 
+func (eps *Endpoints) key() string {
+	return eps.Metadata.Namespace + "/" + eps.Metadata.Name
+}
+
 // EndpointSubset implements k8s endpoint subset.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointsubset-v1-core
@@ -105,6 +57,10 @@ type ObjectReference struct {
 	Namespace string
 }
 
+func (or ObjectReference) key() string {
+	return or.Namespace + "/" + or.Name
+}
+
 // EndpointPort implements k8s endpoint port.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointport-v1beta1-discovery-k8s-io
@@ -127,13 +83,16 @@ func parseEndpointsList(data []byte) (*EndpointsList, error) {
 // 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)
+func (eps *Endpoints) appendTargetLabels(ms []map[string]string, podsCache, servicesCache *sync.Map) []map[string]string {
+	var svc *Service
+	if svco, ok := servicesCache.Load(eps.key()); ok {
+		svc = svco.(*Service)
+	}
 	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")
+			ms = appendEndpointLabelsForAddresses(ms, podPortsSeen, eps, ess.Addresses, epp, podsCache, svc, "true")
+			ms = appendEndpointLabelsForAddresses(ms, podPortsSeen, eps, ess.NotReadyAddresses, epp, podsCache, svc, "false")
 		}
 	}
 
@@ -169,9 +128,13 @@ func (eps *Endpoints) appendTargetLabels(ms []map[string]string, pods []Pod, svc
 }
 
 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 {
+	podsCache *sync.Map, svc *Service, ready string) []map[string]string {
 	for _, ea := range eas {
-		p := getPod(pods, ea.TargetRef.Namespace, ea.TargetRef.Name)
+		var p *Pod
+		if po, ok := podsCache.Load(ea.TargetRef.key()); ok {
+			p = po.(*Pod)
+		}
+		//p := getPod(pods, ea.TargetRef.Namespace, ea.TargetRef.Name)
 		m := getEndpointLabelsForAddressAndPort(podPortsSeen, eps, ea, epp, p, svc, ready)
 		ms = append(ms, m)
 	}
@@ -223,3 +186,24 @@ func getEndpointLabels(om ObjectMeta, ea EndpointAddress, epp EndpointPort, read
 	}
 	return m
 }
+
+func processEndpoints(cfg *apiConfig, sc *SharedKubernetesCache, p *Endpoints, action string) {
+	key := buildSyncKey("endpoints", cfg.setName, p.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		lbs := p.appendTargetLabels(nil, sc.Pods, sc.Services)
+		cfg.targetChan <- SyncEvent{
+			Labels:           lbs,
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+}
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices.go b/lib/promscrape/discovery/kubernetes/endpointslices.go
index f1f9da645..9a0a2c9d9 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices.go
@@ -4,69 +4,12 @@ import (
 	"encoding/json"
 	"fmt"
 	"strconv"
+	"sync"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getEndpointSlicesLabels returns labels for k8s endpointSlices obtained from the given cfg.
-func getEndpointSlicesLabels(cfg *apiConfig) ([]map[string]string, error) {
-	eps, err := getEndpointSlices(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
-}
-
-// getEndpointSlices retrieves endpointSlice with given apiConfig
-func getEndpointSlices(cfg *apiConfig) ([]EndpointSlice, error) {
-	if len(cfg.namespaces) == 0 {
-		return getEndpointSlicesByPath(cfg, "/apis/discovery.k8s.io/v1beta1/endpointslices")
-	}
-	// 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 []EndpointSlice
-	for _, ns := range namespaces {
-		path := fmt.Sprintf("/apis/discovery.k8s.io/v1beta1/namespaces/%s/endpointslices", ns)
-		eps, err := getEndpointSlicesByPath(cfg, path)
-		if err != nil {
-			return nil, err
-		}
-		result = append(result, eps...)
-	}
-	return result, nil
-}
-
-// getEndpointSlicesByPath retrieves endpointSlices from k8s api by given path
-func getEndpointSlicesByPath(cfg *apiConfig, path string) ([]EndpointSlice, error) {
-	data, err := getAPIResponse(cfg, "endpointslices", path)
-	if err != nil {
-		return nil, fmt.Errorf("cannot obtain endpointslices data from API server: %w", err)
-	}
-	epl, err := parseEndpointSlicesList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse endpointslices response from API server: %w", err)
-	}
-	return epl.Items, nil
-
-}
-
 // parseEndpointsList parses EndpointSliceList from data.
 func parseEndpointSlicesList(data []byte) (*EndpointSliceList, error) {
 	var esl EndpointSliceList
@@ -79,11 +22,17 @@ func parseEndpointSlicesList(data []byte) (*EndpointSliceList, error) {
 
 // appendTargetLabels injects labels for endPointSlice to slice map
 // follows TargetRef for enrich labels with pod and service metadata
-func (eps *EndpointSlice) appendTargetLabels(ms []map[string]string, pods []Pod, svcs []Service) []map[string]string {
-	svc := getService(svcs, eps.Metadata.Namespace, eps.Metadata.Name)
+func (eps *EndpointSlice) appendTargetLabels(ms []map[string]string, podsCache, servicesCache *sync.Map) []map[string]string {
+	var svc *Service
+	if s, ok := servicesCache.Load(eps.key()); ok {
+		svc = s.(*Service)
+	}
 	podPortsSeen := make(map[*Pod][]int)
 	for _, ess := range eps.Endpoints {
-		pod := getPod(pods, ess.TargetRef.Namespace, ess.TargetRef.Name)
+		var pod *Pod
+		if p, ok := podsCache.Load(ess.TargetRef.key()); ok {
+			pod = p.(*Pod)
+		}
 		for _, epp := range eps.Ports {
 			for _, addr := range ess.Addresses {
 				ms = append(ms, getEndpointSliceLabelsForAddressAndPort(podPortsSeen, addr, eps, ess, epp, pod, svc))
@@ -186,7 +135,8 @@ func getEndpointSliceLabels(eps *EndpointSlice, addr string, ea Endpoint, epp En
 // that groups service endpoints slices.
 // https://v1-17.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpointslice-v1beta1-discovery-k8s-io
 type EndpointSliceList struct {
-	Items []EndpointSlice
+	Items    []EndpointSlice
+	Metadata listMetadata `json:"metadata"`
 }
 
 // EndpointSlice - implements kubernetes endpoint slice.
@@ -198,6 +148,10 @@ type EndpointSlice struct {
 	Ports       []EndpointPort
 }
 
+func (eps EndpointSlice) key() string {
+	return eps.Metadata.Namespace + "/" + eps.Metadata.Name
+}
+
 // Endpoint implements kubernetes object endpoint for endpoint slice.
 // https://v1-17.docs.kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#endpoint-v1beta1-discovery-k8s-io
 type Endpoint struct {
@@ -213,3 +167,23 @@ type Endpoint struct {
 type EndpointConditions struct {
 	Ready bool
 }
+
+func processEndpointSlices(cfg *apiConfig, sc *SharedKubernetesCache, p *EndpointSlice, action string) {
+	key := buildSyncKey("endpointslices", cfg.setName, p.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		cfg.targetChan <- SyncEvent{
+			Labels:           p.appendTargetLabels(nil, sc.Pods, sc.Services),
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+}
diff --git a/lib/promscrape/discovery/kubernetes/framer.go b/lib/promscrape/discovery/kubernetes/framer.go
new file mode 100644
index 000000000..3aff2c927
--- /dev/null
+++ b/lib/promscrape/discovery/kubernetes/framer.go
@@ -0,0 +1,56 @@
+package kubernetes
+
+import (
+	"encoding/json"
+	"io"
+)
+
+type jsonFrameReader struct {
+	r         io.ReadCloser
+	decoder   *json.Decoder
+	remaining []byte
+}
+
+func newJSONFramedReader(r io.ReadCloser) io.ReadCloser {
+	return &jsonFrameReader{
+		r:       r,
+		decoder: json.NewDecoder(r),
+	}
+}
+
+// ReadFrame decodes the next JSON object in the stream, or returns an error. The returned
+// byte slice will be modified the next time ReadFrame is invoked and should not be altered.
+func (r *jsonFrameReader) Read(data []byte) (int, error) {
+	// Return whatever remaining data exists from an in progress frame
+	if n := len(r.remaining); n > 0 {
+		if n <= len(data) {
+			data = append(data[0:0], r.remaining...)
+			r.remaining = nil
+			return n, nil
+		}
+		n = len(data)
+		data = append(data[0:0], r.remaining[:n]...)
+		r.remaining = r.remaining[n:]
+		return n, io.ErrShortBuffer
+	}
+
+	n := len(data)
+	m := json.RawMessage(data[:0])
+	if err := r.decoder.Decode(&m); err != nil {
+		return 0, err
+	}
+
+	// If capacity of data is less than length of the message, decoder will allocate a new slice
+	// and set m to it, which means we need to copy the partial result back into data and preserve
+	// the remaining result for subsequent reads.
+	if len(m) > n {
+		data = append(data[0:0], m[:n]...)
+		r.remaining = m[n:]
+		return n, io.ErrShortBuffer
+	}
+	return len(m), nil
+}
+
+func (r *jsonFrameReader) Close() error {
+	return r.r.Close()
+}
diff --git a/lib/promscrape/discovery/kubernetes/ingress.go b/lib/promscrape/discovery/kubernetes/ingress.go
index 5aad6d903..7a17854df 100644
--- a/lib/promscrape/discovery/kubernetes/ingress.go
+++ b/lib/promscrape/discovery/kubernetes/ingress.go
@@ -3,60 +3,16 @@ package kubernetes
 import (
 	"encoding/json"
 	"fmt"
-)
-
-// getIngressesLabels returns labels for k8s ingresses obtained from the given cfg.
-func getIngressesLabels(cfg *apiConfig) ([]map[string]string, error) {
-	igs, err := getIngresses(cfg)
-	if err != nil {
-		return nil, err
-	}
-	var ms []map[string]string
-	for _, ig := range igs {
-		ms = ig.appendTargetLabels(ms)
-	}
-	return ms, nil
-}
 
-func getIngresses(cfg *apiConfig) ([]Ingress, error) {
-	if len(cfg.namespaces) == 0 {
-		return getIngressesByPath(cfg, "/apis/extensions/v1beta1/ingresses")
-	}
-	// 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 []Ingress
-	for _, ns := range namespaces {
-		path := fmt.Sprintf("/apis/extensions/v1beta1/namespaces/%s/ingresses", ns)
-		igs, err := getIngressesByPath(cfg, path)
-		if err != nil {
-			return nil, err
-		}
-		result = append(result, igs...)
-	}
-	return result, nil
-}
-
-func getIngressesByPath(cfg *apiConfig, path string) ([]Ingress, error) {
-	data, err := getAPIResponse(cfg, "ingress", path)
-	if err != nil {
-		return nil, fmt.Errorf("cannot obtain ingresses data from API server: %w", err)
-	}
-	igl, err := parseIngressList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse ingresses response from API server: %w", err)
-	}
-	return igl.Items, nil
-}
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
 
 // IngressList represents ingress list in k8s.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#ingresslist-v1beta1-extensions
 type IngressList struct {
-	Items []Ingress
+	Items    []Ingress
+	Metadata listMetadata `json:"metadata"`
 }
 
 // Ingress represents ingress in k8s.
@@ -67,6 +23,10 @@ type Ingress struct {
 	Spec     IngressSpec
 }
 
+func (ig Ingress) key() string {
+	return ig.Metadata.Namespace + "/" + ig.Metadata.Name
+}
+
 // IngressSpec represents ingress spec in k8s.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#ingressspec-v1beta1-extensions
@@ -164,3 +124,23 @@ func getIngressRulePaths(paths []HTTPIngressPath) []string {
 	}
 	return result
 }
+
+func processIngress(cfg *apiConfig, p *Ingress, action string) {
+	key := buildSyncKey("ingress", cfg.setName, p.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		cfg.targetChan <- SyncEvent{
+			Labels:           p.appendTargetLabels(nil),
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "ERROR":
+	default:
+		logger.Infof("unexpected action: %s", action)
+	}
+}
diff --git a/lib/promscrape/discovery/kubernetes/kubernetes.go b/lib/promscrape/discovery/kubernetes/kubernetes.go
index d1f29ed75..aea328438 100644
--- a/lib/promscrape/discovery/kubernetes/kubernetes.go
+++ b/lib/promscrape/discovery/kubernetes/kubernetes.go
@@ -37,26 +37,16 @@ type Selector struct {
 	Field string `yaml:"field"`
 }
 
-// GetLabels returns labels for the given sdc and baseDir.
-func (sdc *SDConfig) GetLabels(baseDir string) ([]map[string]string, error) {
-	cfg, err := getAPIConfig(sdc, baseDir)
+// StartWatchOnce returns init labels for the given sdc and baseDir.
+// and starts watching for changes.
+func StartWatchOnce(watchCfg *WatchConfig, setName string, sdc *SDConfig, baseDir string) ([]map[string]string, error) {
+	cfg, err := getAPIConfig(watchCfg, setName, sdc, baseDir)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create API config: %w", err)
 	}
-	switch sdc.Role {
-	case "node":
-		return getNodesLabels(cfg)
-	case "service":
-		return getServicesLabels(cfg)
-	case "pod":
-		return getPodsLabels(cfg)
-	case "endpoints":
-		return getEndpointsLabels(cfg)
-	case "endpointslices":
-		return getEndpointSlicesLabels(cfg)
-	case "ingress":
-		return getIngressesLabels(cfg)
-	default:
-		return nil, fmt.Errorf("unexpected `role`: %q; must be one of `node`, `service`, `pod`, `endpoints` or `ingress`; skipping it", sdc.Role)
-	}
+	var ms []map[string]string
+	cfg.watchOnce.Do(func() {
+		ms = startWatcherByRole(watchCfg.Ctx, sdc.Role, cfg, watchCfg.SC)
+	})
+	return ms, nil
 }
diff --git a/lib/promscrape/discovery/kubernetes/node.go b/lib/promscrape/discovery/kubernetes/node.go
index 9a584c67e..134bdc77a 100644
--- a/lib/promscrape/discovery/kubernetes/node.go
+++ b/lib/promscrape/discovery/kubernetes/node.go
@@ -4,32 +4,16 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getNodesLabels returns labels for k8s nodes obtained from the given cfg.
-func getNodesLabels(cfg *apiConfig) ([]map[string]string, error) {
-	data, err := getAPIResponse(cfg, "node", "/api/v1/nodes")
-	if err != nil {
-		return nil, fmt.Errorf("cannot obtain nodes data from API server: %w", err)
-	}
-	nl, err := parseNodeList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse nodes response from API server: %w", err)
-	}
-	var ms []map[string]string
-	for _, n := range nl.Items {
-		// Do not apply namespaces, since they are missing in nodes.
-		ms = n.appendTargetLabels(ms)
-	}
-	return ms, nil
-}
-
 // NodeList represents NodeList from k8s API.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#nodelist-v1-core
 type NodeList struct {
-	Items []Node
+	Items    []Node
+	Metadata listMetadata `json:"metadata"`
 }
 
 // Node represents Node from k8s API.
@@ -40,6 +24,10 @@ type Node struct {
 	Status   NodeStatus
 }
 
+func (n Node) key() string {
+	return n.Metadata.Name
+}
+
 // NodeStatus represents NodeStatus from k8s API.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#nodestatus-v1-core
@@ -131,3 +119,24 @@ func getAddrByType(nas []NodeAddress, typ string) string {
 	}
 	return ""
 }
+
+func processNode(cfg *apiConfig, n *Node, action string) {
+	key := buildSyncKey("nodes", cfg.setName, n.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		lbs := n.appendTargetLabels(nil)
+		cfg.targetChan <- SyncEvent{
+			Labels:           lbs,
+			ConfigSectionSet: cfg.setName,
+			Key:              key,
+		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			ConfigSectionSet: cfg.setName,
+			Key:              key,
+		}
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+}
diff --git a/lib/promscrape/discovery/kubernetes/pod.go b/lib/promscrape/discovery/kubernetes/pod.go
index 3bc3495b0..1d5c384e4 100644
--- a/lib/promscrape/discovery/kubernetes/pod.go
+++ b/lib/promscrape/discovery/kubernetes/pod.go
@@ -6,61 +6,16 @@ import (
 	"strconv"
 	"strings"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getPodsLabels returns labels for k8s pods obtained from the given cfg
-func getPodsLabels(cfg *apiConfig) ([]map[string]string, error) {
-	pods, err := getPods(cfg)
-	if err != nil {
-		return nil, err
-	}
-	var ms []map[string]string
-	for _, p := range pods {
-		ms = p.appendTargetLabels(ms)
-	}
-	return ms, nil
-}
-
-func getPods(cfg *apiConfig) ([]Pod, error) {
-	if len(cfg.namespaces) == 0 {
-		return getPodsByPath(cfg, "/api/v1/pods")
-	}
-	// 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 []Pod
-	for _, ns := range namespaces {
-		path := fmt.Sprintf("/api/v1/namespaces/%s/pods", ns)
-		pods, err := getPodsByPath(cfg, path)
-		if err != nil {
-			return nil, err
-		}
-		result = append(result, pods...)
-	}
-	return result, nil
-}
-
-func getPodsByPath(cfg *apiConfig, path string) ([]Pod, error) {
-	data, err := getAPIResponse(cfg, "pod", path)
-	if err != nil {
-		return nil, fmt.Errorf("cannot obtain pods data from API server: %w", err)
-	}
-	pl, err := parsePodList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse pods response from API server: %w", err)
-	}
-	return pl.Items, nil
-}
-
 // PodList implements k8s pod list.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#podlist-v1-core
 type PodList struct {
-	Items []Pod
+	Items    []Pod
+	Metadata listMetadata `json:"metadata"`
 }
 
 // Pod implements k8s pod.
@@ -72,6 +27,10 @@ type Pod struct {
 	Status   PodStatus
 }
 
+func (p Pod) key() string {
+	return p.Metadata.Namespace + "/" + p.Metadata.Name
+}
+
 // PodSpec implements k8s pod spec.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#podspec-v1-core
@@ -211,12 +170,22 @@ func getPodReadyStatus(conds []PodCondition) string {
 	return "unknown"
 }
 
-func getPod(pods []Pod, namespace, name string) *Pod {
-	for i := range pods {
-		pod := &pods[i]
-		if pod.Metadata.Name == name && pod.Metadata.Namespace == namespace {
-			return pod
+func processPods(cfg *apiConfig, p *Pod, action string) {
+	key := buildSyncKey("pods", cfg.setName, p.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		cfg.targetChan <- SyncEvent{
+			Labels:           p.appendTargetLabels(nil),
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
 		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
 	}
-	return nil
 }
diff --git a/lib/promscrape/discovery/kubernetes/service.go b/lib/promscrape/discovery/kubernetes/service.go
index 5ce94a4c9..3723493bb 100644
--- a/lib/promscrape/discovery/kubernetes/service.go
+++ b/lib/promscrape/discovery/kubernetes/service.go
@@ -4,61 +4,16 @@ import (
 	"encoding/json"
 	"fmt"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
 )
 
-// getServicesLabels returns labels for k8s services obtained from the given cfg.
-func getServicesLabels(cfg *apiConfig) ([]map[string]string, error) {
-	svcs, err := getServices(cfg)
-	if err != nil {
-		return nil, err
-	}
-	var ms []map[string]string
-	for _, svc := range svcs {
-		ms = svc.appendTargetLabels(ms)
-	}
-	return ms, nil
-}
-
-func getServices(cfg *apiConfig) ([]Service, error) {
-	if len(cfg.namespaces) == 0 {
-		return getServicesByPath(cfg, "/api/v1/services")
-	}
-	// 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 []Service
-	for _, ns := range namespaces {
-		path := fmt.Sprintf("/api/v1/namespaces/%s/services", ns)
-		svcs, err := getServicesByPath(cfg, path)
-		if err != nil {
-			return nil, err
-		}
-		result = append(result, svcs...)
-	}
-	return result, nil
-}
-
-func getServicesByPath(cfg *apiConfig, path string) ([]Service, error) {
-	data, err := getAPIResponse(cfg, "service", path)
-	if err != nil {
-		return nil, fmt.Errorf("cannot obtain services data from API server: %w", err)
-	}
-	sl, err := parseServiceList(data)
-	if err != nil {
-		return nil, fmt.Errorf("cannot parse services response from API server: %w", err)
-	}
-	return sl.Items, nil
-}
-
 // ServiceList is k8s service list.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#servicelist-v1-core
 type ServiceList struct {
-	Items []Service
+	Items    []Service
+	Metadata listMetadata `json:"metadata"`
 }
 
 // Service is k8s service.
@@ -69,6 +24,10 @@ type Service struct {
 	Spec     ServiceSpec
 }
 
+func (s Service) key() string {
+	return s.Metadata.Namespace + "/" + s.Metadata.Name
+}
+
 // ServiceSpec is k8s service spec.
 //
 // See https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.17/#servicespec-v1-core
@@ -127,12 +86,22 @@ func (s *Service) appendCommonLabels(m map[string]string) {
 	s.Metadata.registerLabelsAndAnnotations("__meta_kubernetes_service", m)
 }
 
-func getService(svcs []Service, namespace, name string) *Service {
-	for i := range svcs {
-		svc := &svcs[i]
-		if svc.Metadata.Name == name && svc.Metadata.Namespace == namespace {
-			return svc
+func processService(cfg *apiConfig, svc *Service, action string) {
+	key := buildSyncKey("service", cfg.setName, svc.key())
+	switch action {
+	case "ADDED", "MODIFIED":
+		cfg.targetChan <- SyncEvent{
+			Labels:           svc.appendTargetLabels(nil),
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
+		}
+	case "DELETED":
+		cfg.targetChan <- SyncEvent{
+			Key:              key,
+			ConfigSectionSet: cfg.setName,
 		}
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
 	}
-	return nil
 }
diff --git a/lib/promscrape/discovery/kubernetes/shared_cache.go b/lib/promscrape/discovery/kubernetes/shared_cache.go
new file mode 100644
index 000000000..72c763c39
--- /dev/null
+++ b/lib/promscrape/discovery/kubernetes/shared_cache.go
@@ -0,0 +1,76 @@
+package kubernetes
+
+import (
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+// SharedKubernetesCache holds cache of kubernetes objects for current config.
+type SharedKubernetesCache struct {
+	Endpoints       *sync.Map
+	EndpointsSlices *sync.Map
+	Pods            *sync.Map
+	Services        *sync.Map
+}
+
+// NewSharedKubernetesCache returns new cache.
+func NewSharedKubernetesCache() *SharedKubernetesCache {
+	return &SharedKubernetesCache{
+		Endpoints:       new(sync.Map),
+		EndpointsSlices: new(sync.Map),
+		Pods:            new(sync.Map),
+		Services:        new(sync.Map),
+	}
+}
+
+func updatePodCache(cache *sync.Map, p *Pod, action string) {
+	switch action {
+	case "ADDED":
+		cache.Store(p.key(), p)
+	case "DELETED":
+		cache.Delete(p.key())
+	case "MODIFIED":
+		cache.Store(p.key(), p)
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+}
+
+func updateServiceCache(cache *sync.Map, p *Service, action string) {
+	switch action {
+	case "ADDED", "MODIFIED":
+		cache.Store(p.key(), p)
+	case "DELETED":
+		cache.Delete(p.key())
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+
+}
+
+func updateEndpointsCache(cache *sync.Map, p *Endpoints, action string) {
+	switch action {
+	case "ADDED", "MODIFIED":
+		cache.Store(p.key(), p)
+	case "DELETED":
+		cache.Delete(p.key())
+	case "ERROR":
+	default:
+		logger.Warnf("unexpected action: %s", action)
+	}
+}
+
+func updateEndpointsSliceCache(cache *sync.Map, p *EndpointSlice, action string) {
+	switch action {
+	case "ADDED", "MODIFIED":
+		cache.Store(p.key(), p)
+	case "DELETED":
+		cache.Delete(p.key())
+	case "ERROR":
+	default:
+		logger.Infof("unexpected action: %s", action)
+	}
+}
diff --git a/lib/promscrape/discovery/kubernetes/watch.go b/lib/promscrape/discovery/kubernetes/watch.go
new file mode 100644
index 000000000..1cec161a7
--- /dev/null
+++ b/lib/promscrape/discovery/kubernetes/watch.go
@@ -0,0 +1,510 @@
+package kubernetes
+
+import (
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+)
+
+// SyncEvent represent kubernetes resource watch event.
+type SyncEvent struct {
+	// object type + set name + ns + name
+	// must be unique.
+	Key string
+	// Labels targets labels for given resource
+	Labels []map[string]string
+	// job name + position id
+	ConfigSectionSet string
+}
+
+type watchResponse struct {
+	Action string          `json:"type"`
+	Object json.RawMessage `json:"object"`
+}
+
+// WatchConfig holds objects for watch handler start.
+type WatchConfig struct {
+	Ctx       context.Context
+	SC        *SharedKubernetesCache
+	WG        *sync.WaitGroup
+	WatchChan chan SyncEvent
+}
+
+// NewWatchConfig returns new config with given context.
+func NewWatchConfig(ctx context.Context) *WatchConfig {
+	return &WatchConfig{
+		Ctx:       ctx,
+		SC:        NewSharedKubernetesCache(),
+		WG:        new(sync.WaitGroup),
+		WatchChan: make(chan SyncEvent, 100),
+	}
+}
+
+func buildSyncKey(objType string, setName string, objKey string) string {
+	return objType + "/" + setName + "/" + objKey
+}
+
+func startWatcherByRole(ctx context.Context, role string, cfg *apiConfig, sc *SharedKubernetesCache) []map[string]string {
+	var ms []map[string]string
+	switch role {
+	case "pod":
+		startWatchForObject(ctx, cfg, "pods", func(wr *watchResponse) {
+			var p Pod
+			if err := json.Unmarshal(wr.Object, &p); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processPods(cfg, &p, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			pods, err := parsePodList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, pod := range pods.Items {
+				ms = pod.appendTargetLabels(ms)
+				processPods(cfg, &pod, "ADDED")
+			}
+			return pods.Metadata.ResourceVersion, nil
+		})
+	case "node":
+		startWatchForObject(ctx, cfg, "nodes", func(wr *watchResponse) {
+			var n Node
+			if err := json.Unmarshal(wr.Object, &n); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processNode(cfg, &n, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			nodes, err := parseNodeList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, node := range nodes.Items {
+				processNode(cfg, &node, "ADDED")
+				ms = node.appendTargetLabels(ms)
+			}
+			return nodes.Metadata.ResourceVersion, nil
+		})
+	case "endpoints":
+		startWatchForObject(ctx, cfg, "pods", func(wr *watchResponse) {
+			var p Pod
+			if err := json.Unmarshal(wr.Object, &p); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			updatePodCache(sc.Pods, &p, wr.Action)
+			if wr.Action == "MODIFIED" {
+				eps, ok := sc.Endpoints.Load(p.key())
+				if ok {
+					ep := eps.(*Endpoints)
+					processEndpoints(cfg, sc, ep, wr.Action)
+				}
+			}
+		}, func(bytes []byte) (string, error) {
+			pods, err := parsePodList(bytes)
+			if err != nil {
+				return "", err
+			}
+			for _, pod := range pods.Items {
+				updatePodCache(sc.Pods, &pod, "ADDED")
+			}
+			return pods.Metadata.ResourceVersion, nil
+		})
+		startWatchForObject(ctx, cfg, "services", func(wr *watchResponse) {
+			var svc Service
+			if err := json.Unmarshal(wr.Object, &svc); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			updateServiceCache(sc.Services, &svc, wr.Action)
+			if wr.Action == "MODIFIED" {
+				linkedEps, ok := sc.Endpoints.Load(svc.key())
+				if ok {
+					ep := linkedEps.(*Endpoints)
+					processEndpoints(cfg, sc, ep, wr.Action)
+				}
+			}
+		}, func(bytes []byte) (string, error) {
+			svcs, err := parseServiceList(bytes)
+			if err != nil {
+				return "", err
+			}
+			for _, svc := range svcs.Items {
+				updateServiceCache(sc.Services, &svc, "ADDED")
+			}
+			return svcs.Metadata.ResourceVersion, nil
+		})
+		startWatchForObject(ctx, cfg, "endpoints", func(wr *watchResponse) {
+			var eps Endpoints
+			if err := json.Unmarshal(wr.Object, &eps); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processEndpoints(cfg, sc, &eps, wr.Action)
+			updateEndpointsCache(sc.Endpoints, &eps, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			eps, err := parseEndpointsList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, ep := range eps.Items {
+				ms = ep.appendTargetLabels(ms, sc.Pods, sc.Services)
+				processEndpoints(cfg, sc, &ep, "ADDED")
+				updateEndpointsCache(sc.Endpoints, &ep, "ADDED")
+			}
+			return eps.Metadata.ResourceVersion, nil
+		})
+	case "service":
+		startWatchForObject(ctx, cfg, "services", func(wr *watchResponse) {
+			var svc Service
+			if err := json.Unmarshal(wr.Object, &svc); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processService(cfg, &svc, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			svcs, err := parseServiceList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, svc := range svcs.Items {
+				processService(cfg, &svc, "ADDED")
+				ms = svc.appendTargetLabels(ms)
+			}
+			return svcs.Metadata.ResourceVersion, nil
+		})
+	case "ingress":
+		startWatchForObject(ctx, cfg, "ingresses", func(wr *watchResponse) {
+			var ig Ingress
+			if err := json.Unmarshal(wr.Object, &ig); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processIngress(cfg, &ig, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			igs, err := parseIngressList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, ig := range igs.Items {
+				processIngress(cfg, &ig, "ADDED")
+				ms = ig.appendTargetLabels(ms)
+			}
+			return igs.Metadata.ResourceVersion, nil
+		})
+	case "endpointslices":
+		startWatchForObject(ctx, cfg, "pods", func(wr *watchResponse) {
+			var p Pod
+			if err := json.Unmarshal(wr.Object, &p); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			updatePodCache(sc.Pods, &p, wr.Action)
+			if wr.Action == "MODIFIED" {
+				eps, ok := sc.EndpointsSlices.Load(p.key())
+				if ok {
+					ep := eps.(*EndpointSlice)
+					processEndpointSlices(cfg, sc, ep, wr.Action)
+				}
+			}
+		}, func(bytes []byte) (string, error) {
+			pods, err := parsePodList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, pod := range pods.Items {
+				updatePodCache(sc.Pods, &pod, "ADDED")
+			}
+			return pods.Metadata.ResourceVersion, nil
+		})
+		startWatchForObject(ctx, cfg, "services", func(wr *watchResponse) {
+			var svc Service
+			if err := json.Unmarshal(wr.Object, &svc); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			updateServiceCache(sc.Services, &svc, wr.Action)
+			if wr.Action == "MODIFIED" {
+				linkedEps, ok := sc.EndpointsSlices.Load(svc.key())
+				if ok {
+					ep := linkedEps.(*EndpointSlice)
+					processEndpointSlices(cfg, sc, ep, wr.Action)
+				}
+			}
+		}, func(bytes []byte) (string, error) {
+			svcs, err := parseServiceList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, svc := range svcs.Items {
+				updateServiceCache(sc.Services, &svc, "ADDED")
+			}
+			return svcs.Metadata.ResourceVersion, nil
+		})
+		startWatchForObject(ctx, cfg, "endpointslices", func(wr *watchResponse) {
+			var eps EndpointSlice
+			if err := json.Unmarshal(wr.Object, &eps); err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return
+			}
+			processEndpointSlices(cfg, sc, &eps, wr.Action)
+			updateEndpointsSliceCache(sc.EndpointsSlices, &eps, wr.Action)
+		}, func(bytes []byte) (string, error) {
+			epss, err := parseEndpointSlicesList(bytes)
+			if err != nil {
+				logger.Errorf("failed to parse object, err: %v", err)
+				return "", err
+			}
+			for _, eps := range epss.Items {
+				ms = eps.appendTargetLabels(ms, sc.Pods, sc.Services)
+				processEndpointSlices(cfg, sc, &eps, "ADDED")
+			}
+			return epss.Metadata.ResourceVersion, nil
+		})
+	default:
+		logger.Errorf("unexpected role: %s", role)
+	}
+	return ms
+}
+
+func startWatchForObject(ctx context.Context, cfg *apiConfig, objectName string, wh func(wr *watchResponse), getSync func([]byte) (string, error)) {
+	if len(cfg.namespaces) > 0 {
+		for _, ns := range cfg.namespaces {
+			path := fmt.Sprintf("/api/v1/namespaces/%s/%s", ns, objectName)
+			// special case.
+			if objectName == "endpointslices" {
+				path = fmt.Sprintf("/apis/discovery.k8s.io/v1beta1/namespaces/%s/%s", ns, objectName)
+			}
+			query := joinSelectors(objectName, nil, cfg.selectors)
+			if len(query) > 0 {
+				path += "?" + query
+			}
+			data, err := cfg.wc.getBlockingAPIResponse(path)
+			if err != nil {
+				logger.Errorf("cannot get latest resource version: %v", err)
+			}
+			version, err := getSync(data)
+			if err != nil {
+				logger.Errorf("cannot get latest resource version: %v", err)
+			}
+			cfg.wc.wg.Add(1)
+			go func(path, version string) {
+				cfg.wc.startWatchForResource(ctx, path, wh, version)
+			}(path, version)
+		}
+	} else {
+		path := "/api/v1/" + objectName
+		if objectName == "endpointslices" {
+			// special case.
+			path = fmt.Sprintf("/apis/discovery.k8s.io/v1beta1/%s", objectName)
+		}
+		query := joinSelectors(objectName, nil, cfg.selectors)
+		if len(query) > 0 {
+			path += "?" + query
+		}
+		data, err := cfg.wc.getBlockingAPIResponse(path)
+		if err != nil {
+			logger.Errorf("cannot get latest resource version: %v", err)
+		}
+		version, err := getSync(data)
+		if err != nil {
+			logger.Errorf("cannot get latest resource version: %v", err)
+		}
+		cfg.wc.wg.Add(1)
+		go func() {
+			cfg.wc.startWatchForResource(ctx, path, wh, version)
+		}()
+	}
+}
+
+type watchClient struct {
+	c         *http.Client
+	ac        *promauth.Config
+	apiServer string
+	wg        *sync.WaitGroup
+}
+
+func (wc *watchClient) startWatchForResource(ctx context.Context, path string, wh func(wr *watchResponse), initResourceVersion string) {
+	defer wc.wg.Done()
+	path += "?watch=1"
+	maxBackOff := time.Second * 30
+	backoff := time.Second
+	for {
+		err := wc.getStreamAPIResponse(ctx, path, initResourceVersion, wh)
+		if errors.Is(err, context.Canceled) {
+			return
+		}
+		if !errors.Is(err, io.EOF) {
+			logger.Errorf("got unexpected error : %v", err)
+		}
+		// reset version.
+		initResourceVersion = ""
+		if backoff < maxBackOff {
+			backoff += time.Second * 5
+		}
+		time.Sleep(backoff)
+	}
+}
+
+func (wc *watchClient) getBlockingAPIResponse(path string) ([]byte, error) {
+	req, err := http.NewRequest("GET", wc.apiServer+path, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept-Encoding", "gzip")
+	if wc.ac != nil && wc.ac.Authorization != "" {
+		req.Header.Set("Authorization", wc.ac.Authorization)
+	}
+	resp, err := wc.c.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("get unexpected code: %d, at blocking api request path: %q", resp.StatusCode, path)
+	}
+	if ce := resp.Header.Get("Content-Encoding"); ce == "gzip" {
+		gr, err := gzip.NewReader(resp.Body)
+		if err != nil {
+			return nil, fmt.Errorf("cannot create gzip reader: %w", err)
+		}
+		return ioutil.ReadAll(gr)
+	}
+	return ioutil.ReadAll(resp.Body)
+}
+
+func (wc *watchClient) getStreamAPIResponse(ctx context.Context, path, resouceVersion string, wh func(wr *watchResponse)) error {
+	if resouceVersion != "" {
+		path += "&resourceVersion=" + resouceVersion
+	}
+	req, err := http.NewRequestWithContext(ctx, "GET", wc.apiServer+path, nil)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Accept-Encoding", "gzip")
+	if wc.ac != nil && wc.ac.Authorization != "" {
+		req.Header.Set("Authorization", wc.ac.Authorization)
+	}
+	resp, err := wc.c.Do(req)
+	if err != nil {
+		return err
+	}
+	if resp.StatusCode != 200 {
+		return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
+	br := resp.Body
+	if ce := resp.Header.Get("Content-Encoding"); ce == "gzip" {
+		br, err = gzip.NewReader(resp.Body)
+		if err != nil {
+			return fmt.Errorf("cannot create gzip reader: %w", err)
+		}
+	}
+	r := newJSONFramedReader(br)
+	for {
+		b := make([]byte, 1024)
+		b, err := readJSONObject(r, b)
+		if err != nil {
+			return err
+		}
+		var rObject watchResponse
+		err = json.Unmarshal(b, &rObject)
+		if err != nil {
+			logger.Errorf("failed to parse watch api response as json, err %v, response: %v", err, string(b))
+			continue
+		}
+		wh(&rObject)
+	}
+}
+
+func readJSONObject(r io.Reader, b []byte) ([]byte, error) {
+	offset := 0
+	for {
+		n, err := r.Read(b[offset:])
+		if err == io.ErrShortBuffer {
+			if n == 0 {
+				return nil, fmt.Errorf("got short buffer with n=0, cap=%d", cap(b))
+			}
+			// double buffer..
+			b = bytesutil.Resize(b, len(b)*2)
+			offset += n
+			continue
+		}
+		if err != nil {
+			return nil, err
+		}
+		offset += n
+		break
+	}
+	return b[:offset], nil
+}
+
+func newWatchClient(wg *sync.WaitGroup, sdc *SDConfig, baseDir string) (*watchClient, error) {
+	ac, err := promauth.NewConfig(baseDir, sdc.BasicAuth, sdc.BearerToken, sdc.BearerTokenFile, sdc.TLSConfig)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse auth config: %w", err)
+	}
+	apiServer := sdc.APIServer
+	if len(apiServer) == 0 {
+		// Assume we run at k8s pod.
+		// Discover apiServer and auth config according to k8s docs.
+		// See https://kubernetes.io/docs/reference/access-authn-authz/service-accounts-admin/#service-account-admission-controller
+		host := os.Getenv("KUBERNETES_SERVICE_HOST")
+		port := os.Getenv("KUBERNETES_SERVICE_PORT")
+		if len(host) == 0 {
+			return nil, fmt.Errorf("cannot find KUBERNETES_SERVICE_HOST env var; it must be defined when running in k8s; " +
+				"probably, `kubernetes_sd_config->api_server` is missing in Prometheus configs?")
+		}
+		if len(port) == 0 {
+			return nil, fmt.Errorf("cannot find KUBERNETES_SERVICE_PORT env var; it must be defined when running in k8s; "+
+				"KUBERNETES_SERVICE_HOST=%q", host)
+		}
+		apiServer = "https://" + net.JoinHostPort(host, port)
+		tlsConfig := promauth.TLSConfig{
+			CAFile: "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
+		}
+		acNew, err := promauth.NewConfig(".", nil, "", "/var/run/secrets/kubernetes.io/serviceaccount/token", &tlsConfig)
+		if err != nil {
+			return nil, fmt.Errorf("cannot initialize service account auth: %w; probably, `kubernetes_sd_config->api_server` is missing in Prometheus configs?", err)
+		}
+		ac = acNew
+	}
+	var proxy func(*http.Request) (*url.URL, error)
+	if proxyURL := sdc.ProxyURL.URL(); proxyURL != nil {
+		proxy = http.ProxyURL(proxyURL)
+	}
+	c := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig:     ac.NewTLSConfig(),
+			Proxy:               proxy,
+			TLSHandshakeTimeout: 10 * time.Second,
+			IdleConnTimeout:     2 * time.Minute,
+		},
+	}
+	wc := watchClient{
+		c:         c,
+		apiServer: apiServer,
+		ac:        ac,
+		wg:        wg,
+	}
+	return &wc, nil
+}
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index 5e3aef789..ddef4861d 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -23,6 +23,7 @@ var (
 	kubernetesSDCheckInterval = flag.Duration("promscrape.kubernetesSDCheckInterval", 30*time.Second, "Interval for checking for changes in Kubernetes API server. "+
 		"This works only if `kubernetes_sd_configs` is configured in '-promscrape.config' file. "+
 		"See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#kubernetes_sd_config for details")
+
 	openstackSDCheckInterval = flag.Duration("promscrape.openstackSDCheckInterval", 30*time.Second, "Interval for checking for changes in openstack API server. "+
 		"This works only if `openstack_sd_configs` is configured in '-promscrape.config' file. "+
 		"See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#openstack_sd_config for details")
@@ -97,7 +98,9 @@ func runScraper(configFile string, pushData func(wr *prompbmarshal.WriteRequest)
 	scs := newScrapeConfigs(pushData)
 	scs.add("static_configs", 0, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getStaticScrapeWork() })
 	scs.add("file_sd_configs", *fileSDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getFileSDScrapeWork(swsPrev) })
-	scs.add("kubernetes_sd_configs", *kubernetesSDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getKubernetesSDScrapeWork(swsPrev) })
+	scs.add("kubernetes_sd_configs", *kubernetesSDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork {
+		return getKubernetesSDScrapeWorkStream(cfg, swsPrev)
+	})
 	scs.add("openstack_sd_configs", *openstackSDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getOpenStackSDScrapeWork(swsPrev) })
 	scs.add("consul_sd_configs", *consul.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getConsulSDScrapeWork(swsPrev) })
 	scs.add("eureka_sd_configs", *eurekaSDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getEurekaSDScrapeWork(swsPrev) })
diff --git a/lib/promscrape/watch_handler.go b/lib/promscrape/watch_handler.go
new file mode 100644
index 000000000..d940e08fe
--- /dev/null
+++ b/lib/promscrape/watch_handler.go
@@ -0,0 +1,83 @@
+package promscrape
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kubernetes"
+)
+
+type kubernetesWatchHandler struct {
+	ctx       context.Context
+	cancel    context.CancelFunc
+	startOnce sync.Once
+	watchCfg  *kubernetes.WatchConfig
+	// guards cache and set
+	mu             sync.Mutex
+	lastAccessTime time.Time
+	swCache        map[string][]*ScrapeWork
+	sdcSet         map[string]*scrapeWorkConfig
+}
+
+func newKubernetesWatchHandler() *kubernetesWatchHandler {
+	ctx, cancel := context.WithCancel(context.Background())
+	kwh := &kubernetesWatchHandler{
+		ctx:      ctx,
+		cancel:   cancel,
+		swCache:  map[string][]*ScrapeWork{},
+		sdcSet:   map[string]*scrapeWorkConfig{},
+		watchCfg: kubernetes.NewWatchConfig(ctx),
+	}
+	go kwh.waitForStop()
+	return kwh
+}
+
+func (ksc *kubernetesWatchHandler) waitForStop() {
+	t := time.NewTicker(time.Second * 5)
+	for range t.C {
+		ksc.mu.Lock()
+		lastTime := time.Since(ksc.lastAccessTime)
+		ksc.mu.Unlock()
+		if lastTime > *kubernetesSDCheckInterval*30 {
+			t1 := time.Now()
+			ksc.cancel()
+			ksc.watchCfg.WG.Wait()
+			close(ksc.watchCfg.WatchChan)
+			logger.Infof("stopped kubernetes api watcher handler, after: %.3f seconds", time.Since(t1).Seconds())
+			ksc.watchCfg.SC = nil
+			t.Stop()
+			return
+		}
+	}
+}
+
+func processKubernetesSyncEvents(cfg *Config) {
+	for {
+		select {
+		case <-cfg.kwh.ctx.Done():
+			return
+		case se, ok := <-cfg.kwh.watchCfg.WatchChan:
+			if !ok {
+				return
+			}
+			if se.Labels == nil {
+				cfg.kwh.mu.Lock()
+				delete(cfg.kwh.swCache, se.Key)
+				cfg.kwh.mu.Unlock()
+				continue
+			}
+			cfg.kwh.mu.Lock()
+			swc, ok := cfg.kwh.sdcSet[se.ConfigSectionSet]
+			cfg.kwh.mu.Unlock()
+			if !ok {
+				logger.Fatalf("bug config section not found: %v", se.ConfigSectionSet)
+			}
+			ms := appendScrapeWorkForTargetLabels(nil, swc, se.Labels, "kubernetes_sd_config")
+			cfg.kwh.mu.Lock()
+			cfg.kwh.swCache[se.Key] = ms
+			cfg.kwh.mu.Unlock()
+		}
+	}
+}


Test patch:
diff --git a/lib/promscrape/discovery/kubernetes/endpoints_test.go b/lib/promscrape/discovery/kubernetes/endpoints_test.go
index e5f0f4251..8f154b5c6 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints_test.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints_test.go
@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"reflect"
+	"sync"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -89,7 +90,8 @@ func TestParseEndpointsListSuccess(t *testing.T) {
 	endpoint := els.Items[0]
 
 	// Check endpoint.appendTargetLabels()
-	labelss := endpoint.appendTargetLabels(nil, nil, nil)
+	var pc, sc sync.Map
+	labelss := endpoint.appendTargetLabels(nil, &pc, &sc)
 	var sortedLabelss [][]prompbmarshal.Label
 	for _, labels := range labelss {
 		sortedLabelss = append(sortedLabelss, discoveryutils.GetSortedLabels(labels))
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices_test.go b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
index 900751b74..b1c3731ef 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices_test.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
@@ -2,6 +2,7 @@ package kubernetes
 
 import (
 	"reflect"
+	"sync"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -186,7 +187,8 @@ func Test_parseEndpointSlicesListSuccess(t *testing.T) {
 	}
 
 	firstEsl := esl.Items[0]
-	got := firstEsl.appendTargetLabels(nil, nil, nil)
+	var pc, sc sync.Map
+	got := firstEsl.appendTargetLabels(nil, &pc, &sc)
 	sortedLables := [][]prompbmarshal.Label{}
 	for _, labels := range got {
 		sortedLables = append(sortedLables, discoveryutils.GetSortedLabels(labels))
@@ -439,7 +441,17 @@ func TestEndpointSlice_appendTargetLabels(t *testing.T) {
 				AddressType: tt.fields.AddressType,
 				Ports:       tt.fields.Ports,
 			}
-			got := eps.appendTargetLabels(tt.args.ms, tt.args.pods, tt.args.svcs)
+			pc := sync.Map{}
+			sc := sync.Map{}
+			for _, p := range tt.args.pods {
+				p := &p
+				pc.Store(p.key(), p)
+			}
+			for _, s := range tt.args.svcs {
+				s := &s
+				sc.Store(s.key(), s)
+			}
+			got := eps.appendTargetLabels(tt.args.ms, &pc, &sc)
 			var sortedLabelss [][]prompbmarshal.Label
 			for _, labels := range got {
 				sortedLabelss = append(sortedLabelss, discoveryutils.GetSortedLabels(labels))