Implement entity change events. Fixes openziti/fabric#562

controller/api/context.go

@@ -79,11 +79,22 @@ func (rc *RequestContextImpl) GetEntitySubId() (string, error) {
 }
 
 func (rc *RequestContextImpl) NewChangeContext() *change.Context {
-	src := fmt.Sprintf("rest[auth=fabric/host=%v/method=%v/remote=%v]", rc.GetRequest().Host, rc.GetRequest().Method, rc.GetRequest().RemoteAddr)
-	changeCtx := change.New().SetSource(src).SetChangeAuthorType("fabric.admin")
-	if rc.Request.Form.Has("traceId") {
-		changeCtx.SetChangeAuthorId(rc.Request.Form.Get("traceId"))
+	changeCtx := change.New().SetSourceType(change.SourceTypeRest).
+		SetSourceAuth("fabric").
+		SetSourceMethod(rc.GetRequest().Method).
+		SetSourceLocal(rc.GetRequest().Host).
+		SetSourceRemote(rc.GetRequest().RemoteAddr)
+
+	changeCtx.SetChangeAuthorType(change.AuthorTypeCert)
+
+	if rc.Request.TLS != nil {
+		for _, cert := range rc.Request.TLS.PeerCertificates {
+			if !cert.IsCA {
+				changeCtx.SetChangeAuthorId(cert.Subject.CommonName)
+			}
+		}
 	}
+
 	return changeCtx
 }
 

controller/change/context.go

@@ -31,7 +31,27 @@ const (
 	AuthorNameKey = "authorName"
 	AuthorTypeKey = "authorType"
 	TraceIdKey    = "traceId"
-	Source        = "source"
+	SourceType    = "src.type"
+	SourceAuth    = "src.auth"
+	SourceMethod  = "src.method"
+	SourceLocal   = "src.local"
+	SourceRemote  = "src.remote"
+)
+
+type AuthorType string
+
+const (
+	AuthorTypeCert         = "cert"
+	AuthorTypeIdentity     = "identity"
+	AuthorTypeRouter       = "router"
+	AuthorTypeController   = "controller"
+	AuthorTypeUnattributed = "unattributed"
+)
+
+const (
+	SourceTypeControlChannel = "ctrl.channel"
+	SourceTypeRest           = "rest"
+	SourceTypeXt             = "xt"
 )
 
 func New() *Context {
@@ -45,6 +65,25 @@ type Context struct {
 	RaftIndex  uint64
 }
 
+type Author struct {
+	Type string `json:"type"`
+	Id   string `json:"id,omitempty"`
+	Name string `json:"name,omitempty"`
+}
+
+type Source struct {
+	Type       string `json:"type"`
+	Auth       string `json:"auth,omitempty"`
+	LocalAddr  string `json:"local_addr,omitempty"`
+	RemoteAddr string `json:"remote_addr,omitempty"`
+	Method     string `json:"method,omitempty"`
+}
+
+func (self *Context) SetChangeAuthorType(val AuthorType) *Context {
+	self.Attributes[AuthorTypeKey] = string(val)
+	return self
+}
+
 func (self *Context) SetChangeAuthorId(val string) *Context {
 	self.Attributes[AuthorIdKey] = val
 	return self
@@ -55,25 +94,73 @@ func (self *Context) SetChangeAuthorName(val string) *Context {
 	return self
 }
 
-func (self *Context) SetChangeAuthorType(val string) *Context {
-	self.Attributes[AuthorTypeKey] = val
-	return self
-}
-
 func (self *Context) SetTraceId(val string) *Context {
 	self.Attributes[TraceIdKey] = val
 	return self
 }
 
-func (self *Context) SetSource(val string) *Context {
-	self.Attributes[Source] = val
+func (self *Context) SetSourceType(val string) *Context {
+	self.Attributes[SourceType] = val
 	return self
 }
 
+func (self *Context) SetSourceAuth(val string) *Context {
+	self.Attributes[SourceAuth] = val
+	return self
+}
+
+func (self *Context) SetSourceMethod(val string) *Context {
+	self.Attributes[SourceMethod] = val
+	return self
+}
+
+func (self *Context) SetSourceLocal(val string) *Context {
+	self.Attributes[SourceLocal] = val
+	return self
+}
+
+func (self *Context) SetSourceRemote(val string) *Context {
+	self.Attributes[SourceRemote] = val
+	return self
+}
+
+func (self *Context) GetAuthor() *Author {
+	if self == nil {
+		return nil
+	}
+	return &Author{
+		Type: self.Attributes[AuthorTypeKey],
+		Id:   self.Attributes[AuthorIdKey],
+		Name: self.Attributes[AuthorNameKey],
+	}
+}
+
+func (self *Context) GetSource() *Source {
+	if self == nil {
+		return nil
+	}
+	return &Source{
+		Type:       self.Attributes[SourceType],
+		Auth:       self.Attributes[SourceAuth],
+		LocalAddr:  self.Attributes[SourceLocal],
+		RemoteAddr: self.Attributes[SourceRemote],
+		Method:     self.Attributes[SourceMethod],
+	}
+}
+
+func (self *Context) PopulateMetadata(meta map[string]any) {
+	meta["author"] = self.GetAuthor()
+	meta["source"] = self.GetSource()
+	if traceId, found := self.Attributes[TraceIdKey]; found {
+		meta["trace_id"] = traceId
+	}
+}
+
 func (self *Context) ToProtoBuf() *cmd_pb.ChangeContext {
 	if self == nil {
 		return nil
 	}
+
 	return &cmd_pb.ChangeContext{
 		Attributes: self.Attributes,
 		RaftIndex:  self.RaftIndex,

controller/command/command.go

@@ -78,6 +78,11 @@ func (self *LocalDispatcher) Dispatch(command Command) error {
 		}
 	}()
 
+	changeCtx := command.GetChangeContext()
+	if changeCtx == nil {
+		changeCtx = change.New().SetSourceType("unattributed").SetChangeAuthorType(change.AuthorTypeUnattributed)
+	}
+	ctx := changeCtx.NewMutateContext()
 	if self.EncodeDecodeCommands {
 		bytes, err := command.Encode()
 		if err != nil {
@@ -87,10 +92,10 @@ func (self *LocalDispatcher) Dispatch(command Command) error {
 		if err != nil {
 			return err
 		}
-		return cmd.Apply(change.New().NewMutateContext())
+		return cmd.Apply(ctx)
 	}
 
-	return command.Apply(change.New().NewMutateContext())
+	return command.Apply(ctx)
 }
 
 // Decoder instances know how to decode encoded commands

controller/db/db.go

@@ -21,7 +21,8 @@ import (
 )
 
 const (
-	RootBucket = "ziti"
+	RootBucket     = "ziti"
+	MetadataBucket = "metadata"
 )
 
 func Open(path string) (boltz.Db, error) {

controller/db/router_store.go

@@ -32,11 +32,11 @@ const (
 
 type Router struct {
 	boltz.BaseExtEntity
-	Name        string
-	Fingerprint *string
-	Cost        uint16
-	NoTraversal bool
-	Disabled    bool
+	Name        string  `json:"name"`
+	Fingerprint *string `json:"fingerprint"`
+	Cost        uint16  `json:"cost"`
+	NoTraversal bool    `json:"no_traversal"`
+	Disabled    bool    `json:"disabled"`
 }
 
 func (entity *Router) GetEntityType() string {

controller/db/service_store.go

@@ -31,8 +31,8 @@ const (
 
 type Service struct {
 	boltz.BaseExtEntity
-	Name               string
-	TerminatorStrategy string
+	Name               string `json:"name"`
+	TerminatorStrategy string `json:"terminator_strategy"`
 }
 
 func (entity *Service) GetEntityType() string {

controller/db/terminator_store.go

@@ -42,16 +42,16 @@ const (
 
 type Terminator struct {
 	boltz.BaseExtEntity
-	Service        string
-	Router         string
-	Binding        string
-	Address        string
-	InstanceId     string
-	InstanceSecret []byte
-	Cost           uint16
-	Precedence     string
-	PeerData       xt.PeerData
-	HostId         string
+	Service        string      `json:"service"`
+	Router         string      `json:"router"`
+	Binding        string      `json:"binding"`
+	Address        string      `json:"address"`
+	InstanceId     string      `json:"instance_id"`
+	InstanceSecret []byte      `json:"instance_secret"`
+	Cost           uint16      `json:"cost"`
+	Precedence     string      `json:"precedence"`
+	PeerData       xt.PeerData `json:"peer_data"`
+	HostId         string      `json:"host_id"`
 }
 
 func (entity *Terminator) GetCost() uint16 {

controller/handler_ctrl/base.go

@@ -17,7 +17,6 @@
 package handler_ctrl
 
 import (
-	"fmt"
 	"github.com/openziti/channel/v2"
 	"github.com/openziti/fabric/controller/change"
 	"github.com/openziti/fabric/controller/network"
@@ -28,10 +27,13 @@ type baseHandler struct {
 	network *network.Network
 }
 
-func (self *baseHandler) newChangeContext(ch channel.Channel) *change.Context {
+func (self *baseHandler) newChangeContext(ch channel.Channel, method string) *change.Context {
 	return change.New().
 		SetChangeAuthorId(self.router.Id).
 		SetChangeAuthorName(self.router.Name).
-		SetChangeAuthorType("router").
-		SetSource(fmt.Sprintf("ctrl[%v]", ch.Underlay().GetRemoteAddr().String()))
+		SetChangeAuthorType(change.AuthorTypeRouter).
+		SetSourceType(change.SourceTypeControlChannel).
+		SetSourceMethod(method).
+		SetSourceLocal(ch.Underlay().GetLocalAddr().String()).
+		SetSourceRemote(ch.Underlay().GetRemoteAddr().String())
 }

controller/handler_ctrl/create_terminator.go

@@ -72,7 +72,7 @@ func (self *createTerminatorHandler) handleCreateTerminator(msg *channel.Message
 		Cost:           uint16(request.Cost),
 	}
 
-	if err := self.network.Terminators.Create(terminator, self.newChangeContext(ch)); err == nil {
+	if err := self.network.Terminators.Create(terminator, self.newChangeContext(ch, "fabric.create.terminator")); err == nil {
 		pfxlog.Logger().Infof("created terminator [t/%s]", terminator.Id)
 		handler_common.SendSuccess(msg, ch, terminator.Id)
 	} else {

controller/handler_ctrl/remove_terminator.go

@@ -63,7 +63,7 @@ func (self *removeTerminatorHandler) handleRemoveTerminator(msg *channel.Message
 		return
 	}
 
-	if err := self.network.Terminators.Delete(request.TerminatorId, self.newChangeContext(ch)); err == nil {
+	if err := self.network.Terminators.Delete(request.TerminatorId, self.newChangeContext(ch, "fabric.remove.terminator")); err == nil {
 		log.
 			WithField("routerId", ch.Id()).
 			WithField("serviceId", terminator.Service).

controller/handler_ctrl/remove_terminators.go

@@ -57,7 +57,7 @@ func (self *removeTerminatorsHandler) HandleReceive(msg *channel.Message, ch cha
 func (self *removeTerminatorsHandler) handleRemoveTerminators(msg *channel.Message, ch channel.Channel, request *ctrl_pb.RemoveTerminatorsRequest) {
 	log := pfxlog.ContextLogger(ch.Label())
 
-	if err := self.network.Terminators.DeleteBatch(request.TerminatorIds, self.newChangeContext(ch)); err == nil {
+	if err := self.network.Terminators.DeleteBatch(request.TerminatorIds, self.newChangeContext(ch, "fabric.remove.terminators.batch")); err == nil {
 		log.
 			WithField("routerId", ch.Id()).
 			WithField("terminatorIds", request.TerminatorIds).

controller/handler_ctrl/update_terminator.go

@@ -97,7 +97,7 @@ func (self *updateTerminatorHandler) handleUpdateTerminator(msg *channel.Message
 		checker[db.FieldTerminatorPrecedence] = struct{}{}
 	}
 
-	if err := self.network.Terminators.Update(terminator, checker, self.newChangeContext(ch)); err != nil {
+	if err := self.network.Terminators.Update(terminator, checker, self.newChangeContext(ch, "fabric.update.terminator")); err != nil {
 		handler_common.SendFailure(msg, ch, err.Error())
 		return
 	}

controller/network/routesender.go

@@ -165,8 +165,15 @@ func (self *routeSender) handleRouteSend(attempt uint32, path *Path, strategy xt
 				changeCtx := change.New().
 					SetChangeAuthorId(status.Router.Id).
 					SetChangeAuthorName(status.Router.Name).
-					SetChangeAuthorType("router").
-					SetSource("router reported invalid terminator")
+					SetChangeAuthorType(change.AuthorTypeRouter).
+					SetSourceType(change.SourceTypeControlChannel).
+					SetSourceMethod("route.response")
+				if ch := status.Router.Control; ch != nil {
+					changeCtx.
+						SetSourceLocal(ch.Underlay().GetLocalAddr().String()).
+						SetSourceRemote(ch.Underlay().GetRemoteAddr().String())
+				}
+
 				if err := self.terminators.Delete(terminator.GetId(), changeCtx); err != nil {
 					logger.WithError(fmt.Errorf("unable to delete invalid terminator: %v", err))
 				}

controller/network/terminator.go

@@ -196,7 +196,7 @@ func (self *TerminatorManager) handlePrecedenceChange(terminatorId string, prece
 		db.FieldTerminatorPrecedence: struct{}{},
 	}
 
-	if err = self.Update(terminator, checker, change.New().SetChangeAuthorId("xt").SetChangeAuthorType("controller")); err != nil {
+	if err = self.Update(terminator, checker, change.New().SetSourceType(change.SourceTypeXt).SetChangeAuthorType(change.AuthorTypeController)); err != nil {
 		pfxlog.Logger().Errorf("unable to update precedence for terminator %v to %v (%v)", terminatorId, precedence, err)
 	}
 }

controller/raft/fsm.go

@@ -35,8 +35,7 @@ import (
 )
 
 const (
-	bucketName = "raft"
-	fieldIndex = "index"
+	fieldIndex = "raftIndex"
 )
 
 func NewFsm(dataDir string, decoders command.Decoders, indexTracker IndexTracker, eventDispatcher event.Dispatcher) *BoltDbFsm {
@@ -85,7 +84,7 @@ func (self *BoltDbFsm) GetDb() boltz.Db {
 func (self *BoltDbFsm) loadCurrentIndex() (uint64, error) {
 	var result uint64
 	err := self.db.View(func(tx *bbolt.Tx) error {
-		if raftBucket := boltz.Path(tx, db.RootBucket, bucketName); raftBucket != nil {
+		if raftBucket := boltz.Path(tx, db.RootBucket, db.MetadataBucket); raftBucket != nil {
 			if val := raftBucket.GetInt64(fieldIndex); val != nil {
 				result = uint64(*val)
 			}
@@ -96,7 +95,7 @@ func (self *BoltDbFsm) loadCurrentIndex() (uint64, error) {
 }
 
 func (self *BoltDbFsm) updateIndexInTx(tx *bbolt.Tx, index uint64) error {
-	raftBucket := boltz.GetOrCreatePath(tx, db.RootBucket, bucketName)
+	raftBucket := boltz.GetOrCreatePath(tx, db.RootBucket, db.MetadataBucket)
 	raftBucket.SetInt64(fieldIndex, int64(index), nil)
 	return raftBucket.GetError()
 }
@@ -156,7 +155,7 @@ func (self *BoltDbFsm) Apply(log *raft.Log) interface{} {
 			logger.Infof("apply log with type %T", cmd)
 			changeCtx := cmd.GetChangeContext()
 			if changeCtx == nil {
-				changeCtx = change.New().SetSource("untracked")
+				changeCtx = change.New().SetSourceType("unattributed").SetChangeAuthorType(change.AuthorTypeUnattributed)
 			}
 			changeCtx.RaftIndex = log.Index