Merge branch 'litmuschaos:master' into singleton-client

Author: shovan-mondal

CONTRIBUTING.md

@@ -15,7 +15,7 @@ We welcome contributions of all kinds:
 ## Getting Started
 
 ### Ask Questions or Discuss Ideas
-- Join our community on [Slack](http://slack.litmuschaos.io) and post in **#litmus** for general questions or **#litmus-dev** for technical discussions.
+- Join our community on [Slack](https://slack.litmuschaos.io) and post in **#litmus** for general questions or **#litmus-dev** for technical discussions.
 
 ### Report Issues or Propose Changes
 - Open a new [GitHub issue](https://github.com/litmuschaos/litmus/issues/new) describing your request, bug, or feature idea.  
@@ -27,7 +27,7 @@ We welcome contributions of all kinds:
 - **Manual setup**  
   Follow the [Local Development Guide](https://github.com/litmuschaos/litmus/wiki/ChaosCenter-Development-Guide) if you prefer to set up and configure the environment yourself.
 - Review [Development Best Practices](https://github.com/litmuschaos/litmus/wiki/Development-Best-Practices)  
-- For Go contributors, read [Effective Go](https://golang.org/doc/effective_go.html) and [Go Code Review Comments](https://go.dev/wiki/CodeReviewComments)  
+- For Go contributors, read [Effective Go](https://go.dev/doc/effective_go) and [Go Code Review Comments](https://go.dev/wiki/CodeReviewComments)  
 
 You can contribute fixes and improvements by submitting a Pull Request (PR) on GitHub. Each PR will be reviewed by one or more maintainers and merged once it meets the project’s standards.
 
@@ -64,8 +64,8 @@ For full details, see the [DCO documentation](https://developercertificate.org/)
 
 3. **Develop & Test**  
 - Follow backend and frontend coding guidelines:  
-  - **Backend:** [Go Code Review Comments](https://code.google.com/p/go-wiki/wiki/CodeReviewComments), [Best Practices](https://peter.bourgon.org/go-in-production/#formatting-and-style)  
-  - **Frontend:** [Airbnb React Style Guide](https://airbnb.io/javascript/react/)  
+  - **Backend:** [Go Code Review Comments](https://go.dev/wiki/CodeReviewComments), [Best Practices](https://peter.bourgon.org/go-in-production/#formatting-and-style)  
+  - **Frontend:** [Airbnb React Style Guide](https://javascript.airbnb.tech)  
 
 - If you are making any changes in backend, make sure you have run and tested the code locally, the reviewers might ask for relevant screenshots in the comments.
 - Include relevant tests for new code or bug fixes  
@@ -82,7 +82,7 @@ For full details, see the [DCO documentation](https://developercertificate.org/)
   - Merge occurs when all checks pass and reviews are approved  
   - If your PR is large or high-impact, coordinate with maintainers in **#litmus-dev** Slack
 
-If you are new to Go, consider reading [Effective Go](https://golang.org/doc/effective_go.html) and [Go Code Review Comments](https://github.com/golang/go/wiki/CodeReviewComments) for guidance on writing idiomatic Go code.
+If you are new to Go, consider reading [Effective Go](https://go.dev/doc/effective_go) and [Go Code Review Comments](https://go.dev/wiki/CodeReviewComments) for guidance on writing idiomatic Go code.
 
 ## Pull Request Checklist
 - [ ] Rebase to the current master branch before submitting your pull request.

chaoscenter/event-tracker/controllers/eventtrackerpolicy_controller.go

@@ -20,11 +20,11 @@ import (
 	"context"
 	"encoding/json"
 	"strings"
-	"sync"
 
 	"github.com/litmuschaos/litmus/chaoscenter/event-tracker/pkg/utils"
 	"github.com/sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/util/retry"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -62,50 +62,99 @@ type apiResponse struct {
 func (r *EventTrackerPolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	_ = log.FromContext(ctx)
 
-	var mutex = &sync.Mutex{}
-	mutex.Lock()
+	// Phase 1: atomically claim pending triggers by flipping IsTriggered to "true"
+	// before any side effect. A concurrent reconcile that loses the CAS will re-read
+	// and observe the winner's claim, so each trigger is owned by exactly one loop.
+	var claimed []string
+	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+		claimed = nil
+
+		var etp eventtrackerv1.EventTrackerPolicy
+		if err := r.Client.Get(ctx, req.NamespacedName, &etp); err != nil {
+			if errors.IsNotFound(err) {
+				logrus.Infof("EventTrackerPolicy %s not found", req.NamespacedName)
+				return nil
+			}
+			return err
+		}
 
-	var etp eventtrackerv1.EventTrackerPolicy
-	err := r.Client.Get(context.Background(), req.NamespacedName, &etp)
-	if errors.IsNotFound(err) {
-		logrus.Infof("namespace: %s not found", req.NamespacedName)
-		return ctrl.Result{}, nil
-	} else if err != nil {
+		for i, s := range etp.Statuses {
+			if s.Result == utils.ConditionPassed && strings.ToLower(s.IsTriggered) == "false" {
+				etp.Statuses[i].IsTriggered = "true"
+				claimed = append(claimed, s.ExperimentID)
+				logrus.Printf("claiming trigger resource=%s experimentID=%s", s.ResourceName, s.ExperimentID)
+			}
+		}
+		if len(claimed) == 0 {
+			return nil
+		}
+		return r.Client.Update(ctx, &etp)
+	})
+	if err != nil {
 		return ctrl.Result{}, err
 	}
 
-	for index, status := range etp.Statuses {
-		if status.Result == utils.ConditionPassed && strings.ToLower(status.IsTriggered) == "false" {
-			logrus.Print("ResourceName: " + status.ResourceName + ", ExperimentID: " + status.ExperimentID)
-			response, err := utils.SendRequest(status.ExperimentID)
-			if err != nil {
-				return ctrl.Result{}, err
-			}
-
-			logrus.Print(response)
+	// Phase 2: side-effect dispatch. Executed outside the retry loop so each
+	// claim triggers exactly one SendRequest. On a Gitops-Disabled response we
+	// restore IsTriggered="false" to match the pre-fix behavior (retryable).
+	var toRevert []string
+	var dispatchErr error
+	for _, experimentID := range claimed {
+		response, sendErr := utils.SendRequest(experimentID)
+		if sendErr != nil {
+			logrus.WithError(sendErr).Errorf("failed to trigger experiment %s", experimentID)
+			toRevert = append(toRevert, experimentID)
+			dispatchErr = sendErr
+			continue
+		}
 
-			var res apiResponse
-			err = json.Unmarshal([]byte(response), &res)
-			if err != nil {
-				return ctrl.Result{}, err
-			}
+		var res apiResponse
+		if jsonErr := json.Unmarshal([]byte(response), &res); jsonErr != nil {
+			logrus.WithError(jsonErr).Errorf("failed to parse response for experiment %s", experimentID)
+			toRevert = append(toRevert, experimentID)
+			dispatchErr = jsonErr
+			continue
+		}
 
-			if res.Data.GitopsNotifer == "Gitops Disabled" {
-				etp.Statuses[index].IsTriggered = "false"
-			} else {
-				etp.Statuses[index].IsTriggered = "true"
-			}
+		if res.Data.GitopsNotifer == "Gitops Disabled" {
+			logrus.Infof("gitops disabled for experiment %s; releasing claim", experimentID)
+			toRevert = append(toRevert, experimentID)
+			continue
 		}
+		logrus.Infof("successfully triggered experiment %s", experimentID)
 	}
 
-	err = r.Client.Update(context.Background(), &etp)
-	if err != nil {
-		return ctrl.Result{}, err
+	if len(toRevert) > 0 {
+		revertErr := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
+			var etp eventtrackerv1.EventTrackerPolicy
+			if err := r.Client.Get(ctx, req.NamespacedName, &etp); err != nil {
+				if errors.IsNotFound(err) {
+					return nil
+				}
+				return err
+			}
+			revertSet := make(map[string]struct{}, len(toRevert))
+			for _, id := range toRevert {
+				revertSet[id] = struct{}{}
+			}
+			changed := false
+			for i, s := range etp.Statuses {
+				if _, ok := revertSet[s.ExperimentID]; ok && strings.ToLower(s.IsTriggered) == "true" {
+					etp.Statuses[i].IsTriggered = "false"
+					changed = true
+				}
+			}
+			if !changed {
+				return nil
+			}
+			return r.Client.Update(ctx, &etp)
+		})
+		if revertErr != nil {
+			return ctrl.Result{}, revertErr
+		}
 	}
 
-	defer mutex.Unlock()
-
-	return ctrl.Result{}, nil
+	return ctrl.Result{}, dispatchErr
 }
 
 // SetupWithManager sets up the controller with the Manager.

chaoscenter/graphql/server/Dockerfile

@@ -1,5 +1,5 @@
 # BUILD STAGE
-FROM golang:1.24 AS builder
+FROM golang:1.26 AS builder
 
 LABEL maintainer="LitmusChaos"
 
@@ -17,7 +17,7 @@ RUN CGO_ENABLED=0 go build -o /output/server -v
 
 # DEPLOY STAGE
 # Use Red Hat UBI minimal image as base
-FROM registry.access.redhat.com/ubi9/ubi-minimal:9.6
+FROM registry.access.redhat.com/ubi9/ubi-minimal:9.7
 
 LABEL maintainer="LitmusChaos"
 

chaoscenter/graphql/server/pkg/chaos_experiment/handler/handler.go

@@ -1551,7 +1551,14 @@ func (c *ChaosExperimentHandler) StopExperimentRuns(ctx context.Context, project
 	}
 
 	for _, runID := range experimentRunsID {
-		err = c.chaosExperimentRunService.ProcessExperimentRunStop(ctx, query, &runID, experiment, username, projectID, r)
+		// scope the update to the specific run so we don't accidentally touch sibling runs
+		runQuery := bson.D{
+			{"experiment_id", experimentID},
+			{"project_id", projectID},
+			{"experiment_run_id", runID},
+			{"is_removed", false},
+		}
+		err = c.chaosExperimentRunService.ProcessExperimentRunStop(ctx, runQuery, &runID, experiment, username, projectID, r)
 		if err != nil {
 			return false, err
 		}

chaoscenter/graphql/server/pkg/chaos_experiment_run/service.go

@@ -71,10 +71,15 @@ func (c *chaosExperimentRunService) ProcessExperimentRunDelete(ctx context.Conte
 	return nil
 }
 
-// ProcessExperimentRunStop deletes a workflow entry and updates the database
+// ProcessExperimentRunStop marks an experiment run as stopped in the database and notifies
+// the infra subscriber if it's currently connected. Updating the DB here is important because
+// the infra might be disconnected (e.g. agent redeployed), in which case the stop signal
+// never reaches the agent and the run would be stuck in Running state forever.
 func (c *chaosExperimentRunService) ProcessExperimentRunStop(ctx context.Context, query bson.D, experimentRunID *string, experiment dbChaosExperiment.ChaosExperimentRequest, username string, projectID string, r *store.StateData) error {
 	update := bson.D{
 		{"$set", bson.D{
+			{"phase", string(model.ExperimentRunStatusStopped)},
+			{"completed", true},
 			{"updated_at", time.Now().UnixMilli()},
 			{"updated_by", mongodb.UserDetailResponse{
 				Username: username,
@@ -86,6 +91,8 @@ func (c *chaosExperimentRunService) ProcessExperimentRunStop(ctx context.Context
 	if err != nil {
 		return err
 	}
+	// Best-effort: notify the agent to clean up in-flight resources. If the infra
+	// is not connected the channel send is a no-op and the DB state is still correct.
 	if r != nil {
 		chaos_infrastructure.SendExperimentToSubscriber(projectID, &model.ChaosExperimentRequest{
 			InfraID: experiment.InfraID,

chaoscenter/graphql/server/pkg/chaos_experiment_run/service_test.go

@@ -155,6 +155,97 @@ func Test_chaosExperimentRunService_ProcessExperimentRunDelete(t *testing.T) {
 	}
 }
 
+func Test_chaosExperimentRunService_ProcessExperimentRunStop(t *testing.T) {
+	type args struct {
+		ctx           context.Context
+		query         bson.D
+		experimentRun dbChaosExperimentRun.ChaosExperimentRun
+		workflow      dbChaosExperiment.ChaosExperimentRequest
+		username      string
+		r             *store.StateData
+	}
+	experimentRunID := uuid.New().String()
+	projectID := uuid.New().String()
+	experimentID := uuid.New().String()
+	infraID := uuid.New().String()
+
+	testcases := []struct {
+		name    string
+		args    args
+		given   func()
+		wantErr bool
+	}{
+		{
+			name: "success: stop marks run as Stopped in DB even when infra is disconnected",
+			args: args{
+				ctx: context.Background(),
+				query: bson.D{
+					{Key: "experiment_id", Value: experimentID},
+					{Key: "project_id", Value: projectID},
+					{Key: "experiment_run_id", Value: experimentRunID},
+					{Key: "is_removed", Value: false},
+				},
+				experimentRun: dbChaosExperimentRun.ChaosExperimentRun{
+					ProjectID:       projectID,
+					ExperimentID:    experimentID,
+					ExperimentRunID: experimentRunID,
+					InfraID:         infraID,
+					Phase:           "Running",
+				},
+				workflow: dbChaosExperiment.ChaosExperimentRequest{
+					ProjectID:    projectID,
+					InfraID:      infraID,
+					ExperimentID: experimentID,
+				},
+				username: "test",
+				// nil store means no subscriber channel — simulates disconnected infra
+				r: nil,
+			},
+			given: func() {
+				mongodbMockOperator.On("Update", mock.Anything, mongodb.ChaosExperimentRunsCollection, mock.Anything, mock.Anything, mock.Anything).Return(&mongo.UpdateResult{MatchedCount: 1}, nil).Once()
+			},
+			wantErr: false,
+		},
+		{
+			name: "failure: DB error propagates",
+			args: args{
+				ctx: context.Background(),
+				query: bson.D{
+					{Key: "experiment_id", Value: experimentID},
+					{Key: "project_id", Value: projectID},
+					{Key: "experiment_run_id", Value: experimentRunID},
+					{Key: "is_removed", Value: false},
+				},
+				experimentRun: dbChaosExperimentRun.ChaosExperimentRun{
+					ProjectID:       projectID,
+					ExperimentID:    experimentID,
+					ExperimentRunID: experimentRunID,
+					InfraID:         infraID,
+				},
+				workflow: dbChaosExperiment.ChaosExperimentRequest{
+					ProjectID:    projectID,
+					InfraID:      infraID,
+					ExperimentID: experimentID,
+				},
+				username: "test",
+				r:        nil,
+			},
+			given: func() {
+				mongodbMockOperator.On("Update", mock.Anything, mongodb.ChaosExperimentRunsCollection, mock.Anything, mock.Anything, mock.Anything).Return(&mongo.UpdateResult{MatchedCount: 0}, errors.New("db error")).Once()
+			},
+			wantErr: true,
+		},
+	}
+	for _, tc := range testcases {
+		tc.given()
+		t.Run(tc.name, func(t *testing.T) {
+			if err := chaosExperimentRunTestService.ProcessExperimentRunStop(tc.args.ctx, tc.args.query, &experimentRunID, tc.args.workflow, tc.args.username, projectID, tc.args.r); (err != nil) != tc.wantErr {
+				t.Errorf("chaosExperimentRunService.ProcessExperimentRunStop() error = %v, wantErr %v", err, tc.wantErr)
+			}
+		})
+	}
+}
+
 func Test_chaosExperimentRunService_ProcessCompletedExperimentRun(t *testing.T) {
 	type args struct {
 		execData ExecutionData

chaoscenter/graphql/server/pkg/chaos_infrastructure/service.go

@@ -899,7 +899,9 @@ func (in *infraService) GetVersionDetails() (*model.InfraVersionDetails, error)
 	compatibleVersions := utils.Config.InfraCompatibleVersions
 
 	var compatibleArray []string
-	_ = json.Unmarshal([]byte(compatibleVersions), &compatibleArray)
+	if err := json.Unmarshal([]byte(compatibleVersions), &compatibleArray); err != nil {
+		return &model.InfraVersionDetails{}, fmt.Errorf("failed to parse InfraCompatibleVersions config: %w", err)
+	}
 
 	// To find the latest compatible version
 	compatibleMap := make(map[int]string)