vid-app-common/src/main/java/org/onap/vid/job/impl/JobWorker.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * VID
   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
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  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=========================================================
  19  */
  20
  21 package org.onap.vid.job.impl;
  22
  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;
  27
  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;
  41 import org.slf4j.MDC;
  42 import org.springframework.scheduling.quartz.QuartzJobBean;
  43 import org.springframework.stereotype.Component;
  44
  45 @Component
  46 public class JobWorker extends QuartzJobBean {
  47
  48     private static final EELFLoggerDelegate LOGGER = EELFLoggerDelegate.getLogger(JobWorker.class);
  49
  50     private JobsBrokerService jobsBrokerService;
  51     private JobCommandFactory jobCommandFactory;
  52     private Job.JobStatus topic;
  53
  54     @Override
  55     protected void executeInternal(JobExecutionContext context) {
  56         Optional<Job> job;
  57
  58         job = pullJob();
  59
  60         while (job.isPresent()) {
  61             Job nextJob = executeJobAndGetNext(job.get());
  62             pushBack(nextJob);
  63
  64             job = pullJob();
  65         }
  66     }
  67
  68     private Optional<Job> pullJob() {
  69         try {
  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);
  74
  75             return Optional.empty();
  76         }
  77     }
  78
  79     private void pushBack(Job nextJob) {
  80         try {
  81             jobsBrokerService.pushBack(nextJob);
  82         } catch (Exception e) {
  83             LOGGER.error("failed pushing back job to queue: {}", e, e);
  84         }
  85     }
  86
  87     protected Job executeJobAndGetNext(Job job) {
  88         setThreadWithRandomRequestId();
  89
  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),
  95                 job.getStatus(),
  96                 job.getType(),
  97                 actionPhase,
  98                 internalState
  99                 );
 100
 101         NextCommand nextCommand = executeCommandAndGetNext(job);
 102
 103         return setNextCommandInJob(nextCommand, job);
 104     }
 105
 106     private void setThreadWithRandomRequestId() {
 107         //make sure requestIds in outgoing requests are not reused
 108         MDC.put(ONAPLogConstants.MDCs.REQUEST_ID, UUID.randomUUID().toString());
 109     }
 110
 111     private NextCommand executeCommandAndGetNext(Job job) {
 112         NextCommand nextCommand;
 113         try {
 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);
 119         }
 120
 121         if (nextCommand == null) {
 122             nextCommand = new NextCommand(STOPPED);
 123         }
 124         return nextCommand;
 125     }
 126
 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()) : "");
 134
 135         job.setStatus(nextCommand.getStatus());
 136
 137         if (nextCommand.getCommand() != null) {
 138             job.setTypeAndData(nextCommand.getCommand().getType(), nextCommand.getCommand().getData());
 139         }
 140
 141         return job;
 142     }
 143
 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);
 149
 150         if (indexOfJobException >= 0) {
 151             try {
 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());
 155                 if (!success) {
 156                     LOGGER.error("failed to mute job {}", jobException.getJobUuid());
 157                 }
 158             } catch (Exception e1) {
 159                 LOGGER.error("failed to mute job: {}", e1, e1);
 160             }
 161         }
 162     }
 163
 164     //used by quartz to inject JobsBrokerService into the job
 165     //see JobSchedulerInitializer
 166     public void setJobsBrokerService(JobsBrokerService jobsBrokerService) {
 167         this.jobsBrokerService = jobsBrokerService;
 168     }
 169
 170     public void setJobCommandFactory(JobCommandFactory jobCommandFactory) {
 171         this.jobCommandFactory = jobCommandFactory;
 172     }
 173
 174     public void setTopic(Job.JobStatus topic) {
 175         this.topic = topic;
 176     }
 177 }