Add API endpoints to pause/unpause new job submissions (#1872)

* Initial plan

* Add pause/unpause functionality for new job submissions

Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

* Update OpenAPI spec with pause/unpause endpoints

Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

* Fix test port conflict - use port 8097 instead of 8096

Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

* Fix test port collision - use port 8102 instead of 8097

Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: vcastellm <47026+vcastellm@users.noreply.github.com>

Author: Copilot

dkron/agent.go

@@ -128,6 +128,10 @@ type Agent struct {
 	logger *logrus.Entry
 
 	tracer trace.Tracer
+
+	// pauseNewJobs controls whether new jobs can be created or updated
+	pauseNewJobs bool
+	pauseMu      sync.RWMutex
 }
 
 // ProcessorFactory is a function type that creates a new instance
@@ -965,3 +969,26 @@ func (a *Agent) startReporter() {
 		}
 	}()
 }
+
+// PauseNewJobs pauses new job submissions
+func (a *Agent) PauseNewJobs() {
+	a.pauseMu.Lock()
+	defer a.pauseMu.Unlock()
+	a.pauseNewJobs = true
+	a.logger.Info("agent: New job submissions paused")
+}
+
+// UnpauseNewJobs resumes new job submissions
+func (a *Agent) UnpauseNewJobs() {
+	a.pauseMu.Lock()
+	defer a.pauseMu.Unlock()
+	a.pauseNewJobs = false
+	a.logger.Info("agent: New job submissions resumed")
+}
+
+// IsNewJobsPaused returns whether new job submissions are paused
+func (a *Agent) IsNewJobsPaused() bool {
+	a.pauseMu.RLock()
+	defer a.pauseMu.RUnlock()
+	return a.pauseNewJobs
+}

dkron/api.go

@@ -109,6 +109,10 @@ func (h *HTTPTransport) APIRoutes(r *gin.RouterGroup, middleware ...gin.HandlerF
 
 	v1.GET("/busy", h.busyHandler)
 
+	v1.GET("/pause", h.pauseStatusHandler)
+	v1.POST("/pause", h.pauseHandler)
+	v1.POST("/unpause", h.unpauseHandler)
+
 	v1.POST("/jobs", h.jobCreateOrUpdateHandler)
 	v1.PATCH("/jobs", h.jobCreateOrUpdateHandler)
 	// Place fallback routes last
@@ -217,6 +221,13 @@ func (h *HTTPTransport) jobGetHandler(c *gin.Context) {
 }
 
 func (h *HTTPTransport) jobCreateOrUpdateHandler(c *gin.Context) {
+	// Check if new job submissions are paused
+	if h.agent.IsNewJobsPaused() {
+		c.AbortWithStatus(http.StatusServiceUnavailable)
+		_, _ = c.Writer.WriteString("New job submissions are currently paused")
+		return
+	}
+
 	// Init the Job object with defaults
 	job := Job{
 		Concurrency: ConcurrencyAllow,
@@ -508,3 +519,18 @@ func (h *HTTPTransport) busyHandler(c *gin.Context) {
 	c.Header("X-Total-Count", strconv.Itoa(len(executions)))
 	renderJSON(c, http.StatusOK, executions)
 }
+
+func (h *HTTPTransport) pauseHandler(c *gin.Context) {
+	h.agent.PauseNewJobs()
+	renderJSON(c, http.StatusOK, gin.H{"paused": true})
+}
+
+func (h *HTTPTransport) unpauseHandler(c *gin.Context) {
+	h.agent.UnpauseNewJobs()
+	renderJSON(c, http.StatusOK, gin.H{"paused": false})
+}
+
+func (h *HTTPTransport) pauseStatusHandler(c *gin.Context) {
+	paused := h.agent.IsNewJobsPaused()
+	renderJSON(c, http.StatusOK, gin.H{"paused": paused})
+}

dkron/api_test.go

@@ -540,3 +540,71 @@ func TestAPILeaderEndpointsNoRaftNoPanic(t *testing.T) {
 			"leader endpoint should return valid status code")
 	}
 }
+
+func TestAPIPauseUnpause(t *testing.T) {
+	port := "8102"
+	baseURL := fmt.Sprintf("http://localhost:%s/v1", port)
+	dir, a := setupAPITest(t, port)
+	defer os.RemoveAll(dir)
+	defer a.Stop() // nolint: errcheck
+
+	// Check initial pause status (should be false)
+	resp, err := http.Get(baseURL + "/pause")
+	require.NoError(t, err)
+	body, _ := ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(body), `"paused":false`)
+
+	// Pause new job submissions
+	resp, err = http.Post(baseURL+"/pause", "application/json", nil)
+	require.NoError(t, err)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(body), `"paused":true`)
+
+	// Verify pause status
+	resp, err = http.Get(baseURL + "/pause")
+	require.NoError(t, err)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(body), `"paused":true`)
+
+	// Try to create a job while paused (should fail)
+	jsonStr := []byte(`{
+		"name": "test_job_paused",
+		"schedule": "@every 1m",
+		"executor": "shell",
+		"executor_config": {"command": "date"}
+	}`)
+	resp, err = http.Post(baseURL+"/jobs", "application/json", bytes.NewBuffer(jsonStr))
+	require.NoError(t, err)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
+	assert.Contains(t, string(body), "paused")
+
+	// Unpause new job submissions
+	resp, err = http.Post(baseURL+"/unpause", "application/json", nil)
+	require.NoError(t, err)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(body), `"paused":false`)
+
+	// Verify unpause status
+	resp, err = http.Get(baseURL + "/pause")
+	require.NoError(t, err)
+	body, _ = ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusOK, resp.StatusCode)
+	assert.Contains(t, string(body), `"paused":false`)
+
+	// Try to create a job after unpause (should succeed)
+	resp, err = http.Post(baseURL+"/jobs", "application/json", bytes.NewBuffer(jsonStr))
+	require.NoError(t, err)
+	resp.Body.Close()
+	assert.Equal(t, http.StatusCreated, resp.StatusCode)
+}

website/static/openapi/openapi.yaml

@@ -434,6 +434,61 @@ paths:
                 items:
                   $ref: '#/components/schemas/execution'
 
+  /pause:
+    get:
+      tags:
+        - default
+      description: |
+        Returns the current pause status for new job submissions.
+      operationId: pauseStatus
+      responses:
+        "200":
+          description: Successful response
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  paused:
+                    type: boolean
+                    description: Whether new job submissions are currently paused
+    post:
+      tags:
+        - default
+      description: |
+        Pauses new job submissions. Existing jobs continue to run, but no new jobs can be created or updated.
+      operationId: pause
+      responses:
+        "200":
+          description: Successful response
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  paused:
+                    type: boolean
+                    description: Whether new job submissions are currently paused
+
+  /unpause:
+    post:
+      tags:
+        - default
+      description: |
+        Resumes new job submissions after being paused.
+      operationId: unpause
+      responses:
+        "200":
+          description: Successful response
+          content:
+            application/json:
+              schema:
+                type: object
+                properties:
+                  paused:
+                    type: boolean
+                    description: Whether new job submissions are currently paused
+
   /acl/policies/{name}:
     get:
       tags: