Assemble isolated Dispatcher runtime command

2026-09-30 08:21:08 +08:00
parent 5da33c0d5b
commit 99a06b740f
3 changed files with 254 additions and 0 deletions
@@ -0,0 +1,170 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"time"
+
+	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
+	"git.ipao.vip/rogee/go-sip/internal/config"
+	"git.ipao.vip/rogee/go-sip/internal/dispatcher"
+	"git.ipao.vip/rogee/go-sip/internal/mq"
+	"git.ipao.vip/rogee/go-sip/internal/oss"
+	"git.ipao.vip/rogee/go-sip/internal/rpc"
+	"git.ipao.vip/rogee/go-sip/internal/store"
+	"github.com/google/uuid"
+	"github.com/spf13/cobra"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+func newCurrentDispatcherCommand() *cobra.Command {
+	var mode string
+	command := &cobra.Command{
+		Use: "dispatcher", Short: "Run the isolated Dispatcher", Args: cobra.NoArgs,
+		SilenceUsage: true,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return runCurrentDispatcher(cmd.Context(), mode)
+		},
+	}
+	command.Flags().StringVar(&mode, "mode", "mock", "isolated Mock mode only")
+	return command
+}
+
+// runCurrentDispatcher does not publish or consume until the predeclared MQ
+// topology, Agent session and approved SIP load have all been verified.
+func runCurrentDispatcher(ctx context.Context, mode string) (result error) {
+	settings, err := config.LoadDispatcherRuntimeEnvironment(mode)
+	if err != nil {
+		return err
+	}
+	agentEndpoint, err := config.LoadCurrentMockAgentEndpoint(settings.AgentEndpointsFile)
+	if err != nil {
+		return err
+	}
+	ossConfiguration, err := config.LoadCurrentOSSConfig(settings.OSSConfigFile, settings.DispatcherID)
+	if err != nil {
+		return err
+	}
+	ca, err := readAgentPEM("MTLS_CA_FILE", settings.CAFile)
+	if err != nil {
+		return err
+	}
+	cert, err := readAgentPEM("MTLS_CERT_FILE", settings.CertFile)
+	if err != nil {
+		return err
+	}
+	key, err := readAgentPEM("MTLS_KEY_FILE", settings.KeyFile)
+	if err != nil {
+		return err
+	}
+	serverTLS, err := rpc.NewServerTLSConfig(ca, cert, key)
+	if err != nil {
+		return errors.New("Dispatcher mTLS listener certificate is invalid")
+	}
+	clientTLS, err := rpc.NewClientTLSConfig(ca, cert, key, agentEndpoint.ServerName)
+	if err != nil {
+		return errors.New("Dispatcher mTLS Agent certificate configuration is invalid")
+	}
+	httpClient := localMockHTTPClient()
+	httpClient.Timeout = 10 * time.Second
+	_, reader, err := dispatcherConfigurationClient(mode, httpClient)
+	if err != nil {
+		return err
+	}
+	ossClient, err := oss.NewClient(ossConfiguration)
+	if err != nil {
+		return fmt.Errorf("initialize bounded OSS grant signer: %w", err)
+	}
+	database, err := store.OpenCurrent(settings.SQLitePath)
+	if err != nil {
+		return err
+	}
+	defer func() { result = errors.Join(result, database.Close()) }()
+	if err := database.CloseAdmission(settings.DispatcherID); err != nil {
+		return fmt.Errorf("close task admission before connecting external adapters: %w", err)
+	}
+	broker, err := mq.OpenCurrent(settings.RabbitMQURL, settings.DispatcherID, 1)
+	if err != nil {
+		return err
+	}
+	defer func() { result = errors.Join(result, broker.Close()) }()
+	connection, err := grpc.NewClient(agentEndpoint.Address, grpc.WithTransportCredentials(credentials.NewTLS(clientTLS)))
+	if err != nil {
+		return errors.New("Dispatcher cannot create verified Agent connection")
+	}
+	defer func() { result = errors.Join(result, connection.Close()) }()
+	agentClient := agentpb.NewAgentControlServiceClient(connection)
+	coordinator := dispatcher.NewAgentCoordinator(time.Now)
+	if err := coordinator.Register(agentEndpoint.AgentID, agentClient); err != nil {
+		return err
+	}
+	probeCtx, cancelProbe := context.WithTimeout(ctx, 5*time.Second)
+	status, err := coordinator.Probe(probeCtx, agentEndpoint.AgentID, agentEndpoint.CellID)
+	if err != nil {
+		cancelProbe()
+		return fmt.Errorf("probe assigned Agent before admission: %w", err)
+	}
+	epoch, err := uuid.NewRandom()
+	if err != nil {
+		cancelProbe()
+		return fmt.Errorf("create Dispatcher epoch: %w", err)
+	}
+	if _, err := coordinator.Activate(probeCtx, settings.DispatcherID, agentEndpoint.AgentID, agentEndpoint.CellID, status.BootId, epoch.String(), 0); err != nil {
+		cancelProbe()
+		return fmt.Errorf("activate assigned Agent before admission: %w", err)
+	}
+	cancelProbe()
+	originator := &dispatcher.ApprovedOriginator{
+		DispatcherID: settings.DispatcherID,
+		Client:       agentClient,
+		Meta: func(callCtx context.Context) (*agentpb.RequestMeta, error) {
+			return coordinator.ApprovedMeta(callCtx, agentEndpoint.AgentID)
+		},
+	}
+	worker := &dispatcher.CurrentRuntime{
+		Broker: broker,
+		Bootstrap: dispatcher.CurrentBootstrap{
+			DispatcherID: settings.DispatcherID, Client: reader, Store: database, VerifySIP: originator.VerifySIP,
+		},
+		Execute: dispatcher.CurrentExecuteController{
+			DispatcherID: settings.DispatcherID, Store: database, Originator: originator, Publisher: broker, Now: time.Now,
+		},
+		Control: dispatcher.CurrentControlController{
+			DispatcherID: settings.DispatcherID, Store: database, Client: reader, Agent: originator, VerifySIP: originator.VerifySIP, Now: time.Now,
+		},
+		PollInterval: time.Second, DiscoveryInterval: time.Minute, Logger: slog.Default(),
+	}
+	listener, err := net.Listen("tcp", settings.Listen)
+	if err != nil {
+		return fmt.Errorf("Dispatcher cannot listen on configured Mock address: %w", err)
+	}
+	server := grpc.NewServer(grpc.Creds(credentials.NewTLS(serverTLS)))
+	agentpb.RegisterAgentControlServiceServer(server, &rpc.RecordingServer{
+		DispatcherID: settings.DispatcherID, Store: database, OSS: ossClient,
+		TrustedFingerprints: settings.PeerFingerprints, AuthorizeSession: coordinator.AuthorizeInboundMeta,
+	})
+	serveCtx, cancelServe := context.WithCancel(ctx)
+	defer cancelServe()
+	serverErrors := make(chan error, 1)
+	go func() {
+		serverErrors <- server.Serve(listener)
+		cancelServe()
+	}()
+	runtimeErr := worker.Serve(serveCtx)
+	server.GracefulStop()
+	serverErr := <-serverErrors
+	if serverErr != nil && !errors.Is(serverErr, grpc.ErrServerStopped) {
+		return errors.Join(runtimeErr, fmt.Errorf("Dispatcher recording listener stopped: %w", serverErr))
+	}
+	if runtimeErr != nil {
+		return runtimeErr
+	}
+	if ctx.Err() == nil {
+		return errors.New("Dispatcher stopped without process shutdown")
+	}
+	return nil
+}
@@ -0,0 +1,83 @@
+package main
+
+import (
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestCurrentDispatcherCommandRejectsOldFlagsAndModesBeforeResources(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		args []string
+		want string
+	}{
+		{"real", []string{"--mode", "real"}, "Mock"},
+		{"mixed", []string{"--mode", "mixed"}, "Mock"},
+		{"old config", []string{"--config", "/nonexistent"}, "unknown flag"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			database := filepath.Join(t.TempDir(), "dispatcher.sqlite")
+			t.Setenv("DISPATCHER_SQLITE_PATH", database)
+			command := newCurrentDispatcherCommand()
+			command.SetOut(io.Discard)
+			command.SetErr(io.Discard)
+			command.SetArgs(tc.args)
+			if err := command.Execute(); err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("unapproved Dispatcher entry was admitted: %v", err)
+			}
+			if _, err := os.Stat(database); !os.IsNotExist(err) {
+				t.Fatalf("rejected entry opened business state: %v", err)
+			}
+		})
+	}
+}
+
+func TestCurrentDispatcherCommandValidatesDeploymentBeforeOpeningSQLite(t *testing.T) {
+	root := t.TempDir()
+	endpointFile := filepath.Join(root, "agent-endpoints.json")
+	ossFile := filepath.Join(root, "oss.json")
+	database := filepath.Join(root, "dispatcher.sqlite")
+	for name, value := range map[string]string{
+		"DISPATCHER_ID":                   "c046b893-8628-4589-ae50-619d049248a6",
+		"DISPATCHER_SECRET_KEY":           "isolated-placeholder",
+		"SAAS_BASE_URL":                   "http://127.0.0.1:18770",
+		"RABBITMQ_URL":                    "amqp://127.0.0.1:18771",
+		"DISPATCHER_SQLITE_PATH":          database,
+		"DISPATCHER_GRPC_LISTEN":          "127.0.0.1:0",
+		"DISPATCHER_AGENT_ENDPOINTS_FILE": endpointFile,
+		"DISPATCHER_OSS_CONFIG_FILE":      ossFile,
+		"MTLS_CA_FILE":                    filepath.Join(root, "ca.pem"),
+		"MTLS_CERT_FILE":                  filepath.Join(root, "dispatcher.pem"),
+		"MTLS_KEY_FILE":                   filepath.Join(root, "dispatcher.key"),
+		"MTLS_PEER_CERT_FINGERPRINTS":     strings.Repeat("a", 64),
+	} {
+		t.Setenv(name, value)
+	}
+	if err := os.WriteFile(endpointFile, []byte(`[{"agent_id":"agent-mock","cell_id":"cell-mock","address":"saas.example.invalid:443","server_name":"agent.local"}]`), 0600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(ossFile, []byte(`{"dispatcher_id":"22222222-2222-4222-8222-222222222222","oss":{}}`), 0600); err != nil {
+		t.Fatal(err)
+	}
+	check := func(want string) {
+		t.Helper()
+		command := newCurrentDispatcherCommand()
+		command.SetOut(io.Discard)
+		command.SetErr(io.Discard)
+		command.SetArgs([]string{"--mode", "mock"})
+		if err := command.Execute(); err == nil || !strings.Contains(err.Error(), want) || strings.Contains(err.Error(), "saas.example.invalid") {
+			t.Fatalf("unapproved Dispatcher deployment was admitted or echoed: %v", err)
+		}
+		if _, err := os.Stat(database); !os.IsNotExist(err) {
+			t.Fatalf("rejected deployment opened SQLite: %v", err)
+		}
+	}
+	check("local")
+	if err := os.WriteFile(endpointFile, []byte(`[{"agent_id":"agent-mock","cell_id":"cell-mock","address":"127.0.0.1:19090","server_name":"agent.local"}]`), 0600); err != nil {
+		t.Fatal(err)
+	}
+	check("not assigned")
+}