read metric from urls for listen and peers

This commit is contained in:
Arceliar 2021-05-23 20:58:34 -05:00
parent 70c5b06286
commit fd5cda6329
4 changed files with 26 additions and 31 deletions

View File

@ -106,15 +106,15 @@ func (c *Core) GetSessions() []Session {
// ListenTCP starts a new TCP listener. The input URI should match that of the // ListenTCP starts a new TCP listener. The input URI should match that of the
// "Listen" configuration item, e.g. // "Listen" configuration item, e.g.
// tcp://a.b.c.d:e // tcp://a.b.c.d:e
func (c *Core) ListenTCP(uri string) (*TcpListener, error) { func (c *Core) ListenTCP(uri string, metric uint8) (*TcpListener, error) {
return c.links.tcp.listen(uri, nil) return c.links.tcp.listen(uri, nil, metric)
} }
// ListenTLS starts a new TLS listener. The input URI should match that of the // ListenTLS starts a new TLS listener. The input URI should match that of the
// "Listen" configuration item, e.g. // "Listen" configuration item, e.g.
// tls://a.b.c.d:e // tls://a.b.c.d:e
func (c *Core) ListenTLS(uri string) (*TcpListener, error) { func (c *Core) ListenTLS(uri string, metric uint8) (*TcpListener, error) {
return c.links.tcp.listen(uri, c.links.tcp.tls.forListener) return c.links.tcp.listen(uri, c.links.tcp.tls.forListener, metric)
} }
// Address gets the IPv6 address of the Yggdrasil node. This is always a /128 // Address gets the IPv6 address of the Yggdrasil node. This is always a /128

View File

@ -8,6 +8,7 @@ import (
"io" "io"
"net" "net"
"net/url" "net/url"
"strconv"
"strings" "strings"
"sync" "sync"
@ -52,6 +53,7 @@ type link struct {
type linkOptions struct { type linkOptions struct {
pinnedEd25519Keys map[keyArray]struct{} pinnedEd25519Keys map[keyArray]struct{}
metric uint8
} }
func (l *links) init(c *Core) error { func (l *links) init(c *Core) error {
@ -90,6 +92,10 @@ func (l *links) call(u *url.URL, sintf string) error {
} }
} }
} }
if ms := u.Query()["metric"]; len(ms) == 1 {
m64, _ := strconv.ParseUint(ms[0], 10, 8)
tcpOpts.metric = uint8(m64)
}
switch u.Scheme { switch u.Scheme {
case "tcp": case "tcp":
l.tcp.call(u.Host, tcpOpts, sintf) l.tcp.call(u.Host, tcpOpts, sintf)
@ -110,23 +116,6 @@ func (l *links) call(u *url.URL, sintf string) error {
return nil return nil
} }
func (l *links) listen(uri string) error {
u, err := url.Parse(uri)
if err != nil {
return fmt.Errorf("listener %s is not correctly formatted (%s)", uri, err)
}
switch u.Scheme {
case "tcp":
_, err := l.tcp.listen(u.Host, nil)
return err
case "tls":
_, err := l.tcp.listen(u.Host, l.tcp.tls.forListener)
return err
default:
return errors.New("unknown listen scheme: " + u.Scheme)
}
}
func (l *links) create(conn net.Conn, name, linkType, local, remote string, incoming, force bool, options linkOptions) (*link, error) { func (l *links) create(conn net.Conn, name, linkType, local, remote string, incoming, force bool, options linkOptions) (*link, error) {
// Technically anything unique would work for names, but let's pick something human readable, just for debugging // Technically anything unique would work for names, but let's pick something human readable, just for debugging
intf := link{ intf := link{
@ -158,7 +147,7 @@ func (intf *link) handler() (chan struct{}, error) {
defer intf.conn.Close() defer intf.conn.Close()
meta := version_getBaseMetadata() meta := version_getBaseMetadata()
meta.key = intf.links.core.public meta.key = intf.links.core.public
// TODO set meta.metric meta.metric = intf.options.metric
metric := uint64(meta.metric) metric := uint64(meta.metric)
metaBytes := meta.encode() metaBytes := meta.encode()
// TODO timeouts on send/recv (goroutine for send/recv, channel select w/ timer) // TODO timeouts on send/recv (goroutine for send/recv, channel select w/ timer)

View File

@ -20,6 +20,7 @@ import (
"math/rand" "math/rand"
"net" "net"
"net/url" "net/url"
"strconv"
"strings" "strings"
"sync" "sync"
"time" "time"
@ -50,7 +51,7 @@ type tcp struct {
// multicast interfaces. // multicast interfaces.
type TcpListener struct { type TcpListener struct {
Listener net.Listener Listener net.Listener
upgrade *TcpUpgrade opts tcpOptions
stop chan struct{} stop chan struct{}
} }
@ -112,13 +113,18 @@ func (t *tcp) init(l *links) error {
if err != nil { if err != nil {
t.links.core.log.Errorln("Failed to parse listener: listener", listenaddr, "is not correctly formatted, ignoring") t.links.core.log.Errorln("Failed to parse listener: listener", listenaddr, "is not correctly formatted, ignoring")
} }
var metric uint8 // TODO parse from url
if ms := u.Query()["metric"]; len(ms) == 1 {
m64, _ := strconv.ParseUint(ms[0], 10, 8)
metric = uint8(m64)
}
switch u.Scheme { switch u.Scheme {
case "tcp": case "tcp":
if _, err := t.listen(u.Host, nil); err != nil { if _, err := t.listen(u.Host, nil, metric); err != nil {
return err return err
} }
case "tls": case "tls":
if _, err := t.listen(u.Host, t.tls.forListener); err != nil { if _, err := t.listen(u.Host, t.tls.forListener, metric); err != nil {
return err return err
} }
default: default:
@ -179,7 +185,7 @@ func (t *tcp) reconfigure() {
*/ */
} }
func (t *tcp) listen(listenaddr string, upgrade *TcpUpgrade) (*TcpListener, error) { func (t *tcp) listen(listenaddr string, upgrade *TcpUpgrade, metric uint8) (*TcpListener, error) {
var err error var err error
ctx := context.Background() ctx := context.Background()
@ -190,9 +196,10 @@ func (t *tcp) listen(listenaddr string, upgrade *TcpUpgrade) (*TcpListener, erro
if err == nil { if err == nil {
l := TcpListener{ l := TcpListener{
Listener: listener, Listener: listener,
upgrade: upgrade, opts: tcpOptions{upgrade: upgrade},
stop: make(chan struct{}), stop: make(chan struct{}),
} }
l.opts.metric = metric
t.waitgroup.Add(1) t.waitgroup.Add(1)
go t.listener(&l, listenaddr) go t.listener(&l, listenaddr)
return &l, nil return &l, nil
@ -243,9 +250,7 @@ func (t *tcp) listener(l *TcpListener, listenaddr string) {
continue continue
} }
t.waitgroup.Add(1) t.waitgroup.Add(1)
options := tcpOptions{ options := l.opts
upgrade: l.upgrade,
}
go t.handler(sock, true, options) go t.handler(sock, true, options)
} }
} }

View File

@ -274,7 +274,8 @@ func (m *Multicast) _announce() {
if nfo, ok := m.listeners[iface.Name]; !ok || nfo.listener.Listener == nil { if nfo, ok := m.listeners[iface.Name]; !ok || nfo.listener.Listener == nil {
// No listener was found - let's create one // No listener was found - let's create one
listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, iface.Name, m.listenPort) listenaddr := fmt.Sprintf("[%s%%%s]:%d", addrIP, iface.Name, m.listenPort)
if li, err := m.core.ListenTCP(listenaddr); err == nil { var metric uint8 // TODO parse this from config
if li, err := m.core.ListenTCP(listenaddr, metric); err == nil {
m.log.Debugln("Started multicasting on", iface.Name) m.log.Debugln("Started multicasting on", iface.Name)
// Store the listener so that we can stop it later if needed // Store the listener so that we can stop it later if needed
info = &listenerInfo{listener: li, time: time.Now()} info = &listenerInfo{listener: li, time: time.Now()}