Merge pull request #4 from kristofferlind/custom-output

* add custom output type where artifact check is a command
* streaming logs when singlethreaded

examples/.knega.root.yml

@@ -1,6 +1,6 @@
 componentPaths: ["components"]
 componentSearchDepth: 2
-concurrency: 2
+concurrency: 4
 
 defaultConfiguration:
   inputs:
@@ -17,7 +17,4 @@ defaultConfiguration:
   actions:
     build:
       commands:
-      # - 'echo "Component name: $COMPONENT_NAME"'
-      # - 'echo "Root path: $ROOT"'
-      # - 'echo "Inputs hash: $INPUTS_HASH"'
       - $ROOT/scripts/build.sh

examples/components/custom-output-1/.component.yml

@@ -0,0 +1,5 @@
+name: "custom-output-1"
+
+outputs:
+- type: "Custom"
+  command: "docker manifest inspect localhost:5000/$COMPONENT_NAME:$INPUTS_HASH --insecure"

examples/components/custom-output-1/Dockerfile

@@ -0,0 +1 @@
+FROM alpine

internal/command/command.go

@@ -11,7 +11,16 @@ import (
 )
 
 func ExecuteStopOnError(command string, directory string, variables map[string]string) string {
-	result, err := Execute(command, directory, variables)
+	result, err := ComplexExecute(command, directory, variables, false)
+	if err != nil {
+		log.Print(result)
+		log.Fatal(err)
+	}
+	return result
+}
+
+func ExecuteStopOnErrorWithStreamingLogs(command string, directory string, variables map[string]string, shouldStreamLogs bool) string {
+	result, err := ComplexExecute(command, directory, variables, shouldStreamLogs)
 	if err != nil {
 		log.Print(result)
 		log.Fatal(err)
@@ -20,6 +29,10 @@ func ExecuteStopOnError(command string, directory string, variables map[string]s
 }
 
 func Execute(command string, directory string, variables map[string]string) (string, error) {
+	return ComplexExecute(command, directory, variables, false)
+}
+
+func ComplexExecute(command string, directory string, variables map[string]string, shouldStreamLogs bool) (string, error) {
 	expandedCommand := command
 	environment := os.Environ()
 
@@ -42,6 +55,10 @@ func Execute(command string, directory string, variables map[string]string) (str
 
 	commandExecutor.Stderr = commandExecutor.Stdout
 
+	if shouldStreamLogs == true {
+		commandExecutor.Stdout = os.Stdout
+	}
+
 	startError := commandExecutor.Start()
 
 	if startError != nil {

internal/configuration/component.go

@@ -21,7 +21,7 @@ type Component struct {
 	Path         string
 	Name         string
 	Inputs       []input
-	Outputs      []output
+	Outputs      []Output
 	Actions      map[string]action
 	InputsHash   string
 	ChangeStatus ChangeStatus
@@ -63,9 +63,10 @@ func (component *Component) HasChanges() bool {
 	variables := map[string]string{
 		"COMPONENT_NAME": component.Name,
 		"INPUTS_HASH":    component.InputsHash,
+		"ROOT":           GetRepository().RootPath,
 	}
 	for _, output := range component.Outputs {
-		if !output.exists(variables) {
+		if !output.Exists(variables) {
 			component.ChangeStatus = Dirty
 			hasChanges = true
 		}

internal/configuration/configuration.go

@@ -23,6 +23,7 @@ type configurationOutput struct {
 	Type       string `yaml:"type"`
 	Repository string `yaml:"repository"`
 	Tag        string `yaml:"tag"`
+	Command    string `yaml:"command"`
 }
 
 type configurationAction struct {
@@ -53,12 +54,12 @@ func parseRootConfiguration(path string) rootConfiguration {
 func parseComponentConfiguration(path string) componentConfiguration {
 	fileContent, err := ioutil.ReadFile(path)
 	if err != nil {
-		log.Fatalf("Error reading content of root config file: %v", err)
+		log.Fatalf("Error reading content of component config file: %v", err)
 	}
 	configuration := componentConfiguration{}
 	err = yaml.Unmarshal(fileContent, &configuration)
 	if err != nil {
-		log.Fatalf("Error parsing root config: %v", err)
+		log.Fatalf("Error parsing component config: %v at %s", err, path)
 	}
 
 	return configuration

internal/configuration/output.go

@@ -8,26 +8,43 @@ import (
 	"github.com/kristofferlind/knega/internal/command"
 )
 
-type output struct {
+type Output interface {
+	Exists(map[string]string) bool
+}
+
+type ociOutput struct {
 	repository string
 	tag        string
 }
 
-func initializeOutputs(configurationOutputs []configurationOutput) []output {
-	var outputs []output
+type customOutput struct {
+	command string
+}
+
+func initializeOutputs(configurationOutputs []configurationOutput) []Output {
+	var outputs []Output
 
 	for _, configurationOutput := range configurationOutputs {
-		output := output{
-			repository: configurationOutput.Repository,
-			tag:        configurationOutput.Tag,
+		switch configurationOutput.Type {
+		case "OCI":
+			output := ociOutput{
+				repository: configurationOutput.Repository,
+				tag:        configurationOutput.Tag,
+			}
+			outputs = append(outputs, &output)
+
+		case "Custom":
+			output := customOutput{
+				command: configurationOutput.Command,
+			}
+			outputs = append(outputs, &output)
 		}
-		outputs = append(outputs, output)
 	}
 
 	return outputs
 }
 
-func (output *output) exists(variables map[string]string) bool {
+func (output *ociOutput) Exists(variables map[string]string) bool {
 	repository := GetRepository()
 	registrySecurityArg := ""
 	allowInsecureRegistry := os.Getenv("KNEGA_ALLOW_INSECURE_REGISTRY")
@@ -37,11 +54,24 @@ func (output *output) exists(variables map[string]string) bool {
 
 	result, err := command.Execute("docker manifest inspect "+output.repository+":"+output.tag+registrySecurityArg, repository.RootPath, variables)
 
-	log.Print(result)
-
 	if err != nil {
+		log.Print(result)
+		log.Print(err)
 		return false
 	}
 
 	return !strings.Contains(result, "no such manifest")
 }
+
+func (output *customOutput) Exists(variables map[string]string) bool {
+	repository := GetRepository()
+	result, err := command.Execute(output.command, repository.RootPath, variables)
+
+	if err != nil {
+		log.Print(result)
+		log.Print(err)
+		return false
+	}
+
+	return true
+}

internal/orchestrator/pipeline.go

@@ -27,19 +27,25 @@ func createPipeline(jobs []job) {
 
 	for workerId := 1; workerId <= threads; workerId++ {
 		asyncWorkers.Add(1)
-		go createWorker(&asyncWorkers, queue)
+		go createWorker(&asyncWorkers, queue, threads)
 	}
 
 	asyncWorkers.Wait()
 }
 
-func createWorker(asyncWorkers *sync.WaitGroup, jobs <-chan job) {
+func createWorker(asyncWorkers *sync.WaitGroup, jobs <-chan job, threadCount int) {
 	for job := range jobs {
 		jobStartTime := time.Now()
 
+		var shouldStreamLogs bool = false
+
+		if threadCount < 2 && IsTrace() {
+			shouldStreamLogs = true
+		}
+
 		for _, jobCommand := range job.commands {
-			output := command.ExecuteStopOnError(jobCommand, job.workingDirectory, job.variables)
-			if IsTrace() {
+			output := command.ExecuteStopOnErrorWithStreamingLogs(jobCommand, job.workingDirectory, job.variables, shouldStreamLogs)
+			if threadCount > 1 && IsTrace() {
 				log.Printf("%s", output)
 			}
 		}

test/tests.bats

@@ -23,7 +23,9 @@ load assert.bash
   assert_component_status "example-1" "Rebuild required"
   assert_component_status "example-2" "Rebuild required"
   assert_component_status "with-test-1" "Rebuild required"
+  assert_component_status "custom-output-1" "Rebuild required"
   refute_output_contains "test-no-build-1"
+  assert_output_contains "Total time taken"
 }
 
 @test "knega changed test should skip components that were built" {
@@ -37,11 +39,12 @@ load assert.bash
   assert_component_status "example-1" "Artifacts found"
   assert_component_status "example-2" "Artifacts found"
   assert_component_status "with-test-1" "Artifacts found"
+  assert_component_status "custom-output-1" "Artifacts found"
 }
 
 @test "knega all build should build everything even if artifacts already exist" {
   run $KNEGA all build
-  assert_output_contains "Created 3 jobs"
+  assert_output_contains "Created 4 jobs"
 }
 
 @test "knega all test should test everything even if artifacts already exist" {