From be3a00e320509d6ad7f09fdf1f8a8790c520966a Mon Sep 17 00:00:00 2001 From: Dan Fuhry Date: Tue, 13 Oct 2026 09:29:50 -0400 Subject: [PATCH] [sd] always publish on non-loopback addresses MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit When adding support for service discovery publishing as a sidecar, I realized that health probes are often performed against the loopback address, but the sd libraries don't support separate addresses for probing and publication. This commit adds that by detecting the local primary IPv4 and IPv6 addresses when the probe address is a loopback IP. Supporting changes: - Add `net2` library with `MyIP4()` and `MyIP6()` **🚫🤖 No AI** --- sd/BUILD.bazel | 1 + sd/healthcheck.go | 20 +++--- sd/healthcheck_http.go | 63 ++++++++++++++---- sd/healthcheck_ldap.go | 20 +++--- sd/monitor.go | 11 +++- sd/watcher.go | 4 +- utils/net2/BUILD.bazel | 23 +++++++ utils/net2/myip.go | 130 +++++++++++++++++++++++++++++++++++++ utils/net2/myip_generic.go | 20 ++++++ utils/net2/myip_linux.go | 64 ++++++++++++++++++ 10 files changed, 322 insertions(+), 34 deletions(-) create mode 100644 utils/net2/BUILD.bazel create mode 100644 utils/net2/myip.go create mode 100644 utils/net2/myip_generic.go create mode 100644 utils/net2/myip_linux.go diff --git a/sd/BUILD.bazel b/sd/BUILD.bazel index 3242353..fbcd623 100644 --- a/sd/BUILD.bazel +++ b/sd/BUILD.bazel @@ -22,6 +22,7 @@ go_multi_library( "//utils/context", "//utils/hostname", "//utils/log", + "//utils/net2", "@com_github_go_ldap_ldap_v3//:ldap", "@io_etcd_go_etcd_api_v3//v3rpc/rpctypes", "@io_etcd_go_etcd_client_pkg_v3//srv", diff --git a/sd/healthcheck.go b/sd/healthcheck.go index 9948c50..242f47f 100644 --- a/sd/healthcheck.go +++ b/sd/healthcheck.go @@ -23,8 +23,8 @@ type HealthCheckEngine interface { type HealthCheckService interface { Check() error ServiceDefinition() *ServiceDefinition - NewPublisher(domain string, regions []string) *SDPublisher - NewDefaultPublisher() *SDPublisher + NewPublisher(domain string, regions []string) (*SDPublisher, error) + NewDefaultPublisher() (*SDPublisher, error) } type PublishAddress struct { @@ -44,9 +44,11 @@ const ( var ( ErrTLSServerOptionConflict = errors.New( "mTLS related options (\"peer_mtls_id\") are mutually exclusive with \"server_name\" " + - "and \"ca_certificate\"") + "and \"ca_certificate\"", + ) ErrTLSClientOptionConflict = errors.New( - "cannot set client cert or private key when mtls identities are used") + "cannot set client cert or private key when mtls identities are used", + ) ) type HealthStatus uint @@ -83,8 +85,10 @@ type tlsService struct { CaCert string `json:"ca_certificate"` } -var engines []HealthCheckEngine -var enginesOnce sync.Once +var ( + engines []HealthCheckEngine + enginesOnce sync.Once +) func (p Layer4Protocol) DNSComponent() string { switch p { @@ -141,7 +145,7 @@ func HealthCheckLoop(ctx context.Context, svc HealthCheckService) <-chan HealthC ticker := time.NewTicker(time.Second * time.Duration(svc.ServiceDefinition().PollInterval)) eventsChan := make(chan HealthChangeEvent) - go (func() { + go func() { var lastEvent *HealthChangeEvent doHealthCheck := func() { @@ -181,7 +185,7 @@ func HealthCheckLoop(ctx context.Context, svc HealthCheckService) <-chan HealthC doHealthCheck() } } - })() + }() return eventsChan } diff --git a/sd/healthcheck_http.go b/sd/healthcheck_http.go index bc55e4f..44ea919 100644 --- a/sd/healthcheck_http.go +++ b/sd/healthcheck_http.go @@ -2,12 +2,15 @@ package sd import ( "encoding/json" + "errors" "fmt" + "net" "net/http" - "strings" "go.fuhry.dev/runtime/constants" "go.fuhry.dev/runtime/utils/hostname" + "go.fuhry.dev/runtime/utils/log" + "go.fuhry.dev/runtime/utils/net2" ) type httpHealthCheckEngine struct { @@ -26,15 +29,21 @@ type httpHealthCheckService struct { HttpHost string `json:"http_host"` Http2 bool `json:"http2"` Path string `json:"path"` + + logger log.Logger `json:"-"` } +var errMustSetIP = errors.New("must configure IP on SDPublisher") + func (hce *httpHealthCheckEngine) Name() string { return "http" } func (hce *httpHealthCheckEngine) Load(serviceDef *ServiceDefinition, conf []byte) (HealthCheckService, error) { + logPrefix := fmt.Sprintf("httpHealthCheck[%s]", serviceDef.ServiceName) svc := &httpHealthCheckService{ serviceDef: serviceDef, + logger: log.WithPrefix(logPrefix), } err := json.Unmarshal(conf, svc) @@ -61,27 +70,43 @@ func (hcs *httpHealthCheckService) ServiceDefinition() *ServiceDefinition { return hcs.serviceDef } -func (hcs *httpHealthCheckService) NewPublisher(domain string, regions []string) *SDPublisher { - _, port := hcs.PublishAddress() - shard := hcs.serviceDef.Shard - if shard == "" { - shard = strings.Split(hcs.SocketHost, ".")[0] - } +func (hcs *httpHealthCheckService) NewPublisher(domain string, regions []string) (*SDPublisher, error) { + host, port, err := hcs.PublishAddress() + var ip4Str, ip6Str string + if err != nil && !errors.Is(err, errMustSetIP) { + return nil, err + } else if errors.Is(err, errMustSetIP) { + ip4, err := net2.MyIP4() + if err != nil { + return nil, err + } + ip6, err := net2.MyIP6() + if err != nil { + return nil, err + } + host += fmt.Sprintf(".%s.%s", hcs.serviceDef.ServiceName, domain) + ip4Str = ip4.String() + ip6Str = ip6.String() + hcs.logger.Noticef("publish address updated to hostname %q, IPv4 %s, IPv6 %s", + host, ip4Str, ip6Str) + } publisher := &SDPublisher{ - AdvertiseHost: hcs.SocketHost, + AdvertiseHost: host, AdvertisePort: port, Domain: domain, Protocol: ProtocolTCP, Service: hcs.serviceDef.ServiceName, - ShardName: shard, + ShardName: hcs.serviceDef.Shard, Regions: regions, + IP4: ip4Str, + IP6: ip6Str, } - return publisher + return publisher, nil } -func (hcs *httpHealthCheckService) NewDefaultPublisher() *SDPublisher { +func (hcs *httpHealthCheckService) NewDefaultPublisher() (*SDPublisher, error) { domain := constants.SDDomain regions := []string{ hostname.RegionName(), @@ -90,7 +115,7 @@ func (hcs *httpHealthCheckService) NewDefaultPublisher() *SDPublisher { return hcs.NewPublisher(domain, regions) } -func (hcs *httpHealthCheckService) PublishAddress() (string, uint16) { +func (hcs *httpHealthCheckService) PublishAddress() (string, uint16, error) { port := hcs.SocketPort if port == 0 { port = 80 @@ -99,7 +124,19 @@ func (hcs *httpHealthCheckService) PublishAddress() (string, uint16) { } } - return hcs.SocketHost, port + host := hcs.SocketHost + if ip := net.ParseIP(host); ip != nil { + if !ip.IsLoopback() { + return "", 0, fmt.Errorf( + "socketHost must be either a loopback IP address or DNS name: %q", host, + ) + } + + hcs.logger.V(1).Debugf("SocketHost is a loopback address: %s", host) + return hostname.Hostname(), port, errMustSetIP + } + + return hcs.SocketHost, port, nil } func (hcs *httpHealthCheckService) Check() error { diff --git a/sd/healthcheck_ldap.go b/sd/healthcheck_ldap.go index cc00084..7e253b9 100644 --- a/sd/healthcheck_ldap.go +++ b/sd/healthcheck_ldap.go @@ -7,6 +7,7 @@ import ( "strings" ldap "github.com/go-ldap/ldap/v3" + "go.fuhry.dev/runtime/constants" "go.fuhry.dev/runtime/utils/hostname" "go.fuhry.dev/runtime/utils/log" @@ -56,15 +57,18 @@ func (lcs *ldapHealthCheckService) ServiceDefinition() *ServiceDefinition { return lcs.serviceDef } -func (lcs *ldapHealthCheckService) NewPublisher(domain string, regions []string) *SDPublisher { - _, port := lcs.PublishAddress() +func (lcs *ldapHealthCheckService) NewPublisher(domain string, regions []string) (*SDPublisher, error) { + host, port, err := lcs.PublishAddress() + if err != nil { + return nil, err + } shard := lcs.serviceDef.Shard if shard == "" { - shard = strings.Split(lcs.SocketHost, ".")[0] + shard = strings.Split(host, ".")[0] } publisher := &SDPublisher{ - AdvertiseHost: lcs.SocketHost, + AdvertiseHost: host, AdvertisePort: port, Domain: domain, Protocol: ProtocolTCP, @@ -73,10 +77,10 @@ func (lcs *ldapHealthCheckService) NewPublisher(domain string, regions []string) Regions: regions, } - return publisher + return publisher, nil } -func (lcs *ldapHealthCheckService) NewDefaultPublisher() *SDPublisher { +func (lcs *ldapHealthCheckService) NewDefaultPublisher() (*SDPublisher, error) { domain := constants.SDDomain region := []string{ hostname.RegionName(), @@ -85,7 +89,7 @@ func (lcs *ldapHealthCheckService) NewDefaultPublisher() *SDPublisher { return lcs.NewPublisher(domain, region) } -func (lcs *ldapHealthCheckService) PublishAddress() (string, uint16) { +func (lcs *ldapHealthCheckService) PublishAddress() (string, uint16, error) { port := lcs.SocketPort if port == 0 { port = 389 @@ -94,7 +98,7 @@ func (lcs *ldapHealthCheckService) PublishAddress() (string, uint16) { } } - return lcs.SocketHost, port + return lcs.SocketHost, port, nil } func (lcs *ldapHealthCheckService) Check() error { diff --git a/sd/monitor.go b/sd/monitor.go index 58a45c0..4aba33f 100644 --- a/sd/monitor.go +++ b/sd/monitor.go @@ -64,11 +64,11 @@ func (sdp *healthCheckingPublisher) MonitorAndPublish(ctx context.Context, wg *s } logger.V(2).Debugf("leaseID: %016x", statLease.ID) - defer (func() { + defer func() { if statLease != nil { leaseApi.Revoke(context.Background(), statLease.ID) } - })() + }() events := HealthCheckLoop(ctx, sdp.healthcheck) @@ -148,8 +148,13 @@ func NewHealthCheckPublisher(hcs HealthCheckService, domain string, regions []st return nil, err } + publisher, err := hcs.NewPublisher(domain, regions) + if err != nil { + return nil, err + } + return &healthCheckingPublisher{ - publisher: hcs.NewPublisher(domain, regions), + publisher: publisher, healthcheck: hcs, etcdClient: etcd, logger: log.WithPrefix(fmt.Sprintf("HealthCheckingPublisher(%s)", hcs.ServiceDefinition().ServiceName)), diff --git a/sd/watcher.go b/sd/watcher.go index e43e859..273b196 100644 --- a/sd/watcher.go +++ b/sd/watcher.go @@ -156,7 +156,7 @@ func (w *SDWatcher) WatchUpdates(ctx context.Context) chan AddressUpdate { w.watcherCount++ och := make(chan AddressUpdate) - go (func() { + go func() { defer func() { w.watcherCount-- }() for { @@ -168,7 +168,7 @@ func (w *SDWatcher) WatchUpdates(ctx context.Context) chan AddressUpdate { return } } - })() + }() return och } diff --git a/utils/net2/BUILD.bazel b/utils/net2/BUILD.bazel new file mode 100644 index 0000000..da5ca3c --- /dev/null +++ b/utils/net2/BUILD.bazel @@ -0,0 +1,23 @@ +load("//bazel:go.bzl", "go_multi_library") + +go_multi_library( + name = "net2", + srcs = [ + "myip.go", + "myip_generic.go", + "myip_linux.go", + ], + importpath = "go.fuhry.dev/runtime/utils/net2", + visibility = ["//visibility:public"], + deps = [ + "//utils/log", + ] + select({ + "@rules_go//go/platform:android": [ + "@com_github_vishvananda_netlink//:netlink", + ], + "@rules_go//go/platform:linux": [ + "@com_github_vishvananda_netlink//:netlink", + ], + "//conditions:default": [], + }), +) diff --git a/utils/net2/myip.go b/utils/net2/myip.go new file mode 100644 index 0000000..45554cb --- /dev/null +++ b/utils/net2/myip.go @@ -0,0 +1,130 @@ +package net2 + +import ( + "bytes" + "errors" + "fmt" + "net" + "sync" + + "go.fuhry.dev/runtime/utils/log" +) + +var sixToFourPrefix = []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff} + +func is6to4(ip net.IP) bool { + return (len(ip) == net.IPv6len && + bytes.Equal([]byte(ip[:len(sixToFourPrefix)]), sixToFourPrefix)) +} + +var ( + myip4Once sync.Once + myip4 net.IP + myip4Err error +) + +func MyIP4() (net.IP, error) { + myip4Once.Do(func() { + myip4, myip4Err = func() (net.IP, error) { + dri, err := defaultRouteInterface() + if err != nil { + return nil, fmt.Errorf("failed to detect default route interface: %v", err) + } + + addrs, err := dri.Addrs() + if err != nil { + return nil, fmt.Errorf("failed to list addresses for default route iface %s", dri.Name) + } + + for i, addr := range addrs { + ipnet, ok := addr.(*net.IPNet) + if !ok { + // not an IPNet + log.Default().V(2).Debugf("addr: (%T) %+v", addr, addr) + continue + } + + log.Default().V(2).Debugf( + "%s addr %d: (%T) %+v (len=%d)", + dri.Name, i, addr, addr, len(ipnet.IP), + ) + + if len(ipnet.IP) == net.IPv4len || is6to4(ipnet.IP) { + return ipnet.IP, nil + } + } + + return nil, fmt.Errorf("found no eligible addresses on default route interface %q", dri.Name) + }() + }) + return myip4, myip4Err +} + +var ( + myip6Once sync.Once + myip6 net.IP + myip6Err error +) + +func MyIP6() (net.IP, error) { + myip6Once.Do(func() { + myip6, myip6Err = func() (net.IP, error) { + dri, err := defaultRouteInterface() + if err != nil { + return nil, fmt.Errorf("failed to detect default route interface: %v", err) + } + + addrs, err := dri.Addrs() + if err != nil { + return nil, fmt.Errorf("failed to list addresses for default route iface %s", dri.Name) + } + + for _, addr := range addrs { + ipnet, ok := addr.(*net.IPNet) + if !ok { + // not an IPNet + log.Default().V(2).Debugf("addr: (%T) %+v", addr, addr) + continue + } + + if len(ipnet.IP) == net.IPv6len && !is6to4(ipnet.IP) { + return ipnet.IP, nil + } + } + + return nil, fmt.Errorf("found no eligible addresses on default route interface %q", dri.Name) + }() + }) + + return myip6, myip6Err +} + +func defaultRouteInterface() (*net.Interface, error) { + ifaces, err := net.Interfaces() + if err != nil { + return nil, err + } + + for i, iface := range ifaces { + log.Default().V(3).Debugf("iface %d name: %q flags: %d", i, iface.Name, iface.Flags) + flags := iface.Flags + + if (flags & net.FlagUp) != net.FlagUp { + // interface not up + log.Default().V(3).Debugf("iface %s is not up, skipping", iface.Name) + continue + } + + if (flags & net.FlagLoopback) == net.FlagLoopback { + // default route can't be the loopback + log.Default().V(3).Debugf("iface %s is a loopback, skipping", iface.Name) + continue + } + + if ifaceIsDefaultRoute(iface) { + return &iface, nil + } + } + + return nil, errors.New("cannot find any up, non-loopback network interfaces") +} diff --git a/utils/net2/myip_generic.go b/utils/net2/myip_generic.go new file mode 100644 index 0000000..e3fe770 --- /dev/null +++ b/utils/net2/myip_generic.go @@ -0,0 +1,20 @@ +//go:build !linux + +package net2 + +import ( + "net" + + "go.fuhry.dev/runtime/utils/log" +) + +func ifaceIsDefaultRoute(iface net.Interface) bool { + addrs, err := iface.Addrs() + if err != nil { + // can't get addresses + log.Default().V(2).Debugf("failed to get addrs for iface %s: %v", iface.Name, err) + return false + } + + return len(addrs) > 0 +} diff --git a/utils/net2/myip_linux.go b/utils/net2/myip_linux.go new file mode 100644 index 0000000..1f03ea8 --- /dev/null +++ b/utils/net2/myip_linux.go @@ -0,0 +1,64 @@ +//go:build linux + +package net2 + +import ( + "errors" + "net" + "sync" + + "github.com/vishvananda/netlink" + + "go.fuhry.dev/runtime/utils/log" +) + +func ifaceIsDefaultRoute(iface net.Interface) bool { + dri, err := getDefaultRouteIface() + return err == nil && dri == iface.Name +} + +var ( + defaultRouteOnce sync.Once + defaultRouteIface string + defaultRouteErr error +) + +func getDefaultRouteIface() (string, error) { + defaultRouteOnce.Do(func() { + defaultRouteIface, defaultRouteErr = func() (string, error) { + handle, err := netlink.NewHandle() + if err != nil { + return "", err + } + defer handle.Close() + links, err := handle.LinkList() + if err != nil { + return "", err + } + + for _, link := range links { + routes, err := netlink.RouteList(link, netlink.FAMILY_V4) + if err != nil { + log.Default().V(2).Warningf( + "failed to enumerate routes for link %s: %v", + link.Attrs().Name, + ) + } + + for _, route := range routes { + if route.Dst.IP.Equal(net.IPv4zero) { + if ones, _ := route.Dst.Mask.Size(); ones == 0 { + return link.Attrs().Name, nil + } + } + } + } + + return "", errors.New( + "unable to find any interface with an IPv4 route to 0.0.0.0/0", + ) + }() + }) + + return defaultRouteIface, defaultRouteErr +} -- 2.52.0