Repository: VictoriaMetrics/VictoriaMetrics
Pull request #983 (2020-12-24 08:44:21)

fixes kubernetes_sd
adds missing service metadata for pod ports without endpoint


Patch:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index aba2c78f9..3a05f6677 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 # tip
 
+* FEATURE: vmagent: add support for `proxy_url` config option in Prometheus scrape configs. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/503
 * FEATURE: remove parts with stale data as soon as they go outside the configured `-retentionPeriod`. Previously such parts may remain active for long periods of time. This should help reducing disk usage for `-retentionPeriod` smaller than one month.
 * FEATURE: vmalert: allow setting multiple values for `-notifier.tlsInsecureSkipVerify` command-line flag per each `-notifier.url`.
 
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index 98e1ff230..bec3ef23e 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -8,6 +8,7 @@ import (
 	"io"
 	"io/ioutil"
 	"net/http"
+	"net/url"
 	"strings"
 	"time"
 
@@ -66,10 +67,14 @@ func newClient(sw *ScrapeWork) *client {
 			host += ":443"
 		}
 	}
+	dialFunc, err := newStatDialFunc(sw.ProxyURL, tlsCfg)
+	if err != nil {
+		logger.Fatalf("cannot create dial func: %s", err)
+	}
 	hc := &fasthttp.HostClient{
 		Addr:                         host,
 		Name:                         "vm_promscrape",
-		Dial:                         statDial,
+		Dial:                         dialFunc,
 		IsTLS:                        isTLS,
 		TLSConfig:                    tlsCfg,
 		MaxIdleConnDuration:          2 * sw.ScrapeInterval,
@@ -80,9 +85,14 @@ func newClient(sw *ScrapeWork) *client {
 	}
 	var sc *http.Client
 	if *streamParse || sw.StreamParse {
+		var proxy func(*http.Request) (*url.URL, error)
+		if proxyURL := sw.ProxyURL.URL(); proxyURL != nil {
+			proxy = http.ProxyURL(proxyURL)
+		}
 		sc = &http.Client{
 			Transport: &http.Transport{
 				TLSClientConfig:     tlsCfg,
+				Proxy:               proxy,
 				TLSHandshakeTimeout: 10 * time.Second,
 				IdleConnTimeout:     2 * sw.ScrapeInterval,
 				DisableCompression:  *disableCompression || sw.DisableCompression,
@@ -93,9 +103,8 @@ func newClient(sw *ScrapeWork) *client {
 		}
 	}
 	return &client{
-		hc: hc,
-		sc: sc,
-
+		hc:                 hc,
+		sc:                 sc,
 		scrapeURL:          sw.ScrapeURL,
 		host:               host,
 		requestURI:         requestURI,
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index e4aaca37e..31e1ce858 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -23,6 +23,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/gce"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kubernetes"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/openstack"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 	"gopkg.in/yaml.v2"
 )
 
@@ -71,6 +72,7 @@ type ScrapeConfig struct {
 	BasicAuth            *promauth.BasicAuthConfig   `yaml:"basic_auth,omitempty"`
 	BearerToken          string                      `yaml:"bearer_token,omitempty"`
 	BearerTokenFile      string                      `yaml:"bearer_token_file,omitempty"`
+	ProxyURL             proxy.URL                   `yaml:"proxy_url,omitempty"`
 	TLSConfig            *promauth.TLSConfig         `yaml:"tls_config,omitempty"`
 	StaticConfigs        []StaticConfig              `yaml:"static_configs,omitempty"`
 	FileSDConfigs        []FileSDConfig              `yaml:"file_sd_configs,omitempty"`
@@ -495,6 +497,7 @@ func getScrapeWorkConfig(sc *ScrapeConfig, baseDir string, globalCfg *GlobalConf
 		metricsPath:          metricsPath,
 		scheme:               scheme,
 		params:               params,
+		proxyURL:             sc.ProxyURL,
 		authConfig:           ac,
 		honorLabels:          honorLabels,
 		honorTimestamps:      honorTimestamps,
@@ -516,6 +519,7 @@ type scrapeWorkConfig struct {
 	metricsPath          string
 	scheme               string
 	params               map[string][]string
+	proxyURL             proxy.URL
 	authConfig           *promauth.Config
 	honorLabels          bool
 	honorTimestamps      bool
@@ -750,6 +754,7 @@ func appendScrapeWork(dst []*ScrapeWork, swc *scrapeWorkConfig, target string, e
 		HonorTimestamps:      swc.honorTimestamps,
 		OriginalLabels:       originalLabels,
 		Labels:               labels,
+		ProxyURL:             swc.proxyURL,
 		AuthConfig:           swc.authConfig,
 		MetricRelabelConfigs: swc.metricRelabelConfigs,
 		SampleLimit:          swc.sampleLimit,
diff --git a/lib/promscrape/discovery/consul/api.go b/lib/promscrape/discovery/consul/api.go
index 6a888eabb..eec5a3c9a 100644
--- a/lib/promscrape/discovery/consul/api.go
+++ b/lib/promscrape/discovery/consul/api.go
@@ -58,7 +58,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		}
 		apiServer = scheme + "://" + apiServer
 	}
-	client, err := discoveryutils.NewClient(apiServer, ac)
+	client, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
 	}
diff --git a/lib/promscrape/discovery/consul/consul.go b/lib/promscrape/discovery/consul/consul.go
index 6376d4062..9c980f779 100644
--- a/lib/promscrape/discovery/consul/consul.go
+++ b/lib/promscrape/discovery/consul/consul.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 )
 
 // SDConfig represents service discovery config for Consul.
@@ -16,6 +17,7 @@ type SDConfig struct {
 	Scheme       string              `yaml:"scheme,omitempty"`
 	Username     string              `yaml:"username"`
 	Password     string              `yaml:"password"`
+	ProxyURL     proxy.URL           `yaml:"proxy_url,omitempty"`
 	TLSConfig    *promauth.TLSConfig `yaml:"tls_config,omitempty"`
 	Services     []string            `yaml:"services,omitempty"`
 	Tags         []string            `yaml:"tags,omitempty"`
diff --git a/lib/promscrape/discovery/dockerswarm/api.go b/lib/promscrape/discovery/dockerswarm/api.go
index 3853600a4..c79d12fc7 100644
--- a/lib/promscrape/discovery/dockerswarm/api.go
+++ b/lib/promscrape/discovery/dockerswarm/api.go
@@ -34,11 +34,12 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		port:            sdc.Port,
 		filtersQueryArg: getFiltersQueryArg(sdc.Filters),
 	}
+
 	ac, err := promauth.NewConfig(baseDir, sdc.BasicAuth, sdc.BearerToken, sdc.BearerTokenFile, sdc.TLSConfig)
 	if err != nil {
 		return nil, err
 	}
-	client, err := discoveryutils.NewClient(sdc.Host, ac)
+	client, err := discoveryutils.NewClient(sdc.Host, ac, sdc.ProxyURL)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", sdc.Host, err)
 	}
diff --git a/lib/promscrape/discovery/dockerswarm/dockerswarm.go b/lib/promscrape/discovery/dockerswarm/dockerswarm.go
index 61b4b02f6..bd9564306 100644
--- a/lib/promscrape/discovery/dockerswarm/dockerswarm.go
+++ b/lib/promscrape/discovery/dockerswarm/dockerswarm.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 )
 
 // SDConfig represents docker swarm service discovery configuration
@@ -15,7 +16,7 @@ type SDConfig struct {
 	Port    int      `yaml:"port,omitempty"`
 	Filters []Filter `yaml:"filters,omitempty"`
 
-	// TODO: add support for proxy_url
+	ProxyURL  proxy.URL           `yaml:"proxy_url,omitempty"`
 	TLSConfig *promauth.TLSConfig `yaml:"tls_config,omitempty"`
 	// refresh_interval is obtained from `-promscrape.dockerswarmSDCheckInterval` command-line option
 	BasicAuth       *promauth.BasicAuthConfig `yaml:"basic_auth,omitempty"`
diff --git a/lib/promscrape/discovery/eureka/api.go b/lib/promscrape/discovery/eureka/api.go
index f1a319bc3..255c9005d 100644
--- a/lib/promscrape/discovery/eureka/api.go
+++ b/lib/promscrape/discovery/eureka/api.go
@@ -43,7 +43,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		}
 		apiServer = scheme + "://" + apiServer
 	}
