Scheduler.java
/*
** Module : Scheduler.java
** Abstract : cron-like scheduler for P2j processes.
**
** Copyright (c) 2014-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --------------------------------------Description---------------------------------------
** 001 CA 20140206 First version.
** 002 CA 20140509 Fixed jobs which launch a Java class - it must implement Runnable.
** 003 CA 20161027 Allow Runnable instances to be configured as jobs, at runtime.
** CA 20170228 Moved Scheduler.initialize to Scheduler.scheduleJobs.
** 004 CA 20191211 When spawning a Java process programmatically, allow the caller to pass
** environment options specific to that call (beside any jvmArgs configured in
** the directory).
** 005 CA 20200116 Javadoc fixes.
** 006 GES 20210415 Changed RECURRENT to RECURRING.
** 007 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 008 GBB 20241010 Adding getSchedulerModeFor to find the mode for an appserver scheduler.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.scheduler;
import java.lang.reflect.*;
import java.text.*;
import java.util.*;
import java.util.logging.*;
import com.goldencode.p2j.util.logging.*;
import org.quartz.*;
import com.goldencode.p2j.directory.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.security.SecurityManager;
import com.goldencode.p2j.util.*;
/**
* Implements a cron-like scheduler, which can run P2J processes or arbitrary Java classes or
* methods.
* <p>
* The scheduler is configured in the {@code scheduler} container, which resides as a child of the
* {@code server/<server-id>} or the {@code server/default} nodes.
* <p>
* The {@code scheduler} node has a children nodes of class {@code job}. For each {@code job} node,
* the node's name represents the job name (which must be unique accross all jobs). Also,
* {@code node-attribute} nodes can be provided, with the following name:
* <ul>
* <li>{@code type}, mandatory, case insensitive: string representation of the job type,
* following the description at the {@link JobType} enum.</li>
* <li>{@code target}, mandatory: the job target, which can be a process name, a java class or
* a java method (see javadoc for {@link Job} class for more details.</li>
* <li>{@code enabled}, mandatory: boolean value flag indicating if this job is enabled.</li>
* <li>{@code mode}, mandatory: string representation of the job mode, following the
* description at the {@link JobMode} enum.</li>
* <li>{@code minute}: the scheduled minute.</li>
* <li>{@code hour}: the scheduled hour.</li>
* <li>{@code day-of-week}: the scheduled day of the week.</li>
* <li>{@code day-of-month}: the scheduled day of the month.</li>
* <li>{@code month}: the scheduled month.</li>
* </ul>
* <p>
* An example of a job starting an appserver at server startup:
* <pre>
* {@code
* <node class="container" name="scheduler">
* <node class="job" name="start_appserver">
* <node-attribute name="type" value="process"/>
* <node-attribute name="target" value="appserver-name"/>
* <node-attribute name="enabled" value="TRUE"/>
* <node-attribute name="mode" value="now"/>
* </node>
* </node>
* }
* </pre>
* The {@link JobType#INSTANCE} can't be configured from the directory.
* <p>
* WARNING: when searching for the scheduler node, if there are both
* {@code server/<server-id>/scheduler} and {@code server/default/scheduler}, only the server's
* scheduler node will be used. Currently, there is no way of merging the default and server's
* scheduler.
* <p>
* When scheduling jobs for execution, each job must run under the server's context. This is not
* possible using the {@link #timer}'s threads. To solve this, the approach is to use the
* {@link #timer} as a proxy which sends the jobs to the {@link JobProcessor}, which runs in its
* own thread, under the server's context. When the {@link JobProcessor} receives a new job,
* it will create a dedicated thread (with the server's context) and use that thread to execute
* the job.
*
* @see Job
* For more details about how jobs are scheduled.
*/
public final class Scheduler
{
/** Logger. */
private static final CentralLogger LOG = CentralLogger.get(Scheduler.class);
/** A map identifying all the available jobs. */
static Map<String, Job> jobs = null;
/** A registry with the enabled jobs, scheduled to be ran. */
static Map<String, JobTask> jobRegistry = new HashMap<>();
/** Job launcher timer. Will execute each job at its scheduled time. */
static Timer timer = new Timer("Job launcher", true);
/**
* This instance will run on a dedicated thread, under the server's context, as job launching
* requires access to the directory.
*/
static final JobProcessor processor = new JobProcessor();
/**
* Do not allow instances of this class.
*/
private Scheduler()
{
// do not allow instances of this class
}
/**
* Returns the string representation of the scheduler mode for the appserver, or <code>null</code> if
* no job is defined, or no job is enabled.
*
* @param appserverProcess
* The appserver process name.
*
* @return See above.
*/
public static String getSchedulerModeFor(String appserverProcess)
{
if (jobs == null || jobs.size() == 0)
{
return null;
}
return jobs.values()
.stream()
.filter(job -> job.getTarget().equals(appserverProcess) && job.isEnabled())
.map(job -> job.getMode().getMode())
.findAny()
.orElse(null);
}
/**
* Check if the specified job is scheduled for execution.
*
* @param jobName
* The job name.
*
* @return See above.
*/
public static boolean isScheduled(String jobName)
{
jobName = jobName.toLowerCase();
synchronized (jobRegistry)
{
return jobRegistry.containsKey(jobName);
}
}
/**
* Get the next execution time for the given job.
*
* @param jobName
* The job name.
*
* @return The next execution time or {@code -1}, if not scheduled.
*/
public static long scheduledExecutionTime(String jobName)
{
jobName = jobName.toLowerCase();
synchronized (jobRegistry)
{
if (jobRegistry.containsKey(jobName))
{
return jobRegistry.get(jobName).scheduledExecutionTime();
}
}
return -1;
}
/**
* Remove the given job from the {@link #jobs global job registry}. Active jobs (which are
* scheduled for execution) can not be removed. Use {@link #cancelJob} to cancel it first.
*
* @param jobName
* The job name.
*
* @return An {@link Job} instance with the removed job details or {@code null} if the job
* is scheduled for execution or it does not exist.
*/
public static Job removeJob(String jobName)
{
jobName = jobName.toLowerCase();
synchronized (jobRegistry)
{
if (jobRegistry.containsKey(jobName))
{
if (LOG.isLoggable(Level.WARNING))
{
String msg = "Can not remove job - there are pending tasks! Cancel the job first.";
LOG.logp(Level.WARNING, "Scheduler", "removeJob", msg);
return null;
}
}
}
synchronized (jobs)
{
return jobs.remove(jobName);
}
}
/**
* Get the {@link Job} with the given name.
*
* @param jobName
* The job name.
*
* @return The {@link Job} instance or {@code null} if it does not exist.
*/
public static Job getJob(String jobName)
{
jobName = jobName.toLowerCase();
synchronized (jobs)
{
return jobs.get(jobName);
}
}
/**
* Add a new {@link Job} to the {@link #jobs global registry}.
*
* @param job
* The job name. Must not already be in use for another job.
*
* @return <code>true</code> if the job was added. <code>false</code> if there is already
* a job with the same name,
*/
public static boolean addJob(Job job)
{
String jobName = job.getName().toLowerCase();
synchronized (jobs)
{
if (jobs.containsKey(jobName))
{
return false;
}
jobs.put(jobName, job);
}
return true;
}
/**
* Cancel the job with the given name.
*
* @param jobName
* The job name.
*
* @return <code>true</code> if the job was scheduled for execution and it was canceled.
*/
public static boolean cancelJob(String jobName)
{
jobName = jobName.toLowerCase();
synchronized (jobRegistry)
{
JobTask task = jobRegistry.get(jobName);
if (task != null)
{
task.cancel();
// remove the reference from the timer
timer.purge();
}
return (task != null);
}
}
/**
* Schedule a job for execution.
* <p>
* First, this will cancel any pending schedules and schedule it again, using the latest
* {@link Job} configuration.
* <p>
* Second, it will create a new {@link JobTask} which will be registered with the
* {@link #timer}. At appropriate time, the {@link #timer} will pick up the {@link JobTask},
* use {@link #getRunner} to create a new {@link Runnable} instance for this job and
* send it to the {@link JobProcessor} thread.
*
* @param jobName
* The job name.
* @param lastExecutionTime
* The last time the job was executed. {@code null} if this is the first time the
* job needs to run.
*
* @return {@code true} if the job was scheduled.
*/
public static boolean scheduleJob(String jobName, Long lastExecutionTime)
{
return scheduleJob(jobName, lastExecutionTime, null);
}
/**
* Schedule a job for execution.
* <p>
* First, this will cancel any pending schedules and schedule it again, using the latest
* {@link Job} configuration.
* <p>
* Second, it will create a new {@link JobTask} which will be registered with the
* {@link #timer}. At appropriate time, the {@link #timer} will pick up the {@link JobTask},
* use {@link #getRunner} to create a new {@link Runnable} instance for this job and
* send it to the {@link JobProcessor} thread.
*
* @param jobName
* The job name.
* @param lastExecutionTime
* The last time the job was executed. {@code null} if this is the first time the
* job needs to run.
* @param env
* Map of environment properties, in the <code>key=value</code> form. For processes,
* these will be passed as JVM args.
*
* @return {@code true} if the job was scheduled.
*/
public static boolean scheduleJob(String jobName, Long lastExecutionTime, Map<?, ?> env)
{
jobName = jobName.toLowerCase();
synchronized (jobs)
{
Job job = getJob(jobName);
if (job == null)
{
if (LOG.isLoggable(Level.WARNING))
{
String msg = String.format("Job '%s' is not registered with the scheduler!",
jobName);
LOG.logp(Level.WARNING, "Scheduler", "scheduleJob", msg);
}
return false;
}
synchronized (jobRegistry)
{
// make sure this is not already scheduled
cancelJob(jobName);
// schedule the job for a new run
long next = getNextScheduleTime(job, lastExecutionTime);
if (next < 0)
{
// can not schedule it
return false;
}
// convert the env to string pairs
Map<String, String> jenv = null;
if (env != null)
{
jenv = new HashMap<>();
for (Object key : env.keySet())
{
Object val = env.get(key);
if (key instanceof BaseDataType)
{
key = ((BaseDataType) key).toStringMessage();
}
else
{
key = key.toString();
}
if (val instanceof BaseDataType)
{
val = ((BaseDataType) val).toStringMessage();
}
else
{
val = val.toString();
}
jenv.put((String) key, (String) val);
}
}
JobTask runner = new JobTask(jobName, jenv);
jobRegistry.put(jobName, runner);
if (LOG.isLoggable(Level.FINE))
{
String msg = String.format("Scheduling job %s to execute at %tc.", jobName, next);
LOG.logp(Level.WARNING, "Scheduler", "scheduleJob", msg);
}
timer.schedule(runner, new Date(next));
}
}
return true;
}
/**
* Build a cron expression following the {@link CronExpression} syntax, from the specified
* job.
*
* @param job
* Jhe job.
*
* @return A cron expression.
*/
public static String getCronExpression(Job job)
{
return getCronExpression(job.getSecond(), job.getMinute(), job.getHour(),
job.getDayOfWeek(), job.getDayOfMonth(), job.getMonth(),
job.getYear());
}
/**
* Build a cron expression following the {@link CronExpression} syntax, from the specified
* fields.
*
* @param second
* The second when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param minute
* The minute when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param hour
* The hour when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param dayOfWeek
* The day of the week when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param dayOfMonth
* The day the month when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param month
* The month when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param year
* The year when the job needs to be executed, following {@link CronExpression}
* syntax.
*
* @return A cron expression.
*/
public static String getCronExpression(String second,
String minute,
String hour,
String dayOfWeek,
String dayOfMonth,
String month,
String year)
{
String[] fields = { second, minute, hour, dayOfMonth, month, dayOfWeek, year};
StringBuilder cronExpr = new StringBuilder();
for (int i = 0; i < fields.length; i++)
{
String field = (fields[i] == null ? "*" : fields[i]);
cronExpr.append(field).append(" ");
}
return cronExpr.toString().trim();
}
/**
* Validate the given configuration and create a new job.
*
* @param jobName
* The name of this job.
* @param type
* The type of this job, one of the {@link JobType} values.
* @param target
* The target of this job.
* @param enabled
* Flag indicating if this job is enabled.
* @param mode
* The execution mode of this job, one of the {@link JobMode} values.
* @param second
* The second when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param minute
* The minute when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param hour
* The hour when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param dayOfWeek
* The day of the week when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param dayOfMonth
* The day the month when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param month
* The month when the job needs to be executed, following {@link CronExpression}
* syntax.
* @param year
* The year when the job needs to be executed, following {@link CronExpression}
* syntax.
*
* @return A new job instance.
*
* @throws JobValidationException
* If the configuration can't be validated. See the {@link Job} class javadoc for
* details about the validations being performed.
*/
public static Job createJob(String jobName,
String type,
String target,
boolean enabled,
String mode,
String second,
String minute,
String hour,
String dayOfWeek,
String dayOfMonth,
String month,
String year)
throws JobValidationException
{
if (type == null)
{
final String msg = "Type for job %s is missing: use 'process' or 'java'.";
throw new JobValidationException(String.format(msg, jobName));
}
JobType jobType = JobType.resolveJobType(type);
if (jobType == JobType.PROCESS)
{
SecurityManager sm = SecurityManager.getInstance();
// only processes can be set to be scheduled
if (!sm.isProcessDefined(target))
{
final String msg = "Scheduler: process '%s' does not exist for job %s!";
throw new JobValidationException(String.format(msg, target, jobName));
}
}
else if (jobType == JobType.CLASS)
{
String msg = null;
try
{
Class<?> cls = Class.forName(target);
int mods = cls.getModifiers();
if (Modifier.isAbstract(mods) || Modifier.isInterface(mods))
{
msg = "Job '%s': class '%s' must be a non-abstract Java class!";
}
else if (!Runnable.class.isAssignableFrom(cls))
{
// must implement runnable
msg = "Job '%s': class '%s' must implement Runnable interface!";
}
}
catch (ClassNotFoundException e)
{
msg = "Job '%s': class '%s' could not be found!";
if (LOG.isLoggable(Level.SEVERE))
{
LOG.logp(Level.SEVERE, "Scheduler", "createJob",
String.format(msg, jobName, target), e);
}
}
if (msg != null)
{
throw new JobValidationException(String.format(msg, jobName, target));
}
}
else if (jobType == JobType.METHOD)
{
if (target.indexOf('.') == -1)
{
String msg = "Job '%s': target '%s' has no method name!";
throw new JobValidationException(String.format(msg, jobName, target));
}
else
{
String msg = null;
String clsName = target.substring(0, target.lastIndexOf('.'));
String mthdName = target.substring(target.lastIndexOf('.') + 1);
try
{
Class<?> cls = Class.forName(target);
int mods = cls.getModifiers();
if (Modifier.isAbstract(mods) ||
Modifier.isInterface(mods) ||
!Modifier.isPublic(mods))
{
msg = "Job '%s': class '%s' must be a non-abstract, public Java class!";
}
else if (!Runnable.class.isAssignableFrom(cls))
{
// must implement runnable
msg = "Job '%s': class '%s' must implement Runnable interface!";
}
else
{
// validate the method
Method mthd = cls.getMethod(mthdName);
mods = mthd.getModifiers();
if (!Modifier.isPublic(mods))
{
msg = "Job '%s' class '%s': method '%s' must be public!";
}
}
}
catch (ClassNotFoundException e)
{
msg = "Job '%s': class '%s' for method '%s' could not be found!";
if (LOG.isLoggable(Level.SEVERE))
{
LOG.logp(Level.SEVERE, "Scheduler", "createJob",
String.format(msg, jobName, clsName, mthdName), e);
}
}
catch (NoSuchMethodException | SecurityException e)
{
msg = "Job '%s' class '%s': method '%s' could not be found!";
if (LOG.isLoggable(Level.SEVERE))
{
LOG.logp(Level.SEVERE, "Scheduler", "createJob",
String.format(msg, jobName, clsName, mthdName), e);
}
}
if (msg != null)
{
throw new JobValidationException(String.format(msg, jobName, target));
}
}
}
else
{
String msg = "Invalid type '%s' for job %s (use 'process', 'java' or 'method').";
throw new JobValidationException(String.format(msg, type, jobName));
}
if (mode == null)
{
String msg = "Mode for job %s is missing: use 'now', 'recurrent' or 'one-time'";
throw new JobValidationException(String.format(msg, jobName));
}
JobMode jobMode = JobMode.resolveJobMode(mode);
if (jobMode == null)
{
String msg = "Invalid mode '%s' for job %s (use 'now', 'recurrent' or 'one-time').";
throw new JobValidationException(String.format(msg, mode, jobName));
}
// validate the scheduled time
if (jobMode.isTimed())
{
String cronExpr = getCronExpression(second, minute, hour,
dayOfWeek, dayOfMonth, month, year);
if (!CronExpression.isValidExpression(cronExpr))
{
final String msg = "The %s cron expression is not valid!";
throw new JobValidationException(String.format(msg, cronExpr));
}
}
Job job = new Job(jobName, jobType, target, enabled, jobMode);
job.setSecond(second);
job.setMinute(minute);
job.setHour(hour);
job.setDayOfWeek(dayOfWeek);
job.setDayOfMonth(dayOfMonth);
job.setMonth(month);
job.setYear(year);
return job;
}
/**
* Must be called only on server startup. It starts new {@link #processor job processor}
* thread using the server's context. Once the server's sockets have been connected, it reads
* the scheduler configuration and schedules the jobs.
*/
public static void scheduleJobs()
{
Scheduler.initialize();
Thread t = new AssociatedThread(processor);
t.setDaemon(true);
t.setName(CentralLogger.generate("Job processor"));
t.start();
}
/**
* Read the scheduler configuration and schedule the jobs.
*/
synchronized static void initialize()
{
if (jobs != null)
{
return;
}
jobs = new HashMap<>();
// access the directory service
DirectoryService ds = DirectoryService.getInstance();
if (!ds.bind())
throw new RuntimeException("directory bind failed");
String schedulerPath = Utils.findDirectoryNodePath(ds, "scheduler", "container", false);
if (schedulerPath == null)
{
if (LOG.isLoggable(Level.WARNING))
{
LOG.logp(Level.WARNING, "Scheduler", "initialize",
"No scheduler configured.");
}
return;
}
// enumerate jobs
String[] jobNodes = ds.enumerateNodes(schedulerPath);
for (String jobName : jobNodes)
{
String jobPath = schedulerPath + "/" + jobName;
String type = ds.getNodeString(jobPath, "type");
String target = ds.getNodeString(jobPath, "target");
Boolean oenabled = ds.getNodeBoolean(jobPath, "enabled");
String mode = ds.getNodeString(jobPath, "mode");
String second = ds.getNodeString(jobPath, "second");
String minute = ds.getNodeString(jobPath, "minute");
String hour = ds.getNodeString(jobPath, "hour");
String dayOfWeek = ds.getNodeString(jobPath, "day-of-Week");
String dayOfMonth = ds.getNodeString(jobPath, "day-of-Month");
String month = ds.getNodeString(jobPath, "month");
String year = ds.getNodeString(jobPath, "year");
Job job = createJob(jobName, type, target, (oenabled != null && oenabled.booleanValue()),
mode, second, minute, hour, dayOfWeek, dayOfMonth, month, year);
if (!addJob(job))
{
throw new RuntimeException(String.format("Job '%s' is already defined!", jobName));
}
}
// last use of the directory, end our session with it
ds.unbind();
}
/**
* Get a runner to execute this job, depending on the job's {@link Job#getType() type}.
*
* @param job
* The job to be executed.
* @param env
* Map of environment properties, in the <code>key=value</code> form. For processes,
* these will be passed as JVM args.
*
* @return A runnable instance for this job.
*/
static Runnable getRunner(Job job, Map<String, String> env)
{
switch (job.getType())
{
case PROCESS:
return new ProcessLaunchTask(job.getTarget(), env);
case CLASS:
return new JavaLaunchTask(job.getTarget(), "run");
case METHOD:
String target = job.getTarget();
int idx = target.lastIndexOf(".");
String cls = target.substring(0, idx);
String method = target.substring(idx + 1);
return new JavaLaunchTask(cls, method);
case INSTANCE:
return job.getInstance();
}
return null;
}
/**
* Get the next job execution time.
*
* @param job
* The job name.
* @param lastExecutionTime
* The last time the job was executed, in millis. If {@code null}, it means it is
* the first execution and the current time will be used.
*
* @return The next execution time, in millis, or -1 if the job's cron expression could not
* be parsed.
*/
private static long getNextScheduleTime(Job job, Long lastExecutionTime)
throws JobValidationException
{
if (lastExecutionTime != null && job.getMode() != JobMode.RECURRING)
{
// only recurrent mode allows more than one time
return -1;
}
if (job.getMode() == JobMode.NOW)
{
// return a past time, so it will be executed immediately
return System.currentTimeMillis() - 1000;
}
if (lastExecutionTime == null)
{
// use current time as offset, when scheduling for the first time
lastExecutionTime = System.currentTimeMillis() + 60 * 1000;
}
// compute the next execution time
String cronExpression = getCronExpression(job);
try
{
CronExpression ce = new CronExpression(cronExpression);
Date next = ce.getNextValidTimeAfter(new Date(lastExecutionTime));
return next.getTime();
}
catch (ParseException e)
{
if (LOG.isLoggable(Level.SEVERE))
{
final String msg = "Cron expression %s could not be parsed!";
LOG.logp(Level.SEVERE, "Scheduler", "getNextScheduleTime",
String.format(msg, cronExpression), e);
}
return -1;
}
}
}