mirror of
https://github.com/openziti/ziti.git
synced 2026-10-24 04:32:31 +00:00
Paul Lorenz
444999bc5b
- adds RemoveTerminatorsV2, an asynchronous control-channel removal, so a router no longer holds a request slot open waiting for a synchronous reply; gated on a new capability so routers fall back to the synchronous path against controllers that do not advertise it - paces establish and remove retries with a per-terminator exponential backoff after a controller rejection, so a rejected operation is not re-sent on the next event-loop tick, which is what let a rate-limited removal sustain itself - routes v2 removes to the leader and skips the ordered delete for a terminator the router confirms it created that no longer exists, so a storm of no-op retries cannot consume the command rate limiter - gates that skip on the leader having applied everything committed when it took leadership, since a leader reports itself elected before those applies finish and would otherwise read a committed create as absent and drop its delete - guards the leader term and every write to the caught-up flag with a dedicated lock, so a barrier completing alongside a leadership change cannot record a result for a term that has already ended - scopes v2 removals to the requesting router, so a router can only remove terminators it owns on the asynchronous path as well as the synchronous one - tracks in-flight removal batches on the router, resolving or requeueing them when the response arrives; lost responses self-heal via rate-limiter expiry and the retry scan - waits for an outstanding operation before sending a terminator delete, and checks that before dequeuing, so a delete neither races a create the controller has committed but not applied nor falls out of the queue until the retry scan finds it minutes later - matches a create response to the attempt that produced it, by control channel and reply sequence, since attempts overlap once one overruns the establishment timeout and every attempt carries the same terminator id; only the outstanding attempt's response may confirm the create or report the terminator idle - records the attempt as the sender assigns its sequence, so a response cannot be processed before the request it answers is known, and withdraws an attempt whose send never queued - reports a removal that succeeds past the establishment timeout as congestion rather than success, matching how the establish path classifies its own latency, since both resolve into the same limiter - extracts the transient-error test both remove handlers share, so the v2 path treats a leaderless cluster as busy like v1 does instead of as a permanent failure, and the two cannot drift again - applies the same delete sequencing and response correlation to the router-hosted (ER/T) registry, which cleared the same in-flight marker unguarded and dropped a deferred delete from its queue rather than leaving it there - bounds the v2 response send with a timeout, so a router that stops draining its send queue cannot accumulate blocked responder goroutines on the controller
108 lines
3.9 KiB
Go
108 lines
3.9 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 handler_ctrl
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/michaelquigley/pfxlog"
|
|
"github.com/openziti/channel/v5"
|
|
"github.com/openziti/channel/v5/protobufs"
|
|
"github.com/openziti/ziti/v2/common/pb/ctrl_pb"
|
|
"github.com/openziti/ziti/v2/controller/model"
|
|
"github.com/openziti/ziti/v2/controller/network"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
// removeTerminatorsV2Handler handles RemoveTerminatorsV2Request messages. It performs the same
|
|
// batch removal as removeTerminatorsHandler, but answers asynchronously with a
|
|
// RemoveTerminatorsV2Response instead of a synchronous result, so the router does not hold a
|
|
// request slot open waiting for the reply.
|
|
type removeTerminatorsV2Handler struct {
|
|
baseHandler
|
|
}
|
|
|
|
func newRemoveTerminatorsV2Handler(network *network.Network, router *model.Router) *removeTerminatorsV2Handler {
|
|
return &removeTerminatorsV2Handler{
|
|
baseHandler: baseHandler{
|
|
router: router,
|
|
network: network,
|
|
},
|
|
}
|
|
}
|
|
|
|
func (self *removeTerminatorsV2Handler) ContentType() int32 {
|
|
return int32(ctrl_pb.ContentType_RemoveTerminatorsV2RequestType)
|
|
}
|
|
|
|
func (self *removeTerminatorsV2Handler) HandleReceive(msg *channel.Message, ch channel.Channel) {
|
|
log := pfxlog.ContextLogger(ch.Label())
|
|
|
|
request := &ctrl_pb.RemoveTerminatorsV2Request{}
|
|
if err := proto.Unmarshal(msg.Body, request); err != nil {
|
|
log.WithError(err).Error("failed to unmarshal remove terminators v2 message")
|
|
return
|
|
}
|
|
|
|
go self.handleRemoveTerminators(ch, request)
|
|
}
|
|
|
|
func (self *removeTerminatorsV2Handler) handleRemoveTerminators(ch channel.Channel, request *ctrl_pb.RemoveTerminatorsV2Request) {
|
|
log := pfxlog.ContextLogger(ch.Label()).WithField("requestId", request.RequestId)
|
|
|
|
response := &ctrl_pb.RemoveTerminatorsV2Response{RequestId: request.RequestId}
|
|
|
|
toDelete := self.selectRemovableTerminators(request.TerminatorIds, request.CreateConfirmed)
|
|
|
|
// Everything was either not owned by this router or a confirmed no-op (already gone), so there's
|
|
// nothing to order through raft.
|
|
if len(toDelete) == 0 {
|
|
response.Success = true
|
|
self.sendResponse(ch, response)
|
|
return
|
|
}
|
|
|
|
if err := self.network.Terminator.DeleteBatch(toDelete, self.newChangeContext(ch, "fabric.remove.terminators.batch.v2")); err == nil {
|
|
log.
|
|
WithField("routerId", ch.Id()).
|
|
WithField("terminatorIds", toDelete).
|
|
Info("removed terminators")
|
|
response.Success = true
|
|
} else if wasControllerBusy(err) {
|
|
log.WithError(err).WithField("terminatorIds", toDelete).
|
|
Info("unable to remove terminators, controller busy")
|
|
response.WasRateLimited = true
|
|
response.Msg = err.Error()
|
|
} else {
|
|
log.WithError(err).WithField("terminatorIds", toDelete).
|
|
Error("unable to remove terminators")
|
|
response.Msg = err.Error()
|
|
}
|
|
|
|
self.sendResponse(ch, response)
|
|
}
|
|
|
|
// sendResponse queues the response, giving up if the channel's send queue stays full. A response is
|
|
// only ever logged here, and the router recovers a lost one through rate-limiter expiry and its retry
|
|
// scan, so blocking a per-request goroutine on an undrained queue buys nothing.
|
|
func (self *removeTerminatorsV2Handler) sendResponse(ch channel.Channel, response *ctrl_pb.RemoveTerminatorsV2Response) {
|
|
if err := protobufs.MarshalTyped(response).WithTimeout(5 * time.Second).Send(ch); err != nil {
|
|
pfxlog.ContextLogger(ch.Label()).WithError(err).WithField("requestId", response.RequestId).
|
|
Error("failed to send remove terminators v2 response")
|
|
}
|
|
}
|