ziti/controller/model/controller_manager.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 model

import (
	"crypto/x509"
	"time"

	"github.com/michaelquigley/pfxlog"
	nfpem "github.com/openziti/foundation/v2/pem"
	"github.com/openziti/storage/boltz"
	"github.com/openziti/ziti/common/pb/edge_cmd_pb"
	"github.com/openziti/ziti/controller/change"
	"github.com/openziti/ziti/controller/command"
	"github.com/openziti/ziti/controller/db"
	"github.com/openziti/ziti/controller/event"
	"github.com/openziti/ziti/controller/fields"
	"github.com/openziti/ziti/controller/models"
	"google.golang.org/protobuf/proto"
)

func NewControllerManager(env Env) *ControllerManager {
	manager := &ControllerManager{
		baseEntityManager: newBaseEntityManager[*Controller, *db.Controller](env, env.GetStores().Controller),
	}
	manager.impl = manager

	RegisterManagerDecoder[*Controller](env, manager)

	return manager
}

type ControllerManager struct {
	baseEntityManager[*Controller, *db.Controller]
}

func (self *ControllerManager) NewModelEntity() *Controller {
	return &Controller{}
}

func (self *ControllerManager) Create(entity *Controller, ctx *change.Context) error {
	return DispatchCreate[*Controller](self, entity, ctx)
}

func (self *ControllerManager) ApplyCreate(cmd *command.CreateEntityCommand[*Controller], ctx boltz.MutateContext) error {
	_, err := self.createEntity(cmd.Entity, ctx)
	return err
}

func (self *ControllerManager) Update(entity *Controller, checker fields.UpdatedFields, ctx *change.Context) error {
	return DispatchUpdate[*Controller](self, entity, checker, ctx)
}

func (self *ControllerManager) ApplyUpdate(cmd *command.UpdateEntityCommand[*Controller], ctx boltz.MutateContext) error {
	return self.updateEntity(cmd.Entity, cmd.UpdatedFields, ctx)
}

func (self *ControllerManager) Read(id string) (*Controller, error) {
	modelEntity := &Controller{}
	if err := self.readEntity(id, modelEntity); err != nil {
		return nil, err
	}
	return modelEntity, nil
}

func (self *ControllerManager) ReadByName(name string) (*Controller, error) {
	modelEntity := &Controller{}
	nameIndex := self.env.GetStores().Controller.GetNameIndex()
	if err := self.readEntityWithIndex("name", []byte(name), nameIndex, modelEntity); err != nil {
		return nil, err
	}
	return modelEntity, nil
}

func (self *ControllerManager) Marshall(entity *Controller) ([]byte, error) {
	msg := &edge_cmd_pb.Controller{
		Id:           entity.Id,
		Name:         entity.Name,
		Address:      entity.CtrlAddress,
		CertPem:      entity.CertPem,
		Fingerprint:  entity.Fingerprint,
		IsOnline:     entity.IsOnline,
		LastJoinedAt: timePtrToPb(&entity.LastJoinedAt),
		ApiAddresses: map[string]*edge_cmd_pb.ApiAddressList{},
	}

	for apiKey, instances := range entity.ApiAddresses {
		msg.ApiAddresses[apiKey] = &edge_cmd_pb.ApiAddressList{}
		for _, instance := range instances {
			msg.ApiAddresses[apiKey].Addresses = append(msg.ApiAddresses[apiKey].Addresses, &edge_cmd_pb.ApiAddress{
				Url:     instance.Url,
				Version: instance.Version,
			})
		}
	}

	return proto.Marshal(msg)
}

func (self *ControllerManager) Unmarshall(bytes []byte) (*Controller, error) {
	msg := &edge_cmd_pb.Controller{}

	if err := proto.Unmarshal(bytes, msg); err != nil {
		return nil, err
	}

	lastJoinedAt := time.Time{}
	if msg.LastJoinedAt != nil {
		lastJoinedAt = *pbTimeToTimePtr(msg.LastJoinedAt)
	}

	controller := &Controller{
		BaseEntity: models.BaseEntity{
			Id:   msg.Id,
			Tags: edge_cmd_pb.DecodeTags(msg.Tags),
		},
		Name:         msg.Name,
		CtrlAddress:  msg.Address,
		CertPem:      msg.CertPem,
		Fingerprint:  msg.Fingerprint,
		IsOnline:     msg.IsOnline,
		LastJoinedAt: lastJoinedAt,
		ApiAddresses: map[string][]ApiAddress{},
	}

	for apiKey, instanceList := range msg.ApiAddresses {
		controller.ApiAddresses[apiKey] = nil
		if instanceList != nil {
			for _, instance := range instanceList.Addresses {
				controller.ApiAddresses[apiKey] = append(controller.ApiAddresses[apiKey], ApiAddress{
					Url:     instance.Url,
					Version: instance.Version,
				})
			}
		}
	}

	return controller, nil
}

