Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: add transactions for numbers #5729

Draft
wants to merge 4 commits into
base: develop
Choose a base branch
from
Draft

fix: add transactions for numbers #5729

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
00b6f5c
fix: add transactions for numbers
vsukhin Aug 5, 2024
809d898
fix: integration test
vsukhin Aug 5, 2024
aeb7ce6
fix: integration test
vsukhin Aug 5, 2024
922469d
fix: nil check
vsukhin Aug 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions pkg/api/v1/testkube/model_test_workflow_extended.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,9 @@ func (w *TestWorkflow) ConvertDots(fn func(string) string) *TestWorkflow {
if w.Labels == nil {
w.Labels = convertDotsInMap(w.Labels, fn)
}
if w.Spec == nil {
return w
}
if w.Spec.Pod != nil {
w.Spec.Pod.Labels = convertDotsInMap(w.Spec.Pod.Labels, fn)
w.Spec.Pod.Annotations = convertDotsInMap(w.Spec.Pod.Annotations, fn)
Expand Down
4 changes: 2 additions & 2 deletions pkg/cloud/data/testworkflow/execution.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,8 @@ func (r *CloudRepository) GetExecutionsSummary(ctx context.Context, filter testw
return pass(r.executor, ctx, req, process)
}

func (r *CloudRepository) Insert(ctx context.Context, result testkube.TestWorkflowExecution) (err error) {
req := ExecutionInsertRequest{WorkflowExecution: result}
func (r *CloudRepository) Insert(ctx context.Context, result *testkube.TestWorkflowExecution) (err error) {
req := ExecutionInsertRequest{WorkflowExecution: *result}
return passNoContent(r.executor, ctx, req)
}

Expand Down
2 changes: 1 addition & 1 deletion pkg/repository/testworkflow/interface.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ type Repository interface {
// GetPreviousFinishedState gets previous finished execution state by test
GetPreviousFinishedState(ctx context.Context, testName string, date time.Time) (testkube.TestWorkflowStatus, error)
// Insert inserts new execution result
Insert(ctx context.Context, result testkube.TestWorkflowExecution) error
Insert(ctx context.Context, result *testkube.TestWorkflowExecution) error
// Update updates execution
Update(ctx context.Context, result testkube.TestWorkflowExecution) error
// UpdateResult updates execution result
Expand Down
2 changes: 1 addition & 1 deletion pkg/repository/testworkflow/mock_repository.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

35 changes: 33 additions & 2 deletions pkg/repository/testworkflow/mongo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"go.mongodb.org/mongo-driver/bson/primitive"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/options"
"go.mongodb.org/mongo-driver/mongo/writeconcern"

"github.com/kubeshop/testkube/pkg/api/v1/testkube"
"github.com/kubeshop/testkube/pkg/repository/sequence"
Expand Down Expand Up @@ -268,12 +269,42 @@ func (r *MongoRepository) GetExecutionsSummary(ctx context.Context, filter Filte
return
}

func (r *MongoRepository) Insert(ctx context.Context, result testkube.TestWorkflowExecution) (err error) {
func (r *MongoRepository) Insert(ctx context.Context, result *testkube.TestWorkflowExecution) (err error) {
result.EscapeDots()
if result.Reports == nil {
result.Reports = []testkube.TestWorkflowReport{}
}
_, err = r.Coll.InsertOne(ctx, result)

wc := writeconcern.New(writeconcern.WMajority())
txnOptions := options.Transaction().SetWriteConcern(wc)
session, err := r.db.Client().StartSession()
if err != nil {
return err
}
defer session.EndSession(ctx)

_, err = session.WithTransaction(ctx, func(sessCtx mongo.SessionContext) (interface{}, error) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Either that should be an atomic MongoDB query, or better to go with an external lock (NATS or Mongo?) - mixing the Golang logic with MongoDB queries will lead to broken results anyway; in best case, for concurrent transactions, it will increment sequence correctly, but will try with the duplicated execution name

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, transactions are supported only for replicaset and will not help, external lock with Mongo might work

Copy link
Collaborator Author

@vsukhin vsukhin Aug 5, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

actualy, distributed lock is also not a silver bullet, if connection fails before the lock is removed, it will be necessary to wait till it is expired. so instead of simply rerunning the test and have a missed number, user will need to wait sometime and not sure, it's a better option for him. I would rather discuss it first, if we need it or not

number, err := r.sequenceRepository.GetNextExecutionNumber(sessCtx, result.Workflow.Name, sequence.ExecutionTypeTestWorkflow)
if err != nil {
return nil, err
}

result.Number = number
if result.Name == "" {
result.Name = fmt.Sprintf("%s-%d", result.Workflow.Name, result.Number)
}

// Ensure it is unique name
// TODO: Consider if we shouldn't make name unique across all TestWorkflows
next, _ := r.GetByNameAndTestWorkflow(sessCtx, result.Name, result.Workflow.Name)
if next.Name == result.Name {
return nil, errors.New("execution name already exists")
}

_, err = r.Coll.InsertOne(sessCtx, result)
return nil, err
}, txnOptions)

return
}

Expand Down
12 changes: 8 additions & 4 deletions pkg/repository/testworkflow/mongo_integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import (
"go.mongodb.org/mongo-driver/mongo/options"

"github.com/kubeshop/testkube/pkg/api/v1/testkube"
"github.com/kubeshop/testkube/pkg/repository/sequence"
)

var (
Expand All @@ -32,13 +33,16 @@ func TestNewMongoRepository_UpdateReport_Integration(t *testing.T) {
db.Drop(ctx)
})

repo := NewMongoRepository(db, false)
seq := sequence.NewMongoRepository(db)
repo := NewMongoRepository(db, false, WithMongoRepositorySequence(seq))

execution := testkube.TestWorkflowExecution{
Id: "test-id",
Name: "test-name",
Id: "test-id",
Workflow: &testkube.TestWorkflow{
Name: "test-name",
},
}
if err := repo.Insert(ctx, execution); err != nil {
if err := repo.Insert(ctx, &execution); err != nil {
t.Fatalf("error inserting execution: %v", err)
}

Expand Down
63 changes: 25 additions & 38 deletions pkg/testworkflows/testworkflowexecutor/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -464,69 +464,56 @@ func (e *executor) Execute(ctx context.Context, workflow testworkflowsv1.TestWor
return execution, errors.Wrap(err, "processing error")
}

// Load execution identifier data
number, err := e.repository.GetNextExecutionNumber(context.Background(), workflow.Name)
if err != nil {
log.DefaultLogger.Errorw("failed to retrieve TestWorkflow execution number", "id", id, "error", err)
}

executionName := request.Name
if executionName == "" {
executionName = fmt.Sprintf("%s-%d", workflow.Name, number)
}

testWorkflowExecutionName := request.TestWorkflowExecutionName
// Ensure it is unique name
// TODO: Consider if we shouldn't make name unique across all TestWorkflows
next, _ := e.repository.GetByNameAndTestWorkflow(ctx, executionName, workflow.Name)
if next.Name == executionName {
return execution, errors.Wrap(err, "execution name already exists")
}

// Build machine with actual execution data
executionMachine := expressions.NewMachine().Register("execution", map[string]interface{}{
"id": id,
"name": executionName,
"number": number,
"scheduledAt": now.UTC().Format(constants.RFC3339Millis),
"disableWebhooks": request.DisableWebhooks,
})

// Process the TestWorkflow
bundle, err := e.processor.Bundle(ctx, &workflow, machine, executionMachine)
if err != nil {
return execution, errors.Wrap(err, "processing error")
}

// Build Execution entity
// TODO: Consider storing "config" as well
execution = testkube.TestWorkflowExecution{
Id: id,
Name: executionName,
Name: request.Name,
Namespace: namespace,
Number: number,
ScheduledAt: now,
StatusAt: now,
Signature: stage.MapSignatureListToInternal(bundle.Signature),
Result: &testkube.TestWorkflowResult{
Status: common.Ptr(testkube.QUEUED_TestWorkflowStatus),
PredictedStatus: common.Ptr(testkube.PASSED_TestWorkflowStatus),
Initialization: &testkube.TestWorkflowStepResult{
Status: common.Ptr(testkube.QUEUED_TestWorkflowStepStatus),
},
Steps: stage.MapSignatureListToStepResults(bundle.Signature),
},
Output: []testkube.TestWorkflowOutput{},
Workflow: testworkflowmappers.MapKubeToAPI(initialWorkflow),
ResolvedWorkflow: testworkflowmappers.MapKubeToAPI(resolvedWorkflow),
TestWorkflowExecutionName: testWorkflowExecutionName,
DisableWebhooks: request.DisableWebhooks,
}
err = e.repository.Insert(ctx, execution)
err = e.repository.Insert(ctx, &execution)
if err != nil {
return execution, errors.Wrap(err, "inserting execution to storage")
}

// Build machine with actual execution data
executionMachine := expressions.NewMachine().Register("execution", map[string]interface{}{
"id": id,
"name": execution.Name,
"number": execution.Number,
"scheduledAt": now.UTC().Format(constants.RFC3339Millis),
"disableWebhooks": request.DisableWebhooks,
})

// Process the TestWorkflow
bundle, err := e.processor.Bundle(ctx, &workflow, machine, executionMachine)
if err != nil {
return execution, errors.Wrap(err, "processing error")
}

execution.Signature = stage.MapSignatureListToInternal(bundle.Signature)
execution.Result.Steps = stage.MapSignatureListToStepResults(bundle.Signature)

err = e.repository.Update(ctx, execution)
if err != nil {
return execution, errors.Wrap(err, "updating execution to storage")
}

// Inform about execution start
e.emitter.Notify(testkube.NewEventQueueTestWorkflow(&execution))

Expand Down
Loading