Working on outputs for jobs

Author: Gergely Brautigam

gaia.go

@@ -219,6 +219,7 @@ type Job struct {
 	Status       JobStatus   `json:"status,omitempty"`
 	Args         []*Argument `json:"args,omitempty"`
 	FailPipeline bool        `json:"failpipeline,omitempty"`
+	Outs         []*Output   `json:"outs,omitempty"`
 }
 
 // Argument represents a single argument of a job
@@ -229,6 +230,12 @@ type Argument struct {
 	Value       string `json:"value,omitempty"`
 }
 
+// Output represents a single output of a job
+type Output struct {
+	Key   string `json:"key,omitempty"`
+	Value string `json:"value,omitempty"`
+}
+
 // CreatePipeline represents a pipeline which is not yet
 // compiled.
 type CreatePipeline struct {

go.mod

@@ -5,6 +5,7 @@ require (
 	github.com/GeertJohan/go.rice v1.0.0
 	github.com/Microsoft/go-winio v0.4.14 // indirect
 	github.com/Pallinder/go-randomdata v1.1.0
+	github.com/Skarlso/protobuf v0.0.0-20200307140829-e2fe748a72a5
 	github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 // indirect
 	github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 // indirect
 	github.com/boltdb/bolt v1.3.1 // indirect
@@ -62,3 +63,5 @@ require (
 	gopkg.in/yaml.v2 v2.2.2
 	gotest.tools v2.2.0+incompatible // indirect
 )
+
+go 1.13

go.sum

@@ -9,6 +9,12 @@ github.com/Microsoft/go-winio v0.4.14 h1:+hMXMk01us9KgxGb7ftKQt2Xpf5hH/yky+TDA+q
 github.com/Microsoft/go-winio v0.4.14/go.mod h1:qXqCSQ3Xa7+6tgxaGTIe4Kpcdsi+P8jBhyzoq1bpyYA=
 github.com/Pallinder/go-randomdata v1.1.0 h1:gUubB1IEUliFmzjqjhf+bgkg1o6uoFIkRsP3VrhEcx8=
 github.com/Pallinder/go-randomdata v1.1.0/go.mod h1:yHmJgulpD2Nfrm0cR9tI/+oAgRqCQQixsA8HyRZfV9Y=
+github.com/Skarlso/protobuf v0.0.0-20180812091451-7be8a901b55a h1:4HqYngkMAP3NjNrDoLMeUL+UWAhSb1Y2NucsOSHAUeM=
+github.com/Skarlso/protobuf v0.0.0-20180812091451-7be8a901b55a/go.mod h1:D/ResgH7nQRyfO3W9YadC7oTWlF9M5YSGvbrRebmLhg=
+github.com/Skarlso/protobuf v0.0.0-20200307130307-c786b2468c2e h1:N8RjWy1dTXt85FSO9rnaQXC33s0XAQlnVSReKGGBAWs=
+github.com/Skarlso/protobuf v0.0.0-20200307130307-c786b2468c2e/go.mod h1:D/ResgH7nQRyfO3W9YadC7oTWlF9M5YSGvbrRebmLhg=
+github.com/Skarlso/protobuf v0.0.0-20200307140829-e2fe748a72a5 h1:GthyChPbR7o0AjYNtdFPCf73/NfWKAnc7riHfxrOFTg=
+github.com/Skarlso/protobuf v0.0.0-20200307140829-e2fe748a72a5/go.mod h1:D/ResgH7nQRyfO3W9YadC7oTWlF9M5YSGvbrRebmLhg=
 github.com/akavel/rsrc v0.8.0 h1:zjWn7ukO9Kc5Q62DOJCcxGpXC18RawVtYAGdz2aLlfw=
 github.com/akavel/rsrc v0.8.0/go.mod h1:uLoCtb9J+EyAqh+26kdrTgmzRBFPGOolLWKpdxkKq+c=
 github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 h1:uSoVVbwJiQipAclBbw+8quDsfcvFjOpI5iCf4p/cqCs=

plugin/grpc.go

@@ -3,7 +3,7 @@ package plugin
 import (
 	"context"
 
-	proto "github.com/gaia-pipeline/protobuf"
+	proto "github.com/Skarlso/protobuf"
 	plugin "github.com/hashicorp/go-plugin"
 	"google.golang.org/grpc"
 )

plugin/plugin.go

@@ -11,9 +11,9 @@ import (
 	"sync"
 	"time"
 
+	proto "github.com/Skarlso/protobuf"
 	"github.com/gaia-pipeline/gaia"
 	"github.com/gaia-pipeline/gaia/security"
-	proto "github.com/gaia-pipeline/protobuf"
 	"github.com/hashicorp/go-plugin"
 )
 
@@ -269,6 +269,18 @@ func (p *GoPlugin) Execute(j *gaia.Job) error {
 		timeString := time.Now().Format(timeFormat)
 		_, _ = p.logger.WriteString(fmt.Sprintf("%s Job '%s' threw an error: %s\n", timeString, j.Title, err.Error()))
 	} else {
+		// We set up the job's output if there was any
+		outs := resultObj.GetOutput()
+		if outs != nil {
+			o := make([]*gaia.Output, 0)
+			for _, out := range outs {
+				o = append(o, &gaia.Output{
+					Key:   out.GetKey(),
+					Value: out.GetValue(),
+				})
+			}
+			j.Outs = o
+		}
 		j.Status = gaia.JobSuccess
 	}
 
@@ -313,6 +325,15 @@ func (p *GoPlugin) GetJobs() ([]*gaia.Job, error) {
 			args = append(args, a)
 		}
 
+		outs := make([]*gaia.Output, 0, len(job.GetOuts()))
+		for _, out := range job.GetOuts() {
+			o := &gaia.Output{
+				Key:   out.GetKey(),
+				Value: out.GetValue(),
+			}
+			outs = append(outs, o)
+		}
+
 		// add proto object to separate list to rebuild dep later.
 		pList = append(pList, job)
 
@@ -323,6 +344,7 @@ func (p *GoPlugin) GetJobs() ([]*gaia.Job, error) {
 			Description: job.Description,
 			Status:      gaia.JobWaitingExec,
 			Args:        args,
+			Outs:        outs,
 		}
 		l = append(l, j)
 		jobsMap[j.ID] = j

plugin/plugin_test.go

@@ -13,8 +13,8 @@ import (
 	"testing"
 	"time"
 
+	proto "github.com/Skarlso/protobuf"
 	"github.com/gaia-pipeline/gaia"
-	proto "github.com/gaia-pipeline/protobuf"
 	"github.com/hashicorp/go-hclog"
 	"google.golang.org/grpc/metadata"
 )

workers/agent/agent.go

@@ -351,6 +351,15 @@ func (a *Agent) scheduleWork() {
 				}
 				j.Args = append(j.Args, a)
 			}
+			// Outputs
+			j.Outs = make([]*gaia.Output, 0, len(job.GetOuts()))
+			for _, out := range job.GetOuts() {
+				o := &gaia.Output{
+					Key:   out.GetKey(),
+					Value: out.GetValue(),
+				}
+				j.Outs = append(j.Outs, o)
+			}
 		}
 
 		// Convert dependencies