Merge pull request #1 from asokolov365/spec_http_grpc_src_ip

allow configuration of source ip for http and grpc probers

CONFIGURATION.md

@@ -153,6 +153,9 @@ modules:
   [ preferred_ip_protocol: <string> | default = "ip6" ]
   [ ip_protocol_fallback: <boolean> | default = true ]
 
+  # The source IP address.
+  [ source_ip_address: <string> ]
+
   # The body of the HTTP request used in probe.
   [ body: <string> ]
 
@@ -310,6 +313,9 @@ validate_additional_rrs:
 [ preferred_ip_protocol: <string> ]
 [ ip_protocol_fallback: <boolean> | default = true ]
 
+# The source IP address.
+[ source_ip_address: <string> ]
+
 # Whether to connect to the endpoint with TLS.
 [ tls: <boolean | default = false> ]
 

config/config.go

@@ -208,6 +208,7 @@ type HTTPProbe struct {
 	ValidHTTPVersions            []string                `yaml:"valid_http_versions,omitempty"`
 	IPProtocol                   string                  `yaml:"preferred_ip_protocol,omitempty"`
 	IPProtocolFallback           bool                    `yaml:"ip_protocol_fallback,omitempty"`
+	SourceIPAddress              string                  `yaml:"source_ip_address,omitempty"`
 	SkipResolvePhaseWithProxy    bool                    `yaml:"skip_resolve_phase_with_proxy,omitempty"`
 	NoFollowRedirects            *bool                   `yaml:"no_follow_redirects,omitempty"`
 	FailIfSSL                    bool                    `yaml:"fail_if_ssl,omitempty"`
@@ -231,6 +232,7 @@ type GRPCProbe struct {
 	TLSConfig           config.TLSConfig `yaml:"tls_config,omitempty"`
 	IPProtocolFallback  bool             `yaml:"ip_protocol_fallback,omitempty"`
 	PreferredIPProtocol string           `yaml:"preferred_ip_protocol,omitempty"`
+	SourceIPAddress     string           `yaml:"source_ip_address,omitempty"`
 }
 
 type HeaderMatch struct {

config/testdata/blackbox-good.yml

@@ -3,6 +3,7 @@ modules:
     prober: http
     timeout: 5s
     http:
+      source_ip_address: 127.0.0.1
   http_post_2xx:
     prober: http
     timeout: 5s
@@ -64,6 +65,7 @@ modules:
     dns:
       query_name: example.com
       preferred_ip_protocol: ip4
+      source_ip_address: 127.0.0.1
       ip_protocol_fallback: false
       validate_answer_rrs:
         fail_if_matches_regexp: [test]

prober/grpc.go

@@ -15,6 +15,11 @@ package prober
 
 import (
 	"context"
+	"net"
+	"net/url"
+	"strings"
+	"time"
+
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/prometheus/blackbox_exporter/config"
@@ -27,10 +32,6 @@ import (
 	"google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
-	"net"
-	"net/url"
-	"strings"
-	"time"
 )
 
 type GRPCHealthCheck interface {
@@ -167,6 +168,18 @@ func ProbeGRPC(ctx context.Context, target string, module config.Module, registr
 	}
 
 	var opts []grpc.DialOption
+	if len(module.GRPC.SourceIPAddress) > 0 {
+		srcIP := net.ParseIP(module.GRPC.SourceIPAddress)
+		if srcIP == nil {
+			level.Error(logger).Log("msg", "Error parsing source ip address", "srcIP", module.GRPC.SourceIPAddress)
+			return false
+		}
+		level.Info(logger).Log("msg", "Using local address", "srcIP", srcIP)
+		opts = append(opts, grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) {
+			return (&net.Dialer{LocalAddr: &net.TCPAddr{IP: srcIP}}).DialContext(ctx, "tcp", addr)
+		}))
+	}
+
 	target = targetHost + ":" + targetPort
 	if !module.GRPC.TLS {
 		level.Debug(logger).Log("msg", "Dialing GRPC without TLS")

prober/grpc_test.go

@@ -414,3 +414,64 @@ func TestGRPCHealthCheckUnimplemented(t *testing.T) {
 
 	checkRegistryResults(expectedResults, mfs, t)
 }
+
+func TestGrpcSourceIPAddress(t *testing.T) {
+
+	ln, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Fatalf("Error listening on socket: %s", err)
+	}
+	defer ln.Close()
+
+	_, port, err := net.SplitHostPort(ln.Addr().String())
+	if err != nil {
+		t.Fatalf("Error retrieving port for socket: %s", err)
+	}
+	s := grpc.NewServer()
+	healthServer := health.NewServer()
+	healthServer.SetServingStatus("service", grpc_health_v1.HealthCheckResponse_SERVING)
+	grpc_health_v1.RegisterHealthServer(s, healthServer)
+
+	go func() {
+		if err := s.Serve(ln); err != nil {
+			t.Errorf("failed to serve: %v", err)
+			return
+		}
+	}()
+	defer s.GracefulStop()
+
+	ifaces, err := net.Interfaces()
+	if err != nil {
+		t.Fatalf("Error retrieving network interfaces: %s", err)
+	}
+	for _, iface := range ifaces {
+		addrs, err := iface.Addrs()
+		if err != nil {
+			t.Fatalf("Error retrieving addrs from iface %s: %s", iface.Name, err)
+		}
+		for _, addr := range addrs {
+			var ip net.IP
+			switch v := addr.(type) {
+			case *net.IPNet:
+				ip = v.IP
+			case *net.IPAddr:
+				ip = v.IP
+			}
+			// Skipping IPv6 addrs
+			if ip.To4() == nil {
+				continue
+			}
+			registry := prometheus.NewRegistry()
+			testCTX, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			result := ProbeGRPC(testCTX, "localhost:"+port,
+				config.Module{Timeout: time.Second, GRPC: config.GRPCProbe{
+					IPProtocolFallback: false,
+					SourceIPAddress:    ip.String(),
+				}}, registry, log.NewNopLogger())
+			if result != true {
+				t.Fatalf("Test %s had unexpected result", ip.String())
+			}
+		}
+	}
+}

prober/http.go

@@ -348,14 +348,30 @@ func ProbeHTTP(ctx context.Context, target string, module config.Module, registr
 			}
 		}
 	}