func (self *ControllerManager) getCurrentAsClusterPeer() *event.ClusterPeer {
	addr, id, version := self.env.GetRaftInfo()
	clientApiCert := self.env.GetRootTlsJwtSigner()
	var leaderCerts []*x509.Certificate

	for _, certBytes := range clientApiCert.TlsCerts.Certificate {
		if cert, err := x509.ParseCertificate(certBytes); err == nil {
			leaderCerts = append(leaderCerts, cert)
		}
	}

	apiAddresses, _ := self.env.GetApiAddresses()

	return &event.ClusterPeer{
		Id:           id,
		Addr:         addr,
		Version:      version,
		ServerCert:   leaderCerts,
		ApiAddresses: apiAddresses,
	}
}

func (self *ControllerManager) UpdateControllerState(peers []*event.ClusterPeer, peerConnectedEvent bool) {
	controllers := map[string]*Controller{}

	result, err := self.BaseList("true limit none")
	if err != nil {
		pfxlog.Logger().WithError(err).Error("failed to list controllers")
		return
	} else {
		for _, ctrl := range result.Entities {
			controllers[ctrl.Id] = ctrl
		}
	}

	changeCtx := change.New()
	if peerConnectedEvent {
		changeCtx.SetSourceType("raft.peers.connected").
			SetChangeAuthorType(change.AuthorTypeController)
	} else {
		changeCtx.SetSourceType("raft.leadership.gained").
			SetChangeAuthorType(change.AuthorTypeController)
	}

	selfAsPeer := self.getCurrentAsClusterPeer()
	peerFingerprints := ""
	for _, peer := range peers {
		if len(peer.ServerCert) > 0 {
			fingerprint := nfpem.FingerprintFromCertificate(peer.ServerCert[0])

			if peerFingerprints == "" {
				peerFingerprints = fingerprint
			} else {
				peerFingerprints = peerFingerprints + ", " + fingerprint
			}
		}
	}

	pfxlog.Logger().Infof("acting as leader, updating controllers from peers, connectEvt? %v, self: %s, peer count: %d, peers: %s",
		peerConnectedEvent, nfpem.FingerprintFromCertificate(selfAsPeer.ServerCert[0]), len(peers), peerFingerprints)

	if !peerConnectedEvent {
		// add this controller as a "peer" when leadership is gained
		peers = append(peers, selfAsPeer)
	}

	for _, peer := range peers {
		// Use our locally built peer instance to represent ourselves in the list
		if peer.Id == selfAsPeer.Id && peer != selfAsPeer {
			continue
		}

		if len(peer.ServerCert) < 1 {
			pfxlog.Logger().Errorf("peer %s has no certificate", peer.Id)
			continue
		}

		newController := &Controller{
			BaseEntity: models.BaseEntity{
				Id: peer.Id,
			},
			Name:         peer.ServerCert[0].Subject.CommonName,
			CertPem:      nfpem.EncodeToString(peer.ServerCert[0]),
			Fingerprint:  nfpem.FingerprintFromCertificate(peer.ServerCert[0]),
			CtrlAddress:  peer.Addr,
			IsOnline:     true,
			LastJoinedAt: time.Now(),
			ApiAddresses: apiAddressesFromPeer(peer),
		}

		existing := controllers[peer.Id]
		if existing == nil {
			if err = self.Create(newController, changeCtx); err != nil {
				pfxlog.Logger().WithError(err).WithField("ctrlId", peer.Id).
					Error("could not create controller during peer(s) connection")
			}
		} else if peerConnectedEvent || existing.IsChanged(newController) {
			if err = self.Update(newController, nil, changeCtx); err != nil {
				pfxlog.Logger().WithError(err).WithField("ctrlId", peer.Id).
					Error("could not update controller during peer(s) connection")
			}
		}
	}

	// If we're the new leader, marking any controller not connected to us as offline
	if !peerConnectedEvent {
		connectedPeers := map[string]struct{}{}
		for _, peer := range self.env.GetCommandDispatcher().GetPeers() {
			connectedPeers[peer.Id()] = struct{}{}
		}

		disconnectFields := fields.UpdatedFieldsMap{
			db.FieldControllerIsOnline: struct{}{},
		}

		for _, controller := range controllers {
			if controller.IsOnline && controller.Id != selfAsPeer.Id {
				if _, ok := connectedPeers[controller.Id]; !ok {
					controller.IsOnline = false

					if err := self.Update(controller, disconnectFields, changeCtx); err != nil {
						pfxlog.Logger().WithError(err).Error("could not update controller marking peer disconnected")
					}
				}
			}
		}
	}
}

