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 019package org.apache.oozie.store; 020 021import java.sql.Connection; 022import java.sql.SQLException; 023import java.sql.Timestamp; 024import java.util.ArrayList; 025import java.util.Date; 026import java.util.List; 027import java.util.Map; 028import java.util.concurrent.Callable; 029 030import javax.persistence.EntityManager; 031import javax.persistence.Query; 032 033import org.apache.oozie.ErrorCode; 034import org.apache.oozie.WorkflowActionBean; 035import org.apache.oozie.WorkflowJobBean; 036import org.apache.oozie.WorkflowsInfo; 037import org.apache.oozie.client.OozieClient; 038import org.apache.oozie.client.WorkflowJob.Status; 039import org.apache.oozie.executor.jpa.JPAExecutorException; 040import org.apache.oozie.executor.jpa.WorkflowActionQueryExecutor; 041import org.apache.oozie.executor.jpa.WorkflowJobQueryExecutor; 042import org.apache.oozie.service.InstrumentationService; 043import org.apache.oozie.service.SchemaService; 044import org.apache.oozie.service.Services; 045import org.apache.oozie.service.SchemaService.SchemaName; 046import org.apache.oozie.util.Instrumentation; 047import org.apache.oozie.util.ParamChecker; 048import org.apache.oozie.util.XLog; 049import org.apache.oozie.workflow.WorkflowException; 050import org.apache.openjpa.persistence.OpenJPAEntityManager; 051import org.apache.openjpa.persistence.OpenJPAPersistence; 052import org.apache.openjpa.persistence.OpenJPAQuery; 053import org.apache.openjpa.persistence.jdbc.FetchDirection; 054import org.apache.openjpa.persistence.jdbc.JDBCFetchPlan; 055import org.apache.openjpa.persistence.jdbc.LRSSizeAlgorithm; 056import org.apache.openjpa.persistence.jdbc.ResultSetType; 057 058/** 059 * DB Implementation of Workflow Store 060 */ 061public class WorkflowStore extends Store { 062 private Connection conn; 063 private EntityManager entityManager; 064 private boolean selectForUpdate; 065 private static final String INSTR_GROUP = "db"; 066 public static final int LOCK_TIMEOUT = 50000; 067 private static final String seletStr = "Select w.id, w.appName, w.statusStr, w.run, w.user, w.group, w.createdTimestamp, " 068 + "w.startTimestamp, w.lastModifiedTimestamp, w.endTimestamp from WorkflowJobBean w"; 069 private static final String countStr = "Select count(w) from WorkflowJobBean w"; 070 071 public WorkflowStore() { 072 } 073 074 public WorkflowStore(Connection connection, boolean selectForUpdate) throws StoreException { 075 super(); 076 conn = ParamChecker.notNull(connection, "conn"); 077 entityManager = getEntityManager(); 078 this.selectForUpdate = selectForUpdate; 079 } 080 081 public WorkflowStore(Connection connection, Store store, boolean selectForUpdate) throws StoreException { 082 super(store); 083 conn = ParamChecker.notNull(connection, "conn"); 084 entityManager = getEntityManager(); 085 this.selectForUpdate = selectForUpdate; 086 } 087 088 public WorkflowStore(boolean selectForUpdate) throws StoreException { 089 super(); 090 entityManager = getEntityManager(); 091 javax.xml.validation.Schema schema = Services.get().get(SchemaService.class).getSchema(SchemaName.WORKFLOW); 092 OpenJPAEntityManager kem = OpenJPAPersistence.cast(entityManager); 093 conn = (Connection) kem.getConnection(); 094 this.selectForUpdate = selectForUpdate; 095 } 096 097 public WorkflowStore(Store store, boolean selectForUpdate) throws StoreException { 098 super(store); 099 entityManager = getEntityManager(); 100 this.selectForUpdate = selectForUpdate; 101 } 102 103 /** 104 * Create a Workflow and return a WorkflowJobBean. It also creates the process instance for the job. 105 * 106 * @param workflow workflow bean 107 * @throws StoreException 108 */ 109 110 public void insertWorkflow(final WorkflowJobBean workflow) throws StoreException { 111 ParamChecker.notNull(workflow, "workflow"); 112 113 doOperation("insertWorkflow", new Callable<Void>() { 114 public Void call() throws SQLException, StoreException, WorkflowException { 115 entityManager.persist(workflow); 116 return null; 117 } 118 }); 119 } 120 121 /** 122 * Load the Workflow into a Bean and return it. Also load the Workflow Instance into the bean. And lock the Workflow 123 * depending on the locking parameter. 124 * 125 * @param id Workflow ID 126 * @param locking true if Workflow is to be locked 127 * @return WorkflowJobBean 128 * @throws StoreException 129 */ 130 public WorkflowJobBean getWorkflow(final String id, final boolean locking) throws StoreException { 131 ParamChecker.notEmpty(id, "WorkflowID"); 132 WorkflowJobBean wfBean = doOperation("getWorkflow", new Callable<WorkflowJobBean>() { 133 public WorkflowJobBean call() throws SQLException, StoreException, WorkflowException, InterruptedException { 134 WorkflowJobBean wfBean = null; 135 wfBean = getWorkflowOnly(id, locking); 136 if (wfBean == null) { 137 throw new StoreException(ErrorCode.E0604, id); 138 } 139 /* 140 * WorkflowInstance wfInstance; //krishna and next line 141 * wfInstance = workflowLib.get(id); wfInstance = 142 * wfBean.get(wfBean.getWfInstance()); 143 * wfBean.setWorkflowInstance(wfInstance); 144 * wfBean.setWfInstance(wfInstance); 145 */ 146 return wfBean; 147 } 148 }); 149 return wfBean; 150 } 151 152 /** 153 * Get the number of Workflows with the given status. 154 * 155 * @param status Workflow Status. 156 * @return number of Workflows with given status. 157 * @throws StoreException 158 */ 159 public int getWorkflowCountWithStatus(final String status) throws StoreException { 160 ParamChecker.notEmpty(status, "status"); 161 Integer cnt = doOperation("getWorkflowCountWithStatus", new Callable<Integer>() { 162 public Integer call() throws SQLException { 163 Query q = entityManager.createNamedQuery("GET_WORKFLOWS_COUNT_WITH_STATUS"); 164 q.setParameter("status", status); 165 Long count = (Long) q.getSingleResult(); 166 return Integer.valueOf(count.intValue()); 167 } 168 }); 169 return cnt.intValue(); 170 } 171 172 /** 173 * Get the number of Workflows with the given status which was modified in given time limit. 174 * 175 * @param status Workflow Status. 176 * @param secs No. of seconds within which the workflow got modified. 177 * @return number of Workflows modified within given time with given status. 178 * @throws StoreException 179 */ 180 public int getWorkflowCountWithStatusInLastNSeconds(final String status, final int secs) throws StoreException { 181 ParamChecker.notEmpty(status, "status"); 182 ParamChecker.notEmpty(status, "secs"); 183 Integer cnt = doOperation("getWorkflowCountWithStatusInLastNSecs", new Callable<Integer>() { 184 public Integer call() throws SQLException { 185 Query q = entityManager.createNamedQuery("GET_WORKFLOWS_COUNT_WITH_STATUS_IN_LAST_N_SECS"); 186 Timestamp ts = new Timestamp(System.currentTimeMillis() - (secs * 1000)); 187 q.setParameter("status", status); 188 q.setParameter("lastModTime", ts); 189 Long count = (Long) q.getSingleResult(); 190 return Integer.valueOf(count.intValue()); 191 } 192 }); 193 return cnt.intValue(); 194 } 195 196 /** 197 * Update the data from Workflow Bean to DB along with the workflow instance data. Action table is not updated 198 * 199 * @param wfBean Workflow Bean 200 * @throws StoreException If Workflow doesn't exist 201 */ 202 public void updateWorkflow(final WorkflowJobBean wfBean) throws StoreException { 203 ParamChecker.notNull(wfBean, "WorkflowJobBean"); 204 doOperation("updateWorkflow", new Callable<Void>() { 205 public Void call() throws SQLException, StoreException, WorkflowException, JPAExecutorException { 206 WorkflowJobQueryExecutor.getInstance().executeUpdate( 207 WorkflowJobQueryExecutor.WorkflowJobQuery.UPDATE_WORKFLOW, wfBean); 208 return null; 209 } 210 }); 211 } 212 213 /** 214 * Create a new Action record in the ACTIONS table with the given Bean. 215 * 216 * @param action WorkflowActionBean 217 * @throws StoreException If the action is already present 218 */ 219 public void insertAction(final WorkflowActionBean action) throws StoreException { 220 ParamChecker.notNull(action, "WorkflowActionBean"); 221 doOperation("insertAction", new Callable<Void>() { 222 public Void call() throws SQLException, StoreException, WorkflowException { 223 entityManager.persist(action); 224 return null; 225 } 226 }); 227 } 228 229 /** 230 * Load the action data and returns a bean. 231 * 232 * @param id Action Id 233 * @param locking true if the action is to be locked 234 * @return Action Bean 235 * @throws StoreException If action doesn't exist 236 */ 237 public WorkflowActionBean getAction(final String id, final boolean locking) throws StoreException { 238 ParamChecker.notEmpty(id, "ActionID"); 239 WorkflowActionBean action = doOperation("getAction", new Callable<WorkflowActionBean>() { 240 public WorkflowActionBean call() throws SQLException, StoreException, WorkflowException, 241 InterruptedException { 242 Query q = entityManager.createNamedQuery("GET_ACTION"); 243 /* 244 * if (locking) { OpenJPAQuery oq = OpenJPAPersistence.cast(q); 245 * FetchPlan fetch = oq.getFetchPlan(); 246 * fetch.setReadLockMode(LockModeType.WRITE); 247 * fetch.setLockTimeout(1000); // 1 seconds } 248 */ 249 WorkflowActionBean action = null; 250 q.setParameter("id", id); 251 List<WorkflowActionBean> actions = q.getResultList(); 252 // action = (WorkflowActionBean) q.getSingleResult(); 253 if (actions.size() > 0) { 254 action = actions.get(0); 255 } 256 else { 257 throw new StoreException(ErrorCode.E0605, id); 258 } 259 260 /* 261 * if (locking) return action; else 262 */ 263 // return action; 264 return getBeanForRunningAction(action); 265 } 266 }); 267 return action; 268 } 269 270 /** 271 * Update the given action bean to DB. 272 * 273 * @param action Action Bean 274 * @throws StoreException if action doesn't exist 275 */ 276 public void updateAction(final WorkflowActionBean action) throws StoreException { 277 ParamChecker.notNull(action, "WorkflowActionBean"); 278 doOperation("updateAction", new Callable<Void>() { 279 public Void call() throws SQLException, StoreException, WorkflowException, JPAExecutorException { 280 WorkflowActionQueryExecutor.getInstance().executeUpdate( 281 WorkflowActionQueryExecutor.WorkflowActionQuery.UPDATE_ACTION, action); 282 return null; 283 } 284 }); 285 } 286 287 /** 288 * Delete the Action with given id. 289 * 290 * @param id Action ID 291 * @throws StoreException if Action doesn't exist 292 */ 293 public void deleteAction(final String id) throws StoreException { 294 ParamChecker.notEmpty(id, "ActionID"); 295 doOperation("deleteAction", new Callable<Void>() { 296 public Void call() throws SQLException, StoreException, WorkflowException { 297 /* 298 * Query q = entityManager.createNamedQuery("DELETE_ACTION"); 299 * q.setParameter("id", id); q.executeUpdate(); 300 */ 301 WorkflowActionBean action = entityManager.find(WorkflowActionBean.class, id); 302 if (action != null) { 303 entityManager.remove(action); 304 } 305 return null; 306 } 307 }); 308 } 309 310 /** 311 * Loads all the actions for the given Workflow. Also locks all the actions if locking is true. 312 * 313 * @param wfId Workflow ID 314 * @param locking true if Actions are to be locked 315 * @return A List of WorkflowActionBean 316 * @throws StoreException 317 */ 318 public List<WorkflowActionBean> getActionsForWorkflow(final String wfId, final boolean locking) 319 throws StoreException { 320 ParamChecker.notEmpty(wfId, "WorkflowID"); 321 List<WorkflowActionBean> actions = doOperation("getActionsForWorkflow", 322 new Callable<List<WorkflowActionBean>>() { 323 public List<WorkflowActionBean> call() throws SQLException, StoreException, WorkflowException, 324 InterruptedException { 325 List<WorkflowActionBean> actions; 326 List<WorkflowActionBean> actionList = new ArrayList<WorkflowActionBean>(); 327 try { 328 Query q = entityManager.createNamedQuery("GET_ACTIONS_FOR_WORKFLOW"); 329 330 /* 331 * OpenJPAQuery oq = OpenJPAPersistence.cast(q); 332 * if (locking) { // 333 * q.setHint("openjpa.FetchPlan.ReadLockMode" 334 * ,"WRITE"); FetchPlan fetch = oq.getFetchPlan(); 335 * fetch.setReadLockMode(LockModeType.WRITE); 336 * fetch.setLockTimeout(1000); // 1 seconds } 337 */ 338 q.setParameter("wfId", wfId); 339 actions = q.getResultList(); 340 for (WorkflowActionBean a : actions) { 341 WorkflowActionBean aa = getBeanForRunningAction(a); 342 actionList.add(aa); 343 } 344 } 345 catch (IllegalStateException e) { 346 throw new StoreException(ErrorCode.E0601, e.getMessage(), e); 347 } 348 /* 349 * if (locking) { return actions; } else { 350 */ 351 return actionList; 352 // } 353 } 354 }); 355 return actions; 356 } 357 358 /** 359 * Loads given number of actions for the given Workflow. Also locks all the actions if locking is true. 360 * 361 * @param wfId Workflow ID 362 * @param start offset for select statement 363 * @param len number of Workflow Actions to be returned 364 * @param locking true if Actions are to be locked 365 * @return A List of WorkflowActionBean 366 * @throws StoreException 367 */ 368 public List<WorkflowActionBean> getActionsSubsetForWorkflow(final String wfId, final int start, final int len) 369 throws StoreException { 370 ParamChecker.notEmpty(wfId, "WorkflowID"); 371 List<WorkflowActionBean> actions = doOperation("getActionsForWorkflow", 372 new Callable<List<WorkflowActionBean>>() { 373 public List<WorkflowActionBean> call() throws SQLException, StoreException, WorkflowException, 374 InterruptedException { 375 List<WorkflowActionBean> actions; 376 List<WorkflowActionBean> actionList = new ArrayList<WorkflowActionBean>(); 377 try { 378 Query q = entityManager.createNamedQuery("GET_ACTIONS_FOR_WORKFLOW"); 379 OpenJPAQuery oq = OpenJPAPersistence.cast(q); 380 q.setParameter("wfId", wfId); 381 q.setFirstResult(start - 1); 382 q.setMaxResults(len); 383 actions = q.getResultList(); 384 for (WorkflowActionBean a : actions) { 385 WorkflowActionBean aa = getBeanForRunningAction(a); 386 actionList.add(aa); 387 } 388 } 389 catch (IllegalStateException e) { 390 throw new StoreException(ErrorCode.E0601, e.getMessage(), e); 391 } 392 return actionList; 393 } 394 }); 395 return actions; 396 } 397 398 /** 399 * Load All the actions that are pending for more than given time. 400 * 401 * @param minimumPendingAgeSecs Minimum Pending age in seconds 402 * @return List of action beans 403 * @throws StoreException 404 */ 405 public List<WorkflowActionBean> getPendingActions(final long minimumPendingAgeSecs) throws StoreException { 406 List<WorkflowActionBean> actions = doOperation("getPendingActions", new Callable<List<WorkflowActionBean>>() { 407 public List<WorkflowActionBean> call() throws SQLException, StoreException, WorkflowException { 408 Timestamp ts = new Timestamp(System.currentTimeMillis() - minimumPendingAgeSecs * 1000); 409 List<WorkflowActionBean> actionList = null; 410 try { 411 Query q = entityManager.createNamedQuery("GET_PENDING_ACTIONS"); 412 q.setParameter("pendingAge", ts); 413 actionList = q.getResultList(); 414 } 415 catch (IllegalStateException e) { 416 throw new StoreException(ErrorCode.E0601, e.getMessage(), e); 417 } 418 return actionList; 419 } 420 }); 421 return actions; 422 } 423 424 /** 425 * Load All the actions that are running and were last checked after now - miminumCheckAgeSecs 426 * 427 * @param checkAgeSecs check age in seconds. 428 * @return List of action beans. 429 * @throws StoreException 430 */ 431 public List<WorkflowActionBean> getRunningActions(final long checkAgeSecs) throws StoreException { 432 List<WorkflowActionBean> actions = doOperation("getRunningActions", new Callable<List<WorkflowActionBean>>() { 433 434 public List<WorkflowActionBean> call() throws SQLException, StoreException, WorkflowException { 435 List<WorkflowActionBean> actions = new ArrayList<WorkflowActionBean>(); 436 Timestamp ts = new Timestamp(System.currentTimeMillis() - checkAgeSecs * 1000); 437 try { 438 Query q = entityManager.createNamedQuery("GET_RUNNING_ACTIONS"); 439 q.setParameter("lastCheckTime", ts); 440 actions = q.getResultList(); 441 } 442 catch (IllegalStateException e) { 443 throw new StoreException(ErrorCode.E0601, e.getMessage(), e); 444 } 445 446 return actions; 447 } 448 }); 449 return actions; 450 } 451 452 /** 453 * Load All the actions that are START_RETRY or START_MANUAL or END_RETRY or END_MANUAL. 454 * 455 * @param wfId String 456 * @return List of action beans 457 * @throws StoreException 458 */ 459 public List<WorkflowActionBean> getRetryAndManualActions(final String wfId) throws StoreException { 460 List<WorkflowActionBean> actions = doOperation("GET_RETRY_MANUAL_ACTIONS", 461 new Callable<List<WorkflowActionBean>>() { 462 public List<WorkflowActionBean> call() throws SQLException, StoreException, WorkflowException { 463 List<WorkflowActionBean> actionList = null; 464 try { 465 Query q = entityManager.createNamedQuery("GET_RETRY_MANUAL_ACTIONS"); 466 q.setParameter("wfId", wfId); 467 actionList = q.getResultList(); 468 } 469 catch (IllegalStateException e) { 470 throw new StoreException(ErrorCode.E0601, e.getMessage(), e); 471 } 472 473 return actionList; 474 } 475 }); 476 return actions; 477 } 478 479 /** 480 * Loads all the jobs that are satisfying the given filter condition. Filters can be applied on user, group, 481 * appName, status. 482 * 483 * @param filter Filter condition 484 * @param start offset for select statement 485 * @param len number of Workflows to be returned 486 * @return A list of workflows 487 * @throws StoreException 488 */ 489 public WorkflowsInfo getWorkflowsInfo(final Map<String, List<String>> filter, final int start, final int len) 490 throws StoreException { 491 492 WorkflowsInfo workFlowsInfo = doOperation("getWorkflowsInfo", new Callable<WorkflowsInfo>() { 493 @SuppressWarnings("unchecked") 494 public WorkflowsInfo call() throws SQLException, StoreException { 495 496 List<String> orArray = new ArrayList<String>(); 497 List<String> colArray = new ArrayList<String>(); 498 List<String> valArray = new ArrayList<String>(); 499 StringBuilder sb = new StringBuilder(""); 500 boolean isStatus = false; 501 boolean isGroup = false; 502 boolean isAppName = false; 503 boolean isUser = false; 504 boolean isEnabled = false; 505 int index = 0; 506 for (Map.Entry<String, List<String>> entry : filter.entrySet()) { 507 String colName = null; 508 String colVar = null; 509 if (entry.getKey().equals(OozieClient.FILTER_GROUP)) { 510 List<String> values = filter.get(OozieClient.FILTER_GROUP); 511 colName = "group"; 512 for (int i = 0; i < values.size(); i++) { 513 colVar = "group"; 514 colVar = colVar + index; 515 if (!isEnabled && !isGroup) { 516 sb.append(seletStr).append(" where w.group IN (:group" + index); 517 isGroup = true; 518 isEnabled = true; 519 } 520 else { 521 if (isEnabled && !isGroup) { 522 sb.append(" and w.group IN (:group" + index); 523 isGroup = true; 524 } 525 else { 526 if (isGroup) { 527 sb.append(", :group" + index); 528 } 529 } 530 } 531 if (i == values.size() - 1) { 532 sb.append(")"); 533 } 534 index++; 535 valArray.add(values.get(i)); 536 orArray.add(colName); 537 colArray.add(colVar); 538 } 539 } 540 else { 541 if (entry.getKey().equals(OozieClient.FILTER_STATUS)) { 542 List<String> values = filter.get(OozieClient.FILTER_STATUS); 543 colName = "status"; 544 for (int i = 0; i < values.size(); i++) { 545 colVar = "status"; 546 colVar = colVar + index; 547 if (!isEnabled && !isStatus) { 548 sb.append(seletStr).append(" where w.statusStr IN (:status" + index); 549 isStatus = true; 550 isEnabled = true; 551 } 552 else { 553 if (isEnabled && !isStatus) { 554 sb.append(" and w.statusStr IN (:status" + index); 555 isStatus = true; 556 } 557 else { 558 if (isStatus) { 559 sb.append(", :status" + index); 560 } 561 } 562 } 563 if (i == values.size() - 1) { 564 sb.append(")"); 565 } 566 index++; 567 valArray.add(values.get(i)); 568 orArray.add(colName); 569 colArray.add(colVar); 570 } 571 } 572 else { 573 if (entry.getKey().equals(OozieClient.FILTER_NAME)) { 574 List<String> values = filter.get(OozieClient.FILTER_NAME); 575 colName = "appName"; 576 for (int i = 0; i < values.size(); i++) { 577 colVar = "appName"; 578 colVar = colVar + index; 579 if (!isEnabled && !isAppName) { 580 sb.append(seletStr).append(" where w.appName IN (:appName" + index); 581 isAppName = true; 582 isEnabled = true; 583 } 584 else { 585 if (isEnabled && !isAppName) { 586 sb.append(" and w.appName IN (:appName" + index); 587 isAppName = true; 588 } 589 else { 590 if (isAppName) { 591 sb.append(", :appName" + index); 592 } 593 } 594 } 595 if (i == values.size() - 1) { 596 sb.append(")"); 597 } 598 index++; 599 valArray.add(values.get(i)); 600 orArray.add(colName); 601 colArray.add(colVar); 602 } 603 } 604 else { 605 if (entry.getKey().equals(OozieClient.FILTER_USER)) { 606 List<String> values = filter.get(OozieClient.FILTER_USER); 607 colName = "user"; 608 for (int i = 0; i < values.size(); i++) { 609 colVar = "user"; 610 colVar = colVar + index; 611 if (!isEnabled && !isUser) { 612 sb.append(seletStr).append(" where w.user IN (:user" + index); 613 isUser = true; 614 isEnabled = true; 615 } 616 else { 617 if (isEnabled && !isUser) { 618 sb.append(" and w.user IN (:user" + index); 619 isUser = true; 620 } 621 else { 622 if (isUser) { 623 sb.append(", :user" + index); 624 } 625 } 626 } 627 if (i == values.size() - 1) { 628 sb.append(")"); 629 } 630 index++; 631 valArray.add(values.get(i)); 632 orArray.add(colName); 633 colArray.add(colVar); 634 } 635 } 636 } 637 } 638 } 639 } 640 641 int realLen = 0; 642 643 Query q = null; 644 Query qTotal = null; 645 if (orArray.size() == 0) { 646 q = entityManager.createNamedQuery("GET_WORKFLOWS_COLUMNS"); 647 q.setFirstResult(start - 1); 648 q.setMaxResults(len); 649 qTotal = entityManager.createNamedQuery("GET_WORKFLOWS_COUNT"); 650 } 651 else { 652 if (orArray.size() > 0) { 653 StringBuilder sbTotal = new StringBuilder(sb); 654 sb.append(" order by w.startTimestamp desc "); 655 XLog.getLog(getClass()).debug("Created String is **** " + sb.toString()); 656 q = entityManager.createQuery(sb.toString()); 657 q.setFirstResult(start - 1); 658 q.setMaxResults(len); 659 qTotal = entityManager.createQuery(sbTotal.toString().replace(seletStr, countStr)); 660 for (int i = 0; i < orArray.size(); i++) { 661 q.setParameter(colArray.get(i), valArray.get(i)); 662 qTotal.setParameter(colArray.get(i), valArray.get(i)); 663 } 664 } 665 } 666 667 OpenJPAQuery kq = OpenJPAPersistence.cast(q); 668 JDBCFetchPlan fetch = (JDBCFetchPlan) kq.getFetchPlan(); 669 fetch.setFetchBatchSize(20); 670 fetch.setResultSetType(ResultSetType.SCROLL_INSENSITIVE); 671 fetch.setFetchDirection(FetchDirection.FORWARD); 672 fetch.setLRSSizeAlgorithm(LRSSizeAlgorithm.LAST); 673 List<?> resultList = q.getResultList(); 674 List<Object[]> objectArrList = (List<Object[]>) resultList; 675 List<WorkflowJobBean> wfBeansList = new ArrayList<WorkflowJobBean>(); 676 677 for (Object[] arr : objectArrList) { 678 WorkflowJobBean ww = getBeanForWorkflowFromArray(arr); 679 wfBeansList.add(ww); 680 } 681 682 realLen = ((Long) qTotal.getSingleResult()).intValue(); 683 684 return new WorkflowsInfo(wfBeansList, start, len, realLen); 685 } 686 }); 687 return workFlowsInfo; 688 689 } 690 691 /** 692 * Load the Workflow and all Action details and return a WorkflowJobBean. Workflow Instance is not loaded 693 * 694 * @param id Workflow Id 695 * @return Workflow Bean 696 * @throws StoreException If Workflow doesn't exist 697 */ 698 public WorkflowJobBean getWorkflowInfo(final String id) throws StoreException { 699 ParamChecker.notEmpty(id, "WorkflowID"); 700 WorkflowJobBean wfBean = doOperation("getWorkflowInfo", new Callable<WorkflowJobBean>() { 701 public WorkflowJobBean call() throws SQLException, StoreException, InterruptedException { 702 WorkflowJobBean wfBean = null; 703 wfBean = getWorkflowforInfo(id, false); 704 if (wfBean == null) { 705 throw new StoreException(ErrorCode.E0604, id); 706 } 707 else { 708 wfBean.setActions(getActionsForWorkflow(id, false)); 709 } 710 return wfBean; 711 } 712 }); 713 return wfBean; 714 } 715 716 /** 717 * Load the Workflow and subset Actions details and return a WorkflowJobBean. Workflow Instance is not loaded 718 * 719 * @param id Workflow Id 720 * @param start offset for select statement for actions 721 * @param len number of Workflow Actions to be returned 722 * @return Workflow Bean 723 * @throws StoreException If Workflow doesn't exist 724 */ 725 public WorkflowJobBean getWorkflowInfoWithActionsSubset(final String id, final int start, final int len) throws StoreException { 726 ParamChecker.notEmpty(id, "WorkflowID"); 727 WorkflowJobBean wfBean = doOperation("getWorkflowInfo", new Callable<WorkflowJobBean>() { 728 public WorkflowJobBean call() throws SQLException, StoreException, InterruptedException { 729 WorkflowJobBean wfBean = null; 730 wfBean = getWorkflowforInfo(id, false); 731 if (wfBean == null) { 732 throw new StoreException(ErrorCode.E0604, id); 733 } 734 else { 735 wfBean.setActions(getActionsSubsetForWorkflow(id, start, len)); 736 } 737 return wfBean; 738 } 739 }); 740 return wfBean; 741 } 742 743 /** 744 * Get the Workflow ID with given external ID which will be assigned for the subworkflows. 745 * 746 * @param externalId external ID 747 * @return Workflow ID 748 * @throws StoreException if there is no job with external ID 749 */ 750 public String getWorkflowIdForExternalId(final String externalId) throws StoreException { 751 ParamChecker.notEmpty(externalId, "externalId"); 752 String wfId = doOperation("getWorkflowIdForExternalId", new Callable<String>() { 753 public String call() throws SQLException, StoreException { 754 String id = ""; 755 Query q = entityManager.createNamedQuery("GET_WORKFLOW_ID_FOR_EXTERNAL_ID"); 756 q.setParameter("externalId", externalId); 757 List<String> w = q.getResultList(); 758 if (w.size() == 0) { 759 id = ""; 760 } 761 else { 762 int index = w.size() - 1; 763 id = w.get(index); 764 } 765 return id; 766 } 767 }); 768 return wfId; 769 } 770 771 private static final long DAY_IN_MS = 24 * 60 * 60 * 1000; 772 773 /** 774 * Purge the Workflows Completed older than given days. 775 * 776 * @param olderThanDays number of days for which to preserve the workflows 777 * @throws StoreException 778 */ 779 public void purge(final long olderThanDays, final int limit) throws StoreException { 780 doOperation("purge", new Callable<Void>() { 781 public Void call() throws SQLException, StoreException, WorkflowException { 782 Timestamp maxEndTime = new Timestamp(System.currentTimeMillis() - (olderThanDays * DAY_IN_MS)); 783 Query q = entityManager.createNamedQuery("GET_COMPLETED_WORKFLOWS_OLDER_THAN"); 784 q.setParameter("endTime", maxEndTime); 785 q.setMaxResults(limit); 786 List<WorkflowJobBean> workflows = q.getResultList(); 787 int actionDeleted = 0; 788 if (workflows.size() != 0) { 789 for (WorkflowJobBean w : workflows) { 790 String wfId = w.getId(); 791 entityManager.remove(w); 792 Query g = entityManager.createNamedQuery("DELETE_ACTIONS_FOR_WORKFLOW"); 793 g.setParameter("wfId", wfId); 794 actionDeleted += g.executeUpdate(); 795 } 796 } 797 XLog.getLog(getClass()).debug("ENDED Workflow Purge deleted jobs :" + workflows.size() + " and actions " + actionDeleted); 798 return null; 799 } 800 }); 801 } 802 803 private <V> V doOperation(String name, Callable<V> command) throws StoreException { 804 try { 805 Instrumentation.Cron cron = new Instrumentation.Cron(); 806 cron.start(); 807 V retVal; 808 try { 809 retVal = command.call(); 810 } 811 finally { 812 cron.stop(); 813 } 814 Services.get().get(InstrumentationService.class).get().addCron(INSTR_GROUP, name, cron); 815 return retVal; 816 } 817 catch (StoreException ex) { 818 throw ex; 819 } 820 catch (SQLException ex) { 821 throw new StoreException(ErrorCode.E0611, name, ex.getMessage(), ex); 822 } 823 catch (Exception e) { 824 throw new StoreException(ErrorCode.E0607, name, e.getMessage(), e); 825 } 826 } 827 828 private WorkflowJobBean getWorkflowOnly(final String id, boolean locking) throws SQLException, 829 InterruptedException, StoreException { 830 WorkflowJobBean wfBean = null; 831 Query q = entityManager.createNamedQuery("GET_WORKFLOW"); 832 /* 833 * if (locking) { // q.setHint("openjpa.FetchPlan.ReadLockMode","READ"); 834 * OpenJPAQuery oq = OpenJPAPersistence.cast(q); FetchPlan fetch = 835 * oq.getFetchPlan(); fetch.setReadLockMode(LockModeType.WRITE); 836 * fetch.setLockTimeout(-1); // unlimited } 837 */ 838 q.setParameter("id", id); 839 List<WorkflowJobBean> w = q.getResultList(); 840 if (w.size() > 0) { 841 wfBean = w.get(0); 842 } 843 return wfBean; 844 // return getBeanForRunningWorkflow(wfBean); 845 } 846 847 private WorkflowJobBean getWorkflowforInfo(final String id, boolean locking) throws SQLException, 848 InterruptedException, StoreException { 849 WorkflowJobBean wfBean = null; 850 Query q = entityManager.createNamedQuery("GET_WORKFLOW"); 851 q.setParameter("id", id); 852 List<WorkflowJobBean> w = q.getResultList(); 853 if (w.size() > 0) { 854 wfBean = w.get(0); 855 return getBeanForRunningWorkflow(wfBean); 856 } 857 return null; 858 } 859 860 private WorkflowJobBean getBeanForRunningWorkflow(WorkflowJobBean w) throws SQLException { 861 WorkflowJobBean wfBean = new WorkflowJobBean(); 862 wfBean.setId(w.getId()); 863 wfBean.setAppName(w.getAppName()); 864 wfBean.setAppPath(w.getAppPath()); 865 wfBean.setConfBlob(w.getConfBlob()); 866 wfBean.setGroup(w.getGroup()); 867 wfBean.setRun(w.getRun()); 868 wfBean.setUser(w.getUser()); 869 wfBean.setCreatedTime(w.getCreatedTime()); 870 wfBean.setEndTime(w.getEndTime()); 871 wfBean.setExternalId(w.getExternalId()); 872 wfBean.setLastModifiedTime(w.getLastModifiedTime()); 873 wfBean.setLogToken(w.getLogToken()); 874 wfBean.setProtoActionConfBlob(w.getProtoActionConfBlob()); 875 wfBean.setSlaXmlBlob(w.getSlaXmlBlob()); 876 wfBean.setStartTime(w.getStartTime()); 877 wfBean.setStatus(w.getStatus()); 878 wfBean.setWfInstanceBlob(w.getWfInstanceBlob()); 879 return wfBean; 880 } 881 882 private WorkflowJobBean getBeanForWorkflowFromArray(Object[] arr) { 883 884 WorkflowJobBean wfBean = new WorkflowJobBean(); 885 wfBean.setId((String) arr[0]); 886 if (arr[1] != null) { 887 wfBean.setAppName((String) arr[1]); 888 } 889 if (arr[2] != null) { 890 wfBean.setStatus(Status.valueOf((String) arr[2])); 891 } 892 if (arr[3] != null) { 893 wfBean.setRun((Integer) arr[3]); 894 } 895 if (arr[4] != null) { 896 wfBean.setUser((String) arr[4]); 897 } 898 if (arr[5] != null) { 899 wfBean.setGroup((String) arr[5]); 900 } 901 if (arr[6] != null) { 902 wfBean.setCreatedTime((Timestamp) arr[6]); 903 } 904 if (arr[7] != null) { 905 wfBean.setStartTime((Timestamp) arr[7]); 906 } 907 if (arr[8] != null) { 908 wfBean.setLastModifiedTime((Timestamp) arr[8]); 909 } 910 if (arr[9] != null) { 911 wfBean.setEndTime((Timestamp) arr[9]); 912 } 913 return wfBean; 914 } 915 916 private WorkflowActionBean getBeanForRunningAction(WorkflowActionBean a) throws SQLException { 917 if (a != null) { 918 WorkflowActionBean action = new WorkflowActionBean(); 919 action.setId(a.getId()); 920 action.setConfBlob(a.getConfBlob()); 921 action.setConsoleUrl(a.getConsoleUrl()); 922 action.setDataBlob(a.getDataBlob()); 923 action.setStatsBlob(a.getStatsBlob()); 924 action.setExternalChildIDsBlob(a.getExternalChildIDsBlob()); 925 action.setErrorInfo(a.getErrorCode(), a.getErrorMessage()); 926 action.setExternalId(a.getExternalId()); 927 action.setExternalStatus(a.getExternalStatus()); 928 action.setName(a.getName()); 929 action.setCred(a.getCred()); 930 action.setRetries(a.getRetries()); 931 action.setTrackerUri(a.getTrackerUri()); 932 action.setTransition(a.getTransition()); 933 action.setType(a.getType()); 934 action.setEndTime(a.getEndTime()); 935 action.setExecutionPath(a.getExecutionPath()); 936 action.setLastCheckTime(a.getLastCheckTime()); 937 action.setLogToken(a.getLogToken()); 938 if (a.isPending() == true) { 939 action.setPending(); 940 } 941 action.setPendingAge(a.getPendingAge()); 942 action.setSignalValue(a.getSignalValue()); 943 action.setSlaXmlBlob(a.getSlaXmlBlob()); 944 action.setStartTime(a.getStartTime()); 945 action.setStatus(a.getStatus()); 946 action.setJobId(a.getWfId()); 947 action.setUserRetryCount(a.getUserRetryCount()); 948 action.setUserRetryInterval(a.getUserRetryInterval()); 949 action.setUserRetryMax(a.getUserRetryMax()); 950 return action; 951 } 952 return null; 953 } 954}