feat: [TKC-2581] add testworkflow / template cloud client

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
+}