001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements. See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership. The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License. You may obtain a copy of the License at
009 *
010 * http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018 package org.apache.oozie.command.coord;
019
020 import java.util.Date;
021
022 import org.apache.oozie.CoordinatorActionBean;
023 import org.apache.oozie.ErrorCode;
024 import org.apache.oozie.WorkflowJobBean;
025 import org.apache.oozie.XException;
026 import org.apache.oozie.service.JPAService;
027 import org.apache.oozie.service.Services;
028 import org.apache.oozie.util.LogUtils;
029 import org.apache.oozie.util.db.SLADbOperations;
030 import org.apache.oozie.client.CoordinatorAction;
031 import org.apache.oozie.client.WorkflowJob;
032 import org.apache.oozie.client.SLAEvent.SlaAppType;
033 import org.apache.oozie.client.SLAEvent.Status;
034 import org.apache.oozie.command.CommandException;
035 import org.apache.oozie.command.PreconditionException;
036 import org.apache.oozie.executor.jpa.CoordActionGetForExternalIdJPAExecutor;
037 import org.apache.oozie.executor.jpa.CoordActionUpdateJPAExecutor;
038 import org.apache.oozie.executor.jpa.JPAExecutorException;
039
040 public class CoordActionUpdateXCommand extends CoordinatorXCommand<Void> {
041 private WorkflowJobBean workflow;
042 private CoordinatorActionBean coordAction = null;
043 private JPAService jpaService = null;
044 private int maxRetries = 1;
045
046 public CoordActionUpdateXCommand(WorkflowJobBean workflow) {
047 super("coord-action-update", "coord-action-update", 1);
048 this.workflow = workflow;
049 }
050
051 public CoordActionUpdateXCommand(WorkflowJobBean workflow, int maxRetries) {
052 super("coord-action-update", "coord-action-update", 1);
053 this.workflow = workflow;
054 this.maxRetries = maxRetries;
055 }
056
057 @Override
058 protected Void execute() throws CommandException {
059 try {
060 LOG.debug("STARTED CoordActionUpdateXCommand for wfId=" + workflow.getId());
061
062 Status slaStatus = null;
063 CoordinatorAction.Status preCoordStatus = coordAction.getStatus();
064 if (workflow.getStatus() == WorkflowJob.Status.SUCCEEDED) {
065 coordAction.setStatus(CoordinatorAction.Status.SUCCEEDED);
066 coordAction.setPending(0);
067 slaStatus = Status.SUCCEEDED;
068 }
069 else if (workflow.getStatus() == WorkflowJob.Status.FAILED) {
070 coordAction.setStatus(CoordinatorAction.Status.FAILED);
071 coordAction.setPending(0);
072 slaStatus = Status.FAILED;
073 }
074 else if (workflow.getStatus() == WorkflowJob.Status.KILLED) {
075 coordAction.setStatus(CoordinatorAction.Status.KILLED);
076 coordAction.setPending(0);
077 slaStatus = Status.KILLED;
078 }
079 else if (workflow.getStatus() == WorkflowJob.Status.SUSPENDED) {
080 coordAction.setStatus(CoordinatorAction.Status.SUSPENDED);
081 coordAction.decrementAndGetPending();
082 }
083 else if (workflow.getStatus() == WorkflowJob.Status.RUNNING) {
084 // resume workflow job and update coord action accordingly
085 coordAction.setStatus(CoordinatorAction.Status.RUNNING);
086 coordAction.decrementAndGetPending();
087 }
088 else {
089 LOG.warn("Unexpected workflow " + workflow.getId() + " STATUS " + workflow.getStatus());
090 // update lastModifiedTime
091 coordAction.setLastModifiedTime(new Date());
092 jpaService.execute(new CoordActionUpdateJPAExecutor(coordAction));
093
094 return null;
095 }
096
097 LOG.info("Updating Coordintaor action id :" + coordAction.getId() + " status from " + preCoordStatus
098 + " to " + coordAction.getStatus() + ", pending = " + coordAction.getPending());
099
100 coordAction.setLastModifiedTime(new Date());
101 jpaService.execute(new CoordActionUpdateJPAExecutor(coordAction));
102 if (slaStatus != null) {
103 SLADbOperations.writeStausEvent(coordAction.getSlaXml(), coordAction.getId(), slaStatus,
104 SlaAppType.COORDINATOR_ACTION, LOG);
105 }
106 if (workflow.getStatus() != WorkflowJob.Status.SUSPENDED
107 && workflow.getStatus() != WorkflowJob.Status.RUNNING) {
108 queue(new CoordActionReadyXCommand(coordAction.getJobId()));
109 }
110 LOG.debug("ENDED CoordActionUpdateXCommand for wfId=" + workflow.getId());
111 }
112 catch (XException ex) {
113 LOG.warn("CoordActionUpdate Failed ", ex.getMessage());
114 throw new CommandException(ex);
115 }
116 return null;
117 }
118
119 /* (non-Javadoc)
120 * @see org.apache.oozie.command.XCommand#getEntityKey()
121 */
122 @Override
123 protected String getEntityKey() {
124 return coordAction.getJobId();
125 }
126
127 /* (non-Javadoc)
128 * @see org.apache.oozie.command.XCommand#isLockRequired()
129 */
130 @Override
131 protected boolean isLockRequired() {
132 return true;
133 }
134
135 /* (non-Javadoc)
136 * @see org.apache.oozie.command.XCommand#eagerLoadState()
137 */
138 @Override
139 protected void eagerLoadState() throws CommandException {
140 jpaService = Services.get().get(JPAService.class);
141 if (jpaService == null) {
142 throw new CommandException(ErrorCode.E0610);
143 }
144
145 int retries = 0;
146 while (retries++ < maxRetries) {
147 try {
148 coordAction = jpaService.execute(new CoordActionGetForExternalIdJPAExecutor(workflow.getId()));
149 if (coordAction != null) {
150 break;
151 }
152
153 if (retries < maxRetries) {
154 Thread.sleep(500);
155 }
156 }
157 catch (JPAExecutorException je) {
158 LOG.warn("Could not load coord action {0}", je.getMessage(), je);
159 }
160 catch (InterruptedException ex) {
161 LOG.warn("Retry to load coord action is interrupted {0}", ex.getMessage(), ex);
162 }
163 }
164
165 if (coordAction != null) {
166 LogUtils.setLogInfo(coordAction, logInfo);
167 }
168 }
169
170 /* (non-Javadoc)
171 * @see org.apache.oozie.command.XCommand#eagerVerifyPrecondition()
172 */
173 @Override
174 protected void eagerVerifyPrecondition() throws CommandException, PreconditionException {
175 if (coordAction == null) {
176 throw new PreconditionException(ErrorCode.E1100, ", coord action is null");
177 }
178 }
179
180 /* (non-Javadoc)
181 * @see org.apache.oozie.command.XCommand#loadState()
182 */
183 @Override
184 protected void loadState() throws CommandException {
185 }
186
187 /* (non-Javadoc)
188 * @see org.apache.oozie.command.XCommand#verifyPrecondition()
189 */
190 @Override
191 protected void verifyPrecondition() throws CommandException, PreconditionException {
192
193 // if coord action is RUNNING and pending false and workflow is RUNNING, this doesn't need to be updated.
194 if (workflow.getStatus() == WorkflowJob.Status.RUNNING
195 && coordAction.getStatus() == CoordinatorAction.Status.RUNNING && !coordAction.isPending()) {
196 // update lastModifiedTime
197 coordAction.setLastModifiedTime(new Date());
198 try {
199 jpaService.execute(new org.apache.oozie.executor.jpa.CoordActionUpdateJPAExecutor(coordAction));
200 }
201 catch (JPAExecutorException je) {
202 throw new CommandException(je);
203 }
204 throw new PreconditionException(ErrorCode.E1100, ", workflow is RUNNING and coordinator action is RUNNING and pending false");
205 }
206 }
207
208 }