initial

Author: Ole

.gitignore

@@ -0,0 +1,4 @@
+.idea/
+src
+cbsd-mq-router
+log

Makefile

@@ -0,0 +1,10 @@
+all: 
+	@./build.sh
+clean:
+	rm -f bs_router
+install: all
+	install bs_router /usr/local/sbin
+	install rc.d/bs_router /usr/local/etc/rc.d/bs_router
+uninstall: 
+	rm -f /usr/local/sbin/bs_router /usr/local/etc/rc.d/bs_router
+

README.md

@@ -0,0 +1,25 @@
+# cbsd-mq-router
+
+CBSD message queue router
+
+Client + sample: https://github.com/cbsd/bs_router-client
+
+# Installation
+
+setenv GOPATH $( realpath . )
+
+go get
+
+go build
+
+pkg update -f
+
+pkg install -y beanstalkd
+
+service beanstalkd enable
+
+sysrc beanstalkd_flags="-l 127.0.0.1"
+
+service beanstalkd start
+
+./bs_router

beanstalk.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"fmt"
+	"time"
+	"encoding/json"
+	"github.com/beanstalkd/go-beanstalk"
+)
+
+
+// beanstalk config struct
+type BeanstalkConfig struct {
+	Uri			string	`json:"uri"`
+	Tube			string	`json:"tube"`
+	ReplyTubePrefix		string	`json:"reply_tube_prefix"`
+	ReconnectTimeout	int	`json:"reconnect_timeout"`
+	ReserveTimeout		int	`json:"reserve_timeout"`
+	PublishTimeout		int	`json:"publish_timeout"`
+	LogDir			string	`json:"logdir"`
+}
+
+func beanstalkdPublish(config BeanstalkConfig, tube string, body []byte) error {
+
+	amqpURI := config.Uri
+	c, err := beanstalk.Dial("tcp", amqpURI)
+
+	if err != nil {
+		Infof("Publish/callback: unable connect to beanstalkd broker:%s", err)
+		return nil
+	}
+
+	mytube := &beanstalk.Tube{Conn: c, Name: tube}
+	id, err := mytube.Put([]byte(body), 1, 0, time.Duration(config.PublishTimeout)*time.Second)
+	if err != nil {
+		Infof("\nPublish err: %d\n",err)
+		return err
+	}
+	Infof("\nPublish id: %d\n",id)
+
+	return nil
+}
+
+func beanstalkdLoop(config BeanstalkConfig) error {
+	for {
+		beanstalkdConsume(config)
+		Infof("broker disconnected, sleep and retry:%d\n", config.ReconnectTimeout)
+		time.Sleep(time.Duration(config.ReconnectTimeout) * time.Second)
+	}
+	return nil
+}
+
+func WakeOnJob(ch chan bool, config BeanstalkConfig, id uint64, body []byte) {
+
+	Infof("\nwake up and delete job id: %d\n",id)
+	comment := Comment{}
+	Infof("\nWI: %d\n",id)
+	comment.JobID = id
+	response := fmt.Sprintf("%v", comment.JobID)
+	Infof("response %s\n", response)
+	//callback
+	Infof("recv msg: %s", string(body))
+	err := json.Unmarshal(body, &comment)
+	if err != nil {
+			Infof("json decode error %s", err)
+	}
+	callbackQueueName := fmt.Sprintf("%s%d",config.ReplyTubePrefix,comment.JobID)
+	Infof("callback queue name: %s\n",callbackQueueName)
+	err, cbsdTask := DoProcess(&comment, config.LogDir)
+	if err != nil {
+		Errorf("doprocess error:", err)
+		panic(err)
+	}
+	b, err := json.Marshal(cbsdTask)
+	if err != nil {
+		Errorf("error:", err)
+	}
+	Infof("FINE: %s\n",b)
+	err = beanstalkdPublish(config,callbackQueueName,b)
+	ch <- true
+}
+
+
+func beanstalkdConsume(config BeanstalkConfig) error {
+
+	amqpURI := config.Uri
+	tube := config.Tube
+
+	c, err := beanstalk.Dial("tcp", amqpURI)
+
+	if err != nil {
+		Infof("Unable connect to beanstalkd broker:%s", err)
+		return nil
+	}
+
+	Infof("Subscribe tube: %s, reserve timeout: %d", tube, config.ReserveTimeout)
+
+	c.TubeSet = *beanstalk.NewTubeSet(c, tube)
+
+	for {
+		// The BS library does not understand the network/BS problems and hangs forever.
+		// ping in backround?
+		id, body, err := c.Reserve(time.Duration(config.ReserveTimeout) * time.Second)
+
+		if err != nil {
+			Infof("\nid: %d, res: %s\n",id, err.Error())
+		}
+
+		if id == 0 {
+			continue
+		}
+		c.Delete(id)
+
+		ch := make(chan bool)
+		go WakeOnJob(ch, config, id, body)
+	}
+
+	return nil
+}

