WorkflowServiceImpl.java
/*
* Licensed to The Apereo Foundation under one or more contributor license
* agreements. See the NOTICE file distributed with this work for additional
* information regarding copyright ownership.
*
*
* The Apereo Foundation licenses this file to you under the Educational
* Community License, Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain a copy of the License
* at:
*
* http://opensource.org/licenses/ecl2.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*
*/
package org.opencastproject.workflow.impl;
import static org.opencastproject.security.api.SecurityConstants.GLOBAL_ADMIN_ROLE;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.FAILED;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.FAILING;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.INSTANTIATED;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.PAUSED;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.RUNNING;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.STOPPED;
import static org.opencastproject.workflow.api.WorkflowInstance.WorkflowState.SUCCEEDED;
import org.opencastproject.assetmanager.api.AssetManager;
import org.opencastproject.assetmanager.api.query.RichAResult;
import org.opencastproject.assetmanager.util.WorkflowPropertiesUtil;
import org.opencastproject.elasticsearch.api.SearchIndexException;
import org.opencastproject.elasticsearch.api.SearchResult;
import org.opencastproject.elasticsearch.api.SearchResultItem;
import org.opencastproject.elasticsearch.index.ElasticsearchIndex;
import org.opencastproject.elasticsearch.index.objects.event.Event;
import org.opencastproject.elasticsearch.index.objects.event.EventSearchQuery;
import org.opencastproject.elasticsearch.index.rebuild.AbstractIndexProducer;
import org.opencastproject.elasticsearch.index.rebuild.IndexProducer;
import org.opencastproject.elasticsearch.index.rebuild.IndexRebuildException;
import org.opencastproject.elasticsearch.index.rebuild.IndexRebuildService;
import org.opencastproject.job.api.Job;
import org.opencastproject.job.api.Job.Status;
import org.opencastproject.job.api.JobProducer;
import org.opencastproject.mediapackage.MediaPackage;
import org.opencastproject.mediapackage.MediaPackageElement;
import org.opencastproject.mediapackage.MediaPackageParser;
import org.opencastproject.mediapackage.MediaPackageSupport;
import org.opencastproject.mediapackage.Publication;
import org.opencastproject.metadata.api.MediaPackageMetadata;
import org.opencastproject.metadata.api.MediaPackageMetadataService;
import org.opencastproject.metadata.api.MetadataService;
import org.opencastproject.metadata.api.util.MediaPackageMetadataSupport;
import org.opencastproject.security.api.AccessControlList;
import org.opencastproject.security.api.AccessControlUtil;
import org.opencastproject.security.api.AclScope;
import org.opencastproject.security.api.AuthorizationService;
import org.opencastproject.security.api.Organization;
import org.opencastproject.security.api.OrganizationDirectoryService;
import org.opencastproject.security.api.Permissions;
import org.opencastproject.security.api.SecurityService;
import org.opencastproject.security.api.UnauthorizedException;
import org.opencastproject.security.api.User;
import org.opencastproject.security.api.UserDirectoryService;
import org.opencastproject.series.api.SeriesException;
import org.opencastproject.series.api.SeriesService;
import org.opencastproject.serviceregistry.api.ServiceRegistry;
import org.opencastproject.serviceregistry.api.ServiceRegistryException;
import org.opencastproject.serviceregistry.api.UndispatchableJobException;
import org.opencastproject.util.NotFoundException;
import org.opencastproject.util.ReadinessIndicator;
import org.opencastproject.util.data.Tuple;
import org.opencastproject.workflow.api.ResumableWorkflowOperationHandler;
import org.opencastproject.workflow.api.RetryStrategy;
import org.opencastproject.workflow.api.WorkflowDatabaseException;
import org.opencastproject.workflow.api.WorkflowDefinition;
import org.opencastproject.workflow.api.WorkflowException;
import org.opencastproject.workflow.api.WorkflowIdentifier;
import org.opencastproject.workflow.api.WorkflowIndexData;
import org.opencastproject.workflow.api.WorkflowInstance;
import org.opencastproject.workflow.api.WorkflowInstance.WorkflowState;
import org.opencastproject.workflow.api.WorkflowListener;
import org.opencastproject.workflow.api.WorkflowOperationDefinition;
import org.opencastproject.workflow.api.WorkflowOperationDefinitionImpl;
import org.opencastproject.workflow.api.WorkflowOperationHandler;
import org.opencastproject.workflow.api.WorkflowOperationInstance;
import org.opencastproject.workflow.api.WorkflowOperationInstance.OperationState;
import org.opencastproject.workflow.api.WorkflowOperationResult;
import org.opencastproject.workflow.api.WorkflowOperationResult.Action;
import org.opencastproject.workflow.api.WorkflowOperationResultImpl;
import org.opencastproject.workflow.api.WorkflowParsingException;
import org.opencastproject.workflow.api.WorkflowService;
import org.opencastproject.workflow.api.WorkflowServiceDatabase;
import org.opencastproject.workflow.api.WorkflowStateException;
import org.opencastproject.workflow.api.WorkflowStateMapping;
import org.opencastproject.workflow.api.WorkflowUtil;
import org.opencastproject.workflow.api.XmlWorkflowParser;
import org.opencastproject.workspace.api.Workspace;
import com.google.common.util.concurrent.Striped;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.osgi.framework.InvalidSyntaxException;
import org.osgi.framework.ServiceReference;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.Callable;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* Implements WorkflowService with in-memory data structures to hold WorkflowOperations and WorkflowInstances.
* WorkflowOperationHandlers are looked up in the OSGi service registry based on the "workflow.operation" property. If
* the WorkflowOperationHandler's "workflow.operation" service registration property matches
* WorkflowOperation.getName(), then the factory returns a WorkflowOperationRunner to handle that operation. This allows
* for custom runners to be added or modified without affecting the workflow service itself.
*/
@Component(
property = {
"service.description=Workflow Service",
"service.pid=org.opencastproject.workflow.impl.WorkflowServiceImpl"
},
immediate = true,
service = { WorkflowService.class, WorkflowServiceImpl.class, IndexProducer.class }
)
public class WorkflowServiceImpl extends AbstractIndexProducer implements WorkflowService, JobProducer {
/** Retry strategy property name */
private static final String RETRY_STRATEGY = "retryStrategy";
/** Logging facility */
private static final Logger logger = LoggerFactory.getLogger(WorkflowServiceImpl.class);
/** List of available operations on jobs */
enum Operation {
START_WORKFLOW, RESUME, START_OPERATION
}
/** Constant value indicating a <code>null</code> parent id */
private static final String NULL_PARENT_ID = "-";
/** The load imposed on the system by a workflow job.
* We are keeping this hardcoded because otherwise bad things will likely happen,
* like an inability to process a workflow past a certain point in high-load conditions
*/
private static final float WORKFLOW_JOB_LOAD = 0.0f;
/** Error resolution handler id constant */
public static final String ERROR_RESOLUTION_HANDLER_ID = "error-resolution";
/** Remove references to the component context once felix scr 1.2 becomes available */
protected ComponentContext componentContext = null;
/** The metadata services */
private SortedSet<MediaPackageMetadataService> metadataServices;
/** Persistent storage */
protected WorkflowServiceDatabase persistence;
/** The list of workflow listeners */
private final List<WorkflowListener> listeners = new CopyOnWriteArrayList<WorkflowListener>();
/** The thread pool to use for firing listeners and handling dispatched jobs */
protected ThreadPoolExecutor executorService;
/** The workspace */
protected Workspace workspace = null;
/** The service registry */
protected ServiceRegistry serviceRegistry = null;
/** The security service */
protected SecurityService securityService = null;
/** The authorization service */
protected AuthorizationService authorizationService = null;
/** The user directory service */
protected UserDirectoryService userDirectoryService = null;
/** The organization directory service */
protected OrganizationDirectoryService organizationDirectoryService = null;
/** The series service */
protected SeriesService seriesService;
/** The asset manager */
protected AssetManager assetManager = null;
/** The workflow definition scanner */
private WorkflowDefinitionScanner workflowDefinitionScanner;
/** List of initially delayed workflows */
private final List<Long> delayedWorkflows = new ArrayList<Long>();
/** Striped locks for synchronization */
private final Striped<Lock> lock = Striped.lazyWeakLock(1024);
private final Striped<Lock> updateLock = Striped.lazyWeakLock(1024);
private final Striped<Lock> mediaPackageLocks = Striped.lazyWeakLock(1024);
/** The Elasticsearch indices */
private ElasticsearchIndex index;
/**
* Constructs a new workflow service impl, with a priority-sorted map of metadata services
*/
public WorkflowServiceImpl() {
metadataServices = new TreeSet<>(Comparator.comparingInt(MetadataService::getPriority));
}
/**
* Activate this service implementation via the OSGI service component runtime.
*
* @param componentContext
* the component context
*/
@Activate
public void activate(ComponentContext componentContext) {
this.componentContext = componentContext;
executorService = (ThreadPoolExecutor) Executors.newCachedThreadPool();
logger.info("Activate Workflow service");
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#addWorkflowListener(org.opencastproject.workflow.api.WorkflowListener)
*/
@Override
public void addWorkflowListener(WorkflowListener listener) {
listeners.add(listener);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#removeWorkflowListener(org.opencastproject.workflow.api.WorkflowListener)
*/
@Override
public void removeWorkflowListener(WorkflowListener listener) {
listeners.remove(listener);
}
/**
* Fires the workflow listeners on workflow updates.
*/
protected void fireListeners(final WorkflowInstance oldWorkflowInstance, final WorkflowInstance newWorkflowInstance) {
final User currentUser = securityService.getUser();
final Organization currentOrganization = securityService.getOrganization();
for (final WorkflowListener listener : listeners) {
if (oldWorkflowInstance == null || !oldWorkflowInstance.getState().equals(newWorkflowInstance.getState())) {
Runnable runnable = () -> {
try {
securityService.setUser(currentUser);
securityService.setOrganization(currentOrganization);
listener.stateChanged(newWorkflowInstance);
} finally {
securityService.setUser(null);
securityService.setOrganization(null);
}
};
executorService.execute(runnable);
} else {
logger.debug("Not notifying {} because the workflow state has not changed", listener);
}
if (newWorkflowInstance.getCurrentOperation() != null) {
if (oldWorkflowInstance == null || oldWorkflowInstance.getCurrentOperation() == null
|| !oldWorkflowInstance.getCurrentOperation().equals(newWorkflowInstance.getCurrentOperation())) {
Runnable runnable = () -> {
try {
securityService.setUser(currentUser);
securityService.setOrganization(currentOrganization);
listener.operationChanged(newWorkflowInstance);
} finally {
securityService.setUser(null);
securityService.setOrganization(null);
}
};
executorService.execute(runnable);
}
} else {
logger.debug("Not notifying {} because the workflow operation has not changed", listener);
}
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#listAvailableWorkflowDefinitions()
*/
@Override
public List<WorkflowDefinition> listAvailableWorkflowDefinitions() {
return workflowDefinitionScanner
.getAvailableWorkflowDefinitions(securityService.getOrganization(), securityService.getUser())
.sorted()
.collect(Collectors.toList());
}
/**
* {@inheritDoc}
*/
public boolean isRunnable(WorkflowDefinition workflowDefinition) {
List<String> availableOperations = listAvailableOperationNames();
List<WorkflowDefinition> checkedWorkflows = new ArrayList<>();
boolean runnable = isRunnable(workflowDefinition, availableOperations, checkedWorkflows);
int wfCount = checkedWorkflows.size() - 1;
if (runnable)
logger.info("Workflow {}, containing {} derived workflows, is runnable", workflowDefinition, wfCount);
else
logger.warn("Workflow {}, containing {} derived workflows, is not runnable", workflowDefinition, wfCount);
return runnable;
}
/**
* Tests the workflow definition for its runnability. This method is a helper for
* {@link #isRunnable(WorkflowDefinition)} that is suited for recursive calling.
*
* @param workflowDefinition
* the definition to test
* @param availableOperations
* list of currently available operation handlers
* @param checkedWorkflows
* list of checked workflows, used to avoid circular checking
* @return <code>true</code> if all bits and pieces used for executing <code>workflowDefinition</code> are in place
*/
private boolean isRunnable(WorkflowDefinition workflowDefinition, List<String> availableOperations,
List<WorkflowDefinition> checkedWorkflows) {
if (checkedWorkflows.contains(workflowDefinition))
return true;
// Test availability of operation handler and catch workflows
for (WorkflowOperationDefinition op : workflowDefinition.getOperations()) {
if (!availableOperations.contains(op.getId())) {
logger.info("{} is not runnable due to missing operation {}", workflowDefinition, op);
return false;
}
String catchWorkflow = op.getExceptionHandlingWorkflow();
if (catchWorkflow != null) {
WorkflowDefinition catchWorkflowDefinition;
try {
catchWorkflowDefinition = getWorkflowDefinitionById(catchWorkflow);
} catch (NotFoundException e) {
logger.info("{} is not runnable due to missing catch workflow {} on operation {}", workflowDefinition,
catchWorkflow, op);
return false;
}
if (!isRunnable(catchWorkflowDefinition, availableOperations, checkedWorkflows))
return false;
}
}
// Add the workflow to the list of checked workflows
if (!checkedWorkflows.contains(workflowDefinition))
checkedWorkflows.add(workflowDefinition);
return true;
}
/**
* Gets the currently registered workflow operation handlers.
*
* @return All currently registered handlers
*/
public Set<HandlerRegistration> getRegisteredHandlers() {
Set<HandlerRegistration> set = new HashSet<>();
ServiceReference[] refs;
try {
refs = componentContext.getBundleContext().getServiceReferences(WorkflowOperationHandler.class.getName(), null);
} catch (InvalidSyntaxException e) {
throw new IllegalStateException(e);
}
if (refs != null) {
for (ServiceReference ref : refs) {
WorkflowOperationHandler handler = (WorkflowOperationHandler) componentContext.getBundleContext().getService(
ref);
set.add(new HandlerRegistration((String) ref.getProperty(WORKFLOW_OPERATION_PROPERTY), handler));
}
} else {
logger.warn("No registered workflow operation handlers found");
}
return set;
}
protected WorkflowOperationHandler getWorkflowOperationHandler(String operationId) {
for (HandlerRegistration reg : getRegisteredHandlers()) {
if (reg.operationName.equals(operationId))
return reg.handler;
}
return null;
}
/**
* Lists the names of each workflow operation. Operation names are availalbe for use if there is a registered
* {@link WorkflowOperationHandler} with an equal {@link WorkflowServiceImpl#WORKFLOW_OPERATION_PROPERTY} property.
*
* @return The {@link List} of available workflow operation names
*/
protected List<String> listAvailableOperationNames() {
return getRegisteredHandlers().parallelStream().map(op -> op.operationName).collect(Collectors.toList());
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#getWorkflowById(long)
*/
@Override
public WorkflowInstance getWorkflowById(long id) throws NotFoundException,
UnauthorizedException {
try {
WorkflowInstance workflow = persistence.getWorkflow(id);
assertPermission(workflow, Permissions.Action.READ.toString(), workflow.getOrganizationId());
return workflow;
} catch (WorkflowDatabaseException e) {
throw new IllegalStateException("Got not get workflow from database with id ");
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#start(org.opencastproject.workflow.api.WorkflowDefinition,
* org.opencastproject.mediapackage.MediaPackage)
*/
@Override
public WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage)
throws WorkflowDatabaseException, UnauthorizedException, WorkflowParsingException {
return start(workflowDefinition, mediaPackage, new HashMap<>());
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#start(org.opencastproject.workflow.api.WorkflowDefinition,
* org.opencastproject.mediapackage.MediaPackage)
*/
@Override
public WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage mediaPackage,
Map<String, String> properties)
throws WorkflowDatabaseException, UnauthorizedException, WorkflowParsingException {
try {
return start(workflowDefinition, mediaPackage, null, properties);
} catch (NotFoundException e) {
// should never happen
throw new IllegalStateException("a null workflow ID caused a NotFoundException. This is a programming error.");
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#start(org.opencastproject.workflow.api.WorkflowDefinition,
* org.opencastproject.mediapackage.MediaPackage, Long, java.util.Map)
*/
@Override
public WorkflowInstance start(WorkflowDefinition workflowDefinition, MediaPackage sourceMediaPackage,
Long parentWorkflowId, Map<String, String> originalProperties) throws WorkflowDatabaseException,
NotFoundException, UnauthorizedException, WorkflowParsingException, IllegalStateException {
final String mediaPackageId = sourceMediaPackage.getIdentifier().toString();
Map<String, String> properties = null;
// WorkflowPropertiesUtil.storeProperties will take a snapshot if there isn't one
// and we want the mp in the snapshot to have all the metadata populated.
populateMediaPackageMetadata(sourceMediaPackage);
if (originalProperties != null) {
WorkflowPropertiesUtil.storeProperties(assetManager, sourceMediaPackage, originalProperties);
properties = WorkflowPropertiesUtil.getLatestWorkflowProperties(assetManager, mediaPackageId);
}
// We have to synchronize per media package to avoid starting multiple simultaneous workflows for one media package.
final Lock lock = mediaPackageLocks.get(mediaPackageId);
lock.lock();
try {
if (workflowDefinition == null)
throw new IllegalArgumentException("workflow definition must not be null");
for (List<String> errors : MediaPackageSupport.sanityCheck(sourceMediaPackage)) {
throw new IllegalArgumentException("Insane media package cannot be processed: " + String.join("; ", errors));
}
if (parentWorkflowId != null) {
try {
getWorkflowById(parentWorkflowId); // Let NotFoundException bubble up
} catch (UnauthorizedException e) {
throw new IllegalArgumentException("Parent workflow " + parentWorkflowId + " not visible to this user");
}
} else {
if (persistence.mediaPackageHasActiveWorkflows(mediaPackageId)) {
throw new IllegalStateException(String.format(
"Can't start workflow '%s' for media package '%s' because another workflow is currently active.",
workflowDefinition.getTitle(),
sourceMediaPackage.getIdentifier().toString()));
}
}
// Get the current user
User currentUser = securityService.getUser();
validUserOrThrow(currentUser);
// Get the current organization
Organization organization = securityService.getOrganization();
if (organization == null)
throw new SecurityException("Current organization is unknown");
WorkflowInstance workflowInstance = new WorkflowInstance(workflowDefinition, sourceMediaPackage,
currentUser, organization, properties);
workflowInstance = updateConfiguration(workflowInstance, properties);
// Create and configure the workflow instance
try {
// Create a new job for this workflow instance
String workflowDefinitionXml = XmlWorkflowParser.toXml(workflowDefinition);
String mediaPackageXml = MediaPackageParser.getAsXml(sourceMediaPackage);
List<String> arguments = new ArrayList<>();
arguments.add(workflowDefinitionXml);
arguments.add(mediaPackageXml);
if (parentWorkflowId != null || properties != null) {
String parentWorkflowIdString = (parentWorkflowId != null) ? parentWorkflowId.toString() : NULL_PARENT_ID;
arguments.add(parentWorkflowIdString);
}
if (properties != null) {
arguments.add(mapToString(properties));
}
Job job = serviceRegistry.createJob(JOB_TYPE, Operation.START_WORKFLOW.toString(), arguments,
null, false, null, WORKFLOW_JOB_LOAD);
// Have the workflow take on the job's identity
workflowInstance.setId(job.getId());
// Add the workflow to the search index and have the job enqueued for dispatch.
// Update also sets ACL and mediapackage metadata
update(workflowInstance);
return workflowInstance;
} catch (Throwable t) {
try {
workflowInstance.setState(FAILED);
update(workflowInstance);
} catch (Exception failureToFail) {
logger.warn("Unable to update workflow to failed state", failureToFail);
}
try {
throw t;
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
}
}
} finally {
lock.unlock();
}
}
protected WorkflowInstance updateConfiguration(WorkflowInstance instance, Map<String, String> properties) {
if (properties != null) {
for (Entry<String, String> entry : properties.entrySet()) {
instance.setConfiguration(entry.getKey(), entry.getValue());
}
}
return instance;
}
/**
* Does a lookup of available operation handlers for the given workflow operation.
*
* @param operation
* the operation definition
* @return the handler or <code>null</code>
*/
protected WorkflowOperationHandler selectOperationHandler(WorkflowOperationInstance operation) {
List<WorkflowOperationHandler> handlerList = new ArrayList<>();
for (HandlerRegistration handlerReg : getRegisteredHandlers()) {
if (handlerReg.operationName != null && handlerReg.operationName.equals(operation.getTemplate())) {
handlerList.add(handlerReg.handler);
}
}
if (handlerList.size() > 1) {
throw new IllegalStateException("Multiple operation handlers found for operation '" + operation.getTemplate()
+ "'");
} else if (handlerList.size() == 1) {
return handlerList.get(0);
}
logger.warn("No workflow operation handlers found for operation '{}'", operation.getTemplate());
return null;
}
/**
* Executes the workflow.
*
* @param workflow
* the workflow instance
* @throws WorkflowException
* if there is a problem processing the workflow
* @throws UnauthorizedException
*/
protected Job runWorkflow(WorkflowInstance workflow) throws WorkflowException, UnauthorizedException {
if (INSTANTIATED != workflow.getState()) {
// If the workflow is "running", we need to determine if there is an operation being executed or not.
// When a workflow has been restarted, this might not be the case and the status might not have been
// updated accordingly.
if (RUNNING == workflow.getState()) {
WorkflowOperationInstance currentOperation = workflow.getCurrentOperation();
if (currentOperation != null) {
if (currentOperation.getId() != null) {
try {
Job operationJob = serviceRegistry.getJob(currentOperation.getId());
if (Job.Status.RUNNING.equals(operationJob.getStatus())) {
logger.debug("Not starting workflow {}, it is already in running state", workflow);
return null;
} else {
logger.info("Scheduling next operation of workflow {}", workflow);
operationJob.setStatus(Status.QUEUED);
operationJob.setDispatchable(true);
return serviceRegistry.updateJob(operationJob);
}
} catch (Exception e) {
logger.warn("Error determining status of current workflow operation in {}", workflow, e);
return null;
}
}
} else {
throw new IllegalStateException("Cannot start a workflow '" + workflow + "' with no current operation");
}
} else {
throw new IllegalStateException("Cannot start a workflow in state '" + workflow.getState() + "'");
}
}
// If this is a new workflow, move to the first operation
workflow.setState(RUNNING);
update(workflow);
WorkflowOperationInstance operation = workflow.getCurrentOperation();
if (operation == null) {
throw new IllegalStateException("Cannot start a workflow without a current operation");
}
if (!operation.equals(workflow.getOperations().get(0))) {
throw new IllegalStateException("Current operation expected to be first");
}
try {
logger.info("Scheduling workflow {} for execution", workflow.getId());
Job job = serviceRegistry.createJob(JOB_TYPE, Operation.START_OPERATION.toString(),
Collections.singletonList(Long.toString(workflow.getId())), null, false, null, WORKFLOW_JOB_LOAD);
operation.setId(job.getId());
update(workflow);
job.setStatus(Status.QUEUED);
job.setDispatchable(true);
return serviceRegistry.updateJob(job);
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
} catch (NotFoundException e) {
// this should be impossible
throw new IllegalStateException("Unable to find a job that was just created");
}
}
/**
* Executes the workflow's current operation.
*
* @param workflow
* the workflow
* @param properties
* the properties that are passed in on resume
* @return the processed workflow operation
* @throws WorkflowException
* if there is a problem processing the workflow
* @throws UnauthorizedException
*/
protected WorkflowOperationInstance runWorkflowOperation(WorkflowInstance workflow, Map<String, String> properties)
throws WorkflowException, UnauthorizedException {
WorkflowOperationInstance processingOperation = workflow.getCurrentOperation();
if (processingOperation == null)
throw new IllegalStateException("Workflow '" + workflow + "' has no operation to run");
// Keep the current state for later reference, it might have been changed from the outside
WorkflowState initialState = workflow.getState();
// Execute the operation handler
WorkflowOperationHandler operationHandler = selectOperationHandler(processingOperation);
WorkflowOperationWorker worker = new WorkflowOperationWorker(operationHandler, workflow, properties, this);
workflow = worker.execute();
Long currentOperationJobId = processingOperation.getId();
try {
updateOperationJob(currentOperationJobId, processingOperation.getState());
} catch (NotFoundException e) {
throw new IllegalStateException("Unable to find a job that has already been running");
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
}
// Move on to the next workflow operation
WorkflowOperationInstance currentOperation = workflow.getCurrentOperation();
// Is the workflow done?
if (currentOperation == null) {
// If we are in failing mode, we were simply working off an error handling workflow
if (FAILING.equals(workflow.getState())) {
workflow.setState(FAILED);
}
// Otherwise, let's make sure we didn't miss any failed operation, since the workflow state could have been
// switched to paused while processing the error handling workflow extension
else if (!FAILED.equals(workflow.getState())) {
workflow.setState(SUCCEEDED);
for (WorkflowOperationInstance op : workflow.getOperations()) {
if (op.getState().equals(WorkflowOperationInstance.OperationState.FAILED)) {
if (op.isFailOnError()) {
workflow.setState(FAILED);
break;
}
}
}
}
// Save the updated workflow to the database
logger.debug("{} has {}", workflow, workflow.getState());
update(workflow);
} else {
// Somebody might have set the workflow to "paused" from the outside, so take a look a the database first
WorkflowState dbWorkflowState;
try {
dbWorkflowState = getWorkflowById(workflow.getId()).getState();
} catch (NotFoundException e) {
throw new IllegalStateException("The workflow with ID " + workflow.getId()
+ " can not be found in the database", e);
} catch (UnauthorizedException e) {
throw new IllegalStateException("The workflow with ID " + workflow.getId() + " can not be read", e);
}
// If somebody changed the workflow state from the outside, that state should take precedence
if (!dbWorkflowState.equals(initialState)) {
logger.info("Workflow state for {} was changed to '{}' from the outside", workflow, dbWorkflowState);
workflow.setState(dbWorkflowState);
}
// Save the updated workflow to the database
Job job;
switch (workflow.getState()) {
case FAILED:
update(workflow);
break;
case FAILING:
case RUNNING:
try {
job = serviceRegistry.createJob(JOB_TYPE, Operation.START_OPERATION.toString(),
Collections.singletonList(Long.toString(workflow.getId())), null, false, null, WORKFLOW_JOB_LOAD);
currentOperation.setId(job.getId());
update(workflow);
job.setStatus(Status.QUEUED);
job.setDispatchable(true);
serviceRegistry.updateJob(job);
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
} catch (NotFoundException e) {
// this should be impossible
throw new IllegalStateException("Unable to find a job that was just created");
}
break;
case PAUSED:
case STOPPED:
case SUCCEEDED:
update(workflow);
break;
case INSTANTIATED:
update(workflow);
throw new IllegalStateException("Impossible workflow state found during processing");
default:
throw new IllegalStateException("Unknown workflow state found during processing");
}
}
return processingOperation;
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#getWorkflowDefinitionById(String)
*/
@Override
public WorkflowDefinition getWorkflowDefinitionById(String id) throws NotFoundException {
final WorkflowIdentifier workflowIdentifier = new WorkflowIdentifier(id, securityService.getOrganization().getId());
final WorkflowDefinition def = workflowDefinitionScanner
.getWorkflowDefinition(securityService.getUser(), workflowIdentifier);
if (def == null) {
throw new NotFoundException("Workflow definition '" + workflowIdentifier + "' not found or inaccessible");
}
return def;
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#stop(long)
*/
@Override
public WorkflowInstance stop(long workflowInstanceId) throws WorkflowException, NotFoundException,
UnauthorizedException {
final Lock lock = this.lock.get(workflowInstanceId);
lock.lock();
try {
WorkflowInstance instance = getWorkflowById(workflowInstanceId);
if (instance.getState() != STOPPED) {
// Update the workflow instance
instance.setState(STOPPED);
update(instance);
}
try {
removeTempFiles(instance);
} catch (Exception e) {
logger.warn("Cannot remove temp files for workflow instance {}", workflowInstanceId, e);
}
return instance;
} finally {
lock.unlock();
}
}
/**
* Checks whether user is set and is known to the userDirectoryService
*/
private void validUserOrThrow(User user) {
if (user == null)
throw new SecurityException("Current user is unknown");
if (userDirectoryService.loadUser(user.getUsername()) == null)
throw new SecurityException(String.format("Current user '%s' can not be loaded", user.getUsername()));
}
private void removeTempFiles(WorkflowInstance workflowInstance) {
logger.info("Removing temporary files for workflow {}", workflowInstance.getId());
MediaPackage mp = workflowInstance.getMediaPackage();
if (null == mp) {
logger.warn("Workflow instance {} does not have an media package set", workflowInstance.getId());
return;
}
for (MediaPackageElement elem : mp.getElements()) {
// Publications should not link to temporary files and can be skipped
if (elem instanceof Publication) {
continue;
}
if (null == elem.getURI()) {
logger.warn("Media package element {} from the media package {} does not have an URI set",
elem.getIdentifier(), mp.getIdentifier());
continue;
}
try {
logger.debug("Removing temporary file {} for workflow {}", elem.getURI(), workflowInstance);
workspace.delete(elem.getURI());
} catch (IOException e) {
logger.warn("Unable to delete mediapackage element", e);
} catch (NotFoundException e) {
// File was probably already deleted before...
}
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#remove(long)
*/
@Override
public void remove(long workflowInstanceId) throws WorkflowDatabaseException, NotFoundException,
UnauthorizedException, WorkflowParsingException, WorkflowStateException {
remove(workflowInstanceId, false);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#remove(long,boolean)
*/
@Override
public void remove(long workflowInstanceId, boolean force) throws WorkflowDatabaseException, NotFoundException,
UnauthorizedException, WorkflowStateException {
final Lock lock = this.lock.get(workflowInstanceId);
lock.lock();
try {
WorkflowInstance instance = getWorkflowById(workflowInstanceId);
WorkflowInstance.WorkflowState state = instance.getState();
if (!state.isTerminated() && !force) {
throw new WorkflowStateException("Workflow instance with state '" + state + "' cannot be removed "
+ "since it is not yet terminated.");
}
assertPermission(instance, Permissions.Action.WRITE.toString(), instance.getOrganizationId());
// First, remove temporary files
removeTempFiles(instance);
// Second, remove jobs related to operations which belong to the workflow instance
List<WorkflowOperationInstance> operations = instance.getOperations();
List<Long> jobsToDelete = new ArrayList<>();
for (WorkflowOperationInstance op : operations) {
if (op.getId() != null) {
long workflowOpId = op.getId();
if (workflowOpId != workflowInstanceId) {
jobsToDelete.add(workflowOpId);
}
}
}
try {
serviceRegistry.removeJobs(jobsToDelete);
} catch (ServiceRegistryException e) {
logger.warn("Problems while removing jobs related to workflow operations '{}'", jobsToDelete, e);
} catch (NotFoundException e) {
logger.debug("No jobs related to one of the workflow operation '{}' found in service registry", jobsToDelete);
}
// Third, remove workflow instance job itself
try {
serviceRegistry.removeJobs(Collections.singletonList(workflowInstanceId));
removeWorkflowInstanceFromIndex(instance.getId());
} catch (ServiceRegistryException e) {
logger.warn("Problems while removing workflow instance job '{}'", workflowInstanceId, e);
} catch (NotFoundException e) {
logger.info("No workflow instance job '{}' found in the service registry", workflowInstanceId);
}
// Remove workflow from database
persistence.removeFromDatabase(instance);
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#suspend(long)
*/
@Override
public WorkflowInstance suspend(long workflowInstanceId) throws WorkflowException, NotFoundException,
UnauthorizedException {
final Lock lock = this.lock.get(workflowInstanceId);
lock.lock();
try {
WorkflowInstance instance = getWorkflowById(workflowInstanceId);
instance.setState(PAUSED);
update(instance);
return instance;
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#resume(long)
*/
@Override
public WorkflowInstance resume(long id) throws WorkflowException, NotFoundException, IllegalStateException,
UnauthorizedException {
return resume(id, null);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#resume(long, Map)
*/
@Override
public WorkflowInstance resume(long workflowInstanceId, Map<String, String> properties) throws WorkflowException,
NotFoundException, IllegalStateException, UnauthorizedException {
WorkflowInstance workflowInstance = getWorkflowById(workflowInstanceId);
if (!WorkflowState.PAUSED.equals(workflowInstance.getState()))
throw new IllegalStateException("Can not resume a workflow where the current state is not in paused");
workflowInstance = updateConfiguration(workflowInstance, properties);
update(workflowInstance);
WorkflowOperationInstance currentOperation = workflowInstance.getCurrentOperation();
// Is the workflow done?
if (currentOperation == null) {
// Let's make sure we didn't miss any failed operation, since the workflow state could have been
// switched to paused while processing the error handling workflow extension
workflowInstance.setState(SUCCEEDED);
for (WorkflowOperationInstance op : workflowInstance.getOperations()) {
if (op.getState().equals(WorkflowOperationInstance.OperationState.FAILED)) {
if (op.isFailOnError()) {
workflowInstance.setState(FAILED);
break;
}
}
}
// Save the resumed workflow to the database
logger.debug("{} has {}", workflowInstance, workflowInstance.getState());
update(workflowInstance);
return workflowInstance;
}
// We can resume workflows when they are in either the paused state, or they are being advanced manually passed
// certain operations. In the latter case, there is no current paused operation.
if (OperationState.INSTANTIATED.equals(currentOperation.getState())) {
try {
// the operation has its own job. Update that too.
Job operationJob = serviceRegistry.createJob(JOB_TYPE, Operation.START_OPERATION.toString(),
Collections.singletonList(Long.toString(workflowInstanceId)), null, false, null, WORKFLOW_JOB_LOAD);
// this method call is publicly visible, so it doesn't necessarily go through the accept method. Set the
// workflow state manually.
workflowInstance.setState(RUNNING);
currentOperation.setId(operationJob.getId());
// update the workflow and its associated job
update(workflowInstance);
// Now set this job to be queued so it can be dispatched
operationJob.setStatus(Status.QUEUED);
operationJob.setDispatchable(true);
serviceRegistry.updateJob(operationJob);
return workflowInstance;
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
}
}
Long operationJobId = workflowInstance.getCurrentOperation().getId();
if (operationJobId == null)
throw new IllegalStateException("Can not resume a workflow where the current operation has no associated id");
// Set the current operation's job to queued, so it gets picked up again
Job workflowJob;
try {
workflowJob = serviceRegistry.getJob(workflowInstanceId);
workflowJob.setStatus(Status.RUNNING);
persistence.updateInDatabase(workflowInstance);
serviceRegistry.updateJob(workflowJob);
Job operationJob = serviceRegistry.getJob(operationJobId);
operationJob.setStatus(Status.QUEUED);
operationJob.setDispatchable(true);
if (properties != null) {
Properties props = new Properties();
props.putAll(properties);
ByteArrayOutputStream out = new ByteArrayOutputStream();
props.store(out, null);
List<String> newArguments = new ArrayList<String>(operationJob.getArguments());
newArguments.add(new String(out.toByteArray(), StandardCharsets.UTF_8));
operationJob.setArguments(newArguments);
}
serviceRegistry.updateJob(operationJob);
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(e);
} catch (IOException e) {
throw new WorkflowParsingException("Unable to parse workflow and/or workflow properties");
}
return workflowInstance;
}
/**
* Asserts that the current user has permission to take the provided action on a workflow instance.
*
* @param workflow
* the workflow instance
* @param action
* the action to ensure is permitted
* @throws UnauthorizedException
* if the action is not authorized
*/
protected void assertPermission(WorkflowInstance workflow, String action, String workflowOrgId) throws UnauthorizedException {
User currentUser = securityService.getUser();
Organization currentOrg = securityService.getOrganization();
String currentOrgAdminRole = currentOrg.getAdminRole();
String currentOrgId = currentOrg.getId();
MediaPackage mediapackage = workflow.getMediaPackage();
WorkflowState state = workflow.getState();
if (state != INSTANTIATED && state != RUNNING && workflow.getState() != FAILING) {
Optional<MediaPackage> assetMediapackage = assetManager.getMediaPackage(mediapackage.getIdentifier().toString());
if (assetMediapackage.isPresent()) {
mediapackage = assetMediapackage.get();
}
}
var creatorName = workflow.getCreatorName();
var workflowCreator = creatorName == null ? null : userDirectoryService.loadUser(creatorName);
boolean authorized = currentUser.hasRole(GLOBAL_ADMIN_ROLE)
|| (currentUser.hasRole(currentOrgAdminRole) && currentOrgId.equals(workflowOrgId))
|| (currentUser.equals(workflowCreator))
|| (authorizationService.hasPermission(mediapackage, action) && currentOrgId.equals(workflowOrgId));
if (!authorized) {
throw new UnauthorizedException(currentUser, action);
}
}
protected boolean assertMediaPackagePermission(String mediaPackageId, String action) {
var currentUser = securityService.getUser();
Optional<MediaPackage> mp = assetManager.getMediaPackage(mediaPackageId);
// asset manager already checks if user is admin, org admin for same org as mp, or has explicit read rights
// global admins can still get workflow instances if mp is gone from asset manager
// org admins can't because then we don't know if mp belonged to same org as user
return currentUser.hasRole(GLOBAL_ADMIN_ROLE)
|| mp.isPresent() && currentUser.hasRole(securityService.getOrganization().getAdminRole())
|| mp.isPresent() && authorizationService.hasPermission(mp.get(), action);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#update(org.opencastproject.workflow.api.WorkflowInstance)
*/
@Override
public void update(final WorkflowInstance workflowInstance) throws WorkflowDatabaseException, UnauthorizedException {
final Lock lock = updateLock.get(workflowInstance.getId());
lock.lock();
try {
WorkflowInstance originalWorkflowInstance = null;
try {
// get workflow and assert permissions
originalWorkflowInstance = getWorkflowById(workflowInstance.getId());
} catch (NotFoundException e) {
// That's fine, it's a new workflow instance
}
MediaPackage updatedMediaPackage = null;
try {
// Before we persist this, extract the metadata
updatedMediaPackage = workflowInstance.getMediaPackage();
populateMediaPackageMetadata(updatedMediaPackage);
String seriesId = updatedMediaPackage.getSeries();
if (seriesId != null && workflowInstance.getCurrentOperation() != null) {
// If the mediapackage contains a series, find the series ACLs and add the security information to the
// mediapackage
try {
AccessControlList acl = seriesService.getSeriesAccessControl(seriesId);
Tuple<AccessControlList, AclScope> activeAcl = authorizationService.getAcl(
updatedMediaPackage, AclScope.Series);
// Update series ACL if it differs from the active series ACL on the media package
if (!AclScope.Series.equals(activeAcl.getB()) || !AccessControlUtil.equals(activeAcl.getA(), acl)) {
authorizationService.setAcl(updatedMediaPackage, AclScope.Series, acl);
}
} catch (NotFoundException e) {
logger.debug("Not updating series ACL on event {} since series {} has no ACL set",
updatedMediaPackage, seriesId, e);
}
}
workflowInstance.setMediaPackage(updatedMediaPackage);
} catch (SeriesException e) {
throw new WorkflowDatabaseException(e);
} catch (Exception e) {
logger.error("Metadata for media package {} could not be updated", updatedMediaPackage, e);
}
// Synchronize the job status with the workflow
WorkflowState workflowState = workflowInstance.getState();
Job job;
try {
job = serviceRegistry.getJob(workflowInstance.getId());
job.setPayload(Long.toString(workflowInstance.getId()));
// Synchronize workflow and job state
switch (workflowState) {
case FAILED:
job.setStatus(Status.FAILED);
break;
case FAILING:
break;
case INSTANTIATED:
job.setDispatchable(true);
job.setStatus(Status.QUEUED);
break;
case PAUSED:
job.setStatus(Status.PAUSED);
break;
case RUNNING:
job.setStatus(Status.RUNNING);
break;
case STOPPED:
job.setStatus(Status.CANCELLED);
break;
case SUCCEEDED:
job.setStatus(Status.FINISHED);
break;
default:
throw new IllegalStateException("Found a workflow state that is not handled");
}
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException(
"Unable to read workflow job " + workflowInstance.getId() + " from service registry", e);
} catch (NotFoundException e) {
throw new WorkflowDatabaseException(
"Job for workflow " + workflowInstance.getId() + " not found in service registry", e);
}
// Update both workflow and workflow job
try {
//Update the database
persistence.updateInDatabase(workflowInstance);
job = serviceRegistry.updateJob(job);
WorkflowOperationInstance op = workflowInstance.getCurrentOperation();
// Update index used for UI. Note that we only need certain metadata and we can safely filter out workflow
// updates for running operations since we updated the metadata right before these operations and will do so
// again right after those operations.
if (op == null || op.getState() != OperationState.RUNNING) {
// Collect necessary information only for index update
long id = workflowInstance.getId();
int state = workflowInstance.getState().ordinal();
var template = workflowInstance.getTemplate();
String mpId = workflowInstance.getMediaPackage().getIdentifier().toString();
String orgId = workflowInstance.getOrganizationId();
updateWorkflowInstanceInIndex(id, state, template, mpId, orgId);
}
} catch (ServiceRegistryException e) {
throw new WorkflowDatabaseException("Update of workflow job " + workflowInstance.getId()
+ " in the service registry failed, service registry and workflow table may be out of sync", e);
} catch (NotFoundException e) {
throw new WorkflowDatabaseException("Job for workflow " + workflowInstance.getId()
+ " not found in service registry", e);
} catch (Exception e) {
throw new WorkflowDatabaseException("Update of workflow job " + job.getId() + " in the service registry failed, "
+ "service registry and workflow table may be out of sync", e);
}
try {
fireListeners(originalWorkflowInstance, workflowInstance);
} catch (Exception e) {
// Can't happen, since we are converting from an in-memory object
throw new IllegalStateException("In-memory workflow instance could not be serialized", e);
}
} finally {
lock.unlock();
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#countWorkflowInstances()
*/
@Override
public long countWorkflowInstances() throws WorkflowDatabaseException {
return countWorkflowInstances(null);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#countWorkflowInstances(org.opencastproject.workflow.api.WorkflowInstance.WorkflowState)
*/
@Override
public long countWorkflowInstances(WorkflowState state) throws WorkflowDatabaseException {
return persistence.countWorkflows(state);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#getWorkflowInstancesByMediaPackage(String)
*/
@Override
public List<WorkflowInstance> getWorkflowInstancesByMediaPackage(String mediaPackageId)
throws WorkflowDatabaseException, UnauthorizedException {
// If we have read permission to the media package, return all workflows
if (!assertMediaPackagePermission(mediaPackageId, Permissions.Action.READ.toString())) {
throw new UnauthorizedException("Not allowed to access event");
}
return persistence.getWorkflowInstancesByMediaPackage(mediaPackageId);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#getRunningWorkflowInstanceByMediaPackage(String, String)
*/
@Override
public Optional<WorkflowInstance> getRunningWorkflowInstanceByMediaPackage(String mediaPackageId, String action)
throws WorkflowException, UnauthorizedException, WorkflowDatabaseException {
List<WorkflowInstance> workflowInstances = persistence.getRunningWorkflowInstancesByMediaPackage(mediaPackageId);
// If there is more than workflow running something is very wrong
if (workflowInstances.size() > 1) {
throw new WorkflowException("Multiple workflows are active on mediapackage " + mediaPackageId);
}
Optional<WorkflowInstance> optWorkflowInstance = Optional.empty();
if (workflowInstances.size() == 1) {
WorkflowInstance wfInstance = workflowInstances.get(0);
optWorkflowInstance = Optional.of(wfInstance);
assertPermission(wfInstance, action, wfInstance.getOrganizationId());
}
return optWorkflowInstance;
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#mediaPackageHasActiveWorkflows(String)
*/
@Override
public boolean mediaPackageHasActiveWorkflows(String mediaPackageId) throws WorkflowDatabaseException {
return persistence.mediaPackageHasActiveWorkflows(mediaPackageId);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.workflow.api.WorkflowService#userHasActiveWorkflows(String)
*/
@Override
public boolean userHasActiveWorkflows(String userId) throws WorkflowDatabaseException {
return persistence.userHasActiveWorkflows(userId);
}
/**
* Callback for workflow operations that were throwing an exception. This implementation assumes that the operation
* worker has already adjusted the current operation's state appropriately.
*
* @param workflow
* the workflow instance
* @param currentOperation
* the current workflow operation
* @return the workflow instance
*/
protected WorkflowInstance handleOperationException(
WorkflowInstance workflow,
WorkflowOperationInstance currentOperation) {
int failedAttempt = currentOperation.getFailedAttempts() + 1;
currentOperation.setFailedAttempts(failedAttempt);
// Operation was aborted by the user, after going into hold state
if (ERROR_RESOLUTION_HANDLER_ID.equals(currentOperation.getTemplate())
&& OperationState.FAILED.equals(currentOperation.getState())) {
int position = workflow.getOperations().indexOf(currentOperation);
// Advance to operation that actually failed
if (workflow.getOperations().size() > position + 1) { // This should always be true...
currentOperation = workflow.getOperations().get(position + 1);
// It's currently in RETRY state, change to FAILED
currentOperation.setState(OperationState.FAILED);
}
handleFailedOperation(workflow, currentOperation);
} else if (currentOperation.getMaxAttempts() != -1 && failedAttempt == currentOperation.getMaxAttempts()) {
handleFailedOperation(workflow, currentOperation);
} else {
switch (currentOperation.getRetryStrategy()) {
case NONE:
handleFailedOperation(workflow, currentOperation);
break;
case RETRY:
currentOperation.setState(OperationState.RETRY);
break;
case HOLD:
currentOperation.setState(OperationState.RETRY);
List<WorkflowOperationInstance> operations = workflow.getOperations();
WorkflowOperationDefinitionImpl errorResolutionDefinition = new WorkflowOperationDefinitionImpl(
ERROR_RESOLUTION_HANDLER_ID, "Error Resolution Operation", "error", false);
var errorResolutionInstance = new WorkflowOperationInstance(errorResolutionDefinition);
errorResolutionInstance.setExceptionHandlingWorkflow(currentOperation.getExceptionHandlingWorkflow());
var index = workflow.getOperations().indexOf(currentOperation);
operations.add(index, errorResolutionInstance);
workflow.setOperations(operations);
break;
default:
break;
}
}
return workflow;
}
/**
* Handles the workflow for a failing operation.
*
* @param workflow
* the workflow
* @param currentOperation
* the failing workflow operation instance
*/
private void handleFailedOperation(WorkflowInstance workflow, WorkflowOperationInstance currentOperation) {
String errorDefId = currentOperation.getExceptionHandlingWorkflow();
// Adjust the workflow state according to the setting on the operation
if (currentOperation.isFailOnError()) {
if (StringUtils.isBlank(errorDefId)) {
workflow.setState(FAILED);
} else {
workflow.setState(FAILING);
// Remove the rest of the original workflow
int currentOperationPosition = workflow.getOperations().indexOf(currentOperation);
List<WorkflowOperationInstance> operations = new ArrayList<>(
workflow.getOperations().subList(0, currentOperationPosition + 1));
workflow.setOperations(operations);
// Determine the current workflow configuration
Map<String, String> configuration = new HashMap<>();
for (String configKey : workflow.getConfigurationKeys()) {
configuration.put(configKey, workflow.getConfiguration(configKey));
}
// Append the operations
WorkflowDefinition errorDef = null;
try {
errorDef = getWorkflowDefinitionById(errorDefId);
workflow.extend(errorDef);
workflow.setOperations(updateConfiguration(workflow, configuration).getOperations());
} catch (NotFoundException notFoundException) {
throw new IllegalStateException("Unable to find the error workflow definition '" + errorDefId + "'");
}
}
}
// Fail the current operation
currentOperation.setState(OperationState.FAILED);
}
/**
* Callback for workflow operation handlers that executed and finished without exception. This implementation assumes
* that the operation worker has already adjusted the current operation's state appropriately.
*
* @param workflow
* the workflow instance
* @param result
* the workflow operation result
* @return the workflow instance
* @throws WorkflowDatabaseException
* if updating the workflow fails
*/
protected WorkflowInstance handleOperationResult(WorkflowInstance workflow, WorkflowOperationResult result)
throws WorkflowDatabaseException {
// Get the operation and its handler
WorkflowOperationInstance currentOperation = workflow.getCurrentOperation();
WorkflowOperationHandler handler = getWorkflowOperationHandler(currentOperation.getTemplate());
// Create an operation result for the lazy or else update the workflow's media package
if (result == null) {
logger.warn("Handling a null operation result for workflow {} in operation {}", workflow.getId(),
currentOperation.getTemplate());
result = new WorkflowOperationResultImpl(workflow.getMediaPackage(), null, Action.CONTINUE, 0);
} else {
MediaPackage mp = result.getMediaPackage();
if (mp != null) {
workflow.setMediaPackage(mp);
}
}
// The action to take
Action action = result.getAction();
// Update the workflow configuration.
workflow = updateConfiguration(workflow, result.getProperties());
// Adjust workflow statistics
currentOperation.setTimeInQueue(result.getTimeInQueue());
// Adjust the operation state
switch (action) {
case CONTINUE:
currentOperation.setState(OperationState.SUCCEEDED);
break;
case PAUSE:
if (!(handler instanceof ResumableWorkflowOperationHandler)) {
throw new IllegalStateException("Operation " + currentOperation.getTemplate() + " is not resumable");
}
// Set abortable and continuable to default values
currentOperation.setContinuable(result.allowsContinue());
currentOperation.setAbortable(result.allowsAbort());
workflow.setState(PAUSED);
currentOperation.setState(OperationState.PAUSED);
break;
case SKIP:
currentOperation.setState(OperationState.SKIPPED);
break;
default:
throw new IllegalStateException("Unknown action '" + action + "' returned");
}
if (ERROR_RESOLUTION_HANDLER_ID.equals(currentOperation.getTemplate()) && result.getAction() == Action.CONTINUE) {
Map<String, String> resultProperties = result.getProperties();
if (resultProperties == null || StringUtils.isBlank(resultProperties.get(RETRY_STRATEGY)))
throw new WorkflowDatabaseException("Retry strategy not present in properties!");
RetryStrategy retryStrategy = RetryStrategy.valueOf(resultProperties.get(RETRY_STRATEGY));
switch (retryStrategy) {
case NONE:
handleFailedOperation(workflow, workflow.getCurrentOperation());
break;
case RETRY:
break;
default:
throw new WorkflowDatabaseException("Retry strategy not implemented yet!");
}
}
return workflow;
}
/**
* Reads the available metadata from the dublin core catalog (if there is one) and updates the mediapackage.
*
* @param mp
* the media package
*/
protected void populateMediaPackageMetadata(MediaPackage mp) {
if (metadataServices.isEmpty()) {
logger.warn("No metadata services are registered, so no media package metadata can be extracted from catalogs");
return;
}
for (MediaPackageMetadataService metadataService : metadataServices) {
MediaPackageMetadata metadata = metadataService.getMetadata(mp);
MediaPackageMetadataSupport.populateMediaPackageMetadata(mp, metadata);
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.job.api.JobProducer#isReadyToAcceptJobs(String)
*/
@Override
public boolean isReadyToAcceptJobs(String operation) {
return true;
}
/**
* {@inheritDoc}
*
* If we are already running the maximum number of workflows, don't accept another START_WORKFLOW job
*
* @see org.opencastproject.job.api.AbstractJobProducer#isReadyToAccept(org.opencastproject.job.api.Job)
*/
@Override
public boolean isReadyToAccept(Job job) throws UndispatchableJobException {
String operation = job.getOperation();
// Only restrict execution of new jobs
if (!Operation.START_WORKFLOW.toString().equals(operation))
return true;
// If the first operation is guaranteed to pause, run the job.
if (job.getArguments().size() > 1 && job.getArguments().get(0) != null) {
try {
WorkflowDefinition workflowDef = XmlWorkflowParser.parseWorkflowDefinition(job.getArguments().get(0));
if (workflowDef.getOperations().size() > 0) {
String firstOperationId = workflowDef.getOperations().get(0).getId();
WorkflowOperationHandler handler = getWorkflowOperationHandler(firstOperationId);
if (handler instanceof ResumableWorkflowOperationHandler) {
if (((ResumableWorkflowOperationHandler) handler).isAlwaysPause()) {
return true;
}
}
}
} catch (WorkflowParsingException e) {
throw new UndispatchableJobException(job + " is not a proper job to start a workflow", e);
}
}
WorkflowInstance workflow;
Optional<WorkflowInstance> workflowInstance;
String mediaPackageId;
// Fetch all workflows that are running with the current media package
try {
workflow = getWorkflowById(job.getId());
mediaPackageId = workflow.getMediaPackage().getIdentifier().toString();
} catch (NotFoundException e) {
throw new UndispatchableJobException("Trying to start workflow with job id " + job.getId()
+ " but no corresponding instance is available from the workflow service", e);
} catch (UnauthorizedException e) {
throw new UndispatchableJobException(
"Authorization denied while requesting to loading workflow instance. Job: " + job.getId(), e);
}
try {
workflowInstance = getRunningWorkflowInstanceByMediaPackage(
workflow.getMediaPackage().getIdentifier().toString(), Permissions.Action.READ.toString());
} catch (UnauthorizedException e) {
throw new UndispatchableJobException("Authorization denied while requesting to loading workflow instance " + workflow.getId(), e);
} catch (WorkflowDatabaseException e) {
throw new UndispatchableJobException("An database error occurred while checking if a workflow is already active "
+ "(job: " + job.getId() + ")", e);
} catch (WorkflowException e) {
// Avoid running multiple workflows with same media package id at the same time
delayWorkflow(workflow, mediaPackageId);
return false;
}
// Make sure we are not excluding ourselves
if (workflowInstance.isPresent() && workflow.getId() != workflowInstance.get().getId()) {
delayWorkflow(workflow, mediaPackageId);
return false;
}
return true;
}
private void delayWorkflow(WorkflowInstance workflow, String mediaPackageId) {
if (!delayedWorkflows.contains(workflow.getId())) {
logger.info("Delaying start of workflow {}, another workflow on media package {} is still running",
workflow.getId(), mediaPackageId);
delayedWorkflows.add(workflow.getId());
}
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.job.api.AbstractJobProducer#acceptJob(org.opencastproject.job.api.Job)
*/
@Override
public synchronized void acceptJob(Job job) throws ServiceRegistryException {
User originalUser = securityService.getUser();
Organization originalOrg = securityService.getOrganization();
try {
Organization organization = organizationDirectoryService.getOrganization(job.getOrganization());
securityService.setOrganization(organization);
User user = userDirectoryService.loadUser(job.getCreator());
securityService.setUser(user);
job.setStatus(Job.Status.RUNNING);
job = serviceRegistry.updateJob(job);
// Check if this workflow was initially delayed
if (delayedWorkflows.contains(job.getId())) {
delayedWorkflows.remove(job.getId());
logger.info("Starting initially delayed workflow {}, {} more waiting", job.getId(), delayedWorkflows.size());
}
executorService.submit(new JobRunner(job, serviceRegistry.getCurrentJob()));
} catch (Exception e) {
if (e instanceof ServiceRegistryException)
throw (ServiceRegistryException) e;
throw new ServiceRegistryException(e);
} finally {
securityService.setUser(originalUser);
securityService.setOrganization(originalOrg);
}
}
/**
* Processes the workflow job.
*
* @param job
* the job
* @return the job payload
* @throws Exception
* if job processing fails
*/
protected String process(Job job) throws Exception {
List<String> arguments = job.getArguments();
Operation op = null;
WorkflowInstance workflowInstance = null;
WorkflowOperationInstance wfo;
String operation = job.getOperation();
try {
try {
op = Operation.valueOf(operation);
switch (op) {
case START_WORKFLOW:
workflowInstance = persistence.getWorkflow(Long.parseLong(job.getPayload()));
logger.debug("Starting new workflow {}", workflowInstance);
runWorkflow(workflowInstance);
break;
case RESUME:
workflowInstance = getWorkflowById(Long.parseLong(arguments.get(0)));
Map<String, String> properties = null;
if (arguments.size() > 1) {
Properties props = new Properties();
props.load(IOUtils.toInputStream(arguments.get(arguments.size() - 1), StandardCharsets.UTF_8));
properties = new HashMap<>();
for (Entry<Object, Object> entry : props.entrySet()) {
properties.put(entry.getKey().toString(), entry.getValue().toString());
}
}
logger.debug("Resuming {} at {}", workflowInstance, workflowInstance.getCurrentOperation());
workflowInstance.setState(RUNNING);
update(workflowInstance);
runWorkflowOperation(workflowInstance, properties);
break;
case START_OPERATION:
workflowInstance = getWorkflowById(Long.parseLong(arguments.get(0)));
wfo = workflowInstance.getCurrentOperation();
if (OperationState.RUNNING.equals(wfo.getState()) || OperationState.PAUSED.equals(wfo.getState())) {
logger.info("Reset operation state {} {} to INSTANTIATED due to job restart", workflowInstance, wfo);
wfo.setState(OperationState.INSTANTIATED);
}
wfo.setExecutionHost(job.getProcessingHost());
logger.debug("Running {} {}", workflowInstance, wfo);
wfo = runWorkflowOperation(workflowInstance, null);
updateOperationJob(job.getId(), wfo.getState());
break;
default:
throw new IllegalStateException("Don't know how to handle operation '" + operation + "'");
}
} catch (IllegalArgumentException e) {
throw new ServiceRegistryException("This service can't handle operations of type '" + op + "'", e);
} catch (IndexOutOfBoundsException e) {
throw new ServiceRegistryException(
"The argument list for operation '" + op + "' (job: " + job.getId() + ") does not meet expectations", e);
} catch (NotFoundException e) {
logger.warn("Not found processing job {}", job, e);
updateOperationJob(job.getId(), OperationState.FAILED);
}
return null;
} catch (Exception e) {
logger.warn("Exception while accepting job {}", job, e);
try {
if (workflowInstance != null) {
logger.warn("Marking job {} and workflow instance {} as failed", job, workflowInstance);
updateOperationJob(job.getId(), OperationState.FAILED);
workflowInstance.setState(FAILED);
update(workflowInstance);
} else {
logger.warn("Unable to parse workflow instance", e);
}
} catch (WorkflowDatabaseException e1) {
throw new ServiceRegistryException(e1);
}
if (e instanceof ServiceRegistryException)
throw e;
throw new ServiceRegistryException("Error handling operation '" + op + "'", e);
}
}
/**
* Synchronizes the workflow operation's job with the operation status if the operation has a job associated with it,
* which is determined by looking at the operation's job id.
*
* @param state
* the operation state
* @param jobId
* the associated job
* @return the updated job or <code>null</code> if there is no job for this operation
* @throws ServiceRegistryException
* if the job can't be updated in the service registry
* @throws NotFoundException
* if the job can't be found
*/
private Job updateOperationJob(Long jobId, OperationState state) throws NotFoundException, ServiceRegistryException {
if (jobId == null)
return null;
Job job = serviceRegistry.getJob(jobId);
switch (state) {
case FAILED:
case RETRY:
job.setStatus(Status.FAILED);
break;
case PAUSED:
job.setStatus(Status.PAUSED);
job.setOperation(Operation.RESUME.toString());
break;
case SKIPPED:
case SUCCEEDED:
job.setStatus(Status.FINISHED);
break;
default:
throw new IllegalStateException("Unexpected state '" + state + "' found");
}
return serviceRegistry.updateJob(job);
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.job.api.JobProducer#countJobs(org.opencastproject.job.api.Job.Status)
*/
@Override
public long countJobs(Status status) throws ServiceRegistryException {
return serviceRegistry.count(JOB_TYPE, status);
}
/**
* Converts a Map<String, String> to s key=value\n string, suitable for the properties form parameter expected by the
* workflow rest endpoint.
*
* @param props
* The map of strings
* @return the string representation
*/
private String mapToString(Map<String, String> props) {
if (props == null)
return null;
StringBuilder sb = new StringBuilder();
for (Entry<String, String> entry : props.entrySet()) {
sb.append(entry.getKey());
sb.append("=");
sb.append(entry.getValue());
sb.append("\n");
}
return sb.toString();
}
/**
* Dummy callback for osgi
*
* @param unused
* the unused ReadinessIndicator
*/
@Reference(target = "(artifact=workflowdefinition)")
protected void setProfilesReadyIndicator(ReadinessIndicator unused) { }
/**
* Callback for the OSGi environment to register with the <code>Workspace</code>.
*
* @param workspace
* the workspace
*/
@Reference
protected void setWorkspace(Workspace workspace) {
this.workspace = workspace;
}
/**
* Callback for the OSGi environment to register with the <code>ServiceRegistry</code>.
*
* @param registry
* the service registry
*/
@Reference
protected void setServiceRegistry(ServiceRegistry registry) {
this.serviceRegistry = registry;
}
public ServiceRegistry getServiceRegistry() {
return serviceRegistry;
}
/**
* Callback for setting the security service.
*
* @param securityService
* the securityService to set
*/
@Reference
public void setSecurityService(SecurityService securityService) {
this.securityService = securityService;
}
/**
* Callback for setting the authorization service.
*
* @param authorizationService
* the authorizationService to set
*/
@Reference
public void setAuthorizationService(AuthorizationService authorizationService) {
this.authorizationService = authorizationService;
}
/**
* Callback for setting the user directory service
*
* @param userDirectoryService
* the userDirectoryService to set
*/
@Reference
public void setUserDirectoryService(UserDirectoryService userDirectoryService) {
this.userDirectoryService = userDirectoryService;
}
/**
* Sets a reference to the organization directory service.
*
* @param organizationDirectory
* the organization directory
*/
@Reference
public void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectory) {
this.organizationDirectoryService = organizationDirectory;
}
/**
* Sets the series service
*
* @param seriesService
* the seriesService to set
*/
@Reference
public void setSeriesService(SeriesService seriesService) {
this.seriesService = seriesService;
}
/**
* Sets the asset manager
*
* @param assetManager
* the assetManager to set
*/
@Reference
public void setAssetManager(AssetManager assetManager) {
this.assetManager = assetManager;
}
/**
* Callback to set the metadata service
*
* @param service
* the metadata service
*/
@Reference(cardinality = ReferenceCardinality.AT_LEAST_ONE, policy = ReferencePolicy.DYNAMIC, unbind = "removeMetadataService")
protected void addMetadataService(MediaPackageMetadataService service) {
metadataServices.add(service);
}
/**
* Callback to remove a mediapackage metadata service.
*
* @param service
* the mediapackage metadata service to remove
*/
protected void removeMetadataService(MediaPackageMetadataService service) {
metadataServices.remove(service);
}
/**
* Callback to set the workflow definition scanner
*
* @param scanner
* the workflow definition scanner
*/
@Reference
protected void addWorkflowDefinitionScanner(WorkflowDefinitionScanner scanner) {
workflowDefinitionScanner = scanner;
}
/**
* Callback to set the Admin UI index.
*
* @param index
* the admin UI index.
*/
@Reference
public void setIndex(ElasticsearchIndex index) {
this.index = index;
}
/**
* Callback to set the workflow database
*
* @param persistence
* the workflow database
*/
@Reference(name = "workflow-persistence")
public void setPersistence(WorkflowServiceDatabase persistence) {
this.persistence = persistence;
}
/**
* {@inheritDoc}
*
* @see org.opencastproject.job.api.JobProducer#getJobType()
*/
@Override
public String getJobType() {
return JOB_TYPE;
}
/**
* A tuple of a workflow operation handler and the name of the operation it handles
*/
public static class HandlerRegistration {
protected WorkflowOperationHandler handler;
protected String operationName;
public HandlerRegistration(String operationName, WorkflowOperationHandler handler) {
if (operationName == null)
throw new IllegalArgumentException("Operation name cannot be null");
if (handler == null)
throw new IllegalArgumentException("Handler cannot be null");
this.operationName = operationName;
this.handler = handler;
}
public WorkflowOperationHandler getHandler() {
return handler;
}
/**
* {@inheritDoc}
*
* @see java.lang.Object#hashCode()
*/
@Override
public int hashCode() {
final int prime = 31;
int result = 1;
result = prime * result + handler.hashCode();
result = prime * result + operationName.hashCode();
return result;
}
/**
* {@inheritDoc}
*
* @see java.lang.Object#equals(java.lang.Object)
*/
@Override
public boolean equals(Object obj) {
if (this == obj)
return true;
if (obj == null)
return false;
if (getClass() != obj.getClass())
return false;
HandlerRegistration other = (HandlerRegistration) obj;
if (!handler.equals(other.handler))
return false;
return operationName.equals(other.operationName);
}
}
/**
* A utility class to run jobs
*/
class JobRunner implements Callable<Void> {
/** The job */
private Job job = null;
/** The current job */
private final Job currentJob;
/**
* Constructs a new job runner
*
* @param job
* the job to run
* @param currentJob
* the current running job
*/
JobRunner(Job job, Job currentJob) {
this.job = job;
this.currentJob = currentJob;
}
/**
* {@inheritDoc}
*
* @see java.util.concurrent.Callable#call()
*/
@Override
public Void call() throws Exception {
Organization jobOrganization = organizationDirectoryService.getOrganization(job.getOrganization());
try {
serviceRegistry.setCurrentJob(currentJob);
securityService.setOrganization(jobOrganization);
User jobUser = userDirectoryService.loadUser(job.getCreator());
securityService.setUser(jobUser);
process(job);
} finally {
serviceRegistry.setCurrentJob(null);
securityService.setUser(null);
securityService.setOrganization(null);
}
return null;
}
}
@Override
public synchronized void cleanupWorkflowInstances(int buffer, WorkflowState state) throws UnauthorizedException,
WorkflowDatabaseException {
logger.info("Start cleaning up workflow instances older than {} days with status '{}'", buffer, state);
int instancesCleaned = 0;
int cleaningFailed = 0;
Date priorTo = DateUtils.addDays(new Date(), -buffer);
try {
for (WorkflowInstance workflowInstance : persistence.getWorkflowInstancesForCleanup(state, priorTo)) {
try {
logger.debug("Deleting workflow instance {}", workflowInstance.getId());
remove(workflowInstance.getId());
instancesCleaned++;
} catch (WorkflowDatabaseException | UnauthorizedException e) {
throw e;
} catch (NotFoundException e) {
// Since we are in a cleanup operation, we don't have to care about NotFoundExceptions
logger.debug("Workflow instance '{}' could not be removed", workflowInstance.getId(), e);
} catch (WorkflowParsingException | WorkflowStateException e) {
logger.warn("Workflow instance '{}' could not be removed", workflowInstance.getId(), e);
cleaningFailed++;
}
}
} catch (WorkflowDatabaseException e) {
throw new WorkflowDatabaseException(e);
}
if (instancesCleaned == 0 && cleaningFailed == 0) {
logger.info("No workflow instances found to clean up");
return;
}
if (instancesCleaned > 0)
logger.info("Cleaned up '{}' workflow instances", instancesCleaned);
if (cleaningFailed > 0) {
logger.warn("Cleaning failed for '{}' workflow instances", cleaningFailed);
throw new WorkflowDatabaseException("Unable to clean all workflow instances, see logs!");
}
}
@Override
public Map<String, Map<String, String>> getWorkflowStateMappings() {
return workflowDefinitionScanner.workflowStateMappings.entrySet().stream().collect(Collectors.toMap(
Entry::getKey, e -> e.getValue().stream()
.collect(Collectors.toMap(m -> m.getState().name(), WorkflowStateMapping::getValue))
));
}
@Override
public void repopulate(IndexRebuildService.DataType type) throws IndexRebuildException {
try {
final int total;
try {
total = persistence.countMediaPackages();
} catch (WorkflowDatabaseException e) {
logIndexRebuildError(logger, e);
throw new IndexRebuildException(getService(), e);
}
if (total > 0) {
logIndexRebuildBegin(logger, total, "workflows");
int current = 0;
int n = 20;
List<WorkflowIndexData> workflowIndexData;
int limit = 1000;
int offset = 0;
String currentMediapackageId;
String lastMediapackageId = "";
var updatedWorkflowRange = new ArrayList<Event>();
do {
try {
workflowIndexData = persistence.getWorkflowIndexData(limit, offset);
} catch (WorkflowDatabaseException e) {
logIndexRebuildError(logger, e);
throw new IndexRebuildException(getService(), e);
}
if (!workflowIndexData.isEmpty()) {
offset += limit;
logger.debug("Got {} workflows for re-indexing", workflowIndexData.size());
for (WorkflowIndexData indexData : workflowIndexData) {
currentMediapackageId = indexData.getMediaPackageId();
if (currentMediapackageId.equals(lastMediapackageId)) {
continue;
}
current++;
// Include PAUSED; otherwise, paused workflows will show up as "Finished"
if (!WorkflowUtil.isActive(WorkflowInstance.WorkflowState.values()[indexData.getState()].toString())
|| WorkflowState.PAUSED == WorkflowInstance.WorkflowState.values()[indexData.getState()]) {
String orgid = indexData.getOrganizationId();
if (null == orgid) {
String mpId = indexData.getMediaPackageId();
//We're assuming here that mediapackages don't change orgs
RichAResult results = assetManager.getSnapshotsById(mpId);
if (results.getSize() == 0) {
logger.debug("Dropping {} from the index since it is missing from the database", mpId);
continue;
}
orgid = results.getSnapshots().stream().findFirst().get().getOrganizationId();
//We try-catch here since it's possible for the WF to exist in the *index* but not in the *DB*
// It probably shouldn't be, but that won't keep it from happening anyway.
try {
//NB: This version of getWorkflow takes the org id, which in this case is null
// Using the normal version filters by org, and since this workflow has a NULL org it can't be found
WorkflowInstance instance = persistence.getWorkflow(indexData.getId(), null);
instance.setOrganizationId(orgid);
persistence.updateInDatabase(instance);
} catch (NotFoundException e) {
//Technically this should never happen, but getWorkflow throws it.
}
}
var updatedWorkflowData = index.getEvent(indexData.getMediaPackageId(), orgid, securityService.getUser());
updatedWorkflowData = getStateUpdateFunction(indexData).apply(updatedWorkflowData);
updatedWorkflowRange.add(updatedWorkflowData.get());
if (updatedWorkflowRange.size() >= n || current >= total) {
index.bulkEventUpdate(updatedWorkflowRange);
logIndexRebuildProgress(logger, total, current, n);
updatedWorkflowRange.clear();
}
}
else {
logger.info("Skipping. Workflow {} is currently active.", indexData.getId());
}
lastMediapackageId = currentMediapackageId;
}
}
} while (!workflowIndexData.isEmpty());
}
} catch (Exception e) {
logIndexRebuildError(logger, e);
throw new IndexRebuildException(getService(), e);
}
}
@Override
public IndexRebuildService.Service getService() {
return IndexRebuildService.Service.Workflow;
}
/**
* Remove a workflow instance from the Elasticsearch index.
*
* @param workflowInstanceId
* the identifier of the workflow instance to remove
* @param index
* the index to update
*/
private void removeWorkflowInstanceFromIndex(long workflowInstanceId) {
final String orgId = securityService.getOrganization().getId();
final User user = securityService.getUser();
// find events
SearchResult<Event> results;
try {
results = index.getByQuery(new EventSearchQuery(orgId, user).withWorkflowId(workflowInstanceId));
} catch (SearchIndexException e) {
logger.error("Error retrieving the events for workflow instance {} from the {} index.", workflowInstanceId,
index.getIndexName(), e);
return;
}
if (results.getItems().length == 0) {
logger.warn("No events for workflow instance {} found in the {} index.", workflowInstanceId,
index.getIndexName());
return;
}
// should be only one event, but better safe than sorry
for (SearchResultItem<Event> item: results.getItems()) {
String eventId = item.getSource().getIdentifier();
logger.debug("Removing workflow instance {} of event {} from the {} index.", workflowInstanceId, eventId,
index.getIndexName());
Function<Optional<Event>, Optional<Event>> updateFunction = (Optional<Event> eventOpt) -> {
if (!eventOpt.isPresent()) {
logger.warn("Event {} of workflow instance {} not found in the {} index.", workflowInstanceId,
eventId, index.getIndexName());
return Optional.empty();
}
Event event = eventOpt.get();
if (event.getWorkflowId() != null && event.getWorkflowId().equals(workflowInstanceId)) {
logger.debug("Workflow {} is the current workflow of event {}. Removing it from event.", eventId,
workflowInstanceId);
event.setWorkflowId(null);
event.setWorkflowDefinitionId(null);
event.setWorkflowState(null);
return Optional.of(event);
}
return Optional.empty();
};
try {
index.addOrUpdateEvent(eventId, updateFunction, orgId, user);
logger.debug("Workflow instance {} of event {} removed from the {} index.", workflowInstanceId, eventId,
index.getIndexName());
} catch (SearchIndexException e) {
logger.error("Error removing the workflow instance {} of event {} from the {} index.", workflowInstanceId,
eventId, index.getIndexName(), e);
}
}
}
/**
* Update a workflow instance in the Elasticsearch index.
*
* @param id
* workflow id
* @param state
* workflow state as int
* @param mpId
* corresponding mediapackage id
* @param orgId
* workflow organization id
*/
private void updateWorkflowInstanceInIndex(long id, int state, String wfDefId, String mpId, String orgId) {
final WorkflowState workflowState = WorkflowState.values()[state];
final User user = securityService.getUser();
logger.debug("Updating workflow instance {} of event {} in the {} index.", id, mpId,
index.getIndexName());
Function<Optional<Event>, Optional<Event>> updateFunction = (Optional<Event> eventOpt) -> {
Event event = eventOpt.orElse(new Event(mpId, orgId));
event.setWorkflowId(id);
event.setWorkflowState(workflowState);
event.setWorkflowDefinitionId(wfDefId);
return Optional.of(event);
};
try {
index.addOrUpdateEvent(mpId, updateFunction, orgId, user);
logger.debug("Workflow instance {} of event {} updated in the {} index.", id, mpId,
index.getIndexName());
} catch (SearchIndexException e) {
logger.error("Error updating the workflow instance {} of event {} in the {} index.", id, mpId,
index.getIndexName(), e);
}
}
/**
* Get the function to update the workflow state for an event in the Elasticsearch index.
*
* @param wfData
* The workflow index data package
* @return the function to do the update
*/
private Function<Optional<Event>, Optional<Event>> getStateUpdateFunction(WorkflowIndexData wfData) {
return (Optional<Event> eventOpt) -> {
Event event = eventOpt.orElse(new Event(wfData.getMediaPackageId(), wfData.getOrganizationId()));
event.setWorkflowId(wfData.getId());
event.setWorkflowState(WorkflowState.values()[wfData.getState()]);
return Optional.of(event);
};
}
}