From f2a04dd4c268cd3c47a8d430936df5d75709e275 Mon Sep 17 00:00:00 2001 From: Ben Duncan <5178243+benduncan@users.noreply.github.com> Date: Fri, 21 Nov 2025 23:06:14 +1100 Subject: [PATCH] Added multi-node support and updated INSTALLER. Fix vm.go bug on qemu exec with args error --- INSTALL.md | 127 ++++- TODO.md | 2 +- cmd/hive/cmd/admin.go | 593 ++++++++++++----------- cmd/hive/cmd/service.go | 24 +- cmd/hive/cmd/templates/hive.toml | 8 +- cmd/hive/cmd/templates/nats.conf | 43 +- go.mod | 4 +- go.sum | 2 + hive/admin/admin.go | 342 +++++++++++++ hive/config/config.go | 40 +- hive/daemon/daemon.go | 286 ++++++++++- hive/services/hive/hive.go | 12 +- hive/services/nats/nats.go | 61 ++- hive/services/viperblockd/viperblockd.go | 24 +- hive/vm/vm.go | 2 +- scripts/create-multi-node.sh | 67 +++ scripts/start-dev.sh | 49 +- scripts/start-multi-nats.sh | 3 + scripts/start-multi-node.sh | 12 + scripts/stop-dev.sh | 35 +- scripts/stop-multi-node.sh | 17 + 21 files changed, 1362 insertions(+), 391 deletions(-) create mode 100644 hive/admin/admin.go create mode 100755 scripts/create-multi-node.sh create mode 100644 scripts/start-multi-nats.sh create mode 100755 scripts/start-multi-node.sh create mode 100755 scripts/stop-multi-node.sh diff --git a/INSTALL.md b/INSTALL.md index 646bbb040..de0b38c5c 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -23,7 +23,7 @@ go version go version go1.25.4 linux/amd64 ``` -Hive provided AWS API/SDK layer functionality, which requires the AWS CLI tool to be installed to interface with the system. +Hive provides AWS API/SDK layer functionality, which requires the AWS CLI tool to be installed to interface with the system. ```bash sudo apt install awscli @@ -459,4 +459,127 @@ On success no data will be returned, since the instance is no longer available. # Multi-node configuration -Details for setting up a multi-node configuration of Mulga for development use. +To create a simulate multi-node installation on a single server (e.g node1, node2, node3) review the script `./scripts/create-multi-node.sh` + +Specifically this will create 3 IPs on a specified ethernet adapter + +```bash +ip addr add 10.11.12.1/24 dev eth0 +ip addr add 10.11.12.2/24 dev eth0 +ip addr add 10.11.12.3/24 dev eth0 +``` + +Next, `hive` will be installed and initiated in `$HOME/node1` for configuration files. + +```bash +# Initialize node1 +./bin/hive admin init \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node1 \ +--bind 10.11.12.1 \ +--cluster-bind 10.11.12.1 \ +--port 4432 \ +--hive-dir ~/node1/ \ +--config-dir ~/node1/config/ +``` + +Two other nodes will be created, joining node1 to exchange configuration files and identify as a new node. + +```bash +# Join cluster +./bin/hive admin join \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node2 \ +--bind 10.11.12.2 \ +--cluster-bind 10.11.12.2 \ +--cluster-routes 10.11.12.1:4248 \ +--host 10.11.12.1:4432 \ +--data-dir ~/node2/ \ +--config-dir ~/node2/config/ \ + +# Start node2 +./scripts/start-dev.sh ~/node2/ + +echo "Node3 Setup:" + +./bin/hive admin join \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node3 \ +--bind 10.11.12.3 \ +--cluster-bind 10.11.12.3 \ +--cluster-routes 10.11.12.1:4248 \ +--host 10.11.12.1:4432 \ +--data-dir ~/node3/ \ +--config-dir ~/node3/config/ + +./scripts/start-dev.sh ~/node3/ +``` + +Once configured, each node will have it's own storage for config, data and state in `~/node[1,2,3]` to simulate a unique node and network. + +Once the multi-node is configured, follow the original installation instructions above to: + +* Import an SSH key +* Clone an AMI +* Launch an instance + +Note, when using the AWS CLI tool you must specify the IP address of the node (10.11.12.1, 10.11.12.2, or 10.11.12.3 ) and ignore SSL verification for development purposes, specifically appending the arguments `--endpoint-url https://10.11.12.3:9999/ --no-verify-ssl` + +```bash +aws --endpoint-url https://10.11.12.3:9999/ --no-verify-ssl ec2 describe-instances --insta +nce-ids i-9f5f648adc57ea46d + +urllib3/connectionpool.py:1064: InsecureRequestWarning: Unverified HTTPS request is being made to host '10.11.12.3'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/1.26.x/advanced-usage.html#ssl-warnings +``` + +```json +{ + "Reservations": [ + { + "ReservationId": "r-9d26866c3b0d53bf4", + "OwnerId": "123456789012", + "Instances": [ + { + "InstanceId": "i-9f5f648adc57ea46d", + "ImageId": "ami-d0de73dd18fa33ac9", + "State": { + "Code": 16, + "Name": "running" + }, + "KeyName": "hive-key", + "InstanceType": "t3.micro", + "LaunchTime": "2025-11-21T10:36:42.834000+00:00" + } + ] + } + ] +``` + +For debugging, reference the node unique configuration files, e.g + +```bash +tail -n 100 ~/node3/logs/hive.log +``` + +```bash +... +2025/11/21 22:59:32 Received message on subject: ec2.DescribeInstances +2025/11/21 22:59:32 Message data: {"DryRun":null,"Filters":null,"InstanceIds":["i-9f5f648adc57ea46d"],"MaxResults":null,"NextToken":null} +2025/11/21 22:59:32 INFO Processing DescribeInstances request from this node +2025/11/21 22:59:32 INFO handleEC2DescribeInstances completed count=1 +... +``` + +### Stop multi-node instance + +To stop a simulated multi-node instance + +```bash +./scripts/stop-multi-node.sh +``` + +Note if multiple EC2 instances are running, it make take a few minutes to gracefully terminate the instance, unmount the attached EBS volume (via NBD) and push the write-ahead-log (WAL) to the S3 server (predastore) + diff --git a/TODO.md b/TODO.md index b9225c999..0037fb305 100644 --- a/TODO.md +++ b/TODO.md @@ -26,7 +26,7 @@ Design layout for multi-node configuration. ``` # node1 -hive admin init --region ap-southeast-2 --az ap-southeast-2a --node node1 --nodes 1 --hive-dir ~/hive/ +hive admin init --region ap-southeast-2 --az ap-southeast-2a --node node1 --nodes 3 --hive-dir ~/hive/ # node2 hive admin join --region ap-southeast-2 --az ap-southeast-2a --node node2 --host node1.local:8443 --data-dir ~/hive/ diff --git a/cmd/hive/cmd/admin.go b/cmd/hive/cmd/admin.go index b62a10e62..470d07884 100644 --- a/cmd/hive/cmd/admin.go +++ b/cmd/hive/cmd/admin.go @@ -16,32 +16,30 @@ limitations under the License. package cmd import ( - "crypto/rand" - "crypto/rsa" - "crypto/x509" - "crypto/x509/pkix" + "bytes" + "crypto/tls" _ "embed" - "encoding/base64" "encoding/json" - "encoding/pem" "fmt" - "math/big" + "io" "net" + "net/http" "os" "path" "path/filepath" "sort" - "text/template" + "strings" "time" + "github.com/mulgadc/hive/hive/admin" "github.com/mulgadc/hive/hive/config" "github.com/mulgadc/hive/hive/utils" "github.com/mulgadc/viperblock/viperblock" "github.com/mulgadc/viperblock/viperblock/backends/s3" vutils "github.com/mulgadc/viperblock/viperblock/utils" + "github.com/pelletier/go-toml/v2" "github.com/pterm/pterm" "github.com/spf13/cobra" - "gopkg.in/ini.v1" ) //go:embed templates/hive.toml @@ -83,6 +81,15 @@ configuration files in ~/hive/config.`, Run: runAdminInit, } +var adminJoinCmd = &cobra.Command{ + Use: "join", + Short: "Join an existing Hive cluster", + Long: `Join an existing Hive cluster by connecting to a leader node and retrieving +the cluster configuration. This command will configure the local node to join +the cluster and participate in distributed operations.`, + Run: runAdminJoin, +} + var imagesCmd = &cobra.Command{ Use: "images", Short: "Manage OS images", @@ -125,6 +132,7 @@ hive admin images import --file /path/to/image --distro debian --version 12 --ar func init() { rootCmd.AddCommand(adminCmd) adminCmd.AddCommand(adminInitCmd) + adminCmd.AddCommand(adminJoinCmd) adminCmd.AddCommand(imagesCmd) imagesCmd.AddCommand(imagesImportCmd) @@ -145,6 +153,22 @@ func init() { 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") + adminInitCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.1 for multi-node)") + adminInitCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") + adminInitCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)") + + // Flags for admin join + adminJoinCmd.Flags().String("region", "ap-southeast-2", "Region for this node") + adminJoinCmd.Flags().String("az", "ap-southeast-2a", "Availability zone for this node") + adminJoinCmd.Flags().String("node", "", "Node name (required)") + adminJoinCmd.Flags().String("host", "", "Leader node host:port (e.g., node1.local:4432) (required)") + adminJoinCmd.Flags().String("data-dir", "", "Data directory for this node (default: ~/hive)") + adminJoinCmd.Flags().Int("port", 4432, "Port to bind cluster services on") + adminJoinCmd.Flags().String("bind", "0.0.0.0", "IP address to bind services to (e.g., 10.11.12.2 for multi-node on single host)") + adminJoinCmd.Flags().String("cluster-bind", "", "IP address to bind NATS cluster services to (e.g., 10.11.12.1 for multi-node)") + adminJoinCmd.Flags().String("cluster-routes", "", "NATS cluster hosts for routing specify multiple with comma (e.g., 10.11.12.1:4248,10.11.12.2:4248 for multi-node)") + adminJoinCmd.MarkFlagRequired("node") + adminJoinCmd.MarkFlagRequired("host") imagesImportCmd.Flags().String("tmp-dir", os.TempDir(), "Temporary directory for image import processing") @@ -185,7 +209,7 @@ func runimagesImportCmd(cmd *cobra.Command, args []string) { // Check the base dir has our images path, and correctlty init imageDir := fmt.Sprintf("%s/images", baseDir) - if !fileExists(imageDir) { + if !admin.FileExists(imageDir) { fmt.Fprintf(os.Stderr, "Image directory does not exist. Base path specified correctly? %s", imageDir) os.Exit(1) } @@ -268,7 +292,7 @@ func runimagesImportCmd(cmd *cobra.Command, args []string) { fmt.Printf("Downloading image %s to %s\n", image.URL, imageFile) // If image path exists, skip - if fileExists(imageFile) && !forceCmd { + if admin.FileExists(imageFile) && !forceCmd { fmt.Printf("Image file already exists, skipping download, use --force to overwrite: %s\n", imageFile) } else { @@ -436,15 +460,33 @@ func runimagesListCmd(cmd *cobra.Command, args []string) { pterm.Println("hive admin images import --name ") } +// TODO: Move all logic to a module, use minimal application logic in viper commands func runAdminInit(cmd *cobra.Command, args []string) { force, _ := cmd.Flags().GetBool("force") configDir, _ := cmd.Flags().GetString("config-dir") + hiveRoot, _ := cmd.Flags().GetString("hive-dir") region, _ := cmd.Flags().GetString("region") az, _ := cmd.Flags().GetString("az") node, _ := cmd.Flags().GetString("node") port, _ := cmd.Flags().GetInt("port") + bindIP, _ := cmd.Flags().GetString("bind") + clusterBind, _ := cmd.Flags().GetString("cluster-bind") + //clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") + // := strings.Split(clusterRoutesStr, ",") - fmt.Println("Config", az, node, port) + // Validate IP address format + if net.ParseIP(bindIP) == nil { + fmt.Fprintf(os.Stderr, "❌ Error: Invalid IP address for --bind: %s\n", bindIP) + os.Exit(1) + } + + // Validate port range + if port < 1 || port > 65535 { + fmt.Fprintf(os.Stderr, "❌ Error: Port must be between 1 and 65535, got: %d\n", port) + os.Exit(1) + } + + fmt.Printf("Initializing Hive with bind IP: %s, port: %d\n", bindIP, port) // Default config directory if configDir == "" { @@ -461,7 +503,7 @@ func runAdminInit(cmd *cobra.Command, args []string) { // Check if already initialized hiveTomlPath := filepath.Join(configDir, "hive.toml") - if !force && fileExists(hiveTomlPath) { + if !force && admin.FileExists(hiveTomlPath) { fmt.Println("⚠️ Hive already initialized!") fmt.Printf("Config file exists: %s\n", hiveTomlPath) fmt.Println("\nTo re-initialize, run with --force flag:") @@ -477,9 +519,9 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Printf("✅ Created config directory: %s\n", configDir) // Generate AWS credentials - accessKey := generateAWSAccessKey() - secretKey := generateAWSSecretKey() - accountID := generateAccountID() + accessKey := admin.GenerateAWSAccessKey() + secretKey := admin.GenerateAWSSecretKey() + accountID := admin.GenerateAccountID() fmt.Println("\n🔑 Generated AWS credentials:") fmt.Printf(" Access Key: %s\n", accessKey) @@ -487,29 +529,18 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Printf(" Account ID: %s\n", accountID) // Generate SSL certificates - certPath := filepath.Join(configDir, "server.pem") - keyPath := filepath.Join(configDir, "server.key") - - if force || !fileExists(certPath) || !fileExists(keyPath) { - fmt.Println("\n🔐 Generating SSL certificates...") - if err := generateSelfSignedCert(certPath, keyPath); err != nil { - fmt.Fprintf(os.Stderr, "Error generating SSL certificates: %v\n", err) - os.Exit(1) - } - fmt.Printf("✅ SSL certificates generated:\n") - fmt.Printf(" Certificate: %s\n", certPath) - fmt.Printf(" Key: %s\n", keyPath) - } else { - fmt.Println("\n✅ SSL certificates already exist") - } + certPath := admin.GenerateCertificatesIfNeeded(configDir, force) // Generate NATS token - natsToken := generateNATSToken() + natsToken := admin.GenerateNATSToken() fmt.Println("\n🔒 Generated NATS authentication token") - // Get home directory for data path - homeDir, _ := os.UserHomeDir() - hiveRoot := filepath.Join(homeDir, "hive") + if hiveRoot == "" { + // Get home directory for data path + homeDir, _ := os.UserHomeDir() + hiveRoot = filepath.Join(homeDir, "hive") + + } // Create config files from embedded templates fmt.Println("\n📝 Creating configuration files...") @@ -527,55 +558,63 @@ func runAdminInit(cmd *cobra.Command, args []string) { } } - // Generate all config files - configs := []struct { - path string - template string - name string - }{ - {hiveTomlPath, hiveTomlTemplate, "hive.toml"}, - {filepath.Join(awsgwDir, "awsgw.toml"), awsgwTomlTemplate, "awsgw/awsgw.toml"}, - {filepath.Join(predastoreDir, "predastore.toml"), predastoreTomlTemplate, "predastore/predastore.toml"}, - {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, az, node, portStr); err != nil { - fmt.Fprintf(os.Stderr, "Error creating %s: %v\n", cfg.name, err) - os.Exit(1) - } - fmt.Printf("✅ Created: %s\n", cfg.name) + configSettings := admin.ConfigSettings{ + AccessKey: accessKey, + SecretKey: secretKey, + AccountID: accountID, + Region: region, + NatsToken: natsToken, + DataDir: hiveRoot, + + Node: node, + Az: az, + Port: portStr, + BindIP: bindIP, + ClusterBindIP: clusterBind, + //ClusterRoutes: clusterRoutes, + } + + // Generate all config files + configs := []admin.ConfigFile{ + { + Name: "hive.toml", + Path: hiveTomlPath, + Template: hiveTomlTemplate, + }, + { + Name: filepath.Join(awsgwDir, "awsgw.toml"), + Path: filepath.Join(awsgwDir, "awsgw.toml"), + Template: awsgwTomlTemplate, + }, + { + Name: filepath.Join(predastoreDir, "predastore.toml"), + Path: filepath.Join(predastoreDir, "predastore.toml"), + Template: predastoreTomlTemplate, + }, + { + Name: filepath.Join(natsDir, "nats.conf"), + Path: filepath.Join(natsDir, "nats.conf"), + Template: natsConfTemplate, + }, + } + + err := admin.GenerateConfigFiles(configs, configSettings) + if err != nil { + fmt.Fprintf(os.Stderr, "Error generating configuration files: %v\n", err) + os.Exit(1) } // Update ~/.aws/credentials and ~/.aws/config fmt.Println("\n🔧 Configuring AWS credentials...") - if err := setupAWSCredentials(accessKey, secretKey, region, certPath); err != nil { + if err := admin.SetupAWSCredentials(accessKey, secretKey, region, certPath); err != nil { fmt.Fprintf(os.Stderr, "Warning: Could not update AWS credentials: %v\n", err) } else { fmt.Println("✅ AWS credentials configured") } - // Create additional directories - dirs := []string{ - filepath.Join(hiveRoot, "images"), - filepath.Join(hiveRoot, "amis"), - filepath.Join(hiveRoot, "volumes"), - filepath.Join(hiveRoot, "state"), - filepath.Join(hiveRoot, "logs"), - filepath.Join(hiveRoot, "nats"), - filepath.Join(hiveRoot, "predastore"), - filepath.Join(hiveRoot, "viperblock"), - } - - fmt.Println("\n📁 Creating directory structure...") - for _, dir := range dirs { - if err := os.MkdirAll(dir, 0755); err != nil { - fmt.Fprintf(os.Stderr, "Warning: Could not create %s: %v\n", dir, err) - } - } - fmt.Printf("✅ Directory structure created in %s\n", hiveRoot) + admin.CreateServiceDirectories(hiveRoot) // Print success message fmt.Println("\n🎉 Hive initialization complete!") @@ -593,263 +632,233 @@ func runAdminInit(cmd *cobra.Command, args []string) { fmt.Println() } -// generateAWSAccessKey generates an AWS-style access key -// Format: AKIA + 16 random uppercase alphanumeric characters -func generateAWSAccessKey() string { - const prefix = "AKIA" - const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" - const length = 16 +func runAdminJoin(cmd *cobra.Command, args []string) { + node, _ := cmd.Flags().GetString("node") + leaderHost, _ := cmd.Flags().GetString("host") + region, _ := cmd.Flags().GetString("region") + az, _ := cmd.Flags().GetString("az") + dataDir, _ := cmd.Flags().GetString("data-dir") + port, _ := cmd.Flags().GetInt("port") + bindIP, _ := cmd.Flags().GetString("bind") + configDir, _ := cmd.Flags().GetString("config-dir") + clusterBind, _ := cmd.Flags().GetString("cluster-bind") + // Receive as string, split later + // TODO: Use GetStringArray + clusterRoutesStr, _ := cmd.Flags().GetString("cluster-routes") - result := make([]byte, length) - for i := range result { - num, _ := rand.Int(rand.Reader, big.NewInt(int64(len(charset)))) - result[i] = charset[num.Int64()] + // Supply as array to template + clusterRoutes := strings.Split(clusterRoutesStr, ",") + + // Validate required parameters + if node == "" { + fmt.Fprintf(os.Stderr, "❌ Error: --node is required\n") + os.Exit(1) + } + if leaderHost == "" { + fmt.Fprintf(os.Stderr, "❌ Error: --host is required\n") + os.Exit(1) } - return prefix + string(result) -} - -// generateAWSSecretKey generates an AWS-style secret key -// 40 character base64-encoded string -func generateAWSSecretKey() string { - bytes := make([]byte, 30) // 30 bytes = 40 chars in base64 - if _, err := rand.Read(bytes); err != nil { - panic(err) + // Validate IP address format + if net.ParseIP(bindIP) == nil { + fmt.Fprintf(os.Stderr, "❌ Error: Invalid IP address for --bind: %s\n", bindIP) + os.Exit(1) } - return base64.StdEncoding.EncodeToString(bytes) -} -// generateAccountID generates a 12-digit AWS account ID -func generateAccountID() string { - // Generate random 12-digit number - num, _ := rand.Int(rand.Reader, big.NewInt(900000000000)) - accountID := num.Int64() + 100000000000 // Ensure it's 12 digits - return fmt.Sprintf("%012d", accountID) -} - -// generateNATSToken generates a secure random token for NATS -func generateNATSToken() string { - bytes := make([]byte, 32) - if _, err := rand.Read(bytes); err != nil { - panic(err) + // Validate port range + if port < 1 || port > 65535 { + fmt.Fprintf(os.Stderr, "❌ Error: Port must be between 1 and 65535, got: %d\n", port) + os.Exit(1) } - return "nats_" + base64.URLEncoding.EncodeToString(bytes)[:32] -} -// generateSelfSignedCert generates a self-signed SSL certificate -func generateSelfSignedCert(certPath, keyPath string) error { - // Generate private key - privateKey, err := rsa.GenerateKey(rand.Reader, 4096) + // Set default data directory + if dataDir == "" { + homeDir, err := os.UserHomeDir() + if err != nil { + fmt.Fprintf(os.Stderr, "❌ Error getting home directory: %v\n", err) + os.Exit(1) + } + dataDir = filepath.Join(homeDir, "hive") + } + + // Set daemon host for this node + daemonHost := fmt.Sprintf("%s:%d", bindIP, port) + + fmt.Println("🚀 Joining Hive cluster...") + fmt.Printf("Node: %s\n", node) + fmt.Printf("Leader: %s\n", leaderHost) + fmt.Printf("Region: %s\n", region) + fmt.Printf("AZ: %s\n", az) + fmt.Printf("Bind IP: %s\n", bindIP) + fmt.Printf("Port: %d\n\n", port) + + // Create join request + joinReq := config.NodeJoinRequest{ + Node: node, + Region: region, + AZ: az, + DataDir: dataDir, + DaemonHost: daemonHost, + } + + reqBody, err := json.Marshal(joinReq) if err != nil { - return fmt.Errorf("failed to generate private key: %w", err) + fmt.Fprintf(os.Stderr, "Error marshaling join request: %v\n", err) + os.Exit(1) } - // Create certificate template - notBefore := time.Now() - notAfter := notBefore.Add(3650 * 24 * time.Hour) // 10 years - - serialNumber, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) - if err != nil { - return fmt.Errorf("failed to generate serial number: %w", err) - } - - template := x509.Certificate{ - SerialNumber: serialNumber, - Subject: pkix.Name{ - CommonName: "localhost", - Organization: []string{"Hive Platform"}, + // Send join request to leader + client := &http.Client{ + Timeout: 30 * time.Second, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // Skip TLS verification for self-signed certs }, - NotBefore: notBefore, - NotAfter: notAfter, - KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, - BasicConstraintsValid: true, - DNSNames: []string{"localhost"}, - IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, } - // Create certificate - derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &privateKey.PublicKey, privateKey) + joinURL := fmt.Sprintf("http://%s/join", leaderHost) + resp, err := client.Post(joinURL, "application/json", bytes.NewBuffer(reqBody)) if err != nil { - return fmt.Errorf("failed to create certificate: %w", err) + fmt.Fprintf(os.Stderr, "❌ Error connecting to leader node: %v\n", err) + fmt.Fprintf(os.Stderr, "Make sure the leader node is running and accessible at %s\n", leaderHost) + os.Exit(1) + } + defer resp.Body.Close() + + body, _ := io.ReadAll(resp.Body) + + if resp.StatusCode != 200 { + var errResp config.NodeJoinResponse + json.Unmarshal(body, &errResp) + fmt.Fprintf(os.Stderr, "❌ Failed to join cluster: %s\n", errResp.Message) + os.Exit(1) } - // Write certificate to file - certOut, err := os.Create(certPath) + var joinResp config.NodeJoinResponse + if err := json.Unmarshal(body, &joinResp); err != nil { + fmt.Fprintf(os.Stderr, "Error parsing join response: %v\n", err) + os.Exit(1) + } + + if !joinResp.Success { + fmt.Fprintf(os.Stderr, "❌ Failed to join cluster: %s\n", joinResp.Message) + os.Exit(1) + } + + fmt.Println("✅ Successfully joined cluster!") + fmt.Printf("Epoch: %d\n", joinResp.SharedData.Epoch) + fmt.Printf("Config hash: %s\n\n", joinResp.ConfigHash) + + // Save cluster config locally + // Confirm if default config directory required + if configDir == "" { + homeDir, err := os.UserHomeDir() + if err != nil { + fmt.Fprintf(os.Stderr, "Error getting home directory: %v\n", err) + os.Exit(1) + } + configDir = filepath.Join(homeDir, "hive", "config") + } + + hiveTomlPath := filepath.Join(configDir, "hive.toml") + + // Create config directory if needed + if err := os.MkdirAll(configDir, 0700); err != nil { + fmt.Fprintf(os.Stderr, "Error creating config directory: %v\n", err) + os.Exit(1) + } + + // Construct local node config with THIS node's name at top level + localConfig := config.ClusterConfig{ + Epoch: joinResp.SharedData.Epoch, + Node: node, // THIS node's name, not the leader's + Version: joinResp.SharedData.Version, + Nodes: joinResp.SharedData.Nodes, + } + + // Marshal config to TOML format + configTOML, err := toml.Marshal(&localConfig) if err != nil { - return fmt.Errorf("failed to create cert file: %w", err) - } - defer certOut.Close() - - if err := pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes}); err != nil { - return fmt.Errorf("failed to write cert: %w", err) + fmt.Fprintf(os.Stderr, "Error marshaling config to TOML: %v\n", err) + os.Exit(1) } - // Write private key to file - keyOut, err := os.OpenFile(keyPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600) - if err != nil { - return fmt.Errorf("failed to create key file: %w", err) - } - defer keyOut.Close() - - privBytes, err := x509.MarshalPKCS8PrivateKey(privateKey) - if err != nil { - return fmt.Errorf("failed to marshal private key: %w", err) + // Write config to file + if err := os.WriteFile(hiveTomlPath, configTOML, 0600); err != nil { + fmt.Fprintf(os.Stderr, "Error writing config file: %v\n", err) + os.Exit(1) } - if err := pem.Encode(keyOut, &pem.Block{Type: "PRIVATE KEY", Bytes: privBytes}); err != nil { - return fmt.Errorf("failed to write key: %w", err) - } + fmt.Printf("✅ Config saved to: %s\n\n", hiveTomlPath) - return nil -} + // Generate certificates if needed + certPath := admin.GenerateCertificatesIfNeeded(configDir, false) -// generateConfigFile creates a configuration file from a template -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 { - return fmt.Errorf("failed to parse template: %w", err) - } + fmt.Printf("✅ SSL certificates available at: %s\n\n", certPath) - // Template data - data := struct { - AccessKey string - SecretKey string - AccountID string - Region string - NatsToken string - DataDir string + // Write individual node config files + portStr := fmt.Sprintf("%d", port) - // Add more fields as needed - Node string - Az string - Port string - }{ - AccessKey: accessKey, - SecretKey: secretKey, - AccountID: accountID, + configSettings := admin.ConfigSettings{ + AccessKey: localConfig.Nodes[node].AccessKey, + SecretKey: localConfig.Nodes[node].SecretKey, Region: region, - NatsToken: natsToken, + NatsToken: localConfig.Nodes[node].NATS.ACL.Token, DataDir: dataDir, - Node: Node, - Az: Az, - Port: Port, + Node: node, + Az: az, + Port: portStr, + BindIP: bindIP, + ClusterBindIP: clusterBind, + ClusterRoutes: clusterRoutes, } - // Create file with secure permissions - f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600) - if err != nil { - return fmt.Errorf("failed to create config file: %w", err) - } - defer f.Close() + awsgwDir := filepath.Join(configDir, "awsgw") + predastoreDir := filepath.Join(configDir, "predastore") + natsDir := filepath.Join(configDir, "nats") - // Execute template - if err := tmpl.Execute(f, data); err != nil { - return fmt.Errorf("failed to execute template: %w", err) - } - - return nil -} - -// setupAWSCredentials updates ~/.aws/credentials and ~/.aws/config -func setupAWSCredentials(accessKey, secretKey, region, certPath string) error { - homeDir, err := os.UserHomeDir() - if err != nil { - return err - } - - awsDir := filepath.Join(homeDir, ".aws") - if err := os.MkdirAll(awsDir, 0700); err != nil { - return err - } - - credPath := filepath.Join(awsDir, "credentials") - configPath := filepath.Join(awsDir, "config") - - // Determine profile name - //profileName := "default" - - // Use hive as the default profile - profileName := "hive" - - if fileExists(credPath) { - // Check if default profile already exists - cfg, err := ini.Load(credPath) - if err == nil && cfg.HasSection("default") { - profileName = "hive" + for _, dir := range []string{awsgwDir, predastoreDir, natsDir} { + if err := os.MkdirAll(dir, 0700); err != nil { + fmt.Fprintf(os.Stderr, "Error creating directory %s: %v\n", dir, err) + os.Exit(1) } } - // Update credentials file - if err := updateAWSINIFile(credPath, profileName, map[string]string{ - "aws_access_key_id": accessKey, - "aws_secret_access_key": secretKey, - }); err != nil { - return err + // Generate all config files + configs := []admin.ConfigFile{ + { + Name: "hive.toml", + Path: hiveTomlPath, + Template: hiveTomlTemplate, + }, + { + Name: filepath.Join(awsgwDir, "awsgw.toml"), + Path: filepath.Join(awsgwDir, "awsgw.toml"), + Template: awsgwTomlTemplate, + }, + { + Name: filepath.Join(predastoreDir, "predastore.toml"), + Path: filepath.Join(predastoreDir, "predastore.toml"), + Template: predastoreTomlTemplate, + }, + { + Name: filepath.Join(natsDir, "nats.conf"), + Path: filepath.Join(natsDir, "nats.conf"), + Template: natsConfTemplate, + }, } - // Update config file - configSection := profileName - if profileName != "default" { - configSection = "profile " + profileName - } - - if err := updateAWSINIFile(configPath, configSection, map[string]string{ - "region": region, - "endpoint_url": "https://localhost:9999", - "ca_bundle": certPath, - "output": "json", - }); err != nil { - return err - } - - fmt.Printf(" Profile: %s\n", profileName) - if profileName != "default" { - fmt.Printf(" Use: export AWS_PROFILE=%s\n", profileName) - } - - return nil -} - -// Helper functions - -func fileExists(path string) bool { - _, err := os.Stat(path) - return err == nil -} - -// updateAWSINIFile updates or creates an AWS INI file section with given key-value pairs -func updateAWSINIFile(path, section string, values map[string]string) error { - var cfg *ini.File - var err error - - // Load existing file or create new one - if fileExists(path) { - cfg, err = ini.Load(path) - if err != nil { - return fmt.Errorf("failed to load INI file: %w", err) - } - } else { - cfg = ini.Empty() - } - - // Get or create section - sec, err := cfg.NewSection(section) + err = admin.GenerateConfigFiles(configs, configSettings) if err != nil { - // Section already exists, get it - sec, err = cfg.GetSection(section) - if err != nil { - return fmt.Errorf("failed to get section: %w", err) - } + fmt.Fprintf(os.Stderr, "Error generating configuration files: %v\n", err) + os.Exit(1) } - // Set key-value pairs - for key, value := range values { - sec.Key(key).SetValue(value) - } + admin.CreateServiceDirectories(dataDir) - // Save with proper permissions - return cfg.SaveTo(path) + fmt.Println("🎉 Node successfully joined cluster!") + fmt.Println("\n📋 Next steps:") + fmt.Println(" 1. Start the hive service:") + fmt.Printf(" hive service hive start --config %s\n", hiveTomlPath) + fmt.Println() } diff --git a/cmd/hive/cmd/service.go b/cmd/hive/cmd/service.go index 9c6887b44..8de8af7bb 100644 --- a/cmd/hive/cmd/service.go +++ b/cmd/hive/cmd/service.go @@ -307,12 +307,15 @@ var natsStartCmd = &cobra.Command{ dataDir := viper.GetString("data-dir") jetStream := viper.GetBool("jetstream") + cfgFile := viper.GetString("config") + service, err := service.New("nats", &nats.Config{ - Port: port, - Host: host, - Debug: debug, - DataDir: dataDir, - JetStream: jetStream, + ConfigFile: cfgFile, + Port: port, + Host: host, + Debug: debug, + DataDir: dataDir, + JetStream: jetStream, }) if err != nil { @@ -394,14 +397,19 @@ var hiveStartCmd = &cobra.Command{ // Apply changes back to cluster config clusterConfig.Nodes[clusterConfig.Node] = nodeConfig - service, err := service.New("hive", clusterConfig) + svc, err := service.New("hive", clusterConfig) if err != nil { fmt.Println("Error starting hive service:", err) return } - service.Start() + // Set config path for cluster manager + if hiveSvc, ok := svc.(interface{ SetConfigPath(string) }); ok { + hiveSvc.SetConfigPath(cfgFile) + } + + svc.Start() fmt.Println("HIVE service started") }, } @@ -463,7 +471,7 @@ var awsgwStartCmd = &cobra.Command{ awsgwHost := viper.GetString("host") if awsgwHost != "" { fmt.Println("Overwriting awsgw host to:", awsgwHost) - nodeConfig.AWSGW.Host = awsgwHost + //nodeConfig.AWSGW.Host = awsgwHost } awsgwTlsCert := viper.GetString("tls-cert") diff --git a/cmd/hive/cmd/templates/hive.toml b/cmd/hive/cmd/templates/hive.toml index 0ab695e49..9861ba292 100644 --- a/cmd/hive/cmd/templates/hive.toml +++ b/cmd/hive/cmd/templates/hive.toml @@ -13,19 +13,19 @@ access_key = "{{.AccessKey}}" secret_key = "{{.SecretKey}}" [nodes.{{.Node}}.daemon] -host = "0.0.0.0:{{.Port}}" +host = "{{.BindIP}}:{{.Port}}" tlskey = "config/server.key" tlscert = "config/server.pem" [nodes.{{.Node}}.awsgw] -host = "0.0.0.0:9999" +host = "{{.BindIP}}:9999" tlskey = "config/server.key" tlscert = "config/server.pem" debug = true config = "awsgw/awsgw.toml" [nodes.{{.Node}}.nats] -host = "0.0.0.0:4222" +host = "{{.BindIP}}:4222" [nodes.{{.Node}}.nats.acl] token = "{{.NatsToken}}" @@ -33,7 +33,7 @@ token = "{{.NatsToken}}" [nodes.{{.Node}}.nats.sub] [nodes.{{.Node}}.predastore] -host = "0.0.0.0:8443" +host = "{{.BindIP}}:8443" region = "{{.Region}}" bucket = "predastore" access_key = "{{.AccessKey}}" diff --git a/cmd/hive/cmd/templates/nats.conf b/cmd/hive/cmd/templates/nats.conf index 7fa47ba52..5ba94281a 100644 --- a/cmd/hive/cmd/templates/nats.conf +++ b/cmd/hive/cmd/templates/nats.conf @@ -2,20 +2,42 @@ # Generated by Hive admin init # Server configuration -port: 4222 -server_name: hive-nats +{{ if .BindIP }} +listen: {{.BindIP}}:4222 +{{ else }} +listen: 0.0.0.0:4222 +{{ end }} + + +{{ if .ClusterBindIP }} +cluster { + name: C1 + listen: {{.ClusterBindIP}}:4248 + +{{ if .ClusterRoutes }} +routes = [ +{{- range .ClusterRoutes }} + "nats://{{ . }}", +{{- end }} +] +{{ end }} +} +{{ end }} + + +server_name: hive-nats-{{.Node}} # Authorization authorization { - token: "{{.NatsToken}}" +# token: "{{.NatsToken}}" } # JetStream for persistent messaging -jetstream { - store_dir: "{{.DataDir}}/nats/storage" - max_memory_store: 1GB - max_file_store: 10GB -} +#jetstream { +# store_dir: "{{.DataDir}}/nats/" +# max_memory_store: 1GB +# max_file_store: 10GB +#} # Logging debug: false @@ -29,4 +51,9 @@ max_payload: 1MB max_pending: 64MB # Monitoring +{{ if .BindIP }} +http: {{.BindIP}}:8222 +{{ else }} http_port: 8222 +{{ end }} + diff --git a/go.mod b/go.mod index 36704c338..fb94f762d 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/google/uuid v1.6.0 github.com/kdomanski/iso9660 v0.4.0 github.com/mulgadc/predastore v0.1.5 - github.com/mulgadc/viperblock v0.1.1 + github.com/mulgadc/viperblock v0.1.2 github.com/nats-io/nats-server/v2 v2.11.6 github.com/nats-io/nats.go v1.43.0 github.com/pterm/pterm v0.12.82 @@ -63,4 +63,4 @@ require ( golang.org/x/text v0.26.0 // indirect golang.org/x/time v0.12.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect -) \ No newline at end of file +) diff --git a/go.sum b/go.sum index dbc57d4e8..6c1481042 100644 --- a/go.sum +++ b/go.sum @@ -87,6 +87,8 @@ github.com/mulgadc/predastore v0.1.5 h1:7SySkQQ5MQ1NTcKcB14pCUIJipq0YVgCJZeKM6yt github.com/mulgadc/predastore v0.1.5/go.mod h1:BhMymRV3zMTu2S+5pJnNDDm5+AGakjhSc/PKX0q8Xfg= github.com/mulgadc/viperblock v0.1.1 h1:R1DVX1S+/xPPMQ4LIFkwIX7Es+RS3Wgdd8bsWLTEvgY= github.com/mulgadc/viperblock v0.1.1/go.mod h1:zCjdQ76fi2Qs2a5ign9A/Uqgbz77xts5WV0OR1be48I= +github.com/mulgadc/viperblock v0.1.2 h1:h5yAha2CIsSp/x7oPvqByqlsYzZYSLLim15AL1o7Dm0= +github.com/mulgadc/viperblock v0.1.2/go.mod h1:zCjdQ76fi2Qs2a5ign9A/Uqgbz77xts5WV0OR1be48I= github.com/nats-io/jwt/v2 v2.7.4 h1:jXFuDDxs/GQjGDZGhNgH4tXzSUK6WQi2rsj4xmsNOtI= github.com/nats-io/jwt/v2 v2.7.4/go.mod h1:me11pOkwObtcBNR8AiMrUbtVOUGkqYjMQZ6jnSdVUIA= github.com/nats-io/nats-server/v2 v2.11.6 h1:4VXRjbTUFKEB+7UoaKL3F5Y83xC7MxPoIONOnGgpkHw= diff --git a/hive/admin/admin.go b/hive/admin/admin.go new file mode 100644 index 000000000..6d46f357d --- /dev/null +++ b/hive/admin/admin.go @@ -0,0 +1,342 @@ +package admin + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "encoding/pem" + "fmt" + "math/big" + "net" + "os" + "path/filepath" + "text/template" + "time" + + "gopkg.in/ini.v1" +) + +type ConfigSettings struct { + AccessKey string + SecretKey string + AccountID string + Region string + NatsToken string + DataDir string + + // Add more fields as needed + Node string + Az string + Port string + BindIP string + + // Cluster settings + ClusterBindIP string + ClusterRoutes []string +} + +type ConfigFile struct { + Name string + Path string + Template string +} + +func GenerateConfigFiles(configs []ConfigFile, configSettings ConfigSettings) error { + + for _, cfg := range configs { + if err := GenerateConfigFile(cfg.Path, cfg.Template, configSettings); err != nil { + return fmt.Errorf("Error creating %s: %v\n", cfg.Name, err) + } + fmt.Printf("✅ Created: %s\n", cfg.Name) + } + + return nil +} + +// generateConfigFile creates a configuration file from a template +func GenerateConfigFile(configPath string, configTemplate string, configSettings ConfigSettings) error { + // Parse the embedded template + tmpl, err := template.New("config").Parse(configTemplate) + if err != nil { + return fmt.Errorf("failed to parse template: %w", err) + } + + // Create file with secure permissions + f, err := os.OpenFile(configPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600) + if err != nil { + return fmt.Errorf("failed to create config file: %w", err) + } + defer f.Close() + + // Execute template + if err := tmpl.Execute(f, configSettings); err != nil { + return fmt.Errorf("failed to execute template: %w", err) + } + + return nil +} + +func GenerateCertificatesIfNeeded(configDir string, force bool) (certPath string) { + // Generate SSL certificates + certPath = filepath.Join(configDir, "server.pem") + keyPath := filepath.Join(configDir, "server.key") + + if force || !FileExists(certPath) || !FileExists(keyPath) { + fmt.Println("\n🔐 Generating SSL certificates...") + if err := GenerateSelfSignedCert(certPath, keyPath); err != nil { + fmt.Fprintf(os.Stderr, "Error generating SSL certificates: %v\n", err) + os.Exit(1) + } + fmt.Printf("✅ SSL certificates generated:\n") + fmt.Printf(" Certificate: %s\n", certPath) + fmt.Printf(" Key: %s\n", keyPath) + } else { + fmt.Println("\n✅ SSL certificates already exist") + } + + return certPath + +} + +func CreateServiceDirectories(hiveRoot string) { + + // Create additional directories + dirs := []string{ + filepath.Join(hiveRoot, "images"), + filepath.Join(hiveRoot, "amis"), + filepath.Join(hiveRoot, "volumes"), + filepath.Join(hiveRoot, "state"), + filepath.Join(hiveRoot, "logs"), + filepath.Join(hiveRoot, "nats"), + filepath.Join(hiveRoot, "predastore"), + filepath.Join(hiveRoot, "viperblock"), + filepath.Join(hiveRoot, "hive"), + } + + fmt.Println("\n📁 Creating directory structure...") + for _, dir := range dirs { + + // Check if directory exists + if _, err := os.Stat(dir); os.IsNotExist(err) { + if err := os.MkdirAll(dir, 0755); err != nil { + fmt.Fprintf(os.Stderr, "Warning: Could not create %s: %v\n", dir, err) + } + } + } + fmt.Printf("✅ Directory structure created in %s\n", hiveRoot) + +} + +// Helper functions + +func FileExists(path string) bool { + _, err := os.Stat(path) + return err == nil +} + +// updateAWSINIFile updates or creates an AWS INI file section with given key-value pairs +func UpdateAWSINIFile(path, section string, values map[string]string) error { + var cfg *ini.File + var err error + + // Load existing file or create new one + if FileExists(path) { + cfg, err = ini.Load(path) + if err != nil { + return fmt.Errorf("failed to load INI file: %w", err) + } + } else { + cfg = ini.Empty() + } + + // Get or create section + sec, err := cfg.NewSection(section) + if err != nil { + // Section already exists, get it + sec, err = cfg.GetSection(section) + if err != nil { + return fmt.Errorf("failed to get section: %w", err) + } + } + + // Set key-value pairs + for key, value := range values { + sec.Key(key).SetValue(value) + } + + // Save with proper permissions + return cfg.SaveTo(path) +} + +// generateAWSAccessKey generates an AWS-style access key +// Format: AKIA + 16 random uppercase alphanumeric characters +func GenerateAWSAccessKey() string { + const prefix = "AKIA" + const charset = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + const length = 16 + + result := make([]byte, length) + for i := range result { + num, _ := rand.Int(rand.Reader, big.NewInt(int64(len(charset)))) + result[i] = charset[num.Int64()] + } + + return prefix + string(result) +} + +// generateAWSSecretKey generates an AWS-style secret key +// 40 character base64-encoded string +func GenerateAWSSecretKey() string { + bytes := make([]byte, 30) // 30 bytes = 40 chars in base64 + if _, err := rand.Read(bytes); err != nil { + panic(err) + } + return base64.StdEncoding.EncodeToString(bytes) +} + +// generateAccountID generates a 12-digit AWS account ID +func GenerateAccountID() string { + // Generate random 12-digit number + num, _ := rand.Int(rand.Reader, big.NewInt(900000000000)) + accountID := num.Int64() + 100000000000 // Ensure it's 12 digits + return fmt.Sprintf("%012d", accountID) +} + +// generateNATSToken generates a secure random token for NATS +func GenerateNATSToken() string { + bytes := make([]byte, 32) + if _, err := rand.Read(bytes); err != nil { + panic(err) + } + return "nats_" + base64.URLEncoding.EncodeToString(bytes)[:32] +} + +// generateSelfSignedCert generates a self-signed SSL certificate +func GenerateSelfSignedCert(certPath, keyPath string) error { + // Generate private key + privateKey, err := rsa.GenerateKey(rand.Reader, 4096) + if err != nil { + return fmt.Errorf("failed to generate private key: %w", err) + } + + // Create certificate template + notBefore := time.Now() + notAfter := notBefore.Add(3650 * 24 * time.Hour) // 10 years + + serialNumber, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) + if err != nil { + return fmt.Errorf("failed to generate serial number: %w", err) + } + + template := x509.Certificate{ + SerialNumber: serialNumber, + Subject: pkix.Name{ + CommonName: "localhost", + Organization: []string{"Hive Platform"}, + }, + NotBefore: notBefore, + NotAfter: notAfter, + KeyUsage: x509.KeyUsageKeyEncipherment | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + DNSNames: []string{"localhost"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.ParseIP("::1")}, + } + + // Create certificate + derBytes, err := x509.CreateCertificate(rand.Reader, &template, &template, &privateKey.PublicKey, privateKey) + if err != nil { + return fmt.Errorf("failed to create certificate: %w", err) + } + + // Write certificate to file + certOut, err := os.Create(certPath) + if err != nil { + return fmt.Errorf("failed to create cert file: %w", err) + } + defer certOut.Close() + + if err := pem.Encode(certOut, &pem.Block{Type: "CERTIFICATE", Bytes: derBytes}); err != nil { + return fmt.Errorf("failed to write cert: %w", err) + } + + // Write private key to file + keyOut, err := os.OpenFile(keyPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600) + if err != nil { + return fmt.Errorf("failed to create key file: %w", err) + } + defer keyOut.Close() + + privBytes, err := x509.MarshalPKCS8PrivateKey(privateKey) + if err != nil { + return fmt.Errorf("failed to marshal private key: %w", err) + } + + if err := pem.Encode(keyOut, &pem.Block{Type: "PRIVATE KEY", Bytes: privBytes}); err != nil { + return fmt.Errorf("failed to write key: %w", err) + } + + return nil +} + +// setupAWSCredentials updates ~/.aws/credentials and ~/.aws/config +func SetupAWSCredentials(accessKey, secretKey, region, certPath string) error { + homeDir, err := os.UserHomeDir() + if err != nil { + return err + } + + awsDir := filepath.Join(homeDir, ".aws") + if err := os.MkdirAll(awsDir, 0700); err != nil { + return err + } + + credPath := filepath.Join(awsDir, "credentials") + configPath := filepath.Join(awsDir, "config") + + // Determine profile name + //profileName := "default" + + // Use hive as the default profile + profileName := "hive" + + if FileExists(credPath) { + // Check if default profile already exists + cfg, err := ini.Load(credPath) + if err == nil && cfg.HasSection("default") { + profileName = "hive" + } + } + + // Update credentials file + if err := UpdateAWSINIFile(credPath, profileName, map[string]string{ + "aws_access_key_id": accessKey, + "aws_secret_access_key": secretKey, + }); err != nil { + return err + } + + // Update config file + configSection := profileName + if profileName != "default" { + configSection = "profile " + profileName + } + + if err := UpdateAWSINIFile(configPath, configSection, map[string]string{ + "region": region, + "endpoint_url": "https://localhost:9999", + "ca_bundle": certPath, + "output": "json", + }); err != nil { + return err + } + + fmt.Printf(" Profile: %s\n", profileName) + if profileName != "default" { + fmt.Printf(" Use: export AWS_PROFILE=%s\n", profileName) + } + + return nil +} diff --git a/hive/config/config.go b/hive/config/config.go index 117de6216..1435f95e0 100644 --- a/hive/config/config.go +++ b/hive/config/config.go @@ -34,8 +34,8 @@ type Config struct { // Authentication // TODO: Move to more appropriate setting above - AccessKey string `mapstructure:"access_key"` - SecretKey string `mapstructure:"secret_key"` + AccessKey string `mapstructure:"accesskey"` + SecretKey string `mapstructure:"secretkey"` BaseDir string `mapstructure:"base_dir"` WalDir string `mapstructure:"wal_dir"` } @@ -54,8 +54,8 @@ type PredastoreConfig struct { Host string `mapstructure:"host"` Bucket string `mapstructure:"bucket"` Region string `mapstructure:"region"` - AccessKey string `mapstructure:"access_key"` - SecretKey string `mapstructure:"secret_key"` + AccessKey string `mapstructure:"accesskey"` + SecretKey string `mapstructure:"secretkey"` BaseDir string `mapstructure:"base_dir"` } @@ -143,6 +143,38 @@ func (ec2StartInstanceResponse EC2StartInstancesResponse) Respond(msg *nats.Msg) } +// Cluster manager API types +type NodeJoinRequest struct { + Node string `json:"node"` + Region string `json:"region"` + AZ string `json:"az"` + DataDir string `json:"data_dir"` + DaemonHost string `json:"daemon_host"` // Host:port where this node's daemon is accessible +} + +// SharedClusterData contains only the shared cluster information (no node-specific top-level fields) +type SharedClusterData struct { + Epoch uint64 `json:"epoch" toml:"epoch"` + Version string `json:"version" toml:"version"` + Nodes map[string]Config `json:"nodes" toml:"nodes"` +} + +type NodeJoinResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + SharedData *SharedClusterData `json:"shared_data,omitempty"` + ConfigHash string `json:"config_hash,omitempty"` + JoiningNode string `json:"joining_node,omitempty"` // The node name that is joining +} + +type NodeHealthResponse struct { + Node string `json:"node"` + Status string `json:"status"` + ConfigHash string `json:"config_hash"` + Epoch uint64 `json:"epoch"` + Uptime int64 `json:"uptime"` +} + // LoadConfig loads the configuration from file and environment variables func LoadConfig(configPath string) (*ClusterConfig, error) { // Set environment variable prefix diff --git a/hive/daemon/daemon.go b/hive/daemon/daemon.go index 3c94544cb..26ac83de5 100644 --- a/hive/daemon/daemon.go +++ b/hive/daemon/daemon.go @@ -3,6 +3,8 @@ package daemon import ( "bufio" "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -20,6 +22,7 @@ import ( "time" "github.com/aws/aws-sdk-go/service/ec2" + "github.com/gofiber/fiber/v2" "github.com/mulgadc/hive/hive/awserrors" "github.com/mulgadc/hive/hive/config" gateway_ec2_instance "github.com/mulgadc/hive/hive/gateway/ec2/instance" @@ -31,6 +34,7 @@ import ( "github.com/mulgadc/hive/hive/vm" "github.com/mulgadc/viperblock/viperblock" "github.com/nats-io/nats.go" + "github.com/pelletier/go-toml/v2" ) type BlockDeviceMapping struct { @@ -89,6 +93,11 @@ type Daemon struct { // NAT Subscriptions natsSubscriptions map[string]*nats.Subscription + // Cluster manager + clusterApp *fiber.App + startTime time.Time + configPath string + mu sync.Mutex } @@ -179,6 +188,11 @@ func NewResourceManager() *ResourceManager { } } +// SetConfigPath sets the configuration file path for cluster management +func (d *Daemon) SetConfigPath(path string) { + d.configPath = path +} + // NewDaemon creates a new daemon instance func NewDaemon(cfg *config.ClusterConfig) *Daemon { ctx, cancel := context.WithCancel(context.Background()) @@ -200,6 +214,7 @@ func NewDaemon(cfg *config.ClusterConfig) *Daemon { cancel: cancel, Instances: vm.Instances{VMS: make(map[string]*vm.VM)}, natsSubscriptions: make(map[string]*nats.Subscription), + startTime: time.Now(), } } @@ -371,6 +386,20 @@ func (d *Daemon) Start() error { return fmt.Errorf("failed to subscribe to NATS ec2.launch: %w", err) } + // Subscribe to health check for this node + healthSubject := fmt.Sprintf("hive.admin.%s.health", d.node) + log.Printf("Subscribing to health check: %s", healthSubject) + + d.natsSubscriptions[healthSubject], err = d.natsConn.Subscribe(healthSubject, d.handleHealthCheck) + if err != nil { + return fmt.Errorf("failed to subscribe to NATS %s: %w", healthSubject, err) + } + + // Start cluster manager HTTP server + if err := d.ClusterManager(); err != nil { + return fmt.Errorf("failed to start cluster manager: %w", err) + } + // Setup graceful shutdown d.setupShutdown() @@ -388,6 +417,180 @@ func (d *Daemon) Start() error { return nil } +// computeConfigHash computes SHA256 hash of the shared cluster config (excluding node-specific fields) +func (d *Daemon) computeConfigHash() (string, error) { + // Only hash the shared cluster data, not the node-specific top-level field + sharedData := config.SharedClusterData{ + Epoch: d.clusterConfig.Epoch, + Version: d.clusterConfig.Version, + Nodes: d.clusterConfig.Nodes, + } + + configJSON, err := json.Marshal(sharedData) + if err != nil { + return "", err + } + hash := sha256.Sum256(configJSON) + return hex.EncodeToString(hash[:]), nil +} + +// saveClusterConfig writes the cluster config to disk in TOML format +func (d *Daemon) saveClusterConfig() error { + if d.configPath == "" { + return fmt.Errorf("config path not set") + } + + // Marshal to TOML + configTOML, err := toml.Marshal(d.clusterConfig) + if err != nil { + return fmt.Errorf("failed to marshal config to TOML: %w", err) + } + + // Write to config file + if err := os.WriteFile(d.configPath, configTOML, 0600); err != nil { + return fmt.Errorf("failed to write config: %w", err) + } + + slog.Info("Cluster config saved", "path", d.configPath, "epoch", d.clusterConfig.Epoch) + return nil +} + +// ClusterManager starts the HTTP cluster management server +func (d *Daemon) ClusterManager() error { + + // Get daemon host from config + daemonHost := d.config.Daemon.Host + if daemonHost == "" { + return fmt.Errorf("daemon.host not configured") + } + + // Create Fiber app + d.clusterApp = fiber.New(fiber.Config{ + DisableStartupMessage: true, + AppName: "Hive Cluster Manager", + }) + + // Health endpoint - responds to HTTP and NATS + d.clusterApp.Get("/health", func(c *fiber.Ctx) error { + configHash, err := d.computeConfigHash() + if err != nil { + return c.Status(500).JSON(fiber.Map{ + "error": "failed to compute config hash", + }) + } + + response := config.NodeHealthResponse{ + Node: d.node, + Status: "running", + ConfigHash: configHash, + Epoch: d.clusterConfig.Epoch, + Uptime: int64(time.Since(d.startTime).Seconds()), + } + + return c.JSON(response) + }) + + // Join endpoint - accepts new nodes joining the cluster + d.clusterApp.Post("/join", func(c *fiber.Ctx) error { + var req config.NodeJoinRequest + if err := c.BodyParser(&req); err != nil { + return c.Status(400).JSON(config.NodeJoinResponse{ + Success: false, + Message: "invalid request body", + }) + } + + slog.Info("Node join request received", "node", req.Node, "region", req.Region, "az", req.AZ) + + // Validate request + if req.Node == "" || req.Region == "" || req.AZ == "" { + return c.Status(400).JSON(config.NodeJoinResponse{ + Success: false, + Message: "node, region, and az are required", + }) + } + + // Check if node already exists + if _, exists := d.clusterConfig.Nodes[req.Node]; exists { + return c.Status(409).JSON(config.NodeJoinResponse{ + Success: false, + Message: fmt.Sprintf("node %s already exists in cluster", req.Node), + }) + } + + // Add new node to cluster config + d.mu.Lock() + newNodeConfig := config.Config{ + Node: req.Node, + Region: req.Region, + AZ: req.AZ, + DataDir: req.DataDir, + Daemon: config.DaemonConfig{ + Host: req.DaemonHost, + }, + // Copy shared config from current node + NATS: d.config.NATS, + Predastore: d.config.Predastore, + AWSGW: d.config.AWSGW, + AccessKey: d.config.AccessKey, + SecretKey: d.config.SecretKey, + BaseDir: req.DataDir, + } + + d.clusterConfig.Nodes[req.Node] = newNodeConfig + d.clusterConfig.Epoch++ // Increment epoch for version tracking + d.mu.Unlock() + + // Save updated config + if err := d.saveClusterConfig(); err != nil { + slog.Error("Failed to save cluster config", "error", err) + return c.Status(500).JSON(config.NodeJoinResponse{ + Success: false, + Message: "failed to save cluster config", + }) + } + + configHash, _ := d.computeConfigHash() + + slog.Info("Node joined cluster", "node", req.Node, "epoch", d.clusterConfig.Epoch) + + // Send only shared cluster data (exclude node-specific top-level fields) + sharedData := &config.SharedClusterData{ + Epoch: d.clusterConfig.Epoch, + Version: d.clusterConfig.Version, + Nodes: d.clusterConfig.Nodes, + } + + return c.JSON(config.NodeJoinResponse{ + Success: true, + Message: fmt.Sprintf("node %s successfully joined cluster", req.Node), + SharedData: sharedData, + ConfigHash: configHash, + JoiningNode: req.Node, + }) + }) + + // Get cluster config endpoint + d.clusterApp.Get("/config", func(c *fiber.Ctx) error { + configHash, _ := d.computeConfigHash() + + return c.JSON(fiber.Map{ + "config": d.clusterConfig, + "config_hash": configHash, + }) + }) + + // Start HTTP server in goroutine + go func() { + slog.Info("Starting cluster manager", "host", daemonHost) + if err := d.clusterApp.Listen(daemonHost); err != nil { + slog.Error("Cluster manager failed to start", "error", err) + } + }() + + return nil +} + // Write the state to disk func (d *Daemon) WriteState() error { d.Instances.Mu.Lock() @@ -683,8 +886,8 @@ func (d *Daemon) handleEC2Events(msg *nats.Msg) { // handleEC2RunInstances processes incoming EC2 RunInstances requests func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { - log.Printf("Received message on subject: %s", msg.Subject) - log.Printf("Message data: %s", string(msg.Data)) + slog.Debug("Received message on subject", "subject", msg.Subject) + slog.Debug("Message data", "data", string(msg.Data)) // Initialize runInstancesInput before unmarshaling into it runInstancesInput := &ec2.RunInstancesInput{} @@ -760,7 +963,7 @@ func (d *Daemon) handleEC2RunInstances(msg *nats.Msg) { } msg.Respond(jsonResponse) - // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions) + // Launch the instance infrastructure (QEMU, QMP, NATS subscriptions), this can take sometime err = d.LaunchInstance(instance) if err != nil { @@ -991,6 +1194,14 @@ func (d *Daemon) setupShutdown() { // Close NATS connection d.natsConn.Close() + // Shutdown cluster manager + if d.clusterApp != nil { + log.Println("Shutting down cluster manager...") + if err := d.clusterApp.Shutdown(); err != nil { + log.Printf("Error shutting down cluster manager: %v", err) + } + } + // Write the state to disk err := d.WriteState() if err != nil { @@ -1192,6 +1403,7 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { drive.File = v.NBDURI // Cleanup hostname to point to nbd://localhost from [::] // TODO: Make NBD host config defined, or remote NBD server if not running locally. + // TODO: Add socket support for nbdkit, much faster than TCP drive.File = strings.Replace(drive.File, "[::]", "nbd://127.0.0.1", 1) if v.Boot { @@ -1259,6 +1471,9 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { instance.Config.QMPSocket = qmpSocket + // Temp, wait for nbdkit to start + time.Sleep(2 * time.Second) + // Create a unique error channel for this specific mount request processChan := make(chan int, 1) exitChan := make(chan int, 1) @@ -1280,7 +1495,22 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { return } - cmd.Start() + VMstderr, err := cmd.StderrPipe() + if err != nil { + slog.Error("Failed to pipe STDERR VM", "err", err) + processChan <- 0 + return + } + + err = cmd.Start() + + if err != nil { + slog.Error("Failed to start VM", "err", err) + processChan <- 0 + return + } + + slog.Info("VM started successfully", "pid", cmd.Process.Pid) // TODO: Consider workaround using QMP // (QEMU) query-chardev @@ -1289,15 +1519,19 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { go func() { // TODO: Add a timeout to the scanner scanner := bufio.NewScanner(VMstdout) + + slog.Info("QEMU stdout reader started") + re := regexp.MustCompile(`/dev/pts/(\d+)`) for scanner.Scan() { line := scanner.Text() - slog.Debug("[qemu stderr]", "line", line) + slog.Info("[qemu]", "line", line) matches := re.FindStringSubmatch(line) if len(matches) == 2 { ptsInt, err := strconv.Atoi(matches[1]) + slog.Info("Extracted pts from QEMU output", "pts", ptsInt) if err != nil { slog.Error("Failed to convert pts to int:", "err", err) @@ -1312,16 +1546,20 @@ func (d *Daemon) StartInstance(instance *vm.VM) error { } }() - if err != nil { - slog.Error("Failed to launch VM", "err", err) - processChan <- 0 - return - } + // --- reader for STDERR --- + go func() { + scanner := bufio.NewScanner(VMstderr) + slog.Info("QEMU stderr reader started") + + for scanner.Scan() { + line := scanner.Text() + slog.Error("[qemu-stderr]", "line", line) + } + }() processChan <- cmd.Process.Pid // Read the pts from launch - err = cmd.Wait() if err != nil { @@ -1681,3 +1919,29 @@ func (d *Daemon) handleEC2DescribeInstances(msg *nats.Msg) { slog.Info("handleEC2DescribeInstances completed", "count", len(reservations)) } + +// handleHealthCheck processes NATS health check requests +func (d *Daemon) handleHealthCheck(msg *nats.Msg) { + configHash, err := d.computeConfigHash() + if err != nil { + slog.Error("Failed to compute config hash for health check", "error", err) + configHash = "error" + } + + response := config.NodeHealthResponse{ + Node: d.node, + Status: "running", + ConfigHash: configHash, + Epoch: d.clusterConfig.Epoch, + Uptime: int64(time.Since(d.startTime).Seconds()), + } + + jsonResponse, err := json.Marshal(response) + if err != nil { + slog.Error("handleHealthCheck failed to marshal response", "err", err) + return + } + + msg.Respond(jsonResponse) + slog.Debug("Health check responded", "node", d.node, "epoch", d.clusterConfig.Epoch) +} diff --git a/hive/services/hive/hive.go b/hive/services/hive/hive.go index c5aa2f7e5..d4f2a5027 100644 --- a/hive/services/hive/hive.go +++ b/hive/services/hive/hive.go @@ -12,7 +12,8 @@ import ( var serviceName = "hive" type Service struct { - Config *config.ClusterConfig + Config *config.ClusterConfig + ConfigPath string } func New(cfg interface{}) (svc *Service, err error) { @@ -22,9 +23,13 @@ func New(cfg interface{}) (svc *Service, err error) { return svc, nil } +func (svc *Service) SetConfigPath(path string) { + svc.ConfigPath = path +} + func (svc *Service) Start() (int, error) { utils.WritePidFile(serviceName, os.Getpid()) - err := launchService(svc.Config) + err := launchService(svc.Config, svc.ConfigPath) if err != nil { return 0, err } @@ -49,9 +54,10 @@ func (svc *Service) Reload() (err error) { return nil } -func launchService(config *config.ClusterConfig) (err error) { +func launchService(config *config.ClusterConfig, configPath string) (err error) { d := daemon.NewDaemon(config) + d.SetConfigPath(configPath) slog.Info("Starting Hive daemon ...") err = d.Start() diff --git a/hive/services/nats/nats.go b/hive/services/nats/nats.go index 813cc3525..254ba9878 100644 --- a/hive/services/nats/nats.go +++ b/hive/services/nats/nats.go @@ -1,6 +1,7 @@ package nats import ( + "fmt" "log/slog" "os" @@ -12,12 +13,13 @@ import ( var serviceName = "nats" type Config struct { - Port int `json:"port"` - Host string `json:"host"` - Debug bool `json:"debug"` - LogFile string `json:"log_file"` - DataDir string `json:"data_dir"` - JetStream bool `json:"jetstream"` + ConfigFile string `json:"config_file"` + Port int `json:"port"` + Host string `json:"host"` + Debug bool `json:"debug"` + LogFile string `json:"log_file"` + DataDir string `json:"data_dir"` + JetStream bool `json:"jetstream"` } type Service struct { @@ -56,23 +58,40 @@ func (svc *Service) Reload() (err error) { func launchService(config *Config) (err error) { // Create proper server options - opts := &server.Options{ - Port: config.Port, - Host: config.Host, - Debug: config.Debug, - LogFile: config.LogFile, - JetStream: config.JetStream, + opts := &server.Options{} + + // If configFile set use, otherwise set defaults + if config.ConfigFile != "" { + + opts, err = server.ProcessConfigFile(config.ConfigFile) + + if err != nil { + slog.Error("Failed to process NATS config file", "err", err) + return err + } + + } else { + + opts = &server.Options{ + ConfigFile: config.ConfigFile, + Port: config.Port, + Host: config.Host, + Debug: config.Debug, + LogFile: config.LogFile, + JetStream: config.JetStream, + } + + // Set defaults if not provided + if opts.Port == 0 { + opts.Port = 4222 + } + if opts.Host == "" { + opts.Host = "0.0.0.0" + } + } - //fmt.Println("opts", opts) - - // Set defaults if not provided - if opts.Port == 0 { - opts.Port = 4222 - } - if opts.Host == "" { - opts.Host = "0.0.0.0" - } + fmt.Println(opts) // Initialize new server with options ns, err := server.NewServer(opts) diff --git a/hive/services/viperblockd/viperblockd.go b/hive/services/viperblockd/viperblockd.go index 5172335d2..1a9ac2944 100644 --- a/hive/services/viperblockd/viperblockd.go +++ b/hive/services/viperblockd/viperblockd.go @@ -3,7 +3,6 @@ package viperblockd import ( "encoding/json" "fmt" - "log" "log/slog" "os" "os/signal" @@ -103,10 +102,10 @@ func launchService(cfg *Config) (err error) { } // Subscribe to the viperblock.mount subject - fmt.Println("Connected. Waiting for EBS events") + slog.Info("Connected. Waiting for EBS events") // TODO: Support volume delete and predastore delete bucket - nc.Subscribe("ebs.delete", func(msg *nats.Msg) { + nc.QueueSubscribe("ebs.delete", "hive-workers", func(msg *nats.Msg) { slog.Info("Received message", "data", string(msg.Data)) // Parse the message @@ -122,7 +121,7 @@ func launchService(cfg *Config) (err error) { }) - nc.Subscribe("ebs.unmount", func(msg *nats.Msg) { + nc.QueueSubscribe("ebs.unmount", "hive-workers", func(msg *nats.Msg) { slog.Info("Received message", "data", string(msg.Data)) // Parse the message @@ -171,7 +170,7 @@ func launchService(cfg *Config) (err error) { nc.Publish("ebs.unmount.response", response) }) - nc.Subscribe("ebs.mount", func(msg *nats.Msg) { + nc.QueueSubscribe("ebs.mount", "hive-workers", func(msg *nats.Msg) { slog.Info("Received message:", "data", string(msg.Data)) // Parse the message @@ -182,7 +181,7 @@ func launchService(cfg *Config) (err error) { return } - fmt.Println("Request =>", ebsRequest) + slog.Info("ebs.mount", "request", ebsRequest) var ebsResponse config.EBSMountResponse ebsResponse.Mounted = false @@ -280,8 +279,7 @@ func launchService(cfg *Config) (err error) { return } - fmt.Println("NBD URI =>", nbdUri) - fmt.Println("PORT =>", port) + slog.Info("Mounting volume", "name", ebsRequest.Name, "port", port, "uri", nbdUri) // Execute nbdkit @@ -339,7 +337,7 @@ func launchService(cfg *Config) (err error) { exitChan <- exitCode - fmt.Println("NBDKit exited,, code", exitCode) + slog.Error("NBDKit exited", "code", exitCode) }() // Wait for startup result @@ -369,7 +367,7 @@ func launchService(cfg *Config) (err error) { } default: // nbdkit is still running after 1 second, which means it started successfully - fmt.Println("NBDKit started successfully and is running") + slog.Info("NBDKit started successfully and is running") } ebsResponse.Mounted = true @@ -394,7 +392,7 @@ func launchService(cfg *Config) (err error) { nc.Publish("ebs.mount.response", response) - fmt.Println("Response =>", string(response)) + slog.Debug("Sent ebs.mount response", "response", string(response)) }) // Create a channel to receive shutdown signals @@ -403,14 +401,14 @@ func launchService(cfg *Config) (err error) { // Wait for shutdown signal <-sigChan - log.Println("Shutting down gracefully...") + slog.Info("Shutting down gracefully...") nc.Close() // Close all nbdkit processes cfg.mu.Lock() for _, volume := range cfg.MountedVolumes { - fmt.Println("Killing nbdkit process", volume.PID) + slog.Info("Killing nbdkit process", "pid", volume.PID) utils.KillProcess(volume.PID) } cfg.mu.Unlock() diff --git a/hive/vm/vm.go b/hive/vm/vm.go index 070c39ba1..43804561d 100644 --- a/hive/vm/vm.go +++ b/hive/vm/vm.go @@ -138,7 +138,7 @@ func (cfg *Config) Execute() (*exec.Cmd, error) { } if cfg.NoGraphic { - args = append(args, "-display none") + args = append(args, "-display", "none") } if cfg.MachineType != "" { diff --git a/scripts/create-multi-node.sh b/scripts/create-multi-node.sh new file mode 100755 index 000000000..decdb90e8 --- /dev/null +++ b/scripts/create-multi-node.sh @@ -0,0 +1,67 @@ +# simulate node1 +echo "Multi-node simulation environment" +make + +echo "Adding simulated IPs: 10.11.12.1, 10.11.12.2, 10.11.12.3" + +sudo ip addr add 10.11.12.1/24 dev eth0 + +# simulate node2 +sudo ip addr add 10.11.12.2/24 dev eth0 + +# simulate node3 +sudo ip addr add 10.11.12.3/24 dev eth0 + +echo "Removing old node directories..." + +rm -rf ~/node1/ +rm -rf ~/node2/ +rm -rf ~/node3/ + +echo "Node1 Setup:" + +# Initialize node1 +./bin/hive admin init \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node1 \ +--bind 10.11.12.1 \ +--cluster-bind 10.11.12.1 \ +--port 4432 \ +--hive-dir ~/node1/ \ +--config-dir ~/node1/config/ + +# Start node1 +./scripts/start-dev.sh ~/node1/ + +echo "Node2 Setup:" + +# Join cluster +./bin/hive admin join \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node2 \ +--bind 10.11.12.2 \ +--cluster-bind 10.11.12.2 \ +--cluster-routes 10.11.12.1:4248 \ +--host 10.11.12.1:4432 \ +--data-dir ~/node2/ \ +--config-dir ~/node2/config/ \ + +# Start node2 +./scripts/start-dev.sh ~/node2/ + +echo "Node3 Setup:" + +./bin/hive admin join \ +--region ap-southeast-2 \ +--az ap-southeast-2a \ +--node node3 \ +--bind 10.11.12.3 \ +--cluster-bind 10.11.12.3 \ +--cluster-routes 10.11.12.1:4248 \ +--host 10.11.12.1:4432 \ +--data-dir ~/node3/ \ +--config-dir ~/node3/config/ + +./scripts/start-dev.sh ~/node3/ \ No newline at end of file diff --git a/scripts/start-dev.sh b/scripts/start-dev.sh index 03b7c5a17..05190c62d 100755 --- a/scripts/start-dev.sh +++ b/scripts/start-dev.sh @@ -2,6 +2,8 @@ # Start Hive development environment # This script starts all required services in the correct order using Hive service commands +# Usage: ./scripts/start-dev.sh [data-dir] +# data-dir: Optional data directory path (default: ~/hive) set -e @@ -9,11 +11,14 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" MULGA_ROOT="$(cd "$PROJECT_ROOT/.." && pwd)" +# Accept optional data directory argument +DATA_DIR="${1:-$HOME/hive}" + # Configuration paths -# Use CONFIG_DIR environment variable if set, otherwise default to ~/hive/config -CONFIG_DIR="${CONFIG_DIR:-$HOME/hive/config}" +# Use CONFIG_DIR environment variable if set, otherwise derive from DATA_DIR +CONFIG_DIR="${CONFIG_DIR:-$DATA_DIR/config}" +echo "Using data directory: $DATA_DIR" echo "Using configuration directory: $CONFIG_DIR" -DATA_DIR="$HOME/hive" LOGS_DIR="$DATA_DIR/logs" WAL_DIR="$DATA_DIR/hive" @@ -106,7 +111,7 @@ check_service() { done echo " ⚠️ $name may not be responding on port $port (continuing anyway)" - return 1 + } # Pre-flight, compile latest @@ -117,11 +122,18 @@ make build echo "" echo "1️⃣ Starting NATS server..." -export HIVE_NATS_HOST=0.0.0.0 -export HIVE_NATS_PORT=4222 -export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/ -export HIVE_NATS_JETSTREAM=false -export HIVE_NATS_DEBUG=true +if [ -f "$CONFIG_DIR/nats/nats.conf" ]; then + export HIVE_CONFIG_PATH=$CONFIG_DIR/nats/nats.conf + echo " Using NATS config file: $CONFIG_DIR/nats/nats.conf" +else + echo " ⚠️ NATS config file not found at $CONFIG_DIR/nats/nats.conf, using defaults" + # TODO: Confirm, settings file will overwrite env vars (e.g multi-node config) + export HIVE_NATS_HOST=0.0.0.0 + export HIVE_NATS_PORT=4222 + export HIVE_NATS_DATA_DIR=$DATA_DIR/nats/ + export HIVE_NATS_JETSTREAM=false + export HIVE_NATS_DEBUG=true +fi # Use air for hot reloading (dev!) #NATS_CMD="air -c .air-nats.toml" @@ -130,13 +142,13 @@ NATS_CMD="./bin/hive service nats start" start_service "nats" "$NATS_CMD" #start_service "nats" "go run cmd/hive/main.go service nats start" -check_service "NATS" "4222" +#check_service "NATS" "4222" # 2️⃣ Start Predastore -echo "" echo "2️⃣ Starting Predastore..." +unset HIVE_CONFIG_PATH -export HIVE_PREDASTORE_BASE_PATH=~/hive/predastore/ +export HIVE_PREDASTORE_BASE_PATH=$DATA_DIR/predastore/ export HIVE_PREDASTORE_CONFIG_PATH=$CONFIG_DIR/predastore/predastore.toml export HIVE_PREDASTORE_TLS_CERT=$CONFIG_DIR/server.pem export HIVE_PREDASTORE_TLS_KEY=$CONFIG_DIR/server.key @@ -151,6 +163,12 @@ PREDASTORE_CMD="./bin/hive service predastore start" start_service "predastore" "$PREDASTORE_CMD" check_service "Predastore" "8443" +#else +# echo "" +# echo "2️⃣ Skipping Predastore for multi-node setup (TODO)" +#fi + + # 3️⃣ Start Viperblock echo "" echo "3️⃣ Starting Viperblock..." @@ -219,7 +237,8 @@ unset HIVE_NATS_HOST unset HIVE_PREDASTORE_HOST export HIVE_BASE_DIR=$CONFIG_DIR export HIVE_CONFIG_PATH=$CONFIG_DIR/hive.toml -export HIVE_AWSGW_HOST="0.0.0.0:9999" +# Overwritten by config file +#export HIVE_AWSGW_HOST="0.0.0.0:9999" export HIVE_AWSGW_TLS_CERT=$CONFIG_DIR/server.pem export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key @@ -228,7 +247,9 @@ export HIVE_AWSGW_TLS_KEY=$CONFIG_DIR/server.key AWSGW_CMD="./bin/hive service awsgw start" start_service "awsgw" "$AWSGW_CMD" -check_service "awsgw" "9999" + +# TODO: Need host:ip support, depending on node +#check_service "awsgw" "9999" echo "" diff --git a/scripts/start-multi-nats.sh b/scripts/start-multi-nats.sh new file mode 100644 index 000000000..8b1a4bb14 --- /dev/null +++ b/scripts/start-multi-nats.sh @@ -0,0 +1,3 @@ +#!/bin/sh + +HIVE_CONFIG_PATH=~/node1/config/nats/nats.conf ./bin/hive service nats start diff --git a/scripts/start-multi-node.sh b/scripts/start-multi-node.sh new file mode 100755 index 000000000..2c8cf7e22 --- /dev/null +++ b/scripts/start-multi-node.sh @@ -0,0 +1,12 @@ +#!/bin/sh + +# Start node1 +./scripts/start-dev.sh ~/node1/ + + +# Start node2 +./scripts/start-dev.sh ~/node2/ + + +# Start node3 +./scripts/start-dev.sh ~/node3/ diff --git a/scripts/stop-dev.sh b/scripts/stop-dev.sh index d2cc5924c..8463aa3f8 100755 --- a/scripts/stop-dev.sh +++ b/scripts/stop-dev.sh @@ -2,6 +2,19 @@ # Stop Hive development environment # This script stops all services started by start-dev.sh +# Usage: ./scripts/stop-dev.sh +# Note: Services are stopped using PID files, so data-dir is not required + +# Accept optional data directory argument +DATA_DIR="${1:-$HOME/hive}" + +# If path is provided, use for pid location, else the default for Hive +if [ -n "$1" ]; then + PID_DIR="$DATA_DIR/logs" +else + PID_DIR="" +fi + SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" @@ -18,10 +31,18 @@ echo "Stopping Hive development environment..." # Function to stop service by PID file stop_service() { local name="$1" + local pidpath="$2" - echo "🛑 Stopping $name..." + echo "🛑 Stopping $name $pidpath..." - $PROJECT_ROOT/bin/hive service $name stop + # Workaround for local development for multi-node config on single instance + if [ -n "$pidpath" ]; then + kill -SIGTERM `cat $pidpath/$name.pid` + sleep 1 + else + # Correct graceful shutdown via hive binary, waits for clean exit + $PROJECT_ROOT/bin/hive service $name stop + fi echo "Status: $?" @@ -39,19 +60,19 @@ stop_service() { echo "Stopping services..." echo "" # Stop Hive daemon/gateway first (it will terminate running instances, unmount nbd devices) -stop_service "hive" +stop_service "hive" "$PID_DIR" # Stop AWSGW -stop_service "awsgw" +stop_service "awsgw" "$PID_DIR" # Stop Viperblock -stop_service "viperblock" +stop_service "viperblock" "$PID_DIR" # Stop Predastore -stop_service "predastore" +stop_service "predastore" "$PID_DIR" # Stop NATS -stop_service "nats" +stop_service "nats" "$PID_DIR" echo "" echo "✅ Hive development environment stopped" diff --git a/scripts/stop-multi-node.sh b/scripts/stop-multi-node.sh new file mode 100755 index 000000000..958dccca3 --- /dev/null +++ b/scripts/stop-multi-node.sh @@ -0,0 +1,17 @@ + +#sudo ip addr del 10.11.12.1/24 dev eth0 +#sudo ip addr del 10.11.12.2/24 dev eth0 +#sudo ip addr del 10.11.12.3/24 dev eth0 + +# TODO: Reverse order, predastore running on node1 +# BUG: Need to wait for each QEMU to flush/commit WAL to disk > S3, before shutdown, else corruption may occur +# BUG: nbdkit: ramdisk.3: debug: sending error reply: Cannot send after transport endpoint shutdown +# TODO: Fallback, if WAL on local disk not flushed, on boot, viperblock should commit and verify. Requires UT and test-cases + +./scripts/stop-dev.sh ~/node3/ + +./scripts/stop-dev.sh ~/node2/ + +./scripts/stop-dev.sh ~/node1/ + +