[WIP] feat : Add logic in manager to monitor Pending PipelineRuns and run them

go.mod

@@ -86,7 +86,7 @@ require (
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.7.0 // indirect
 	go.uber.org/zap v1.19.1 // indirect
-	golang.org/x/crypto v0.0.0-20220214200702-86341886e292 // indirect
+	golang.org/x/crypto v0.0.0-20220214200702-86341886e292
 	golang.org/x/net v0.0.0-20220325170049-de3da57026de // indirect
 	golang.org/x/oauth2 v0.0.0-20220309155454-6242fa91716a
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect

go.sum

@@ -187,8 +187,6 @@ github.com/go-openapi/swag v0.19.15 h1:D2NRCBzS9/pEY3gP9Nl8aDqGUcPFrwG2p+CNFrLyr
 github.com/go-openapi/swag v0.19.15/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
 github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
-github.com/go-stomp/stomp/v3 v3.0.0 h1:SnKnOoBkx/2MLxBeANKlyr+O8+Zx10H0ONub5ax6v/Y=
-github.com/go-stomp/stomp/v3 v3.0.0/go.mod h1:jTrybHBK20jPdM9iyh65m6GusX6aMf7atfEFZ1nIcgc=
 github.com/go-stomp/stomp/v3 v3.0.5 h1:yOORvXLqSu0qF4loJjfWrcVE1o0+9cFudclcP0an36Y=
 github.com/go-stomp/stomp/v3 v3.0.5/go.mod h1:ztzZej6T2W4Y6FlD+Tb5n7HQP3/O5UNQiuC169pIp10=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
@@ -212,6 +210,7 @@ github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
 github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -247,7 +246,6 @@ github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
 github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
 github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
@@ -570,8 +568,6 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
-golang.org/x/exp v0.0.0-20220518171630-0b5c67f07fdf h1:oXVg4h2qJDd9htKxb5SCpFBHLipW6hXmL3qpUixS2jw=
-golang.org/x/exp v0.0.0-20220518171630-0b5c67f07fdf/go.mod h1:yh0Ynu2b5ZUe3MQfp2nM0ecK7wsgouWTDN0FNeJuIys=
 golang.org/x/exp v0.0.0-20231127185646-65229373498e h1:Gvh4YaCaXNs6dKTlfgismwWZKyjVZXwOPfIyUaqU3No=
 golang.org/x/exp v0.0.0-20231127185646-65229373498e/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
@@ -600,7 +596,7 @@ golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.5.0/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
-golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 h1:kQgndtyPBW/JIYERgdxfwMYh3AVStj88WQTlNDi2a+o=
+golang.org/x/mod v0.14.0 h1:dGoOF9QVLYng8IHTm7BAyWqCqSheQ5pYWGhzW00YJr0=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -845,11 +841,10 @@ golang.org/x/tools v0.1.2/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.3/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.4/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
-golang.org/x/tools v0.1.10 h1:QjFRCZxdOhBJ/UNgnBZLbNV13DlbnK0quyivTnXJM20=
+golang.org/x/tools v0.16.0 h1:GO788SKMRunPIBCXiQyo2AaexLstOrVhuAL5YwsckQM=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gomodules.xyz/jsonpatch/v2 v2.2.0 h1:4pT439QV83L+G9FkcCriY6EkpcK6r6bK+A5FBUMI7qY=
 gomodules.xyz/jsonpatch/v2 v2.2.0/go.mod h1:WXp+iVDkoLQqPudfQ9GBlwB2eZ5DKOnjQZCYdOS8GPY=
@@ -1066,17 +1061,13 @@ k8s.io/gengo v0.0.0-20210813121822-485abfe95c7c/go.mod h1:FiNAH4ZV3gBg2Kwh89tzAE
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0/go.mod h1:Od+F08eJP+W3HUb4pSrPpgp9DGU4GzlpG/TmITuYh/Y=
 k8s.io/klog/v2 v2.30.0/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
-k8s.io/klog/v2 v2.60.1-0.20220317184644-43cc75f9ae89 h1:bUNlsw5yb353zbKMj8srOr6V2Ajhz1VkTKonP1L8r2o=
-k8s.io/klog/v2 v2.60.1-0.20220317184644-43cc75f9ae89/go.mod h1:N3kgBtsFxMb4nQ0eBDgbHEt/dtxBuTkSFQ+7K5OUoz4=
 k8s.io/klog/v2 v2.80.1 h1:atnLQ121W371wYYFawwYx1aEY2eUfs4l3J72wtgAwV4=
 k8s.io/klog/v2 v2.80.1/go.mod h1:y1WjHnz7Dj687irZUWR/WLkLc5N1YHtjLdmgWjndZn0=
 k8s.io/kube-openapi v0.0.0-20211115234752-e816edb12b65/go.mod h1:sX9MT8g7NVZM5lVL/j8QyCCJe8YSMW30QvGZWaCIDIk=
 k8s.io/kube-openapi v0.0.0-20220124234850-424119656bbf h1:M9XBsiMslw2lb2ZzglC0TOkBPK5NQi0/noUrdnoFwUg=
 k8s.io/kube-openapi v0.0.0-20220124234850-424119656bbf/go.mod h1:sX9MT8g7NVZM5lVL/j8QyCCJe8YSMW30QvGZWaCIDIk=
 k8s.io/utils v0.0.0-20210802155522-efc7438f0176/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20211116205334-6203023598ed/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 h1:HNSDgDCrr/6Ly3WEGKZftiE7IY19Vz2GdbOCyI4qqhc=
-k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20231127182322-b307cd553661 h1:FepOBzJ0GXm8t0su67ln2wAZjbQ6RxQGZDnzuLcrUTI=
 k8s.io/utils v0.0.0-20231127182322-b307cd553661/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
 knative.dev/pkg v0.0.0-20220329144915-0a1ec2e0d46c h1:PLzCwnuPt+bjwiyYrRrsbmiBWR8a5cIwKtTd3LK1TqU=

pkg/manager/manager.go

@@ -3,6 +3,7 @@ package manager
 import (
 	"encoding/base64"
 	"fmt"
+	"github.com/devtools-qe-incubator/eventmanager/pkg/manager/scheduling/pipelines"
 	"os"
 	"os/signal"
 	"strings"
@@ -36,6 +37,11 @@ func Initialize(providersFilePath string, flowsFilePath []string) {
 		logging.Error(err)
 		os.Exit(1)
 	}
+	stopChan := make(chan bool)
+	if err := pipelines.ManagePipelineRuns(stopChan); err != nil {
+		logging.Error(err)
+		os.Exit(1)
+	}
 	if err := status.Init(); err != nil {
 		logging.Error(err)
 		os.Exit(1)

pkg/manager/scheduling/baremetal.go

@@ -0,0 +1,143 @@
+package scheduling
+
+import (
+	"fmt"
+	"github.com/devtools-qe-incubator/eventmanager/pkg/util/logging"
+	"golang.org/x/crypto/ssh"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+var (
+	BareMetalMachineSecretMountPath = "/etc/%s"
+)
+
+type BareMetalMachineConfig struct {
+	Username  string `json:"username"`
+	Host      string `json:"host"`
+	PublicKey string `json:"id_rsa"`
+	// TODO: Should we also specify what command to run to find whether machine is free
+	// and expected output?
+	Command        string `json:"command"`
+	ExpectedResult string `json:"expectedResult"`
+}
+
+func InspectFreeBareMetalMachines() []string {
+	machinesAvailableForUse := make([]string, 0)
+	machinesAsCommaSeparatedStr := os.Getenv("BAREMETAL_MACHINES_LIST")
+	if machinesAsCommaSeparatedStr != "" {
+		machines := strings.Split(machinesAsCommaSeparatedStr, ",")
+		for _, machine := range machines {
+			logging.Infof("Checking machine %v", machine)
+			if isBareMetalMachineAvailable(machine) {
+				machinesAvailableForUse = append(machinesAvailableForUse, machine)
+			}
+		}
+	}
+
+	return machinesAvailableForUse
+}
+
+func isBareMetalMachineAvailable(machine string) bool {
+	config, err := getBareMetalMachineConfigFromMountedSecret(machine)
+	if err != nil {
+		logging.Errorf("Error getting bare metal machine config from mounted secret for %s: %v", machine, err)
+		return false
+	}
+
+	return createSshConnectionAndExecuteCommand(config)
+}
+
+func getBareMetalMachineConfigFromMountedSecret(machine string) (*BareMetalMachineConfig, error) {
+	machineSecretDir := fmt.Sprintf(BareMetalMachineSecretMountPath, machine)
+	if _, err := os.Stat(machineSecretDir); os.IsNotExist(err) {
+		return nil, fmt.Errorf("baremetal machine secret directory does not exist: %s", machineSecretDir)
+	} else if err != nil {
+		return nil, fmt.Errorf("error checking baremetal machine secret directory: %w", err)
+	}
+	logging.Info("Found baremetal machine secret directory")
+	config := &BareMetalMachineConfig{}
+
+	dir, err := os.Open(machineSecretDir)
+	if err != nil {
+		return nil, fmt.Errorf("error opening secret directory: %w", err)
+	}
+	defer dir.Close()
+
+	fileNames, err := dir.Readdirnames(-1) // -1 reads all entries
+	if err != nil {
+		return nil, fmt.Errorf("error reading file names from baremetal machine secret directory: %w", err)
+	}
+
+	for _, filename := range fileNames {
+		filePath := filepath.Join(machineSecretDir, filename)
+		info, _ := os.Stat(filePath)
+		if !info.IsDir() {
+			contentBytes, err := os.ReadFile(filePath)
+			if err != nil {
+				return nil, fmt.Errorf("error reading file %s: %w", filePath, err)
+			}
+			content := strings.TrimSuffix(string(contentBytes), "\n")
+
+			switch filename {
+			case "user":
+				config.Username = content
+			case "host":
+				config.Host = content
+			case "key":
+				config.PublicKey = content
+				// TODO: Shall we also specify what command to run to find a specific instance is available or not?
+			case "command":
+				config.Command = content
+				// TODO: If we specify what command to run, we should also specify expected output to consider a machine available.
+			case "expectedResult":
+				config.ExpectedResult = content
+			}
+		}
+	}
+
+	logging.Info("Baremetal machine secret directory found, created obj ", config)
+	return config, nil
+}
+
+func createSshConnectionAndExecuteCommand(config *BareMetalMachineConfig) bool {
+	signer, err := ssh.ParsePrivateKey([]byte(config.PublicKey))
+	if err != nil {
+		logging.Info("Failed to parse private key:", err)
+		return false
+	}
+
+	connectionConfig := &ssh.ClientConfig{
+		User: config.Username,
+		Auth: []ssh.AuthMethod{
+			ssh.PublicKeys(signer),
+		},
+		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
+		Timeout:         10 * time.Second,
+	}
+
+	conn, err := ssh.Dial("tcp", fmt.Sprintf("%s:22", config.Host), connectionConfig)
+	if err != nil {
+		logging.Info("Failed to connect to server:", err)
+		return false
+	}
+	defer conn.Close()
+
+	session, err := conn.NewSession()
+	if err != nil {
+		logging.Info("Failed to create SSH session:", err)
+		return false
+	}
+	defer session.Close()
+
+	output, err := session.CombinedOutput(config.Command)
+	if err != nil {
+		logging.Info("Failed to execute command:", err)
+		return false
+	}
+
+	logging.Info("Command Output:\n", string(output))
+	return config.ExpectedResult == string(output)
+}

pkg/manager/scheduling/pipelines/pipelinerun_manager.go

@@ -0,0 +1,64 @@
+package pipelines
+
+import (
+	"github.com/devtools-qe-incubator/eventmanager/pkg/manager/scheduling"
+	tektonClient "github.com/devtools-qe-incubator/eventmanager/pkg/services/cicd/tekton"
+	"github.com/devtools-qe-incubator/eventmanager/pkg/util/logging"
+	"golang.org/x/exp/slices"
+	"os"
+	"strconv"
+	"time"
+)
+
+const (
+	DefaultPipelineRunSyncInterval = 5 * time.Second
+)
+
+func ManagePipelineRuns(stopChan chan bool) error {
+	ticker := time.NewTicker(readPipelineRunSyncInterval())
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			availableMachines := scheduling.InspectFreeBareMetalMachines()
+			err := listAndSchedulePipelineRuns(availableMachines)
+			if err != nil {
+				logging.Errorf("problem in listing PipelineRuns: %s", err)
+			}
+			logging.Infof("finished listing PipelineRuns")
+		case <-stopChan:
+			logging.Info("Received termination signal, shutting down...")
+		}
+	}
+}
+
+func readPipelineRunSyncInterval() time.Duration {
+	intervalStr := os.Getenv("PIPELINE_RUN_SYNC_INTERVAL")
+	if intervalStr == "" {
+		logging.Infof("PIPELINE_RUN_SYNC_INTERVAL not set, using default interval of 30 seconds")
+		return DefaultPipelineRunSyncInterval
+	}
+
+	interval, err := strconv.Atoi(intervalStr)
+	if err != nil || interval <= 0 {
+		logging.Infof("Invalid PIPELINE_RUN_SYNC_INTERVAL value, using default interval of 30 seconds")
+		return DefaultPipelineRunSyncInterval
+	}
+
+	return time.Duration(interval) * time.Second
+}
+
+func listAndSchedulePipelineRuns(availableMachines []string) error {
+	pendingPipelineRuns, err := tektonClient.ListPendingPipelineRuns()
+	if err != nil {
+		return err
+	}
+	for _, pendingPipelineRun := range pendingPipelineRuns {
+		targetMachine := pendingPipelineRun.GetLabels()["targetBareMetalMachine"]
+		if slices.Contains(availableMachines, targetMachine) {
+			tektonClient.UpdatePipelineRunStatus(pendingPipelineRun)
+		}
+	}
+	return nil
+}

pkg/services/cicd/tekton/tekton.go

@@ -63,6 +63,21 @@ func GetPipelinerunDashboardUrl(pipelinerunName string) string {
 		_client.consoleURL, _client.namespace, pipelinerunName)
 }
 
+func ListPendingPipelineRuns() ([]v1beta1.PipelineRun, error) {
+	logging.Info("Listing pending pipelineruns")
+	pipelineRunList, err := _client.clientset.TektonV1beta1().PipelineRuns(_client.namespace).List(context.TODO(), v1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	pendingPipelineRuns := make([]v1beta1.PipelineRun, 0)
+	for _, pipelineRun := range pipelineRunList.Items {
+		if pipelineRun.Spec.Status == "PipelineRunPending" {
+			pendingPipelineRuns = append(pendingPipelineRuns, pipelineRun)
+		}
+	}
+	return pendingPipelineRuns, nil
+}
+
 func ApplyPipelinerun(spec *v1beta1.PipelineRun) (*v1beta1.PipelineRun, error) {
 	if err := checkInitialization(); err != nil {
 		return nil, err
@@ -155,3 +170,12 @@ func fillWorkspaceBinding(workspacesInfo []WorkspaceBinding) {
 		_client.workspaces = append(_client.workspaces, workspace)
 	}
 }
+
+func UpdatePipelineRunStatus(pendingPipelineRun v1beta1.PipelineRun) {
+	pendingPipelineRun.Spec.Status = ""
+	updatedPipelineRun, err := _client.clientset.TektonV1beta1().PipelineRuns(_client.namespace).Update(context.Background(), &pendingPipelineRun, v1.UpdateOptions{})
+	if err != nil {
+		logging.Errorf("error updating pipelinerun status: %v", err)
+	}
+	logging.Infof("updated pipelinerun status: %v", updatedPipelineRun)
+}