diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..b010a96
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,66 @@
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion, Android Studio and WebStorm
+# Reference: https://intellij-support.jetbrains.com/hc/en-us/articles/206544839
+
+# User-specific stuff
+.idea/**/workspace.xml
+.idea/**/tasks.xml
+.idea/**/usage.statistics.xml
+.idea/**/dictionaries
+.idea/**/shelf
+
+# Generated files
+.idea/**/contentModel.xml
+
+# Sensitive or high-churn files
+.idea/**/dataSources/
+.idea/**/dataSources.ids
+.idea/**/dataSources.local.xml
+.idea/**/sqlDataSources.xml
+.idea/**/dynamic.xml
+.idea/**/uiDesigner.xml
+.idea/**/dbnavigator.xml
+
+# Gradle
+.idea/**/gradle.xml
+.idea/**/libraries
+
+# Gradle and Maven with auto-import
+# When using Gradle or Maven with auto-import, you should exclude module files,
+# since they will be recreated, and may cause churn. Uncomment if using
+# auto-import.
+# .idea/modules.xml
+# .idea/*.iml
+# .idea/modules
+
+# CMake
+cmake-build-*/
+
+# Mongo Explorer plugin
+.idea/**/mongoSettings.xml
+
+# File-based project format
+*.iws
+
+# IntelliJ
+out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Cursive Clojure plugin
+.idea/replstate.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+fabric.properties
+
+# Editor-based Rest Client
+.idea/httpRequests
+
+# Android studio 3.1+ serialized cache file
+.idea/caches/build_file_checksums.ser
\ No newline at end of file
diff --git a/.idea/go-scheduler.iml b/.idea/go-scheduler.iml
new file mode 100644
index 0000000..01865b7
--- /dev/null
+++ b/.idea/go-scheduler.iml
@@ -0,0 +1,9 @@
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/misc.xml b/.idea/misc.xml
new file mode 100644
index 0000000..28a804d
--- /dev/null
+++ b/.idea/misc.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/modules.xml b/.idea/modules.xml
new file mode 100644
index 0000000..c4b1501
--- /dev/null
+++ b/.idea/modules.xml
@@ -0,0 +1,8 @@
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/.idea/vcs.xml b/.idea/vcs.xml
new file mode 100644
index 0000000..94a25f7
--- /dev/null
+++ b/.idea/vcs.xml
@@ -0,0 +1,6 @@
+
+
+
+
+
+
\ No newline at end of file
diff --git a/CLI b/CLI
new file mode 100755
index 0000000..ccf41cb
Binary files /dev/null and b/CLI differ
diff --git a/Makefile b/Makefile
new file mode 100644
index 0000000..72a3425
--- /dev/null
+++ b/Makefile
@@ -0,0 +1,41 @@
+
+.PHONY: test
+
+all: | clean deps gen_protobuf build gen_mocks test
+
+clean:
+ @rm -rf ./gen/*
+ @rm -rf ./test/cli/mocks/*
+ @rm -rf ./test/common/mocks/*
+
+deps:
+ @echo syncing dependencies...
+ @go get github.com/onsi/ginkgo
+ @go get github.com/onsi/gomega
+ @go get github.com/vektra/mockery/.../
+ @go get github.com/stretchr/testify/mock
+ @go get -u google.golang.org/grpc
+ @go get -u github.com/golang/protobuf/protoc-gen-go
+
+build: build_cli
+
+build_cli:
+ @echo building cli...
+ @go build -o go-scheduler-cli ./cmd/go-scheduler-cli
+ @chmod +x go-scheduler-cli
+
+gen_mocks:
+ @echo generating mocks...
+ @${GOPATH}/bin/mockery -all -dir pkg/cli -output test/cli/mocks -case=underscore
+ @${GOPATH}/bin/mockery -all -dir pkg/common -output test/common/mocks -case=underscore
+ @${GOPATH}/bin/mockery -all -dir pkg/master -output test/master/mocks -case=underscore
+ @${GOPATH}/bin/mockery -all -dir gen/protobuf/master -output test/master/mocks -case=underscore
+
+gen_protobuf:
+ @echo generating protobuf APIs...
+ @command -v protoc >/dev/null 2>&1 || { echo >&2 "protoc is not installed. "; exit 1; }
+ @protoc --go_out=plugins=grpc,paths=source_relative:./gen ./protobuf/master/master.proto
+
+test:
+ @echo running tests...
+ @${GOPATH}/bin/ginkgo -r ./test
\ No newline at end of file
diff --git a/README.md b/README.md
index 204ff45..794adbd 100644
--- a/README.md
+++ b/README.md
@@ -1,2 +1,47 @@
# go-scheduler
-This is a distributed job scheduler for linux systems built in Go.
+This is a distributed job scheduler for linux systems built in Go. It can run arbitrary docker images on a cluster of computers.
+
+# Install
+
+## Get the package
+`go get github.com/kmacmcfarlane/go-scheduler`
+
+## Build the project
+`make`
+
+## Generate Certificates with EasyRSA
+The cluster uses x509 certificates to authenticate the client, master, and node applications.
+
+# CLI
+The go-scheduler CLI client allows you to control work run on the cluster.
+
+## Install Client Certificate
+The client certificate is used to authenticate with your cluster. The certificate must have file permission 600.
+
+`mkdir ~/.go-scheduler`
+`cp ~/Downloads/client.crt ~/.go-scheduler/client.crt`
+`chmod 600 ~/.go-scheduler/client.crt`
+
+## Start a Job
+
+`go-scheduler-cli start --image redis --name redisCache --master 192.168.1.80`
+
+## Stop a Job
+
+`go-scheduler-cli stop --name redisCache --master 192.168.1.80`
+
+## Query Job Status
+
+`go-scheduler-cli query --name redisCache --master 192.168.1.80`
+
+## Stream Log Output
+
+`go-scheduler-cli log --name redisCache --master 192.168.1.80`
+
+## Status Codes
+go-scheduler-cli returns the following status codes:
+
+0: success
+1: error parsing command-line arguments
+2: error parsing command-line arguments
+3: application error
\ No newline at end of file
diff --git a/cmd/go-scheduler-cli/main.go b/cmd/go-scheduler-cli/main.go
new file mode 100644
index 0000000..d1134c9
--- /dev/null
+++ b/cmd/go-scheduler-cli/main.go
@@ -0,0 +1,33 @@
+package main
+
+import (
+ "context"
+ "fmt"
+ "github.com/kmacmcfarlane/go-scheduler/internal/grpc"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/cli"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/common"
+ "os"
+)
+
+func main() {
+
+ ctx := context.Background()
+
+ logger := common.NewConsoleLogger()
+
+ clientFactory := grpc.NewMasterClientFactory()
+
+ clientService := cli.NewClientService(ctx, clientFactory, logger)
+
+ commandParser := cli.NewCommandParser(clientService, logger)
+
+ // Parse and execute Command
+ err := commandParser.Parse(os.Args)
+
+ if nil != err {
+ fmt.Fprintln(os.Stderr, err.Error())
+ os.Exit(1)
+ }
+
+ os.Exit(0)
+}
\ No newline at end of file
diff --git a/gen/protobuf/master/master.pb.go b/gen/protobuf/master/master.pb.go
new file mode 100644
index 0000000..da110ef
--- /dev/null
+++ b/gen/protobuf/master/master.pb.go
@@ -0,0 +1,597 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: protobuf/master/master.proto
+
+package master
+
+import (
+ context "context"
+ fmt "fmt"
+ proto "github.com/golang/protobuf/proto"
+ grpc "google.golang.org/grpc"
+ math "math"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion3 // please upgrade the proto package
+
+type StartRequest struct {
+ DockerImage string `protobuf:"bytes,1,opt,name=dockerImage,proto3" json:"dockerImage,omitempty"`
+ JobName string `protobuf:"bytes,2,opt,name=jobName,proto3" json:"jobName,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *StartRequest) Reset() { *m = StartRequest{} }
+func (m *StartRequest) String() string { return proto.CompactTextString(m) }
+func (*StartRequest) ProtoMessage() {}
+func (*StartRequest) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{0}
+}
+
+func (m *StartRequest) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_StartRequest.Unmarshal(m, b)
+}
+func (m *StartRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_StartRequest.Marshal(b, m, deterministic)
+}
+func (m *StartRequest) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_StartRequest.Merge(m, src)
+}
+func (m *StartRequest) XXX_Size() int {
+ return xxx_messageInfo_StartRequest.Size(m)
+}
+func (m *StartRequest) XXX_DiscardUnknown() {
+ xxx_messageInfo_StartRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_StartRequest proto.InternalMessageInfo
+
+func (m *StartRequest) GetDockerImage() string {
+ if m != nil {
+ return m.DockerImage
+ }
+ return ""
+}
+
+func (m *StartRequest) GetJobName() string {
+ if m != nil {
+ return m.JobName
+ }
+ return ""
+}
+
+type StopRequest struct {
+ JobName string `protobuf:"bytes,1,opt,name=jobName,proto3" json:"jobName,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *StopRequest) Reset() { *m = StopRequest{} }
+func (m *StopRequest) String() string { return proto.CompactTextString(m) }
+func (*StopRequest) ProtoMessage() {}
+func (*StopRequest) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{1}
+}
+
+func (m *StopRequest) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_StopRequest.Unmarshal(m, b)
+}
+func (m *StopRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_StopRequest.Marshal(b, m, deterministic)
+}
+func (m *StopRequest) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_StopRequest.Merge(m, src)
+}
+func (m *StopRequest) XXX_Size() int {
+ return xxx_messageInfo_StopRequest.Size(m)
+}
+func (m *StopRequest) XXX_DiscardUnknown() {
+ xxx_messageInfo_StopRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_StopRequest proto.InternalMessageInfo
+
+func (m *StopRequest) GetJobName() string {
+ if m != nil {
+ return m.JobName
+ }
+ return ""
+}
+
+type QueryRequest struct {
+ JobName string `protobuf:"bytes,1,opt,name=jobName,proto3" json:"jobName,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *QueryRequest) Reset() { *m = QueryRequest{} }
+func (m *QueryRequest) String() string { return proto.CompactTextString(m) }
+func (*QueryRequest) ProtoMessage() {}
+func (*QueryRequest) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{2}
+}
+
+func (m *QueryRequest) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_QueryRequest.Unmarshal(m, b)
+}
+func (m *QueryRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_QueryRequest.Marshal(b, m, deterministic)
+}
+func (m *QueryRequest) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_QueryRequest.Merge(m, src)
+}
+func (m *QueryRequest) XXX_Size() int {
+ return xxx_messageInfo_QueryRequest.Size(m)
+}
+func (m *QueryRequest) XXX_DiscardUnknown() {
+ xxx_messageInfo_QueryRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_QueryRequest proto.InternalMessageInfo
+
+func (m *QueryRequest) GetJobName() string {
+ if m != nil {
+ return m.JobName
+ }
+ return ""
+}
+
+type LogRequest struct {
+ JobName string `protobuf:"bytes,1,opt,name=jobName,proto3" json:"jobName,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *LogRequest) Reset() { *m = LogRequest{} }
+func (m *LogRequest) String() string { return proto.CompactTextString(m) }
+func (*LogRequest) ProtoMessage() {}
+func (*LogRequest) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{3}
+}
+
+func (m *LogRequest) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_LogRequest.Unmarshal(m, b)
+}
+func (m *LogRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_LogRequest.Marshal(b, m, deterministic)
+}
+func (m *LogRequest) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_LogRequest.Merge(m, src)
+}
+func (m *LogRequest) XXX_Size() int {
+ return xxx_messageInfo_LogRequest.Size(m)
+}
+func (m *LogRequest) XXX_DiscardUnknown() {
+ xxx_messageInfo_LogRequest.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogRequest proto.InternalMessageInfo
+
+func (m *LogRequest) GetJobName() string {
+ if m != nil {
+ return m.JobName
+ }
+ return ""
+}
+
+type StartResponse struct {
+ Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *StartResponse) Reset() { *m = StartResponse{} }
+func (m *StartResponse) String() string { return proto.CompactTextString(m) }
+func (*StartResponse) ProtoMessage() {}
+func (*StartResponse) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{4}
+}
+
+func (m *StartResponse) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_StartResponse.Unmarshal(m, b)
+}
+func (m *StartResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_StartResponse.Marshal(b, m, deterministic)
+}
+func (m *StartResponse) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_StartResponse.Merge(m, src)
+}
+func (m *StartResponse) XXX_Size() int {
+ return xxx_messageInfo_StartResponse.Size(m)
+}
+func (m *StartResponse) XXX_DiscardUnknown() {
+ xxx_messageInfo_StartResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_StartResponse proto.InternalMessageInfo
+
+func (m *StartResponse) GetError() string {
+ if m != nil {
+ return m.Error
+ }
+ return ""
+}
+
+type StopResponse struct {
+ Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *StopResponse) Reset() { *m = StopResponse{} }
+func (m *StopResponse) String() string { return proto.CompactTextString(m) }
+func (*StopResponse) ProtoMessage() {}
+func (*StopResponse) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{5}
+}
+
+func (m *StopResponse) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_StopResponse.Unmarshal(m, b)
+}
+func (m *StopResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_StopResponse.Marshal(b, m, deterministic)
+}
+func (m *StopResponse) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_StopResponse.Merge(m, src)
+}
+func (m *StopResponse) XXX_Size() int {
+ return xxx_messageInfo_StopResponse.Size(m)
+}
+func (m *StopResponse) XXX_DiscardUnknown() {
+ xxx_messageInfo_StopResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_StopResponse proto.InternalMessageInfo
+
+func (m *StopResponse) GetError() string {
+ if m != nil {
+ return m.Error
+ }
+ return ""
+}
+
+type QueryResponse struct {
+ Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
+ Status string `protobuf:"bytes,2,opt,name=status,proto3" json:"status,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *QueryResponse) Reset() { *m = QueryResponse{} }
+func (m *QueryResponse) String() string { return proto.CompactTextString(m) }
+func (*QueryResponse) ProtoMessage() {}
+func (*QueryResponse) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{6}
+}
+
+func (m *QueryResponse) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_QueryResponse.Unmarshal(m, b)
+}
+func (m *QueryResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_QueryResponse.Marshal(b, m, deterministic)
+}
+func (m *QueryResponse) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_QueryResponse.Merge(m, src)
+}
+func (m *QueryResponse) XXX_Size() int {
+ return xxx_messageInfo_QueryResponse.Size(m)
+}
+func (m *QueryResponse) XXX_DiscardUnknown() {
+ xxx_messageInfo_QueryResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_QueryResponse proto.InternalMessageInfo
+
+func (m *QueryResponse) GetError() string {
+ if m != nil {
+ return m.Error
+ }
+ return ""
+}
+
+func (m *QueryResponse) GetStatus() string {
+ if m != nil {
+ return m.Status
+ }
+ return ""
+}
+
+type LogResponse struct {
+ Error string `protobuf:"bytes,1,opt,name=error,proto3" json:"error,omitempty"`
+ LogMessages string `protobuf:"bytes,2,opt,name=logMessages,proto3" json:"logMessages,omitempty"`
+ XXX_NoUnkeyedLiteral struct{} `json:"-"`
+ XXX_unrecognized []byte `json:"-"`
+ XXX_sizecache int32 `json:"-"`
+}
+
+func (m *LogResponse) Reset() { *m = LogResponse{} }
+func (m *LogResponse) String() string { return proto.CompactTextString(m) }
+func (*LogResponse) ProtoMessage() {}
+func (*LogResponse) Descriptor() ([]byte, []int) {
+ return fileDescriptor_cc3a3ace9647fdbe, []int{7}
+}
+
+func (m *LogResponse) XXX_Unmarshal(b []byte) error {
+ return xxx_messageInfo_LogResponse.Unmarshal(m, b)
+}
+func (m *LogResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+ return xxx_messageInfo_LogResponse.Marshal(b, m, deterministic)
+}
+func (m *LogResponse) XXX_Merge(src proto.Message) {
+ xxx_messageInfo_LogResponse.Merge(m, src)
+}
+func (m *LogResponse) XXX_Size() int {
+ return xxx_messageInfo_LogResponse.Size(m)
+}
+func (m *LogResponse) XXX_DiscardUnknown() {
+ xxx_messageInfo_LogResponse.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_LogResponse proto.InternalMessageInfo
+
+func (m *LogResponse) GetError() string {
+ if m != nil {
+ return m.Error
+ }
+ return ""
+}
+
+func (m *LogResponse) GetLogMessages() string {
+ if m != nil {
+ return m.LogMessages
+ }
+ return ""
+}
+
+func init() {
+ proto.RegisterType((*StartRequest)(nil), "master.StartRequest")
+ proto.RegisterType((*StopRequest)(nil), "master.StopRequest")
+ proto.RegisterType((*QueryRequest)(nil), "master.QueryRequest")
+ proto.RegisterType((*LogRequest)(nil), "master.LogRequest")
+ proto.RegisterType((*StartResponse)(nil), "master.StartResponse")
+ proto.RegisterType((*StopResponse)(nil), "master.StopResponse")
+ proto.RegisterType((*QueryResponse)(nil), "master.QueryResponse")
+ proto.RegisterType((*LogResponse)(nil), "master.LogResponse")
+}
+
+func init() { proto.RegisterFile("protobuf/master/master.proto", fileDescriptor_cc3a3ace9647fdbe) }
+
+var fileDescriptor_cc3a3ace9647fdbe = []byte{
+ // 343 bytes of a gzipped FileDescriptorProto
+ 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x84, 0x92, 0x4f, 0x4f, 0xc2, 0x40,
+ 0x10, 0xc5, 0xa9, 0x4a, 0x8d, 0x53, 0xb8, 0x2c, 0x68, 0x08, 0xf1, 0x40, 0x1a, 0xff, 0x70, 0x91,
+ 0x1a, 0x48, 0x48, 0x8c, 0xf1, 0x62, 0xe2, 0x41, 0x03, 0x26, 0xc2, 0xcd, 0xdb, 0xb6, 0x0c, 0x8b,
+ 0x42, 0x19, 0xdc, 0xdd, 0x1e, 0xfc, 0xb8, 0x7e, 0x13, 0xc3, 0xb2, 0xa5, 0x2d, 0x07, 0x38, 0x35,
+ 0xf3, 0xf6, 0x37, 0x3b, 0xf3, 0xf6, 0x15, 0x2e, 0x57, 0x92, 0x34, 0x85, 0xc9, 0x34, 0x88, 0xb9,
+ 0xd2, 0x28, 0xed, 0xa7, 0x63, 0x64, 0xe6, 0x6e, 0x2a, 0xff, 0x0d, 0x2a, 0x63, 0xcd, 0xa5, 0x1e,
+ 0xe1, 0x4f, 0x82, 0x4a, 0xb3, 0x16, 0x78, 0x13, 0x8a, 0xe6, 0x28, 0x5f, 0x63, 0x2e, 0xb0, 0xe1,
+ 0xb4, 0x9c, 0xf6, 0xd9, 0x28, 0x2f, 0xb1, 0x06, 0x9c, 0x7e, 0x53, 0xf8, 0xce, 0x63, 0x6c, 0x1c,
+ 0x99, 0xd3, 0xb4, 0xf4, 0x6f, 0xc1, 0x1b, 0x6b, 0x5a, 0xa5, 0x57, 0xe5, 0x40, 0xa7, 0x08, 0xb6,
+ 0xa1, 0xf2, 0x91, 0xa0, 0xfc, 0x3d, 0x4c, 0xde, 0x00, 0x0c, 0x48, 0x1c, 0xe6, 0xae, 0xa1, 0x6a,
+ 0x6d, 0xa8, 0x15, 0x2d, 0x15, 0xb2, 0x3a, 0x94, 0x51, 0x4a, 0x92, 0x16, 0xdc, 0x14, 0xfe, 0xd5,
+ 0xda, 0xed, 0x7a, 0xc3, 0xbd, 0xd4, 0x13, 0x54, 0xed, 0x7a, 0xfb, 0x30, 0x76, 0x01, 0xae, 0xd2,
+ 0x5c, 0x27, 0xca, 0xbe, 0x83, 0xad, 0xfc, 0x17, 0xf0, 0xcc, 0xce, 0x7b, 0x9b, 0x5b, 0xe0, 0x2d,
+ 0x48, 0x0c, 0x51, 0x29, 0x2e, 0x30, 0xbd, 0x21, 0x2f, 0x75, 0xff, 0x1c, 0x70, 0x87, 0x26, 0x24,
+ 0xd6, 0x87, 0xb2, 0x71, 0xc7, 0xea, 0x1d, 0x1b, 0x62, 0x3e, 0xb3, 0xe6, 0xf9, 0x8e, 0xba, 0x19,
+ 0xec, 0x97, 0x58, 0x0f, 0x4e, 0xd6, 0x76, 0x59, 0x2d, 0x03, 0xb6, 0xf1, 0x34, 0xeb, 0x45, 0x71,
+ 0xdb, 0xd4, 0x87, 0xb2, 0x71, 0x9f, 0x0d, 0xcb, 0x67, 0x95, 0x0d, 0x2b, 0x3c, 0x91, 0x5f, 0x62,
+ 0x5d, 0x38, 0x1e, 0x90, 0x60, 0x2c, 0x3d, 0xcf, 0x72, 0x6b, 0xd6, 0x0a, 0x5a, 0xda, 0x71, 0xef,
+ 0x3c, 0x3f, 0x7e, 0x3e, 0x88, 0x2f, 0x3d, 0x4b, 0xc2, 0x4e, 0x44, 0x71, 0x30, 0x8f, 0x79, 0x14,
+ 0x47, 0x53, 0x2e, 0x17, 0x7c, 0x89, 0x81, 0xa0, 0x3b, 0x15, 0xcd, 0x70, 0x92, 0x2c, 0x50, 0x06,
+ 0x02, 0x97, 0xc1, 0xce, 0xff, 0x1c, 0xba, 0x46, 0xe8, 0xfd, 0x07, 0x00, 0x00, 0xff, 0xff, 0x5c,
+ 0xdf, 0xe3, 0x1e, 0xe9, 0x02, 0x00, 0x00,
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// MasterClient is the client API for Master service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type MasterClient interface {
+ Start(ctx context.Context, in *StartRequest, opts ...grpc.CallOption) (*StartResponse, error)
+ Stop(ctx context.Context, in *StopRequest, opts ...grpc.CallOption) (*StopResponse, error)
+ Query(ctx context.Context, in *QueryRequest, opts ...grpc.CallOption) (*QueryResponse, error)
+ Log(ctx context.Context, in *LogRequest, opts ...grpc.CallOption) (Master_LogClient, error)
+}
+
+type masterClient struct {
+ cc *grpc.ClientConn
+}
+
+func NewMasterClient(cc *grpc.ClientConn) MasterClient {
+ return &masterClient{cc}
+}
+
+func (c *masterClient) Start(ctx context.Context, in *StartRequest, opts ...grpc.CallOption) (*StartResponse, error) {
+ out := new(StartResponse)
+ err := c.cc.Invoke(ctx, "/master.Master/Start", in, out, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *masterClient) Stop(ctx context.Context, in *StopRequest, opts ...grpc.CallOption) (*StopResponse, error) {
+ out := new(StopResponse)
+ err := c.cc.Invoke(ctx, "/master.Master/Stop", in, out, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *masterClient) Query(ctx context.Context, in *QueryRequest, opts ...grpc.CallOption) (*QueryResponse, error) {
+ out := new(QueryResponse)
+ err := c.cc.Invoke(ctx, "/master.Master/Query", in, out, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+func (c *masterClient) Log(ctx context.Context, in *LogRequest, opts ...grpc.CallOption) (Master_LogClient, error) {
+ stream, err := c.cc.NewStream(ctx, &_Master_serviceDesc.Streams[0], "/master.Master/Log", opts...)
+ if err != nil {
+ return nil, err
+ }
+ x := &masterLogClient{stream}
+ if err := x.ClientStream.SendMsg(in); err != nil {
+ return nil, err
+ }
+ if err := x.ClientStream.CloseSend(); err != nil {
+ return nil, err
+ }
+ return x, nil
+}
+
+type Master_LogClient interface {
+ Recv() (*LogResponse, error)
+ grpc.ClientStream
+}
+
+type masterLogClient struct {
+ grpc.ClientStream
+}
+
+func (x *masterLogClient) Recv() (*LogResponse, error) {
+ m := new(LogResponse)
+ if err := x.ClientStream.RecvMsg(m); err != nil {
+ return nil, err
+ }
+ return m, nil
+}
+
+// MasterServer is the server API for Master service.
+type MasterServer interface {
+ Start(context.Context, *StartRequest) (*StartResponse, error)
+ Stop(context.Context, *StopRequest) (*StopResponse, error)
+ Query(context.Context, *QueryRequest) (*QueryResponse, error)
+ Log(*LogRequest, Master_LogServer) error
+}
+
+func RegisterMasterServer(s *grpc.Server, srv MasterServer) {
+ s.RegisterService(&_Master_serviceDesc, srv)
+}
+
+func _Master_Start_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(StartRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MasterServer).Start(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: "/master.Master/Start",
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MasterServer).Start(ctx, req.(*StartRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _Master_Stop_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(StopRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MasterServer).Stop(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: "/master.Master/Stop",
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MasterServer).Stop(ctx, req.(*StopRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _Master_Query_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+ in := new(QueryRequest)
+ if err := dec(in); err != nil {
+ return nil, err
+ }
+ if interceptor == nil {
+ return srv.(MasterServer).Query(ctx, in)
+ }
+ info := &grpc.UnaryServerInfo{
+ Server: srv,
+ FullMethod: "/master.Master/Query",
+ }
+ handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+ return srv.(MasterServer).Query(ctx, req.(*QueryRequest))
+ }
+ return interceptor(ctx, in, info, handler)
+}
+
+func _Master_Log_Handler(srv interface{}, stream grpc.ServerStream) error {
+ m := new(LogRequest)
+ if err := stream.RecvMsg(m); err != nil {
+ return err
+ }
+ return srv.(MasterServer).Log(m, &masterLogServer{stream})
+}
+
+type Master_LogServer interface {
+ Send(*LogResponse) error
+ grpc.ServerStream
+}
+
+type masterLogServer struct {
+ grpc.ServerStream
+}
+
+func (x *masterLogServer) Send(m *LogResponse) error {
+ return x.ServerStream.SendMsg(m)
+}
+
+var _Master_serviceDesc = grpc.ServiceDesc{
+ ServiceName: "master.Master",
+ HandlerType: (*MasterServer)(nil),
+ Methods: []grpc.MethodDesc{
+ {
+ MethodName: "Start",
+ Handler: _Master_Start_Handler,
+ },
+ {
+ MethodName: "Stop",
+ Handler: _Master_Stop_Handler,
+ },
+ {
+ MethodName: "Query",
+ Handler: _Master_Query_Handler,
+ },
+ },
+ Streams: []grpc.StreamDesc{
+ {
+ StreamName: "Log",
+ Handler: _Master_Log_Handler,
+ ServerStreams: true,
+ },
+ },
+ Metadata: "protobuf/master/master.proto",
+}
diff --git a/go-scheduler-cli b/go-scheduler-cli
new file mode 100755
index 0000000..224fa08
Binary files /dev/null and b/go-scheduler-cli differ
diff --git a/internal/grpc/master_client_factory.go b/internal/grpc/master_client_factory.go
new file mode 100644
index 0000000..9b6288a
--- /dev/null
+++ b/internal/grpc/master_client_factory.go
@@ -0,0 +1,32 @@
+package grpc
+
+import (
+ gen_master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/master"
+ "google.golang.org/grpc"
+ "io"
+)
+
+var _ master.MasterClientFactory = MasterClientFactory{}
+
+type MasterClientFactory struct {}
+
+func NewMasterClientFactory() MasterClientFactory{
+ return MasterClientFactory{}
+}
+
+func (cf MasterClientFactory) CreateClient(host string) (client gen_master.MasterClient, closer io.Closer, err error) {
+
+ //TODO: dial options
+ opts := []grpc.DialOption{}
+
+ conn, err := grpc.Dial(host, opts...)
+
+ if err != nil {
+ return client, conn, err
+ }
+
+ client = gen_master.NewMasterClient(conn)
+
+ return client, conn, err
+}
\ No newline at end of file
diff --git a/pkg/cli/client_service.go b/pkg/cli/client_service.go
new file mode 100644
index 0000000..922466a
--- /dev/null
+++ b/pkg/cli/client_service.go
@@ -0,0 +1,142 @@
+package cli
+
+import (
+ "context"
+ "errors"
+ gen_master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/common"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/master"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/model/status"
+ "io"
+)
+
+type ClientService interface {
+ Start(dockerImage string, jobName string, host string) error
+ Stop(jobName string, host string) error
+ Query(jobName string, host string) (status status.Status, err error)
+ Log(jobName string, host string) (logReader io.ReadCloser, err error)
+}
+
+type GrpcClientService struct {
+ ctx context.Context
+ clientFactory master.MasterClientFactory
+ logger common.Logger
+}
+
+func NewClientService(
+ ctx context.Context,
+ clientFactory master.MasterClientFactory,
+ logger common.Logger) *GrpcClientService {
+
+ return &GrpcClientService{
+ ctx: ctx,
+ clientFactory: clientFactory,
+ logger: logger}
+}
+
+// Start creates or starts a job and assigns a name to it
+func (cs *GrpcClientService) Start(dockerImage string, name string, host string) (err error) {
+
+ client, conn, err := cs.clientFactory.CreateClient(host)
+
+ if err != nil {
+ return err
+ }
+
+ defer conn.Close()
+
+ req := &gen_master.StartRequest{
+ DockerImage: dockerImage,
+ JobName: name}
+
+ resp, err := client.Start(cs.ctx, req)
+
+ if err != nil {
+ return err
+ }
+
+ if resp.Error != "" {
+ return errors.New(resp.Error)
+ }
+
+ return err
+}
+
+// Stop terminates a job
+func (cs *GrpcClientService) Stop(name string, host string) (err error) {
+
+ client, conn, err := cs.clientFactory.CreateClient(host)
+
+ if err != nil {
+ return err
+ }
+
+ defer conn.Close()
+
+ req := &gen_master.StopRequest{
+ JobName: name}
+
+ resp, err := client.Stop(cs.ctx, req)
+
+ if err != nil {
+ return err
+ }
+
+ if resp.Error != "" {
+ return errors.New(resp.Error)
+ }
+
+ return err
+}
+
+// Query provides the current status of the given job
+func (cs *GrpcClientService) Query(name string, host string) (result status.Status, err error) {
+
+ client, conn, err := cs.clientFactory.CreateClient(host)
+
+ if err != nil {
+ return result, err
+ }
+
+ defer conn.Close()
+
+ req := &gen_master.QueryRequest{
+ JobName: name}
+
+ resp, err := client.Query(cs.ctx, req)
+
+ if err != nil {
+ return result, err
+ }
+
+ if resp.Error != "" {
+ return result, errors.New(resp.Error)
+ }
+
+ result = status.New(resp.Status)
+
+ return result, err
+}
+
+// Log provides a reader that streams the log output of the job
+func (cs *GrpcClientService) Log(name string, host string) (logReader io.ReadCloser, err error) {
+
+ client, conn, err := cs.clientFactory.CreateClient(host)
+
+ if err != nil {
+ return logReader, err
+ }
+
+ req := &gen_master.LogRequest{
+ JobName: name}
+
+ logClient, err := client.Log(cs.ctx, req)
+
+ if err != nil {
+ return logReader, err
+ }
+
+ logReader = NewLogReader(logClient, conn)
+
+ return logReader, err
+}
diff --git a/pkg/cli/command_parser.go b/pkg/cli/command_parser.go
new file mode 100644
index 0000000..41cef8a
--- /dev/null
+++ b/pkg/cli/command_parser.go
@@ -0,0 +1,184 @@
+package cli
+
+import (
+ "bufio"
+ "errors"
+ "flag"
+ "fmt"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/common"
+ "io"
+)
+
+type CommandParser struct {
+ clientService ClientService
+ logger common.Logger
+}
+
+func NewCommandParser(clientService ClientService, logger common.Logger) *CommandParser {
+ return &CommandParser{
+ clientService: clientService,
+ logger: logger}
+}
+
+func (cp *CommandParser) Parse(args []string) (err error) {
+
+ // Start
+ startCommand := flag.NewFlagSet("start", flag.ContinueOnError)
+ startCommand.SetOutput(cp.logger)
+
+ startImage := startCommand.String("image", "", "The docker image name (Required)")
+ startName := startCommand.String("name", "", "The name of the job (Required)")
+ startHost := startCommand.String("host", "localhost", "The hostname of the master node")
+ // Improvement: port number flag
+
+ // Stop
+ stopCommand := flag.NewFlagSet("stop", flag.ContinueOnError)
+ stopCommand.SetOutput(cp.logger)
+
+ stopName := stopCommand.String("name", "", "The name of the job (Required)")
+ stopHost := stopCommand.String("host", "localhost", "The hostname of the master node")
+
+ // Query
+ queryCommand := flag.NewFlagSet("query", flag.ContinueOnError)
+ queryCommand.SetOutput(cp.logger)
+
+ queryName := queryCommand.String("name", "", "The name of the job (Required)")
+ queryHost := queryCommand.String("host", "localhost", "The hostname of the master node")
+
+ // Stream Logs
+ logCommand := flag.NewFlagSet("log", flag.ContinueOnError)
+ logCommand.SetOutput(cp.logger)
+
+ logName := logCommand.String("name", "", "The name of the job (Required)")
+ logHost := logCommand.String("host", "localhost", "The hostname of the master node")
+
+ // Validate command input
+ defaultFlags := flag.NewFlagSet("default", flag.ContinueOnError)
+ helpShort := defaultFlags.Bool("h", false, "Print this usage")
+ helpLong := defaultFlags.Bool("help", false, "Print this usage")
+
+ defaultFlags.Parse(args)
+
+ if len(args) < 2 || *helpShort || *helpLong {
+
+ cp.logger.Println("start")
+ startCommand.PrintDefaults()
+ cp.logger.Println("stop")
+ stopCommand.PrintDefaults()
+ cp.logger.Println("query")
+ queryCommand.PrintDefaults()
+ cp.logger.Println("log")
+ logCommand.PrintDefaults()
+
+ return errors.New("sub-command is required: start, stop, query, or log")
+ }
+
+ // Parse sub-command
+ switch args[1] {
+ case "start":
+
+ err = startCommand.Parse(args[2:])
+
+ if err != nil {
+ return err
+ }
+
+ // Assert required flags
+ if *startImage == "" {
+
+ startCommand.PrintDefaults()
+ return errors.New("image name is required")
+ }
+
+ if *startName == "" {
+
+ startCommand.PrintDefaults()
+ return errors.New("job name is required")
+ }
+
+ // Call master node
+ err := cp.clientService.Start(*startImage, *startName, *startHost)
+
+ if nil != err {
+ return err
+ } else {
+ cp.logger.Println("job started")
+ }
+ case "stop":
+
+ stopCommand.Parse(args[2:])
+
+ // Assert required flags
+ if *stopName == "" {
+ stopCommand.PrintDefaults()
+ return errors.New("job name is required")
+ }
+
+ // Call master node
+ err := cp.clientService.Stop(*stopName, *stopHost)
+
+ if nil != err {
+ return err
+ } else {
+ cp.logger.Println("job stopped")
+ }
+ case "query":
+
+ queryCommand.Parse(args[2:])
+
+ // Assert required flags
+ if *queryName == "" {
+ queryCommand.PrintDefaults()
+ return errors.New("job name is required")
+ }
+
+ // Call master node
+ status, err := cp.clientService.Query(*queryName, *queryHost)
+
+ if nil != err {
+ return err
+ }
+
+ cp.logger.Printf("job status: %s\n", status.String())
+ case "log":
+
+ logCommand.Parse(args[2:])
+
+ // Assert required flags
+ if *logName == "" {
+ logCommand.PrintDefaults()
+ return errors.New("job name is required")
+ }
+
+ // Call master node
+ logReader, err := cp.clientService.Log(*logName, *logHost)
+
+ if nil != err {
+ return err
+ }
+
+ defer logReader.Close()
+
+ // Print out the streaming log data
+ bufferedReader := bufio.NewReader(logReader)
+
+ for {
+
+ line, err := bufferedReader.ReadString('\n')
+
+ if err != nil {
+ if err == io.EOF {
+ break
+ } else {
+ return err
+ }
+ }
+
+ cp.logger.Print(line) // line contains the newline char at the end
+ }
+ default:
+ return errors.New(fmt.Sprintf("unrecognized command: %s\nsub-command is required: start, stop, query, or log", args[1]))
+ }
+
+ return err
+}
\ No newline at end of file
diff --git a/pkg/cli/log_reader.go b/pkg/cli/log_reader.go
new file mode 100644
index 0000000..862dda4
--- /dev/null
+++ b/pkg/cli/log_reader.go
@@ -0,0 +1,50 @@
+package cli
+
+import (
+ "bytes"
+ "errors"
+ "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+ "io"
+)
+
+var _ io.ReadCloser = &LogReader{}
+
+// LogReader exposes the grpc log stream as an io.Reader interface
+type LogReader struct {
+ logClient master.Master_LogClient
+ closer io.Closer
+ buffer *bytes.Buffer
+}
+ // NewLogReader wraps the logClient and closer implementing ReadCloser and unboxing log messages from the grpc messages
+func NewLogReader(logClient master.Master_LogClient, closer io.Closer) *LogReader {
+ return &LogReader{
+ logClient: logClient,
+ closer: closer,
+ buffer: new(bytes.Buffer)}
+}
+
+// Read handles filling the buffer from the LogClient
+func (lr *LogReader) Read(p []byte) (n int, err error){
+
+ if lr.buffer.Len() == 0 {
+
+ resp, err := lr.logClient.Recv()
+
+ if err != nil {
+ return n, err
+ }
+
+ if resp.Error != "" {
+ return n, errors.New(resp.Error)
+ }
+
+ lr.buffer.WriteString(resp.LogMessages)
+ }
+
+ return lr.buffer.Read(p)
+}
+
+// Close closes the underlying connection
+func (lr *LogReader) Close() error {
+ return lr.closer.Close()
+}
\ No newline at end of file
diff --git a/pkg/common/closer.go b/pkg/common/closer.go
new file mode 100644
index 0000000..b5e504f
--- /dev/null
+++ b/pkg/common/closer.go
@@ -0,0 +1,5 @@
+package common
+
+type Closer interface {
+ Close() error
+}
\ No newline at end of file
diff --git a/pkg/common/logger.go b/pkg/common/logger.go
new file mode 100644
index 0000000..0d8ca2e
--- /dev/null
+++ b/pkg/common/logger.go
@@ -0,0 +1,43 @@
+package common
+
+import (
+ "bytes"
+ "fmt"
+)
+
+// Wrap common output functions for testability
+type Logger interface {
+ Print(args ...interface{})
+ Printf(format string, args ...interface{})
+ Println(args ...interface{})
+ Write(p []byte) (n int, err error)
+}
+
+var _ Logger = &ConsoleLogger{}
+
+// ConsoleLogger implements Logger interface and logs to Stdout
+type ConsoleLogger struct {}
+
+func NewConsoleLogger() *ConsoleLogger {
+ return &ConsoleLogger{}
+}
+
+func (l *ConsoleLogger) Print(args ...interface{}) {
+ fmt.Print(args)
+}
+
+func (l *ConsoleLogger) Printf(format string, args ...interface{}) {
+ fmt.Printf(format, args)
+}
+
+func (l *ConsoleLogger) Println(args ...interface{}) {
+ fmt.Println(args)
+}
+
+// io.Writer interface
+func (l *ConsoleLogger) Write(p []byte) (n int, err error) {
+
+ buf := bytes.NewBuffer(p)
+
+ return fmt.Print(buf.String())
+}
\ No newline at end of file
diff --git a/pkg/master/master_client_factory.go b/pkg/master/master_client_factory.go
new file mode 100644
index 0000000..3912cb0
--- /dev/null
+++ b/pkg/master/master_client_factory.go
@@ -0,0 +1,10 @@
+package master
+
+import (
+ "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+ "io"
+)
+
+type MasterClientFactory interface {
+ CreateClient(host string) (result master.MasterClient, closer io.Closer, err error)
+}
\ No newline at end of file
diff --git a/pkg/model/status/Status.go b/pkg/model/status/Status.go
new file mode 100644
index 0000000..63af854
--- /dev/null
+++ b/pkg/model/status/Status.go
@@ -0,0 +1,53 @@
+package status
+
+type Status string
+
+var enums []string
+
+func New(str string) Status {
+
+ for _, s := range enums {
+ if s == str {
+ return Status(s)
+ }
+ }
+
+ panic("Unknown Status: " + str)
+}
+
+func (kind Status) String() string {
+ return string(kind)
+}
+
+func List() []string {
+ return enums
+}
+
+func ListGeneric() []interface{} {
+
+ result := make([]interface{}, len(enums))
+
+ for i, s := range enums {
+ result[i] = s
+ }
+
+ return result
+}
+
+func createStatus(s string) Status {
+
+ enums = append(enums, s)
+
+ return Status(s)
+}
+
+var (
+ Unknown = createStatus("")
+ Created = createStatus("created")
+ Restarting = createStatus("restarting")
+ Running = createStatus("running")
+ Removing = createStatus("removing")
+ Paused = createStatus("paused")
+ Exited = createStatus("exited")
+ Dead = createStatus("dead")
+)
diff --git a/protobuf/master/master.proto b/protobuf/master/master.proto
new file mode 100644
index 0000000..d46f683
--- /dev/null
+++ b/protobuf/master/master.proto
@@ -0,0 +1,47 @@
+syntax = "proto3";
+
+package master;
+
+option go_package = "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master";
+
+service Master {
+ rpc Start (StartRequest) returns (StartResponse) {}
+ rpc Stop (StopRequest) returns (StopResponse) {}
+ rpc Query (QueryRequest) returns (QueryResponse) {}
+ rpc Log (LogRequest) returns (stream LogResponse) {}
+}
+
+message StartRequest {
+ string dockerImage = 1;
+ string jobName = 2;
+}
+
+message StopRequest {
+ string jobName = 1;
+}
+
+message QueryRequest {
+ string jobName = 1;
+}
+
+message LogRequest {
+ string jobName = 1;
+}
+
+message StartResponse {
+ string error = 1;
+}
+
+message StopResponse {
+ string error = 1;
+}
+
+message QueryResponse {
+ string error = 1;
+ string status = 2;
+}
+
+message LogResponse {
+ string error = 1;
+ string logMessages = 2;
+}
\ No newline at end of file
diff --git a/test/cli/cli_suite_test.go b/test/cli/cli_suite_test.go
new file mode 100644
index 0000000..3817d32
--- /dev/null
+++ b/test/cli/cli_suite_test.go
@@ -0,0 +1,13 @@
+package cli_test
+
+import (
+ "testing"
+
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+)
+
+func TestCli(t *testing.T) {
+ RegisterFailHandler(Fail)
+ RunSpecs(t, "Cli Suite")
+}
diff --git a/test/cli/client_service_test.go b/test/cli/client_service_test.go
new file mode 100644
index 0000000..c658dc7
--- /dev/null
+++ b/test/cli/client_service_test.go
@@ -0,0 +1,500 @@
+package cli
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/cli"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/model/status"
+ common_mocks "github.com/kmacmcfarlane/go-scheduler/test/common/mocks"
+ "github.com/kmacmcfarlane/go-scheduler/test/master/mocks"
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+ "io"
+)
+
+var _ = Describe("Client Service", func(){
+
+ var (
+ clientService cli.ClientService
+ clientFactory *mocks.MasterClientFactory
+ client *mocks.MasterClient
+ logClient *mocks.Master_LogClient
+ closer *common_mocks.Closer
+ ctx context.Context
+ logger *common_mocks.Logger
+ )
+
+ BeforeEach(func(){
+
+ logger = new(common_mocks.Logger)
+
+ closer = new(common_mocks.Closer)
+
+ logClient = new(mocks.Master_LogClient)
+
+ client = new(mocks.MasterClient)
+
+ clientFactory = new(mocks.MasterClientFactory)
+
+ clientService = cli.NewClientService(ctx, clientFactory, logger)
+ })
+
+ AfterEach(func(){
+ logger.AssertExpectations(GinkgoT())
+ })
+
+ Describe("Start Job", func(){
+
+ Context("Client factory error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, errors.New("error message")).
+ Once()
+
+ err := clientService.Start("imageName", "jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client call error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StartRequest{
+ JobName:"jobName",
+ DockerImage:"imageName"}
+
+ client.
+ On("Start", ctx, request).
+ Return(nil, errors.New("error message"))
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Start("imageName", "jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client response contains error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StartRequest{
+ JobName:"jobName",
+ DockerImage:"imageName"}
+
+ response := &master.StartResponse{
+ Error:"error message"}
+
+ client.
+ On("Start", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Start("imageName", "jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("No errors", func(){
+ It("Returns no errors and has an empty error field in the response", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StartRequest{
+ JobName:"jobName",
+ DockerImage:"imageName"}
+
+ response := &master.StartResponse{
+ Error:""}
+
+ client.
+ On("Start", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Start("imageName", "jobName", "hostName")
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+
+ Describe("Stop Job", func(){
+
+ Context("Client factory error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, errors.New("error message")).
+ Once()
+
+ err := clientService.Stop("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client call error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StopRequest{
+ JobName:"jobName"}
+
+ client.
+ On("Stop", ctx, request).
+ Return(nil, errors.New("error message"))
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Stop("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client response contains error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StopRequest{
+ JobName:"jobName"}
+
+ response := &master.StopResponse{
+ Error:"error message"}
+
+ client.
+ On("Stop", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Stop("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("No errors", func(){
+ It("Returns no errors and has an empty error field in the response", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.StopRequest{
+ JobName:"jobName"}
+
+ response := &master.StopResponse{
+ Error:""}
+
+ client.
+ On("Stop", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ err := clientService.Stop("jobName", "hostName")
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+
+ Describe("Query Job", func(){
+
+ Context("Client factory error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, errors.New("error message")).
+ Once()
+
+ _, err := clientService.Query("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client call error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.QueryRequest{
+ JobName:"jobName"}
+
+ client.
+ On("Query", ctx, request).
+ Return(nil, errors.New("error message"))
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ _, err := clientService.Query("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client response contains error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.QueryRequest{
+ JobName:"jobName"}
+
+ response := &master.QueryResponse{
+ Error:"error message",
+ Status: status.Unknown.String()}
+
+ client.
+ On("Query", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ _, err := clientService.Query("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("No errors", func(){
+ It("Returns no errors and has an empty error field in the response", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.QueryRequest{
+ JobName:"jobName"}
+
+ response := &master.QueryResponse{
+ Error:"",
+ Status: status.Running.String()}
+
+ client.
+ On("Query", ctx, request).
+ Return(response, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ result, err := clientService.Query("jobName", "hostName")
+
+ Ω(err).Should(BeNil())
+ Ω(result).Should(Equal(status.Running))
+ })
+ })
+ })
+
+ Describe("Log Stream From Job", func(){
+
+ Context("Client factory error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, errors.New("error message")).
+ Once()
+
+ _, err := clientService.Log("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Client call error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.LogRequest{
+ JobName:"jobName"}
+
+ client.
+ On("Log", ctx, request).
+ Return(nil, errors.New("error message"))
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ _, err := clientService.Log("jobName", "hostName")
+
+ Ω(err.Error()).Should(Equal("error message"))
+
+ })
+ })
+
+ Context("Client response contains error", func(){
+ It("Returns the error", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.LogRequest{
+ JobName:"jobName"}
+
+ response := &master.LogResponse{
+ Error:"error message",
+ LogMessages: "one\ntwo\n"}
+
+ logClient.
+ On("Recv").
+ Return(response, nil).
+ On("Recv").
+ Return(nil, io.EOF)
+
+ client.
+ On("Log", ctx, request).
+ Return(logClient, nil)
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ reader, err := clientService.Log("jobName", "hostName")
+
+ Ω(err).Should(BeNil())
+
+ defer reader.Close()
+
+ // Drain the reader into the buffer
+ buf := new(bytes.Buffer)
+ _, err = buf.ReadFrom(reader)
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("No errors", func(){
+ It("Returns no errors and reader gives us one\\ntwo\\nthree\\nfour\\nEOF", func(){
+
+ clientFactory.
+ On("CreateClient", "hostName").
+ Return(client, closer, nil).
+ Once()
+
+ request := &master.LogRequest{
+ JobName:"jobName"}
+
+ response := &master.LogResponse{
+ Error:"",
+ LogMessages: "one\ntwo\n"}
+
+ response2 := &master.LogResponse{
+ Error:"",
+ LogMessages: "three\nfour\n"}
+
+ logClient.
+ On("Recv").
+ Return(response, nil).
+ Once().
+ On("Recv").
+ Return(response2, nil).
+ Once().
+ On("Recv").
+ Return(nil, io.EOF).
+ Once()
+
+ client.
+ On("Log", ctx, request).
+ Return(logClient, nil).
+ Once()
+
+ closer.
+ On("Close").
+ Return(nil).
+ Once()
+
+ reader, err := clientService.Log("jobName", "hostName")
+
+ defer reader.Close()
+
+ Ω(err).Should(BeNil())
+
+ // Drain the reader into the buffer
+ buf := new(bytes.Buffer)
+ _, err = buf.ReadFrom(reader)
+
+ Ω(err).Should(BeNil())
+
+ Ω(buf.String()).Should(Equal("one\ntwo\nthree\nfour\n"))
+ })
+ })
+ })
+})
\ No newline at end of file
diff --git a/test/cli/command_parser_test.go b/test/cli/command_parser_test.go
new file mode 100644
index 0000000..077026a
--- /dev/null
+++ b/test/cli/command_parser_test.go
@@ -0,0 +1,282 @@
+package cli
+
+import (
+ "errors"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/cli"
+ "github.com/kmacmcfarlane/go-scheduler/pkg/model/status"
+ "github.com/kmacmcfarlane/go-scheduler/test"
+ "github.com/kmacmcfarlane/go-scheduler/test/cli/mocks"
+ common_mocks "github.com/kmacmcfarlane/go-scheduler/test/common/mocks"
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+ "github.com/stretchr/testify/mock"
+ "strings"
+)
+
+var _ = Describe("Command Parser", func(){
+
+ var (
+ parser *cli.CommandParser
+ clientService *mocks.ClientService
+ logger *common_mocks.Logger
+ )
+
+ BeforeEach(func(){
+
+ clientService = new(mocks.ClientService)
+ logger = new(common_mocks.Logger)
+
+ parser = cli.NewCommandParser(clientService, logger)
+ })
+
+ AfterEach(func(){
+ clientService.AssertExpectations(GinkgoT())
+ logger.AssertExpectations(GinkgoT())
+ })
+
+ Describe("Subcommand", func(){
+ Context("Missing subcommand", func(){
+ It("Prints message and returns error", func(){
+
+ logger.
+ On("Println", "start").
+ Once().
+ On("Println", "stop").
+ Once().
+ On("Println", "query").
+ Once().
+ On("Println", "log").
+ Once()
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe"})
+
+ Ω(err.Error()).Should(Equal("sub-command is required: start, stop, query, or log"))
+ })
+ })
+
+ Context("Invalid subcommand", func(){
+ It("Prints message and returns error", func(){
+
+ err := parser.Parse([]string{"foo.exe", "jump"})
+
+ Ω(err.Error()).Should(Equal("unrecognized command: jump\nsub-command is required: start, stop, query, or log"))
+ })
+ })
+ })
+
+ Describe("Start Command", func(){
+
+ Context("Missing image name", func(){
+ It("Prints usage and returns error", func(){
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe", "start", "-name", "jobName"})
+
+ Ω(err.Error()).Should(Equal("image name is required"))
+ })
+ })
+
+ Context("Missing job name", func(){
+ It("Prints usage and returns error", func(){
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe", "start", "-image", "imageName"})
+
+ Ω(err.Error()).Should(Equal("job name is required"))
+ })
+ })
+
+ Context("Error returned from ClientService", func(){
+ It("Print error and return", func(){
+
+ clientService.
+ On("Start", "imageName", "jobName", "hostName").
+ Return(errors.New("error message")).
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "start", "-image", "imageName", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Success", func(){
+ It("Returns no error", func(){
+
+ clientService.
+ On("Start", "imageName", "jobName", "hostName").
+ Return(nil).
+ Times(1)
+
+ logger.
+ On("Println", "job started").
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "start", "-image", "imageName", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+
+ Describe("Stop Command", func(){
+
+ Context("Missing job name", func(){
+ It("Prints usage and returns error", func(){
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe", "stop"})
+
+ Ω(err.Error()).Should(Equal("job name is required"))
+ })
+ })
+
+ Context("Error returned from ClientService", func(){
+ It("Print error and return", func(){
+
+ clientService.
+ On("Stop", "jobName", "hostName").
+ Return(errors.New("error message")).
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "stop", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Success", func(){
+ It("Returns no error", func(){
+
+ clientService.
+ On("Stop", "jobName", "hostName").
+ Return(nil).
+ Times(1)
+
+ logger.
+ On("Println", "job stopped").
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "stop", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+
+ Describe("Query Command", func(){
+
+ Context("Missing job name", func(){
+ It("Prints usage and returns error", func(){
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe", "query"})
+
+ Ω(err.Error()).Should(Equal("job name is required"))
+ })
+ })
+
+ Context("Error returned from ClientService", func(){
+ It("Print error and return", func(){
+
+ clientService.
+ On("Query", "jobName", "hostName").
+ Return(status.Status(""), errors.New("error message")).
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "query", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Success", func(){
+ It("Returns no error", func(){
+
+ clientService.
+ On("Query", "jobName", "hostName").
+ Return(status.Running, nil).
+ Times(1)
+
+ logger.
+ On("Printf", "job status: %s\n", status.Running.String()).
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "query", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+
+ Describe("Log Stream Command", func(){
+
+ Context("Missing job name", func(){
+ It("Prints usage and returns error", func(){
+
+ logger.
+ On("Write", mock.AnythingOfType("[]uint8")).
+ Return(123, nil) // called for usage details
+
+ err := parser.Parse([]string{"foo.exe", "log"})
+
+ Ω(err.Error()).Should(Equal("job name is required"))
+ })
+ })
+
+ Context("Error returned from ClientService", func(){
+ It("Returns error", func(){
+
+ clientService.
+ On("Log", "jobName", "hostName").
+ Return(nil, errors.New("error message")).
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "log", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err.Error()).Should(Equal("error message"))
+ })
+ })
+
+ Context("Success", func(){
+ It("Returns no error", func(){
+
+ logReader := strings.NewReader("one\ntwo\n")
+ readCloser := test.NewMockReadCloser(logReader)
+
+ clientService.
+ On("Log", "jobName", "hostName").
+ Return(readCloser, nil).
+ Times(1)
+
+ logger.
+ On("Print", "one\n").
+ Times(1)
+
+ logger.
+ On("Print", "two\n").
+ Times(1)
+
+ err := parser.Parse([]string{"foo.exe", "log", "-name", "jobName", "-host", "hostName"})
+
+ Ω(err).Should(BeNil())
+ })
+ })
+ })
+})
\ No newline at end of file
diff --git a/test/cli/mocks/client_service.go b/test/cli/mocks/client_service.go
new file mode 100644
index 0000000..cb4d3e3
--- /dev/null
+++ b/test/cli/mocks/client_service.go
@@ -0,0 +1,84 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import io "io"
+import mock "github.com/stretchr/testify/mock"
+import status "github.com/kmacmcfarlane/go-scheduler/pkg/model/status"
+
+// ClientService is an autogenerated mock type for the ClientService type
+type ClientService struct {
+ mock.Mock
+}
+
+// Log provides a mock function with given fields: jobName, host
+func (_m *ClientService) Log(jobName string, host string) (io.ReadCloser, error) {
+ ret := _m.Called(jobName, host)
+
+ var r0 io.ReadCloser
+ if rf, ok := ret.Get(0).(func(string, string) io.ReadCloser); ok {
+ r0 = rf(jobName, host)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(io.ReadCloser)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(string, string) error); ok {
+ r1 = rf(jobName, host)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Query provides a mock function with given fields: jobName, host
+func (_m *ClientService) Query(jobName string, host string) (status.Status, error) {
+ ret := _m.Called(jobName, host)
+
+ var r0 status.Status
+ if rf, ok := ret.Get(0).(func(string, string) status.Status); ok {
+ r0 = rf(jobName, host)
+ } else {
+ r0 = ret.Get(0).(status.Status)
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(string, string) error); ok {
+ r1 = rf(jobName, host)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Start provides a mock function with given fields: dockerImage, jobName, host
+func (_m *ClientService) Start(dockerImage string, jobName string, host string) error {
+ ret := _m.Called(dockerImage, jobName, host)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(string, string, string) error); ok {
+ r0 = rf(dockerImage, jobName, host)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// Stop provides a mock function with given fields: jobName, host
+func (_m *ClientService) Stop(jobName string, host string) error {
+ ret := _m.Called(jobName, host)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(string, string) error); ok {
+ r0 = rf(jobName, host)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
diff --git a/test/common/mocks/closer.go b/test/common/mocks/closer.go
new file mode 100644
index 0000000..05a97a1
--- /dev/null
+++ b/test/common/mocks/closer.go
@@ -0,0 +1,24 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// Closer is an autogenerated mock type for the Closer type
+type Closer struct {
+ mock.Mock
+}
+
+// Close provides a mock function with given fields:
+func (_m *Closer) Close() error {
+ ret := _m.Called()
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func() error); ok {
+ r0 = rf()
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
diff --git a/test/common/mocks/logger.go b/test/common/mocks/logger.go
new file mode 100644
index 0000000..35ae71e
--- /dev/null
+++ b/test/common/mocks/logger.go
@@ -0,0 +1,53 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import mock "github.com/stretchr/testify/mock"
+
+// Logger is an autogenerated mock type for the Logger type
+type Logger struct {
+ mock.Mock
+}
+
+// Print provides a mock function with given fields: args
+func (_m *Logger) Print(args ...interface{}) {
+ var _ca []interface{}
+ _ca = append(_ca, args...)
+ _m.Called(_ca...)
+}
+
+// Printf provides a mock function with given fields: format, args
+func (_m *Logger) Printf(format string, args ...interface{}) {
+ var _ca []interface{}
+ _ca = append(_ca, format)
+ _ca = append(_ca, args...)
+ _m.Called(_ca...)
+}
+
+// Println provides a mock function with given fields: args
+func (_m *Logger) Println(args ...interface{}) {
+ var _ca []interface{}
+ _ca = append(_ca, args...)
+ _m.Called(_ca...)
+}
+
+// Write provides a mock function with given fields: p
+func (_m *Logger) Write(p []byte) (int, error) {
+ ret := _m.Called(p)
+
+ var r0 int
+ if rf, ok := ret.Get(0).(func([]byte) int); ok {
+ r0 = rf(p)
+ } else {
+ r0 = ret.Get(0).(int)
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func([]byte) error); ok {
+ r1 = rf(p)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
diff --git a/test/master/mocks/master__log_client.go b/test/master/mocks/master__log_client.go
new file mode 100644
index 0000000..1e55682
--- /dev/null
+++ b/test/master/mocks/master__log_client.go
@@ -0,0 +1,133 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import context "context"
+import master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+import metadata "google.golang.org/grpc/metadata"
+import mock "github.com/stretchr/testify/mock"
+
+// Master_LogClient is an autogenerated mock type for the Master_LogClient type
+type Master_LogClient struct {
+ mock.Mock
+}
+
+// CloseSend provides a mock function with given fields:
+func (_m *Master_LogClient) CloseSend() error {
+ ret := _m.Called()
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func() error); ok {
+ r0 = rf()
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// Context provides a mock function with given fields:
+func (_m *Master_LogClient) Context() context.Context {
+ ret := _m.Called()
+
+ var r0 context.Context
+ if rf, ok := ret.Get(0).(func() context.Context); ok {
+ r0 = rf()
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(context.Context)
+ }
+ }
+
+ return r0
+}
+
+// Header provides a mock function with given fields:
+func (_m *Master_LogClient) Header() (metadata.MD, error) {
+ ret := _m.Called()
+
+ var r0 metadata.MD
+ if rf, ok := ret.Get(0).(func() metadata.MD); ok {
+ r0 = rf()
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(metadata.MD)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func() error); ok {
+ r1 = rf()
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Recv provides a mock function with given fields:
+func (_m *Master_LogClient) Recv() (*master.LogResponse, error) {
+ ret := _m.Called()
+
+ var r0 *master.LogResponse
+ if rf, ok := ret.Get(0).(func() *master.LogResponse); ok {
+ r0 = rf()
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.LogResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func() error); ok {
+ r1 = rf()
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// RecvMsg provides a mock function with given fields: m
+func (_m *Master_LogClient) RecvMsg(m interface{}) error {
+ ret := _m.Called(m)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(interface{}) error); ok {
+ r0 = rf(m)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// SendMsg provides a mock function with given fields: m
+func (_m *Master_LogClient) SendMsg(m interface{}) error {
+ ret := _m.Called(m)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(interface{}) error); ok {
+ r0 = rf(m)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// Trailer provides a mock function with given fields:
+func (_m *Master_LogClient) Trailer() metadata.MD {
+ ret := _m.Called()
+
+ var r0 metadata.MD
+ if rf, ok := ret.Get(0).(func() metadata.MD); ok {
+ r0 = rf()
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(metadata.MD)
+ }
+ }
+
+ return r0
+}
diff --git a/test/master/mocks/master__log_server.go b/test/master/mocks/master__log_server.go
new file mode 100644
index 0000000..d70c0a0
--- /dev/null
+++ b/test/master/mocks/master__log_server.go
@@ -0,0 +1,104 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import context "context"
+import master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+import metadata "google.golang.org/grpc/metadata"
+import mock "github.com/stretchr/testify/mock"
+
+// Master_LogServer is an autogenerated mock type for the Master_LogServer type
+type Master_LogServer struct {
+ mock.Mock
+}
+
+// Context provides a mock function with given fields:
+func (_m *Master_LogServer) Context() context.Context {
+ ret := _m.Called()
+
+ var r0 context.Context
+ if rf, ok := ret.Get(0).(func() context.Context); ok {
+ r0 = rf()
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(context.Context)
+ }
+ }
+
+ return r0
+}
+
+// RecvMsg provides a mock function with given fields: m
+func (_m *Master_LogServer) RecvMsg(m interface{}) error {
+ ret := _m.Called(m)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(interface{}) error); ok {
+ r0 = rf(m)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// Send provides a mock function with given fields: _a0
+func (_m *Master_LogServer) Send(_a0 *master.LogResponse) error {
+ ret := _m.Called(_a0)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(*master.LogResponse) error); ok {
+ r0 = rf(_a0)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// SendHeader provides a mock function with given fields: _a0
+func (_m *Master_LogServer) SendHeader(_a0 metadata.MD) error {
+ ret := _m.Called(_a0)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(metadata.MD) error); ok {
+ r0 = rf(_a0)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// SendMsg provides a mock function with given fields: m
+func (_m *Master_LogServer) SendMsg(m interface{}) error {
+ ret := _m.Called(m)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(interface{}) error); ok {
+ r0 = rf(m)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// SetHeader provides a mock function with given fields: _a0
+func (_m *Master_LogServer) SetHeader(_a0 metadata.MD) error {
+ ret := _m.Called(_a0)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(metadata.MD) error); ok {
+ r0 = rf(_a0)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// SetTrailer provides a mock function with given fields: _a0
+func (_m *Master_LogServer) SetTrailer(_a0 metadata.MD) {
+ _m.Called(_a0)
+}
diff --git a/test/master/mocks/master_client.go b/test/master/mocks/master_client.go
new file mode 100644
index 0000000..39274e0
--- /dev/null
+++ b/test/master/mocks/master_client.go
@@ -0,0 +1,133 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import context "context"
+import grpc "google.golang.org/grpc"
+import master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+import mock "github.com/stretchr/testify/mock"
+
+// MasterClient is an autogenerated mock type for the MasterClient type
+type MasterClient struct {
+ mock.Mock
+}
+
+// Log provides a mock function with given fields: ctx, in, opts
+func (_m *MasterClient) Log(ctx context.Context, in *master.LogRequest, opts ...grpc.CallOption) (master.Master_LogClient, error) {
+ _va := make([]interface{}, len(opts))
+ for _i := range opts {
+ _va[_i] = opts[_i]
+ }
+ var _ca []interface{}
+ _ca = append(_ca, ctx, in)
+ _ca = append(_ca, _va...)
+ ret := _m.Called(_ca...)
+
+ var r0 master.Master_LogClient
+ if rf, ok := ret.Get(0).(func(context.Context, *master.LogRequest, ...grpc.CallOption) master.Master_LogClient); ok {
+ r0 = rf(ctx, in, opts...)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(master.Master_LogClient)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.LogRequest, ...grpc.CallOption) error); ok {
+ r1 = rf(ctx, in, opts...)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Query provides a mock function with given fields: ctx, in, opts
+func (_m *MasterClient) Query(ctx context.Context, in *master.QueryRequest, opts ...grpc.CallOption) (*master.QueryResponse, error) {
+ _va := make([]interface{}, len(opts))
+ for _i := range opts {
+ _va[_i] = opts[_i]
+ }
+ var _ca []interface{}
+ _ca = append(_ca, ctx, in)
+ _ca = append(_ca, _va...)
+ ret := _m.Called(_ca...)
+
+ var r0 *master.QueryResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.QueryRequest, ...grpc.CallOption) *master.QueryResponse); ok {
+ r0 = rf(ctx, in, opts...)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.QueryResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.QueryRequest, ...grpc.CallOption) error); ok {
+ r1 = rf(ctx, in, opts...)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Start provides a mock function with given fields: ctx, in, opts
+func (_m *MasterClient) Start(ctx context.Context, in *master.StartRequest, opts ...grpc.CallOption) (*master.StartResponse, error) {
+ _va := make([]interface{}, len(opts))
+ for _i := range opts {
+ _va[_i] = opts[_i]
+ }
+ var _ca []interface{}
+ _ca = append(_ca, ctx, in)
+ _ca = append(_ca, _va...)
+ ret := _m.Called(_ca...)
+
+ var r0 *master.StartResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.StartRequest, ...grpc.CallOption) *master.StartResponse); ok {
+ r0 = rf(ctx, in, opts...)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.StartResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.StartRequest, ...grpc.CallOption) error); ok {
+ r1 = rf(ctx, in, opts...)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Stop provides a mock function with given fields: ctx, in, opts
+func (_m *MasterClient) Stop(ctx context.Context, in *master.StopRequest, opts ...grpc.CallOption) (*master.StopResponse, error) {
+ _va := make([]interface{}, len(opts))
+ for _i := range opts {
+ _va[_i] = opts[_i]
+ }
+ var _ca []interface{}
+ _ca = append(_ca, ctx, in)
+ _ca = append(_ca, _va...)
+ ret := _m.Called(_ca...)
+
+ var r0 *master.StopResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.StopRequest, ...grpc.CallOption) *master.StopResponse); ok {
+ r0 = rf(ctx, in, opts...)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.StopResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.StopRequest, ...grpc.CallOption) error); ok {
+ r1 = rf(ctx, in, opts...)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
diff --git a/test/master/mocks/master_client_factory.go b/test/master/mocks/master_client_factory.go
new file mode 100644
index 0000000..4dfb9bf
--- /dev/null
+++ b/test/master/mocks/master_client_factory.go
@@ -0,0 +1,44 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import io "io"
+import master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+import mock "github.com/stretchr/testify/mock"
+
+// MasterClientFactory is an autogenerated mock type for the MasterClientFactory type
+type MasterClientFactory struct {
+ mock.Mock
+}
+
+// CreateClient provides a mock function with given fields: host
+func (_m *MasterClientFactory) CreateClient(host string) (master.MasterClient, io.Closer, error) {
+ ret := _m.Called(host)
+
+ var r0 master.MasterClient
+ if rf, ok := ret.Get(0).(func(string) master.MasterClient); ok {
+ r0 = rf(host)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(master.MasterClient)
+ }
+ }
+
+ var r1 io.Closer
+ if rf, ok := ret.Get(1).(func(string) io.Closer); ok {
+ r1 = rf(host)
+ } else {
+ if ret.Get(1) != nil {
+ r1 = ret.Get(1).(io.Closer)
+ }
+ }
+
+ var r2 error
+ if rf, ok := ret.Get(2).(func(string) error); ok {
+ r2 = rf(host)
+ } else {
+ r2 = ret.Error(2)
+ }
+
+ return r0, r1, r2
+}
diff --git a/test/master/mocks/master_server.go b/test/master/mocks/master_server.go
new file mode 100644
index 0000000..4457200
--- /dev/null
+++ b/test/master/mocks/master_server.go
@@ -0,0 +1,95 @@
+// Code generated by mockery v1.0.0. DO NOT EDIT.
+
+package mocks
+
+import context "context"
+import master "github.com/kmacmcfarlane/go-scheduler/gen/protobuf/master"
+import mock "github.com/stretchr/testify/mock"
+
+// MasterServer is an autogenerated mock type for the MasterServer type
+type MasterServer struct {
+ mock.Mock
+}
+
+// Log provides a mock function with given fields: _a0, _a1
+func (_m *MasterServer) Log(_a0 *master.LogRequest, _a1 master.Master_LogServer) error {
+ ret := _m.Called(_a0, _a1)
+
+ var r0 error
+ if rf, ok := ret.Get(0).(func(*master.LogRequest, master.Master_LogServer) error); ok {
+ r0 = rf(_a0, _a1)
+ } else {
+ r0 = ret.Error(0)
+ }
+
+ return r0
+}
+
+// Query provides a mock function with given fields: _a0, _a1
+func (_m *MasterServer) Query(_a0 context.Context, _a1 *master.QueryRequest) (*master.QueryResponse, error) {
+ ret := _m.Called(_a0, _a1)
+
+ var r0 *master.QueryResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.QueryRequest) *master.QueryResponse); ok {
+ r0 = rf(_a0, _a1)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.QueryResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.QueryRequest) error); ok {
+ r1 = rf(_a0, _a1)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Start provides a mock function with given fields: _a0, _a1
+func (_m *MasterServer) Start(_a0 context.Context, _a1 *master.StartRequest) (*master.StartResponse, error) {
+ ret := _m.Called(_a0, _a1)
+
+ var r0 *master.StartResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.StartRequest) *master.StartResponse); ok {
+ r0 = rf(_a0, _a1)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.StartResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.StartRequest) error); ok {
+ r1 = rf(_a0, _a1)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
+// Stop provides a mock function with given fields: _a0, _a1
+func (_m *MasterServer) Stop(_a0 context.Context, _a1 *master.StopRequest) (*master.StopResponse, error) {
+ ret := _m.Called(_a0, _a1)
+
+ var r0 *master.StopResponse
+ if rf, ok := ret.Get(0).(func(context.Context, *master.StopRequest) *master.StopResponse); ok {
+ r0 = rf(_a0, _a1)
+ } else {
+ if ret.Get(0) != nil {
+ r0 = ret.Get(0).(*master.StopResponse)
+ }
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func(context.Context, *master.StopRequest) error); ok {
+ r1 = rf(_a0, _a1)
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
diff --git a/test/read_closer.go b/test/read_closer.go
new file mode 100644
index 0000000..7ea35ed
--- /dev/null
+++ b/test/read_closer.go
@@ -0,0 +1,22 @@
+package test
+
+import "io"
+
+// MockReadCloser wraps an io.Reader adding an impotent Close method to make it a ReadCloser for testing
+type MockReadCloser struct {
+ reader io.Reader
+}
+
+func NewMockReadCloser(reader io.Reader) MockReadCloser {
+ return MockReadCloser{
+ reader: reader}
+}
+
+func (rc MockReadCloser) Read(p []byte) (n int, err error) {
+ return rc.reader.Read(p)
+}
+
+func (rc MockReadCloser) Close() error {
+ // do nothing
+ return nil
+}
\ No newline at end of file