func (self *ControllerManager) DeleteRemovedPeers(peers []*event.ClusterPeer) {
	controllers := map[string]*Controller{}

	result, err := self.BaseList("true limit none")
	if err != nil {
		pfxlog.Logger().WithError(err).Error("failed to list controllers")
		return
	}

	for _, ctrl := range result.Entities {
		controllers[ctrl.Id] = ctrl
	}

	changeCtx := change.New()
	changeCtx.SetSourceType("raft.peers.changed").
		SetChangeAuthorType(change.AuthorTypeController)

	peers = append(peers, self.getCurrentAsClusterPeer())

	for _, peer := range peers {
		delete(controllers, peer.Id)
	}

	for _, controller := range controllers {
		if err = self.Delete(controller.Id, changeCtx); err != nil {
			pfxlog.Logger().WithError(err).WithField("ctrlId", controller.Id).
				Error("could not delete controller during cluster member change")
		}
	}
}

func (self *ControllerManager) PeersDisconnected(peers []*event.ClusterPeer) {
	changeCtx := change.New()
	changeCtx.SetSourceType("raft.peers.disconnected").
		SetChangeAuthorType(change.AuthorTypeController)

	disconnectFields := fields.UpdatedFieldsMap{
		db.FieldControllerIsOnline: struct{}{},
	}
	for _, peer := range peers {
		controller := &Controller{
			BaseEntity: models.BaseEntity{
				Id: peer.Id,
			},
			IsOnline: false,
		}

		if err := self.Update(controller, disconnectFields, changeCtx); err != nil {
			pfxlog.Logger().WithError(err).Error("could not update controller during peer(s) disconnection")
		}
	}
}

// UpdateSelfOnNewLeader has each controller update itself with its current configuration. This is then
// forwarded to the current leader for updates. Ensure that new leaders always have the most update-to-date configuration
// from each controller and avoid stale or out-of-date information.
func (self *ControllerManager) UpdateSelfOnNewLeader() {
	peer := self.getCurrentAsClusterPeer()

	newController := &Controller{
		BaseEntity: models.BaseEntity{
			Id: peer.Id,
		},
		Name:         peer.ServerCert[0].Subject.CommonName,
		CertPem:      nfpem.EncodeToString(peer.ServerCert[0]),
		Fingerprint:  nfpem.FingerprintFromCertificate(peer.ServerCert[0]),
		CtrlAddress:  peer.Addr,
		IsOnline:     true,
		ApiAddresses: apiAddressesFromPeer(peer),
	}
	disconnectFields := fields.UpdatedFieldsMap{
		db.FieldControllerIsOnline:          struct{}{},
		db.FieldControllerCertPem:           struct{}{},
		db.FieldControllerFingerprint:       struct{}{},
		db.FieldControllerCtrlAddress:       struct{}{},
		db.FieldControllerApiAddresses:      struct{}{},
		db.FieldControllerApiAddressUrl:     struct{}{},
		db.FieldControllerApiAddressVersion: struct{}{},
		db.FieldName:                        struct{}{},
	}

	changeCtx := change.New()
	changeCtx.SetSourceType("raft.state.has_leader").
		SetChangeAuthorType(change.AuthorTypeController)

	if err := self.Update(newController, disconnectFields, changeCtx); err != nil {
		pfxlog.Logger().WithError(err).Error("could not update self")
	}
}

// apiAddressFromPeer converts event.ClusterPeer API Addresses to model API Addresses
func apiAddressesFromPeer(peer *event.ClusterPeer) map[string][]ApiAddress {
	result := map[string][]ApiAddress{}

	for apiKey, instances := range peer.ApiAddresses {
		result[apiKey] = nil

		for _, instance := range instances {
			result[apiKey] = append(result[apiKey], ApiAddress{
				Url:     instance.Url,
				Version: instance.Version,
			})
		}
	}

	return result
}