feat: [TKC-2581] add testworkflow / template cloud client (#5882)

cmd/api-server/main.go

@@ -235,8 +235,10 @@ func main() {
 	testsourcesClient := testsourcesclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
 	testExecutionsClient := testexecutionsclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
 	testsuiteExecutionsClient := testsuiteexecutionsclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
-	testWorkflowsClient := testworkflowsclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
-	testWorkflowTemplatesClient := testworkflowsclientv1.NewTestWorkflowTemplatesClient(kubeClient, cfg.TestkubeNamespace)
+	var testWorkflowsClient testworkflowsclientv1.Interface
+	testWorkflowsClient = testworkflowsclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
+	var testWorkflowTemplatesClient testworkflowsclientv1.TestWorkflowTemplatesInterface
+	testWorkflowTemplatesClient = testworkflowsclientv1.NewTestWorkflowTemplatesClient(kubeClient, cfg.TestkubeNamespace)
 	testWorkflowExecutionsClient := testworkflowsclientv1.NewTestWorkflowExecutionsClient(kubeClient, cfg.TestkubeNamespace)
 	templatesClient := templatesclientv1.NewClient(kubeClient, cfg.TestkubeNamespace)
 
@@ -274,6 +276,11 @@ func main() {
 		resultsRepository = cloudresult.NewCloudResultRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
 		testResultsRepository = cloudtestresult.NewCloudRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
 		configRepository = cloudconfig.NewCloudResultRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
+
+		if cfg.WorkflowStorage == "control-plane" {
+			testWorkflowsClient = cloudtestworkflow.NewCloudTestWorkflowRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
+			testWorkflowTemplatesClient = cloudtestworkflow.NewCloudTestWorkflowTemplateRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
+		}
 		// Pro edition only (tcl protected code)
 		testWorkflowResultsRepository = cloudtestworkflow.NewCloudRepository(grpcClient, grpcConn, cfg.TestkubeProAPIKey)
 		var opts []cloudtestworkflow.Option

internal/app/api/v1/server.go

@@ -82,8 +82,8 @@ func NewTestkubeAPI(
 	clientset kubernetes.Interface,
 	testkubeClientset testkubeclientset.Interface,
 	testsourcesClient *testsourcesclientv1.TestSourcesClient,
-	testWorkflowsClient *testworkflowsv1.TestWorkflowsClient,
-	testWorkflowTemplatesClient *testworkflowsv1.TestWorkflowTemplatesClient,
+	testWorkflowsClient testworkflowsv1.Interface,
+	testWorkflowTemplatesClient testworkflowsv1.TestWorkflowTemplatesInterface,
 	configMap repoConfig.Repository,
 	clusterId string,
 	eventsEmitter *event.Emitter,

internal/config/config.go

@@ -37,6 +37,7 @@ type Config struct {
 	ScrapperEnabled           bool   `envconfig:"SCRAPPERENABLED" default:"false"`
 	LogsBucket                string `envconfig:"LOGS_BUCKET" default:""`
 	LogsStorage               string `envconfig:"LOGS_STORAGE" default:""`
+	WorkflowStorage           string `envconfig:"WORKFLOW_STORAGE" default:"crd"`
 	// WhitelistedContainers is a list of containers from which logs should be collected.
 	WhitelistedContainers                       []string      `envconfig:"WHITELISTED_CONTAINERS" default:"init,logs,scraper"`
 	NatsEmbedded                                bool          `envconfig:"NATS_EMBEDDED" default:"false"`

pkg/agent/agent.go

@@ -9,14 +9,13 @@ import (
 	"os"
 	"time"
 
-	"google.golang.org/grpc/keepalive"
-
 	"github.com/kubeshop/testkube/pkg/executor/output"
 	"github.com/kubeshop/testkube/pkg/version"
 
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
+	"google.golang.org/grpc/keepalive"
 
 	"github.com/pkg/errors"
 	"github.com/valyala/fasthttp"

pkg/cloud/data/testworkflow/commands.go

@@ -29,6 +29,9 @@ const (
 	CmdTestWorkflowOutputHasLog                 executor.Command = "workflow_output_has_log"
 	CmdTestWorkflowOutputDeleteByTestWorkflow   executor.Command = "workflow_output_delete_by_test_workflow"
 	CmdTestworkflowOutputDeleteForTestWorkflows executor.Command = "workflow_output_delete_for_test_workflows"
+
+	CmdTestWorkflowGet         executor.Command = "workflow_get"
+	CmdTestWorkflowTemplateGet executor.Command = "workflow_template_get"
 )
 
 func command(v interface{}) executor.Command {
@@ -82,6 +85,11 @@ func command(v interface{}) executor.Command {
 		return CmdTestWorkflowOutputDeleteByTestWorkflow
 	case ExecutionDeleteOutputForTestWorkflowsRequest:
 		return CmdTestworkflowOutputDeleteForTestWorkflows
+
+	case TestWorkflowGetRequest:
+		return CmdTestWorkflowGet
+	case TestWorkflowTemplateGetRequest:
+		return CmdTestWorkflowTemplateGet
 	}
 	panic("unknown test workflows Cloud request")
 }

pkg/cloud/data/testworkflow/execution_models.go

@@ -186,3 +186,27 @@ type ExecutionGetExecutionTagsRequest struct {
 type ExecutionGetExecutionTagsResponse struct {
 	Tags map[string][]string `json:"tags"`
 }
+
+type TestWorkflowListRequest struct {
+	Selector string `json:"selector"`
+}
+
+type TestWorkflowListResponse struct {
+	TestWorkflows []testkube.TestWorkflow `json:"testWorkflows"`
+}
+
+type TestWorkflowGetRequest struct {
+	Name string `json:"name"`
+}
+
+type TestWorkflowGetResponse struct {
+	TestWorkflow testkube.TestWorkflow `json:"testWorkflow"`
+}
+
+type TestWorkflowTemplateGetRequest struct {
+	Name string `json:"name"`
+}
+
+type TestWorkflowTemplateGetResponse struct {
+	TestWorkflowTemplate testkube.TestWorkflowTemplate `json:"testWorkflowTemplate"`
+}

pkg/cloud/data/testworkflow/templates.go

@@ -0,0 +1,77 @@
+package testworkflow
+
+import (
+	"context"
+	"encoding/json"
+
+	testworkflowsv1 "github.com/kubeshop/testkube-operator/api/testworkflows/v1"
+	testworkflowsclientv1 "github.com/kubeshop/testkube-operator/pkg/client/testworkflows/v1"
+	"github.com/kubeshop/testkube/pkg/cloud"
+	"github.com/kubeshop/testkube/pkg/cloud/data/executor"
+	testworkflowmappers "github.com/kubeshop/testkube/pkg/mapper/testworkflows"
+
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+)
+
+var _ testworkflowsclientv1.TestWorkflowTemplatesInterface = (*CloudTestWorkflowTemplateRepository)(nil)
+
+type CloudTestWorkflowTemplateRepository struct {
+	executor executor.Executor
+}
+
+func NewCloudTestWorkflowTemplateRepository(client cloud.TestKubeCloudAPIClient, grpcConn *grpc.ClientConn, apiKey string) *CloudTestWorkflowTemplateRepository {
+	return &CloudTestWorkflowTemplateRepository{executor: executor.NewCloudGRPCExecutor(client, grpcConn, apiKey)}
+}
+
+func (r *CloudTestWorkflowTemplateRepository) List(selector string) (*testworkflowsv1.TestWorkflowTemplateList, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) ListLabels() (map[string][]string, error) {
+	return make(map[string][]string), nil
+}
+
+func (r *CloudTestWorkflowTemplateRepository) Get(name string) (*testworkflowsv1.TestWorkflowTemplate, error) {
+	req := TestWorkflowTemplateGetRequest{Name: name}
+	response, err := r.executor.Execute(context.Background(), CmdTestWorkflowTemplateGet, req)
+	if err != nil {
+		return nil, err
+	}
+	var commandResponse TestWorkflowTemplateGetResponse
+	if err := json.Unmarshal(response, &commandResponse); err != nil {
+		return nil, err
+	}
+	return testworkflowmappers.MapTemplateAPIToKube(&commandResponse.TestWorkflowTemplate), nil
+}
+
+// Create creates new TestWorkflow
+func (r *CloudTestWorkflowTemplateRepository) Create(workflow *testworkflowsv1.TestWorkflowTemplate) (*testworkflowsv1.TestWorkflowTemplate, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) Update(workflow *testworkflowsv1.TestWorkflowTemplate) (*testworkflowsv1.TestWorkflowTemplate, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) Apply(workflow *testworkflowsv1.TestWorkflowTemplate) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) Delete(name string) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) DeleteAll() error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) DeleteByLabels(selector string) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowTemplateRepository) UpdateStatus(workflow *testworkflowsv1.TestWorkflowTemplate) error {
+	// This is the actual implementation, as update status
+	// should update k8s crd's status field, but we don't have it when stored in mongo
+	return nil
+}

pkg/cloud/data/testworkflow/workflows.go

@@ -0,0 +1,77 @@
+package testworkflow
+
+import (
+	"context"
+	"encoding/json"
+
+	testworkflowsv1 "github.com/kubeshop/testkube-operator/api/testworkflows/v1"
+	testworkflowsclientv1 "github.com/kubeshop/testkube-operator/pkg/client/testworkflows/v1"
+	"github.com/kubeshop/testkube/pkg/cloud"
+	"github.com/kubeshop/testkube/pkg/cloud/data/executor"
+	testworkflowmappers "github.com/kubeshop/testkube/pkg/mapper/testworkflows"
+
+	"github.com/pkg/errors"
+	"google.golang.org/grpc"
+)
+
+var _ testworkflowsclientv1.Interface = (*CloudTestWorkflowRepository)(nil)
+
+type CloudTestWorkflowRepository struct {
+	executor executor.Executor
+}
+
+func NewCloudTestWorkflowRepository(client cloud.TestKubeCloudAPIClient, grpcConn *grpc.ClientConn, apiKey string) *CloudTestWorkflowRepository {
+	return &CloudTestWorkflowRepository{executor: executor.NewCloudGRPCExecutor(client, grpcConn, apiKey)}
+}
+
+func (r *CloudTestWorkflowRepository) List(selector string) (*testworkflowsv1.TestWorkflowList, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) ListLabels() (map[string][]string, error) {
+	return make(map[string][]string), errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) Get(name string) (*testworkflowsv1.TestWorkflow, error) {
+	req := TestWorkflowGetRequest{Name: name}
+	response, err := r.executor.Execute(context.Background(), CmdTestWorkflowGet, req)
+	if err != nil {
+		return nil, err
+	}
+	var commandResponse TestWorkflowGetResponse
+	if err := json.Unmarshal(response, &commandResponse); err != nil {
+		return nil, err
+	}
+	return testworkflowmappers.MapAPIToKube(&commandResponse.TestWorkflow), nil
+}
+
+// Create creates new TestWorkflow
+func (r *CloudTestWorkflowRepository) Create(workflow *testworkflowsv1.TestWorkflow) (*testworkflowsv1.TestWorkflow, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) Update(workflow *testworkflowsv1.TestWorkflow) (*testworkflowsv1.TestWorkflow, error) {
+	return nil, errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) Apply(workflow *testworkflowsv1.TestWorkflow) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) Delete(name string) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) DeleteAll() error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) DeleteByLabels(selector string) error {
+	return errors.New("unimplemented")
+}
+
+func (r *CloudTestWorkflowRepository) UpdateStatus(workflow *testworkflowsv1.TestWorkflow) error {
+	// This is the actual implementation, as update status
+	// should update k8s crd's status field, but we don't have it when stored in mongo
+	return nil
+}