iop/apps/control-plane/cmd/control-plane/main.go

package main

import (
	"context"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"strings"
	"time"

	"github.com/spf13/cobra"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"

	"iop/apps/control-plane/internal/wire"
	"iop/packages/observability"
)

var cfgFile string

type controlPlaneConfig struct {
	Server struct {
		Listen     string `yaml:"listen"`
		WireListen string `yaml:"wire_listen"`
	} `yaml:"server"`
	Database struct {
		URL string `yaml:"url"`
	} `yaml:"database"`
	Redis struct {
		URL       string `yaml:"url"`
		KeyPrefix string `yaml:"key_prefix"`
	} `yaml:"redis"`
	Logging struct {
		Level  string `yaml:"level"`
		Pretty bool   `yaml:"pretty"`
	} `yaml:"logging"`
}

func main() {
	if err := rootCmd().Execute(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

func rootCmd() *cobra.Command {
	root := &cobra.Command{
		Use:   "control-plane",
		Short: "IOP Control Plane - multi-edge management server",
		Long: `control-plane will manage Edge connections, observe Edge state,
adjust Edge configuration, deliver commands to Edge, and receive Edge events.

It controls the system through Edge, not by connecting to or scheduling Node
directly. The main IOP communication layer is protobuf-socket; net/http is kept
for health, readiness, and bootstrap endpoints.`,
	}
	root.PersistentFlags().StringVarP(&cfgFile, "config", "c", "configs/control-plane.yaml", "config file path")
	root.AddCommand(serveCmd())
	return root
}

func serveCmd() *cobra.Command {
	return &cobra.Command{
		Use:   "serve",
		Short: "Start the control-plane server",
		RunE: func(_ *cobra.Command, _ []string) error {
			cfg, err := loadConfig(cfgFile)
			if err != nil {
				return fmt.Errorf("load config: %w", err)
			}
			logger, err := observability.NewLogger(cfg.Logging.Level, cfg.Logging.Pretty)
			if err != nil {
				return fmt.Errorf("logger: %w", err)
			}
			defer func() { _ = logger.Sync() }()

			ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
			defer stop()
			return run(ctx, cfg, logger)
		},
	}
}

func defaultConfig() controlPlaneConfig {
	var cfg controlPlaneConfig
	cfg.Server.Listen = "0.0.0.0:9080"
	cfg.Server.WireListen = "0.0.0.0:19080"
	cfg.Database.URL = "postgres://nomadcode:nomadcode@code-server-postgres:5432/iop-control-plane-local?sslmode=disable"
	cfg.Redis.URL = "redis://code-server-redis:6379/1"
	cfg.Redis.KeyPrefix = "iop:control-plane:local:"
	cfg.Logging.Level = "info"
	return cfg
}

func loadConfig(path string) (controlPlaneConfig, error) {
	cfg := defaultConfig()
	if path != "" {
		b, err := os.ReadFile(path)
		if err != nil {
			return cfg, err
		}
		if err := yaml.Unmarshal(b, &cfg); err != nil {
			return cfg, err
		}
	}
	applyEnvOverrides(&cfg)
	return cfg, nil
}

func applyEnvOverrides(cfg *controlPlaneConfig) {
	if listen := os.Getenv("IOP_LISTEN"); listen != "" {
		cfg.Server.Listen = listen
	}
	if wireListen := os.Getenv("IOP_WIRE_LISTEN"); wireListen != "" {
		cfg.Server.WireListen = wireListen
	}
	if databaseURL := os.Getenv("IOP_DATABASE_URL"); databaseURL != "" {
		cfg.Database.URL = databaseURL
	}
	if redisURL := os.Getenv("IOP_REDIS_URL"); redisURL != "" {
		cfg.Redis.URL = redisURL
	}
	if redisKeyPrefix := os.Getenv("IOP_REDIS_KEY_PREFIX"); redisKeyPrefix != "" {
		cfg.Redis.KeyPrefix = redisKeyPrefix
	}
}

func run(ctx context.Context, cfg controlPlaneConfig, logger *zap.Logger) error {
	mux := http.NewServeMux()
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ok\n"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ready\n"))
	})

	wireEndpoint := wire.Endpoint{Listen: cfg.Server.WireListen}
	logger.Info("control-plane wire endpoint reserved",
		zap.String("protocol", wire.Protocol),
		zap.String("listen", wireEndpoint.Listen),
	)
	if cfg.Database.URL != "" {
		logger.Info("control-plane database configured", databaseLogFields(cfg.Database.URL)...)
	}
	if cfg.Redis.URL != "" {
		logger.Info("control-plane redis configured", redisLogFields(cfg.Redis.URL, cfg.Redis.KeyPrefix)...)
	}
	clientServer, err := wire.NewClientServer(cfg.Server.WireListen, logger)
	if err != nil {
		return fmt.Errorf("wire server: %w", err)
	}
	if err := clientServer.Start(ctx); err != nil {
		return fmt.Errorf("start wire server: %w", err)
	}
	defer func() { _ = clientServer.Stop() }()

	server := &http.Server{
		Addr:              cfg.Server.Listen,
		Handler:           mux,
		ReadHeaderTimeout: 5 * time.Second,
	}

	errCh := make(chan error, 1)
	go func() {
		logger.Info("control-plane http endpoint listening", zap.String("listen", cfg.Server.Listen))
		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
			errCh <- err
			return
		}
		errCh <- nil
	}()

	select {
	case <-ctx.Done():
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		return server.Shutdown(shutdownCtx)
	case err := <-errCh:
		return err
	}
}

func databaseLogFields(rawURL string) []zap.Field {
	return serviceURLLogFields(rawURL)
}

func redisLogFields(rawURL string, keyPrefix string) []zap.Field {
	fields := serviceURLLogFields(rawURL)
	if keyPrefix != "" {
		fields = append(fields, zap.String("key_prefix", keyPrefix))
	}
	return fields
}

func serviceURLLogFields(rawURL string) []zap.Field {
	fields := []zap.Field{zap.Bool("configured", true)}
	parsed, err := url.Parse(rawURL)
	if err != nil {
		return fields
	}
	if parsed.Host != "" {
		fields = append(fields, zap.String("host", parsed.Host))
	}
	if dbName := strings.TrimPrefix(parsed.Path, "/"); dbName != "" {
		fields = append(fields, zap.String("database", dbName))
	}
	return fields
}