9 Commits

Author SHA1 Message Date
714320c018 Add edge routing and health check RPCs (Phase 2)
New agent RPCs for v2 multi-node orchestration:

- SetupEdgeRoute: provisions TLS cert from Metacrypt, resolves backend
  hostname to Tailnet IP, validates it's in 100.64.0.0/10, registers
  L7 route in mc-proxy. Rejects backend_tls=false.
- RemoveEdgeRoute: removes mc-proxy route, cleans up TLS cert, removes
  registry entry.
- ListEdgeRoutes: returns all edge routes with cert serial/expiry.
- HealthCheck: returns agent health and container count.

New database table (migration 4): edge_routes stores hostname, backend
info, and cert paths for persistence across agent restarts.

ProxyRouter gains CertPath/KeyPath helpers for consistent cert path
construction.

Security:
- Backend hostname must resolve to a Tailnet IP (100.64.0.0/10)
- backend_tls=false is rejected (no cleartext to backends)
- Cert provisioning failure fails the setup (no route to missing cert)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-02 13:13:10 -07:00
fa8ba6fac1 Move ARCHITECTURE_V2.md to metacircular docs
The v2 architecture doc is platform-wide (covers master, agents,
edge routing, snapshots, migration across all nodes). Moved to
docs/architecture-v2.md in the metacircular workspace repo.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-02 11:09:09 -07:00
f66758b92b Hardcode version in flake.nix
The git fetcher doesn't provide gitDescribe, so the Nix build was
falling through to shortRev and producing commit-hash versions instead
of tag-based ones.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-30 17:46:12 -07:00
09d0d197c3 Add component-level targeting to start, stop, and restart
Allow start/stop/restart to target a single component via
<service>/<component> syntax, matching deploy/logs/purge. When a
component is specified, start/stop skip toggling the service-level
active flag. Agent-side filtering returns NotFound for unknown
components.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-30 17:26:05 -07:00
52914d50b0 Pass mode, backend-tls, and tls cert/key through route add
The --mode flag was defined but never wired through to the RPC.
Add tls_cert and tls_key fields to AddProxyRouteRequest so L7
routes can be created via mcp route add.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-29 20:44:44 -07:00
bb4bee51ba Add mono-repo consideration to ARCHITECTURE_V2.md open questions
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-29 20:40:32 -07:00
4ac8a6d60b Add ARCHITECTURE_V2.md for multi-node master/agent topology
Documents the planned v2 architecture: mcp-master on straylight
coordinates deployments across worker (rift) and edge (svc) nodes.
Includes edge routing flow, agent RPCs, migration plan, and
operational issues from v1 that motivate the redesign.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-29 20:37:24 -07:00
d8f45ca520 Merge explicit ports with route-allocated ports during deploy
Previously, explicit port mappings from the service definition were
ignored when routes were present. Now both are included, allowing
services to have stable external port bindings alongside dynamic
route-allocated ports.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-29 19:28:40 -07:00
95f86157b4 Add mcp route command for managing mc-proxy routes
New top-level command with list, add, remove subcommands. Supports
-n/--node to target a specific node. Adds AddProxyRoute and
RemoveProxyRoute RPCs to the agent. Moves route listing from
mcp node routes to mcp route list.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-29 19:10:11 -07:00
15 changed files with 1819 additions and 141 deletions

View File

