Add pluggable repository layer

Part of theycallmemac#34

odin-engine/api/main.go

@@ -1,20 +1,20 @@
 package api
 
 import (
-        "os"
-        "syscall"
+	"os"
+	"syscall"
 
+	"github.com/theycallmemac/odin/odin-engine/pkg/fsm"
+	"github.com/theycallmemac/odin/odin-engine/pkg/jobs"
+	"github.com/theycallmemac/odin/odin-engine/pkg/repository"
 	"github.com/valyala/fasthttp"
-        "github.com/theycallmemac/odin/odin-engine/pkg/fsm"
-        "github.com/theycallmemac/odin/odin-engine/pkg/jobs"
 )
 
 var (
-    // HTTPAddr contains the port used by this node
-    HTTPAddr string
+	// HTTPAddr contains the port used by this node
+	HTTPAddr string
 )
 
-
 // SetOdinEnv is used to set variables to be used by running jobs via Odin SDK
 // parameters: mongoURL (a string of the address for the MongoDB instance)
 // returns: nil
@@ -31,15 +31,17 @@ func SetOdinEnv(mongoURL string) {
 type Service struct {
 	addr  string
 	store fsm.Store
+	repo  repository.Repository
 }
 
 // NewService is used to initialize a new service struct
 // parameters: addr (a string of a http address), store (a store of node details)
 // returns: *Service (a newly initialized service struct)
-func NewService(addr string, store fsm.Store) *Service {
+func NewService(addr string, store fsm.Store, repo repository.Repository) *Service {
 	return &Service{
 		addr:  addr,
 		store: store,
+		repo:  repo,
 	}
 }
 
@@ -49,44 +51,44 @@ func NewService(addr string, store fsm.Store) *Service {
 func (service *Service) Start() {
 	routes := func(ctx *fasthttp.RequestCtx) {
 		switch string(ctx.Path()) {
-                        case "/cluster/join":
-                                service.JoinCluster(ctx)
-                        case "/cluster/leave":
-                                service.LeaveCluster(ctx)
-                        case "/execute":
-                                Executor(ctx)
-                        case "/execute/yaml":
-                                ExecuteYaml(ctx)
-                        case "/jobs/add":
-                                AddJob(ctx)
-                        case "/jobs/delete":
-                                DeleteJob(ctx)
-                        case "/jobs/info/update":
-                                UpdateJob(ctx)
-                        case "/jobs/info/description":
-                                GetJobDescription(ctx)
-                        case "/jobs/info/runs":
-                                UpdateJobRuns(ctx)
-                        case "/jobs/list":
-                                ListJobs(ctx)
-                        case "/jobs/logs":
-                                GetJobLogs(ctx)
-                        case "/links/add":
-                                LinkJobs(ctx)
-                        case "/links/delete":
-                                UnlinkJobs(ctx)
-		        case "/schedule":
-			        GetJobSchedule(ctx)
-                        case "/stats/add":
-                                AddJobStats(ctx)
-                        case "/stats/get":
-                                GetJobStats(ctx)
+		case "/cluster/join":
+			service.JoinCluster(ctx)
+		case "/cluster/leave":
+			service.LeaveCluster(ctx)
+		case "/execute":
+			Executor(ctx)
+		case "/execute/yaml":
+			ExecuteYaml(ctx)
+		case "/jobs/add":
+			AddJob(ctx)
+		case "/jobs/delete":
+			DeleteJob(ctx)
+		case "/jobs/info/update":
+			UpdateJob(ctx)
+		case "/jobs/info/description":
+			GetJobDescription(ctx)
+		case "/jobs/info/runs":
+			UpdateJobRuns(ctx)
+		case "/jobs/list":
+			ListJobs(ctx)
+		case "/jobs/logs":
+			GetJobLogs(ctx)
+		case "/links/add":
+			LinkJobs(ctx)
+		case "/links/delete":
+			UnlinkJobs(ctx)
+		case "/schedule":
+			GetJobSchedule(ctx)
+		case "/stats/add":
+			AddJobStats(ctx)
+		case "/stats/get":
+			GetJobStats(service.repo, ctx)
 		}
 	}
 
-        // start the countdown timer for the execution until the first job
-        go jobs.StartTicker(service.store, service.addr)
+	// start the countdown timer for the execution until the first job
+	go jobs.StartTicker(service.store, service.addr)
 
-        HTTPAddr = service.addr
+	HTTPAddr = service.addr
 	fasthttp.ListenAndServe(HTTPAddr, routes)
 }

odin-engine/api/stats.go

@@ -2,11 +2,12 @@ package api
 
 import (
 	"context"
-        "fmt"
+	"fmt"
 	"strings"
 
 	"github.com/theycallmemac/odin/odin-engine/pkg/jobs"
-        "github.com/valyala/fasthttp"
+	"github.com/theycallmemac/odin/odin-engine/pkg/repository"
+	"github.com/valyala/fasthttp"
 
 	"go.mongodb.org/mongo-driver/mongo"
 )
@@ -29,9 +30,9 @@ func AddJobStats(ctx *fasthttp.RequestCtx) {
 		fmt.Fprintf(ctx, "MongoDB cannot be accessed at the moment\n")
 	} else {
 		if InsertIntoMongo(client, typeOfValue, desc, value, id, timestamp) {
-			fmt.Fprintf(ctx,"200")
+			fmt.Fprintf(ctx, "200")
 		} else {
-			fmt.Fprintf(ctx,"500")
+			fmt.Fprintf(ctx, "500")
 		}
 	}
 }
@@ -56,13 +57,11 @@ func InsertIntoMongo(client *mongo.Client, typeOfValue string, desc string, valu
 }
 
 // GetJobStats is used to show stats collected by a specified job
-func GetJobStats(ctx *fasthttp.RequestCtx) {
-	client, err := jobs.SetupClient()
+func GetJobStats(repo repository.Repository, ctx *fasthttp.RequestCtx) {
+	statsList, err := repo.GetJobStats(ctx, string(ctx.PostBody()))
 	if err != nil {
-		fmt.Fprintf(ctx, "MongoDB cannot be accessed at the moment\n")
+		fmt.Fprintf(ctx, "[ERROR] Cannot get job stats: %v", err)
 	} else {
-		statsList := jobs.GetJobStats(client, string(ctx.PostBody()))
-		fmt.Fprintf(ctx, jobs.Format("ID", "DESCRIPTION", "TYPE", "VALUE"))
 		for _, stat := range statsList {
 			fmt.Fprintf(ctx, jobs.Format(stat.ID, stat.Description, stat.Type, stat.Value))
 		}

odin-engine/cmd/odin-engine/main.go

@@ -15,7 +15,10 @@ import (
 
 	"github.com/theycallmemac/odin/odin-engine/api"
 	"github.com/theycallmemac/odin/odin-engine/pkg/fsm"
+	"github.com/theycallmemac/odin/odin-engine/pkg/repository"
+	_ "github.com/theycallmemac/odin/odin-engine/pkg/repository/all"
 	"github.com/theycallmemac/odin/odin-engine/pkg/resources"
+	"github.com/theycallmemac/odin/odin-engine/pkg/types"
 )
 
 // define constsant default values to be used by the engine
@@ -32,6 +35,7 @@ var (
 	raftAddr string
 	joinAddr string
 	nodeID   string
+	config   types.EngineConfig
 )
 
 // init is used to define options to be used when running the engine
@@ -42,6 +46,12 @@ func init() {
 	flag.StringVar(&raftAddr, "raft", DefaultRaftAddr, "Set Raft bind address")
 	flag.StringVar(&joinAddr, "join", "", "Set join address, if any")
 	flag.StringVar(&nodeID, "id", "", "Node ID")
+
+	usr, err := user.Current()
+	if err != nil {
+		log.Fatal(err)
+	}
+	config = resources.UnmarsharlYaml(resources.ReadFileBytes(usr.HomeDir + "/odin-config.yml"))
 }
 
 // main is used to setup the odin-engine as a single node cluster, as well as allowing for further nodes to joining it. This is all done with the use of flags when initially running the engine.
@@ -67,15 +77,25 @@ func main() {
 		log.Fatalf("%v", err)
 	}
 
+	reg, err := repository.GetRegistration(config.Storage.Name)
+	if err != nil {
+		log.Fatal(err)
+	}
+	// TODO: options is nil for now. Pass proper storage options
+	repo, err := reg.OpenFunc(config.Storage.Address, nil)
+	if err != nil {
+		log.Fatal(err)
+	}
+
 	if httpAddr == "" {
 		usr, _ := user.Current()
 		config := resources.UnmarsharlYaml(resources.ReadFileBytes(usr.HomeDir + "/odin-config.yml"))
 		api.SetOdinEnv(config.Mongo.Address)
 		httpAddr = config.OdinVars.Master + ":" + config.OdinVars.Port
 	}
-	service := api.NewService(httpAddr, *s)
+	service := api.NewService(httpAddr, *s, repo)
 	go service.Start()
-        fmt.Println("ADDR:",httpAddr,"JOIN:",joinAddr, "RAFT:", raftAddr)
+	fmt.Println("ADDR:", httpAddr, "JOIN:", joinAddr, "RAFT:", raftAddr)
 	if joinAddr != "" {
 		if err := join(joinAddr, raftAddr, nodeID); err != nil {
 			log.Fatalf("failed to join node at %s: %s", joinAddr, err.Error())

odin-engine/config/odin-config.yml

@@ -4,3 +4,6 @@ odin:
   port: "3939"
 mongo:
   address: "mongodb://localhost:27017"
+storage:
+  name: "mongodb"
+  address: "localhost:27017"

odin-engine/go.mod

@@ -5,7 +5,9 @@ go 1.13
 require (
 	github.com/gorhill/cronexpr v0.0.0-20180427100037-88b0669f7d75
 	github.com/hashicorp/raft v1.1.2
+	github.com/hidal-go/hidalgo v0.0.0-20190814174001-42e03f3b5eaa
 	github.com/lnquy/cron v1.0.1
+	github.com/stretchr/testify v1.3.0
 	github.com/valyala/fasthttp v1.14.0
 	go.mongodb.org/mongo-driver v1.3.4
 	gopkg.in/yaml.v2 v2.3.0