The control plane declares, and hears back
`declare` sends a node a signed declaration; `serve` now also consumes reports. Signed over the exact bytes published, which is what the node verifies. Anything re-encoding in between would sign one thing and check another, and a difference in key order alone would have a node refuse a declaration that was genuinely the mesh's. Sent to the node's queue directly rather than through the exchange: a declaration is for one node, and routing by name through a shared exchange means a binding per node that nothing removes when a node is retired. Enrolment now issues the node its own broker password, replacing the token's secret, and tells it the broker address, the fingerprint and the signing key -- so a node can reconnect after a restart without a person and a new token, which is what makes disconnection ordinary rather than a crisis. A report is a statement, not a write. What a node says it applied is its own account of its own machine, kept as a copy for recovery rather than as a source.
This commit is contained in:
+41
-5
@@ -24,7 +24,7 @@ const AMQPVar = "MESH_BROKER_AMQP"
|
||||
type Enroller interface {
|
||||
// Enrol spends the token, records the key, and reports the node's name. The error is
|
||||
// returned to the node as a refusal; it must be the same for every reason a token can fail.
|
||||
Enrol(ctx context.Context, secret string, public ed25519.PublicKey, profile map[string]any) (string, error)
|
||||
Enrol(ctx context.Context, secret string, public ed25519.PublicKey, profile map[string]any) (EnrolReply, error)
|
||||
}
|
||||
|
||||
// Server consumes what nodes say.
|
||||
@@ -77,6 +77,17 @@ func Connect(enroller Enroller) (*Server, error) {
|
||||
log: log.New(os.Stdout, "", log.LstdFlags)}, nil
|
||||
}
|
||||
|
||||
func (s *Server) bindOrClose(channel *amqp.Channel, conn *amqp.Connection, key string) error {
|
||||
if err := channel.QueueBind(ControlQueue, key, Exchange, false, nil); err != nil {
|
||||
conn.Close()
|
||||
return fmt.Errorf("cannot bind %s to %s/%s: %w", ControlQueue, Exchange, key, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Channel is the control plane's channel, for sending declarations.
|
||||
func (s *Server) Channel() *amqp.Channel { return s.channel }
|
||||
|
||||
func (s *Server) Close() {
|
||||
if s.channel != nil {
|
||||
_ = s.channel.Close()
|
||||
@@ -106,7 +117,7 @@ func (s *Server) Serve(ctx context.Context) error {
|
||||
}
|
||||
|
||||
closed := s.conn.NotifyClose(make(chan *amqp.Error, 1))
|
||||
s.log.Printf("consuming %s, bound to %s/%s", ControlQueue, Exchange, KeyEnrol)
|
||||
s.log.Printf("consuming %s, bound to %s/{%s,%s}", ControlQueue, Exchange, KeyEnrol, KeyReport)
|
||||
|
||||
for {
|
||||
select {
|
||||
@@ -130,6 +141,8 @@ func (s *Server) handle(ctx context.Context, delivery amqp.Delivery) {
|
||||
switch delivery.RoutingKey {
|
||||
case KeyEnrol:
|
||||
s.handleEnrol(ctx, delivery)
|
||||
case KeyReport:
|
||||
s.handleReport(delivery)
|
||||
default:
|
||||
// Rejected without requeue: a message nothing understands will not be understood on the
|
||||
// next attempt either, and requeuing it would spin.
|
||||
@@ -138,6 +151,29 @@ func (s *Server) handle(ctx context.Context, delivery amqp.Delivery) {
|
||||
}
|
||||
}
|
||||
|
||||
// handleReport records what a node says it did.
|
||||
//
|
||||
// A node states; nothing here writes anything the node claimed about itself beyond that it was
|
||||
// heard from. What it applied is its own account of its own machine, and the mesh keeps the last
|
||||
// one as a copy for recovery rather than as a source (novox/hq 09-the-node-lifecycle).
|
||||
func (s *Server) handleReport(delivery amqp.Delivery) {
|
||||
var report Report
|
||||
if err := json.Unmarshal(delivery.Body, &report); err != nil {
|
||||
s.log.Printf("a report could not be read: %v", err)
|
||||
_ = delivery.Reject(false)
|
||||
return
|
||||
}
|
||||
switch {
|
||||
case report.Refused != "":
|
||||
s.log.Printf("%s refused a declaration: %s", report.Node, report.Refused)
|
||||
case len(report.Failed) > 0:
|
||||
s.log.Printf("%s applied %d and failed: %v", report.Node, len(report.Applied), report.Failed)
|
||||
default:
|
||||
s.log.Printf("%s applied %d resource(s)", report.Node, len(report.Applied))
|
||||
}
|
||||
_ = delivery.Ack(false)
|
||||
}
|
||||
|
||||
func (s *Server) handleEnrol(ctx context.Context, delivery amqp.Delivery) {
|
||||
reply := EnrolReply{Refusal: "that token cannot be used"}
|
||||
|
||||
@@ -145,14 +181,14 @@ func (s *Server) handleEnrol(ctx context.Context, delivery amqp.Delivery) {
|
||||
if err := json.Unmarshal(delivery.Body, &request); err != nil {
|
||||
s.log.Printf("an enrolment request could not be read: %v", err)
|
||||
} else {
|
||||
name, err := s.enroller.Enrol(ctx, request.Secret, request.PublicKey, request.Profile)
|
||||
accepted, err := s.enroller.Enrol(ctx, request.Secret, request.PublicKey, request.Profile)
|
||||
if err != nil {
|
||||
// Logged in full here, where an operator can see it; sent back as one refusal, so
|
||||
// that somebody guessing learns nothing from which reason came back.
|
||||
s.log.Printf("refusing enrolment for %q: %v", request.Node, err)
|
||||
} else {
|
||||
reply = EnrolReply{Accepted: true, Node: name, Queue: QueueFor(name)}
|
||||
s.log.Printf("enrolled %s", name)
|
||||
reply = accepted
|
||||
s.log.Printf("enrolled %s", accepted.Node)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user