From b9e87a1597aca2988e451c90df0a2b9f6aba9f5b Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Tue, 30 Nov 2021 01:03:51 -0500 Subject: [PATCH] recieve handlers should never send as they may block forever if the underlying channel closes --- controller/handler_ctrl/create_terminator.go | 6 +- controller/handler_ctrl/fault.go | 91 +++++++++++--------- controller/handler_ctrl/inspect.go | 9 +- controller/handler_ctrl/link.go | 21 +++-- controller/handler_ctrl/metrics.go | 2 +- controller/handler_ctrl/remove_terminator.go | 8 +- controller/handler_ctrl/route_result.go | 4 + controller/handler_ctrl/trace.go | 2 +- controller/handler_ctrl/update_terminator.go | 7 +- 9 files changed, 93 insertions(+), 57 deletions(-) diff --git a/controller/handler_ctrl/create_terminator.go b/controller/handler_ctrl/create_terminator.go index 15dd98b8a..1f3e717a7 100644 --- a/controller/handler_ctrl/create_terminator.go +++ b/controller/handler_ctrl/create_terminator.go @@ -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 diff --git a/controller/handler_ctrl/fault.go b/controller/handler_ctrl/fault.go index 8b8496990..a53118ce6 100644 --- a/controller/handler_ctrl/fault.go +++ b/controller/handler_ctrl/fault.go @@ -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()) } } diff --git a/controller/handler_ctrl/inspect.go b/controller/handler_ctrl/inspect.go index e18945576..4fbb54f51 100644 --- a/controller/handler_ctrl/inspect.go +++ b/controller/handler_ctrl/inspect.go @@ -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" { diff --git a/controller/handler_ctrl/link.go b/controller/handler_ctrl/link.go index 21a8e4994..daf09f623 100644 --- a/controller/handler_ctrl/link.go +++ b/controller/handler_ctrl/link.go @@ -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") } } diff --git a/controller/handler_ctrl/metrics.go b/controller/handler_ctrl/metrics.go index f229f78ea..48de130db 100644 --- a/controller/handler_ctrl/metrics.go +++ b/controller/handler_ctrl/metrics.go @@ -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) } diff --git a/controller/handler_ctrl/remove_terminator.go b/controller/handler_ctrl/remove_terminator.go index fa75f3ac4..fa0e39a49 100644 --- a/controller/handler_ctrl/remove_terminator.go +++ b/controller/handler_ctrl/remove_terminator.go @@ -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()) diff --git a/controller/handler_ctrl/route_result.go b/controller/handler_ctrl/route_result.go index 2b4b85b8b..48130b05f 100644 --- a/controller/handler_ctrl/route_result.go +++ b/controller/handler_ctrl/route_result.go @@ -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] diff --git a/controller/handler_ctrl/trace.go b/controller/handler_ctrl/trace.go index a3ccf7068..d7592cf3c 100644 --- a/controller/handler_ctrl/trace.go +++ b/controller/handler_ctrl/trace.go @@ -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) } diff --git a/controller/handler_ctrl/update_terminator.go b/controller/handler_ctrl/update_terminator.go index 2d997a6bb..d15131335 100644 --- a/controller/handler_ctrl/update_terminator.go +++ b/controller/handler_ctrl/update_terminator.go @@ -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())