-	client, err := pconfig.NewClientFromConfig(httpClientConfig, "http_probe", pconfig.WithKeepAlivesDisabled())
+
+	httpClientOptions := []pconfig.HTTPClientOption{
+		pconfig.WithKeepAlivesDisabled(),
+	}
+
+	if len(module.HTTP.SourceIPAddress) > 0 {
+		srcIP := net.ParseIP(module.HTTP.SourceIPAddress)
+		if srcIP == nil {
+			level.Error(logger).Log("msg", "Error parsing source ip address", "srcIP", module.HTTP.SourceIPAddress)
+			return false
+		}
+		level.Info(logger).Log("msg", "Using local address", "srcIP", srcIP)
+		httpClientOptions = append(httpClientOptions,
+			pconfig.WithDialContextFunc((&net.Dialer{LocalAddr: &net.TCPAddr{IP: srcIP}}).DialContext))
+	}
+
+	client, err := pconfig.NewClientFromConfig(httpClientConfig, "http_probe", httpClientOptions...)
 	if err != nil {
 		level.Error(logger).Log("msg", "Error generating HTTP client", "err", err)
 		return false
 	}
 
 	httpClientConfig.TLSConfig.ServerName = ""
-	noServerName, err := pconfig.NewRoundTripperFromConfig(httpClientConfig, "http_probe", pconfig.WithKeepAlivesDisabled())
+	noServerName, err := pconfig.NewRoundTripperFromConfig(httpClientConfig, "http_probe", httpClientOptions...)
 	if err != nil {
 		level.Error(logger).Log("msg", "Error generating HTTP client without ServerName", "err", err)
 		return false

prober/http_test.go

@@ -1544,3 +1544,47 @@ func TestBody(t *testing.T) {
 		}
 	}
 }
+
+func TestHttpSourceIPAddress(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer ts.Close()
+
+	ifaces, err := net.Interfaces()
+	if err != nil {
+		t.Fatalf("Error retrieving network interfaces: %s", err)
+	}
+	for _, iface := range ifaces {
+		addrs, err := iface.Addrs()
+		if err != nil {
+			t.Fatalf("Error retrieving addrs from iface %s: %s", iface.Name, err)
+		}
+		for _, addr := range addrs {
+			var ip net.IP
+			switch v := addr.(type) {
+			case *net.IPNet:
+				ip = v.IP
+			case *net.IPAddr:
+				ip = v.IP
+			}
+			// Skipping IPv6 addrs
+			if ip.To4() == nil {
+				continue
+			}
+			registry := prometheus.NewRegistry()
+			recorder := httptest.NewRecorder()
+			testCTX, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			result := ProbeHTTP(testCTX, ts.URL,
+				config.Module{Timeout: time.Second, HTTP: config.HTTPProbe{
+					IPProtocolFallback: true,
+					SourceIPAddress:    ip.String(),
+				}}, registry, log.NewNopLogger())
+			body := recorder.Body.String()
+			if result != true {
+				t.Fatalf("Test %s had unexpected result: %s", ip.String(), body)
+			}
+		}
+	}
+}