-	client, err := discoveryutils.NewClient(apiServer, ac)
+	client, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
 	}
diff --git a/lib/promscrape/discovery/eureka/eureka.go b/lib/promscrape/discovery/eureka/eureka.go
index 5dca81937..c8ebc55ef 100644
--- a/lib/promscrape/discovery/eureka/eureka.go
+++ b/lib/promscrape/discovery/eureka/eureka.go
@@ -5,9 +5,9 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 )
 
 const appsAPIPath = "/apps"
@@ -22,6 +22,7 @@ type SDConfig struct {
 	Scheme     string              `yaml:"scheme,omitempty"`
 	Username   string              `yaml:"username"`
 	Password   string              `yaml:"password"`
+	ProxyURL   proxy.URL           `yaml:"proxy_url,omitempty"`
 	TLSConfig  *promauth.TLSConfig `yaml:"tls_config,omitempty"`
 	// RefreshInterval time.Duration `yaml:"refresh_interval"`
 	// refresh_interval is obtained from `-promscrape.ec2SDCheckInterval` command-line option.
diff --git a/lib/promscrape/discovery/kubernetes/api.go b/lib/promscrape/discovery/kubernetes/api.go
index caa330643..2c0463214 100644
--- a/lib/promscrape/discovery/kubernetes/api.go
+++ b/lib/promscrape/discovery/kubernetes/api.go
@@ -56,7 +56,7 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		}
 		ac = acNew
 	}