bget.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+// get bhyve properties via cbsd bget command
+func bget(jname string, properties string) string {
+	var result string
+	// todo: rewrite to SQLite3
+	cmdStr := fmt.Sprintf("/usr/local/bin/cbsd bget jname=%s mode=quiet %s", jname, properties)
+	cmdArgs := strings.Fields(cmdStr)
+	cmd := exec.Command(cmdArgs[0], cmdArgs[1:len(cmdArgs)]...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		Infof("bget cmd.Run() failed (cbsd bget jname=%s mode=quiet %s) with %s\n", jname, properties, err)
+		return ""
+	}
+
+	result=(string(out))
+	fmt.Printf("bget str: [%s]\n", result)
+
+	return result
+}

bhyve-dsk.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"fmt"
+	"os/exec"
+	"strings"
+)
+
+// get bhyve properties via cbsd bhyve-dsk command
+func bhyvedsk(jname string, properties string) string {
+	var result string
+	// todo: rewrite to SQLite3
+	cmdStr := fmt.Sprintf("/usr/local/bin/cbsd bhyve-dsk mode=get jname=%s %s", jname, properties)
+	cmdArgs := strings.Fields(cmdStr)
+	cmd := exec.Command(cmdArgs[0], cmdArgs[1:len(cmdArgs)]...)
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		Infof("bhyve-dsk cmd.Run() failed (cbsd bhyve-dsk mode=get jname=%s %s) with %s\n", jname, properties, err)
+		return ""
+	}
+	result=(string(out))
+	fmt.Printf("bhyve-dsk str: [%s]\n", result)
+
+	return result
+}

build.sh

@@ -0,0 +1,26 @@
+#!/bin/sh
+export PATH="/sbin:/bin:/usr/sbin:/usr/bin:/usr/local/sbin:/usr/local/bin"
+pgm="${0##*/}"		# Program basename
+progdir="${0%/*}"	# Program directory
+
+cd ${proddir}
+
+# Check go install
+if [ -z "$( which go )" ]; then
+	echo "error: Go is not installed. Please install go: pkg install -y lang/go"
+	exit 1
+fi
+
+# Check go version
+GOVERS="$( go version | cut -d " " -f 3 )"
+if [ -z "${GOVERS}" ]; then
+	echo "unable to determine: go version"
+	exit 1
+fi
+
+export GOPATH="${progdir}"
+export GOBIN="${progdir}"
+
+set -e
+go get
+go build -ldflags "${LDFLAGS} -extldflags '-static'" -o "${progdir}/cbsd-mq-router"

cbsd-mq-router.json

@@ -0,0 +1,15 @@
+{
+    "cbsdenv": "/usr/jails",
+    "cbsdcolor": false,
+    "broker": "beanstalkd",
+    "logfile": "/dev/stdout",
+    "beanstalkd": {
+      "uri": "127.0.0.1:11300",
+      "tube": "cbsd_zpool1",
+      "reply_tube_prefix": "cbsd_zpool1_result_id",
+      "reconnect_timeout": 5,
+      "reserve_timeout": 5,
+      "publish_timeout": 5,
+      "logdir": "/var/log/cbsdmq"
+    }
+}