feature. add cluster_byoh batch job.

cmd/server/cluster_byoh.go

@@ -0,0 +1,127 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+
+	_apiClient "github.com/openinfradev/tks-api/pkg/api-client"
+	"github.com/openinfradev/tks-api/pkg/domain"
+	"github.com/openinfradev/tks-api/pkg/log"
+	"github.com/spf13/viper"
+)
+
+var token string
+
+func processClusterByoh() error {
+	// get clusters
+	clusters, err := clusterAccessor.GetBootstrappedByohClusters()
+	if err != nil {
+		return err
+	}
+	if len(clusters) == 0 {
+		return nil
+	}
+	log.Info("byoh clusters : ", clusters)
+
+	token = getTksApiToken()
+	for _, cluster := range clusters {
+		clusterId := cluster.ID
+
+		// check agent node
+		apiClient, err := _apiClient.New(fmt.Sprintf("%s:%d", viper.GetString("tks-api-address"), viper.GetInt("tks-api-port")), token)
+		if err != nil {
+			log.Error(err)
+			continue
+		}
+
+		url := fmt.Sprintf("clusters/%s/nodes", clusterId)
+		body, err := apiClient.Get(url)
+		if err != nil {
+			return err
+		}
+
+		var out domain.GetClusterNodesResponse
+		transcode(body, &out)
+
+		completed := true
+		for _, node := range out.Nodes {
+			if node.Status != "COMPLETED" {
+				completed = false
+			}
+		}
+		log.Info(out.Nodes)
+
+		//completed = true // FOR TEST
+		if completed {
+			log.Info(fmt.Sprintf("all agents registered! starting stack creation. clusterId %s", clusterId))
+			if err = clusterAccessor.UpdateClusterStatus(clusterId, domain.ClusterStatus_INSTALLING); err != nil {
+				log.Error("Failed to update cluster status err : ", err)
+				continue
+			}
+
+			apiClient, err := _apiClient.New(fmt.Sprintf("%s:%d", viper.GetString("tks-api-address"), viper.GetInt("tks-api-port")), token)
+			if err != nil {
+				log.Error(err)
+				continue
+			}
+
+			if cluster.IsStack {
+				if _, err = apiClient.Post(fmt.Sprintf("organizations/%s/stacks/%s/install", cluster.OrganizationId, clusterId), nil); err != nil {
+					log.Error(err)
+					continue
+				}
+			} else {
+				if _, err = apiClient.Post("clusters/"+clusterId+"/install", nil); err != nil {
+					log.Error(err)
+					continue
+				}
+			}
+
+		}
+	}
+	return nil
+}
+
+func transcode(in, out interface{}) {
+	buf := new(bytes.Buffer)
+	err := json.NewEncoder(buf).Encode(in)
+	if err != nil {
+		fmt.Println(err)
+	}
+	err = json.NewDecoder(buf).Decode(out)
+	if err != nil {
+		fmt.Println(err)
+	}
+}
+
+func getTksApiToken() string {
+	apiClient, err := _apiClient.New(fmt.Sprintf("%s:%d", viper.GetString("tks-api-address"), viper.GetInt("tks-api-port")), "")
+	if err != nil {
+		log.Error(err)
+		return ""
+	}
+
+	_, err = apiClient.Post("auth/ping", domain.PingTokenRequest{
+		Token:          token,
+		OrganizationId: "master",
+	})
+	if err != nil {
+		body, err := apiClient.Post("auth/login", domain.LoginRequest{
+			AccountId:      viper.GetString("tks-api-account"),
+			Password:       viper.GetString("tks-api-password"),
+			OrganizationId: "master",
+		})
+		if err != nil {
+			return ""
+		}
+
+		var out domain.LoginResponse
+		transcode(body, &out)
+
+		log.Info(out.User.Token)
+		token = out.User.Token
+	}
+
+	return token
+}

cmd/server/cluster_status.go

@@ -62,6 +62,17 @@ func processClusterStatus() error {
 				case "Error":
 					newStatus = domain.ClusterStatus_DELETE_ERROR
 				}
