add terminate functionality

Author: iignatevich

action.terminate.yaml

@@ -0,0 +1,5 @@
+runtime: plugin
+action:
+  title: "Test: Terminate"
+  description: >-
+    Removes key-value pair from keyring

plugin.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"time"
 
 	"github.com/launchrctl/launchr"
 	"github.com/launchrctl/launchr/pkg/action"
@@ -26,6 +27,9 @@ const (
 //go:embed action.yaml
 var actionYaml []byte
 
+//go:embed action.terminate.yaml
+var actionTerminateYaml []byte
+
 func init() {
 	launchr.RegisterPlugin(&Plugin{})
 }
@@ -105,5 +109,31 @@ func (p *Plugin) DiscoverActions(_ context.Context) ([]*action.Action, error) {
 
 		return p.runBackgroundWeb(ctx, webRunFlags, webPidFile)
 	}))
-	return []*action.Action{a}, nil
+
+	terminateA := action.NewFromYAML("test:terminate", actionTerminateYaml)
+	terminateA.SetRuntime(action.NewFnRuntime(func(ctx context.Context, a *action.Action) error {
+		ticker := time.NewTicker(10 * time.Second)
+		defer ticker.Stop()
+
+		for i := 1; i <= 6; i++ {
+			select {
+			case <-ctx.Done():
+				fmt.Println("someFunction: Context cancelled, stopping early")
+				str := []byte("someFunction: Context cancelled, stopping early")
+				_, _ = a.Input().Streams().Out().Write(str)
+				return ctx.Err()
+			case <-ticker.C:
+				fmt.Printf("someFunction: Running... %d0 seconds\n", i)
+				str := []byte(fmt.Sprintf("someFunction: Running... %d0 seconds\n", i))
+				_, err := a.Input().Streams().Out().Write(str)
+				if err != nil {
+					return err
+				}
+			}
+		}
+
+		return nil
+	}))
+
+	return []*action.Action{a, terminateA}, nil
 }

server/api.go

@@ -23,6 +23,7 @@ import (
 
 type launchrServer struct {
 	actionMngr  action.Manager
+	stateMngr   *StateManager
 	cfg         launchr.Config
 	ctx         context.Context
 	baseURL     string
@@ -95,6 +96,31 @@ func (l *launchrServer) GetOneRunningActionByID(w http.ResponseWriter, _ *http.R
 	})
 }
 
+func (l *launchrServer) CancelRunningAction(w http.ResponseWriter, _ *http.Request, id ActionId, runID ActionRunInfoId) {
+	ri, ok := l.actionMngr.RunInfoByID(runID)
+	if !ok {
+		sendError(w, http.StatusNotFound, fmt.Sprintf("action run info with id %q is not found", id))
+		return
+	}
+
+	if ri.Status != "running" {
+		sendError(w, http.StatusNotFound, fmt.Sprintf("action %q is not running", id))
+		return
+	}
+
+	as, ok := l.stateMngr.actionStateByID(runID)
+	if !ok {
+		sendError(w, http.StatusNotFound, fmt.Sprintf("action state info with id %q is not found", id))
+		return
+	}
+
+	// Cancel context
+	as.cancelSwitch()
+
+	w.WriteHeader(http.StatusOK)
+	_ = json.NewEncoder(w).Encode(struct{}{})
+}
+
 func (l *launchrServer) GetRunningActionStreams(w http.ResponseWriter, _ *http.Request, id ActionId, runID ActionRunInfoId, params GetRunningActionStreamsParams) {
 	ri, ok := l.actionMngr.RunInfoByID(runID)
 	if !ok {
@@ -376,7 +402,8 @@ func (l *launchrServer) RunAction(w http.ResponseWriter, r *http.Request, id str
 		return
 	}
 
-	ri, chErr := l.actionMngr.RunBackground(l.ctx, a, runID)
+	state := l.stateMngr.registerState(runID)
+	ri, chErr := l.actionMngr.RunBackground(state.context, a, runID)
 
 	go func() {
 		err := <-chErr
@@ -386,6 +413,8 @@ func (l *launchrServer) RunAction(w http.ResponseWriter, r *http.Request, id str
 			if _, writeErr := streams.Err().Write([]byte(err.Error())); writeErr != nil {
 				launchr.Log().Error("Failed to write error to stream", "error", writeErr)
 			}
+
+			l.stateMngr.removeActionState(runID)
 		}
 	}()
 

server/openapi.gen.go

@@ -149,6 +149,20 @@ paths:
                   $ref: '#/components/schemas/ActionRunStreamData'
         default:
           $ref: '#/components/responses/DefaultError'
+  /actions/{id}/running/{runId}/cancel:
+    post:
+      summary: Cancels running action
+      description: Cancels running action
+      operationId: cancelRunningAction
+      parameters:
+        - $ref: '#/components/parameters/ActionId'
+        - $ref: '#/components/parameters/ActionRunInfoId'
+      responses:
+        '200':
+          description: Successful operation
+          content: {}
+        default:
+          $ref: '#/components/responses/DefaultError'
   /wizard:
     get:
       summary: Lists all wizards

server/openapi.yaml

@@ -94,6 +94,7 @@ func Run(ctx context.Context, app launchr.App, opts *RunOptions) error {
 		apiPrefix:   opts.APIPrefix,
 		customize:   opts.FrontendCustomize,
 		logsDirPath: opts.LogsDirPath,
+		stateMngr:   NewStateManager(),
 	}
 	app.GetService(&store.actionMngr)
 	app.GetService(&store.cfg)

server/server.go

@@ -0,0 +1,56 @@
+package server
+
+import (
+	"context"
+	"sync"
+)
+
+// StateManager is a definition of manager for actions states
+type StateManager struct {
+	// @todo merge state into action manager
+	actionState map[string]*ActionState
+	mx          sync.Mutex
+}
+
+// NewStateManager constructs a new state manager.
+func NewStateManager() *StateManager {
+	return &StateManager{
+		actionState: make(map[string]*ActionState),
+	}
+}
+
+func (m *StateManager) registerState(id string) *ActionState {
+	m.mx.Lock()
+	defer m.mx.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	as := &ActionState{
+		id:           id,
+		context:      ctx,
+		cancelSwitch: cancel,
+	}
+	m.actionState[id] = as
+
+	return as
+}
+
+func (m *StateManager) removeActionState(id string) {
+	m.mx.Lock()
+	defer m.mx.Unlock()
+	delete(m.actionState, id)
+}
+
+func (m *StateManager) actionStateByID(id string) (*ActionState, bool) {
+	m.mx.Lock()
+	defer m.mx.Unlock()
+	ri, ok := m.actionState[id]
+	return ri, ok
+}
+
+// ActionState defines running action state
+type ActionState struct {
+	id           string
+	context      context.Context
+	cancelSwitch context.CancelFunc
+}