ziti/controller/sync_strats/rtx.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 sync_strats

import (
	"fmt"
	"reflect"
	"sync"
	"sync/atomic"
	"time"

	"github.com/michaelquigley/pfxlog"
	"github.com/openziti/channel/v5"
	"github.com/openziti/channel/v5/protobufs"
	"github.com/openziti/foundation/v2/stringz"
	"github.com/openziti/ziti/v2/common"
	"github.com/openziti/ziti/v2/common/ctrlchan"
	"github.com/openziti/ziti/v2/common/eid"
	"github.com/openziti/ziti/v2/common/pb/edge_ctrl_pb"
	"github.com/openziti/ziti/v2/controller/env"
	"github.com/openziti/ziti/v2/controller/model"
	cmap "github.com/orcaman/concurrent-map/v2"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"google.golang.org/protobuf/proto"
)

// RouterSender represents a connection from a router to the controller and asynchronously buffers
// and sends messages to it via Send(). It starts running when created and is stopped with Stop().
// Senders exist for both edge and non-edge (transit) routers; see the edge field for what that
// distinction gates.
type RouterSender struct {
	env.RouterState
	Id               string
	Router           *model.Router
	ae               *env.AppEnv
	send             chan *channel.Message
	closeNotify      chan struct{}
	routerDataModel  *common.RouterDataModelSender
	requestModelSync chan *edge_ctrl_pb.SubscribeToDataModelRequest
	modelChange      chan struct{}
	currentIndex     uint64
	syncRdmUntil     time.Time
	lastIndexSent    uint64
	running          atomic.Bool
	timelineId       string
	subscriptionId   string

	SupportsRouterModel bool

	// edge is true for edge-router senders and false for non-edge (transit) senders, which are
	// created on demand when a transit router subscribes to the data model. It gates the legacy
	// api-session/session broadcasts, which only edge routers consume. Atomic because the connect
	// path can set it on a sender that a subscribe already created and published to the map, where
	// the broadcast fanout is already able to see it. Use setEdge and isEdge.
	edge atomic.Bool

	sync.Mutex
}

func newRouterSender(ae *env.AppEnv, router *model.Router, sendBufferSize int, routerDataModel *common.RouterDataModelSender) *RouterSender {
	rtx := &RouterSender{
		Id:               eid.New(),
		Router:           router,
		ae:               ae,
		send:             make(chan *channel.Message, sendBufferSize),
		requestModelSync: make(chan *edge_ctrl_pb.SubscribeToDataModelRequest, 1),
		modelChange:      make(chan struct{}, 1),
		closeNotify:      make(chan struct{}),
		RouterState:      env.NewLockingRouterStatus(),
		routerDataModel:  routerDataModel,
	}
	rtx.running.Store(true)

	go rtx.run()

	return rtx
}

// setEdge marks whether this sender is for an edge router.
func (rtx *RouterSender) setEdge(edge bool) {
	rtx.edge.Store(edge)
}

// isEdge reports whether this sender is for an edge router.
func (rtx *RouterSender) isEdge() bool {
	return rtx.edge.Load()
}

func (rtx *RouterSender) GetState() env.RouterStateValues {
	if rtx == nil {
		return env.NewRouterStatusValues()
	}

	return rtx.Values()
}

func (rtx *RouterSender) Stop() {
	if rtx.running.CompareAndSwap(true, false) {
		close(rtx.closeNotify)
	}
}

func (rtx *RouterSender) run() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-rtx.closeNotify:
			return
		case msg := <-rtx.send:
			_ = rtx.Router.Control.GetDefaultSender().Send(msg)
		case req := <-rtx.requestModelSync:
			rtx.handleSyncRequest(req)
		case <-rtx.modelChange:
			rtx.handleModelChange()
		case <-ticker.C:
			rtx.sendRouterDataModelIndex()
		}
	}
}

