feat: jobs cli commands

Signed-off-by: Valery Piashchynski <[email protected]>

8 files changed, 194 insertions, 14 deletions

diff --git a/internal/cli/jobs/command.go b/internal/cli/jobs/command.go
new file mode 100644
index 00000000..4237839a
--- /dev/null
+++ b/internal/cli/jobs/command.go
@@ -0,0 +1,75 @@
+package jobs
+
+import (
+	"strings"
+
+	internalRpc "github.com/roadrunner-server/roadrunner/v2/internal/rpc"
+
+	"github.com/roadrunner-server/errors"
+	"github.com/spf13/cobra"
+)
+
+const (
+	listRPC    string = "jobs.List"
+	pauseRPC   string = "jobs.Pause"
+	destroyRPC string = "jobs.Destroy"
+	resumeRPC  string = "jobs.Resume"
+)
+
+// NewCommand creates `jobs` command.
+func NewCommand(cfgFile *string, override *[]string, silent *bool) *cobra.Command { //nolint:funlen
+	var (
+		pausePipes   bool
+		destroyPipes bool
+		resumePipes  bool
+		listPipes    bool
+	)
+
+	cmd := &cobra.Command{
+		Use:   "jobs",
+		Short: "Jobs pipelines manipulation",
+		RunE: func(_ *cobra.Command, args []string) error {
+			const op = errors.Op("jobs_command")
+
+			if cfgFile == nil {
+				return errors.E(op, errors.Str("no configuration file provided"))
+			}
+
+			// for the commands other than list, args[1] should contain list of pipelines to pause/resume/destroy
+			if !listPipes && len(args[0]) == 0 {
+				return errors.Str("pause/resume/destroy commands should have list of the pipelines as second arg")
+			}
+
+			client, err := internalRpc.NewClient(*cfgFile, *override)
+			if err != nil {
+				return err
+			}
+
+			defer func() { _ = client.Close() }()
+
+			switch {
+			case pausePipes:
+				split := strings.Split(strings.Trim(args[0], " "), ",")
+				return pause(client, split, silent)
+			case destroyPipes:
+				split := strings.Split(strings.Trim(args[0], " "), ",")
+				return destroy(client, split, silent)
+			case resumePipes:
+				split := strings.Split(strings.Trim(args[0], " "), ",")
+				return resume(client, split, silent)
+			case listPipes:
+				return list(client)
+			default:
+				return errors.Str("command should be in form of: `rr jobs pause pipe1,pipe2,etc`")
+			}
+		},
+	}
+
+	// commands
+	cmd.Flags().BoolVar(&pausePipes, "pause", false, "pause pipelines")
+	cmd.Flags().BoolVar(&destroyPipes, "destroy", false, "destroy pipelines")
+	cmd.Flags().BoolVar(&resumePipes, "resume", false, "resume pipelines")
+	cmd.Flags().BoolVar(&listPipes, "list", false, "list pipelines")
+
+	return cmd
+}
diff --git a/internal/cli/jobs/command_test.go b/internal/cli/jobs/command_test.go
new file mode 100644
index 00000000..74042479
--- /dev/null
+++ b/internal/cli/jobs/command_test.go
@@ -0,0 +1,17 @@
+package jobs_test
+
+import (
+	"testing"
+
+	"github.com/roadrunner-server/roadrunner/v2/internal/cli/jobs"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCommandProperties(t *testing.T) {
+	path := ""
+	f := false
+	cmd := jobs.NewCommand(&path, nil, &f)
+
+	assert.Equal(t, "jobs", cmd.Use)
+	assert.NotNil(t, cmd.RunE)
+}
diff --git a/internal/cli/jobs/render.go b/internal/cli/jobs/render.go
new file mode 100644
index 00000000..c9f71e2d
--- /dev/null
+++ b/internal/cli/jobs/render.go
@@ -0,0 +1,22 @@
+package jobs
+
+import (
+	"io"
+
+	"github.com/olekukonko/tablewriter"
+)
+
+// JobsCommandsRender uses console renderer to show jobs
+func renderPipelines(writer io.Writer, pipelines []string) *tablewriter.Table {
+	tw := tablewriter.NewWriter(writer)
+	tw.SetAutoWrapText(false)
+	tw.SetHeader([]string{"Pipeline(s)"})
+	tw.SetColWidth(50)
+	tw.SetAlignment(tablewriter.ALIGN_LEFT)
+
+	for i := 0; i < len(pipelines); i++ {
+		tw.Append([]string{pipelines[i]})
+	}
+
+	return tw
+}
diff --git a/internal/cli/jobs/subcommands.go b/internal/cli/jobs/subcommands.go
new file mode 100644
index 00000000..fe47f33f
--- /dev/null
+++ b/internal/cli/jobs/subcommands.go
@@ -0,0 +1,70 @@
+package jobs
+
+import (
+	"net/rpc"
+	"os"
+
+	jobsv1 "go.buf.build/protocolbuffers/go/roadrunner-server/api/proto/jobs/v1"
+)
+
+func pause(client *rpc.Client, pause []string, silent *bool) error {
+	pipes := &jobsv1.Pipelines{Pipelines: pause}
+	er := &jobsv1.Empty{}
+
+	err := client.Call(pauseRPC, pipes, er)
+	if err != nil {
+		return err
+	}
+
+	if !*silent {
+		renderPipelines(os.Stdout, pause).Render()
+	}
+
+	return nil
+}
+
+func resume(client *rpc.Client, resume []string, silent *bool) error {
+	pipes := &jobsv1.Pipelines{Pipelines: resume}
+	er := &jobsv1.Empty{}
+
+	err := client.Call(resumeRPC, pipes, er)
+	if err != nil {
+		return err
+	}
+
+	if !*silent {
+		renderPipelines(os.Stdout, resume).Render()
+	}
+
+	return nil
+}
+
+func destroy(client *rpc.Client, destroy []string, silent *bool) error {
+	pipes := &jobsv1.Pipelines{Pipelines: destroy}
+	resp := &jobsv1.Pipelines{}
+
+	err := client.Call(destroyRPC, pipes, resp)
+	if err != nil {
+		return err
+	}
+
+	if !*silent {
+		renderPipelines(os.Stdout, resp.GetPipelines()).Render()
+	}
+
+	return nil
+}
+
+func list(client *rpc.Client) error {
+	resp := &jobsv1.Pipelines{}
+	er := &jobsv1.Empty{}
+
+	err := client.Call(listRPC, er, resp)
+	if err != nil {
+		return err
+	}
+
+	renderPipelines(os.Stdout, resp.GetPipelines()).Render()
+
+	return nil
+}
diff --git a/internal/cli/reset/command.go b/internal/cli/reset/command.go
index 084cfeff..f48856cc 100644
--- a/internal/cli/reset/command.go
+++ b/internal/cli/reset/command.go
@@ -10,18 +10,18 @@ import (
 	"github.com/spf13/cobra"
 )
 
+const (
+	op            = errors.Op("reset_handler")
+	resetterList  = "resetter.List"
+	resetterReset = "resetter.Reset"
+)
+
 // NewCommand creates `reset` command.
 func NewCommand(cfgFile *string, override *[]string, silent *bool) *cobra.Command {
 	return &cobra.Command{
 		Use:   "reset",
 		Short: "Reset workers of all or specific RoadRunner service",
 		RunE: func(_ *cobra.Command, args []string) error {
-			const (
-				op            = errors.Op("reset_handler")
-				resetterList  = "resetter.List"
-				resetterReset = "resetter.Reset"
-			)
-
 			if cfgFile == nil {
 				return errors.E(op, errors.Str("no configuration file provided"))
 			}
diff --git a/internal/cli/reset/command_test.go b/internal/cli/reset/command_test.go
index 509354e2..8192ca9c 100644
--- a/internal/cli/reset/command_test.go
+++ b/internal/cli/reset/command_test.go
@@ -16,7 +16,3 @@ func TestCommandProperties(t *testing.T) {
 	assert.Equal(t, "reset", cmd.Use)
 	assert.NotNil(t, cmd.RunE)
 }
-
-func TestExecution(t *testing.T) {
-	t.Skip("Command execution is not implemented yet")
-}
diff --git a/internal/cli/root.go b/internal/cli/root.go
index 158eb478..81ea75d2 100644
--- a/internal/cli/root.go
+++ b/internal/cli/root.go
@@ -8,6 +8,7 @@ import (
 	"strconv"
 
 	"github.com/roadrunner-server/errors"
+	"github.com/roadrunner-server/roadrunner/v2/internal/cli/jobs"
 	"github.com/roadrunner-server/roadrunner/v2/internal/cli/reset"
 	"github.com/roadrunner-server/roadrunner/v2/internal/cli/serve"
 	"github.com/roadrunner-server/roadrunner/v2/internal/cli/stop"
@@ -129,6 +130,7 @@ func NewCommand(cmdName string) *cobra.Command { //nolint:funlen,gocognit
 		reset.NewCommand(cfgFile, override, silent),
 		serve.NewCommand(override, cfgFile, silent),
 		stop.NewCommand(silent, forceStop),
+		jobs.NewCommand(cfgFile, override, silent),
 	)
 
 	return cmd
diff --git a/internal/cli/workers/command.go b/internal/cli/workers/command.go
index 78d8d1d2..421e9afc 100644
--- a/internal/cli/workers/command.go
+++ b/internal/cli/workers/command.go
@@ -20,10 +20,8 @@ import (
 
 // NewCommand creates `workers` command.
 func NewCommand(cfgFile *string, override *[]string) *cobra.Command { //nolint:funlen
-	var (
-		// interactive workers updates
-		interactive bool
-	)
+	// interactive workers updates
+	var interactive bool
 
 	cmd := &cobra.Command{
 		Use:   "workers",