mirror of
https://github.com/openziti/ziti.git
synced 2026-09-11 13:29:03 +00:00
Paul Lorenz
3b59bfb95e
Canary state was left behind by every router delete, growing without bound under router churn, because nothing enumerated the stores that hold per-router state. The gossip type registry looked like that enumeration but is not: it drives the epoch sweep, and canaries deliberately sit out that sweep, since a canary carries the epoch that detects a change. Opting out of the sweep silently opted them out of delete cleanup as well. - splits the two lifecycles. The gossip type registry keeps driving the epoch sweep; a separate registry takes every store keyed by router id, which is a different membership for a different reason - registers link gossip, link metrics, canary gossip, the canary listener's own map and the link index, so one call site tears down all of them - tests that every per-router store is registered. Without that the registry is a convention, and a convention is what was already being broken - replaces the canary listener's sync.Map with a map under a mutex, so a read and the delete that follows it cannot interleave The cleanup is not ordered against a create of the same router id, and does not try to be. Router ids are assigned at enrollment; the one path that lets an id be chosen is deprecated.
2205 lines
75 KiB
Go
2205 lines
75 KiB
Go
/*
|
|
Copyright NetFoundry Inc.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
https://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package network
|
|
|
|
import (
|
|
"bytes"
|
|
"compress/gzip"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"os"
|
|
"runtime/debug"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/openziti/channel/v5"
|
|
"github.com/openziti/channel/v5/protobufs"
|
|
"github.com/openziti/foundation/v2/concurrenz"
|
|
"github.com/openziti/foundation/v2/debugz"
|
|
"github.com/openziti/foundation/v2/goroutines"
|
|
"github.com/openziti/foundation/v2/versions"
|
|
"github.com/openziti/identity"
|
|
"github.com/openziti/metrics"
|
|
"github.com/openziti/ziti/v2/common/ctrl_msg"
|
|
"github.com/openziti/ziti/v2/common/inspect"
|
|
"github.com/openziti/ziti/v2/common/logcontext"
|
|
"github.com/openziti/ziti/v2/common/pb/cmd_pb"
|
|
"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
|
|
"github.com/openziti/ziti/v2/common/pb/mgmt_pb"
|
|
"github.com/openziti/ziti/v2/common/servermetrics"
|
|
"github.com/openziti/ziti/v2/common/servermetrics/metrics_pb"
|
|
"github.com/openziti/ziti/v2/common/trace"
|
|
"github.com/openziti/ziti/v2/controller/command"
|
|
"github.com/openziti/ziti/v2/controller/config"
|
|
"github.com/openziti/ziti/v2/controller/db"
|
|
"github.com/openziti/ziti/v2/controller/event"
|
|
"github.com/openziti/ziti/v2/controller/gossip"
|
|
"github.com/openziti/ziti/v2/controller/idgen"
|
|
"github.com/openziti/ziti/v2/controller/model"
|
|
"github.com/openziti/ziti/v2/controller/raft"
|
|
"github.com/openziti/ziti/v2/controller/storage/boltz"
|
|
"github.com/openziti/ziti/v2/controller/storage/objectz"
|
|
"github.com/openziti/ziti/v2/controller/xt"
|
|
"github.com/sirupsen/logrus"
|
|
"github.com/teris-io/shortid"
|
|
"go.etcd.io/bbolt"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
const SmartRerouteAttempt = 99969996
|
|
|
|
// Config provides the values needed to create a Network instance
|
|
type Config interface {
|
|
GetId() *identity.TokenId
|
|
GetMetricsRegistry() metrics.Registry
|
|
GetOptions() *config.NetworkConfig
|
|
GetCommandDispatcher() command.Dispatcher
|
|
GetDb() boltz.Db
|
|
GetVersionProvider() versions.VersionProvider
|
|
GetEventDispatcher() event.Dispatcher
|
|
GetCloseNotify() <-chan struct{}
|
|
GetGossipPeering() GossipPeering
|
|
}
|
|
|
|
// GossipPeering describes how a controller reaches its peers, and is supplied when the network is created. A
|
|
// controller that is not part of a cluster supplies the zero value, which runs gossip with no peers and puts
|
|
// the network in single-controller mode.
|
|
//
|
|
// Both fields come from the raft controller, which exists before the network does, so there is nothing to
|
|
// hand over later. Keeping them together means "clustered" is one decision rather than two that could
|
|
// disagree.
|
|
type GossipPeering struct {
|
|
// Mesh is the transport peer gossip travels over. Nil means this controller has no peers.
|
|
Mesh gossip.Mesh
|
|
|
|
// IsLeader reports whether this controller is currently the raft leader, and being non-nil is what puts
|
|
// the network into HA mode. That decides how a router disconnect is handled: in HA a link is marked down,
|
|
// since the router may still be connected to another controller, while a single controller tombstones it.
|
|
// Nil means single-controller mode, where this controller is by definition the designated writer.
|
|
IsLeader func() bool
|
|
}
|
|
|
|
type InspectTarget func(string) (bool, *string, error)
|
|
|
|
// CtrlDialerValidator is a function that validates ctrl dialer states and returns per-router details.
|
|
type CtrlDialerValidator func() ([]*mgmt_pb.ControllerDialerDetails, error)
|
|
|
|
type Network struct {
|
|
*model.Managers
|
|
env model.Env
|
|
nodeId string
|
|
options *config.NetworkConfig
|
|
routeSenderController *routeSenderController
|
|
eventDispatcher event.Dispatcher
|
|
traceController trace.Controller
|
|
routerPresenceHandlers concurrenz.CopyOnWriteSlice[model.RouterPresenceHandler]
|
|
capabilities []string
|
|
closeNotify <-chan struct{}
|
|
watchdogCh chan struct{}
|
|
lock sync.Mutex
|
|
strategyRegistry xt.Registry
|
|
lastSnapshot time.Time
|
|
metricsRegistry metrics.Registry
|
|
VersionProvider versions.VersionProvider
|
|
|
|
GossipStore *gossip.Store
|
|
LinkGossipType *gossip.StateType[*ctrl_pb.RouterLinks_RouterLink]
|
|
LinkMetricsType *gossip.StateType[*ctrl_pb.LinkMetrics]
|
|
CanaryGossipType *gossip.StateType[*CanaryValue]
|
|
// gossipTypes drives the epoch sweep only. Router-deleted cleanup is a separate lifecycle with a
|
|
// different membership, since canaries carry the epoch that detects a change and so must survive one.
|
|
gossipTypes map[string]gossip.StateTypeInfo // non-generic lookup by store type
|
|
routerDeletedCleanups []routerDeletedCleanup
|
|
canaryListener *canaryGossipListener
|
|
isHA bool
|
|
leaderCheck func() bool
|
|
|
|
serviceEventMetrics servermetrics.UsageRegistry
|
|
serviceDialSuccessCounter servermetrics.IntervalCounter
|
|
serviceDialFailCounter servermetrics.IntervalCounter
|
|
serviceDialTimeoutCounter servermetrics.IntervalCounter
|
|
serviceDialOtherErrorCounter servermetrics.IntervalCounter
|
|
|
|
serviceTerminatorTimeoutCounter servermetrics.IntervalCounter
|
|
serviceTerminatorConnectionRefusedCounter servermetrics.IntervalCounter
|
|
serviceInvalidTerminatorCounter servermetrics.IntervalCounter
|
|
serviceMisconfiguredTerminatorCounter servermetrics.IntervalCounter
|
|
|
|
config Config
|
|
|
|
// restartSelfOnSnapshot, when true, restarts this controller in place after a snapshot restore
|
|
// (e.g. a node joining the cluster) instead of exiting and relying on an external process manager.
|
|
restartSelfOnSnapshot bool
|
|
|
|
Inspections *InspectionsManager
|
|
RouterMessaging *RouterMessaging
|
|
routerConnectSem concurrenz.Semaphore
|
|
gossipApplyPool goroutines.Pool
|
|
peerEventsPool goroutines.Pool
|
|
ioPool goroutines.Pool
|
|
inspectionTargets concurrenz.CopyOnWriteSlice[InspectTarget]
|
|
ctrlDialerValidator CtrlDialerValidator
|
|
}
|
|
|
|
func NewNetwork(config Config, env model.Env) (*Network, error) {
|
|
metricsConfig := servermetrics.DefaultUsageRegistryConfig(config.GetId().Token, config.GetCloseNotify())
|
|
if config.GetOptions().IntervalAgeThreshold != 0 {
|
|
metricsConfig.IntervalAgeThreshold = config.GetOptions().IntervalAgeThreshold
|
|
logrus.Infof("set interval age threshold to '%v'", config.GetOptions().IntervalAgeThreshold)
|
|
}
|
|
serviceEventMetrics := servermetrics.NewUsageRegistry(metricsConfig)
|
|
|
|
network := &Network{
|
|
env: env,
|
|
Managers: env.GetManagers(),
|
|
nodeId: config.GetId().Token,
|
|
options: config.GetOptions(),
|
|
routeSenderController: newRouteSenderController(),
|
|
eventDispatcher: config.GetEventDispatcher(),
|
|
traceController: trace.NewController(config.GetCloseNotify()),
|
|
closeNotify: config.GetCloseNotify(),
|
|
watchdogCh: make(chan struct{}, 1),
|
|
strategyRegistry: xt.GlobalRegistry(),
|
|
lastSnapshot: time.Now().Add(-time.Hour),
|
|
metricsRegistry: config.GetMetricsRegistry(),
|
|
VersionProvider: config.GetVersionProvider(),
|
|
|
|
serviceEventMetrics: serviceEventMetrics,
|
|
serviceDialSuccessCounter: serviceEventMetrics.IntervalCounter("service.dial.success", time.Minute),
|
|
serviceDialFailCounter: serviceEventMetrics.IntervalCounter("service.dial.fail", time.Minute),
|
|
serviceDialTimeoutCounter: serviceEventMetrics.IntervalCounter("service.dial.timeout", time.Minute),
|
|
serviceDialOtherErrorCounter: serviceEventMetrics.IntervalCounter("service.dial.error_other", time.Minute),
|
|
|
|
serviceTerminatorTimeoutCounter: serviceEventMetrics.IntervalCounter("service.dial.terminator.timeout", time.Minute),
|
|
serviceTerminatorConnectionRefusedCounter: serviceEventMetrics.IntervalCounter("service.dial.terminator.connection_refused", time.Minute),
|
|
serviceInvalidTerminatorCounter: serviceEventMetrics.IntervalCounter("service.dial.terminator.invalid", time.Minute),
|
|
serviceMisconfiguredTerminatorCounter: serviceEventMetrics.IntervalCounter("service.dial.terminator.misconfigured", time.Minute),
|
|
|
|
config: config,
|
|
}
|
|
|
|
env.GetManagers().Command.Decoders.RegisterF(int32(cmd_pb.CommandType_SyncSnapshot), network.decodeSyncSnapshotCommand)
|
|
|
|
network.routerConnectSem = concurrenz.NewSemaphore(int(config.GetOptions().RouterConnectConcurrency))
|
|
|
|
gossipApplyPool, err := network.createGossipApplyPool(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
network.gossipApplyPool = gossipApplyPool
|
|
|
|
peerEventsPool, err := network.createPeerEventsPool(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
network.peerEventsPool = peerEventsPool
|
|
|
|
ioPool, err := network.createIoPool(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
network.ioPool = ioPool
|
|
|
|
routerCommPool, err := network.createRouterCommPool(config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
network.Inspections = NewInspectionsManager(network)
|
|
network.RouterMessaging = NewRouterMessaging(env, routerCommPool)
|
|
|
|
env.GetManagers().Router.Store.AddEntityIdListener(network.HandleRouterDelete, boltz.EntityDeletedAsync)
|
|
|
|
linkManager := env.GetManagers().Link
|
|
network.onRouterDeleted("linkIndex", linkManager.RouterDeleted)
|
|
|
|
network.AddCapability("ziti.fabric")
|
|
network.showOptions()
|
|
network.relayControllerMetrics()
|
|
servermetrics.RegisterHostStats(network.metricsRegistry, servermetrics.HostStatsConfig{
|
|
Enabled: config.GetOptions().HostMetrics.Enabled,
|
|
})
|
|
|
|
// ctrl.is_leader is 1 on the raft leader, 0 otherwise (always 1 in non-HA).
|
|
// Lets the metrics timeseries correlate per-controller load with leadership,
|
|
// which matters under chaos/partitions that shuffle leadership. The closure
|
|
// reads IsLeader() dynamically, so it tracks leadership set later by HA init.
|
|
network.metricsRegistry.FuncGauge("ctrl.is_leader", func() int64 {
|
|
if network.IsLeader() {
|
|
return 1
|
|
}
|
|
return 0
|
|
})
|
|
network.AddRouterPresenceHandler(network.RouterMessaging)
|
|
go network.RouterMessaging.run()
|
|
|
|
network.initGossip(config.GetGossipPeering())
|
|
|
|
return network, nil
|
|
}
|
|
|
|
// initGossip builds the gossip store and registers every state type on it, once, for the life of the network.
|
|
// A controller that has no peers gets a mesh with none rather than no mesh, so nothing downstream has to
|
|
// special-case single-controller mode.
|
|
func (network *Network) initGossip(peering GossipPeering) {
|
|
network.isHA = peering.IsLeader != nil
|
|
network.leaderCheck = peering.IsLeader
|
|
|
|
mesh := peering.Mesh
|
|
if mesh == nil {
|
|
mesh = gossip.NewNoopMesh()
|
|
}
|
|
|
|
network.GossipStore = gossip.NewStore(network.nodeId, mesh)
|
|
network.GossipStore.SetEventsPool(network.peerEventsPool)
|
|
network.GossipStore.SetIoPool(network.ioPool)
|
|
network.GossipStore.SetMetricsRegistry(network.metricsRegistry)
|
|
|
|
network.initLinkGossip()
|
|
network.initLinkMetricsGossip()
|
|
network.initCanaryGossip()
|
|
}
|
|
|
|
// IsLeader returns true if this controller is the designated gossip writer for
|
|
// old-router link reports. In non-HA mode, always returns true. In HA mode,
|
|
// returns true if this controller is the raft leader.
|
|
func (network *Network) IsLeader() bool {
|
|
if !network.isHA {
|
|
return true
|
|
}
|
|
if network.leaderCheck != nil {
|
|
return network.leaderCheck()
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (self *Network) HandleRouterDelete(id string) {
|
|
self.routerDeleted(id)
|
|
self.RouterMessaging.RouterDeleted(id)
|
|
self.runRouterDeletedCleanups(id)
|
|
}
|
|
|
|
func (self *Network) decodeSyncSnapshotCommand(_ int32, data []byte) (command.Command, error) {
|
|
msg := &cmd_pb.SyncSnapshotCommand{}
|
|
if err := proto.Unmarshal(data, msg); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
cmd := &command.SyncSnapshotCommand{
|
|
TimelineId: msg.SnapshotId,
|
|
Snapshot: msg.Snapshot,
|
|
ClusterId: msg.ClusterId,
|
|
SnapshotSink: self.RestoreSnapshot,
|
|
}
|
|
|
|
return cmd, nil
|
|
}
|
|
|
|
func routerCommunicationsWorker(_ uint32, f func()) {
|
|
f()
|
|
}
|
|
|
|
func (network *Network) createGossipApplyPool(config Config) (goroutines.Pool, error) {
|
|
poolConfig := goroutines.PoolConfig{
|
|
QueueSize: config.GetOptions().GossipApplyPool.QueueSize,
|
|
MinWorkers: 0,
|
|
MaxWorkers: config.GetOptions().GossipApplyPool.MaxWorkers,
|
|
IdleTime: 30 * time.Second,
|
|
CloseNotify: config.GetCloseNotify(),
|
|
PanicHandler: func(err interface{}) {
|
|
pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during gossip apply processing")
|
|
},
|
|
}
|
|
|
|
servermetrics.ConfigureGoroutinesPoolMetrics(&poolConfig, config.GetMetricsRegistry(), "pool.gossip.apply")
|
|
|
|
pool, err := goroutines.NewPool(poolConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating gossip apply pool: %w", err)
|
|
}
|
|
return pool, nil
|
|
}
|
|
|
|
// createIoPool creates the controller's shared pool for outbound, blocking
|
|
// network I/O, organized by kind of work rather than by function. It is kept
|
|
// separate from the CPU/apply pools by design: I/O that stalls on a slow peer or
|
|
// router must not pin a worker that an apply path depends on. Each submitter
|
|
// chooses its own delivery policy (QueueOrError to drop, e.g. gossip broadcast
|
|
// with anti-entropy as the backstop; or a blocking submit for must-deliver work).
|
|
// Gossip broadcast is the first user; other blocking-I/O senders (e.g.
|
|
// RouterMessaging) can migrate here. Sized with more workers than the CPU pools
|
|
// to absorb concurrent slow sends.
|
|
func (network *Network) createIoPool(config Config) (goroutines.Pool, error) {
|
|
poolConfig := goroutines.PoolConfig{
|
|
QueueSize: config.GetOptions().IoPool.QueueSize,
|
|
MinWorkers: 0,
|
|
MaxWorkers: config.GetOptions().IoPool.MaxWorkers,
|
|
IdleTime: 30 * time.Second,
|
|
CloseNotify: config.GetCloseNotify(),
|
|
PanicHandler: func(err interface{}) {
|
|
pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during gossip io")
|
|
},
|
|
}
|
|
|
|
servermetrics.ConfigureGoroutinesPoolMetrics(&poolConfig, config.GetMetricsRegistry(), "pool.io")
|
|
|
|
pool, err := goroutines.NewPool(poolConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating io pool: %w", err)
|
|
}
|
|
return pool, nil
|
|
}
|
|
|
|
func (network *Network) createPeerEventsPool(config Config) (goroutines.Pool, error) {
|
|
poolConfig := goroutines.PoolConfig{
|
|
QueueSize: config.GetOptions().PeerEventsPool.QueueSize,
|
|
MinWorkers: 0,
|
|
MaxWorkers: config.GetOptions().PeerEventsPool.MaxWorkers,
|
|
IdleTime: 30 * time.Second,
|
|
CloseNotify: config.GetCloseNotify(),
|
|
PanicHandler: func(err interface{}) {
|
|
pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during peer event processing")
|
|
},
|
|
}
|
|
|
|
servermetrics.ConfigureGoroutinesPoolMetrics(&poolConfig, config.GetMetricsRegistry(), "pool.peer.events")
|
|
|
|
pool, err := goroutines.NewPool(poolConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating peer events pool: %w", err)
|
|
}
|
|
return pool, nil
|
|
}
|
|
|
|
func (network *Network) createRouterCommPool(config Config) (goroutines.Pool, error) {
|
|
poolConfig := goroutines.PoolConfig{
|
|
QueueSize: config.GetOptions().RouterComm.QueueSize,
|
|
MinWorkers: 0,
|
|
MaxWorkers: config.GetOptions().RouterComm.MaxWorkers,
|
|
IdleTime: 30 * time.Second,
|
|
CloseNotify: config.GetCloseNotify(),
|
|
PanicHandler: func(err interface{}) {
|
|
pfxlog.Logger().WithField(logrus.ErrorKey, err).WithField("backtrace", string(debug.Stack())).Error("panic during message send to router")
|
|
},
|
|
WorkerFunction: routerCommunicationsWorker,
|
|
}
|
|
|
|
servermetrics.ConfigureGoroutinesPoolMetrics(&poolConfig, config.GetMetricsRegistry(), "pool.router.messaging")
|
|
|
|
pool, err := goroutines.NewPool(poolConfig)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error creating router messaging pool (%w)", err)
|
|
}
|
|
return pool, nil
|
|
}
|
|
|
|
func (network *Network) relayControllerMetrics() {
|
|
go func() {
|
|
timer := time.NewTicker(network.options.MetricsReportInterval)
|
|
defer timer.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-timer.C:
|
|
if msg := servermetrics.Poll(network.metricsRegistry); msg != nil {
|
|
network.eventDispatcher.AcceptMetricsMsg(msg)
|
|
}
|
|
case <-network.closeNotify:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (network *Network) InitServiceCounterDispatch(handler servermetrics.Handler) {
|
|
network.serviceEventMetrics.StartReporting(handler, network.GetOptions().MetricsReportInterval, 10)
|
|
}
|
|
|
|
func (network *Network) GetAppId() string {
|
|
return network.nodeId
|
|
}
|
|
|
|
func (network *Network) GetOptions() *config.NetworkConfig {
|
|
return network.options
|
|
}
|
|
|
|
func (network *Network) GetDb() boltz.Db {
|
|
return network.config.GetDb()
|
|
}
|
|
|
|
func (network *Network) GetStores() *db.Stores {
|
|
return network.env.GetStores()
|
|
}
|
|
|
|
func (network *Network) GetConnectedRouter(routerId string) *model.Router {
|
|
return network.Router.GetConnected(routerId)
|
|
}
|
|
|
|
// NewCtrlChanRouter returns the Router instance representing a new control-channel connection, with the
|
|
// channel recorded on it. Each connection gets its own instance, which is what lets the connect and
|
|
// disconnect paths tell two racing connections for one router apart.
|
|
func (network *Network) NewCtrlChanRouter(ch channel.Channel) (*model.Router, error) {
|
|
return network.Router.NewCtrlChanRouter(ch)
|
|
}
|
|
|
|
func (network *Network) GetRouter(routerId string) (*model.Router, error) {
|
|
return network.Router.Read(routerId)
|
|
}
|
|
|
|
func (network *Network) AllConnectedRouters() []*model.Router {
|
|
return network.Router.AllConnected()
|
|
}
|
|
|
|
func (network *Network) GetLink(linkId string) (*model.Link, bool) {
|
|
return network.Link.Get(linkId)
|
|
}
|
|
|
|
func (network *Network) GetAllLinks() []*model.Link {
|
|
return network.Link.All()
|
|
}
|
|
|
|
func (network *Network) GetAllLinksForRouter(routerId string) []*model.Link {
|
|
r := network.GetConnectedRouter(routerId)
|
|
if r == nil {
|
|
return nil
|
|
}
|
|
return r.GetLinks()
|
|
}
|
|
|
|
func (network *Network) GetCircuit(circuitId string) (*model.Circuit, bool) {
|
|
return network.Circuit.Get(circuitId)
|
|
}
|
|
|
|
func (network *Network) GetAllCircuits() []*model.Circuit {
|
|
return network.Circuit.All()
|
|
}
|
|
|
|
func (network *Network) GetCircuitStore() *objectz.ObjectStore[*model.Circuit] {
|
|
return network.Circuit.GetStore()
|
|
}
|
|
|
|
func (network *Network) GetLinkStore() *objectz.ObjectStore[*model.Link] {
|
|
return network.Link.GetStore()
|
|
}
|
|
|
|
func (network *Network) RouteResult(rs *RouteStatus) bool {
|
|
return network.routeSenderController.forwardRouteResult(rs)
|
|
}
|
|
|
|
func (network *Network) newRouteSender(circuitId string) *routeSender {
|
|
rs := newRouteSender(circuitId, network.options.RouteTimeout, network, network.Terminator)
|
|
network.routeSenderController.addRouteSender(rs)
|
|
return rs
|
|
}
|
|
|
|
func (network *Network) removeRouteSender(rs *routeSender) {
|
|
network.routeSenderController.removeRouteSender(rs)
|
|
}
|
|
|
|
func (network *Network) GetEventDispatcher() event.Dispatcher {
|
|
return network.eventDispatcher
|
|
}
|
|
|
|
func (network *Network) GetTraceController() trace.Controller {
|
|
return network.traceController
|
|
}
|
|
|
|
func (network *Network) GetMetricsRegistry() metrics.Registry {
|
|
return network.metricsRegistry
|
|
}
|
|
|
|
func (network *Network) GetServiceEventsMetricsRegistry() servermetrics.UsageRegistry {
|
|
return network.serviceEventMetrics
|
|
}
|
|
|
|
func (network *Network) GetCloseNotify() <-chan struct{} {
|
|
return network.closeNotify
|
|
}
|
|
|
|
// GetGossipApplyPool returns the bounded pool for router-originated gossip
|
|
// and canary message processing.
|
|
func (network *Network) GetGossipApplyPool() goroutines.Pool {
|
|
return network.gossipApplyPool
|
|
}
|
|
|
|
// GetIoPool returns the shared, bounded pool for outbound blocking network I/O.
|
|
// Submitters choose their own delivery policy (drop vs block); gossip sends use
|
|
// it best-effort with anti-entropy / re-trigger as the backstop.
|
|
func (network *Network) GetIoPool() goroutines.Pool {
|
|
return network.ioPool
|
|
}
|
|
|
|
// GetPeerEventsPool returns the bounded pool for peer controller event processing.
|
|
func (network *Network) GetPeerEventsPool() goroutines.Pool {
|
|
return network.peerEventsPool
|
|
}
|
|
|
|
func (network *Network) ConnectedRouter(id string) bool {
|
|
return network.Router.IsConnected(id)
|
|
}
|
|
|
|
var (
|
|
// ErrConnectRejected indicates a router connect was rejected because another connection for the same
|
|
// router is already current. QueueRouterConnect returns it and the bind handler propagates it so
|
|
// NewChannel closes the rejected connection's underlay without starting rx or registering it; the
|
|
// router then redials.
|
|
ErrConnectRejected = errors.New("router connect rejected: another connection is already current")
|
|
|
|
// ErrConnectChannelClosed indicates a router connect was refused because its control channel was
|
|
// already closed by the time the connect decision was made, so the connection must not be registered.
|
|
ErrConnectChannelClosed = errors.New("router connect rejected: control channel already closed")
|
|
)
|
|
|
|
// IsConnectRejected reports whether err is (or wraps) a connect refusal that the router recovers from by
|
|
// redialing, so the accept path can log it at info rather than treating it as a bind failure.
|
|
func IsConnectRejected(err error) bool {
|
|
return errors.Is(err, ErrConnectRejected) || errors.Is(err, ErrConnectChannelClosed)
|
|
}
|
|
|
|
// QueueRouterConnect makes the synchronous connect decision for r, serialized per router. If the slot is
|
|
// already held by a different connection it rejects this one (returning ErrConnectRejected) and displaces
|
|
// the occupant so its teardown runs; the router redials into the freed slot. A connection whose channel is
|
|
// already closed is refused outright (ErrConnectChannelClosed) rather than registered. Otherwise it
|
|
// registers r (MarkConnected), fires synchronous presence handlers, and hands the remaining, non-gating
|
|
// setup to a goroutine bounded by the connect semaphore. The decision and registration are synchronous so
|
|
// the bind handler can fail a refused connect (no rx started, not registered) and so there is at most one
|
|
// connection per router current at a time. Capacity is never a reason to refuse: a refusal fails the bind,
|
|
// which strands the router's channel group, so the setup waits for a slot instead.
|
|
func (network *Network) QueueRouterConnect(r *model.Router) error {
|
|
unlock := network.Router.LockConnectFor(r.Id)
|
|
defer unlock() // leak-safety net; idempotent, so the explicit unlock below is the one that matters
|
|
|
|
if cur := network.Router.GetConnected(r.Id); cur != nil && cur != r {
|
|
// Displace the occupant outside the lock: the teardown acquires the stripe itself (we have
|
|
// released it), so there is no reentrant self-deadlock.
|
|
unlock()
|
|
network.displaceConnection(cur)
|
|
return ErrConnectRejected
|
|
}
|
|
|
|
// Its close handler has already run and never fires again, so nothing would remove it from the
|
|
// connected map and every redial would bounce off a slot that can never be freed.
|
|
if r.Control == nil || r.Control.IsClosed() {
|
|
return ErrConnectChannelClosed
|
|
}
|
|
|
|
// Mark connected before the deferred setup builds this router's links: repairDest re-reads the
|
|
// connected map for links that land in the table after that build, and must find the router there.
|
|
network.Router.MarkConnected(r)
|
|
|
|
for _, h := range network.routerPresenceHandlers.Value() {
|
|
if syncCapableHandler, ok := h.(model.SyncRouterPresenceHandler); ok && syncCapableHandler.InvokeRouterConnectedSynchronously() {
|
|
h.RouterConnected(r)
|
|
}
|
|
}
|
|
unlock()
|
|
|
|
// Off the accept goroutine, so binding stays quick: the listener holds this group's create
|
|
// reservation until the bind returns, and the group's other underlays wait on it. The semaphore
|
|
// bounds how many of these run at once so a mass reconnect cannot starve the gossip and I/O pools,
|
|
// and it blocks rather than refusing, since a refusal here would fail the bind.
|
|
go func() {
|
|
network.routerConnectSem.Acquire()
|
|
defer network.routerConnectSem.Release()
|
|
network.ConnectRouter(r)
|
|
}()
|
|
|
|
return nil
|
|
}
|
|
|
|
// ConnectRouter runs the deferred, non-gating connect setup for r on the router connect pool. It
|
|
// re-acquires the per-router lock and re-checks currency, so its link building and gossip reconcile are
|
|
// serialized against a concurrent disconnect of the same router and are skipped entirely if r was
|
|
// superseded or disconnected while it was queued. Different routers run concurrently on separate stripes.
|
|
func (network *Network) ConnectRouter(r *model.Router) {
|
|
unlock := network.Router.LockConnectFor(r.Id)
|
|
if !network.IsCurrentConnection(r) {
|
|
unlock()
|
|
return
|
|
}
|
|
|
|
network.Link.BuildRouterLinks(r)
|
|
|
|
// When gossip is enabled and a router reconnects, mark its links as
|
|
// usable again — they were set down on disconnect, not removed.
|
|
// Also reconcile gossip entries to update stale Src/Dst pointers and
|
|
// create any links that were missed during initial gossip application.
|
|
if network.GossipStore != nil {
|
|
for _, l := range r.GetLinks() {
|
|
if l.GetSrc().Id == r.Id && l.IsDown() {
|
|
l.SetDown(false)
|
|
}
|
|
}
|
|
network.ReconcileGossipLinksForRouter(r)
|
|
}
|
|
|
|
for _, h := range network.routerPresenceHandlers.Value() {
|
|
if syncHandler, ok := h.(model.SyncRouterPresenceHandler); ok && syncHandler.InvokeRouterConnectedSynchronously() {
|
|
continue // already called synchronously in QueueRouterConnect
|
|
}
|
|
h.RouterConnected(r)
|
|
}
|
|
unlock()
|
|
|
|
network.ValidateTerminators(r)
|
|
}
|
|
|
|
// RegisterGossipType registers a gossip state type for non-generic lookup by
|
|
// store type name. Used by the staleness detection logic.
|
|
func (network *Network) RegisterGossipType(name string, t gossip.StateTypeInfo) {
|
|
if network.gossipTypes == nil {
|
|
network.gossipTypes = map[string]gossip.StateTypeInfo{}
|
|
}
|
|
network.gossipTypes[name] = t
|
|
}
|
|
|
|
// GetGossipType returns the gossip state type for the given store type name.
|
|
func (network *Network) GetGossipType(name string) gossip.StateTypeInfo {
|
|
return network.gossipTypes[name]
|
|
}
|
|
|
|
// GossipStoreTypes returns the names of all registered router-replicated gossip
|
|
// store types. Used to drive per-store-type operations (such as connect-time
|
|
// digests) generically, so a new store type is covered without per-site edits.
|
|
func (network *Network) GossipStoreTypes() []string {
|
|
names := make([]string, 0, len(network.gossipTypes))
|
|
for name := range network.gossipTypes {
|
|
names = append(names, name)
|
|
}
|
|
return names
|
|
}
|
|
|
|
// dropGossipOwner tombstones every registered store type's entries for the owner,
|
|
// so a removed router's gossip state is reclaimed across all stores.
|
|
// routerDeletedCleanup is one subsystem's per-router state teardown, named so a test can check every
|
|
// per-router store is covered.
|
|
type routerDeletedCleanup struct {
|
|
name string
|
|
cleanup func(routerId string)
|
|
}
|
|
|
|
// onRouterDeleted registers cleanup for state keyed by router id. Every store holding per-router state
|
|
// must register, or it leaks on delete: nothing else enumerates them, and the leak is invisible until a
|
|
// memory profile is read.
|
|
//
|
|
// This assumes a deleted router id does not come back. Router ids are generated at create, apart from the
|
|
// deprecated path that creates a fabric router outside enrollment with a caller-supplied id, and cleanup
|
|
// is not ordered against a create of the same id. See the deprecation note on that path.
|
|
func (network *Network) onRouterDeleted(name string, cleanup func(routerId string)) {
|
|
network.routerDeletedCleanups = append(network.routerDeletedCleanups, routerDeletedCleanup{
|
|
name: name,
|
|
cleanup: cleanup,
|
|
})
|
|
}
|
|
|
|
// routerDeletedCleanupNames returns the registered cleanups, for tests that check every per-router store
|
|
// is covered.
|
|
func (network *Network) routerDeletedCleanupNames() []string {
|
|
names := make([]string, 0, len(network.routerDeletedCleanups))
|
|
for _, c := range network.routerDeletedCleanups {
|
|
names = append(names, c.name)
|
|
}
|
|
return names
|
|
}
|
|
|
|
// runRouterDeletedCleanups runs every registered cleanup for a deleted router.
|
|
func (network *Network) runRouterDeletedCleanups(id string) {
|
|
for _, c := range network.routerDeletedCleanups {
|
|
c.cleanup(id)
|
|
}
|
|
}
|
|
|
|
// deleteGossipOwnerBefore removes every registered store type's entries for the
|
|
// owner from a previous lifetime (epoch older than the given one).
|
|
func (network *Network) deleteGossipOwnerBefore(owner string, epoch []byte) {
|
|
for _, t := range network.gossipTypes {
|
|
t.DeleteByOwnerBefore(owner, epoch)
|
|
}
|
|
}
|
|
|
|
// HandleRouterEpoch processes a router epoch from the hello or canary. If the
|
|
// epoch is newer than the stored epoch, old-epoch entries for this router are
|
|
// deleted across all gossip store types.
|
|
func (network *Network) HandleRouterEpoch(routerId string, epoch []byte) {
|
|
if len(epoch) == 0 {
|
|
return
|
|
}
|
|
|
|
// Update the canary listener's epoch tracking and let it handle cleanup.
|
|
// This reuses the same epoch change detection as the canary path.
|
|
if network.canaryListener != nil {
|
|
network.canaryListener.checkEpoch(routerId, epoch)
|
|
}
|
|
}
|
|
|
|
func (network *Network) ValidateTerminators(r *model.Router) {
|
|
logger := pfxlog.Logger().WithField("routerId", r.Id)
|
|
result, err := network.Terminator.Query(fmt.Sprintf(`router.id = "%v" limit none`, r.Id))
|
|
if err != nil {
|
|
logger.WithError(err).Error("failed to get terminators for router")
|
|
return
|
|
}
|
|
|
|
logger.Debugf("%v terminators to validate", len(result.Entities))
|
|
if len(result.Entities) == 0 {
|
|
return
|
|
}
|
|
|
|
network.RouterMessaging.ValidateRouterTerminators(result.Entities)
|
|
}
|
|
|
|
type LinkValidationCallback func(detail *mgmt_pb.RouterLinkDetails)
|
|
|
|
func (n *Network) ValidateLinks(filter string, cb LinkValidationCallback) (int64, func(), error) {
|
|
result, err := n.Router.BaseList(filter)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
sem := concurrenz.NewSemaphore(10)
|
|
|
|
evalF := func() {
|
|
for _, router := range result.Entities {
|
|
connectedRouter := n.GetConnectedRouter(router.Id)
|
|
if connectedRouter != nil {
|
|
sem.Acquire()
|
|
go func() {
|
|
defer sem.Release()
|
|
n.ValidateRouterLinks(connectedRouter, cb)
|
|
}()
|
|
} else {
|
|
n.reportRouterLinksError(router, errors.New("router not connected"), cb)
|
|
}
|
|
}
|
|
}
|
|
|
|
return int64(len(result.Entities)), evalF, nil
|
|
}
|
|
|
|
type SdkTerminatorValidationCallback func(detail *mgmt_pb.RouterSdkTerminatorsDetails)
|
|
|
|
func (n *Network) ValidateRouterSdkTerminators(filter string, cb SdkTerminatorValidationCallback) (int64, func(), error) {
|
|
result, err := n.Router.BaseList(filter)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
sem := concurrenz.NewSemaphore(10)
|
|
|
|
evalF := func() {
|
|
for _, router := range result.Entities {
|
|
connectedRouter := n.GetConnectedRouter(router.Id)
|
|
if connectedRouter != nil {
|
|
sem.Acquire()
|
|
go func() {
|
|
defer sem.Release()
|
|
n.Router.ValidateRouterSdkTerminators(connectedRouter, cb)
|
|
}()
|
|
} else {
|
|
n.Router.ReportRouterSdkTerminatorsError(router, errors.New("router not connected"), cb)
|
|
}
|
|
}
|
|
}
|
|
|
|
return int64(len(result.Entities)), evalF, nil
|
|
}
|
|
|
|
type ErtTerminatorValidationCallback func(detail *mgmt_pb.RouterErtTerminatorsDetails)
|
|
|
|
func (n *Network) ValidateRouterErtTerminators(filter string, cb ErtTerminatorValidationCallback) (int64, func(), error) {
|
|
result, err := n.Router.BaseList(filter)
|
|
if err != nil {
|
|
return 0, nil, err
|
|
}
|
|
|
|
sem := concurrenz.NewSemaphore(10)
|
|
|
|
evalF := func() {
|
|
for _, router := range result.Entities {
|
|
connectedRouter := n.GetConnectedRouter(router.Id)
|
|
if connectedRouter != nil {
|
|
sem.Acquire()
|
|
go func() {
|
|
defer sem.Release()
|
|
n.Router.ValidateRouterErtTerminators(connectedRouter, cb)
|
|
}()
|
|
} else {
|
|
n.Router.ReportRouterErtTerminatorsError(router, errors.New("router not connected"), cb)
|
|
}
|
|
}
|
|
}
|
|
|
|
return int64(len(result.Entities)), evalF, nil
|
|
}
|
|
|
|
// IsCurrentConnection reports whether r is still the router's current, connected connection, by pointer
|
|
// identity against the connected map (mirrors the check in NotifyExistingLink). A stale or superseded
|
|
// connection returns false.
|
|
//
|
|
// Exported for work deferred to a pool: the connection a message arrived on can be given up while the work
|
|
// waits, and state from a connection that has been given up describes a router lifetime that may be over.
|
|
func (network *Network) IsCurrentConnection(r *model.Router) bool {
|
|
return network.Router.GetConnected(r.Id) == r && r.Connected.Load()
|
|
}
|
|
|
|
// displaceConnection removes cur, the connection occupying its router's connected slot, so that a redial
|
|
// can take the slot. Closing the channel is not sufficient on its own: if it is already closed, its close
|
|
// handler has already run and will never run again, so nothing would remove cur and every subsequent
|
|
// connect would be rejected against a slot that can never be freed. The teardown is therefore also
|
|
// invoked directly; it is gated on connection currency, so it is a no-op once the close handler has
|
|
// cleared the slot. Must be called with the router's connect stripe released, since the teardown
|
|
// acquires it.
|
|
func (network *Network) displaceConnection(cur *model.Router) {
|
|
if ch := cur.Control; ch != nil && !ch.IsClosed() {
|
|
if err := ch.Close(); err != nil {
|
|
pfxlog.Logger().WithError(err).WithField("routerId", cur.Id).
|
|
Error("error closing superseded control channel while rejecting connect")
|
|
}
|
|
}
|
|
network.DisconnectRouter(cur)
|
|
}
|
|
|
|
func (network *Network) DisconnectRouter(r *model.Router) {
|
|
// Lock-free pre-check: a stale/superseded disconnect (e.g. the old connection after a takeover) has
|
|
// nothing to tear down and must not touch the live connection's state; bail without blocking.
|
|
if !network.IsCurrentConnection(r) {
|
|
return
|
|
}
|
|
|
|
unlock := network.Router.LockConnectFor(r.Id)
|
|
defer unlock()
|
|
|
|
// Re-check under the stripe: a newer connection may have taken over between the pre-check and the
|
|
// lock. The teardown is all-or-nothing and must not run against a superseded connection.
|
|
if !network.IsCurrentConnection(r) {
|
|
return
|
|
}
|
|
|
|
// Snapshot the router's links before marking it disconnected: MarkDisconnected clears the
|
|
// router's link set (routerLinks.Clear()), so a later r.GetLinks() would return nothing and
|
|
// the link-removal/reroute cascade below would be skipped entirely.
|
|
links := r.GetLinks()
|
|
|
|
// Mark the router disconnected before the RerouteLink cascade, so reroute and everything it
|
|
// calls (shortestPath, connected-map reads) sees the dying router as gone. Otherwise reroute
|
|
// runs while the router still appears connected and can compute a replacement path through the
|
|
// very router that is being removed.
|
|
network.Router.MarkDisconnected(r)
|
|
|
|
// remove Links for Router, rerouting circuits off any that were connected
|
|
for _, l := range links {
|
|
if l.GetSrc().Id != r.Id {
|
|
continue
|
|
}
|
|
wasUsable := l.IsUsable()
|
|
if network.isHA {
|
|
// HA mode: mark links as down — the router may still be
|
|
// connected to other controllers.
|
|
l.SetDown(true)
|
|
} else {
|
|
// Single-controller mode: tombstone the link via gossip so
|
|
// the listener handles removal.
|
|
network.LinkGossipType.Delete(LinkGossipKey(l.Id, l.Iteration), l.GetSrc().Id)
|
|
}
|
|
if wasUsable {
|
|
network.RerouteLink(l)
|
|
}
|
|
}
|
|
|
|
for _, h := range network.routerPresenceHandlers.Value() {
|
|
h.RouterDisconnected(r)
|
|
}
|
|
}
|
|
|
|
func (network *Network) NotifyExistingLink(srcRouter *model.Router, reportedLink *ctrl_pb.RouterLinks_RouterLink) {
|
|
log := pfxlog.Logger().
|
|
WithField("routerId", srcRouter.Id).
|
|
WithField("linkId", reportedLink.Id).
|
|
WithField("destRouterId", reportedLink.DestRouterId).
|
|
WithField("iteration", reportedLink.Iteration)
|
|
|
|
// Publish under the stripe DisconnectRouter holds: checking currency and then publishing without it
|
|
// lets a report recreate a link after the teardown has snapshotted and cleared it. Events go out after
|
|
// the unlock, since a dispatcher may be slow and this stripe is shared with connect and disconnect.
|
|
unlock := network.Router.LockConnectFor(srcRouter.Id)
|
|
|
|
src := network.Router.GetConnected(srcRouter.Id)
|
|
if src == nil {
|
|
unlock()
|
|
log.Info("ignoring links message processed after router disconnected")
|
|
return
|
|
}
|
|
|
|
if src != srcRouter || !srcRouter.Connected.Load() {
|
|
unlock()
|
|
log.Info("ignoring links message processed from old router connection")
|
|
return
|
|
}
|
|
|
|
dst := network.Router.GetConnected(reportedLink.DestRouterId)
|
|
link, created := network.Link.RouterReportedLink(reportedLink, src, dst)
|
|
|
|
unlock()
|
|
|
|
if dst == nil {
|
|
network.NotifyLinkIdEvent(reportedLink.Id, event.LinkFromRouterDisconnectedDest)
|
|
}
|
|
|
|
if link == nil {
|
|
// Refused, because this router is not the link's source. RouterReportedLink has already said why.
|
|
return
|
|
}
|
|
|
|
if created {
|
|
network.NotifyLinkEvent(link, event.LinkFromRouterNew)
|
|
log.Info("router reported link added")
|
|
} else {
|
|
network.NotifyLinkEvent(link, event.LinkFromRouterKnown)
|
|
log.Info("router reported link already known")
|
|
}
|
|
}
|
|
|
|
func (network *Network) LinkFaulted(link *model.Link, dupe bool) {
|
|
wasUsable := link.IsUsable()
|
|
|
|
link.SetState(model.Failed)
|
|
if dupe {
|
|
network.NotifyLinkEvent(link, event.LinkDuplicate)
|
|
} else {
|
|
network.NotifyLinkEvent(link, event.LinkFault)
|
|
}
|
|
|
|
pfxlog.Logger().WithField("linkId", link.Id).Info("removing failed link")
|
|
network.Link.Remove(link)
|
|
|
|
if wasUsable {
|
|
network.RerouteLink(link)
|
|
}
|
|
}
|
|
|
|
func (network *Network) VerifyRouter(routerId string, fingerprints []string) error {
|
|
router, err := network.GetRouter(routerId)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
routerFingerprint := router.Fingerprint
|
|
if routerFingerprint == nil {
|
|
return fmt.Errorf("invalid router %v, not yet enrolled", routerId)
|
|
}
|
|
|
|
for _, fp := range fingerprints {
|
|
if fp == *routerFingerprint {
|
|
return nil
|
|
}
|
|
}
|
|
|
|
return fmt.Errorf("could not verify fingerprint for router %v", routerId)
|
|
}
|
|
|
|
func (network *Network) RerouteLink(l *model.Link) {
|
|
// This is called from Channel.rxer() and thus may not block
|
|
go func() {
|
|
network.handleRerouteLink(l)
|
|
}()
|
|
}
|
|
|
|
func (network *Network) CreateCircuit(params model.CreateCircuitParams) (*model.Circuit, error) {
|
|
clientId := params.GetClientId()
|
|
service := params.GetServiceId()
|
|
ctx := params.GetLogContext()
|
|
deadline := params.GetDeadline()
|
|
|
|
startTime := time.Now()
|
|
|
|
instanceId, serviceId := parseInstanceIdAndService(service)
|
|
|
|
// 1: Allocate Circuit Identifier
|
|
circuitId := params.GetCircuitId()
|
|
if circuitId == "" {
|
|
var err error
|
|
circuitId, err = idgen.NewUUIDString()
|
|
if err != nil {
|
|
network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, CircuitFailureIdGenerationError)
|
|
return nil, err
|
|
}
|
|
}
|
|
ctx.WithFields(map[string]interface{}{
|
|
"circuitId": circuitId,
|
|
"serviceId": service,
|
|
"attemptNumber": 1,
|
|
})
|
|
logger := pfxlog.ChannelLogger(logcontext.SelectPath).Wire(ctx).Entry
|
|
|
|
circuit := &model.Circuit{
|
|
Id: circuitId,
|
|
ClientId: clientId.Token,
|
|
ServiceId: serviceId,
|
|
Path: &model.Path{}, // empty until the circuit is built
|
|
}
|
|
|
|
// Reserve the circuit ID in the map to prevent collisions and protect routing.
|
|
// On any error path, the deferred cleanup removes the stub.
|
|
if !network.Circuit.Reserve(circuit) {
|
|
return nil, fmt.Errorf("circuit id %v already in use", circuitId)
|
|
}
|
|
removeReserved := true
|
|
defer func() {
|
|
if removeReserved {
|
|
network.Circuit.Remove(circuit)
|
|
}
|
|
}()
|
|
|
|
attempt := uint32(0)
|
|
allCleanups := make(map[string]struct{})
|
|
rs := network.newRouteSender(circuitId)
|
|
defer func() { network.removeRouteSender(rs) }()
|
|
for {
|
|
// 2: Find Service
|
|
svc, err := network.Service.Read(serviceId)
|
|
if err != nil {
|
|
network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, CircuitFailureInvalidService)
|
|
network.ServiceDialOtherError(serviceId)
|
|
return circuit, err
|
|
}
|
|
logger = logger.WithField("serviceName", svc.Name)
|
|
|
|
// 3: select terminator
|
|
strategy, terminator, pathNodes, strategyData, circuitErr := network.selectPath(params, svc, instanceId, ctx)
|
|
if circuitErr != nil {
|
|
network.CircuitFailedEvent(circuitId, params, startTime, nil, nil, circuitErr.Cause())
|
|
network.ServiceDialOtherError(serviceId)
|
|
return circuit, circuitErr
|
|
}
|
|
|
|
circuit.Terminator = terminator
|
|
|
|
// 4: Create Path
|
|
path, pathErr := network.CreatePathWithNodes(pathNodes)
|
|
if pathErr != nil {
|
|
network.CircuitFailedEvent(circuitId, params, startTime, nil, terminator, pathErr.Cause())
|
|
network.ServiceDialOtherError(serviceId)
|
|
return circuit, pathErr
|
|
}
|
|
|
|
circuit.Path = path
|
|
|
|
// get circuit tags
|
|
tags := params.GetCircuitTags(terminator)
|
|
|
|
circuit.Tags = tags
|
|
|
|
// 4a: Create Route Messages
|
|
rms := network.CreateRouteMessages(path, attempt, circuitId, terminator, deadline)
|
|
rms[len(rms)-1].Egress.PeerData = clientId.Data
|
|
for _, msg := range rms {
|
|
msg.Context = &ctrl_pb.Context{
|
|
Fields: ctx.GetStringFields(),
|
|
ChannelMask: ctx.GetChannelsMask(),
|
|
}
|
|
msg.Tags = tags
|
|
}
|
|
|
|
// 5: Routing
|
|
logger.Debug("route attempt for circuit")
|
|
peerData, cleanups, circuitErr := rs.route(attempt, path, rms, strategy, terminator, ctx.Clone())
|
|
for k, v := range cleanups {
|
|
allCleanups[k] = v
|
|
}
|
|
if circuitErr != nil {
|
|
logger.WithError(circuitErr).Warn("route attempt for circuit failed")
|
|
network.CircuitFailedEvent(circuitId, params, startTime, path, terminator, circuitErr.Cause())
|
|
attempt++
|
|
ctx.WithField("attemptNumber", attempt)
|
|
logger = logger.WithField("attemptNumber", attempt)
|
|
if attempt < network.options.CreateCircuitRetries {
|
|
continue
|
|
} else {
|
|
// revert successful routes
|
|
logger.Warnf("circuit creation failed after [%d] attempts, sending cleanup unroutes", network.options.CreateCircuitRetries)
|
|
for cleanupRId := range allCleanups {
|
|
if r := network.GetConnectedRouter(cleanupRId); r != nil {
|
|
if err := sendUnroute(r, circuitId, true); err == nil {
|
|
logger.WithField("routerId", cleanupRId).Debug("sent cleanup unroute for circuit")
|
|
} else {
|
|
logger.WithField("routerId", cleanupRId).Error("error sending cleanup unroute for circuit")
|
|
}
|
|
} else {
|
|
logger.WithField("routerId", cleanupRId).Error("router for circuit cleanup not connected")
|
|
}
|
|
}
|
|
|
|
return circuit, fmt.Errorf("exceeded maximum [%d] retries creating circuit [c/%s] (%w)", network.options.CreateCircuitRetries, circuitId, circuitErr)
|
|
}
|
|
}
|
|
|
|
// 5.a: Unroute Abandoned Routers (from Previous Attempts)
|
|
usedRouters := make(map[string]struct{})
|
|
for _, r := range path.Nodes {
|
|
usedRouters[r.Id] = struct{}{}
|
|
}
|
|
cleanupCount := 0
|
|
for cleanupRId := range allCleanups {
|
|
if _, found := usedRouters[cleanupRId]; !found {
|
|
cleanupCount++
|
|
if r := network.GetConnectedRouter(cleanupRId); r != nil {
|
|
if err := sendUnroute(r, circuitId, true); err == nil {
|
|
logger.WithField("routerId", cleanupRId).Debug("sent abandoned cleanup unroute for circuit to router")
|
|
} else {
|
|
logger.WithField("routerId", cleanupRId).WithError(err).Error("error sending abandoned cleanup unroute for circuit to router")
|
|
}
|
|
} else {
|
|
logger.WithField("routerId", cleanupRId).Error("router not connected for circuit, abandoned cleanup")
|
|
}
|
|
}
|
|
}
|
|
logger.Debugf("cleaned up [%d] abandoned routers for circuit", cleanupCount)
|
|
|
|
path.InitiatorLocalAddr = string(clientId.Data[uint32(ctrl_msg.InitiatorLocalAddressHeader)])
|
|
path.InitiatorRemoteAddr = string(clientId.Data[uint32(ctrl_msg.InitiatorRemoteAddressHeader)])
|
|
path.TerminatorLocalAddr = string(peerData[uint32(ctrl_msg.TerminatorLocalAddressHeader)])
|
|
path.TerminatorRemoteAddr = string(peerData[uint32(ctrl_msg.TerminatorRemoteAddressHeader)])
|
|
|
|
delete(peerData, uint32(ctrl_msg.InitiatorLocalAddressHeader))
|
|
delete(peerData, uint32(ctrl_msg.InitiatorRemoteAddressHeader))
|
|
delete(peerData, uint32(ctrl_msg.TerminatorLocalAddressHeader))
|
|
delete(peerData, uint32(ctrl_msg.TerminatorRemoteAddressHeader))
|
|
|
|
for k, v := range strategyData {
|
|
peerData[k] = v
|
|
}
|
|
|
|
now := time.Now()
|
|
// 6: Create Circuit Object
|
|
circuit.PeerData = peerData
|
|
circuit.CreatedAt = now
|
|
circuit.UpdatedAt = now
|
|
|
|
removeReserved = false // circuit is finalized, don't remove on defer
|
|
creationTimespan := time.Since(startTime)
|
|
network.CircuitEvent(event.CircuitCreated, circuit, &creationTimespan)
|
|
strategy.NotifyEvent(xt.NewDialSucceeded(terminator))
|
|
|
|
logger.WithField("path", circuit.Path).
|
|
WithField("terminator_local_address", circuit.Path.TerminatorLocalAddr).
|
|
WithField("terminator_remote_address", circuit.Path.TerminatorRemoteAddr).
|
|
Debug("created circuit")
|
|
return circuit, nil
|
|
}
|
|
}
|
|
|
|
func (network *Network) ReportForwardingFaults(ffr *ForwardingFaultReport) {
|
|
go network.handleForwardingFaults(ffr)
|
|
}
|
|
|
|
func parseInstanceIdAndService(service string) (string, string) {
|
|
atIndex := strings.IndexRune(service, '@')
|
|
if atIndex < 0 {
|
|
return "", service
|
|
}
|
|
identityId := service[0:atIndex]
|
|
serviceId := service[atIndex+1:]
|
|
return identityId, serviceId
|
|
}
|
|
|
|
func (network *Network) selectPath(params model.CreateCircuitParams, svc *model.Service, instanceId string, ctx logcontext.Context) (xt.Strategy, xt.CostedTerminator, []*model.Router, xt.PeerData, CircuitError) {
|
|
paths := map[string]*PathAndCost{}
|
|
var weightedTerminators []xt.CostedTerminator
|
|
var errList []error
|
|
|
|
log := pfxlog.ChannelLogger(logcontext.SelectPath).Wire(ctx)
|
|
|
|
hasOfflineRouters := false
|
|
pathError := false
|
|
|
|
for _, terminator := range svc.Terminators {
|
|
if terminator.InstanceId != instanceId {
|
|
continue
|
|
}
|
|
|
|
pathAndCost, found := paths[terminator.Router]
|
|
if !found {
|
|
dstR := network.Router.GetConnected(terminator.GetRouterId())
|
|
if dstR == nil {
|
|
err := fmt.Errorf("router with id=%v on terminator with id=%v for service name=%v is not online",
|
|
terminator.GetRouterId(), terminator.GetId(), svc.Name)
|
|
log.Debugf("error while calculating path for service %v: %v", svc.Id, err)
|
|
|
|
errList = append(errList, err)
|
|
hasOfflineRouters = true
|
|
continue
|
|
}
|
|
|
|
path, cost, err := network.shortestPath(params.GetSourceRouter(), dstR)
|
|
if err != nil {
|
|
log.Debugf("error while calculating path for service %v: %v", svc.Id, err)
|
|
errList = append(errList, err)
|
|
pathError = true
|
|
continue
|
|
}
|
|
|
|
pathAndCost = newPathAndCost(path, cost)
|
|
paths[terminator.GetRouterId()] = pathAndCost
|
|
}
|
|
|
|
dynamicCost := xt.GlobalCosts().GetDynamicCost(terminator.Id)
|
|
unbiasedCost := uint32(terminator.Cost) + uint32(dynamicCost) + pathAndCost.cost
|
|
biasedCost := terminator.Precedence.GetBiasedCost(unbiasedCost)
|
|
costedTerminator := &model.RoutingTerminator{
|
|
Terminator: terminator,
|
|
RouteCost: biasedCost,
|
|
}
|
|
weightedTerminators = append(weightedTerminators, costedTerminator)
|
|
}
|
|
|
|
if len(svc.Terminators) == 0 {
|
|
return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoTerminators, "service %v has no terminators", svc.Id)
|
|
}
|
|
|
|
if len(weightedTerminators) == 0 {
|
|
if pathError {
|
|
return nil, nil, nil, nil, newCircuitErrWrap(CircuitFailureNoPath, errors.Join(errList...))
|
|
}
|
|
|
|
if hasOfflineRouters {
|
|
return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoOnlineTerminators, "service %v has no online terminators for instanceId %v", svc.Id, instanceId)
|
|
}
|
|
|
|
return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureNoTerminators, "service %v has no terminators for instanceId %v", svc.Id, instanceId)
|
|
}
|
|
|
|
strategy, err := network.strategyRegistry.GetStrategy(svc.TerminatorStrategy)
|
|
if err != nil {
|
|
return nil, nil, nil, nil, newCircuitErrWrap(CircuitFailureInvalidStrategy, err)
|
|
}
|
|
|
|
sort.Slice(weightedTerminators, func(i, j int) bool {
|
|
return weightedTerminators[i].GetRouteCost() < weightedTerminators[j].GetRouteCost()
|
|
})
|
|
|
|
terminator, peerData, err := strategy.Select(params, weightedTerminators)
|
|
|
|
if err != nil {
|
|
return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureStrategyError, "strategy %v errored selecting terminator for service %v: %v", svc.TerminatorStrategy, svc.Id, err)
|
|
}
|
|
|
|
if terminator == nil {
|
|
return nil, nil, nil, nil, newCircuitErrorf(CircuitFailureStrategyError, "strategy %v did not select terminator for service %v", svc.TerminatorStrategy, svc.Id)
|
|
}
|
|
|
|
path := paths[terminator.GetRouterId()].path
|
|
|
|
if log.Logger.IsLevelEnabled(logrus.DebugLevel) {
|
|
buf := strings.Builder{}
|
|
buf.WriteString("[")
|
|
if len(weightedTerminators) > 0 {
|
|
fmt.Fprintf(&buf, "%v: %v", weightedTerminators[0].GetId(), weightedTerminators[0].GetRouteCost())
|
|
for _, t := range weightedTerminators[1:] {
|
|
buf.WriteString(", ")
|
|
fmt.Fprintf(&buf, "%v: %v", t.GetId(), t.GetRouteCost())
|
|
}
|
|
}
|
|
buf.WriteString("]")
|
|
var routerIds []string
|
|
for _, r := range path {
|
|
routerIds = append(routerIds, fmt.Sprintf("r/%s", r.Id))
|
|
}
|
|
pathStr := strings.Join(routerIds, "->")
|
|
log.Debugf("selected terminator %v for path %v from %v", terminator.GetId(), pathStr, buf.String())
|
|
}
|
|
|
|
return strategy, terminator, path, peerData, nil
|
|
}
|
|
|
|
func (network *Network) RemoveCircuit(circuitId string, now bool) error {
|
|
log := pfxlog.Logger().WithField("circuitId", circuitId)
|
|
|
|
if circuit, found := network.Circuit.Get(circuitId); found {
|
|
for _, r := range circuit.Path.Nodes {
|
|
err := sendUnroute(r, circuit.Id, now)
|
|
if err != nil {
|
|
log.Errorf("error sending unroute to [r/%s] (%s)", r.Id, err)
|
|
}
|
|
}
|
|
|
|
network.Circuit.Remove(circuit)
|
|
network.CircuitEvent(event.CircuitDeleted, circuit, nil)
|
|
|
|
if svc, err := network.Service.Read(circuit.ServiceId); err == nil {
|
|
if strategy, err := network.strategyRegistry.GetStrategy(svc.TerminatorStrategy); strategy != nil {
|
|
strategy.NotifyEvent(xt.NewCircuitRemoved(circuit.Terminator))
|
|
} else if err != nil {
|
|
log.WithError(err).WithField("terminatorStrategy", svc.TerminatorStrategy).Warn("failed to notify strategy of circuit end, invalid strategy")
|
|
}
|
|
} else {
|
|
log.WithError(err).Error("unable to get service for circuit")
|
|
}
|
|
|
|
log.Debug("removed circuit")
|
|
|
|
return nil
|
|
}
|
|
return InvalidCircuitError{circuitId: circuitId}
|
|
}
|
|
|
|
func (network *Network) CreatePath(srcR, dstR *model.Router) (*model.Path, error) {
|
|
ingressId, err := idgen.NewUUIDString()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
egressId, err := idgen.NewUUIDString()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
path := &model.Path{
|
|
Links: make([]*model.Link, 0),
|
|
IngressId: ingressId,
|
|
EgressId: egressId,
|
|
Nodes: make([]*model.Router, 0),
|
|
}
|
|
path.Nodes = append(path.Nodes, srcR)
|
|
path.Nodes = append(path.Nodes, dstR)
|
|
|
|
return network.UpdatePath(path)
|
|
}
|
|
|
|
func (network *Network) setLinks(path *model.Path) error {
|
|
if len(path.Nodes) > 1 {
|
|
for i := 0; i < len(path.Nodes)-1; i++ {
|
|
if link, found := network.Link.LeastExpensiveLink(path.Nodes[i], path.Nodes[i+1]); found {
|
|
path.Links = append(path.Links, link)
|
|
} else {
|
|
return fmt.Errorf("no link from r/%v to r/%v", path.Nodes[i].Id, path.Nodes[i+1].Id)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (network *Network) AddRouterPresenceHandler(h model.RouterPresenceHandler) {
|
|
network.routerPresenceHandlers.Append(h)
|
|
}
|
|
|
|
func (network *Network) Run() {
|
|
defer logrus.Info("exited")
|
|
logrus.Info("started")
|
|
|
|
go network.watchdog()
|
|
|
|
ticker := time.NewTicker(time.Duration(network.options.CycleSeconds) * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
network.clean()
|
|
network.smart()
|
|
if !network.isHA {
|
|
network.Link.ScanForDeadLinks()
|
|
}
|
|
|
|
case <-network.closeNotify:
|
|
network.eventDispatcher.RemoveMetricsMessageHandler(network)
|
|
network.metricsRegistry.DisposeAll()
|
|
return
|
|
}
|
|
|
|
// notify the watchdog that we're processing
|
|
select {
|
|
case network.watchdogCh <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (network *Network) watchdog() {
|
|
watchdogInterval := 2 * time.Duration(network.options.CycleSeconds) * time.Second
|
|
consecutiveFails := 0
|
|
|
|
watchDogTicker := time.NewTicker(watchdogInterval)
|
|
defer watchDogTicker.Stop()
|
|
|
|
for {
|
|
// check every 2x cycle seconds
|
|
select {
|
|
case <-watchDogTicker.C:
|
|
case <-network.closeNotify:
|
|
return
|
|
}
|
|
|
|
select {
|
|
case <-network.watchdogCh:
|
|
consecutiveFails = 0
|
|
continue
|
|
case <-network.closeNotify:
|
|
return
|
|
default:
|
|
consecutiveFails++
|
|
// network.Run didn't complete, something is stalling it
|
|
pfxlog.Logger().
|
|
WithField("watchdogInterval", watchdogInterval.String()).
|
|
WithField("consecutiveFails", consecutiveFails).
|
|
Warn("network.Run did not finish within watchdog interval")
|
|
|
|
if consecutiveFails == 3 {
|
|
debugz.DumpStack()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (network *Network) handleRerouteLink(l *model.Link) {
|
|
log := logrus.WithField("linkId", l.Id)
|
|
log.Info("changed link")
|
|
if err := network.rerouteLink(l, time.Now().Add(config.DefaultOptionsRouteTimeout)); err != nil {
|
|
log.WithError(err).Error("unexpected error rerouting link")
|
|
}
|
|
}
|
|
|
|
func (network *Network) handleForwardingFaults(ffr *ForwardingFaultReport) {
|
|
network.fault(ffr)
|
|
}
|
|
|
|
func (network *Network) AddCapability(capability string) {
|
|
network.lock.Lock()
|
|
defer network.lock.Unlock()
|
|
network.capabilities = append(network.capabilities, capability)
|
|
}
|
|
|
|
func (network *Network) GetCapabilities() []string {
|
|
network.lock.Lock()
|
|
defer network.lock.Unlock()
|
|
return network.capabilities
|
|
}
|
|
|
|
func (network *Network) RemoveLink(linkId string) {
|
|
log := pfxlog.Logger().WithField("linkId", linkId)
|
|
|
|
link, _ := network.Link.Get(linkId)
|
|
var iteration uint32
|
|
|
|
var routerList []*model.Router
|
|
if link != nil {
|
|
iteration = link.Iteration
|
|
routerList = []*model.Router{link.GetSrc()}
|
|
if dst := link.GetDest(); dst != nil {
|
|
routerList = append(routerList, dst)
|
|
}
|
|
log = log.WithField("srcRouterId", link.GetSrc().Id).
|
|
WithField("dstRouterId", link.DstId).
|
|
WithField("iteration", iteration)
|
|
log.Info("deleting known link")
|
|
} else {
|
|
routerList = network.AllConnectedRouters()
|
|
log.Info("deleting unknown link (sending link fault to all connected routers)")
|
|
}
|
|
|
|
for _, router := range routerList {
|
|
fault := &ctrl_pb.Fault{
|
|
Subject: ctrl_pb.FaultSubject_LinkFault,
|
|
Id: linkId,
|
|
Iteration: iteration,
|
|
}
|
|
|
|
if ctrl := router.Control; ctrl != nil {
|
|
if err := protobufs.MarshalTyped(fault).WithTimeout(15 * time.Second).Send(ctrl.GetDefaultSender()); err != nil {
|
|
log.WithField("faultDestRouterId", router.Id).WithError(err).
|
|
Error("failed to send link fault to router on link removal")
|
|
} else {
|
|
log.WithField("faultDestRouterId", router.Id).WithError(err).
|
|
Info("sent link fault to router on link removal")
|
|
}
|
|
}
|
|
}
|
|
|
|
if link != nil {
|
|
network.Link.Remove(link)
|
|
network.RerouteLink(link)
|
|
}
|
|
}
|
|
|
|
func (network *Network) rerouteLink(l *model.Link, deadline time.Time) error {
|
|
circuits := network.Circuit.All()
|
|
for _, circuit := range circuits {
|
|
if circuit.Path.UsesLink(l) {
|
|
log := logrus.WithField("linkId", l.Id).
|
|
WithField("circuitId", circuit.Id)
|
|
log.Info("circuit uses link")
|
|
if err := network.rerouteCircuit(circuit, deadline); err != nil {
|
|
log.WithError(err).Error("error rerouting circuit, removing")
|
|
if err := network.RemoveCircuit(circuit.Id, true); err != nil {
|
|
log.WithError(err).Error("error removing circuit after reroute failure")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (network *Network) rerouteCircuitWithTries(circuit *model.Circuit, retries int) bool {
|
|
log := pfxlog.Logger().WithField("circuitId", circuit.Id)
|
|
|
|
// Path is nil for reserved circuits that haven't been built yet
|
|
if circuit.Path.IsValid() {
|
|
for i := 0; i < retries; i++ {
|
|
deadline := time.Now().Add(config.DefaultOptionsRouteTimeout)
|
|
err := network.rerouteCircuit(circuit, deadline)
|
|
if err == nil {
|
|
return true
|
|
}
|
|
|
|
log.WithError(err).WithField("attempt", i).Error("error re-routing circuit")
|
|
}
|
|
}
|
|
|
|
if err := network.RemoveCircuit(circuit.Id, true); err != nil {
|
|
log.WithError(err).Error("failure while removing circuit after failed re-route attempt")
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (network *Network) rerouteCircuit(circuit *model.Circuit, deadline time.Time) error {
|
|
log := pfxlog.Logger().WithField("circuitId", circuit.Id)
|
|
if circuit.Rerouting.CompareAndSwap(false, true) {
|
|
defer circuit.Rerouting.Store(false)
|
|
|
|
log.Warn("rerouting circuit")
|
|
|
|
oldPath := circuit.Path
|
|
if cq, err := network.UpdatePath(circuit.Path); err == nil {
|
|
circuit.Path = cq
|
|
circuit.UpdatedAt = time.Now()
|
|
|
|
rms := network.CreateRouteMessages(cq, SmartRerouteAttempt, circuit.Id, circuit.Terminator, deadline)
|
|
|
|
for i := 0; i < len(cq.Nodes); i++ {
|
|
if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
|
|
log.WithError(err).Errorf("error sending route to [r/%s]", cq.Nodes[i].Id)
|
|
return err
|
|
}
|
|
}
|
|
|
|
network.unrouteRemovedPathNodes(log, circuit.Id, oldPath, cq)
|
|
|
|
log.Info("rerouted circuit")
|
|
|
|
network.CircuitEvent(event.CircuitUpdated, circuit, nil)
|
|
return nil
|
|
} else {
|
|
return err
|
|
}
|
|
} else {
|
|
log.Info("not rerouting circuit, already in progress")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (network *Network) smartReroute(circuit *model.Circuit, cq *model.Path, deadline time.Time) bool {
|
|
retry := false
|
|
log := pfxlog.Logger().WithField("circuitId", circuit.Id)
|
|
if circuit.Rerouting.CompareAndSwap(false, true) {
|
|
defer circuit.Rerouting.Store(false)
|
|
|
|
oldPath := circuit.Path
|
|
circuit.Path = cq
|
|
circuit.UpdatedAt = time.Now()
|
|
|
|
rms := network.CreateRouteMessages(cq, SmartRerouteAttempt, circuit.Id, circuit.Terminator, deadline)
|
|
|
|
for i := 0; i < len(cq.Nodes); i++ {
|
|
if _, err := sendRoute(cq.Nodes[i], rms[i], network.options.RouteTimeout); err != nil {
|
|
retry = true
|
|
log.WithField("routerId", cq.Nodes[i].Id).WithError(err).Error("error sending smart route update to router")
|
|
break
|
|
}
|
|
}
|
|
|
|
if !retry {
|
|
network.unrouteRemovedPathNodes(log, circuit.Id, oldPath, cq)
|
|
logrus.Debug("rerouted circuit")
|
|
network.CircuitEvent(event.CircuitUpdated, circuit, nil)
|
|
}
|
|
}
|
|
return retry
|
|
}
|
|
|
|
func (network *Network) AcceptMetricsMsg(metrics *metrics_pb.MetricsMessage) {
|
|
if metrics.SourceId == network.nodeId {
|
|
return // ignore metrics coming from the controller itself
|
|
}
|
|
|
|
log := pfxlog.Logger()
|
|
|
|
router, err := network.Router.Read(metrics.SourceId)
|
|
if err != nil {
|
|
log.Debugf("could not find router [r/%s] while processing metrics", metrics.SourceId)
|
|
return
|
|
}
|
|
|
|
// When the reporting router publishes per-link latency over gossip, routing
|
|
// latency comes from the link-metrics store instead, so we skip deriving it
|
|
// from this message. The latency histograms still travel in the message and
|
|
// still feed observability events; they just stop feeding routing.
|
|
if metrics.LinkLatencyInGossip {
|
|
return
|
|
}
|
|
|
|
for _, link := range network.GetAllLinksForRouter(router.Id) {
|
|
metricId := "link." + link.Id + ".latency"
|
|
var latencyCost int64
|
|
var found bool
|
|
if latency, ok := metrics.Histograms[metricId]; ok {
|
|
latencyCost = int64(latency.Mean)
|
|
found = true
|
|
|
|
metricId = "link." + link.Id + ".queue_time"
|
|
if queueTime, ok := metrics.Histograms[metricId]; ok {
|
|
latencyCost += int64(queueTime.Mean)
|
|
}
|
|
}
|
|
|
|
if found {
|
|
if link.GetSrc().Id == router.Id {
|
|
link.SetSrcLatency(latencyCost) // latency is in nanoseconds
|
|
} else if link.DstId == router.Id {
|
|
link.SetDstLatency(latencyCost) // latency is in nanoseconds
|
|
} else {
|
|
log.Warnf("link not for router")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func sendRoute(r *model.Router, createMsg *ctrl_pb.Route, timeout time.Duration) (xt.PeerData, error) {
|
|
log := pfxlog.Logger().WithField("routerId", r.Id).
|
|
WithField("circuitId", createMsg.CircuitId)
|
|
|
|
log.Debug("sending create route message")
|
|
|
|
msg, err := protobufs.MarshalTyped(createMsg).WithTimeout(timeout).SendForReply(r.Control.GetHighPrioritySender())
|
|
if err != nil {
|
|
log.WithError(err).WithField("timeout", timeout).Error("error sending route message")
|
|
return nil, err
|
|
}
|
|
|
|
if msg.ContentType == ctrl_msg.RouteResultType {
|
|
_, success := msg.Headers[ctrl_msg.RouteResultSuccessHeader]
|
|
if !success {
|
|
message := "route error, but no error message from router"
|
|
if errMsg, found := msg.Headers[ctrl_msg.RouteResultErrorHeader]; found {
|
|
message = string(errMsg)
|
|
}
|
|
return nil, errors.New(message)
|
|
}
|
|
|
|
peerData := xt.PeerData{}
|
|
for k, v := range msg.Headers {
|
|
if k > 0 {
|
|
peerData[uint32(k)] = v
|
|
}
|
|
}
|
|
|
|
return peerData, nil
|
|
}
|
|
return nil, fmt.Errorf("unexpected response type %v received in reply to route request", msg.ContentType)
|
|
}
|
|
|
|
func sendUnroute(r *model.Router, circuitId string, now bool) error {
|
|
unroute := &ctrl_pb.Unroute{
|
|
CircuitId: circuitId,
|
|
Now: now,
|
|
}
|
|
return protobufs.MarshalTyped(unroute).Send(r.Control.GetHighPrioritySender())
|
|
}
|
|
|
|
// unrouteRemovedPathNodes sends Unroute to any nodes that were in the old path but are not in the new path.
|
|
// This handles the case where a circuit reroute changes intermediate transit nodes.
|
|
func (network *Network) unrouteRemovedPathNodes(log *logrus.Entry, circuitId string, oldPath, newPath *model.Path) {
|
|
newNodeIds := make(map[string]struct{}, len(newPath.Nodes))
|
|
for _, r := range newPath.Nodes {
|
|
newNodeIds[r.Id] = struct{}{}
|
|
}
|
|
|
|
for _, r := range oldPath.Nodes {
|
|
if _, ok := newNodeIds[r.Id]; !ok {
|
|
if cr := network.GetConnectedRouter(r.Id); cr != nil {
|
|
if err := sendUnroute(cr, circuitId, true); err != nil {
|
|
log.WithError(err).Errorf("error sending unroute to removed path node [r/%s]", r.Id)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (network *Network) showOptions() {
|
|
if jsonOptions, err := json.MarshalIndent(network.options, "", " "); err == nil {
|
|
pfxlog.Logger().Infof("network = %s", string(jsonOptions))
|
|
} else {
|
|
panic(err)
|
|
}
|
|
}
|
|
|
|
type renderConfig interface {
|
|
RenderJsonConfig() (string, error)
|
|
}
|
|
|
|
func (network *Network) routerDeleted(routerId string) {
|
|
circuits := network.GetAllCircuits()
|
|
for _, circuit := range circuits {
|
|
if circuit.HasRouter(routerId) {
|
|
path := circuit.Path
|
|
|
|
// If we're either the initiator, terminator (or both), cleanup the circuit since
|
|
// we won't be able to re-establish it, and we'll never get a circuit fault
|
|
if path.Nodes[0].Id == routerId || path.Nodes[len(path.Nodes)-1].Id == routerId {
|
|
if err := network.RemoveCircuit(circuit.Id, true); err != nil {
|
|
pfxlog.Logger().WithField("routerId", routerId).
|
|
WithField("circuitId", circuit.Id).
|
|
WithError(err).Error("unable to remove circuit after router was deleted")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
var DbSnapshotTooFrequentError = dbSnapshotTooFrequentError{}
|
|
|
|
type dbSnapshotTooFrequentError struct{}
|
|
|
|
func (d dbSnapshotTooFrequentError) Error() string {
|
|
return "may snapshot database at most once per minute"
|
|
}
|
|
|
|
func (network *Network) SnapshotDatabase() error {
|
|
_, err := network.SnapshotDatabaseToFile("")
|
|
return err
|
|
}
|
|
|
|
func (network *Network) SnapshotDatabaseToFile(path string) (string, error) {
|
|
network.lock.Lock()
|
|
defer network.lock.Unlock()
|
|
|
|
if network.lastSnapshot.Add(time.Minute).After(time.Now()) {
|
|
return "", DbSnapshotTooFrequentError
|
|
}
|
|
|
|
actualPath := path
|
|
if actualPath == "" {
|
|
actualPath = "__DB_DIR__/__DB_FILE__-__DATE__-__TIME__"
|
|
}
|
|
|
|
err := network.GetDb().View(func(tx *bbolt.Tx) error {
|
|
currentIndex := db.LoadCurrentRaftIndex(tx)
|
|
actualPath = strings.ReplaceAll(actualPath, "__RAFT_INDEX__", fmt.Sprintf("%v", currentIndex))
|
|
actualPath = strings.ReplaceAll(actualPath, "RAFT_INDEX", fmt.Sprintf("%v", currentIndex))
|
|
|
|
var err error
|
|
actualPath, _, err = network.GetDb().SnapshotInTx(tx, actualPath)
|
|
return err
|
|
})
|
|
|
|
if err == nil {
|
|
network.lastSnapshot = time.Now()
|
|
}
|
|
|
|
return actualPath, err
|
|
}
|
|
|
|
func (network *Network) RestoreSnapshot(cmd *command.SyncSnapshotCommand, index uint64) error {
|
|
log := pfxlog.Logger()
|
|
currentTimelineId, err := network.GetDb().GetTimelineId(boltz.TimelineModeDefault, shortid.Generate)
|
|
if err != nil {
|
|
log.WithError(err).Error("unable to get current timeline id")
|
|
}
|
|
if currentTimelineId != "" && currentTimelineId == cmd.TimelineId {
|
|
log.WithField("timelineId", cmd.TimelineId).Info("snapshot already current, skipping reload")
|
|
// DB already restored; ensure cluster id then raft index (index last, see main path).
|
|
if err = network.ensureClusterId(cmd.ClusterId); err != nil {
|
|
return fmt.Errorf("failed to set cluster id for already-current snapshot (%w)", err)
|
|
}
|
|
if err = network.ensureRaftIndex(index); err != nil {
|
|
return fmt.Errorf("failed to set raft index for already-current snapshot (%w)", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
buf := bytes.NewBuffer(cmd.Snapshot)
|
|
reader, err := gzip.NewReader(buf)
|
|
if err != nil {
|
|
return fmt.Errorf("unable to create gz reader for reading migration snapshot during restore (%w)", err)
|
|
}
|
|
|
|
network.GetDb().RestoreFromReader(reader)
|
|
|
|
// Write the cluster id before the raft index. The index is the completion gate on restart (the
|
|
// FSM skips entries at or below the stored index), so persist it last: any earlier failure then
|
|
// replays and retries instead of skipping the command with a blank cluster id.
|
|
if err = network.ensureClusterId(cmd.ClusterId); err != nil {
|
|
return fmt.Errorf("failed to set cluster id after db restore (%w)", err)
|
|
}
|
|
if err = network.ensureRaftIndex(index); err != nil {
|
|
return fmt.Errorf("failed to set raft index after db restore (%w)", err)
|
|
}
|
|
|
|
time.AfterFunc(5*time.Second, func() {
|
|
if network.restartSelfOnSnapshot {
|
|
log.Info("database restore requires controller restart, restarting...")
|
|
// RestartController returns only when the restart failed; on success it replaces this
|
|
// process and never comes back.
|
|
err := raft.RestartController()
|
|
log.WithError(err).Error("failed to restart controller after snapshot restore, exiting...")
|
|
} else {
|
|
log.Info("database restore requires controller restart. exiting...")
|
|
}
|
|
os.Exit(0)
|
|
})
|
|
|
|
return nil
|
|
}
|
|
|
|
// ensureClusterId writes the cluster id if one is not already set (no-op when empty or matching).
|
|
// The snapshot restore replaces the whole db, which carries no cluster id, so it must be set here.
|
|
func (network *Network) ensureClusterId(clusterId string) error {
|
|
if clusterId == "" {
|
|
return nil
|
|
}
|
|
_, err := db.InitClusterId(network.GetDb(), nil, clusterId)
|
|
return err
|
|
}
|
|
|
|
// ensureRaftIndex records the raft index if the stored one is behind it, returning an error on
|
|
// failure so a missed index update surfaces rather than being treated as success.
|
|
func (network *Network) ensureRaftIndex(index uint64) error {
|
|
return network.GetDb().Update(nil, func(ctx boltz.MutateContext) error {
|
|
if db.LoadCurrentRaftIndex(ctx.Tx()) >= index {
|
|
return nil
|
|
}
|
|
raftBucket := boltz.GetOrCreatePath(ctx.Tx(), db.RootBucket, db.MetadataBucket)
|
|
raftBucket.SetInt64(db.FieldRaftIndex, int64(index), nil)
|
|
return raftBucket.GetError()
|
|
})
|
|
}
|
|
|
|
// SetRestartSelfOnSnapshot controls whether the controller restarts itself after a snapshot restore
|
|
// (true) or exits expecting an external restart (false). It mirrors the raft restartSelfOnSnapshot
|
|
// setting and is applied by the owning controller after construction.
|
|
func (network *Network) SetRestartSelfOnSnapshot(v bool) {
|
|
network.restartSelfOnSnapshot = v
|
|
}
|
|
|
|
func (network *Network) AddInspectTarget(target InspectTarget) {
|
|
network.inspectionTargets.Append(target)
|
|
}
|
|
|
|
// SetCtrlDialerValidator registers the ctrl dialer's validation function with the network.
|
|
func (network *Network) SetCtrlDialerValidator(validator CtrlDialerValidator) {
|
|
network.ctrlDialerValidator = validator
|
|
}
|
|
|
|
// GetCtrlDialerValidator returns the registered ctrl dialer validation function, or nil.
|
|
func (network *Network) GetCtrlDialerValidator() CtrlDialerValidator {
|
|
return network.ctrlDialerValidator
|
|
}
|
|
|
|
func (network *Network) ValidateRouterLinks(router *model.Router, cb LinkValidationCallback) {
|
|
request := &ctrl_pb.InspectRequest{RequestedValues: []string{"links"}}
|
|
resp := &ctrl_pb.InspectResponse{}
|
|
respMsg, err := protobufs.MarshalTyped(request).WithTimeout(time.Minute).SendForReply(router.Control.GetDefaultSender())
|
|
if err = protobufs.TypedResponse(resp).Unmarshall(respMsg, err); err != nil {
|
|
network.reportRouterLinksError(router, err, cb)
|
|
return
|
|
}
|
|
|
|
var linkDetails *inspect.LinksInspectResult
|
|
for _, val := range resp.Values {
|
|
if val.Name == "links" {
|
|
if err = json.Unmarshal([]byte(val.Value), &linkDetails); err != nil {
|
|
network.reportRouterLinksError(router, err, cb)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
if linkDetails == nil {
|
|
if len(resp.Errors) > 0 {
|
|
err = errors.New(strings.Join(resp.Errors, ","))
|
|
network.reportRouterLinksError(router, err, cb)
|
|
return
|
|
}
|
|
network.reportRouterLinksError(router, errors.New("no link details returned from router"), cb)
|
|
return
|
|
}
|
|
|
|
linkMap := network.Link.GetLinkMap()
|
|
|
|
result := &mgmt_pb.RouterLinkDetails{
|
|
RouterId: router.Id,
|
|
RouterName: router.Name,
|
|
ValidateSuccess: true,
|
|
}
|
|
|
|
for _, link := range linkDetails.Links {
|
|
detail := &mgmt_pb.RouterLinkDetail{
|
|
LinkId: link.Id,
|
|
RouterState: mgmt_pb.LinkState_LinkEstablished,
|
|
DestRouterId: link.Dest,
|
|
Dialed: link.Dialed,
|
|
}
|
|
detail.DestConnected = network.ConnectedRouter(link.Dest)
|
|
if ctrlLink, found := linkMap[link.Id]; found {
|
|
detail.CtrlState = mgmt_pb.LinkState_LinkEstablished
|
|
detail.IsValid = detail.DestConnected
|
|
|
|
checkLinkDest(ctrlLink, detail)
|
|
|
|
if link.Dialed { // only compare against dialer side of the link, as src/dst will be flipped on the listener side
|
|
network.checkLinkConns(ctrlLink, link, detail)
|
|
}
|
|
} else {
|
|
detail.CtrlState = mgmt_pb.LinkState_LinkUnknown
|
|
detail.IsValid = !detail.DestConnected
|
|
}
|
|
|
|
// Gossip entries are owned by the dialing router. From this side of the
|
|
// validation, if the router we're asking dialed the link, it's the owner;
|
|
// otherwise the destination is.
|
|
gossipOwner := router.Id
|
|
if !link.Dialed {
|
|
gossipOwner = link.Dest
|
|
}
|
|
gossipKey := LinkGossipKey(link.Id, link.Iteration)
|
|
gossipVal, gossipVer, gossipFound := network.LinkGossipType.GetForOwner(gossipOwner, gossipKey)
|
|
if gossipFound {
|
|
detail.InGossipStore = true
|
|
detail.GossipVersion = gossipVer
|
|
detail.GossipIteration = gossipVal.Iteration
|
|
if ctrlLink, ctrlFound := linkMap[link.Id]; ctrlFound && gossipVal.Iteration != ctrlLink.Iteration {
|
|
detail.Messages = append(detail.Messages, fmt.Sprintf(
|
|
"gossip iteration (%d) differs from ctrl iteration (%d)", gossipVal.Iteration, ctrlLink.Iteration))
|
|
}
|
|
} else if detail.CtrlState == mgmt_pb.LinkState_LinkEstablished {
|
|
detail.Messages = append(detail.Messages, "link in ctrl but missing from gossip store")
|
|
}
|
|
|
|
delete(linkMap, link.Id)
|
|
result.LinkDetails = append(result.LinkDetails, detail)
|
|
}
|
|
|
|
for _, link := range linkMap {
|
|
related := false
|
|
dest := ""
|
|
if link.GetSrc().Id == router.Id {
|
|
related = true
|
|
dest = link.DstId
|
|
} else if link.DstId == router.Id {
|
|
related = true
|
|
dest = link.GetSrc().Id
|
|
}
|
|
|
|
if related {
|
|
detail := &mgmt_pb.RouterLinkDetail{
|
|
LinkId: link.Id,
|
|
CtrlState: mgmt_pb.LinkState_LinkEstablished,
|
|
DestConnected: network.ConnectedRouter(dest),
|
|
RouterState: mgmt_pb.LinkState_LinkUnknown,
|
|
IsValid: false,
|
|
DestRouterId: dest,
|
|
Dialed: link.GetSrc().Id == router.Id,
|
|
}
|
|
|
|
gossipKey := LinkGossipKey(link.Id, link.Iteration)
|
|
if gossipVal, gossipVer, gossipFound := network.LinkGossipType.GetForOwner(link.GetSrc().Id, gossipKey); gossipFound {
|
|
detail.InGossipStore = true
|
|
detail.GossipVersion = gossipVer
|
|
detail.GossipIteration = gossipVal.Iteration
|
|
}
|
|
|
|
result.LinkDetails = append(result.LinkDetails, detail)
|
|
}
|
|
}
|
|
|
|
cb(result)
|
|
}
|
|
|
|
// checkLinkDest flags a link that names a connected destination router without referencing it.
|
|
//
|
|
// Such a link carries no adjacency: ConnectedNeighborsOfRouter skips it, so nothing can be routed over it. And
|
|
// nothing else says so. Both ends report it established, the destination reads as connected because that is
|
|
// answered from the connected router map rather than from the link, and the link listing renders the
|
|
// destination by reading it back from the database.
|
|
//
|
|
// Only checked when the destination is connected. A link to a router this controller has no connection to is
|
|
// expected to reference none, and in HA that is routine rather than a fault.
|
|
func checkLinkDest(ctrlLink *model.Link, detail *mgmt_pb.RouterLinkDetail) {
|
|
if !detail.DestConnected || ctrlLink.GetDest() != nil {
|
|
return
|
|
}
|
|
|
|
detail.IsValid = false
|
|
detail.Messages = append(detail.Messages,
|
|
"destination router is connected but the link does not reference it, so the link carries no adjacency and cannot be routed over")
|
|
}
|
|
|
|
func (network *Network) checkLinkConns(ctrlLink *model.Link, routerLink *inspect.LinkInspectDetail, result *mgmt_pb.RouterLinkDetail) {
|
|
sortF := func(v []*ctrl_pb.LinkConn) []*ctrl_pb.LinkConn {
|
|
return slices.SortedFunc(slices.Values(v), func(e *ctrl_pb.LinkConn, e2 *ctrl_pb.LinkConn) int {
|
|
if diff := strings.Compare(e.Type, e2.Type); diff != 0 {
|
|
return diff
|
|
}
|
|
if diff := strings.Compare(e.LocalAddr, e2.LocalAddr); diff != 0 {
|
|
return diff
|
|
}
|
|
return strings.Compare(e.RemoteAddr, e2.RemoteAddr)
|
|
})
|
|
}
|
|
|
|
var routerConns []*ctrl_pb.LinkConn
|
|
|
|
for _, v := range routerLink.Connections {
|
|
routerConns = append(routerConns, &ctrl_pb.LinkConn{
|
|
Type: v.Type,
|
|
LocalAddr: v.Source,
|
|
RemoteAddr: v.Dest,
|
|
})
|
|
}
|
|
|
|
// The version comes from the hello, so only the connected instance has it, and a link endpoint can
|
|
// still be the database-loaded placeholder a gossiped entry created. Not knowing the version means the
|
|
// comparison cannot be made, which is not a fault of the link.
|
|
if srcR := ctrlLink.GetSrc(); srcR != nil {
|
|
connectedSrc := network.Router.GetConnected(srcR.Id)
|
|
if connectedSrc == nil || connectedSrc.VersionInfo == nil {
|
|
return
|
|
}
|
|
hasMinVersion, err := connectedSrc.VersionInfo.HasMinimumVersion("v1.6.6")
|
|
if err != nil {
|
|
result.IsValid = false
|
|
result.Messages = append(result.Messages, err.Error())
|
|
return
|
|
}
|
|
if !hasMinVersion {
|
|
return
|
|
}
|
|
}
|
|
|
|
ctrlConnState := ctrlLink.GetConnsState()
|
|
|
|
// The iteration decides whether the two views are comparable at all. The router bumps it on every
|
|
// underlay change and reports it together with the conns it describes, while the controller only ever
|
|
// replaces its copy with a wholly newer snapshot. So the router running ahead simply means an update is
|
|
// still in flight, and the conns are expected to differ until it lands; the controller running ahead
|
|
// means its copy outlived what the router is reporting. Neither says anything about whether the two
|
|
// agree, so there is nothing to compare and nothing to report.
|
|
if routerLink.ConnStateIteration != ctrlConnState.GetStateIteration() {
|
|
return
|
|
}
|
|
|
|
// Same iteration, so both sides describe the same state and any difference between them is a real
|
|
// disagreement rather than one side being behind.
|
|
ctrlConns := sortF(ctrlConnState.GetConns())
|
|
routerConns = sortF(routerConns)
|
|
|
|
if len(ctrlConns) != len(routerConns) {
|
|
result.IsValid = false
|
|
result.Messages = append(result.Messages, fmt.Sprintf("for link %s, len(ctrlConns): %d != len(routerConns): %d",
|
|
ctrlLink.Id, len(ctrlConns), len(routerConns)))
|
|
return
|
|
}
|
|
|
|
for i := 0; i < len(ctrlConns); i++ {
|
|
ctrlConn := ctrlConns[i]
|
|
routerConn := routerConns[i]
|
|
if ctrlConn.Type != routerConn.Type {
|
|
result.IsValid = false
|
|
result.Messages = append(result.Messages, fmt.Sprintf("for link %s, type doesn't match. ctrl %s != router %s",
|
|
ctrlLink.Id, ctrlConn.Type, routerConn.Type))
|
|
}
|
|
if ctrlConn.LocalAddr != routerConn.LocalAddr {
|
|
result.IsValid = false
|
|
result.Messages = append(result.Messages, fmt.Sprintf("for link %s, local addr doesn't match. ctrl %s != router %s",
|
|
ctrlLink.Id, ctrlConn.LocalAddr, routerConn.LocalAddr))
|
|
}
|
|
if ctrlConn.RemoteAddr != routerConn.RemoteAddr {
|
|
result.IsValid = false
|
|
result.Messages = append(result.Messages, fmt.Sprintf("for link %s, remote addr doesn't match. ctrl %s != router %s",
|
|
ctrlLink.Id, ctrlConn.RemoteAddr, routerConn.RemoteAddr))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (network *Network) reportRouterLinksError(router *model.Router, err error, cb LinkValidationCallback) {
|
|
result := &mgmt_pb.RouterLinkDetails{
|
|
RouterId: router.Id,
|
|
RouterName: router.Name,
|
|
ValidateSuccess: false,
|
|
Message: err.Error(),
|
|
}
|
|
cb(result)
|
|
}
|
|
|
|
func minCost(q map[*model.Router]bool, dist map[*model.Router]int64) *model.Router {
|
|
if len(dist) < 1 {
|
|
return nil
|
|
}
|
|
|
|
currentMin := int64(math.MaxInt64)
|
|
var selected *model.Router
|
|
for r := range q {
|
|
d := dist[r]
|
|
if d <= currentMin {
|
|
selected = r
|
|
currentMin = d
|
|
}
|
|
}
|
|
return selected
|
|
}
|
|
|
|
type Cache interface {
|
|
RemoveFromCache(id string)
|
|
}
|
|
|
|
func newPathAndCost(path []*model.Router, cost int64) *PathAndCost {
|
|
if cost > (1 << 20) {
|
|
cost = 1 << 20
|
|
}
|
|
return &PathAndCost{
|
|
path: path,
|
|
cost: uint32(cost),
|
|
}
|
|
}
|
|
|
|
type PathAndCost struct {
|
|
path []*model.Router
|
|
cost uint32
|
|
}
|
|
|
|
type InvalidCircuitError struct {
|
|
circuitId string
|
|
}
|
|
|
|
func (err InvalidCircuitError) Error() string {
|
|
return fmt.Sprintf("invalid circuit (%s)", err.circuitId)
|
|
}
|