added proxy task type (#185)

Author: roi-codefresh

venona/VERSION

@@ -1 +1 @@
-1.4.25
+1.4.26

venona/go.mod

@@ -4,17 +4,19 @@ go 1.14
 
 require (
 	github.com/evanphx/json-patch v4.5.0+incompatible // indirect
+	github.com/fzipp/gocyclo v0.0.0-20150627053110-6acd4345c835 // indirect
 	github.com/gin-gonic/gin v1.6.3
 	github.com/go-playground/validator/v10 v10.3.0 // indirect
 	github.com/golang/protobuf v1.4.2 // indirect
+	github.com/hashicorp/go-retryablehttp v0.6.7
 	github.com/inconshreveable/log15 v0.0.0-20200109203555-b30bc20e4fd1
 	github.com/json-iterator/go v1.1.10 // indirect
 	github.com/mattn/go-colorable v0.1.6 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/spf13/cobra v1.0.0
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.7.0
-	github.com/stretchr/objx v0.2.0 // indirect
+	github.com/stretchr/objx v0.2.0
 	github.com/stretchr/testify v1.5.1
 	golang.org/x/crypto v0.0.0-20200128174031-69ecbb4d6d5d // indirect
 	golang.org/x/net v0.0.0-20200226121028-0de0cce0169b // indirect

venona/go.sum

@@ -68,6 +68,8 @@ github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fzipp/gocyclo v0.0.0-20150627053110-6acd4345c835 h1:roDmqJ4Qes7hrDOsWsMCce0vQHz3xiMPjJ9m4c2eeNs=
+github.com/fzipp/gocyclo v0.0.0-20150627053110-6acd4345c835/go.mod h1:BjL/N0+C+j9uNX+1xcNuM9vdSIcXCZrQZUYbXOFbgN8=
 github.com/ghodss/yaml v0.0.0-20150909031657-73d445a93680/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
@@ -156,10 +158,15 @@ github.com/grpc-ecosystem/grpc-gateway v1.9.0/go.mod h1:vNeuVxBJEsws4ogUvrchl83t
 github.com/hashicorp/consul/api v1.1.0/go.mod h1:VmuI/Lkw1nC05EYQWNKwWGbkg+FbDBtguAZLlVdkD9Q=
 github.com/hashicorp/consul/sdk v0.1.1/go.mod h1:VKf9jXwCTEY1QZP2MOLRhb5i/I/ssyNV1vwHyQBF0x8=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-cleanhttp v0.5.1 h1:dH3aiDG9Jvb5r5+bYHsikaOUIpcM0xvgMXVoDkXMzJM=
 github.com/hashicorp/go-cleanhttp v0.5.1/go.mod h1:JpRdi6/HCYpAwUzNwuwqhbovhLtngrth3wmdIIUrZ80=
+github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
+github.com/hashicorp/go-hclog v0.9.2/go.mod h1:5CU+agLiy3J7N7QjHK5d05KxGsuXiQLrjA0H7acj2lQ=
 github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
 github.com/hashicorp/go-msgpack v0.5.3/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
 github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
+github.com/hashicorp/go-retryablehttp v0.6.7 h1:8/CAEZt/+F7kR7GevNHulKkUjLht3CPmn7egmhieNKo=
+github.com/hashicorp/go-retryablehttp v0.6.7/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
 github.com/hashicorp/go-rootcerts v1.0.0/go.mod h1:K6zTfqpRlCUIjkwsN4Z+hiSfzSTQa6eBIzfwKfwNnHU=
 github.com/hashicorp/go-sockaddr v1.0.0/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
 github.com/hashicorp/go-syslog v1.0.0/go.mod h1:qPfqrKkXGihmCqbJM2mZgkZGvKG1dFdvsLplgctolz4=

venona/pkg/agent/agent.go

@@ -15,28 +15,42 @@
 package agent
 
 import (
+	"bytes"
+	"encoding/json"
 	"errors"
+	"fmt"
+	"io/ioutil"
 	"sync"
 	"time"
 
 	"github.com/codefresh-io/go/venona/pkg/codefresh"
 	"github.com/codefresh-io/go/venona/pkg/logger"
 	"github.com/codefresh-io/go/venona/pkg/runtime"
 	"github.com/codefresh-io/go/venona/pkg/task"
+	retryablehttp "github.com/hashicorp/go-retryablehttp"
+	"github.com/stretchr/objx"
 )
 
+// internal errors
 var (
-	errAlreadyRunning   = errors.New("Agent already running")
-	errAlreadyStopped   = errors.New("Agent already stopped")
-	errOptionsRequired  = errors.New("Options are required")
-	errIDRequired       = errors.New("ID options is required")
-	errRuntimesRequired = errors.New("Runtimes options is required")
-	errLoggerRequired   = errors.New("Logger options is required")
+	errAlreadyRunning           = errors.New("Agent already running")
+	errAlreadyStopped           = errors.New("Agent already stopped")
+	errOptionsRequired          = errors.New("Options are required")
+	errIDRequired               = errors.New("ID options is required")
+	errRuntimesRequired         = errors.New("Runtimes options is required")
+	errLoggerRequired           = errors.New("Logger options is required")
+	errFailedToParseAgentTask   = errors.New("Failed to parse agent task spec")
+	errUknownAgentTaskType      = errors.New("Agent task has unknown type")
+	errAgentTaskMalformedParams = errors.New("failed to marshal agent task params")
+	errProxyTaskWithoutURL      = errors.New(`url not provided for task of type "proxy"`)
+	errProxyTaskWithoutToken    = errors.New(`token not provided for task of type "proxy"`)
 )
 
 const (
 	defaultTaskPullingInterval     = time.Second * 3
 	defaultStatusReportingInterval = time.Second * 10
+	defaultProxyRequestTimeout     = time.Second * 10
+	defaultProxyRequestRetries     = 3
 )
 
 type (
@@ -77,6 +91,14 @@ type (
 	}
 )
 
+var (
+	httpClient = retryablehttp.NewClient()
+
+	agentTaskExecutors = map[string]func(t *task.AgentTask, log logger.Logger) error{
+		"proxy": proxyRequest,
+	}
+)
+
 // New creates a new Agent instance
 func New(opt *Options) (*Agent, error) {
 	if err := checkOptions(opt); err != nil {
@@ -210,17 +232,33 @@ func pullTasks(client codefresh.Codefresh, logger logger.Logger) []task.Task {
 func startTasks(tasks []task.Task, runtimes map[string]runtime.Runtime, logger logger.Logger) {
 	creationTasks := []task.Task{}
 	deletionTasks := []task.Task{}
+	agentTasks := []task.Task{}
+
+	// divide tasks by types
 	for _, t := range tasks {
-		logger.Debug("Received task", "type", t.Type, "workflow", t.Metadata.Workflow, "runtime", t.Metadata.ReName)
-		if t.Type == task.TypeCreatePod || t.Type == task.TypeCreatePVC {
+		logger.Debug("Received task", "type", t.Type, "tid", t.Metadata.Workflow, "runtime", t.Metadata.ReName)
+		switch t.Type {
+		case task.TypeCreatePod, task.TypeCreatePVC:
 			creationTasks = append(creationTasks, t)
+		case task.TypeDeletePod, task.TypeDeletePVC:
+			deletionTasks = append(deletionTasks, t)
+		case task.TypeAgentTask:
+			agentTasks = append(agentTasks, t)
+		default:
+			logger.Error("unrecognized task type", "type", t.Type, "tid", t.Metadata.Workflow, "runtime", t.Metadata.ReName)
 		}
+	}
 
-		if t.Type == task.TypeDeletePod || t.Type == task.TypeDeletePVC {
-			deletionTasks = append(deletionTasks, t)
+	// process agent tasks
+	for i := range agentTasks {
+		t := agentTasks[i]
+		logger.Info("executing agent task", "tid", t.Metadata.Workflow)
+		if err := executeAgentTask(&t, logger); err != nil {
+			logger.Error(err.Error())
 		}
 	}
 
+	// process creation tasks
 	for _, tasks := range groupTasks(creationTasks) {
 		reName := tasks[0].Metadata.ReName
 		runtime, ok := runtimes[reName]
@@ -233,6 +271,8 @@ func startTasks(tasks []task.Task, runtimes map[string]runtime.Runtime, logger l
 			logger.Error(err.Error())
 		}
 	}
+
+	// process deletion tasks
 	for _, tasks := range groupTasks(deletionTasks) {
 		reName := tasks[0].Metadata.ReName
 		runtime, ok := runtimes[reName]
@@ -249,6 +289,71 @@ func startTasks(tasks []task.Task, runtimes map[string]runtime.Runtime, logger l
 	}
 }
 
+func executeAgentTask(t *task.Task, log logger.Logger) error {
+	specJSON, err := json.Marshal(t.Spec)
+	if err != nil {
+		return errFailedToParseAgentTask
+	}
+
+	spec := task.AgentTask{}
+	if err = json.Unmarshal(specJSON, &spec); err != nil {
+		return errFailedToParseAgentTask
+	}
+
+	e, ok := agentTaskExecutors[spec.Type]
+	if !ok {
+		return errUknownAgentTaskType
+	}
+
+	return e(&spec, log)
+}
+
+func proxyRequest(t *task.AgentTask, log logger.Logger) error {
+	spec := objx.Map(t.Params)
+	vars := objx.Map(spec.Get("runtimeContext.context.variables").MSI())
+	token := spec.Get("runtimeContext.context.eventReporting.token").Str()
+	if token == "" {
+		return errProxyTaskWithoutToken
+	}
+
+	url := vars.Get("proxyUrl").Str()
+	if url == "" {
+		return errProxyTaskWithoutURL
+	}
+
+	method := vars.Get("method").Str("POST")
+
+	json, err := json.Marshal(t.Params)
+	if err != nil {
+		return errAgentTaskMalformedParams
+	}
+	if json == nil {
+		json = []byte{}
+	}
+
+	req, err := retryablehttp.NewRequest(method, url, bytes.NewReader(json))
+	if err != nil {
+		return err
+	}
+
+	req.Header.Add("x-req-type", "workflow-request")
+	req.Header.Add("x-access-token", token)
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("Content-Length", fmt.Sprintf("%v", len(json)))
+
+	log.Info("executing proxy task", "url", url, "method", method)
+
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		return err
+	}
+	body, _ := ioutil.ReadAll(resp.Body)
+
+	log.Info("finished proxy task", "url", url, "method", method, "status", resp.Status, "body", string(body))
+
+	return nil
+}
+
 func groupTasks(tasks []task.Task) map[string][]task.Task {
 	candidates := map[string][]task.Task{}
 	for _, task := range tasks {
@@ -282,3 +387,8 @@ func checkOptions(opt *Options) error {
 
 	return nil
 }
+
+func init() {
+	httpClient.RetryMax = defaultProxyRequestRetries
+	httpClient.HTTPClient.Timeout = defaultProxyRequestTimeout
+}

venona/pkg/agent/agent_test.go

@@ -224,6 +224,114 @@ func TestNew(t *testing.T) {
 	}
 }
 
+func Test_executeAgentTask(t *testing.T) {
+	executorCalled := false
+	okExecutor := func(t *task.AgentTask, log logger.Logger) error {
+		executorCalled = true
+		return nil
+	}
+
+	badExecutor := func(t *task.AgentTask, log logger.Logger) error {
+		executorCalled = true
+		return errProxyTaskWithoutURL
+	}
+
+	type args struct {
+		executorName string
+		executorFunc func(*task.AgentTask, logger.Logger) error
+		task         *task.Task
+	}
+
+	tests := []struct {
+		name    string
+		args    *args
+		wantErr error
+	}{
+		{
+			name: "should successfully run executor and return nil",
+			args: &args{
+				executorName: "test",
+				executorFunc: okExecutor,
+				task: &task.Task{
+					Type:     task.TypeAgentTask,
+					Metadata: task.Metadata{},
+					Spec: task.AgentTask{
+						Type:   "test",
+						Params: nil,
+					},
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "should call an executor and return an error",
+			args: &args{
+				executorName: "test",
+				executorFunc: badExecutor,
+				task: &task.Task{
+					Type:     task.TypeAgentTask,
+					Metadata: task.Metadata{},
+					Spec: task.AgentTask{
+						Type:   "test",
+						Params: nil,
+					},
+				},
+			},
+			wantErr: errProxyTaskWithoutURL,
+		},
+		{
+			name: "should pass the agent task spec to the executor",
+			args: &args{
+				executorName: "test",
+				executorFunc: func(t *task.AgentTask, l logger.Logger) error {
+					executorCalled = true
+					data, ok := t.Params["data"].(float64)
+					if !ok {
+						return fmt.Errorf("expected data to be of type int")
+					}
+					if data != 3 {
+						return fmt.Errorf("expected data to equal 3 but data=%v", data)
+					}
+					return nil
+				},
+				task: &task.Task{
+					Type:     task.TypeAgentTask,
+					Metadata: task.Metadata{},
+					Spec: task.AgentTask{
+						Type: "test",
+						Params: map[string]interface{}{
+							"data": 3,
+						},
+					},
+				},
+			},
+			wantErr: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		executorCalled = false
+		agentTaskExecutors[tt.args.executorName] = tt.args.executorFunc
+		t.Run(tt.name, func(t *testing.T) {
+			ret := executeAgentTask(tt.args.task, getLoggerMock())
+			if !executorCalled {
+				t.Errorf("executor function hasn't been called")
+			}
+			if ret == nil && tt.wantErr != nil {
+				t.Errorf("expected error %v but got nil", tt.wantErr)
+			}
+			if ret != nil && tt.wantErr == nil {
+				t.Errorf("expected nil but got an error: %v", ret)
+			}
+			if ret != nil && ret.Error() != tt.wantErr.Error() {
+				t.Errorf("expected error: %v but got error: %v", tt.wantErr.Error(), ret.Error())
+			}
+
+		})
+		delete(agentTaskExecutors, tt.args.executorName)
+	}
+}
+
 func createMockAgent() *Agent {
 	runtimes := make(map[string]runtime.Runtime)
 	runtimes["x"] = runtime.New(runtime.Options{})

venona/pkg/task/task.go

@@ -14,14 +14,17 @@
 
 package task
 
-import "encoding/json"
+import (
+	"encoding/json"
+)
 
 // Const for task types
 const (
 	TypeCreatePod = "CreatePod"
 	TypeCreatePVC = "CreatePvc"
 	TypeDeletePod = "DeletePod"
 	TypeDeletePVC = "DeletePvc"
+	TypeAgentTask = "AgentTask"
 )
 
 // UnmarshalTasks with json
@@ -53,3 +56,9 @@ type Metadata struct {
 	ReName    string `json:"reName"`
 	Workflow  string `json:"workflow"`
 }
+
+// AgentTask describes a task of type "AgentTask"
+type AgentTask struct {
+	Type   string                 `json:"type"`
+	Params map[string]interface{} `json:"params"`
+}

venonactl/VERSION

@@ -1 +1 @@
-1.4.25
+1.4.26