2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright (C) 2017 - 2019 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 package org.onap.vid.job.impl;
23 import static org.onap.vid.job.Job.JobStatus.FAILED;
24 import static org.onap.vid.job.Job.JobStatus.STOPPED;
25 import static org.onap.vid.job.command.ResourceCommandKt.ACTION_PHASE;
26 import static org.onap.vid.job.command.ResourceCommandKt.INTERNAL_STATE;
28 import java.util.Optional;
29 import java.util.UUID;
30 import org.apache.commons.lang3.StringUtils;
31 import org.apache.commons.lang3.exception.ExceptionUtils;
32 import org.onap.logging.ref.slf4j.ONAPLogConstants;
33 import org.onap.portalsdk.core.logging.logic.EELFLoggerDelegate;
34 import org.onap.vid.job.Job;
35 import org.onap.vid.job.JobCommand;
36 import org.onap.vid.job.JobException;
37 import org.onap.vid.job.JobsBrokerService;
38 import org.onap.vid.job.NextCommand;
39 import org.onap.vid.job.command.JobCommandFactory;
40 import org.quartz.JobExecutionContext;
42 import org.springframework.scheduling.quartz.QuartzJobBean;
43 import org.springframework.stereotype.Component;
46 public class JobWorker extends QuartzJobBean {
48 private static final EELFLoggerDelegate LOGGER = EELFLoggerDelegate.getLogger(JobWorker.class);
50 private JobsBrokerService jobsBrokerService;
51 private JobCommandFactory jobCommandFactory;
52 private Job.JobStatus topic;
55 protected void executeInternal(JobExecutionContext context) {
60 while (job.isPresent()) {
61 Job nextJob = executeJobAndGetNext(job.get());
68 private Optional<Job> pullJob() {
70 return jobsBrokerService.pull(topic, UUID.randomUUID().toString());
71 } catch (Exception e) {
72 LOGGER.error("failed to pull job from queue, breaking: {}", e, e);
73 tryMutingJobFromException(e);
75 return Optional.empty();
79 private void pushBack(Job nextJob) {
81 jobsBrokerService.pushBack(nextJob);
82 } catch (Exception e) {
83 LOGGER.error("failed pushing back job to queue: {}", e, e);
87 protected Job executeJobAndGetNext(Job job) {
88 setThreadWithRandomRequestId();
90 Object internalState = job.getData().get(INTERNAL_STATE);
91 Object actionPhase = job.getData().get(ACTION_PHASE);
92 LOGGER.debug(EELFLoggerDelegate.debugLogger, "going to execute job {} of {}: {}/{} {}/{}",
93 StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
94 StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
101 NextCommand nextCommand = executeCommandAndGetNext(job);
103 return setNextCommandInJob(nextCommand, job);
106 private void setThreadWithRandomRequestId() {
107 //make sure requestIds in outgoing requests are not reused
108 MDC.put(ONAPLogConstants.MDCs.REQUEST_ID, UUID.randomUUID().toString());
111 private NextCommand executeCommandAndGetNext(Job job) {
112 NextCommand nextCommand;
114 final JobCommand jobCommand = jobCommandFactory.toCommand(job);
115 nextCommand = jobCommand.call();
116 } catch (Exception e) {
117 LOGGER.error("error while executing job from queue: {}", e, e);
118 nextCommand = new NextCommand(FAILED);
121 if (nextCommand == null) {
122 nextCommand = new NextCommand(STOPPED);
127 private Job setNextCommandInJob(NextCommand nextCommand, Job job) {
128 LOGGER.debug(EELFLoggerDelegate.debugLogger, "transforming job {} of {}: {}/{} -> {}{}",
129 StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
130 StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
131 job.getStatus(), job.getType(),
132 nextCommand.getStatus(),
133 nextCommand.getCommand() != null ? ("/" + nextCommand.getCommand().getType()) : "");
135 job.setStatus(nextCommand.getStatus());
137 if (nextCommand.getCommand() != null) {
138 job.setTypeAndData(nextCommand.getCommand().getType(), nextCommand.getCommand().getData());
144 private void tryMutingJobFromException(Exception e) {
145 // If there's JobException in the stack, read job uuid from
146 // the exception, and mute it in DB.
147 final int indexOfJobException =
148 ExceptionUtils.indexOfThrowable(e, JobException.class);
150 if (indexOfJobException >= 0) {
152 final JobException jobException = (JobException) ExceptionUtils.getThrowableList(e).get(indexOfJobException);
153 LOGGER.info(EELFLoggerDelegate.debugLogger, "muting job: {} ({})", jobException.getJobUuid(), jobException.toString());
154 final boolean success = jobsBrokerService.mute(jobException.getJobUuid());
156 LOGGER.error("failed to mute job {}", jobException.getJobUuid());
158 } catch (Exception e1) {
159 LOGGER.error("failed to mute job: {}", e1, e1);
164 //used by quartz to inject JobsBrokerService into the job
165 //see JobSchedulerInitializer
166 public void setJobsBrokerService(JobsBrokerService jobsBrokerService) {
167 this.jobsBrokerService = jobsBrokerService;
170 public void setJobCommandFactory(JobCommandFactory jobCommandFactory) {
171 this.jobCommandFactory = jobCommandFactory;
174 public void setTopic(Job.JobStatus topic) {