mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 07:41:27 +00:00
Build real cron engine + service-detail Cron UI
The cron backend was a stub: calculateNextRun returned now+1h for any input and executeCronJob slept 2s then wrote "success" without running the command. Now robfig/cron parses and validates schedules (400 on invalid), a 60s scheduler goroutine executes due enabled jobs, and commands run via docker exec sh -c inside the service's container with stdout+stderr captured into cron_executions. Added Client.ExecRun for attached output capture. UI: Cron tab on the service detail page with create/edit/enable-toggle, manual Run, and execution history with logs. Verified e2e against real Postgres + Docker (trigger and scheduler tick both produce real captured output). Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
1 parent
aab785afc1
commit
a1e43524f5
9 files changed
+498
-26
No files matched your search
@@ -2,12 +2,17 @@ package api
|
||||
|
||||
import (
|
||||
"containr/internal/database"
|
||||
"encoding/json"
|
||||
"containr/internal/docker"
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/google/uuid"
|
||||
"github.com/robfig/cron/v3"
|
||||
)
|
||||
|
||||
type CronJob struct {
|
||||
@@ -62,6 +67,7 @@ func handleGetCronJobs(c *gin.Context) {
|
||||
db := c.MustGet("db").(*database.DB)
|
||||
userID := c.MustGet("user_id").(string)
|
||||
projectID := c.Query("project_id")
|
||||
serviceID := c.Query("service_id")
|
||||
|
||||
query := `SELECT cj.id, cj.project_id, cj.service_id, cj.name, cj.schedule, cj.timezone,
|
||||
cj.enabled, cj.last_run_at, cj.next_run_at, cj.last_status, cj.last_output,
|
||||
@@ -72,8 +78,12 @@ func handleGetCronJobs(c *gin.Context) {
|
||||
args := []interface{}{userID}
|
||||
|
||||
if projectID != "" {
|
||||
query += " AND cj.project_id = $2"
|
||||
args = append(args, projectID)
|
||||
query += fmt.Sprintf(" AND cj.project_id = $%d", len(args))
|
||||
}
|
||||
if serviceID != "" {
|
||||
args = append(args, serviceID)
|
||||
query += fmt.Sprintf(" AND cj.service_id = $%d", len(args))
|
||||
}
|
||||
|
||||
query += " ORDER BY cj.created_at DESC"
|
||||
@@ -130,7 +140,11 @@ func handleCreateCronJob(c *gin.Context) {
|
||||
req.Retention = 30
|
||||
}
|
||||
|
||||
nextRun := calculateNextRun(req.Schedule, req.Timezone)
|
||||
nextRun, err := calculateNextRun(req.Schedule, req.Timezone)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("invalid schedule: %v", err)})
|
||||
return
|
||||
}
|
||||
|
||||
job := CronJob{
|
||||
ID: uuid.New().String(),
|
||||
@@ -227,8 +241,13 @@ func handleUpdateCronJob(c *gin.Context) {
|
||||
updates["name"] = req.Name
|
||||
}
|
||||
if req.Schedule != "" {
|
||||
next, err := calculateNextRun(req.Schedule, req.Timezone)
|
||||
if err != nil {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("invalid schedule: %v", err)})
|
||||
return
|
||||
}
|
||||
updates["schedule"] = req.Schedule
|
||||
updates["next_run_at"] = calculateNextRun(req.Schedule, "UTC")
|
||||
updates["next_run_at"] = next
|
||||
}
|
||||
if req.Command != "" {
|
||||
updates["command"] = req.Command
|
||||
@@ -352,11 +371,12 @@ func handleTriggerCronJob(c *gin.Context) {
|
||||
var job CronJob
|
||||
var ownerCheck string
|
||||
err := db.QueryRow(
|
||||
`SELECT cj.command, p.owner_id FROM cron_jobs cj
|
||||
`SELECT cj.service_id, cj.command, cj.schedule, cj.timezone, cj.retention, p.owner_id
|
||||
FROM cron_jobs cj
|
||||
JOIN projects p ON cj.project_id = p.id
|
||||
WHERE cj.id = $1`,
|
||||
jobID,
|
||||
).Scan(&job.Command, &ownerCheck)
|
||||
).Scan(&job.ServiceID, &job.Command, &job.Schedule, &job.Timezone, &job.Retention, &ownerCheck)
|
||||
|
||||
if err != nil || ownerCheck != userID {
|
||||
c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"})
|
||||
@@ -372,7 +392,9 @@ func handleTriggerCronJob(c *gin.Context) {
|
||||
execID, jobID, now, "running",
|
||||
)
|
||||
|
||||
go executeCronJob(jobID, execID, job.Command)
|
||||
dockerClient, _ := c.Get("docker_client")
|
||||
dockerCli, _ := dockerClient.(*docker.Client)
|
||||
go executeCronJob(db, dockerCli, jobID, job.ServiceID, execID, job.Command, job.Schedule, job.Timezone, job.Retention)
|
||||
|
||||
LogAudit(userID, "cron_job", jobID, "trigger", map[string]interface{}{
|
||||
"execution_id": execID,
|
||||
@@ -384,33 +406,140 @@ func handleTriggerCronJob(c *gin.Context) {
|
||||
})
|
||||
}
|
||||
|
||||
func calculateNextRun(schedule, timezone string) *time.Time {
|
||||
now := time.Now()
|
||||
next := now.Add(1 * time.Hour)
|
||||
return &next
|
||||
// cronScheduleParser parses standard 5-field cron expressions plus
|
||||
// descriptors (@daily etc.) and CRON_TZ= prefixes.
|
||||
var cronScheduleParser = cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
|
||||
|
||||
func calculateNextRun(schedule, timezone string) (*time.Time, error) {
|
||||
expr := strings.TrimSpace(schedule)
|
||||
if tz := strings.TrimSpace(timezone); tz != "" && tz != "UTC" {
|
||||
expr = "CRON_TZ=" + tz + " " + expr
|
||||
}
|
||||
parsed, err := cronScheduleParser.Parse(expr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
next := parsed.Next(time.Now())
|
||||
return &next, nil
|
||||
}
|
||||
|
||||
func executeCronJob(jobID, execID, command string) {
|
||||
db := auditDB
|
||||
if db == nil {
|
||||
// StartCronScheduler ticks every minute and executes enabled cron jobs whose
|
||||
// next_run_at is due. Each run is recorded in cron_executions and output is
|
||||
// captured from docker exec inside the service's container.
|
||||
func StartCronScheduler(ctx context.Context, db *database.DB, dockerClient *docker.Client) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(time.Minute)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
runDueCronJobs(db, dockerClient)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func runDueCronJobs(db *database.DB, dockerClient *docker.Client) {
|
||||
rows, err := db.Query(
|
||||
`SELECT id, project_id, service_id, name, schedule, command, timezone, enabled, retention
|
||||
FROM cron_jobs
|
||||
WHERE enabled = TRUE AND next_run_at IS NOT NULL AND next_run_at <= NOW()`)
|
||||
if err != nil {
|
||||
log.Printf("cron scheduler: failed to list due jobs: %v", err)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
time.Sleep(2 * time.Second)
|
||||
type dueJob struct {
|
||||
CronJob
|
||||
Command string
|
||||
}
|
||||
var jobs []dueJob
|
||||
for rows.Next() {
|
||||
var j dueJob
|
||||
if err := rows.Scan(&j.ID, &j.ProjectID, &j.ServiceID, &j.Name, &j.Schedule,
|
||||
&j.Command, &j.Timezone, &j.Enabled, &j.Retention); err != nil {
|
||||
continue
|
||||
}
|
||||
jobs = append(jobs, j)
|
||||
}
|
||||
|
||||
for _, j := range jobs {
|
||||
execID := uuid.New().String()
|
||||
if _, err := db.Exec(
|
||||
`INSERT INTO cron_executions (id, cron_job_id, started_at, status) VALUES ($1, $2, $3, $4)`,
|
||||
execID, j.ID, time.Now(), "running"); err != nil {
|
||||
log.Printf("cron scheduler: failed to record execution for %s: %v", j.ID, err)
|
||||
continue
|
||||
}
|
||||
go executeCronJob(db, dockerClient, j.ID, j.ServiceID, execID, j.Command, j.Schedule, j.Timezone, j.Retention)
|
||||
}
|
||||
}
|
||||
|
||||
// findServiceContainer returns the first running container whose name matches
|
||||
// the deployment naming convention containr-<serviceID>-*.
|
||||
func findServiceContainer(ctx context.Context, dockerClient *docker.Client, serviceID string) (string, error) {
|
||||
containers, err := dockerClient.ListContainers(ctx, false)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
prefix := "containr-" + serviceID + "-"
|
||||
for _, ctr := range containers {
|
||||
for _, name := range ctr.Names {
|
||||
if strings.HasPrefix(strings.TrimPrefix(name, "/"), prefix) {
|
||||
return ctr.ID, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
return "", fmt.Errorf("no running container for service %s", serviceID)
|
||||
}
|
||||
|
||||
func executeCronJob(db *database.DB, dockerClient *docker.Client, jobID, serviceID, execID, command, schedule, timezone string, retention int) {
|
||||
status := "success"
|
||||
var output, errText string
|
||||
|
||||
if dockerClient == nil {
|
||||
status, errText = "failed", "docker unavailable"
|
||||
} else {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
|
||||
defer cancel()
|
||||
containerID, findErr := findServiceContainer(ctx, dockerClient, serviceID)
|
||||
if findErr != nil {
|
||||
status, errText = "failed", findErr.Error()
|
||||
} else {
|
||||
out, exitCode, runErr := dockerClient.ExecRun(ctx, containerID, []string{"sh", "-c", command})
|
||||
output = out
|
||||
if len(output) > 64*1024 {
|
||||
output = output[:64*1024]
|
||||
}
|
||||
if runErr != nil {
|
||||
status, errText = "failed", runErr.Error()
|
||||
} else if exitCode != 0 {
|
||||
status, errText = "failed", fmt.Sprintf("exit code %d", exitCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
now := time.Now()
|
||||
db.Exec(
|
||||
`UPDATE cron_executions SET finished_at = $1, status = $2, output = $3 WHERE id = $4`,
|
||||
now, "success", "Job completed successfully", execID,
|
||||
`UPDATE cron_executions SET finished_at = $1, status = $2, output = $3, error = $4 WHERE id = $5`,
|
||||
now, status, output, errText, execID,
|
||||
)
|
||||
|
||||
next, _ := calculateNextRun(schedule, timezone)
|
||||
db.Exec(
|
||||
`UPDATE cron_jobs SET last_run_at = $1, last_status = $2, next_run_at = $3 WHERE id = $4`,
|
||||
now, "success", time.Now().Add(1*time.Hour), jobID,
|
||||
`UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_output = $3, next_run_at = $4 WHERE id = $5`,
|
||||
now, status, output, next, jobID,
|
||||
)
|
||||
}
|
||||
|
||||
func init() {
|
||||
cronJobsData, _ := json.Marshal([]CronJob{})
|
||||
_ = cronJobsData
|
||||
if retention > 0 {
|
||||
db.Exec(
|
||||
`DELETE FROM cron_executions WHERE cron_job_id = $1 AND id NOT IN (
|
||||
SELECT id FROM cron_executions WHERE cron_job_id = $1
|
||||
ORDER BY started_at DESC LIMIT $2)`,
|
||||
jobID, retention,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -57,6 +57,9 @@ func SetupRoutes(router *gin.Engine, db *database.DB, redis *database.Redis, cfg
|
||||
// Initialize agent handler (sqlc-backed)
|
||||
agentHandler := NewNodeAgentHandler(db)
|
||||
|
||||
// Start the cron scheduler — executes due cron_jobs via docker exec.
|
||||
StartCronScheduler(context.Background(), db, dockerClient)
|
||||
|
||||
// Initialize database handler
|
||||
databaseHandler := NewDatabaseHandler(db.DB, dockerClient)
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package docker
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"io"
|
||||
@@ -15,6 +16,7 @@ import (
|
||||
"github.com/docker/docker/api/types/system"
|
||||
"github.com/docker/docker/api/types/volume"
|
||||
"github.com/docker/docker/client"
|
||||
"github.com/docker/docker/pkg/stdcopy"
|
||||
)
|
||||
|
||||
// Client wraps the Docker client with additional functionality
|
||||
@@ -313,6 +315,36 @@ func (c *Client) ExecInspect(ctx context.Context, execID string) (container.Exec
|
||||
return c.cli.ContainerExecInspect(ctx, execID)
|
||||
}
|
||||
|
||||
// ExecRun runs a command in a container and returns its combined
|
||||
// stdout/stderr output and exit code.
|
||||
func (c *Client) ExecRun(ctx context.Context, containerID string, cmd []string) (string, int, error) {
|
||||
exec, err := c.cli.ContainerExecCreate(ctx, containerID, container.ExecOptions{
|
||||
Cmd: cmd,
|
||||
AttachStdout: true,
|
||||
AttachStderr: true,
|
||||
})
|
||||
if err != nil {
|
||||
return "", -1, err
|
||||
}
|
||||
|
||||
resp, err := c.cli.ContainerExecAttach(ctx, exec.ID, container.ExecAttachOptions{})
|
||||
if err != nil {
|
||||
return "", -1, err
|
||||
}
|
||||
defer resp.Close()
|
||||
|
||||
var buf bytes.Buffer
|
||||
if _, err := stdcopy.StdCopy(&buf, &buf, resp.Reader); err != nil {
|
||||
return "", -1, err
|
||||
}
|
||||
|
||||
inspect, err := c.cli.ContainerExecInspect(ctx, exec.ID)
|
||||
if err != nil {
|
||||
return buf.String(), -1, err
|
||||
}
|
||||
return buf.String(), inspect.ExitCode, nil
|
||||
}
|
||||
|
||||
// GetImageInfo returns information about a Docker image
|
||||
func (c *Client) GetImageInfo(ctx context.Context, imageName string) (*ImageInfo, error) {
|
||||
images, err := c.cli.ImageList(ctx, image.ListOptions{})
|
||||
|
||||
Reference in new issue
Block a user