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

package main

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

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

	"iop/apps/control-plane/internal/credentialseal"
	"iop/packages/go/config"
	"iop/packages/go/observability"
)

var cfgFile string

type controlPlaneConfig struct {
	Server struct {
		Listen         string         `yaml:"listen"`
		WireListen     string         `yaml:"wire_listen"`
		EdgeWireListen string         `yaml:"edge_wire_listen"`
		EdgeWireTLS    config.TLSConf `yaml:"edge_wire_tls"`
	} `yaml:"server"`
	Database struct {
		URL string `yaml:"url"`
	} `yaml:"database"`
	CredentialEncryption credentialseal.FileConfig `yaml:"credential_encryption"`
	CredentialPlane      struct {
		Enabled          bool           `yaml:"enabled"`
		HTTPS            config.TLSConf `yaml:"https"`
		IssuerKeyID      string         `yaml:"issuer_key_id"`
		IssuerPrivateKey string         `yaml:"issuer_private_key"`
		LeaseTTLSeconds  int            `yaml:"lease_ttl_seconds"`
		LeaseCacheSize   int            `yaml:"lease_cache_size"`
	} `yaml:"credential_plane"`
	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"`
	Metrics struct {
		Port int `yaml:"port"`
	} `yaml:"metrics"`
}

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(), principalCmd())
	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:18000"
	cfg.Server.WireListen = "0.0.0.0:19080"
	cfg.Server.EdgeWireListen = "0.0.0.0:19081"
	cfg.Database.URL = ""
	cfg.Redis.URL = ""
	cfg.Redis.KeyPrefix = "iop:control-plane:"
	cfg.Logging.Level = "info"
	cfg.Metrics.Port = 0
	cfg.CredentialPlane.LeaseTTLSeconds = 30
	cfg.CredentialPlane.LeaseCacheSize = 256
	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)
	if cfg.CredentialPlane.Enabled {
		if strings.TrimSpace(cfg.Database.URL) == "" || strings.TrimSpace(cfg.CredentialEncryption.KeyFile) == "" {
			return cfg, fmt.Errorf("credential_plane requires database.url and credential_encryption")
		}
		if !cfg.CredentialPlane.HTTPS.Enabled {
			return cfg, fmt.Errorf("credential_plane requires HTTPS")
		}
		if err := cfg.CredentialPlane.HTTPS.ValidateHTTPServer(); err != nil {
			return cfg, fmt.Errorf("credential_plane.https: %w", err)
		}
		if !cfg.Server.EdgeWireTLS.Enabled {
			return cfg, fmt.Errorf("credential_plane requires server.edge_wire_tls")
		}
		if err := cfg.Server.EdgeWireTLS.ValidateServer("edge"); err != nil {
			return cfg, fmt.Errorf("server.edge_wire_tls: %w", err)
		}
		if strings.TrimSpace(cfg.CredentialPlane.IssuerKeyID) == "" || strings.TrimSpace(cfg.CredentialPlane.IssuerPrivateKey) == "" {
			return cfg, fmt.Errorf("credential_plane issuer_key_id and issuer_private_key are required")
		}
		if cfg.CredentialPlane.LeaseTTLSeconds < config.MinCredentialLeaseTTLSeconds || cfg.CredentialPlane.LeaseTTLSeconds > config.MaxCredentialLeaseTTLSeconds {
			return cfg, fmt.Errorf("credential_plane.lease_ttl_seconds must be between %d and %d", config.MinCredentialLeaseTTLSeconds, config.MaxCredentialLeaseTTLSeconds)
		}
		if cfg.CredentialPlane.LeaseCacheSize < config.MinCredentialCacheSize || cfg.CredentialPlane.LeaseCacheSize > config.MaxCredentialCacheSize {
			return cfg, fmt.Errorf("credential_plane.lease_cache_size must be between %d and %d", config.MinCredentialCacheSize, config.MaxCredentialCacheSize)
		}
	}
	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 edgeWireListen := os.Getenv("IOP_EDGE_WIRE_LISTEN"); edgeWireListen != "" {
		cfg.Server.EdgeWireListen = edgeWireListen
	}
	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
	}
	if metricsPort := os.Getenv("IOP_METRICS_PORT"); metricsPort != "" {
		var port int
		if _, err := fmt.Sscanf(metricsPort, "%d", &port); err == nil {
			cfg.Metrics.Port = port
		}
	}
}

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
}