From ff22a089481e2eb2ec490b53d76a03038746f210 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Sun, 16 Nov 2025 21:20:08 +0800 Subject: [PATCH] Added support for multi-node config settings --- INSTALL.md | 16 +++--- TODO.md | 12 ++--- cmd/hive/cmd/admin.go | 37 +++++++++++--- cmd/hive/cmd/root.go | 2 +- cmd/hive/cmd/service.go | 66 ++++++++++++++---------- cmd/hive/cmd/templates/hive.toml | 24 +++++---- hive/config/config.go | 88 +++++++------------------------- hive/daemon/daemon.go | 48 ++++------------- hive/daemon/daemon_test.go | 20 +++++++- hive/services/awsgw/awsgw.go | 24 +++++---- hive/services/hive/hive.go | 6 +-- 11 files changed, 162 insertions(+), 181 deletions(-) diff --git a/INSTALL.md b/INSTALL.md index 85de5d97b..f2be3a124 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -53,8 +53,10 @@ Confirm ./bin/hive exists and executable. When running Hive for the first time, run the init function to create the default directories for data, config files and layout required. -```bash -./bin/hive admin init +Example single node installation to get started: + +``` +/bin/hive admin init --region ap-southeast-2 --az ap-southeast-2a --node node1 --nodes 1 ``` Next, set the AWS profile to use `hive` which points to the local environment. @@ -93,7 +95,7 @@ aws ec2 create-key-pair \ | jq -r '.KeyMaterial | rtrimstr("\n")' > ~/.ssh/hive-key ``` -Update permissions to SSH will accept reading the file. +Update permissions for the key for ssh to accept reading the file. ```bash chmod 600 ~/.ssh/hive-key @@ -105,7 +107,7 @@ Next, generate a public key from the specified private key pair. ssh-keygen -y -f ~/.ssh/hive-key > ~/.ssh/hive-key.pub ``` -Validate the new key is available +### Validate the new key is available ```bash aws ec2 describe-key-pairs @@ -164,13 +166,13 @@ Extracted image to: /home/ben/hive/images/debian/12/arm64/disk.raw ✅ Image import complete. Image-ID (AMI): ami-e29fcc65734aec9ea ``` -Next, verify available disk images to confirm the import was successful +Next, verify available disk images to confirm the import was successful, replace `ami-XXX` with the value from the command output. ```bash -aws ec2 describe-images --image-ids ami-e29fcc65734aec9ea +aws ec2 describe-images --image-ids ami-XXX ``` -### Manual import +### Manual AMI import Using this method you can import any OS disk image. For example, download the Debian 12 image from the repository [https://cloud.debian.org/images/cloud/bookworm/latest/](https://cloud.debian.org/images/cloud/bookworm/latest/) diff --git a/TODO.md b/TODO.md index 689968b96..b9225c999 100644 --- a/TODO.md +++ b/TODO.md @@ -26,22 +26,22 @@ Design layout for multi-node configuration. ``` # node1 -hive admin init --region ap-southeast-2 --availability-zone ap-southeast-2a --nodes 3 --bind :8443 --data-dir ~/hive/ +hive admin init --region ap-southeast-2 --az ap-southeast-2a --node node1 --nodes 1 --hive-dir ~/hive/ # node2 -hive admin join --region ap-southeast-2 --availability-zone ap-southeast-2a --host node1.local:8443 --data-dir ~/hive/ +hive admin join --region ap-southeast-2 --az ap-southeast-2a --node node2 --host node1.local:8443 --data-dir ~/hive/ # node3, optionally toggle EBS/EC2/NATs support only -hive admin join --region ap-southeast-2 --availability-zone ap-southeast-2a --host node1.local:8443 --data-dir ~/hive/ --cap ebs,ec2,nats +hive admin join --region ap-southeast-2 --az ap-southeast-2a --node node3 --host node1.local:8443 --data-dir ~/hive/ --cap ebs,ec2,nats ``` -If --host is missing, the join command tries multicast broadcast to find parent node. +If --host is missing, the join command tries multicast broadcast to find parent (leader) node. Set region with `--region ap-southeast-2` which will create a new Hive cluster for the specified region. -Set the availability-zone, for production it is recommended to have at least 2 availability-zones defined per region deployment of Hive. +For production, it is recommended to run Hive on at least three physical nodes. In this minimum setup, Hive S3 and EBS volumes use 2 data shards and 1 parity shard RS(2,1) to distribute each chunk across the cluster. This configuration tolerates a full node failure while adding only 1.5x storage overhead. -Overview: +### Overview: * init creates a cluster-id, node-id, and a short join token, starts a tiny control server on :8443, and writes DNS hint files if available. diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index 9510a1506..b62a10e62 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -140,6 +140,11 @@ func init() { // Flags for admin init adminInitCmd.Flags().Bool("force", false, "Force re-initialization (overwrites existing config)") adminInitCmd.Flags().String("region", "ap-southeast-2", "Mulga region to create") + adminInitCmd.Flags().String("az", "ap-southeast-2a", "Mulga AZ to create") + adminInitCmd.Flags().String("node", "node1", "Node name, increment for additional nodes (default, node1)") + adminInitCmd.Flags().Int("nodes", 3, "Number of nodes to expect for cluster") + adminInitCmd.Flags().String("host", "", "Leader node to join (if not specified, tries multicast discovery)") + adminInitCmd.Flags().Int("port", 4432, "Port to bind cluster services on") imagesImportCmd.Flags().String("tmp-dir", os.TempDir(), "Temporary directory for image import processing") @@ -310,7 +315,7 @@ func runimagesImportCmd(cmd *cobra.Command, args []string) { // Calculate the size manifest.AMIMetadata.Name = fmt.Sprintf("ami-%s-%s-%s", image.Distro, image.Version, image.Arch) - volumeId := viperblock.GenerateVolumeID("ami", manifest.AMIMetadata.Name, appConfig.Predastore.Bucket, time.Now().Unix()) + volumeId := viperblock.GenerateVolumeID("ami", manifest.AMIMetadata.Name, "", time.Now().Unix()) // TODO: Replace with bucket manifest.AMIMetadata.ImageID = volumeId manifest.AMIMetadata.Description = fmt.Sprintf("%s cloud image prepared for Hive", manifest.AMIMetadata.Name) @@ -361,11 +366,11 @@ func runimagesImportCmd(cmd *cobra.Command, args []string) { s3Config := s3.S3Config{ VolumeName: volumeId, VolumeSize: uint64(imageStat.Size()), - Bucket: appConfig.Predastore.Bucket, - Region: appConfig.Predastore.Region, - AccessKey: appConfig.AccessKey, - SecretKey: appConfig.SecretKey, - Host: appConfig.Predastore.Host, + Bucket: appConfig.Nodes[appConfig.Node].Predastore.Bucket, + Region: appConfig.Nodes[appConfig.Node].Predastore.Region, + AccessKey: appConfig.Nodes[appConfig.Node].AccessKey, + SecretKey: appConfig.Nodes[appConfig.Node].SecretKey, + Host: appConfig.Nodes[appConfig.Node].Predastore.Host, } vbConfig := viperblock.VB{ @@ -435,6 +440,11 @@ func runAdminInit(cmd *cobra.Command, args []string) { force, _ := cmd.Flags().GetBool("force") configDir, _ := cmd.Flags().GetString("config-dir") region, _ := cmd.Flags().GetString("region") + az, _ := cmd.Flags().GetString("az") + node, _ := cmd.Flags().GetString("node") + port, _ := cmd.Flags().GetInt("port") + + fmt.Println("Config", az, node, port) // Default config directory if configDir == "" { @@ -529,8 +539,10 @@ func runAdminInit(cmd *cobra.Command, args []string) { {filepath.Join(natsDir, "nats.conf"), natsConfTemplate, "nats/nats.conf"}, } + portStr := fmt.Sprintf("%d", port) + for _, cfg := range configs { - if err := generateConfigFile(cfg.path, cfg.template, accessKey, secretKey, accountID, region, natsToken, hiveRoot); err != nil { + if err := generateConfigFile(cfg.path, cfg.template, accessKey, secretKey, accountID, region, natsToken, hiveRoot, az, node, portStr); err != nil { fmt.Fprintf(os.Stderr, "Error creating %s: %v\n", cfg.name, err) os.Exit(1) } @@ -693,7 +705,7 @@ func generateSelfSignedCert(certPath, keyPath string) error { } // generateConfigFile creates a configuration file from a template -func generateConfigFile(path, templateContent, accessKey, secretKey, accountID, region, natsToken, dataDir string) error { +func generateConfigFile(path, templateContent, accessKey, secretKey, accountID, region, natsToken, dataDir, Az, Node, Port string) error { // Parse the embedded template tmpl, err := template.New("config").Parse(templateContent) if err != nil { @@ -708,6 +720,11 @@ func generateConfigFile(path, templateContent, accessKey, secretKey, accountID, Region string NatsToken string DataDir string + + // Add more fields as needed + Node string + Az string + Port string }{ AccessKey: accessKey, SecretKey: secretKey, @@ -715,6 +732,10 @@ func generateConfigFile(path, templateContent, accessKey, secretKey, accountID, Region: region, NatsToken: natsToken, DataDir: dataDir, + + Node: Node, + Az: Az, + Port: Port, } // Create file with secure permissions diff --git a/cmd/hive/cmd/root.go b/cmd/hive/cmd/root.go index 17b08b96e..106a0904d 100644 --- a/cmd/hive/cmd/root.go +++ b/cmd/hive/cmd/root.go @@ -26,7 +26,7 @@ import ( var ( cfgFile string - appConfig *config.Config + appConfig *config.ClusterConfig ) // rootCmd represents the base command when called without any subcommands diff --git a/cmd/hive/cmd/service.go b/cmd/hive/cmd/service.go index a2589385e..9c6887b44 100644 --- a/cmd/hive/cmd/service.go +++ b/cmd/hive/cmd/service.go @@ -167,7 +167,8 @@ var viperblockStartCmd = &cobra.Command{ fmt.Println("Loading config from:", cfgFile) // TODO: Support ENV vars, CLI, otherwise revert to config.LoadConfig() - appConfig, err := config.LoadConfig(cfgFile) + clusterConfig, err := config.LoadConfig(cfgFile) + nodeConfig := clusterConfig.Nodes[clusterConfig.Node] if err != nil { fmt.Println("Error loading config file:", err) @@ -178,48 +179,51 @@ var viperblockStartCmd = &cobra.Command{ if natsHost != "" { fmt.Println("Overwriting natsHost:", natsHost) - appConfig.NATS.Host = natsHost + nodeConfig.NATS.Host = natsHost } s3Host := viper.GetString("s3-host") if s3Host != "" { fmt.Println("Overwriting s3host:", s3Host) - appConfig.Predastore.Host = s3Host + nodeConfig.Predastore.Host = s3Host } s3Bucket := viper.GetString("s3-bucket") if s3Bucket != "" { fmt.Println("Overwriting s3bucket:", s3Bucket) - appConfig.Predastore.Bucket = s3Bucket + nodeConfig.Predastore.Bucket = s3Bucket } s3Region := viper.GetString("s3-region") if s3Region != "" { fmt.Println("Overwriting s3Region:", s3Region) - appConfig.Predastore.Region = s3Region + nodeConfig.Predastore.Region = s3Region } accessKey := viper.GetString("access-key") if accessKey != "" { fmt.Println("Overwriting access-key:", accessKey) - appConfig.Predastore.AccessKey = accessKey + nodeConfig.Predastore.AccessKey = accessKey } secretKey := viper.GetString("secret-key") if secretKey != "" { fmt.Println("Overwriting secret-key:", secretKey) - appConfig.Predastore.SecretKey = secretKey + nodeConfig.Predastore.SecretKey = secretKey } baseDir := viper.GetString("base-dir") if baseDir != "" { fmt.Println("Overwriting base-dir:", baseDir) - appConfig.Predastore.BaseDir = baseDir + nodeConfig.Predastore.BaseDir = baseDir } + // Apply changes back to cluster config + clusterConfig.Nodes[clusterConfig.Node] = nodeConfig + pluginPath := viper.GetString("plugin-path") if pluginPath == "" { @@ -236,14 +240,14 @@ var viperblockStartCmd = &cobra.Command{ } service, err := service.New("viperblock", &viperblockd.Config{ - NatsHost: appConfig.NATS.Host, + NatsHost: nodeConfig.NATS.Host, PluginPath: pluginPath, - S3Host: appConfig.Predastore.Host, - Bucket: appConfig.Predastore.Bucket, - Region: appConfig.Predastore.Region, - AccessKey: appConfig.Predastore.AccessKey, - SecretKey: appConfig.Predastore.SecretKey, - BaseDir: appConfig.Predastore.BaseDir, + S3Host: nodeConfig.Predastore.Host, + Bucket: nodeConfig.Predastore.Bucket, + Region: nodeConfig.Predastore.Region, + AccessKey: nodeConfig.Predastore.AccessKey, + SecretKey: nodeConfig.Predastore.SecretKey, + BaseDir: nodeConfig.Predastore.BaseDir, }) if err != nil { @@ -363,7 +367,8 @@ var hiveStartCmd = &cobra.Command{ } // TODO: Support ENV vars, CLI, otherwise revert to config.LoadConfig() - appConfig, err := config.LoadConfig(cfgFile) + clusterConfig, err := config.LoadConfig(cfgFile) + nodeConfig := clusterConfig.Nodes[clusterConfig.Node] if err != nil { fmt.Println("Error loading config file:", err) @@ -375,7 +380,7 @@ var hiveStartCmd = &cobra.Command{ if baseDir != "" { fmt.Println("Overwriting base-dir to:", baseDir) - appConfig.BaseDir = baseDir + nodeConfig.BaseDir = baseDir } // Overwrite defaults (CLI first, config second, env third) @@ -383,10 +388,13 @@ var hiveStartCmd = &cobra.Command{ if walDir != "" { fmt.Println("Overwriting wal-dir to:", walDir) - appConfig.WalDir = walDir + nodeConfig.WalDir = walDir } - service, err := service.New("hive", appConfig) + // Apply changes back to cluster config + clusterConfig.Nodes[clusterConfig.Node] = nodeConfig + + service, err := service.New("hive", clusterConfig) if err != nil { fmt.Println("Error starting hive service:", err) @@ -404,7 +412,7 @@ var hiveStopCmd = &cobra.Command{ Run: func(cmd *cobra.Command, args []string) { fmt.Println("Stopping hive service...") - service, err := service.New("hive", &config.Config{}) + service, err := service.New("hive", &config.ClusterConfig{}) if err != nil { fmt.Println("Error stopping hive service:", err) @@ -443,7 +451,8 @@ var awsgwStartCmd = &cobra.Command{ fmt.Println("Loading config from:", cfgFile) // TODO: Support ENV vars, CLI, otherwise revert to config.LoadConfig() - appConfig, err := config.LoadConfig(cfgFile) + clusterConfig, err := config.LoadConfig(cfgFile) + nodeConfig := clusterConfig.Nodes[clusterConfig.Node] if err != nil { fmt.Println("Error loading config file:", err) @@ -454,30 +463,33 @@ var awsgwStartCmd = &cobra.Command{ awsgwHost := viper.GetString("host") if awsgwHost != "" { fmt.Println("Overwriting awsgw host to:", awsgwHost) - appConfig.AWSGW.Host = awsgwHost + nodeConfig.AWSGW.Host = awsgwHost } awsgwTlsCert := viper.GetString("tls-cert") if awsgwTlsCert != "" { fmt.Println("Overwriting awsgw tls-cert to:", awsgwTlsCert) - appConfig.AWSGW.TLSCert = awsgwTlsCert + nodeConfig.AWSGW.TLSCert = awsgwTlsCert } awsgwTlsKey := viper.GetString("tls-key") if awsgwTlsKey != "" { fmt.Println("Overwriting awsgw tls-key to:", awsgwTlsKey) - appConfig.AWSGW.TLSKey = awsgwTlsKey + nodeConfig.AWSGW.TLSKey = awsgwTlsKey } baseDir := viper.GetString("base-dir") if awsgwTlsKey != "" { fmt.Println("Overwriting awsgw base-dir to:", baseDir) - appConfig.BaseDir = baseDir + nodeConfig.BaseDir = baseDir } - service, err := service.New("awsgw", appConfig) + // Apply changes back to cluster config + clusterConfig.Nodes[clusterConfig.Node] = nodeConfig + + service, err := service.New("awsgw", clusterConfig) if err != nil { fmt.Println("Error starting awsgw service:", err) @@ -495,7 +507,7 @@ var awsgwStopCmd = &cobra.Command{ Run: func(cmd *cobra.Command, args []string) { fmt.Println("Stopping awsgw service...") - service, err := service.New("awsgw", &config.Config{}) + service, err := service.New("awsgw", &config.ClusterConfig{}) if err != nil { fmt.Println("Error stopping awsgw service:", err) diff --git a/cmd/hive/cmd/templates/hive.toml b/cmd/hive/cmd/templates/hive.toml index e2ebeb351..0ab695e49 100644 --- a/cmd/hive/cmd/templates/hive.toml +++ b/cmd/hive/cmd/templates/hive.toml @@ -1,32 +1,38 @@ # Domain configuration file in TOML format. version = "1.0" -region = "{{.Region}}" -account_id = "{{.AccountID}}" +epoch = 1 +node = "{{.Node}}" +[nodes.{{.Node}}] +node = "{{.Node}}" +region = "{{.Region}}" +az = "{{.Az}}" + +account_id = "{{.AccountID}}" access_key = "{{.AccessKey}}" secret_key = "{{.SecretKey}}" -[daemon] -host = "0.0.0.0:4432" +[nodes.{{.Node}}.daemon] +host = "0.0.0.0:{{.Port}}" tlskey = "config/server.key" tlscert = "config/server.pem" -[awsgw] +[nodes.{{.Node}}.awsgw] host = "0.0.0.0:9999" tlskey = "config/server.key" tlscert = "config/server.pem" debug = true config = "awsgw/awsgw.toml" -[nats] +[nodes.{{.Node}}.nats] host = "0.0.0.0:4222" -[nats.acl] +[nodes.{{.Node}}.nats.acl] token = "{{.NatsToken}}" -[nats.sub] +[nodes.{{.Node}}.nats.sub] -[predastore] +[nodes.{{.Node}}.predastore] host = "0.0.0.0:8443" region = "{{.Region}}" bucket = "predastore" diff --git a/hive/config/config.go b/hive/config/config.go index 836fd3428..117de6216 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -11,17 +11,31 @@ import ( "github.com/spf13/viper" ) +type ClusterConfig struct { + Epoch uint64 `mapstructure:"epoch"` // bump when leader commits changes + Node string `mapstructure:"node"` // my node name + Version string `mapstructure:"version"` // hive version + Nodes map[string]Config `mapstructure:"nodes"` // full config for every node +} + // Config holds all configuration for the application type Config struct { + // Node config + Node string `mapstructure:"node"` + Host string `mapstructure:"host"` // Unique hostname or IP of this node + Region string `mapstructure:"region"` + AZ string `mapstructure:"az"` + DataDir string `mapstructure:"data_dir"` + Daemon DaemonConfig `mapstructure:"daemon"` NATS NATSConfig `mapstructure:"nats"` Predastore PredastoreConfig `mapstructure:"predastore"` AWSGW AWSGWConfig `mapstructure:"awsgw"` // Authentication + // TODO: Move to more appropriate setting above AccessKey string `mapstructure:"access_key"` SecretKey string `mapstructure:"secret_key"` - Host string `mapstructure:"host"` BaseDir string `mapstructure:"base_dir"` WalDir string `mapstructure:"wal_dir"` } @@ -33,7 +47,7 @@ type AWSGWConfig struct { Config string `mapstructure:"config"` Debug bool `mapstructure:"debug"` - ExpectedNodes int `mapstructure:"expected_nodes"` + ExpectedNodes int `mapstructure:"expected_nodes"` // TODO: Replace with root cluster config } type PredastoreConfig struct { @@ -106,25 +120,6 @@ type EBSDeleteResponse struct { } // EC2, TODO: Move to vm.go or more applicable place -type EC2Response struct { - InstanceID string - Hostname string - Success bool - Status string - Error string -} - -type EC2DescribeRequest struct { - InstanceID string -} - -type EC2DescribeResponse struct { - InstanceID string - Hostname string - Status string - Error string -} - type EC2StartInstancesRequest struct { InstanceID string } @@ -146,45 +141,14 @@ func (ec2StartInstanceResponse EC2StartInstancesResponse) Respond(msg *nats.Msg) msg.Respond(response) -} -func (ec2DescribeResponse EC2DescribeResponse) Respond(msg *nats.Msg) { - - response, err := json.Marshal(ec2DescribeResponse) - if err != nil { - slog.Error("Failed to marshal response", "err", err) - return - } - - msg.Respond(response) - -} - -func (ec2Response EC2Response) Respond(msg *nats.Msg) { - - response, err := json.Marshal(ec2Response) - if err != nil { - slog.Error("Failed to marshal response", "err", err) - return - } - - msg.Respond(response) - } // LoadConfig loads the configuration from file and environment variables -func LoadConfig(configPath string) (*Config, error) { +func LoadConfig(configPath string) (*ClusterConfig, error) { // Set environment variable prefix viper.SetEnvPrefix("HIVE") viper.AutomaticEnv() - // Set default values - //viper.SetDefault("host", "https://localhost:8443/") - //viper.SetDefault("base_dir", "/tmp/vb/") - - //viper.SetDefault("daemon.host", "0.0.0.0:4432") - //viper.SetDefault("nats.host", "0.0.0.0:4222") - //viper.SetDefault("nats.sub.subject", "ec2.>") - // Try to load config file if it exists if configPath != "" { // Check if file exists @@ -202,26 +166,10 @@ func LoadConfig(configPath string) (*Config, error) { } // Create config struct - var config Config + var config ClusterConfig if err := viper.Unmarshal(&config); err != nil { return nil, fmt.Errorf("error unmarshaling config: %w", err) } - // Validate required fields (can be overwritten by env vars) - /* - if config.NATS.Host == "" { - return nil, fmt.Errorf("NATS host is required") - } - - if config.AccessKey == "" { - return nil, fmt.Errorf("access key is required") - } - if config.SecretKey == "" { - return nil, fmt.Errorf("secret key is required") - } - */ - - // fmt.Println("Config: ", config) - return &config, nil } diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 3b5eacaf5..3c94544cb 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -71,6 +71,8 @@ type ResourceManager struct { // Daemon represents the main daemon service type Daemon struct { + node string + clusterConfig *config.ClusterConfig config *config.Config natsConn *nats.Conn resourceMgr *ResourceManager @@ -178,16 +180,21 @@ func NewResourceManager() *ResourceManager { } // NewDaemon creates a new daemon instance -func NewDaemon(cfg *config.Config) *Daemon { +func NewDaemon(cfg *config.ClusterConfig) *Daemon { ctx, cancel := context.WithCancel(context.Background()) // If WalDir is not set, use BaseDir - if cfg.WalDir == "" { - cfg.WalDir = cfg.BaseDir + config := cfg.Nodes[cfg.Node] + if cfg.Nodes[cfg.Node].WalDir == "" { + config.WalDir = config.BaseDir + + cfg.Nodes[cfg.Node] = config } return &Daemon{ - config: cfg, + node: cfg.Node, + clusterConfig: cfg, + config: &config, resourceMgr: NewResourceManager(), ctx: ctx, cancel: cancel, @@ -488,39 +495,6 @@ func (d *Daemon) handleEC2StartInstances(msg *nats.Msg) { } -func (d *Daemon) handleEC2Describe(msg *nats.Msg) { - - var ec2DescribeRequest config.EC2DescribeRequest - - if err := json.Unmarshal(msg.Data, &ec2DescribeRequest); err != nil { - log.Printf("Error unmarshaling EC2 describe request: %v", err) - return - } - - slog.Info("EC2 Describe Request", "instanceId", ec2DescribeRequest.InstanceID) - - var ec2DescribeResponse config.EC2DescribeResponse - - // Check if the instance is running on this node - d.Instances.Mu.Lock() - defer d.Instances.Mu.Unlock() - - instance, ok := d.Instances.VMS[ec2DescribeRequest.InstanceID] - - if !ok { - slog.Error("EC2 Describe Request - Instance not found", "instanceId", ec2DescribeRequest.InstanceID) - ec2DescribeResponse.InstanceID = ec2DescribeRequest.InstanceID - ec2DescribeResponse.Error = fmt.Sprintf("Instance %s not found", ec2DescribeRequest.InstanceID) - ec2DescribeResponse.Respond(msg) - return - } - - ec2DescribeResponse.InstanceID = instance.ID - ec2DescribeResponse.Status = instance.Status - ec2DescribeResponse.Respond(msg) - -} - func (d *Daemon) SendQMPCommand(q *qmp.QMPClient, cmd qmp.QMPCommand, instanceId string) (*qmp.QMPResponse, error) { // Confirm QMP client is initialized diff --git a/hive/daemon/daemon_test.go b/hive/daemon/daemon_test.go index 1d454478d..b965fe3ae 100644 --- a/hive/daemon/daemon_test.go +++ b/hive/daemon/daemon_test.go @@ -55,6 +55,12 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { os.RemoveAll(tmpDir) }) + // New cluster config + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{}, + } + cfg := &config.Config{ BaseDir: tmpDir, WalDir: tmpDir, @@ -76,7 +82,9 @@ func createTestDaemon(t *testing.T, natsURL string) *Daemon { SecretKey: "test-secret-key", } - daemon := NewDaemon(cfg) + clusterCfg.Nodes["node-1"] = *cfg + + daemon := NewDaemon(clusterCfg) // Connect to NATS nc, err := nats.Connect(natsURL) @@ -352,6 +360,12 @@ func TestDaemon_Initialization(t *testing.T) { require.NoError(t, err) defer os.RemoveAll(tmpDir) + // New cluster config + clusterCfg := &config.ClusterConfig{ + Node: "node-1", + Nodes: map[string]config.Config{}, + } + cfg := &config.Config{ BaseDir: tmpDir, WalDir: tmpDir, @@ -362,7 +376,9 @@ func TestDaemon_Initialization(t *testing.T) { SecretKey: "test-secret", } - daemon := NewDaemon(cfg) + clusterCfg.Nodes["node-1"] = *cfg + + daemon := NewDaemon(clusterCfg) assert.NotNil(t, daemon) assert.NotNil(t, daemon.resourceMgr) diff --git a/hive/services/awsgw/awsgw.go b/hive/services/awsgw/awsgw.go index 19564ff9a..d657ad7b6 100644 --- a/hive/services/awsgw/awsgw.go +++ b/hive/services/awsgw/awsgw.go @@ -17,12 +17,12 @@ import ( var serviceName = "awsgw" type Service struct { - Config *config.Config + Config *config.ClusterConfig } func New(cfg interface{}) (svc *Service, err error) { svc = &Service{ - Config: cfg.(*config.Config), + Config: cfg.(*config.ClusterConfig), } return svc, nil } @@ -54,11 +54,13 @@ func (svc *Service) Reload() (err error) { return nil } -func launchService(config *config.Config) (err error) { +func launchService(config *config.ClusterConfig) (err error) { + + nodeConfig := config.Nodes[config.Node] // Connect to NATS for service communication opts := []nats.Option{ - nats.Token(config.NATS.ACL.Token), + nats.Token(nodeConfig.NATS.ACL.Token), nats.ReconnectWait(time.Second), nats.MaxReconnects(-1), // Infinite reconnects nats.DisconnectErrHandler(func(nc *nats.Conn, err error) { @@ -69,26 +71,26 @@ func launchService(config *config.Config) (err error) { }), } - natsConn, err := nats.Connect(config.NATS.Host, opts...) + natsConn, err := nats.Connect(nodeConfig.NATS.Host, opts...) if err != nil { slog.Error("Failed to connect to NATS", "err", err) return err } defer natsConn.Close() - slog.Info("Connected to NATS server", "host", config.NATS.Host) + slog.Info("Connected to NATS server", "host", nodeConfig.NATS.Host) // Append Base dir if config has no leading path - if config.BaseDir != "" && !strings.HasPrefix(config.AWSGW.Config, "/") { - config.AWSGW.Config = fmt.Sprintf("%s/%s", config.BaseDir, config.AWSGW.Config) + if nodeConfig.BaseDir != "" && !strings.HasPrefix(nodeConfig.AWSGW.Config, "/") { + nodeConfig.AWSGW.Config = fmt.Sprintf("%s/%s", nodeConfig.BaseDir, nodeConfig.AWSGW.Config) } // Create gateway with NATS connection gw := gateway.GatewayConfig{ - Debug: config.AWSGW.Debug, + Debug: nodeConfig.AWSGW.Debug, DisableLogging: false, NATSConn: natsConn, - Config: config.AWSGW.Config, + Config: nodeConfig.AWSGW.Config, } app := gw.SetupRoutes() @@ -98,7 +100,7 @@ func launchService(config *config.Config) (err error) { return err } - log.Fatal(app.ListenTLS(config.AWSGW.Host, config.AWSGW.TLSCert, config.AWSGW.TLSKey)) + log.Fatal(app.ListenTLS(nodeConfig.AWSGW.Host, nodeConfig.AWSGW.TLSCert, nodeConfig.AWSGW.TLSKey)) return nil diff --git a/hive/services/hive/hive.go b/hive/services/hive/hive.go index aa4822e0b..c5aa2f7e5 100644 --- a/hive/services/hive/hive.go +++ b/hive/services/hive/hive.go @@ -12,12 +12,12 @@ import ( var serviceName = "hive" type Service struct { - Config *config.Config + Config *config.ClusterConfig } func New(cfg interface{}) (svc *Service, err error) { svc = &Service{ - Config: cfg.(*config.Config), + Config: cfg.(*config.ClusterConfig), } return svc, nil } @@ -49,7 +49,7 @@ func (svc *Service) Reload() (err error) { return nil } -func launchService(config *config.Config) (err error) { +func launchService(config *config.ClusterConfig) (err error) { d := daemon.NewDaemon(config) slog.Info("Starting Hive daemon ...")