-	client, err := discoveryutils.NewClient(apiServer, ac)
+	client, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL)
 	if err != nil {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
 	}
diff --git a/lib/promscrape/discovery/kubernetes/endpoints.go b/lib/promscrape/discovery/kubernetes/endpoints.go
index 1bb3aaeb8..edd538746 100644
--- a/lib/promscrape/discovery/kubernetes/endpoints.go
+++ b/lib/promscrape/discovery/kubernetes/endpoints.go
@@ -158,6 +158,9 @@ func (eps *Endpoints) appendTargetLabels(ms []map[string]string, pods []Pod, svc
 				}
 				p.appendCommonLabels(m)
 				p.appendContainerLabels(m, c, &cp)
+				if svc != nil {
+					svc.appendCommonLabels(m)
+				}
 				ms = append(ms, m)
 			}
 		}
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices.go b/lib/promscrape/discovery/kubernetes/endpointslices.go
index 48a06a25c..f1f9da645 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices.go
@@ -113,6 +113,9 @@ func (eps *EndpointSlice) appendTargetLabels(ms []map[string]string, pods []Pod,
 				}
 				p.appendCommonLabels(m)
 				p.appendContainerLabels(m, c, &cp)
+				if svc != nil {
+					svc.appendCommonLabels(m)
+				}
 				ms = append(ms, m)
 			}
 		}
diff --git a/lib/promscrape/discovery/kubernetes/kubernetes.go b/lib/promscrape/discovery/kubernetes/kubernetes.go
index 87c381963..2c7a48995 100644
--- a/lib/promscrape/discovery/kubernetes/kubernetes.go
+++ b/lib/promscrape/discovery/kubernetes/kubernetes.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 )
 
 // SDConfig represents kubernetes-based service discovery config.
