module go-distributed-system
?
go 1.17
?
require (
  github.com/golang/protobuf v1.5.0
  google.golang.org/grpc v1.27.0
  google.golang.org/protobuf v1.27.1
)

syntax = "proto3";
?
package core;
option go_package = ".;core";
?
message Request {
  string action = 1;
}
?
message Response {
  string data = 1;
}
?
service NodeService {
  rpc ReportStatus(Request) returns (Response){};       // Simple RPC
  rpc AssignTask(Request) returns (stream Response){};  // Server-Side RPC
}

mkdir core
protoc --go_out=./core \
    --go-grpc_out=./core \
    node.proto

package core
?
import (
  "context"
)
?
type NodeServiceGrpcServer struct {
  UnimplementedNodeServiceServer
?
  // channel to receive command
  CmdChannel chan string
}
?
func (n NodeServiceGrpcServer) ReportStatus(ctx context.Context, request *Request) (*Response, error) {
  return &Response{Data: "ok"}, nil
}
?
func (n NodeServiceGrpcServer) AssignTask(request *Request, server NodeService_AssignTaskServer) error {
  for {
    select {
    case cmd := <-n.CmdChannel:
      // receive command and send to worker node (client)
      if err := server.Send(&Response{Data: cmd}); err != nil {
        return err
      }
    }
  }
}
?
var server *NodeServiceGrpcServer
?
// GetNodeServiceGrpcServer singleton service
func GetNodeServiceGrpcServer() *NodeServiceGrpcServer {
  if server == nil {
    server = &NodeServiceGrpcServer{
      CmdChannel: make(chan string),
    }
  }
  return server
}

package core
?
import (
  "github.com/gin-gonic/gin"
  "google.golang.org/grpc"
  "net"
  "net/http"
)
?
// MasterNode is the node instance
type MasterNode struct {
  api     *gin.Engine            // api server
  ln      net.Listener           // listener
  svr     *grpc.Server           // grpc server
  nodeSvr *NodeServiceGrpcServer // node service
}
?
func (n *MasterNode) Init() (err error) {
  // TODO: implement me
  panic("implement me")
}
?
func (n *MasterNode) Start() {
  // TODO: implement me
  panic("implement me")
}
?
var node *MasterNode
?
// GetMasterNode returns the node instance
func GetMasterNode() *MasterNode {
  if node == nil {
    // node
    node = &MasterNode{}
?
    // initialize node
    if err := node.Init(); err != nil {
      panic(err)
    }
  }
?
  return node
}

func (n *MasterNode) Init() (err error) {
  // grpc server listener with port as 50051
  n.ln, err = net.Listen("tcp", ":50051")
  if err != nil {
    return err
  }
?
  // grpc server
  n.svr = grpc.NewServer()
?
  // node service
  n.nodeSvr = GetNodeServiceGrpcServer()
?
  // register node service to grpc server
  RegisterNodeServiceServer(node.svr, n.nodeSvr)
?
  // api
  n.api = gin.Default()
  n.api.POST("/tasks", func(c *gin.Context) {
    // parse payload
    var payload struct {
      Cmd string `json:"cmd"`
    }
    if err := c.ShouldBindJSON(&payload); err != nil {
      c.AbortWithStatus(http.StatusBadRequest)
      return
    }
?
    // send command to node service
    n.nodeSvr.CmdChannel <- payload.Cmd
?
    c.AbortWithStatus(http.StatusOK)
  })
?
  return nil
}

func (n *MasterNode) Start() {
  // start grpc server
  go n.svr.Serve(n.ln)
?
  // start api server
  _ = n.api.Run(":9092")
?
  // wait for exit
  n.svr.Stop()
}

package core
?
import (
  "context"
  "google.golang.org/grpc"
  "os/exec"
)
?
type WorkerNode struct {
  conn *grpc.ClientConn  // grpc client connection
  c    NodeServiceClient // grpc client
}
?
func (n *WorkerNode) Init() (err error) {
  // connect to master node
  n.conn, err = grpc.Dial("localhost:50051", grpc.WithInsecure())
  if err != nil {
    return err
  }
?
  // grpc client
  n.c = NewNodeServiceClient(n.conn)
?
  return nil
}
?
func (n *WorkerNode) Start() {
  // log
  fmt.Println("worker node started")
?
  // report status
  _, _ = n.c.ReportStatus(context.Background(), &Request{})
?
  // assign task
  stream, _ := n.c.AssignTask(context.Background(), &Request{})
  for {
    // receive command from master node
    res, err := stream.Recv()
    if err != nil {
      return
    }
?
    // log command
    fmt.Println("received command: ", res.Data)
?
    // execute command
    parts := strings.Split(res.Data, " ")
    if err := exec.Command(parts[0], parts[1:]...).Run(); err != nil {
      fmt.Println(err)
    }
  }
}
?
var workerNode *WorkerNode
?
func GetWorkerNode() *WorkerNode {
  if workerNode == nil {
    // node
    workerNode = &WorkerNode{}
?
    // initialize node
    if err := workerNode.Init(); err != nil {
      panic(err)
    }
  }
?
  return workerNode
}

package main
?
import (
  "go-distributed-system/core"
  "os"
)
?
func main() {
  nodeType := os.Args[0]
  switch nodeType {
  case "master":
    core.GetMasterNode().Start()
  case "worker":
    core.GetWorkerNode().Start()
  default:
    panic("invalid node type")
  }
}

[GIN-debug] [WARNING] Creating an Engine instance with the Logger and Recovery middleware already attached.
?
[GIN-debug] [WARNING] Running in "debug" mode. Switch to "release" mode in production.
 - using env:   export GIN_MODE=release
 - using code:  gin.SetMode(gin.ReleaseMode)
?
[GIN-debug] POST   /tasks                    --> go-distributed-system/core.(*MasterNode).Init.func1 (3 handlers)
[GIN-debug] [WARNING] You trusted all proxies, this is NOT safe. We recommend you to set a value.
Please check https://pkg.go.dev/github.com/gin-gonic/gin#readme-don-t-trust-all-proxies for details.
[GIN-debug] Listening and serving HTTP on :9092