func (rtx *RouterSender) notifyOfModelChange() {
	select {
	case rtx.modelChange <- struct{}{}:
	default:
	}
}

func (rtx *RouterSender) sendRouterDataModelIndex() {
	if rtx.routerDataModel == nil {
		return
	}

	idx := rtx.routerDataModel.CurrentIndex()
	if idx <= rtx.lastIndexSent {
		return
	}

	msg := channel.NewMessage(int32(edge_ctrl_pb.ContentType_CurrentIndexMessageType), nil)
	msg.PutUint64Header(int32(edge_ctrl_pb.Header_RouterDataModelIndex), idx)

	if err := rtx.Router.Control.GetDefaultSender().Send(msg); err != nil {
		logger := pfxlog.Logger().WithField("routerId", rtx.Router.Id)
		logger.WithError(err).Error("could not send current router data model index")
	} else {
		rtx.lastIndexSent = idx
	}
}

func (rtx *RouterSender) subscribe(request *edge_ctrl_pb.SubscribeToDataModelRequest) {
	logger := pfxlog.Logger().WithField("routerId", rtx.Router.Id)
	select {
	case rtx.requestModelSync <- request:
		return
	default:
		// already a request queued, let's try to clear it
		logger.Debug("data model subscription event received, but one already queued, attempting to clear")
	}

	select {
	case <-rtx.requestModelSync:
	default:
	}

	select {
	case rtx.requestModelSync <- request:
		return
	default:
		logger.Debug("data model subscription event received, still can't queue, exiting")
	}
}

func (rtx *RouterSender) handleSyncRequest(req *edge_ctrl_pb.SubscribeToDataModelRequest) {
	if !req.Renew || req.SubscriptionId != rtx.subscriptionId {
		rtx.currentIndex = req.CurrentIndex
	}

	if req.SubscriptionDurationSeconds < 10 {
		req.SubscriptionDurationSeconds = 10
	}

	if req.SubscriptionDurationSeconds > 360 {
		req.SubscriptionDurationSeconds = 360
	}

	timelineChange := rtx.timelineId != req.TimelineId
	rtx.timelineId = req.TimelineId
	rtx.subscriptionId = req.SubscriptionId

	rtx.syncRdmUntil = time.Now().Add(time.Duration(req.SubscriptionDurationSeconds) * time.Second)
	pfxlog.Logger().WithField("routerId", rtx.Router.Id).
		WithField("routerName", rtx.Router.Name).
		WithField("requestedIndex", req.CurrentIndex).
		WithField("currentIndex", rtx.currentIndex).
		WithField("renew", req.Renew).
		WithField("subscriptionDuration", rtx.syncRdmUntil.String()).
		WithField("timelineId", rtx.timelineId).
		WithField("subscriptionId", rtx.subscriptionId).
		Info("data model subscription started")

	if rtx.routerDataModel.CurrentIndex() > rtx.currentIndex || timelineChange {
		rtx.handleModelChange()
	}
}

func (rtx *RouterSender) handleModelChange() {
	if rtx.syncRdmUntil.Before(time.Now()) {
		return
	}

	logger := pfxlog.Logger().WithField("routerId", rtx.Router.Id).
		WithField("currentIndex", rtx.currentIndex).
		WithField("routerTimelineId", rtx.timelineId).
		WithField("ctrlTimelineId", rtx.routerDataModel.GetTimelineId()).
		WithField("subscriptionId", rtx.subscriptionId)

	var events []*edge_ctrl_pb.DataState_ChangeSet
	replayResult := common.ReplayResultFullSyncRequired

	if rtx.currentIndex > 0 && rtx.timelineId == rtx.routerDataModel.GetTimelineId() {
		events, replayResult = rtx.routerDataModel.ReplayFrom(rtx.currentIndex)
	}

	var err error

	logger.Debugf("event retrieval ok? %s, event count: %d for replay to router", replayResult.String(), len(events))

	if replayResult == common.ReplayResultSuccess {
		// we replay from the current index, that way if we're already caught up due to timing issues with notify events
		// we won't get a 'request from future' result.
		if len(events) > 0 {
			events = events[1:]
		}

		for _, curEvent := range events {
			// Always forward the change set, even when the filter removes every
			// entry. Sending an empty change set advances the receiver's index
			// without altering any entities, which keeps the PreviousIndex chain
			// intact without per-router bookkeeping. The alternative (skipping
			// filter-empty change sets and rewriting PreviousIndex on the next
			// non-empty one) is racy under subscription churn and produced gap
			// detection storms.
			toSend := curEvent
			filtered, changed := rtx.filterEventsForRouter(curEvent.Changes, true)
			if changed {
				toSend = &edge_ctrl_pb.DataState_ChangeSet{
					Index:         curEvent.Index,
					PreviousIndex: curEvent.PreviousIndex,
					IsSynthetic:   curEvent.IsSynthetic,
					TimestampId:   curEvent.TimestampId,
					Changes:       filtered,
				}
			}

			var msg *channel.Message
			if msg, err = rtx.marshal(toSend); err == nil {
				err = rtx.Router.Control.GetDefaultSender().Send(msg)
			}

			if err != nil {
				logger.WithError(err).
					WithField("eventIndex", curEvent.Index).
					WithField("eventType", reflect.TypeOf(curEvent).String()).
					WithField("synthetic", curEvent.IsSynthetic).
					Error("could not send data state event")
				break
			}
			logger.
				WithField("eventIndex", curEvent.Index).
				WithField("eventType", reflect.TypeOf(curEvent).String()).
				WithField("synthetic", curEvent.IsSynthetic).
				Debug("data state event sent to router")
			rtx.currentIndex = curEvent.Index
		}
	}

	var fullSync bool

	if err != nil {
		fullSync = true
		logger.Info("could not send events for router sync, attempting full state sync")
	} else if replayResult == common.ReplayResultFullSyncRequired {
		fullSync = true
		logger.Info("required sync events not found in event cache, attempting full state sync")
	} else if rtx.ae.GetCommandDispatcher().IsLeader() && replayResult == common.ReplayResultRequestFromFuture {
		fullSync = true
		logger.Info("index is in future, and this node is the leader, attempting full state sync")
	}

	if fullSync {
		//full sync — already filtered per-router (router-target configs only)
		dataState := rtx.routerDataModel.GetDataStateForRouter(rtx.Router.Id)

		if dataState == nil {
			return
		}

		var msg *channel.Message
		if msg, err = rtx.marshal(dataState); err == nil {
			err = rtx.Router.Control.GetDefaultSender().Send(msg)
		}

		if err != nil {
			logger.WithError(err).Error("failure sending full data state")
		} else {
			rtx.currentIndex = dataState.EndIndex
			rtx.timelineId = dataState.TimelineId
			logger.Infof("router synced data model to index %d with on timeline: %s", rtx.currentIndex, rtx.timelineId)
		}
	}
}

// filterEventsForRouter delegates to the shared RouterDataModelSender filter.
// See RouterDataModelSender.FilterEventsForRouter for behavior.
func (rtx *RouterSender) filterEventsForRouter(events []*edge_ctrl_pb.DataState_Event, synthesizeMissing bool) ([]*edge_ctrl_pb.DataState_Event, bool) {
	return rtx.routerDataModel.FilterEventsForRouter(rtx.Router.Id, events, synthesizeMissing)
}

func (rtx *RouterSender) marshal(message protobufs.TypedMessage) (*channel.Message, error) {
	b, err := proto.Marshal(message)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal %v (%w)", reflect.TypeOf(message), err)
	}
	result := channel.NewMessage(message.GetContentType(), b)
	if rtx.subscriptionId != "" {
		result.PutStringHeader(int32(edge_ctrl_pb.Header_RouterDataModelSubscriptionId), rtx.subscriptionId)
	}
	return result, nil
}

func (rtx *RouterSender) logger() *logrus.Entry {
	return pfxlog.Logger().
		WithField("routerTxId", rtx.Id).
		WithField("routerId", rtx.Router.Id).
		WithField("routerName", rtx.Router.Name).
		WithField("routerFingerprint", stringz.OrEmpty(rtx.Router.Fingerprint)).
		WithField("routerChannelIsOpen", !rtx.Router.Control.IsClosed())
}

func (rtx *RouterSender) Send(msg *channel.Message) error {
	if !rtx.running.Load() {
		rtx.logger().Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
		return errors.Errorf("cannot send to router [%s], rtx'er is stopped", rtx.Router.Id)
	}

	if rtx.Router.Control.IsClosed() {
		rtx.logger().Errorf("cannot send to router [%s], rtx's channel is closed", rtx.Router.Id)
		rtx.Stop()
		return errors.Errorf("cannot send to router [%s], rtx's channel is closed", rtx.Router.Id)
	}

	select {
	case rtx.send <- msg:
	case <-rtx.closeNotify:
	}

	return nil
}

// Map used make working with internal RouterSender easier as sync.Map accepts and returns interface{}
type routerTxMap struct {
	internalMap cmap.ConcurrentMap[string, *RouterSender] //id -> RouterSender
}

// GetOrCreate atomically returns the RouterSender for the given id whose Router.Control
// matches control, creating one via factory if none exists or if the existing sender is
// bound to a stale control channel. It returns the sender and whether it was newly created.
//
// A replaced stale sender is stopped here rather than left to RouterDisconnected, because the
// broker dispatches the old connection's RouterDisconnected asynchronously: on a reconnect the new
// connection's RouterConnected can install its sender before that cleanup runs, which would
// otherwise orphan the old sender's goroutine.
func (m *routerTxMap) GetOrCreate(id string, control ctrlchan.CtrlChannel, factory func() *RouterSender) (rtx *RouterSender, created bool) {
	m.internalMap.Upsert(id, nil, func(exists bool, current *RouterSender, _ *RouterSender) *RouterSender {
		if exists && current != nil && current.Router.Control == control {
			rtx = current
			return current
		}

		if exists && current != nil {
			current.Stop()
		}

		rtx = factory()
		created = true
		return rtx
	})
	return rtx, created
}

func (m *routerTxMap) Get(id string) *RouterSender {
	val, found := m.internalMap.Get(id)
	if !found {
		return nil
	}
	return val
}

func (m *routerTxMap) GetState(id string) env.RouterStateValues {
	rtx := m.Get(id)
	return rtx.GetState()
}

func (m *routerTxMap) Remove(r *model.Router) {
	var rtx *RouterSender
	m.internalMap.RemoveCb(r.Id, func(key string, v *RouterSender, exists bool) bool {
		if !exists {
			return false
		}
		if v.Router == r {
			rtx = v
			return true
		}
		return false
	})

	if rtx != nil {
		rtx.Stop()
	}
}

// Range creates a snapshot of the rtx's to loop over and will execute callback
// with each rtx.
func (m *routerTxMap) Range(callback func(entries *RouterSender)) {
	var routerSenders []*RouterSender
	m.internalMap.IterCb(func(_ string, v *RouterSender) {
		routerSenders = append(routerSenders, v)
	})
	for _, routerSender := range routerSenders {
		callback(routerSender)
	}
}

// RangeEdge is like Range but only visits senders for edge routers, skipping non-edge
// (transit) senders. Used for the legacy api-session/session broadcasts, which only edge
// routers consume.
func (m *routerTxMap) RangeEdge(callback func(entries *RouterSender)) {
	m.Range(func(rtx *RouterSender) {
		if rtx.isEdge() {
			callback(rtx)
		}
	})
}