[Feature]: Aborting jobs on synapse (#203)

* add build abort functionality for synapse

* uncomment auto remove for removing container

* remove not required log msg

* correct comments

* fix comment on processAbortBuild

* fix typo

* apply suggestion on comment from code review

Co-authored-by: Vikrant Kumar Sinha  <96419531+VikrantKS@users.noreply.github.com>

Co-authored-by: Vikrant Kumar Sinha  <96419531+VikrantKS@users.noreply.github.com>

Author: sachinkumargit

pkg/core/runner.go

@@ -64,6 +64,9 @@ type DockerRunner interface {
 	// KillRunningDocker kills  container spawn by synapse
 	KillRunningDocker(ctx context.Context)
 
+	// KillContainerForBuildID kills synapse container which is running for given buildID
+	KillContainerForBuildID(buildID string) error
+
 	CreateVolume(ctx context.Context, r *RunnerOptions) error
 
 	// RemoveOldVolumes removes volumes that are older than X hours

pkg/core/wsproto.go

@@ -18,13 +18,15 @@ const (
 	MsgError         MessageType = "error"
 	MsgResourceStats MessageType = "resourcestats"
 	MsgJobInfo       MessageType = "jobinfo"
+	MsgBuildAbort    MessageType = "build_abort"
 )
 
 // JobInfo types
 const (
 	JobCompleted StatusType = "complete"
 	JobStarted   StatusType = "started"
 	JobFailed    StatusType = "failed"
+	JobAborted   StatusType = "aborted"
 )
 
 // ResourceStats types
@@ -66,3 +68,8 @@ type JobInfo struct {
 	BuildID string     `json:"build_id"`
 	Message string     `json:"message"`
 }
+
+// BuildAbortMsg struct defines message for aborting a build
+type BuildAbortMsg struct {
+	BuildID string `json:"build_id"`
+}

pkg/runner/docker/docker.go

@@ -30,6 +30,7 @@ import (
 
 const (
 	buildCacheExpiry time.Duration = 4 * time.Hour
+	BuildID                        = "build-id"
 )
 
 var gracefulyContainerStopDuration = time.Second * 10
@@ -337,6 +338,19 @@ func (d *docker) KillRunningDocker(ctx context.Context) {
 	}
 }
 
+func (d *docker) KillContainerForBuildID(buildID string) error {
+	for _, r := range d.RunningContainers {
+		if r.Label[BuildID] == buildID {
+			if err := d.Destroy(context.Background(), r); err != nil {
+				d.logger.Errorf("error while destroying container: %v", err)
+				return err
+			}
+			return nil
+		}
+	}
+	return nil
+}
+
 func (d *docker) PullImage(containerImageConfig *core.ContainerImageConfig, r *core.RunnerOptions) error {
 	if containerImageConfig.PullPolicy == config.PullNever && r.PodType == core.NucleusPod {
 		d.logger.Infof("pull policy %s pod type %s, not pulling any image",

pkg/synapse/synapse.go

@@ -28,6 +28,8 @@ const (
 	duplicateConnectionSleepDuration = 15 * time.Second
 )
 
+var buildAbortMap = make(map[string]bool)
+
 type synapse struct {
 	conn                     *websocket.Conn
 	runner                   core.DockerRunner
@@ -46,7 +48,6 @@ func New(
 	logger lumber.Logger,
 	secretsManager core.SecretsManager,
 ) core.SynapseManager {
-
 	return &synapse{
 		runner:                   runner,
 		logger:                   logger,
@@ -215,6 +216,9 @@ func (s *synapse) processMessage(msg []byte, duplicateConnectionChan chan struct
 	case core.MsgTask:
 		s.logger.Debugf("task message received from server")
 		go s.processTask(message)
+	case core.MsgBuildAbort:
+		s.logger.Debugf("abort-build message received from server")
+		go s.processAbortBuild(message)
 	default:
 		s.logger.Errorf("message type not found")
 	}
@@ -232,6 +236,17 @@ func (s *synapse) processErrorMessage(message core.Message, duplicateConnectionC
 	}
 }
 
+// processAbortBuild handles aborting a running build
+func (s *synapse) processAbortBuild(message core.Message) {
+	buildID := string(message.Content)
+	buildAbortMap[buildID] = true
+	s.logger.Debugf("message received to abort build %s", buildID)
+	if err := s.runner.KillContainerForBuildID(buildID); err != nil {
+		s.logger.Errorf("error while terminating container for buildID: %s, error: %v", buildID, err)
+		return
+	}
+}
+
 // processTask handles task type message
 func (s *synapse) processTask(message core.Message) {
 	var runnerOpts core.RunnerOptions
@@ -270,6 +285,9 @@ func (s *synapse) runAndUpdateJobStatus(runnerOpts *core.RunnerOptions) {
 	if status.Done {
 		jobStatus = core.JobCompleted
 	}
+	if buildAbortMap[runnerOpts.Label[BuildID]] {
+		jobStatus = core.JobAborted
+	}
 	jobInfo := CreateJobInfo(jobStatus, runnerOpts, status.Error.Message)
 	s.logger.Infof("Sending update to neuron %+v", jobInfo)
 	resourceStatsMessage := CreateJobUpdateMessage(jobInfo)