Docker/portainer
Docker/portainer
1
0
Fork 0
mirror of https://github.com/portainer/portainer.git synced 2025-07-24 15:59:41 +02:00
Code Issues Releases Activity

feat(api): revamp scheduling to introduce system schedules (#2433)

Browse source 
* feat(api): revamp scheduling to introduce system schedules

* fix(api): fix linting issues

* fix(api): fix lint issues

* refactor(api): fix lint issues
This commit is contained in:
Anthony Lapenna 2018-11-06 22:49:48 +13:00 • committed by GitHub
parent dbbea0a20f
commit 110fcc46a6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 475 additions and 297 deletions
Download patch file Download diff file Expand all files Collapse all files

94
api/cron/task_endpoint_sync.go → api/cron/job_endpoint_sync.go
View file

@ -9,48 +9,68 @@ import (
"github.com/portainer/portainer"
)
type (
// EndpointSyncTask represents a task used to synchronize endpoints
// based on an external file. It can be scheduled.
EndpointSyncTask struct {
context *EndpointSyncTaskContext
}
// EndpointSyncJobRunner is used to run a EndpointSyncJob
type EndpointSyncJobRunner struct {
job *portainer.EndpointSyncJob
context *EndpointSyncJobContext
}
// EndpointSyncTaskContext represents the context required for the execution
// of an EndpointSyncTask.
EndpointSyncTaskContext struct {
EndpointService portainer.EndpointService
EndpointFilePath string
}
// EndpointSyncJobContext represents the context of execution of a EndpointSyncJob
type EndpointSyncJobContext struct {
endpointService portainer.EndpointService
endpointFilePath string
}
synchronization struct {
endpointsToCreate []*portainer.Endpoint
endpointsToUpdate []*portainer.Endpoint
endpointsToDelete []*portainer.Endpoint
// NewEndpointSyncJobContext returns a new context that can be used to execute a EndpointSyncJob
func NewEndpointSyncJobContext(endpointService portainer.EndpointService, endpointFilePath string) *EndpointSyncJobContext {
return &EndpointSyncJobContext{
endpointService: endpointService,
endpointFilePath: endpointFilePath,
}
}
fileEndpoint struct {
Name string `json:"Name"`
URL string `json:"URL"`
TLS bool `json:"TLS,omitempty"`
TLSSkipVerify bool `json:"TLSSkipVerify,omitempty"`
TLSCACert string `json:"TLSCACert,omitempty"`
TLSCert string `json:"TLSCert,omitempty"`
TLSKey string `json:"TLSKey,omitempty"`
}
)
// NewEndpointSyncTask creates a new EndpointSyncTask using the specified
// context.
func NewEndpointSyncTask(context *EndpointSyncTaskContext) EndpointSyncTask {
return EndpointSyncTask{
// NewEndpointSyncJobRunner returns a new runner that can be scheduled
func NewEndpointSyncJobRunner(job *portainer.EndpointSyncJob, context *EndpointSyncJobContext) *EndpointSyncJobRunner {
return &EndpointSyncJobRunner{
job: job,
context: context,
}
}
type synchronization struct {
endpointsToCreate []*portainer.Endpoint
endpointsToUpdate []*portainer.Endpoint
endpointsToDelete []*portainer.Endpoint
}
type fileEndpoint struct {
Name string `json:"Name"`
URL string `json:"URL"`
TLS bool `json:"TLS,omitempty"`
TLSSkipVerify bool `json:"TLSSkipVerify,omitempty"`
TLSCACert string `json:"TLSCACert,omitempty"`
TLSCert string `json:"TLSCert,omitempty"`
TLSKey string `json:"TLSKey,omitempty"`
}
// GetScheduleID returns the schedule identifier associated to the runner
func (runner *EndpointSyncJobRunner) GetScheduleID() portainer.ScheduleID {
return runner.job.ScheduleID
}
// SetScheduleID sets the schedule identifier associated to the runner
func (runner *EndpointSyncJobRunner) SetScheduleID(ID portainer.ScheduleID) {
runner.job.ScheduleID = ID
}
// GetJobType returns the job type associated to the runner
func (runner *EndpointSyncJobRunner) GetJobType() portainer.JobType {
return portainer.EndpointSyncJobType
}
// Run triggers the execution of the endpoint synchronization process.
func (task EndpointSyncTask) Run() {
data, err := ioutil.ReadFile(task.context.EndpointFilePath)
func (runner *EndpointSyncJobRunner) Run() {
data, err := ioutil.ReadFile(runner.context.endpointFilePath)
if endpointSyncError(err) {
return
}
@ -62,11 +82,11 @@ func (task EndpointSyncTask) Run() {
}
if len(fileEndpoints) == 0 {
log.Println("background task error (endpoint synchronization). External endpoint source is empty")
log.Println("background job error (endpoint synchronization). External endpoint source is empty")
return
}
storedEndpoints, err := task.context.EndpointService.Endpoints()
storedEndpoints, err := runner.context.endpointService.Endpoints()
if endpointSyncError(err) {
return
}
@ -75,7 +95,7 @@ func (task EndpointSyncTask) Run() {
sync := prepareSyncData(storedEndpoints, convertedFileEndpoints)
if sync.requireSync() {
err = task.context.EndpointService.Synchronize(sync.endpointsToCreate, sync.endpointsToUpdate, sync.endpointsToDelete)
err = runner.context.endpointService.Synchronize(sync.endpointsToCreate, sync.endpointsToUpdate, sync.endpointsToDelete)
if endpointSyncError(err) {
return
}
@ -85,7 +105,7 @@ func (task EndpointSyncTask) Run() {
func endpointSyncError(err error) bool {
if err != nil {
log.Printf("background task error (endpoint synchronization). Unable to synchronize endpoints (err=%s)\n", err)
log.Printf("background job error (endpoint synchronization). Unable to synchronize endpoints (err=%s)\n", err)
return true
}
return false

76
api/cron/job_script_execution.go Normal file
View file

@ -0,0 +1,76 @@
package cron
import (
"log"
"github.com/portainer/portainer"
)
// ScriptExecutionJobRunner is used to run a ScriptExecutionJob
type ScriptExecutionJobRunner struct {
job *portainer.ScriptExecutionJob
context *ScriptExecutionJobContext
}
// ScriptExecutionJobContext represents the context of execution of a ScriptExecutionJob
type ScriptExecutionJobContext struct {
jobService portainer.JobService
endpointService portainer.EndpointService
fileService portainer.FileService
}
// NewScriptExecutionJobContext returns a new context that can be used to execute a ScriptExecutionJob
func NewScriptExecutionJobContext(jobService portainer.JobService, endpointService portainer.EndpointService, fileService portainer.FileService) *ScriptExecutionJobContext {
return &ScriptExecutionJobContext{
jobService: jobService,
endpointService: endpointService,
fileService: fileService,
}
}
// NewScriptExecutionJobRunner returns a new runner that can be scheduled
func NewScriptExecutionJobRunner(job *portainer.ScriptExecutionJob, context *ScriptExecutionJobContext) *ScriptExecutionJobRunner {
return &ScriptExecutionJobRunner{
job: job,
context: context,
}
}
// Run triggers the execution of the job.
// It will iterate through all the endpoints specified in the context to
// execute the script associated to the job.
func (runner *ScriptExecutionJobRunner) Run() {
scriptFile, err := runner.context.fileService.GetFileContent(runner.job.ScriptPath)
if err != nil {
log.Printf("scheduled job error (script execution). Unable to retrieve script file (err=%s)\n", err)
return
}
for _, endpointID := range runner.job.Endpoints {
endpoint, err := runner.context.endpointService.Endpoint(endpointID)
if err != nil {
log.Printf("scheduled job error (script execution). Unable to retrieve information about endpoint (id=%d) (err=%s)\n", endpointID, err)
return
}
err = runner.context.jobService.Execute(endpoint, "", runner.job.Image, scriptFile)
if err != nil {
log.Printf("scheduled job error (script execution). Unable to execute scrtip (endpoint=%s) (err=%s)\n", endpoint.Name, err)
}
}
}
// GetScheduleID returns the schedule identifier associated to the runner
func (runner *ScriptExecutionJobRunner) GetScheduleID() portainer.ScheduleID {
return runner.job.ScheduleID
}
// SetScheduleID sets the schedule identifier associated to the runner
func (runner *ScriptExecutionJobRunner) SetScheduleID(ID portainer.ScheduleID) {
runner.job.ScheduleID = ID
}
// GetJobType returns the job type associated to the runner
func (runner *ScriptExecutionJobRunner) GetJobType() portainer.JobType {
return portainer.ScriptExecutionJobType
}

84
api/cron/job_snapshot.go Normal file
View file

@ -0,0 +1,84 @@
package cron
import (
"log"
"github.com/portainer/portainer"
)
// SnapshotJobRunner is used to run a SnapshotJob
type SnapshotJobRunner struct {
job *portainer.SnapshotJob
context *SnapshotJobContext
}
// SnapshotJobContext represents the context of execution of a SnapshotJob
type SnapshotJobContext struct {
endpointService portainer.EndpointService
snapshotter portainer.Snapshotter
}
// NewSnapshotJobContext returns a new context that can be used to execute a SnapshotJob
func NewSnapshotJobContext(endpointService portainer.EndpointService, snapshotter portainer.Snapshotter) *SnapshotJobContext {
return &SnapshotJobContext{
endpointService: endpointService,
snapshotter: snapshotter,
}
}
// NewSnapshotJobRunner returns a new runner that can be scheduled
func NewSnapshotJobRunner(job *portainer.SnapshotJob, context *SnapshotJobContext) *SnapshotJobRunner {
return &SnapshotJobRunner{
job: job,
context: context,
}
}
// GetScheduleID returns the schedule identifier associated to the runner
func (runner *SnapshotJobRunner) GetScheduleID() portainer.ScheduleID {
return runner.job.ScheduleID
}
// SetScheduleID sets the schedule identifier associated to the runner
func (runner *SnapshotJobRunner) SetScheduleID(ID portainer.ScheduleID) {
runner.job.ScheduleID = ID
}
// GetJobType returns the job type associated to the runner
func (runner *SnapshotJobRunner) GetJobType() portainer.JobType {
return portainer.EndpointSyncJobType
}
// Run triggers the execution of the job.
// It will iterate through all the endpoints available in the database to
// create a snapshot of each one of them.
func (runner *SnapshotJobRunner) Run() {
endpoints, err := runner.context.endpointService.Endpoints()
if err != nil {
log.Printf("background job error (endpoint snapshot). Unable to retrieve endpoint list (err=%s)\n", err)
return
}
for _, endpoint := range endpoints {
if endpoint.Type == portainer.AzureEnvironment {
continue
}
snapshot, err := runner.context.snapshotter.CreateSnapshot(&endpoint)
endpoint.Status = portainer.EndpointStatusUp
if err != nil {
log.Printf("background job error (endpoint snapshot). Unable to create snapshot (endpoint=%s, URL=%s) (err=%s)\n", endpoint.Name, endpoint.URL, err)
endpoint.Status = portainer.EndpointStatusDown
}
if snapshot != nil {
endpoint.Snapshots = []portainer.Snapshot{*snapshot}
}
err = runner.context.endpointService.UpdateEndpoint(endpoint.ID, &endpoint)
if err != nil {
log.Printf("background job error (endpoint snapshot). Unable to update endpoint (endpoint=%s, URL=%s) (err=%s)\n", endpoint.Name, endpoint.URL, err)
return
}
}
}

74
api/cron/scheduler.go
View file

@ -5,52 +5,49 @@ import (
"github.com/robfig/cron"
)
// JobScheduler represents a service for managing crons.
// JobScheduler represents a service for managing crons
type JobScheduler struct {
cron *cron.Cron
}
// NewJobScheduler initializes a new service.
// NewJobScheduler initializes a new service
func NewJobScheduler() *JobScheduler {
return &JobScheduler{
cron: cron.New(),
}
}
// UpdateScheduledTask updates a specific scheduled task by re-creating a new cron
// and adding all the existing jobs. It will then re-schedule the new task
// based on the updatedTask parameter.
// CreateSchedule schedules the execution of a job via a runner
func (scheduler *JobScheduler) CreateSchedule(schedule *portainer.Schedule, runner portainer.JobRunner) error {
runner.SetScheduleID(schedule.ID)
return scheduler.cron.AddJob(schedule.CronExpression, runner)
}
// UpdateSchedule updates a specific scheduled job by re-creating a new cron
// and adding all the existing jobs. It will then re-schedule the new job
// via the specified JobRunner parameter.
// NOTE: the cron library do not support updating schedules directly
// hence the work-around.
func (scheduler *JobScheduler) UpdateScheduledTask(scheduleID portainer.ScheduleID, cronExpression string, updatedTask portainer.Task) error {
jobs := scheduler.cron.Entries()
// hence the work-around
func (scheduler *JobScheduler) UpdateSchedule(schedule *portainer.Schedule, runner portainer.JobRunner) error {
cronEntries := scheduler.cron.Entries()
newCron := cron.New()
for _, job := range jobs {
for _, entry := range cronEntries {
switch task := job.Job.(type) {
case ScriptTask:
if task.context.ScheduleID == scheduleID {
err := newCron.AddJob(cronExpression, updatedTask)
if err != nil {
return err
}
if entry.Job.(portainer.JobRunner).GetScheduleID() == schedule.ID {
continue
var jobRunner cron.Job = runner
if entry.Job.(portainer.JobRunner).GetJobType() == portainer.SnapshotJobType {
jobRunner = entry.Job
}
case SnapshotTask:
_, ok := updatedTask.(SnapshotTask)
if ok {
err := newCron.AddJob(cronExpression, job.Job)
if err != nil {
return err
}
continue
err := newCron.AddJob(schedule.CronExpression, jobRunner)
if err != nil {
return err
}
}
newCron.Schedule(job.Schedule, job.Job)
newCron.Schedule(entry.Schedule, entry.Job)
}
scheduler.cron.Stop()
@ -59,25 +56,21 @@ func (scheduler *JobScheduler) UpdateScheduledTask(scheduleID portainer.Schedule
return nil
}
// UnscheduleTask remove a schedule by re-creating a new cron
// RemoveSchedule remove a scheduled job by re-creating a new cron
// and adding all the existing jobs except for the one specified via scheduleID.
// NOTE: the cron library do not support removing schedules directly
// hence the work-around.
func (scheduler *JobScheduler) UnscheduleTask(scheduleID portainer.ScheduleID) {
jobs := scheduler.cron.Entries()
// hence the work-around
func (scheduler *JobScheduler) RemoveSchedule(scheduleID portainer.ScheduleID) {
cronEntries := scheduler.cron.Entries()
newCron := cron.New()
for _, job := range jobs {
for _, entry := range cronEntries {
switch task := job.Job.(type) {
case ScriptTask:
if task.context.ScheduleID == scheduleID {
continue
}
if entry.Job.(portainer.JobRunner).GetScheduleID() == scheduleID {
continue
}
newCron.Schedule(job.Schedule, job.Job)
newCron.Schedule(entry.Schedule, entry.Job)
}
scheduler.cron.Stop()
@ -85,11 +78,6 @@ func (scheduler *JobScheduler) UnscheduleTask(scheduleID portainer.ScheduleID) {
scheduler.cron.Start()
}
// ScheduleTask adds a new task to be scheduled in the cron.
func (scheduler *JobScheduler) ScheduleTask(cronExpression string, task portainer.Task) error {
return scheduler.cron.AddJob(cronExpression, task)
}
// Start starts the scheduled jobs
func (scheduler *JobScheduler) Start() {
if len(scheduler.cron.Entries()) > 0 {

63
api/cron/task_script.go
View file

@ -1,63 +0,0 @@
package cron
import (
"log"
"github.com/portainer/portainer"
)
// ScriptTaskContext represents the context required for the execution
// of a ScriptTask.
type ScriptTaskContext struct {
JobService portainer.JobService
EndpointService portainer.EndpointService
FileService portainer.FileService
ScheduleID portainer.ScheduleID
TargetEndpoints []portainer.EndpointID
}
// ScriptTask represents a task used to execute a script inside a privileged
// container. It can be scheduled.
type ScriptTask struct {
Image string
ScriptPath string
context *ScriptTaskContext
}
// NewScriptTask creates a new ScriptTask using the specified context.
func NewScriptTask(image, scriptPath string, context *ScriptTaskContext) ScriptTask {
return ScriptTask{
Image: image,
ScriptPath: scriptPath,
context: context,
}
}
// SetContext can be used to set/override the task context
func (task ScriptTask) SetContext(context *ScriptTaskContext) {
task.context = context
}
// Run triggers the execution of the task.
// It will iterate through all the endpoints specified in the context to
// execute the script associated to the task.
func (task ScriptTask) Run() {
scriptFile, err := task.context.FileService.GetFileContent(task.ScriptPath)
if err != nil {
log.Printf("scheduled task error (script execution). Unable to retrieve script file (err=%s)\n", err)
return
}
for _, endpointID := range task.context.TargetEndpoints {
endpoint, err := task.context.EndpointService.Endpoint(endpointID)
if err != nil {
log.Printf("scheduled task error (script execution). Unable to retrieve information about endpoint (id=%d) (err=%s)\n", endpointID, err)
return
}
err = task.context.JobService.Execute(endpoint, "", task.Image, scriptFile)
if err != nil {
log.Printf("scheduled task error (script execution). Unable to execute scrtip (endpoint=%s) (err=%s)\n", endpoint.Name, err)
}
}
}

61
api/cron/task_snapshot.go
View file

@ -1,61 +0,0 @@
package cron
import (
"log"
"github.com/portainer/portainer"
)
// SnapshotTaskContext represents the context required for the execution
// of a SnapshotTask.
type SnapshotTaskContext struct {
EndpointService portainer.EndpointService
Snapshotter portainer.Snapshotter
}
// SnapshotTask represents a task used to create endpoint snapshots.
// It can be scheduled.
type SnapshotTask struct {
context *SnapshotTaskContext
}
// NewSnapshotTask creates a new ScriptTask using the specified context.
func NewSnapshotTask(context *SnapshotTaskContext) SnapshotTask {
return SnapshotTask{
context: context,
}
}
// Run triggers the execution of the task.
// It will iterate through all the endpoints available in the database to
// create a snapshot of each one of them.
func (task SnapshotTask) Run() {
endpoints, err := task.context.EndpointService.Endpoints()
if err != nil {
log.Printf("background task error (endpoint snapshot). Unable to retrieve endpoint list (err=%s)\n", err)
return
}
for _, endpoint := range endpoints {
if endpoint.Type == portainer.AzureEnvironment {
continue
}
snapshot, err := task.context.Snapshotter.CreateSnapshot(&endpoint)
endpoint.Status = portainer.EndpointStatusUp
if err != nil {
log.Printf("background task error (endpoint snapshot). Unable to create snapshot (endpoint=%s, URL=%s) (err=%s)\n", endpoint.Name, endpoint.URL, err)
endpoint.Status = portainer.EndpointStatusDown
}
if snapshot != nil {
endpoint.Snapshots = []portainer.Snapshot{*snapshot}
}
err = task.context.EndpointService.UpdateEndpoint(endpoint.ID, &endpoint)
if err != nil {
log.Printf("background task error (endpoint snapshot). Unable to update endpoint (endpoint=%s, URL=%s) (err=%s)\n", endpoint.Name, endpoint.URL, err)
return
}
}
}