gsadmin/ziti
1
0
Fork 0
mirror of https://github.com/openziti/ziti.git synced 2026-10-13 23:59:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request openziti/fabric#292 from openziti/fix-potential-deadlock

Browse Source 
recieve handlers should never send as they may block forever if the u…
This commit is contained in:
Paul Lorenz
2021-11-30 13:40:54 -05:00
committed by GitHub
parent f3de039747 b9e87a1597
commit 390aef981b
9 changed files with 93 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controller/handler_ctrl/create_terminator.go
+5 -1
View File
@@ -44,12 +44,16 @@ func (h *createTerminatorHandler) ContentType() int32 {
}
func (h *createTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
log := pfxlog.ContextLogger(ch.Label())
request := &ctrl_pb.CreateTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, request); err != nil {
handler_common.SendFailure(msg, ch, err.Error())
log.WithError(err).Error("failed to unmarshal create terminator message")
return
}
go h.handleCreateTerminator(msg, ch, request)
}
func (h *createTerminatorHandler) handleCreateTerminator(msg *channel2.Message, ch channel2.Channel, request *ctrl_pb.CreateTerminatorRequest) {
if request.Cost > math.MaxUint16 {
handler_common.SendFailure(msg, ch, fmt.Sprintf("invalid cost %v. cost must be between 0 and %v inclusive", request.Cost, math.MaxUint16))
return
controller/handler_ctrl/fault.go
+49 -42
View File
@@ -44,49 +44,56 @@ func (h *faultHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel)
log := pfxlog.ContextLogger(ch.Label())
fault := &ctrl_pb.Fault{}
if err := proto.Unmarshal(msg.Body, fault); err == nil {
switch fault.Subject {
case ctrl_pb.FaultSubject_LinkFault:
linkId := &identity.TokenId{Token: fault.Id}
if err := h.network.LinkConnected(linkId, false); err == nil {
if link, found := h.network.GetLink(linkId); found {
h.network.LinkChanged(link)
}
log.Infof("link fault [l/%s]", linkId.Token)
if err := proto.Unmarshal(msg.Body, fault); err != nil {
log.WithError(err).Error("failed to unmarshal fault message")
return
}
go h.handleFault(msg, ch, fault)
}
func (h *faultHandler) handleFault(msg *channel2.Message, ch channel2.Channel, fault *ctrl_pb.Fault) {
log := pfxlog.ContextLogger(ch.Label())
switch fault.Subject {
case ctrl_pb.FaultSubject_LinkFault:
linkId := &identity.TokenId{Token: fault.Id}
if err := h.network.LinkConnected(linkId, false); err == nil {
if link, found := h.network.GetLink(linkId); found {
h.network.LinkChanged(link)
}
case ctrl_pb.FaultSubject_IngressFault:
if err := h.network.RemoveCircuit(fault.Id, false); err != nil {
invalidCircuitErr := network.InvalidCircuitError{}
if errors.As(err, &invalidCircuitErr) {
log.Debugf("error handling ingress fault (%s)", err)
} else {
log.Errorf("error handling ingress fault (%s)", err)
}
} else {
log.Debugf("handled ingress fault for (%s)", fault.Id)
}
case ctrl_pb.FaultSubject_EgressFault:
if err := h.network.RemoveCircuit(fault.Id, false); err != nil {
invalidCircuitErr := network.InvalidCircuitError{}
if errors.As(err, &invalidCircuitErr) {
log.Debugf("error handling egress fault (%s)", err)
} else {
log.Errorf("error handling egress fault (%s)", err)
}
} else {
log.Debugf("handled egress fault for (%s)", fault.Id)
}
case ctrl_pb.FaultSubject_ForwardFault:
circuitIds := strings.Split(fault.Id, " ")
h.network.ReportForwardingFaults(&network.ForwardingFaultReport{R: h.r, CircuitIds: circuitIds})
default:
log.Errorf("unexpected subject (%s)", fault.Subject.String())
log.Infof("link fault [l/%s]", linkId.Token)
}
} else {
log.Errorf("unexpected error (%s)", err)
case ctrl_pb.FaultSubject_IngressFault:
if err := h.network.RemoveCircuit(fault.Id, false); err != nil {
invalidCircuitErr := network.InvalidCircuitError{}
if errors.As(err, &invalidCircuitErr) {
log.Debugf("error handling ingress fault (%s)", err)
} else {
log.Errorf("error handling ingress fault (%s)", err)
}
} else {
log.Debugf("handled ingress fault for (%s)", fault.Id)
}
case ctrl_pb.FaultSubject_EgressFault:
if err := h.network.RemoveCircuit(fault.Id, false); err != nil {
invalidCircuitErr := network.InvalidCircuitError{}
if errors.As(err, &invalidCircuitErr) {
log.Debugf("error handling egress fault (%s)", err)
} else {
log.Errorf("error handling egress fault (%s)", err)
}
} else {
log.Debugf("handled egress fault for (%s)", fault.Id)
}
case ctrl_pb.FaultSubject_ForwardFault:
circuitIds := strings.Split(fault.Id, " ")
h.network.ReportForwardingFaults(&network.ForwardingFaultReport{R: h.r, CircuitIds: circuitIds})
default:
log.Errorf("unexpected subject (%s)", fault.Subject.String())
}
}
controller/handler_ctrl/inspect.go
+6 -3
View File
@@ -17,7 +17,6 @@
package handler_ctrl
import (
"fmt"
"github.com/golang/protobuf/proto"
"github.com/michaelquigley/pfxlog"
"github.com/openziti/fabric/controller/network"
@@ -42,10 +41,14 @@ func (h *inspectHandler) HandleReceive(request *channel2.Message, ch channel2.Ch
inspectRequest := &ctrl_pb.InspectRequest{}
if err := proto.Unmarshal(request.Body, inspectRequest); err != nil {
log.Errorf("unexpected error (%s)", err)
h.respondWithError(ch, request, fmt.Errorf("failed to decode request: (%v)", err).Error())
log.WithError(err).Error("failed to unmarshal inspect message")
return
}
go h.handleInspect(request, ch, inspectRequest)
}
func (h *inspectHandler) handleInspect(request *channel2.Message, ch channel2.Channel, inspectRequest *ctrl_pb.InspectRequest) {
response := &ctrl_pb.InspectResponse{Success: true}
for _, value := range inspectRequest.RequestedValues {
if value == "capability" {
controller/handler_ctrl/link.go
+14 -7
View File
@@ -42,13 +42,20 @@ func (h *linkHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel)
log := pfxlog.ContextLogger(ch.Label())
link := &ctrl_pb.Link{}
if err := proto.Unmarshal(msg.Body, link); err == nil {
if err := h.network.LinkConnected(&identity.TokenId{Token: link.Id}, true); err == nil {
log.Infof("link connected [l/%s]", link.Id)
} else {
log.Errorf("unexpected error (%s)", err)
}
if err := proto.Unmarshal(msg.Body, link); err != nil {
log.WithError(err).Error("failed to unmarshal link message")
return
}
go h.HandleLink(msg, ch, link)
}
func (h *linkHandler) HandleLink(msg *channel2.Message, ch channel2.Channel, link *ctrl_pb.Link) {
log := pfxlog.ContextLogger(ch.Label())
if err := h.network.LinkConnected(&identity.TokenId{Token: link.Id}, true); err == nil {
log.Infof("link connected [l/%s]", link.Id)
} else {
log.Errorf("unexpected error (%s)", err)
log.WithError(err).Error("unexpected error marking link connected")
}
}
controller/handler_ctrl/metrics.go
+1 -1
View File
@@ -40,7 +40,7 @@ func (h *metricsHandler) ContentType() int32 {
func (h *metricsHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
metricsMsg := &metrics_pb.MetricsMessage{}
if err := proto.Unmarshal(msg.Body, metricsMsg); err == nil {
h.AcceptMetrics(metricsMsg)
go h.AcceptMetrics(metricsMsg)
} else {
pfxlog.ContextLogger(ch.Label()).Errorf("unexpected error (%s)", err)
}
controller/handler_ctrl/remove_terminator.go
+7 -1
View File
@@ -42,10 +42,16 @@ func (h *removeTerminatorHandler) HandleReceive(msg *channel2.Message, ch channe
request := &ctrl_pb.RemoveTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, request); err != nil {
handler_common.SendFailure(msg, ch, err.Error())
log.WithError(err).Error("failed to unmarshal remove terminator message")
return
}
go h.handleRemoveTerminator(msg, ch, request)
}
func (h *removeTerminatorHandler) handleRemoveTerminator(msg *channel2.Message, ch channel2.Channel, request *ctrl_pb.RemoveTerminatorRequest) {
log := pfxlog.ContextLogger(ch.Label())
_, err := h.network.Terminators.Read(request.TerminatorId)
if err != nil {
handler_common.SendFailure(msg, ch, err.Error())
controller/handler_ctrl/route_result.go
+4
View File
@@ -45,6 +45,10 @@ func (self *routeResultHandler) ContentType() int32 {
}
func (self *routeResultHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
go self.handleRouteResult(msg)
}
func (self *routeResultHandler) handleRouteResult(msg *channel2.Message) {
if v, found := msg.Headers[ctrl_msg.RouteResultAttemptHeader]; found {
_, success := msg.Headers[ctrl_msg.RouteResultSuccessHeader]
rerrv, _ := msg.Headers[ctrl_msg.RouteResultErrorHeader]
controller/handler_ctrl/trace.go
+1 -1
View File
@@ -24,7 +24,7 @@ func (*traceHandler) ContentType() int32 {
func (handler *traceHandler) HandleReceive(msg *channel2.Message, _ channel2.Channel) {
event := &trace_pb.ChannelMessage{}
if err := proto.Unmarshal(msg.Body, event); err == nil {
handler.dispatcher.Accept(event)
go handler.dispatcher.Accept(event)
} else {
pfxlog.Logger().Errorf("unexpected error decoding trace message (%s)", err)
}
controller/handler_ctrl/update_terminator.go
+6 -1
View File
@@ -26,6 +26,7 @@ import (
"github.com/openziti/fabric/pb/ctrl_pb"
"github.com/openziti/foundation/channel2"
"github.com/openziti/foundation/storage/boltz"
log "github.com/sirupsen/logrus"
"math"
)
@@ -44,10 +45,14 @@ func (h *updateTerminatorHandler) ContentType() int32 {
func (h *updateTerminatorHandler) HandleReceive(msg *channel2.Message, ch channel2.Channel) {
request := &ctrl_pb.UpdateTerminatorRequest{}
if err := proto.Unmarshal(msg.Body, request); err != nil {
handler_common.SendFailure(msg, ch, err.Error())
log.WithError(err).Error("failed to unmarshal update terminator message")
return
}
go h.handleUpdateTerminator(msg, ch, request)
}
func (h *updateTerminatorHandler) handleUpdateTerminator(msg *channel2.Message, ch channel2.Channel, request *ctrl_pb.UpdateTerminatorRequest) {
terminator, err := h.network.Terminators.Read(request.TerminatorId)
if err != nil {
handler_common.SendFailure(msg, ch, err.Error())