diff --git a/controller/api_impl/circuit_api_model.go b/controller/api_impl/circuit_api_model.go index 769b4ea17..5d6a3862a 100644 --- a/controller/api_impl/circuit_api_model.go +++ b/controller/api_impl/circuit_api_model.go @@ -17,6 +17,7 @@ package api_impl import ( + "github.com/go-openapi/strfmt" "github.com/openziti/fabric/controller/api" "github.com/openziti/fabric/controller/network" @@ -52,12 +53,15 @@ func MapCircuitToRestModel(_ *network.Network, _ api.RequestContext, circuit *ne for _, link := range circuit.Path.Links { path.Links = append(path.Links, ToEntityRef(link.Id, link, LinkLinkFactory)) } + + createdAt := strfmt.DateTime(circuit.CreatedAt) ret := &rest_model.CircuitDetail{ ID: &circuit.Id, ClientID: circuit.ClientId, Path: path, Service: ToEntityRef(circuit.Service.Name, circuit.Service, ServiceLinkFactory), Terminator: ToEntityRef(circuit.Terminator.GetId(), circuit.Terminator, TerminatorLinkFactory), + CreatedAt: &createdAt, } return ret, nil diff --git a/controller/network/circuit.go b/controller/network/circuit.go index 6c55b257f..ca1d4498f 100644 --- a/controller/network/circuit.go +++ b/controller/network/circuit.go @@ -21,6 +21,7 @@ import ( "github.com/openziti/fabric/controller/xt" "github.com/openziti/foundation/v2/concurrenz" "github.com/orcaman/concurrent-map/v2" + "time" ) type Circuit struct { @@ -31,6 +32,7 @@ type Circuit struct { Path *Path Rerouting concurrenz.AtomicBoolean PeerData xt.PeerData + CreatedAt time.Time } func (self *Circuit) cost() int64 { @@ -68,29 +70,29 @@ func newCircuitController() *circuitController { } } -func (c *circuitController) nextCircuitId() (string, error) { - return c.idGenerator.NextAlphaNumericPrefixedId() +func (self *circuitController) nextCircuitId() (string, error) { + return self.idGenerator.NextAlphaNumericPrefixedId() } -func (c *circuitController) add(sn *Circuit) { - c.circuits.Set(sn.Id, sn) +func (self *circuitController) add(circuit *Circuit) { + self.circuits.Set(circuit.Id, circuit) } -func (c *circuitController) get(id string) (*Circuit, bool) { - if circuit, found := c.circuits.Get(id); found { +func (self *circuitController) get(id string) (*Circuit, bool) { + if circuit, found := self.circuits.Get(id); found { return circuit, true } return nil, false } -func (c *circuitController) all() []*Circuit { +func (self *circuitController) all() []*Circuit { circuits := make([]*Circuit, 0) - for tuple := range c.circuits.IterBuffered() { + for tuple := range self.circuits.IterBuffered() { circuits = append(circuits, tuple.Val) } return circuits } -func (c *circuitController) remove(sn *Circuit) { - c.circuits.Remove(sn.Id) +func (self *circuitController) remove(circuit *Circuit) { + self.circuits.Remove(circuit.Id) } diff --git a/controller/network/network.go b/controller/network/network.go index 32ba4343a..b5fd853b9 100644 --- a/controller/network/network.go +++ b/controller/network/network.go @@ -524,6 +524,7 @@ func (network *Network) CreateCircuit(srcR *Router, clientId *identity.TokenId, Path: path, Terminator: terminator, PeerData: peerData, + CreatedAt: time.Now(), } network.circuitController.add(circuit) creationTimespan := time.Since(startTime) @@ -870,6 +871,24 @@ func (network *Network) rerouteLink(l *Link, deadline time.Time) error { return nil } +func (network *Network) rerouteCircuitWithTries(circuit *Circuit, retries int) { + log := pfxlog.Logger().WithField("circuitId", circuit.Id) + + for i := 0; i < retries; i++ { + deadline := time.Now().Add(DefaultNetworkOptionsRouteTimeout) + err := network.rerouteCircuit(circuit, deadline) + if err == nil { + return + } + + 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") + } +} + func (network *Network) rerouteCircuit(circuit *Circuit, deadline time.Time) error { log := pfxlog.Logger().WithField("circuitId", circuit.Id) if circuit.Rerouting.CompareAndSwap(false, true) { @@ -901,28 +920,30 @@ func (network *Network) rerouteCircuit(circuit *Circuit, deadline time.Time) err } } -func (network *Network) smartReroute(s *Circuit, cq *Path, deadline time.Time) error { - if s.Rerouting.CompareAndSwap(false, true) { - defer s.Rerouting.Set(false) +func (network *Network) smartReroute(circuit *Circuit, cq *Path, deadline time.Time) bool { + retry := false + log := pfxlog.Logger().WithField("circuitId", circuit.Id) + if circuit.Rerouting.CompareAndSwap(false, true) { + defer circuit.Rerouting.Set(false) - s.Path = cq + circuit.Path = cq - rms := cq.CreateRouteMessages(SmartRerouteAttempt, s.Id, s.Terminator, deadline) + rms := cq.CreateRouteMessages(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 { - logrus.Errorf("error sending route to [r/%s] (%s)", cq.Nodes[i].Id, err) + retry = true + log.WithField("routerId", cq.Nodes[i].Id).WithError(err).Error("error sending smart route update to router") + break } } - logrus.Debugf("rerouted circuit [s/%s]", s.Id) - network.CircuitEvent(CircuitUpdated, s, nil, nil) - return nil - - } else { - logrus.Infof("not rerouting [s/%s], already in progress", s.Id) - return nil + if !retry { + logrus.Debug("rerouted circuit") + network.CircuitEvent(CircuitUpdated, circuit, nil, nil) + } } + return retry } func (network *Network) AcceptMetrics(metrics *metrics_pb.MetricsMessage) { diff --git a/controller/network/smart.go b/controller/network/smart.go index b863c3c5d..1be7c101c 100644 --- a/controller/network/smart.go +++ b/controller/network/smart.go @@ -85,8 +85,8 @@ func (network *Network) smart() { * Reroute. */ for _, circuit := range candidates { - if err := network.smartReroute(circuit, newPaths[circuit], time.Now().Add(DefaultNetworkOptionsRouteTimeout)); err != nil { - log.Errorf("error rerouting [s/%s] (%s)", circuit.Id, err) + if retry := network.smartReroute(circuit, newPaths[circuit], time.Now().Add(DefaultNetworkOptionsRouteTimeout)); retry { + go network.rerouteCircuitWithTries(circuit, DefaultNetworkOptionsCreateCircuitRetries) } } /* */ diff --git a/rest_model/circuit_detail.go b/rest_model/circuit_detail.go index 5f3a263e8..226ce561a 100644 --- a/rest_model/circuit_detail.go +++ b/rest_model/circuit_detail.go @@ -47,6 +47,11 @@ type CircuitDetail struct { // client Id ClientID string `json:"clientId,omitempty"` + // created at + // Required: true + // Format: date-time + CreatedAt *strfmt.DateTime `json:"createdAt"` + // id // Required: true ID *string `json:"id"` @@ -68,6 +73,10 @@ type CircuitDetail struct { func (m *CircuitDetail) Validate(formats strfmt.Registry) error { var res []error + if err := m.validateCreatedAt(formats); err != nil { + res = append(res, err) + } + if err := m.validateID(formats); err != nil { res = append(res, err) } @@ -90,6 +99,19 @@ func (m *CircuitDetail) Validate(formats strfmt.Registry) error { return nil } +func (m *CircuitDetail) validateCreatedAt(formats strfmt.Registry) error { + + if err := validate.Required("createdAt", "body", m.CreatedAt); err != nil { + return err + } + + if err := validate.FormatOf("createdAt", "body", "date-time", m.CreatedAt.String(), formats); err != nil { + return err + } + + return nil +} + func (m *CircuitDetail) validateID(formats strfmt.Registry) error { if err := validate.Required("id", "body", m.ID); err != nil { diff --git a/rest_server/embedded_spec.go b/rest_server/embedded_spec.go index ca4f07de3..32d24b18d 100644 --- a/rest_server/embedded_spec.go +++ b/rest_server/embedded_spec.go @@ -1081,12 +1081,17 @@ func init() { "id", "service", "terminator", - "path" + "path", + "createdAt" ], "properties": { "clientId": { "type": "string" }, + "createdAt": { + "type": "string", + "format": "date-time" + }, "id": { "type": "string" }, @@ -5250,12 +5255,17 @@ func init() { "id", "service", "terminator", - "path" + "path", + "createdAt" ], "properties": { "clientId": { "type": "string" }, + "createdAt": { + "type": "string", + "format": "date-time" + }, "id": { "type": "string" }, diff --git a/router/xgress/link_receive_buffer.go b/router/xgress/link_receive_buffer.go index 3c10f9660..27fc040a7 100644 --- a/router/xgress/link_receive_buffer.go +++ b/router/xgress/link_receive_buffer.go @@ -107,8 +107,9 @@ func (buffer *LinkReceiveBuffer) Inspect() *inspect.XgressRecvBufferDetail { func (buffer *LinkReceiveBuffer) inspectComplete() *inspect.XgressRecvBufferDetail { nextPayload := "none" - if head := buffer.PeekHead(); head != nil { - nextPayload = fmt.Sprintf("%v", head.Sequence) + if head := buffer.tree.LeftValue(); head != nil { + payload := head.(*Payload) + nextPayload = fmt.Sprintf("%v", payload.Sequence) } return &inspect.XgressRecvBufferDetail{ diff --git a/router/xgress/messages.go b/router/xgress/messages.go index 7fd9b0ecf..ad4ff886b 100644 --- a/router/xgress/messages.go +++ b/router/xgress/messages.go @@ -66,7 +66,7 @@ type PayloadFlag uint32 const ( PayloadFlagCircuitEnd PayloadFlag = 1 - PayloadFlagEgress PayloadFlag = 2 + PayloadFlagOriginator PayloadFlag = 2 PayloadFlagCircuitStart PayloadFlag = 4 ) @@ -86,7 +86,7 @@ func (header *Header) GetFlags() string { } func (header *Header) GetOriginator() Originator { - if isPayloadFlagSet(header.Flags, PayloadFlagEgress) { + if isPayloadFlagSet(header.Flags, PayloadFlagOriginator) { return Terminator } return Initiator @@ -267,9 +267,9 @@ func (payload *Payload) IsCircuitStartFlagSet() bool { func SetOriginatorFlag(flags uint32, originator Originator) uint32 { if originator == Initiator { - return ^uint32(PayloadFlagEgress) & flags + return ^uint32(PayloadFlagOriginator) & flags } - return uint32(PayloadFlagEgress) | flags + return uint32(PayloadFlagOriginator) | flags } func (payload *Payload) GetLoggerFields() logrus.Fields { diff --git a/router/xgress/messages_test.go b/router/xgress/messages_test.go index ef350bb63..bb19725f0 100644 --- a/router/xgress/messages_test.go +++ b/router/xgress/messages_test.go @@ -55,14 +55,14 @@ func TestSetOriginatorFlag(t *testing.T) { flags: 0, originator: Terminator, }, - want: uint32(PayloadFlagEgress), + want: uint32(PayloadFlagOriginator), }, {name: "set end of circuit to egress", args: args{ flags: uint32(PayloadFlagCircuitEnd), originator: Terminator, }, - want: uint32(PayloadFlagCircuitEnd) | uint32(PayloadFlagEgress), + want: uint32(PayloadFlagCircuitEnd) | uint32(PayloadFlagOriginator), }, } for _, tt := range tests { diff --git a/router/xgress/ordering_test.go b/router/xgress/ordering_test.go index f815627f0..470ea8565 100644 --- a/router/xgress/ordering_test.go +++ b/router/xgress/ordering_test.go @@ -90,7 +90,7 @@ func Test_Ordering(t *testing.T) { payload := &Payload{ Header: Header{ CircuitId: "test", - Flags: 0, + Flags: SetOriginatorFlag(0, Terminator), RecvBufferSize: 16000, RTT: 0, }, diff --git a/router/xgress/xgress.go b/router/xgress/xgress.go index fb36ee552..0fd68f5d0 100644 --- a/router/xgress/xgress.go +++ b/router/xgress/xgress.go @@ -564,7 +564,10 @@ func (self *Xgress) nextReceiveSequence() int32 { func (self *Xgress) PayloadReceived(payload *Payload) { log := pfxlog.ContextLogger(self.Label()).WithFields(payload.GetLoggerFields()) log.Debug("payload received") - if self.linkRxBuffer.ReceiveUnordered(payload, self.Options.RxBufferSize) { + if self.originator == payload.GetOriginator() { + // a payload sent from this xgress has arrived back at this xgress, instead of the other end + log.Warn("ouroboros (circuit cycle) detected, dropping payload") + } else if self.linkRxBuffer.ReceiveUnordered(payload, self.Options.RxBufferSize) { log.Debug("ready to acknowledge") ack := NewAcknowledgement(self.circuitId, self.originator) diff --git a/specs/swagger.yml b/specs/swagger.yml index e27d6a4f8..eb87539b8 100644 --- a/specs/swagger.yml +++ b/specs/swagger.yml @@ -1580,6 +1580,7 @@ definitions: - service - terminator - path + - createdAt properties: id: type: string @@ -1589,6 +1590,9 @@ definitions: $ref: '#/definitions/entityRef' terminator: $ref: '#/definitions/entityRef' + createdAt: + type: string + format: date-time path: type: object properties: