ziti/controller/handler_ctrl/dial_events.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 (
	"container/heap"
	"time"

	"github.com/michaelquigley/pfxlog"
	"github.com/openziti/ziti/v2/common/inspect"
	"github.com/openziti/ziti/v2/controller/model"
)

type dialEvent interface {
	handle(d *CtrlDialer)
}

type routerDisconnectedEvent struct {
	routerId string
	router   *model.Router
}

func (self *routerDisconnectedEvent) handle(d *CtrlDialer) {
	log := pfxlog.Logger().WithField("routerId", self.routerId)

	state, exists := d.states[self.routerId]
	if !exists {
		addresses := d.getRouterIdEndpointsNeedingDial(self.routerId, self.router)
		if len(addresses) == 0 {
			return
		}
		state = &routerDialState{
			routerId:  self.routerId,
			addresses: addresses,
		}
		d.states[self.routerId] = state
	}

	state.connectionLost(d.config)
	heap.Push(&d.retryQueue, state)

	log.WithField("nextDial", time.Until(state.nextDial).Round(time.Millisecond)).
		Debug("router disconnected, queued for redial")
}

type routerUpdatedEvent struct {
	routerId string
}

func (self *routerUpdatedEvent) handle(d *CtrlDialer) {
	state, exists := d.states[self.routerId]
	if exists && state.status == statusConnected {
		return
	}

	addresses := d.getRouterIdEndpointsNeedingDial(self.routerId, nil)
	if len(addresses) == 0 {
		if exists {
			delete(d.states, self.routerId)
		}
		return
	}

	if !exists {
		state = &routerDialState{
			routerId:  self.routerId,
			addresses: addresses,
			nextDial:  time.Now(),
		}
		d.states[self.routerId] = state
		heap.Push(&d.retryQueue, state)
	} else {
		state.addresses = addresses
		if state.addrIndex >= len(state.addresses) {
			state.addrIndex = state.addrIndex % len(state.addresses)
		}
	}
}

type routerConnectedEvent struct {
	routerId string
}

func (self *routerConnectedEvent) handle(d *CtrlDialer) {
	if state, exists := d.states[self.routerId]; exists {
		state.status = statusConnected
		state.connectedAt = time.Now()
		state.retryDelay = 0
		state.dialAttempts = 0
	}
	// don't delete from states map — let scan/evaluate skip connected routers naturally
}

type routerDeletedEvent struct {
	routerId string
}

func (self *routerDeletedEvent) handle(d *CtrlDialer) {
	delete(d.states, self.routerId)
}

type dialResultEvent struct {
	routerId string
	err      error
}

func (self *dialResultEvent) handle(d *CtrlDialer) {
	state, exists := d.states[self.routerId]
	if !exists {
		return
	}

	log := pfxlog.Logger().WithField("routerId", self.routerId)

	if self.err != nil {
		state.dialFailed(d.config)
		log.WithError(self.err).
			WithField("retryDelay", state.retryDelay).
			WithField("nextDial", time.Until(state.nextDial).Round(time.Millisecond)).
			Warn("dial attempt failed, will retry with backoff")
		heap.Push(&d.retryQueue, state)
		return
	}

	state.dialSucceeded()
	// schedule a re-check after FastFailureWindow to detect fast failures
	state.nextDial = time.Now().Add(d.config.FastFailureWindow)
	heap.Push(&d.retryQueue, state)

	log.Info("successfully connected to router ctrl channel")
}

type validateDialStateSnapshot struct {
	routerId string
	status   routerDialStatus
}

type validateDialStatesEvent struct {
	result chan map[string]*validateDialStateSnapshot
}

func (self *validateDialStatesEvent) handle(d *CtrlDialer) {
	states := make(map[string]*validateDialStateSnapshot, len(d.states))
	for k, v := range d.states {
		states[k] = &validateDialStateSnapshot{
			routerId: v.routerId,
			status:   v.status,
		}
	}
	self.result <- states
}

type inspectDialStatesEvent struct {
	result chan *inspect.CtrlDialerInspectResult
}

func (self *inspectDialStatesEvent) handle(d *CtrlDialer) {
	result := &inspect.CtrlDialerInspectResult{
		Enabled: true,
		Config: inspect.CtrlDialerConfigDetail{
			Groups:             d.config.Groups,
			DialDelay:          d.config.DialDelay.String(),
			MinRetryInterval:   d.config.MinRetryInterval.String(),
			MaxRetryInterval:   d.config.MaxRetryInterval.String(),
			RetryBackoffFactor: d.config.RetryBackoffFactor,
			FastFailureWindow:  d.config.FastFailureWindow.String(),
			QueueSize:          d.config.QueueSize,
			MaxWorkers:         d.config.MaxWorkers,
		},
	}

	for _, state := range d.states {
		result.Routers = append(result.Routers, &inspect.CtrlDialerRouterDial{
			RouterId:       state.routerId,
			Addresses:      state.addresses,
			CurrentAddress: state.currentAddress(),
			Status:         state.status.String(),
			DialAttempts:   state.dialAttempts,
			RetryDelay:     state.retryDelay.String(),
			NextDial:       time.Until(state.nextDial).Round(time.Millisecond).String(),
		})
	}

	self.result <- result
}