Fixes:

Updates bulletin board command to use PrettyLogger and slog instead of logrus. Renames `PrettyLogger.InitDefault()` to `pl.SetUpLogrusAndSlog(logLevel)`. Removes `config.InitConfig()` and replaces it with `config.InitGlobal()` in multiple commands. Updates `main.go` in the `bulletin-board` command to use the new `pl` and `slog` packages. Renames `main.go` in the `clients` command to `newClient.go` and updates it to use PrettyLogger and slog instead of logrus. Renames `main.go` in the `node` command to `newNode.go` and updates it to use PrettyLogger and slog instead of logrus. Updates the `client` and `node` packages to use PrettyLogger and slog instead of logrus. Description: This change updates the logging mechanism in the `bulletin-board`, `clients`, and `node` commands to use PrettyLogger and slog instead of logrus. This change removes the `PrettyLogger` and `logrus` packages from the `bulletin-board`, `clients`, and `node` commands and replaces them with `pl` and `slog` packages. The `initConfig` function was also replaced with `initGlobal` in the `clients` and `nodes` commands. This change should improve the stability and reliability of the logging mechanism in these commands.
HannahMarsh · Jun 24, 2024 · 9033c8c · 9033c8c
1 parent f9e1b3c
commit 9033c8c
Show file tree

Hide file tree