@@ -14,8 +14,8 @@ import (
func stopCmd() *cobra.Command {
return &cobra.Command{
Use: "stop <service>",
Short: "Stop all components, set active=false",
Use: "stop <service>[/<component>]",
Short: "Stop components (or all), set active=false",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
cfg, err := config.LoadCLIConfig(cfgPath)
@@ -23,7 +23,7 @@ func stopCmd() *cobra.Command {
return fmt.Errorf("load config: %w", err)
}
serviceName := args[0]
serviceName, component := parseServiceArg(args[0])
defPath := filepath.Join(cfg.Services.Dir, serviceName+".toml")
def, err := servicedef.Load(defPath)
@@ -31,10 +31,13 @@ func stopCmd() *cobra.Command {
return fmt.Errorf("load service def: %w", err)
}
active := false
def.Active = &active
if err := servicedef.Write(defPath, def); err != nil {
return fmt.Errorf("write service def: %w", err)
// Only flip active=false when stopping the whole service.
if component == "" {
active := false
def.Active = &active
if err := servicedef.Write(defPath, def); err != nil {
return fmt.Errorf("write service def: %w", err)
}
}
address, err := findNodeAddress(cfg, def.Node)
@@ -49,7 +52,8 @@ func stopCmd() *cobra.Command {
defer func() { _ = conn.Close() }()
resp, err := client.StopService(context.Background(), &mcpv1.StopServiceRequest{
Name: serviceName,
Name: serviceName,
Component: component,
})
if err != nil {
return fmt.Errorf("stop service: %w", err)
@@ -63,8 +67,8 @@ func stopCmd() *cobra.Command {
func startCmd() *cobra.Command {
return &cobra.Command{
Use: "start <service>",
Short: "Start all components, set active=true",
Use: "start <service>[/<component>]",
Short: "Start components (or all), set active=true",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
cfg, err := config.LoadCLIConfig(cfgPath)
@@ -72,7 +76,7 @@ func startCmd() *cobra.Command {
return fmt.Errorf("load config: %w", err)
}
serviceName := args[0]
serviceName, component := parseServiceArg(args[0])
defPath := filepath.Join(cfg.Services.Dir, serviceName+".toml")
def, err := servicedef.Load(defPath)
@@ -80,10 +84,13 @@ func startCmd() *cobra.Command {
return fmt.Errorf("load service def: %w", err)
}
active := true
def.Active = &active
if err := servicedef.Write(defPath, def); err != nil {
return fmt.Errorf("write service def: %w", err)
// Only flip active=true when starting the whole service.
if component == "" {
active := true
def.Active = &active
if err := servicedef.Write(defPath, def); err != nil {
return fmt.Errorf("write service def: %w", err)
}
}
address, err := findNodeAddress(cfg, def.Node)
@@ -98,7 +105,8 @@ func startCmd() *cobra.Command {
defer func() { _ = conn.Close() }()
resp, err := client.StartService(context.Background(), &mcpv1.StartServiceRequest{
Name: serviceName,
Name: serviceName,
Component: component,
})
if err != nil {
return fmt.Errorf("start service: %w", err)
@@ -112,8 +120,8 @@ func startCmd() *cobra.Command {
func restartCmd() *cobra.Command {
return &cobra.Command{
Use: "restart <service>",
Short: "Restart all components",
Use: "restart <service>[/<component>]",
Short: "Restart components (or all)",
Args: cobra.ExactArgs(1),
RunE: func(cmd *cobra.Command, args []string) error {
cfg, err := config.LoadCLIConfig(cfgPath)
@@ -121,7 +129,7 @@ func restartCmd() *cobra.Command {
return fmt.Errorf("load config: %w", err)
}
serviceName := args[0]
serviceName, component := parseServiceArg(args[0])
defPath := filepath.Join(cfg.Services.Dir, serviceName+".toml")
def, err := servicedef.Load(defPath)
@@ -141,7 +149,8 @@ func restartCmd() *cobra.Command {
defer func() { _ = conn.Close() }()
resp, err := client.RestartService(context.Background(), &mcpv1.RestartServiceRequest{
Name: serviceName,
Name: serviceName,
Component: component,
})
if err != nil {
return fmt.Errorf("restart service: %w", err)

View File

@@ -53,6 +53,7 @@ func main() {
root.AddCommand(logsCmd())
root.AddCommand(editCmd())
root.AddCommand(dnsCmd())
root.AddCommand(routeCmd())
if err := root.Execute(); err != nil {
log.Fatal(err)

View File

@@ -40,62 +40,10 @@ func nodeCmd() *cobra.Command {
RunE: runNodeRemove,
}
routes := &cobra.Command{
Use: "routes",
Short: "List mc-proxy routes on all nodes",
RunE: runNodeRoutes,
}
cmd.AddCommand(list, add, remove, routes)
cmd.AddCommand(list, add, remove)
return cmd
}
func runNodeRoutes(_ *cobra.Command, _ []string) error {
first := true
return forEachNode(func(node config.NodeConfig, client mcpv1.McpAgentServiceClient) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
resp, err := client.ListProxyRoutes(ctx, &mcpv1.ListProxyRoutesRequest{})
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "warning: %s: list routes: %v\n", node.Name, err)
return nil
}
if !first {
fmt.Println()
}
first = false
fmt.Printf("NODE: %s\n", node.Name)
fmt.Printf("mc-proxy %s\n", resp.GetVersion())
if resp.GetStartedAt() != nil {
uptime := time.Since(resp.GetStartedAt().AsTime()).Truncate(time.Second)
fmt.Printf("uptime: %s\n", uptime)
}
fmt.Printf("connections: %d\n", resp.GetTotalConnections())
fmt.Println()
for _, ls := range resp.GetListeners() {
fmt.Printf(" %s routes=%d active=%d\n",
ls.GetAddr(), ls.GetRouteCount(), ls.GetActiveConnections())
for _, r := range ls.GetRoutes() {
mode := r.GetMode()
if mode == "" {
mode = "l4"
}
extra := ""
if r.GetBackendTls() {
extra = " (re-encrypt)"
}
fmt.Printf(" %s %s → %s%s\n", mode, r.GetHostname(), r.GetBackend(), extra)
}
}
return nil
})
}
func runNodeList(_ *cobra.Command, _ []string) error {
cfg, err := config.LoadCLIConfig(cfgPath)
if err != nil {

225
cmd/mcp/route.go Normal file
View File

@@ -0,0 +1,225 @@
package main
import (
"context"
"fmt"
"os"
"time"
"github.com/spf13/cobra"
mcpv1 "git.wntrmute.dev/mc/mcp/gen/mcp/v1"
"git.wntrmute.dev/mc/mcp/internal/config"
)
func routeCmd() *cobra.Command {
var nodeName string
cmd := &cobra.Command{
Use: "route",
Short: "Manage mc-proxy routes",
}
list := &cobra.Command{
Use: "list",
Short: "List mc-proxy routes",
RunE: func(_ *cobra.Command, _ []string) error {
return runRouteList(nodeName)
},
}
var (
routeMode string
backendTLS bool
tlsCert string
tlsKey string
)
add := &cobra.Command{
Use: "add <listener> <hostname> <backend>",
Short: "Add a route to mc-proxy",
Long: "Add a route. Example: mcp route add -n rift :443 mcq.svc.mcp.metacircular.net 127.0.0.1:48080 --mode l7 --tls-cert /srv/mc-proxy/certs/mcq.pem --tls-key /srv/mc-proxy/certs/mcq.key",
Args: cobra.ExactArgs(3),
RunE: func(_ *cobra.Command, args []string) error {
return runRouteAdd(nodeName, args, routeMode, backendTLS, tlsCert, tlsKey)
},
}
add.Flags().StringVar(&routeMode, "mode", "l4", "route mode (l4 or l7)")
add.Flags().BoolVar(&backendTLS, "backend-tls", false, "re-encrypt traffic to backend")
add.Flags().StringVar(&tlsCert, "tls-cert", "", "path to TLS cert on the node (required for l7)")
add.Flags().StringVar(&tlsKey, "tls-key", "", "path to TLS key on the node (required for l7)")
remove := &cobra.Command{
Use: "remove <listener> <hostname>",
Short: "Remove a route from mc-proxy",
Long: "Remove a route. Example: mcp route remove -n rift :443 mcq.metacircular.net",
Args: cobra.ExactArgs(2),
RunE: func(_ *cobra.Command, args []string) error {
return runRouteRemove(nodeName, args)
},
}
cmd.PersistentFlags().StringVarP(&nodeName, "node", "n", "", "target node (required)")
cmd.AddCommand(list, add, remove)
return cmd
}
func runRouteList(nodeName string) error {
if nodeName == "" {
return runRouteListAll()
}
cfg, err := config.LoadCLIConfig(cfgPath)
if err != nil {
return fmt.Errorf("load config: %w", err)
}
address, err := findNodeAddress(cfg, nodeName)
if err != nil {
return err
}
client, conn, err := dialAgent(address, cfg)
if err != nil {
return fmt.Errorf("dial agent: %w", err)
}
defer func() { _ = conn.Close() }()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
resp, err := client.ListProxyRoutes(ctx, &mcpv1.ListProxyRoutesRequest{})
if err != nil {
return fmt.Errorf("list routes: %w", err)
}
printRoutes(nodeName, resp)
return nil
}
func runRouteListAll() error {
first := true
return forEachNode(func(node config.NodeConfig, client mcpv1.McpAgentServiceClient) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
resp, err := client.ListProxyRoutes(ctx, &mcpv1.ListProxyRoutesRequest{})
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "warning: %s: list routes: %v\n", node.Name, err)
return nil
}
if !first {
fmt.Println()
}
first = false
printRoutes(node.Name, resp)
return nil
})
}
func printRoutes(nodeName string, resp *mcpv1.ListProxyRoutesResponse) {
fmt.Printf("NODE: %s\n", nodeName)
fmt.Printf("mc-proxy %s\n", resp.GetVersion())
if resp.GetStartedAt() != nil {
uptime := time.Since(resp.GetStartedAt().AsTime()).Truncate(time.Second)
fmt.Printf("uptime: %s\n", uptime)
}
fmt.Printf("connections: %d\n", resp.GetTotalConnections())
fmt.Println()
for _, ls := range resp.GetListeners() {
fmt.Printf(" %s routes=%d active=%d\n",
ls.GetAddr(), ls.GetRouteCount(), ls.GetActiveConnections())
for _, r := range ls.GetRoutes() {
mode := r.GetMode()
if mode == "" {
mode = "l4"
}
extra := ""
if r.GetBackendTls() {
extra = " (re-encrypt)"
}
fmt.Printf(" %s %s → %s%s\n", mode, r.GetHostname(), r.GetBackend(), extra)
}
}
}
func runRouteAdd(nodeName string, args []string, mode string, backendTLS bool, tlsCert, tlsKey string) error {
if nodeName == "" {
return fmt.Errorf("--node is required")
}
cfg, err := config.LoadCLIConfig(cfgPath)
if err != nil {
return fmt.Errorf("load config: %w", err)
}
address, err := findNodeAddress(cfg, nodeName)
if err != nil {
return err
}
client, conn, err := dialAgent(address, cfg)
if err != nil {
return fmt.Errorf("dial agent: %w", err)
}
defer func() { _ = conn.Close() }()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err = client.AddProxyRoute(ctx, &mcpv1.AddProxyRouteRequest{
ListenerAddr: args[0],
Hostname: args[1],
Backend: args[2],
Mode: mode,
BackendTls: backendTLS,
TlsCert: tlsCert,
TlsKey: tlsKey,
})
if err != nil {
return fmt.Errorf("add route: %w", err)
}
fmt.Printf("Added route: %s %s → %s on %s (%s)\n", mode, args[1], args[2], args[0], nodeName)
return nil
}
func runRouteRemove(nodeName string, args []string) error {
if nodeName == "" {
return fmt.Errorf("--node is required")
}
cfg, err := config.LoadCLIConfig(cfgPath)
if err != nil {
return fmt.Errorf("load config: %w", err)
}
address, err := findNodeAddress(cfg, nodeName)
if err != nil {
return err
}
client, conn, err := dialAgent(address, cfg)
if err != nil {
return fmt.Errorf("dial agent: %w", err)
}
defer func() { _ = conn.Close() }()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
_, err = client.RemoveProxyRoute(ctx, &mcpv1.RemoveProxyRouteRequest{
ListenerAddr: args[0],
Hostname: args[1],
})
if err != nil {
return fmt.Errorf("remove route: %w", err)
}
fmt.Printf("Removed route: %s from %s (%s)\n", args[1], args[0], nodeName)
return nil
}

View File

@@ -10,7 +10,7 @@
let
system = "x86_64-linux";
pkgs = nixpkgs.legacyPackages.${system};
version = pkgs.lib.removePrefix "v" (self.gitDescribe or self.shortRev or self.dirtyShortRev or "unknown");
version = "0.8.3";
in
{
packages.${system} = {

File diff suppressed because it is too large Load Diff

View File

@@ -35,6 +35,12 @@ const (
McpAgentService_NodeStatus_FullMethodName = "/mcp.v1.McpAgentService/NodeStatus"
McpAgentService_ListDNSRecords_FullMethodName = "/mcp.v1.McpAgentService/ListDNSRecords"
McpAgentService_ListProxyRoutes_FullMethodName = "/mcp.v1.McpAgentService/ListProxyRoutes"
McpAgentService_AddProxyRoute_FullMethodName = "/mcp.v1.McpAgentService/AddProxyRoute"
McpAgentService_RemoveProxyRoute_FullMethodName = "/mcp.v1.McpAgentService/RemoveProxyRoute"
McpAgentService_SetupEdgeRoute_FullMethodName = "/mcp.v1.McpAgentService/SetupEdgeRoute"
McpAgentService_RemoveEdgeRoute_FullMethodName = "/mcp.v1.McpAgentService/RemoveEdgeRoute"
McpAgentService_ListEdgeRoutes_FullMethodName = "/mcp.v1.McpAgentService/ListEdgeRoutes"
McpAgentService_HealthCheck_FullMethodName = "/mcp.v1.McpAgentService/HealthCheck"
McpAgentService_Logs_FullMethodName = "/mcp.v1.McpAgentService/Logs"
)
@@ -67,6 +73,14 @@ type McpAgentServiceClient interface {
ListDNSRecords(ctx context.Context, in *ListDNSRecordsRequest, opts ...grpc.CallOption) (*ListDNSRecordsResponse, error)
// Proxy routes (query mc-proxy)
ListProxyRoutes(ctx context.Context, in *ListProxyRoutesRequest, opts ...grpc.CallOption) (*ListProxyRoutesResponse, error)
AddProxyRoute(ctx context.Context, in *AddProxyRouteRequest, opts ...grpc.CallOption) (*AddProxyRouteResponse, error)
RemoveProxyRoute(ctx context.Context, in *RemoveProxyRouteRequest, opts ...grpc.CallOption) (*RemoveProxyRouteResponse, error)
// Edge routing (called by master on edge nodes)
SetupEdgeRoute(ctx context.Context, in *SetupEdgeRouteRequest, opts ...grpc.CallOption) (*SetupEdgeRouteResponse, error)
RemoveEdgeRoute(ctx context.Context, in *RemoveEdgeRouteRequest, opts ...grpc.CallOption) (*RemoveEdgeRouteResponse, error)
ListEdgeRoutes(ctx context.Context, in *ListEdgeRoutesRequest, opts ...grpc.CallOption) (*ListEdgeRoutesResponse, error)
// Health (called by master on missed heartbeats)
HealthCheck(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error)
// Logs
Logs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogsResponse], error)
}
@@ -239,6 +253,66 @@ func (c *mcpAgentServiceClient) ListProxyRoutes(ctx context.Context, in *ListPro
return out, nil
}
func (c *mcpAgentServiceClient) AddProxyRoute(ctx context.Context, in *AddProxyRouteRequest, opts ...grpc.CallOption) (*AddProxyRouteResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(AddProxyRouteResponse)
err := c.cc.Invoke(ctx, McpAgentService_AddProxyRoute_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) RemoveProxyRoute(ctx context.Context, in *RemoveProxyRouteRequest, opts ...grpc.CallOption) (*RemoveProxyRouteResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RemoveProxyRouteResponse)
err := c.cc.Invoke(ctx, McpAgentService_RemoveProxyRoute_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) SetupEdgeRoute(ctx context.Context, in *SetupEdgeRouteRequest, opts ...grpc.CallOption) (*SetupEdgeRouteResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(SetupEdgeRouteResponse)
err := c.cc.Invoke(ctx, McpAgentService_SetupEdgeRoute_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) RemoveEdgeRoute(ctx context.Context, in *RemoveEdgeRouteRequest, opts ...grpc.CallOption) (*RemoveEdgeRouteResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(RemoveEdgeRouteResponse)
err := c.cc.Invoke(ctx, McpAgentService_RemoveEdgeRoute_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) ListEdgeRoutes(ctx context.Context, in *ListEdgeRoutesRequest, opts ...grpc.CallOption) (*ListEdgeRoutesResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(ListEdgeRoutesResponse)
err := c.cc.Invoke(ctx, McpAgentService_ListEdgeRoutes_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) HealthCheck(ctx context.Context, in *HealthCheckRequest, opts ...grpc.CallOption) (*HealthCheckResponse, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(HealthCheckResponse)
err := c.cc.Invoke(ctx, McpAgentService_HealthCheck_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *mcpAgentServiceClient) Logs(ctx context.Context, in *LogsRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[LogsResponse], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &McpAgentService_ServiceDesc.Streams[0], McpAgentService_Logs_FullMethodName, cOpts...)
@@ -287,6 +361,14 @@ type McpAgentServiceServer interface {
ListDNSRecords(context.Context, *ListDNSRecordsRequest) (*ListDNSRecordsResponse, error)
// Proxy routes (query mc-proxy)
ListProxyRoutes(context.Context, *ListProxyRoutesRequest) (*ListProxyRoutesResponse, error)
AddProxyRoute(context.Context, *AddProxyRouteRequest) (*AddProxyRouteResponse, error)
RemoveProxyRoute(context.Context, *RemoveProxyRouteRequest) (*RemoveProxyRouteResponse, error)
// Edge routing (called by master on edge nodes)
SetupEdgeRoute(context.Context, *SetupEdgeRouteRequest) (*SetupEdgeRouteResponse, error)
RemoveEdgeRoute(context.Context, *RemoveEdgeRouteRequest) (*RemoveEdgeRouteResponse, error)
ListEdgeRoutes(context.Context, *ListEdgeRoutesRequest) (*ListEdgeRoutesResponse, error)
// Health (called by master on missed heartbeats)
HealthCheck(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error)
// Logs
Logs(*LogsRequest, grpc.ServerStreamingServer[LogsResponse]) error
mustEmbedUnimplementedMcpAgentServiceServer()
@@ -347,6 +429,24 @@ func (UnimplementedMcpAgentServiceServer) ListDNSRecords(context.Context, *ListD
func (UnimplementedMcpAgentServiceServer) ListProxyRoutes(context.Context, *ListProxyRoutesRequest) (*ListProxyRoutesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListProxyRoutes not implemented")
}
func (UnimplementedMcpAgentServiceServer) AddProxyRoute(context.Context, *AddProxyRouteRequest) (*AddProxyRouteResponse, error) {
return nil, status.Error(codes.Unimplemented, "method AddProxyRoute not implemented")
}
func (UnimplementedMcpAgentServiceServer) RemoveProxyRoute(context.Context, *RemoveProxyRouteRequest) (*RemoveProxyRouteResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RemoveProxyRoute not implemented")
}
func (UnimplementedMcpAgentServiceServer) SetupEdgeRoute(context.Context, *SetupEdgeRouteRequest) (*SetupEdgeRouteResponse, error) {
return nil, status.Error(codes.Unimplemented, "method SetupEdgeRoute not implemented")
}
func (UnimplementedMcpAgentServiceServer) RemoveEdgeRoute(context.Context, *RemoveEdgeRouteRequest) (*RemoveEdgeRouteResponse, error) {
return nil, status.Error(codes.Unimplemented, "method RemoveEdgeRoute not implemented")
}
func (UnimplementedMcpAgentServiceServer) ListEdgeRoutes(context.Context, *ListEdgeRoutesRequest) (*ListEdgeRoutesResponse, error) {
return nil, status.Error(codes.Unimplemented, "method ListEdgeRoutes not implemented")
}
func (UnimplementedMcpAgentServiceServer) HealthCheck(context.Context, *HealthCheckRequest) (*HealthCheckResponse, error) {
return nil, status.Error(codes.Unimplemented, "method HealthCheck not implemented")
}
func (UnimplementedMcpAgentServiceServer) Logs(*LogsRequest, grpc.ServerStreamingServer[LogsResponse]) error {
return status.Error(codes.Unimplemented, "method Logs not implemented")
}
@@ -659,6 +759,114 @@ func _McpAgentService_ListProxyRoutes_Handler(srv interface{}, ctx context.Conte
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_AddProxyRoute_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(AddProxyRouteRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).AddProxyRoute(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_AddProxyRoute_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).AddProxyRoute(ctx, req.(*AddProxyRouteRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_RemoveProxyRoute_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RemoveProxyRouteRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).RemoveProxyRoute(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_RemoveProxyRoute_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).RemoveProxyRoute(ctx, req.(*RemoveProxyRouteRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_SetupEdgeRoute_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(SetupEdgeRouteRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).SetupEdgeRoute(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_SetupEdgeRoute_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).SetupEdgeRoute(ctx, req.(*SetupEdgeRouteRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_RemoveEdgeRoute_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RemoveEdgeRouteRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).RemoveEdgeRoute(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_RemoveEdgeRoute_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).RemoveEdgeRoute(ctx, req.(*RemoveEdgeRouteRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_ListEdgeRoutes_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ListEdgeRoutesRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).ListEdgeRoutes(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_ListEdgeRoutes_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).ListEdgeRoutes(ctx, req.(*ListEdgeRoutesRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_HealthCheck_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(HealthCheckRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(McpAgentServiceServer).HealthCheck(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: McpAgentService_HealthCheck_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(McpAgentServiceServer).HealthCheck(ctx, req.(*HealthCheckRequest))
}
return interceptor(ctx, in, info, handler)
}
func _McpAgentService_Logs_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(LogsRequest)
if err := stream.RecvMsg(m); err != nil {
@@ -741,6 +949,30 @@ var McpAgentService_ServiceDesc = grpc.ServiceDesc{
MethodName: "ListProxyRoutes",
Handler: _McpAgentService_ListProxyRoutes_Handler,
},
{
MethodName: "AddProxyRoute",
Handler: _McpAgentService_AddProxyRoute_Handler,
},
{
MethodName: "RemoveProxyRoute",
Handler: _McpAgentService_RemoveProxyRoute_Handler,
},
{
MethodName: "SetupEdgeRoute",
Handler: _McpAgentService_SetupEdgeRoute_Handler,
},
{
MethodName: "RemoveEdgeRoute",
Handler: _McpAgentService_RemoveEdgeRoute_Handler,
},
{
MethodName: "ListEdgeRoutes",
Handler: _McpAgentService_ListEdgeRoutes_Handler,
},
{
MethodName: "HealthCheck",
Handler: _McpAgentService_HealthCheck_Handler,
},
},
Streams: []grpc.StreamDesc{
{

View File

@@ -134,7 +134,8 @@ func (a *Agent) deployComponent(ctx context.Context, serviceName string, cs *mcp
Error: fmt.Sprintf("allocate route ports: %v", err),
}
}
runSpec.Ports = ports
// Merge explicit ports from the spec with route-allocated ports.
runSpec.Ports = append(cs.GetPorts(), ports...)
runSpec.Env = append(runSpec.Env, env...)
} else {
// Legacy: use ports directly from the spec.

196
internal/agent/edge_rpc.go Normal file
View File

@@ -0,0 +1,196 @@
package agent
import (
"context"
"crypto/x509"
"encoding/pem"
"fmt"
"net"
"os"
"time"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
mcpv1 "git.wntrmute.dev/mc/mcp/gen/mcp/v1"
mcproxy "git.wntrmute.dev/mc/mc-proxy/client/mcproxy"
"git.wntrmute.dev/mc/mcp/internal/registry"
)
// SetupEdgeRoute provisions a TLS cert and registers an mc-proxy route for a
// public hostname. Called by the master on edge nodes.
func (a *Agent) SetupEdgeRoute(ctx context.Context, req *mcpv1.SetupEdgeRouteRequest) (*mcpv1.SetupEdgeRouteResponse, error) {
a.Logger.Info("SetupEdgeRoute", "hostname", req.GetHostname(),
"backend_hostname", req.GetBackendHostname(), "backend_port", req.GetBackendPort())
// Validate required fields.
if req.GetHostname() == "" {
return nil, status.Error(codes.InvalidArgument, "hostname is required")
}
if req.GetBackendHostname() == "" {
return nil, status.Error(codes.InvalidArgument, "backend_hostname is required")
}
if req.GetBackendPort() == 0 {
return nil, status.Error(codes.InvalidArgument, "backend_port is required")
}
if !req.GetBackendTls() {
return nil, status.Error(codes.InvalidArgument, "backend_tls must be true")
}
if a.Proxy == nil {
return nil, status.Error(codes.FailedPrecondition, "mc-proxy not configured")
}
// Resolve the backend hostname to a Tailnet IP.
ips, err := net.LookupHost(req.GetBackendHostname())
if err != nil || len(ips) == 0 {
return nil, status.Errorf(codes.InvalidArgument, "cannot resolve backend_hostname %q: %v", req.GetBackendHostname(), err)
}
backendIP := ips[0]
// Validate the resolved IP is a Tailnet address (100.64.0.0/10).
ip := net.ParseIP(backendIP)
if ip == nil {
return nil, status.Errorf(codes.InvalidArgument, "resolved IP %q is not valid", backendIP)
}
_, tailnet, _ := net.ParseCIDR("100.64.0.0/10")
if !tailnet.Contains(ip) {
return nil, status.Errorf(codes.InvalidArgument, "resolved IP %s is not a Tailnet address", backendIP)
}
backend := fmt.Sprintf("%s:%d", backendIP, req.GetBackendPort())
// Provision TLS cert for the public hostname if cert provisioner is available.
certPath := ""
keyPath := ""
if a.Certs != nil {
if err := a.Certs.EnsureCert(ctx, req.GetHostname(), []string{req.GetHostname()}); err != nil {
return nil, status.Errorf(codes.Internal, "provision cert for %s: %v", req.GetHostname(), err)
}
certPath = a.Proxy.CertPath(req.GetHostname())
keyPath = a.Proxy.KeyPath(req.GetHostname())
} else {
// No cert provisioner — check if certs already exist on disk.
certPath = a.Proxy.CertPath(req.GetHostname())
keyPath = a.Proxy.KeyPath(req.GetHostname())
if _, err := os.Stat(certPath); err != nil {
return nil, status.Errorf(codes.FailedPrecondition, "no cert provisioner and cert not found at %s", certPath)
}
}
// Register the L7 route in mc-proxy.
route := mcproxy.Route{
Hostname: req.GetHostname(),
Backend: backend,
Mode: "l7",
TLSCert: certPath,
TLSKey: keyPath,
BackendTLS: true,
}
if err := a.Proxy.AddRoute(ctx, ":443", route); err != nil {
return nil, status.Errorf(codes.Internal, "add mc-proxy route: %v", err)
}
// Persist the edge route in the registry.
if err := registry.CreateEdgeRoute(a.DB, req.GetHostname(), req.GetBackendHostname(), int(req.GetBackendPort()), certPath, keyPath); err != nil {
a.Logger.Warn("failed to persist edge route", "hostname", req.GetHostname(), "err", err)
}
a.Logger.Info("edge route established",
"hostname", req.GetHostname(), "backend", backend, "cert", certPath)
return &mcpv1.SetupEdgeRouteResponse{}, nil
}
// RemoveEdgeRoute removes an mc-proxy route and cleans up the TLS cert for a
// public hostname. Called by the master on edge nodes.
func (a *Agent) RemoveEdgeRoute(ctx context.Context, req *mcpv1.RemoveEdgeRouteRequest) (*mcpv1.RemoveEdgeRouteResponse, error) {
a.Logger.Info("RemoveEdgeRoute", "hostname", req.GetHostname())
if req.GetHostname() == "" {
return nil, status.Error(codes.InvalidArgument, "hostname is required")
}
if a.Proxy == nil {
return nil, status.Error(codes.FailedPrecondition, "mc-proxy not configured")
}
// Remove the mc-proxy route.
if err := a.Proxy.RemoveRoute(ctx, ":443", req.GetHostname()); err != nil {
a.Logger.Warn("remove mc-proxy route", "hostname", req.GetHostname(), "err", err)
// Continue — clean up cert and registry even if route removal fails.
}
// Remove the TLS cert.
if a.Certs != nil {
if err := a.Certs.RemoveCert(req.GetHostname()); err != nil {
a.Logger.Warn("remove cert", "hostname", req.GetHostname(), "err", err)
}
}
// Remove from registry.
if err := registry.DeleteEdgeRoute(a.DB, req.GetHostname()); err != nil {
a.Logger.Warn("delete edge route from registry", "hostname", req.GetHostname(), "err", err)
}
a.Logger.Info("edge route removed", "hostname", req.GetHostname())
return &mcpv1.RemoveEdgeRouteResponse{}, nil
}
// ListEdgeRoutes returns all edge routes managed by this agent.
func (a *Agent) ListEdgeRoutes(_ context.Context, _ *mcpv1.ListEdgeRoutesRequest) (*mcpv1.ListEdgeRoutesResponse, error) {
a.Logger.Debug("ListEdgeRoutes called")
routes, err := registry.ListEdgeRoutes(a.DB)
if err != nil {
return nil, status.Errorf(codes.Internal, "list edge routes: %v", err)
}
resp := &mcpv1.ListEdgeRoutesResponse{}
for _, r := range routes {
er := &mcpv1.EdgeRoute{
Hostname: r.Hostname,
BackendHostname: r.BackendHostname,
BackendPort: int32(r.BackendPort), //nolint:gosec // port is a small positive integer
}
// Read cert metadata if available.
if r.TLSCert != "" {
if certData, readErr := os.ReadFile(r.TLSCert); readErr == nil { //nolint:gosec // path from registry, not user input
if block, _ := pem.Decode(certData); block != nil {
if cert, parseErr := x509.ParseCertificate(block.Bytes); parseErr == nil {
er.CertSerial = cert.SerialNumber.String()
er.CertExpires = cert.NotAfter.UTC().Format(time.RFC3339)
}
}
}
}
resp.Routes = append(resp.Routes, er)
}
return resp, nil
}
// HealthCheck returns the agent's health status. Called by the master when
// heartbeats are missed.
func (a *Agent) HealthCheck(_ context.Context, _ *mcpv1.HealthCheckRequest) (*mcpv1.HealthCheckResponse, error) {
a.Logger.Debug("HealthCheck called")
st := "healthy"
containers := int32(0)
// Count running containers if the runtime is available.
if a.Runtime != nil {
if list, err := a.Runtime.List(context.Background()); err == nil {
containers = int32(len(list)) //nolint:gosec // container count is small
} else {
st = "degraded"
}
}
return &mcpv1.HealthCheckResponse{
Status: st,
Containers: containers,
}, nil
}

View File

@@ -12,9 +12,9 @@ import (
"google.golang.org/grpc/status"
)
// StopService stops all components of a service.
// StopService stops all components of a service, or a single component if specified.
func (a *Agent) StopService(ctx context.Context, req *mcpv1.StopServiceRequest) (*mcpv1.StopServiceResponse, error) {
a.Logger.Info("StopService", "service", req.GetName())
a.Logger.Info("StopService", "service", req.GetName(), "component", req.GetComponent())
if req.GetName() == "" {
return nil, status.Error(codes.InvalidArgument, "service name is required")
@@ -25,6 +25,13 @@ func (a *Agent) StopService(ctx context.Context, req *mcpv1.StopServiceRequest)
return nil, status.Errorf(codes.Internal, "list components: %v", err)
}
if target := req.GetComponent(); target != "" {
components, err = filterComponents(components, req.GetName(), target)
if err != nil {
return nil, err
}
}
var results []*mcpv1.ComponentResult
for _, c := range components {
containerName := ContainerNameFor(req.GetName(), c.Name)
@@ -59,10 +66,10 @@ func (a *Agent) StopService(ctx context.Context, req *mcpv1.StopServiceRequest)
return &mcpv1.StopServiceResponse{Results: results}, nil
}
// StartService starts all components of a service. If a container already
// exists but is stopped, it is removed first so a fresh one can be created.
// StartService starts all components of a service, or a single component if specified.
// If a container already exists but is stopped, it is removed first so a fresh one can be created.
func (a *Agent) StartService(ctx context.Context, req *mcpv1.StartServiceRequest) (*mcpv1.StartServiceResponse, error) {
a.Logger.Info("StartService", "service", req.GetName())
a.Logger.Info("StartService", "service", req.GetName(), "component", req.GetComponent())
if req.GetName() == "" {
return nil, status.Error(codes.InvalidArgument, "service name is required")
@@ -73,6 +80,13 @@ func (a *Agent) StartService(ctx context.Context, req *mcpv1.StartServiceRequest
return nil, status.Errorf(codes.Internal, "list components: %v", err)
}
if target := req.GetComponent(); target != "" {
components, err = filterComponents(components, req.GetName(), target)
if err != nil {
return nil, err
}
}
var results []*mcpv1.ComponentResult
for _, c := range components {
r := startComponent(ctx, a, req.GetName(), &c)
@@ -82,10 +96,10 @@ func (a *Agent) StartService(ctx context.Context, req *mcpv1.StartServiceRequest
return &mcpv1.StartServiceResponse{Results: results}, nil
}
// RestartService restarts all components of a service by stopping, removing,
// and re-creating each container. The desired_state is not changed.
// RestartService restarts all components of a service, or a single component if specified,
// by stopping, removing, and re-creating each container. The desired_state is not changed.
func (a *Agent) RestartService(ctx context.Context, req *mcpv1.RestartServiceRequest) (*mcpv1.RestartServiceResponse, error) {
a.Logger.Info("RestartService", "service", req.GetName())
a.Logger.Info("RestartService", "service", req.GetName(), "component", req.GetComponent())
if req.GetName() == "" {
return nil, status.Error(codes.InvalidArgument, "service name is required")
@@ -96,6 +110,13 @@ func (a *Agent) RestartService(ctx context.Context, req *mcpv1.RestartServiceReq
return nil, status.Errorf(codes.Internal, "list components: %v", err)
}
if target := req.GetComponent(); target != "" {
components, err = filterComponents(components, req.GetName(), target)
if err != nil {
return nil, err
}
}
var results []*mcpv1.ComponentResult
for _, c := range components {
r := restartComponent(ctx, a, req.GetName(), &c)
@@ -167,6 +188,16 @@ func componentToSpec(service string, c *registry.Component) runtime.ContainerSpe
}
}
// filterComponents returns only the component matching target, or an error if not found.
func filterComponents(components []registry.Component, service, target string) ([]registry.Component, error) {
for _, c := range components {
if c.Name == target {
return []registry.Component{c}, nil
}
}
return nil, status.Errorf(codes.NotFound, "component %q not found in service %q", target, service)
}
// componentExists checks whether a component already exists in the registry.
func componentExists(db *sql.DB, service, name string) bool {
_, err := registry.GetComponent(db, service, name)

View File

@@ -48,6 +48,16 @@ func (p *ProxyRouter) Close() error {
return p.client.Close()
}
// CertPath returns the expected TLS certificate path for a given name.
func (p *ProxyRouter) CertPath(name string) string {
return filepath.Join(p.certDir, name+".pem")
}
// KeyPath returns the expected TLS key path for a given name.
func (p *ProxyRouter) KeyPath(name string) string {
return filepath.Join(p.certDir, name+".key")
}
// GetStatus returns the mc-proxy server status.
func (p *ProxyRouter) GetStatus(ctx context.Context) (*mcproxy.Status, error) {
if p == nil {
@@ -56,6 +66,22 @@ func (p *ProxyRouter) GetStatus(ctx context.Context) (*mcproxy.Status, error) {
return p.client.GetStatus(ctx)
}
// AddRoute adds a single route to mc-proxy.
func (p *ProxyRouter) AddRoute(ctx context.Context, listenerAddr string, route mcproxy.Route) error {
if p == nil {
return fmt.Errorf("mc-proxy not configured")
}
return p.client.AddRoute(ctx, listenerAddr, route)
}
// RemoveRoute removes a single route from mc-proxy.
func (p *ProxyRouter) RemoveRoute(ctx context.Context, listenerAddr, hostname string) error {
if p == nil {
return fmt.Errorf("mc-proxy not configured")
}
return p.client.RemoveRoute(ctx, listenerAddr, hostname)
}
// RegisterRoutes registers all routes for a service component with mc-proxy.
// It uses the assigned host ports from the registry.
func (p *ProxyRouter) RegisterRoutes(ctx context.Context, serviceName string, routes []registry.Route, hostPorts map[string]int) error {

View File

@@ -4,7 +4,10 @@ import (
"context"
"fmt"
"git.wntrmute.dev/mc/mc-proxy/client/mcproxy"
mcpv1 "git.wntrmute.dev/mc/mcp/gen/mcp/v1"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/timestamppb"
)
@@ -44,3 +47,67 @@ func (a *Agent) ListProxyRoutes(ctx context.Context, _ *mcpv1.ListProxyRoutesReq
return resp, nil
}
// AddProxyRoute adds a route to mc-proxy.
func (a *Agent) AddProxyRoute(ctx context.Context, req *mcpv1.AddProxyRouteRequest) (*mcpv1.AddProxyRouteResponse, error) {
if req.GetListenerAddr() == "" {
return nil, status.Error(codes.InvalidArgument, "listener_addr is required")
}
if req.GetHostname() == "" {
return nil, status.Error(codes.InvalidArgument, "hostname is required")
}
if req.GetBackend() == "" {
return nil, status.Error(codes.InvalidArgument, "backend is required")
}
if a.Proxy == nil {
return nil, status.Error(codes.FailedPrecondition, "mc-proxy not configured")
}
route := mcproxy.Route{
Hostname: req.GetHostname(),
Backend: req.GetBackend(),
Mode: req.GetMode(),
BackendTLS: req.GetBackendTls(),
TLSCert: req.GetTlsCert(),
TLSKey: req.GetTlsKey(),
}
if err := a.Proxy.AddRoute(ctx, req.GetListenerAddr(), route); err != nil {
return nil, fmt.Errorf("add route: %w", err)
}
a.Logger.Info("route added",
"listener", req.GetListenerAddr(),
"hostname", req.GetHostname(),
"backend", req.GetBackend(),
"mode", req.GetMode(),
)
return &mcpv1.AddProxyRouteResponse{}, nil
}
// RemoveProxyRoute removes a route from mc-proxy.
func (a *Agent) RemoveProxyRoute(ctx context.Context, req *mcpv1.RemoveProxyRouteRequest) (*mcpv1.RemoveProxyRouteResponse, error) {
if req.GetListenerAddr() == "" {
return nil, status.Error(codes.InvalidArgument, "listener_addr is required")
}
if req.GetHostname() == "" {
return nil, status.Error(codes.InvalidArgument, "hostname is required")
}
if a.Proxy == nil {
return nil, status.Error(codes.FailedPrecondition, "mc-proxy not configured")
}
if err := a.Proxy.RemoveRoute(ctx, req.GetListenerAddr(), req.GetHostname()); err != nil {
return nil, fmt.Errorf("remove route: %w", err)
}
a.Logger.Info("route removed",
"listener", req.GetListenerAddr(),
"hostname", req.GetHostname(),
)
return &mcpv1.RemoveProxyRouteResponse{}, nil
}

View File

@@ -142,4 +142,18 @@ var migrations = []string{
FOREIGN KEY (service, component) REFERENCES components(service, name) ON DELETE CASCADE
);
`,
// Migration 3: service comment
`ALTER TABLE services ADD COLUMN comment TEXT NOT NULL DEFAULT '';`,
// Migration 4: edge routes (v2 — public routes managed by the master)
`CREATE TABLE IF NOT EXISTS edge_routes (
hostname TEXT NOT NULL PRIMARY KEY,
backend_hostname TEXT NOT NULL,
backend_port INTEGER NOT NULL,
tls_cert TEXT NOT NULL DEFAULT '',
tls_key TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);`,
}

View File

@@ -0,0 +1,93 @@
package registry
import (
"database/sql"
"fmt"
"time"
)
// EdgeRoute represents a public edge route managed by the master.
type EdgeRoute struct {
Hostname string
BackendHostname string
BackendPort int
TLSCert string
TLSKey string
CreatedAt time.Time
UpdatedAt time.Time
}
// CreateEdgeRoute inserts or replaces an edge route.
func CreateEdgeRoute(db *sql.DB, hostname, backendHostname string, backendPort int, tlsCert, tlsKey string) error {
_, err := db.Exec(`
INSERT INTO edge_routes (hostname, backend_hostname, backend_port, tls_cert, tls_key, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, datetime('now'), datetime('now'))
ON CONFLICT(hostname) DO UPDATE SET
backend_hostname = excluded.backend_hostname,
backend_port = excluded.backend_port,
tls_cert = excluded.tls_cert,
tls_key = excluded.tls_key,
updated_at = datetime('now')
`, hostname, backendHostname, backendPort, tlsCert, tlsKey)
if err != nil {
return fmt.Errorf("create edge route %s: %w", hostname, err)
}
return nil
}
// GetEdgeRoute returns a single edge route by hostname.
func GetEdgeRoute(db *sql.DB, hostname string) (*EdgeRoute, error) {
var r EdgeRoute
var createdAt, updatedAt string
err := db.QueryRow(`
SELECT hostname, backend_hostname, backend_port, tls_cert, tls_key, created_at, updated_at
FROM edge_routes WHERE hostname = ?
`, hostname).Scan(&r.Hostname, &r.BackendHostname, &r.BackendPort, &r.TLSCert, &r.TLSKey, &createdAt, &updatedAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("get edge route %s: %w", hostname, err)
}
r.CreatedAt, _ = time.Parse("2006-01-02 15:04:05", createdAt)
r.UpdatedAt, _ = time.Parse("2006-01-02 15:04:05", updatedAt)
return &r, nil
}
// ListEdgeRoutes returns all edge routes.
func ListEdgeRoutes(db *sql.DB) ([]*EdgeRoute, error) {
rows, err := db.Query(`
SELECT hostname, backend_hostname, backend_port, tls_cert, tls_key, created_at, updated_at
FROM edge_routes ORDER BY hostname
`)
if err != nil {
return nil, fmt.Errorf("list edge routes: %w", err)
}
defer func() { _ = rows.Close() }()
var routes []*EdgeRoute
for rows.Next() {
var r EdgeRoute
var createdAt, updatedAt string
if err := rows.Scan(&r.Hostname, &r.BackendHostname, &r.BackendPort, &r.TLSCert, &r.TLSKey, &createdAt, &updatedAt); err != nil {
return nil, fmt.Errorf("scan edge route: %w", err)
}
r.CreatedAt, _ = time.Parse("2006-01-02 15:04:05", createdAt)
r.UpdatedAt, _ = time.Parse("2006-01-02 15:04:05", updatedAt)
routes = append(routes, &r)
}
return routes, rows.Err()
}
// DeleteEdgeRoute removes an edge route by hostname.
func DeleteEdgeRoute(db *sql.DB, hostname string) error {
result, err := db.Exec(`DELETE FROM edge_routes WHERE hostname = ?`, hostname)
if err != nil {
return fmt.Errorf("delete edge route %s: %w", hostname, err)
}
n, _ := result.RowsAffected()
if n == 0 {
return fmt.Errorf("edge route %s not found", hostname)
}
return nil
}

View File

@@ -39,6 +39,16 @@ service McpAgentService {
// Proxy routes (query mc-proxy)
rpc ListProxyRoutes(ListProxyRoutesRequest) returns (ListProxyRoutesResponse);
rpc AddProxyRoute(AddProxyRouteRequest) returns (AddProxyRouteResponse);
rpc RemoveProxyRoute(RemoveProxyRouteRequest) returns (RemoveProxyRouteResponse);
// Edge routing (called by master on edge nodes)
rpc SetupEdgeRoute(SetupEdgeRouteRequest) returns (SetupEdgeRouteResponse);
rpc RemoveEdgeRoute(RemoveEdgeRouteRequest) returns (RemoveEdgeRouteResponse);
rpc ListEdgeRoutes(ListEdgeRoutesRequest) returns (ListEdgeRoutesResponse);
// Health (called by master on missed heartbeats)
rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse);
// Logs
rpc Logs(LogsRequest) returns (stream LogsResponse);
@@ -70,6 +80,7 @@ message ServiceSpec {
string name = 1;
bool active = 2;
repeated ComponentSpec components = 3;
string comment = 4;
}
message DeployRequest {
@@ -90,6 +101,7 @@ message ComponentResult {
message StopServiceRequest {
string name = 1;
string component = 2;
}
message StopServiceResponse {
@@ -98,6 +110,7 @@ message StopServiceResponse {
message StartServiceRequest {
string name = 1;
string component = 2;
}
message StartServiceResponse {
@@ -106,6 +119,7 @@ message StartServiceResponse {
message RestartServiceRequest {
string name = 1;
string component = 2;
}
message RestartServiceResponse {
@@ -146,6 +160,7 @@ message ServiceInfo {
string name = 1;
bool active = 2;
repeated ComponentInfo components = 3;
string comment = 4;
}
message ComponentInfo {
@@ -353,3 +368,62 @@ message ListProxyRoutesResponse {
google.protobuf.Timestamp started_at = 3;
repeated ProxyListenerInfo listeners = 4;
}
message AddProxyRouteRequest {
string listener_addr = 1; // e.g. ":443"
string hostname = 2;
string backend = 3;
string mode = 4; // "l4" or "l7"
bool backend_tls = 5;
string tls_cert = 6; // path to TLS cert (required for l7)
string tls_key = 7; // path to TLS key (required for l7)
}
message AddProxyRouteResponse {}
message RemoveProxyRouteRequest {
string listener_addr = 1; // e.g. ":443"
string hostname = 2;
}
message RemoveProxyRouteResponse {}
// --- Edge routes (v2) ---
message SetupEdgeRouteRequest {
string hostname = 1; // public hostname (e.g. "mcq.metacircular.net")
string backend_hostname = 2; // internal .svc.mcp hostname
int32 backend_port = 3; // port on worker's mc-proxy
bool backend_tls = 4; // MUST be true; agent rejects false
}
message SetupEdgeRouteResponse {}
message RemoveEdgeRouteRequest {
string hostname = 1;
}
message RemoveEdgeRouteResponse {}
message ListEdgeRoutesRequest {}
message ListEdgeRoutesResponse {
repeated EdgeRoute routes = 1;
}
message EdgeRoute {
string hostname = 1;
string backend_hostname = 2;
int32 backend_port = 3;
string cert_serial = 4;
string cert_expires = 5; // RFC3339
}
// --- Health check (v2) ---
message HealthCheckRequest {}
message HealthCheckResponse {
string status = 1; // "healthy" or "degraded"
int32 containers = 2;
}