@@ -15,6 +16,7 @@ type SDConfig struct {
 	BasicAuth       *promauth.BasicAuthConfig `yaml:"basic_auth,omitempty"`
 	BearerToken     string                    `yaml:"bearer_token,omitempty"`
 	BearerTokenFile string                    `yaml:"bearer_token_file,omitempty"`
+	ProxyURL        proxy.URL                 `yaml:"proxy_url,omitempty"`
 	TLSConfig       *promauth.TLSConfig       `yaml:"tls_config,omitempty"`
 	Namespaces      Namespaces                `yaml:"namespaces,omitempty"`
 	Selectors       []Selector                `yaml:"selectors,omitempty"`
diff --git a/lib/promscrape/discoveryutils/client.go b/lib/promscrape/discoveryutils/client.go
index 9469fe06d..b971be8ec 100644
--- a/lib/promscrape/discoveryutils/client.go
+++ b/lib/promscrape/discoveryutils/client.go
@@ -10,8 +10,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
 	"github.com/VictoriaMetrics/fasthttp"
 )
@@ -45,11 +45,12 @@ type Client struct {
 }
 
 // NewClient returns new Client for the given apiServer and the given ac.
-func NewClient(apiServer string, ac *promauth.Config) (*Client, error) {
+func NewClient(apiServer string, ac *promauth.Config, proxyURL proxy.URL) (*Client, error) {
 	var (
 		dialFunc fasthttp.DialFunc
 		tlsCfg   *tls.Config
 		u        fasthttp.URI
+		err      error
 	)
 	u.Update(apiServer)
 
@@ -61,6 +62,7 @@ func NewClient(apiServer string, ac *promauth.Config) (*Client, error) {
 			return net.Dial("unix", dialAddr)
 		}
 	}
+
 	hostPort := string(u.Host())
 	isTLS := string(u.Scheme()) == "https"
 	if isTLS && ac != nil {
@@ -73,10 +75,15 @@ func NewClient(apiServer string, ac *promauth.Config) (*Client, error) {
 		}
 		hostPort = net.JoinHostPort(hostPort, port)
 	}
+	if dialFunc == nil {
+		dialFunc, err = proxyURL.NewDialFunc(tlsCfg)
+		if err != nil {
+			return nil, err
+		}
+	}
 	hc := &fasthttp.HostClient{
 		Addr:                hostPort,
 		Name:                "vm_promscrape/discovery",
-		DialDualStack:       netutil.TCP6Enabled(),
 		IsTLS:               isTLS,
 		TLSConfig:           tlsCfg,
 		ReadTimeout:         time.Minute,
@@ -88,7 +95,6 @@ func NewClient(apiServer string, ac *promauth.Config) (*Client, error) {
 	blockingClient := &fasthttp.HostClient{
 		Addr:                hostPort,
 		Name:                "vm_promscrape/discovery",
-		DialDualStack:       netutil.TCP6Enabled(),
 		IsTLS:               isTLS,
 		TLSConfig:           tlsCfg,
 		ReadTimeout:         BlockingClientReadTimeout,
diff --git a/lib/promscrape/scrapework.go b/lib/promscrape/scrapework.go
index 62f759d58..30f28e3a3 100644
--- a/lib/promscrape/scrapework.go
+++ b/lib/promscrape/scrapework.go
@@ -17,6 +17,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promrelabel"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/prometheus"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 	"github.com/VictoriaMetrics/metrics"
 	xxhash "github.com/cespare/xxhash/v2"
 )
@@ -70,6 +71,9 @@ type ScrapeWork struct {
 	// Auth config
 	AuthConfig *promauth.Config
 
+	// ProxyURL HTTP proxy url
+	ProxyURL proxy.URL
+
 	// Optional `metric_relabel_configs`.
 	MetricRelabelConfigs []promrelabel.ParsedRelabelConfig
 
diff --git a/lib/promscrape/statconn.go b/lib/promscrape/statconn.go
index 73c6dcc94..2fefaf420 100644
--- a/lib/promscrape/statconn.go
+++ b/lib/promscrape/statconn.go
@@ -2,6 +2,7 @@ package promscrape
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net"
 	"sync"
@@ -9,6 +10,7 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
 	"github.com/VictoriaMetrics/fasthttp"
 	"github.com/VictoriaMetrics/metrics"
 )
@@ -47,25 +49,28 @@ var (
 	stdDialerOnce sync.Once
 )
 
-func statDial(addr string) (conn net.Conn, err error) {
-	if netutil.TCP6Enabled() {
-		conn, err = fasthttp.DialDualStack(addr)
-	} else {
-		conn, err = fasthttp.Dial(addr)
-	}
-	dialsTotal.Inc()
+func newStatDialFunc(proxyURL proxy.URL, tlsConfig *tls.Config) (fasthttp.DialFunc, error) {
+	dialFunc, err := proxyURL.NewDialFunc(tlsConfig)
 	if err != nil {
-		dialErrors.Inc()
-		if !netutil.TCP6Enabled() {
-			err = fmt.Errorf("%w; try -enableTCP6 command-line flag if you scrape ipv6 addresses", err)
-		}
 		return nil, err
 	}
-	conns.Inc()
-	sc := &statConn{
-		Conn: conn,
+	statDialFunc := func(addr string) (net.Conn, error) {
+		conn, err := dialFunc(addr)
+		dialsTotal.Inc()
+		if err != nil {
+			dialErrors.Inc()
+			if !netutil.TCP6Enabled() {
+				err = fmt.Errorf("%w; try -enableTCP6 command-line flag if you scrape ipv6 addresses", err)
+			}
+			return nil, err
+		}
+		conns.Inc()
+		sc := &statConn{
+			Conn: conn,
+		}
+		return sc, nil
 	}
-	return sc, nil
+	return statDialFunc, nil
 }
 
 var (
diff --git a/lib/proxy/proxy.go b/lib/proxy/proxy.go
new file mode 100644
index 000000000..82cb7b46a
--- /dev/null
+++ b/lib/proxy/proxy.go
@@ -0,0 +1,117 @@
+package proxy
+
+import (
+	"bufio"
+	"crypto/tls"
+	"encoding/base64"
+	"fmt"
+	"net"
+	"net/url"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/netutil"
+	"github.com/VictoriaMetrics/fasthttp"
+)
+
+// URL implements YAML.Marshaler and yaml.Unmarshaler interfaces for url.URL.
+type URL struct {
+	url *url.URL
+}
+
+// URL return the underlying url.
+func (u *URL) URL() *url.URL {
+	if u == nil || u.url == nil {
+		return nil
+	}
+	return u.url
+}
+
+// MarshalYAML implements yaml.Marshaler interface.
+func (u *URL) MarshalYAML() (interface{}, error) {
+	if u.url == nil {
+		return nil, nil
+	}
+	return u.url.String(), nil
+}
+
+// UnmarshalYAML implements yaml.Unmarshaler interface.
+func (u *URL) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	var s string
+	if err := unmarshal(&s); err != nil {
+		return err
+	}
+	parsedURL, err := url.Parse(s)
+	if err != nil {
+		return fmt.Errorf("cannot parse proxy_url=%q as *url.URL: %w", s, err)
+	}
+	u.url = parsedURL
+	return nil
+}
+
+// NewDialFunc returns dial func for the given pu and tlsConfig.
+func (u *URL) NewDialFunc(tlsConfig *tls.Config) (fasthttp.DialFunc, error) {
+	if u == nil || u.url == nil {
+		return defaultDialFunc, nil
+	}
+	pu := u.url
+	if pu.Scheme != "http" && pu.Scheme != "https" {
+		return nil, fmt.Errorf("unknown scheme=%q for proxy_url=%q, must be http or https", pu.Scheme, pu)
+	}
+	var authHeader string
+	if pu.User != nil && len(pu.User.Username()) > 0 {
+		userPasswordEncoded := base64.StdEncoding.EncodeToString([]byte(pu.User.String()))
+		authHeader = "Proxy-Authorization: Basic " + userPasswordEncoded + "\r\n"
+	}
+	dialFunc := func(addr string) (net.Conn, error) {
+		proxyConn, err := defaultDialFunc(pu.Host)
+		if err != nil {
+			return nil, fmt.Errorf("cannot connect to proxy %q: %w", pu, err)
+		}
+		if pu.Scheme == "https" {
+			proxyConn = tls.Client(proxyConn, tlsConfig)
+		}
+		conn, err := sendConnectRequest(proxyConn, addr, authHeader)
+		if err != nil {
+			_ = proxyConn.Close()
+			return nil, fmt.Errorf("error when sending CONNECT request to proxy %q: %w", pu, err)
+		}
+		return conn, nil
+	}
+	return dialFunc, nil
+}
+
+func defaultDialFunc(addr string) (net.Conn, error) {
+	if netutil.TCP6Enabled() {
+		return fasthttp.DialDualStack(addr)
+	}
+	return fasthttp.Dial(addr)
+}
+
+// sendConnectRequest sends CONNECT request to proxyConn for the given addr and authHeader and returns the established connection to dstAddr.
+func sendConnectRequest(proxyConn net.Conn, dstAddr, authHeader string) (net.Conn, error) {
+	req := "CONNECT " + dstAddr + " HTTP/1.1\r\nHost: " + dstAddr + "\r\n" + authHeader + "\r\n"
+	if _, err := proxyConn.Write([]byte(req)); err != nil {
+		return nil, fmt.Errorf("cannot send CONNECT request for dstAddr=%q: %w", dstAddr, err)
+	}
+	var res fasthttp.Response
+	res.SkipBody = true
+	conn := &bufferedReaderConn{
+		br:   bufio.NewReader(proxyConn),
+		Conn: proxyConn,
+	}
+	if err := res.Read(conn.br); err != nil {
+		return nil, fmt.Errorf("cannot read CONNECT response for dstAddr=%q: %w", dstAddr, err)
+	}
+	if statusCode := res.Header.StatusCode(); statusCode != 200 {
+		return nil, fmt.Errorf("unexpected status code received: %d; want: 200", statusCode)
+	}
+	return conn, nil
+}
+
+type bufferedReaderConn struct {
+	net.Conn
+	br *bufio.Reader
+}
+
+func (brc *bufferedReaderConn) Read(p []byte) (int, error) {
+	return brc.br.Read(p)
+}


Test patch:
diff --git a/lib/promscrape/discovery/kubernetes/endpointslices_test.go b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
index 44acef3d7..900751b74 100644
--- a/lib/promscrape/discovery/kubernetes/endpointslices_test.go
+++ b/lib/promscrape/discovery/kubernetes/endpointslices_test.go
@@ -420,6 +420,13 @@ func TestEndpointSlice_appendTargetLabels(t *testing.T) {
 					"__meta_kubernetes_pod_phase":                               "",
 					"__meta_kubernetes_pod_ready":                               "unknown",
 					"__meta_kubernetes_pod_uid":                                 "some-pod-uuid",
+					"__meta_kubernetes_service_cluster_ip":                      "",
+					"__meta_kubernetes_service_label_service_label_1":           "value-1",
+					"__meta_kubernetes_service_label_service_label_2":           "value-2",
+					"__meta_kubernetes_service_labelpresent_service_label_1":    "true",
+					"__meta_kubernetes_service_labelpresent_service_label_2":    "true",
+					"__meta_kubernetes_service_name":                            "custom-esl",
+					"__meta_kubernetes_service_type":                            "ClusterIP",
 				}),
 			},
 		},