1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
package org.onap.vid.job.impl;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.onap.portalsdk.core.logging.logic.EELFLoggerDelegate;
import org.onap.vid.job.*;
import org.onap.vid.job.command.JobCommandFactory;
import org.onap.vid.properties.Features;
import org.quartz.JobExecutionContext;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import org.togglz.core.manager.FeatureManager;
import java.util.Optional;
import java.util.UUID;
import static org.onap.vid.job.Job.JobStatus.FAILED;
import static org.onap.vid.job.Job.JobStatus.STOPPED;
@Component
public class JobWorker extends QuartzJobBean {
private static final EELFLoggerDelegate LOGGER = EELFLoggerDelegate.getLogger(JobWorker.class);
private JobsBrokerService jobsBrokerService;
private FeatureManager featureManager;
private JobCommandFactory jobCommandFactory;
private Job.JobStatus topic;
@Override
protected void executeInternal(JobExecutionContext context) {
Optional<Job> job;
if (!isMsoNewApiActive()) {
return;
}
job = pullJob();
while (job.isPresent()) {
Job nextJob = executeJobAndGetNext(job.get());
pushBack(nextJob);
job = pullJob();
}
}
private Optional<Job> pullJob() {
try {
return jobsBrokerService.pull(topic, UUID.randomUUID().toString());
} catch (Exception e) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed to pull job from queue, breaking: {}", e, e);
tryMutingJobFromException(e);
return Optional.empty();
}
}
private void pushBack(Job nextJob) {
try {
jobsBrokerService.pushBack(nextJob);
} catch (Exception e) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed pushing back job to queue: {}", e, e);
}
}
protected Job executeJobAndGetNext(Job job) {
LOGGER.debug(EELFLoggerDelegate.debugLogger, "going to execute job {} of {}: {}/{}",
StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
job.getStatus(), job.getType());
NextCommand nextCommand = executeCommandAndGetNext(job);
return setNextCommandInJob(nextCommand, job);
}
private NextCommand executeCommandAndGetNext(Job job) {
NextCommand nextCommand;
try {
final JobCommand jobCommand = jobCommandFactory.toCommand(job);
nextCommand = jobCommand.call();
} catch (Exception e) {
LOGGER.error("error while executing job from queue: {}", e);
nextCommand = new NextCommand(FAILED);
}
if (nextCommand == null) {
nextCommand = new NextCommand(STOPPED);
}
return nextCommand;
}
private Job setNextCommandInJob(NextCommand nextCommand, Job job) {
LOGGER.debug(EELFLoggerDelegate.debugLogger, "transforming job {} of {}: {}/{} -> {}{}",
StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
job.getStatus(), job.getType(),
nextCommand.getStatus(),
nextCommand.getCommand() != null ? ("/" + nextCommand.getCommand().getType()) : "");
job.setStatus(nextCommand.getStatus());
if (nextCommand.getCommand() != null) {
job.setTypeAndData(nextCommand.getCommand().getType(), nextCommand.getCommand().getData());
}
return job;
}
private boolean isMsoNewApiActive() {
return featureManager.isActive(Features.FLAG_ASYNC_INSTANTIATION);
}
private void tryMutingJobFromException(Exception e) {
// If there's JobException in the stack, read job uuid from
// the exception, and mute it in DB.
final int indexOfJobException =
ExceptionUtils.indexOfThrowable(e, JobException.class);
if (indexOfJobException >= 0) {
try {
final JobException jobException = (JobException) ExceptionUtils.getThrowableList(e).get(indexOfJobException);
LOGGER.info(EELFLoggerDelegate.debugLogger, "muting job: {} ({})", jobException.getJobUuid(), jobException.toString());
final boolean success = jobsBrokerService.mute(jobException.getJobUuid());
if (!success) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed to mute job {}", jobException.getJobUuid());
}
} catch (Exception e1) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed to mute job: {}", e1, e1);
}
}
}
//used by quartz to inject JobsBrokerService into the job
//see JobSchedulerInitializer
public void setJobsBrokerService(JobsBrokerService jobsBrokerService) {
this.jobsBrokerService = jobsBrokerService;
}
public void setFeatureManager(FeatureManager featureManager) {
this.featureManager = featureManager;
}
public void setJobCommandFactory(JobCommandFactory jobCommandFactory) {
this.jobCommandFactory = jobCommandFactory;
}
public void setTopic(Job.JobStatus topic) {
this.topic = topic;
}
}
|
