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**
"//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",
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 {
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
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 {
ticker := time.NewTicker(time.Second * time.Duration(svc.ServiceDefinition().PollInterval))
eventsChan := make(chan HealthChangeEvent)
- go (func() {
+ go func() {
var lastEvent *HealthChangeEvent
doHealthCheck := func() {
doHealthCheck()
}
}
- })()
+ }()
return eventsChan
}
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 {
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)
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(),
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
}
}
- 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 {
"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"
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,
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(),
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
}
}
- return lcs.SocketHost, port
+ return lcs.SocketHost, port, nil
}
func (lcs *ldapHealthCheckService) Check() error {
}
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)
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)),
w.watcherCount++
och := make(chan AddressUpdate)
- go (func() {
+ go func() {
defer func() { w.watcherCount-- }()
for {
return
}
}
- })()
+ }()
return och
}
--- /dev/null
+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": [],
+ }),
+)
--- /dev/null
+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")
+}
--- /dev/null
+//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
+}
--- /dev/null
+//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
+}
projects / runtime.git / commitdiff