Showing 24 changed files with 855 additions and 306 deletions.
diff --git a/cmd/bulletin-board/main.go b/cmd/bulletin-board/main.go
@@ -1,19 +1,17 @@
 package main
 
 import (
-	"context"
 	"errors"
 	"flag"
 	"fmt"
-	"github.com/HannahMarsh/PrettyLogger"
-	"github.com/HannahMarsh/pi_t-experiment/cmd/config"
+	pl "github.com/HannahMarsh/PrettyLogger"
+	"github.com/HannahMarsh/pi_t-experiment/config"
 	"github.com/HannahMarsh/pi_t-experiment/internal/bulletin_board"
 	"net/http"
 	"os"
 	"os/signal"
 	"syscall"
 
-	"github.com/sirupsen/logrus"
 	"go.uber.org/automaxprocs/maxprocs"
 	"golang.org/x/exp/slog"
 
@@ -32,15 +30,15 @@ func main() {
 
 	flag.Parse()
 
+	pl.SetUpLogrusAndSlog(*logLevel)
+
 	// set GOMAXPROCS
 	if _, err := maxprocs.Set(); err != nil {
 		slog.Error("failed set max procs", err)
 		os.Exit(1)
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
-	if err := config.InitConfig(); err != nil {
+	if err := config.InitGlobal(); err != nil {
 		slog.Error("failed to init config", err)
 		os.Exit(1)
 	}
@@ -49,37 +47,46 @@ func main() {
 
 	host := cfg.BulletinBoard.Host
 	port := cfg.BulletinBoard.Port
+	url := fmt.Sprintf("https://%s:%d", host, port)
 
 	slog.Info("⚡ init Bulletin board")
 
-	PrettyLogger.InitDefault()
-	logrus.SetLevel(PrettyLogger.ConvertLogLevel(*logLevel))
-
 	bulletinBoard := bulletin_board.NewBulletinBoard(cfg)
 
-	go bulletinBoard.StartRuns()
+	go func() {
+		err := bulletinBoard.StartRuns()
+		if err != nil {
+			slog.Error("failed to start runs", err)
+			config.GlobalCancel()
+		}
+	}()
 
-	http.HandleFunc("/register", bulletinBoard.HandleRegisterNode)
+	http.HandleFunc("/registerNode", bulletinBoard.HandleRegisterNode)
+	http.HandleFunc("/registerClient", bulletinBoard.HandleRegisterClient)
+	http.HandleFunc("/registerIntentToSend", bulletinBoard.HandleRegisterIntentToSend)
 	http.HandleFunc("/update", bulletinBoard.HandleUpdateNodeInfo)
 	http.HandleFunc("/nodes", bulletinBoard.HandleGetActiveNodes)
 
 	go func() {
-		address := fmt.Sprintf(":%d", port)
-		if err2 := http.ListenAndServe(address, nil); err2 != nil && !errors.Is(err2, http.ErrServerClosed) {
-			slog.Error("failed to start HTTP server", err2)
+		if err := http.ListenAndServe(fmt.Sprintf(":%d", port), nil); err != nil {
+			if errors.Is(err, http.ErrServerClosed) {
+				slog.Info("HTTP server closed")
+			} else {
+				slog.Error("failed to start HTTP server", err)
+			}
 		}
 	}()
 
-	slog.Info("🌏 start node...", "address", fmt.Sprintf("https://%s:%d", host, port))
+	slog.Info("🌏 starting bulletin board...", "address", url)
 
 	quit := make(chan os.Signal, 1)
 	signal.Notify(quit, os.Interrupt, syscall.SIGTERM)
 
 	select {
 	case v := <-quit:
-		cancel()
+		config.GlobalCancel()
 		slog.Info("signal.Notify", v)
-	case done := <-ctx.Done():
+	case done := <-config.GlobalCtx.Done():
 		slog.Info("ctx.Done", done)
 	}
 }
diff --git a/cmd/clients/main.go b/cmd/clients/main.go
@@ -1,18 +1,15 @@
 package main
 
 import (
-	"bytes"
-	"context"
-	"encoding/json"
+	"errors"
 	"flag"
 	"fmt"
-	"github.com/HannahMarsh/PrettyLogger"
-	"github.com/HannahMarsh/pi_t-experiment/cmd/config"
-	"github.com/HannahMarsh/pi_t-experiment/internal/api"
-	"github.com/sirupsen/logrus"
+	pl "github.com/HannahMarsh/PrettyLogger"
+	"github.com/HannahMarsh/pi_t-experiment/config"
+	"github.com/HannahMarsh/pi_t-experiment/internal/client"
+	"github.com/HannahMarsh/pi_t-experiment/pkg/utils"
 	"go.uber.org/automaxprocs/maxprocs"
 	"golang.org/x/exp/slog"
-	"io"
 	"net/http"
 	"os"
 	"os/signal"
@@ -24,6 +21,7 @@ import (
 
 func main() {
 	// Define command-line flags
+	id := flag.Int("id", -1, "ID of the newClient (required)")
 	logLevel := flag.String("log-level", "debug", "Log level")
 
 	flag.Usage = func() {
@@ -35,82 +33,83 @@ func main() {
 
 	flag.Parse()
 
+	// Check if the required flag is provided
+	if *id == -1 {
+		_, _ = fmt.Fprintf(os.Stderr, "Error: the -id flag is required\n")
+		flag.Usage()
+		os.Exit(2)
+	}
+
+	pl.SetUpLogrusAndSlog(*logLevel)
+
 	// set GOMAXPROCS
-	_, err := maxprocs.Set()
-	if err != nil {
+	if _, err := maxprocs.Set(); err != nil {
 		slog.Error("failed set max procs", err)
 		os.Exit(1)
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
-	if err = config.InitConfig(); err != nil {
+	if err := config.InitGlobal(); err != nil {
 		slog.Error("failed to init config", err)
 		os.Exit(1)
 	}
 
 	cfg := config.GlobalConfig
 
-	slog.Info("⚡ init client", "heartbeat_interval", cfg.HeartbeatInterval)
-
-	PrettyLogger.InitDefault()
-	logrus.SetLevel(PrettyLogger.ConvertLogLevel(*logLevel))
-
-	node_addresses := make(map[int][]string, 0)
-	for _, n := range cfg.Nodes {
-		if _, ok := node_addresses[n.ID]; !ok {
-			node_addresses[n.ID] = make([]string, 0)
+	var clientConfig *config.Client
+	for _, client := range cfg.Clients {
+		if client.ID == *id {
+			clientConfig = &client
+			break
 		}
-		node_addresses[n.ID] = append(node_addresses[n.ID], fmt.Sprintf("http://%s:%d", n.Host, n.Port))
 	}
 
+	if clientConfig == nil {
+		slog.Error("invalid id", errors.New(fmt.Sprintf("failed to get newClient config for id=%d", *id)))
+		os.Exit(1)
+	}
+
+	slog.Info("⚡ init newClient", "heartbeat_interval", cfg.HeartbeatInterval, "id", *id)
+
+	baddress := fmt.Sprintf("http://%s:%d", cfg.BulletinBoard.Host, cfg.BulletinBoard.Port)
+
+	var newClient *client.Client
 	for {
-		for id, addresses := range node_addresses {
-			for _, addr := range addresses {
-				addr := addr
-				go func() {
-					var msgs []api.Message = make([]api.Message, 0)
-					for i, _ := range node_addresses {
-						if i != id {
-							msgs = append(msgs, api.Message{
-								From: id,
-								To:   i,
-								Msg:  fmt.Sprintf("msg %d", i),
-							})
-						}
-					}
-					if data, err2 := json.Marshal(msgs); err2 != nil {
-						slog.Error("failed to marshal msgs", err2)
-					} else {
-						url := addr + "/requestMsg"
-						slog.Info("Sending add msg request.", "url", url, "num_onions", len(msgs))
-						if resp, err3 := http.Post(url, "application/json", bytes.NewBuffer(data)); err3 != nil {
-							slog.Error("failed to send POST request with msgs to node", err3)
-						} else {
-							defer func(Body io.ReadCloser) {
-								if err4 := Body.Close(); err4 != nil {
-									fmt.Printf("error closing response body: %v\n", err2)
-								}
-							}(resp.Body)
-							if resp.StatusCode != http.StatusCreated {
-								slog.Info("failed to send msgs to node", "status_code", resp.StatusCode, "status", resp.Status)
-							}
-						}
-					}
-				}()
-			}
+		if n, err := client.NewClient(clientConfig.ID, clientConfig.Host, clientConfig.Port, baddress); err != nil {
+			slog.Error("failed to create new client. Trying again in 5 seconds. ", err)
+			time.Sleep(5 * time.Second)
+			continue
+		} else {
+			newClient = n
+			break
 		}
-		time.Sleep(time.Duration(2 * time.Second))
 	}
 
+	http.HandleFunc("/receive", newClient.HandleReceive)
+	http.HandleFunc("/start", newClient.HandleStartRun)
+
+	go func() {
+		address := fmt.Sprintf(":%d", clientConfig.Port)
+		if err2 := http.ListenAndServe(address, nil); err2 != nil && !errors.Is(err2, http.ErrServerClosed) {
+			slog.Error("failed to start HTTP server", err2)
+		}
+	}()
+
+	slog.Info("🌏 start newClient...", "address", fmt.Sprintf("http://%s:%d", clientConfig.Host, clientConfig.Port))
+
+	nodeAddresses := utils.Map(cfg.Nodes, func(n config.Node) string {
+		return fmt.Sprintf("http://%s:%d", n.Host, n.Port)
+	})
+
+	go newClient.StartGeneratingMessages(nodeAddresses)
+
 	quit := make(chan os.Signal, 1)
 	signal.Notify(quit, os.Interrupt, syscall.SIGTERM)
 
 	select {
 	case v := <-quit:
-		cancel()
+		config.GlobalCancel()
 		slog.Info("signal.Notify", v)
-	case done := <-ctx.Done():
+	case done := <-config.GlobalCtx.Done():
 		slog.Info("ctx.Done", done)
 	}
 

diff --git a/cmd/node/main.go b/cmd/node/main.go
@@ -1,14 +1,12 @@
 package main
 
 import (
-	"context"
 	"errors"
 	"flag"
 	"fmt"
-	"github.com/HannahMarsh/PrettyLogger"
-	"github.com/HannahMarsh/pi_t-experiment/cmd/config"
+	pl "github.com/HannahMarsh/PrettyLogger"
+	"github.com/HannahMarsh/pi_t-experiment/config"
 	"github.com/HannahMarsh/pi_t-experiment/internal/node"
-	"github.com/sirupsen/logrus"
 	"go.uber.org/automaxprocs/maxprocs"
 	"golang.org/x/exp/slog"
 	"net/http"
@@ -23,6 +21,7 @@ import (
 func main() {
 	// Define command-line flags
 	id := flag.Int("id", -1, "ID of the newNode (required)")
+	isMixer := flag.Bool("mixer", false, "Included if this node is a mixer")
 	logLevel := flag.String("log-level", "debug", "Log level")
 
 	flag.Usage = func() {
@@ -36,23 +35,20 @@ func main() {
 
 	// Check if the required flag is provided
 	if *id == -1 {
-		if _, err := fmt.Fprintf(os.Stderr, "Error: the -id flag is required\n"); err != nil {
-			slog.Error("Error: the -id flag is required\n", err)
-		}
+		_, _ = fmt.Fprintf(os.Stderr, "Error: the -id flag is required\n")
 		flag.Usage()
 		os.Exit(2)
 	}
 
+	pl.SetUpLogrusAndSlog(*logLevel)
+
 	// set GOMAXPROCS
-	_, err := maxprocs.Set()
-	if err != nil {
+	if _, err := maxprocs.Set(); err != nil {
 		slog.Error("failed set max procs", err)
 		os.Exit(1)
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-
-	if err = config.InitConfig(); err != nil {
+	if err := config.InitGlobal(); err != nil {
 		slog.Error("failed to init config", err)
 		os.Exit(1)
 	}
@@ -74,31 +70,30 @@ func main() {
 
 	slog.Info("⚡ init newNode", "heartbeat_interval", cfg.HeartbeatInterval, "id", *id)
 
-	PrettyLogger.InitDefault()
-	logrus.SetLevel(PrettyLogger.ConvertLogLevel(*logLevel))
-
 	baddress := fmt.Sprintf("http://%s:%d", cfg.BulletinBoard.Host, cfg.BulletinBoard.Port)
 
 	var newNode *node.Node
 	for {
-		if newNode, err = node.NewNode(nodeConfig.ID, nodeConfig.Host, nodeConfig.Port, baddress); err != nil {
+		if n, err := node.NewNode(nodeConfig.ID, nodeConfig.Host, nodeConfig.Port, baddress); err != nil {
 			slog.Error("failed to create newNode. Trying again in 5 seconds. ", err)
-
 			time.Sleep(5 * time.Second)
 			continue
 		} else {
+			newNode = n
 			break
 		}
 	}
 
-	http.HandleFunc("/receive", newNode.HandleReceive)
-	http.HandleFunc("/requestMsg", newNode.HandleClientRequest)
+	http.HandleFunc("/receiveOnion", newNode.HandleReceiveOnion)
 	http.HandleFunc("/start", newNode.HandleStartRun)
 
 	go func() {
-		address := fmt.Sprintf(":%d", nodeConfig.Port)
-		if err2 := http.ListenAndServe(address, nil); err2 != nil && !errors.Is(err2, http.ErrServerClosed) {
-			slog.Error("failed to start HTTP server", err2)
+		if err := http.ListenAndServe(fmt.Sprintf(":%d", nodeConfig.Port), nil); err != nil {
+			if errors.Is(err, http.ErrServerClosed) {
+				slog.Info("HTTP server closed")
+			} else {
+				slog.Error("failed to start HTTP server", err)
+			}
 		}
 	}()
 
@@ -109,9 +104,9 @@ func main() {
 
 	select {
 	case v := <-quit:
-		cancel()
+		config.GlobalCancel()
 		slog.Info("signal.Notify", v)
-	case done := <-ctx.Done():
+	case done := <-config.GlobalCtx.Done():
 		slog.Info("ctx.Done", done)
 	}