+			} else if status == domain.ClusterStatus_BOOTSTRAPPING {
+				switch workflow.Status.Phase {
+				case "Running":
+					newStatus = domain.ClusterStatus_BOOTSTRAPPING
+				case "Succeeded":
+					newStatus = domain.ClusterStatus_BOOTSTRAPPED
+				case "Failed":
+					newStatus = domain.ClusterStatus_BOOTSTRAP_ERROR
+				case "Error":
+					newStatus = domain.ClusterStatus_BOOTSTRAP_ERROR
+				}
 			}
 			if newStatus == domain.ClusterStatus_PENDING {
 				continue
@@ -72,7 +83,7 @@ func processClusterStatus() error {
 
 		if status != newStatus || statusDesc != newMessage {
 			log.Debug(fmt.Sprintf("update status!! clusterId [%s], newStatus [%s], newMessage [%s]", clusterId, newStatus, newMessage))
-			err := clusterAccessor.UpdateClusterStatus(clusterId, newStatus, newMessage, workflowId)
+			err := clusterAccessor.UpdateClusterStatusWithWorkflow(clusterId, newStatus, newMessage, workflowId)
 			if err != nil {
 				log.Error("Failed to update cluster status err : ", err)
 				continue

cmd/server/main.go

@@ -17,7 +17,7 @@ import (
 	"github.com/openinfradev/tks-batch/internal/organization"
 )
 
-const INTERVAL_SEC = 1
+const INTERVAL_SEC = 5
 
 var (
 	argowfClient         argo.ArgoClient
@@ -31,6 +31,10 @@ func init() {
 	flag.Int("port", 9112, "service port")
 	flag.String("argo-address", "localhost", "server address for argo-workflow-server")
 	flag.Int("argo-port", 2746, "server port for argo-workflow-server")
+	flag.String("tks-api-address", "http://localhost", "server address for tks-api")
+	flag.Int("tks-api-port", 8080, "server port number for tks-api")
+	flag.String("tks-api-account", "", "account name for tks-api")
+	flag.String("tks-api-password", "", "the password for tks-api account")
 
 	flag.String("dbhost", "localhost", "host of postgreSQL")
 	flag.String("dbport", "5432", "port of postgreSQL")
@@ -86,6 +90,10 @@ func main() {
 		if err != nil {
 			log.Error(err)
 		}
+		err = processClusterByoh()
+		if err != nil {
+			log.Error(err)
+		}
 
 		time.Sleep(time.Second * INTERVAL_SEC)
 	}

go.mod

@@ -3,6 +3,7 @@ module github.com/openinfradev/tks-batch
 go 1.18
 
 require (
+	github.com/openinfradev/tks-api v0.0.0-20231017022609-fe95bb67dc60
 	github.com/spf13/pflag v1.0.5
 	github.com/spf13/viper v1.15.0
 	gorm.io/driver/postgres v1.4.5
@@ -28,8 +29,8 @@ require (
 	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
-	github.com/openinfradev/tks-api v0.0.0-20230621070855-6fc105a25e72 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.6 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/sirupsen/logrus v1.9.0 // indirect
 	github.com/spf13/afero v1.9.3 // indirect
 	github.com/spf13/cast v1.5.0 // indirect
@@ -46,4 +47,3 @@ require (
 replace github.com/openinfradev/tks-batch => ./
 
 //replace github.com/openinfradev/tks-api => ../tks-api
-//replace github.com/openinfradev/tks-proto => ./tks-proto

go.sum

@@ -220,10 +220,8 @@ github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyua
 github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
-github.com/openinfradev/tks-api v0.0.0-20230524092045-07d78d318640 h1:g+vESHF0vnB5ObeTbnb/U3Dwx7kDpAuRc2ftxu/o3x4=
-github.com/openinfradev/tks-api v0.0.0-20230524092045-07d78d318640/go.mod h1:FfhP5GE5TpRaHGUgTlMyxSjRsr6szNy+KhzXWBkMQ4g=
-github.com/openinfradev/tks-api v0.0.0-20230621070855-6fc105a25e72 h1:07S05WaRiIXS+i4Avbavr/vAr+97IKC62/J5YcXDH+c=
-github.com/openinfradev/tks-api v0.0.0-20230621070855-6fc105a25e72/go.mod h1:FfhP5GE5TpRaHGUgTlMyxSjRsr6szNy+KhzXWBkMQ4g=
+github.com/openinfradev/tks-api v0.0.0-20231017022609-fe95bb67dc60 h1:aKcP0zqiuQuruwJ3j+Eb06QKkZKvddFuZ9XPMsANpgg=
+github.com/openinfradev/tks-api v0.0.0-20231017022609-fe95bb67dc60/go.mod h1:91WDknrRu9zZZ1rI1OGhj7dqqc4DQJb830y7EU5t85I=
 github.com/pelletier/go-toml/v2 v2.0.6 h1:nrzqCb7j9cDFj2coyLNLaZuJTLjWjlaz6nvTvIwycIU=
 github.com/pelletier/go-toml/v2 v2.0.6/go.mod h1:eumQOmlWiOPt5WriQQqoM5y18pDHwha2N+QD+EUNTek=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

internal/cluster/cluster.go

@@ -11,10 +11,12 @@ import (
 
 // Cluster represents a kubernetes cluster information.
 type Cluster struct {
-	ID         string `gorm:"primarykey"`
-	WorkflowId string
-	Status     domain.ClusterStatus
-	StatusDesc string
+	ID             string `gorm:"primarykey"`
+	OrganizationId string
+	WorkflowId     string
+	Status         domain.ClusterStatus
+	StatusDesc     string
+	IsStack        bool
 }
 
 // Accessor accesses cluster info in DB.
@@ -38,7 +40,7 @@ func (x *ClusterAccessor) GetIncompleteClusters() ([]Cluster, error) {
 	var clusters []Cluster
 
 	res := x.db.
-		Where("status IN ?", []domain.ClusterStatus{domain.ClusterStatus_INSTALLING, domain.ClusterStatus_DELETING}).
+		Where("status IN ?", []domain.ClusterStatus{domain.ClusterStatus_BOOTSTRAPPING, domain.ClusterStatus_INSTALLING, domain.ClusterStatus_DELETING}).
 		Find(&clusters)
 
 	if res.Error != nil {
@@ -48,7 +50,21 @@ func (x *ClusterAccessor) GetIncompleteClusters() ([]Cluster, error) {
 	return clusters, nil
 }
 
-func (x *ClusterAccessor) UpdateClusterStatus(clusterId string, status domain.ClusterStatus, statusDesc string, workflowId string) error {
+func (x *ClusterAccessor) GetBootstrappedByohClusters() ([]Cluster, error) {
+	var clusters []Cluster
+
+	res := x.db.
+		Where("cloud_service = 'BYOH' AND status IN ?", []domain.ClusterStatus{domain.ClusterStatus_BOOTSTRAPPED}).
+		Find(&clusters)
+
+	if res.Error != nil {
+		return nil, res.Error
+	}
+
+	return clusters, nil
+}
+
+func (x *ClusterAccessor) UpdateClusterStatusWithWorkflow(clusterId string, status domain.ClusterStatus, statusDesc string, workflowId string) error {
 	log.Info(fmt.Sprintf("UpdateClusterStatus. clusterId[%s], status[%d], statusDesc[%s], workflowId[%s]", clusterId, status, statusDesc, workflowId))
 	res := x.db.Model(Cluster{}).
 		Where("ID = ?", clusterId).
@@ -59,3 +75,15 @@ func (x *ClusterAccessor) UpdateClusterStatus(clusterId string, status domain.Cl
 	}
 	return nil
 }
+
+func (x *ClusterAccessor) UpdateClusterStatus(clusterId string, status domain.ClusterStatus) error {
+	log.Info(fmt.Sprintf("UpdateClusterStatus. clusterId[%s], status[%d]", clusterId, status))
+	res := x.db.Model(Cluster{}).
+		Where("ID = ?", clusterId).
+		Updates(map[string]interface{}{"Status": status})
+
+	if res.Error != nil || res.RowsAffected == 0 {
+		return fmt.Errorf("nothing updated in cluster with id %s", clusterId)
+	}
+	return nil
+}