ExecutableNode.java

/*
 * Copyright 2013 LinkedIn Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */

package azkaban.executor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import azkaban.flow.Node;
import azkaban.utils.Props;
import azkaban.utils.PropsUtils;
import azkaban.utils.TypedMapWrapper;

/**
 * Base Executable that nodes and flows are based.
 */
public class ExecutableNode {
  public static final String ID_PARAM = "id";
  public static final String STATUS_PARAM = "status";
  public static final String STARTTIME_PARAM = "startTime";
  public static final String ENDTIME_PARAM = "endTime";
  public static final String UPDATETIME_PARAM = "updateTime";
  public static final String INNODES_PARAM = "inNodes";
  public static final String OUTNODES_PARAM = "outNodes";
  public static final String TYPE_PARAM = "type";
  public static final String PROPS_SOURCE_PARAM = "propSource";
  public static final String JOB_SOURCE_PARAM = "jobSource";
  public static final String OUTPUT_PROPS_PARAM = "outputProps";

  private String id;
  private String type = null;
  private Status status = Status.READY;
  private long startTime = -1;
  private long endTime = -1;
  private long updateTime = -1;

  // Path to Job File
  private String jobSource;
  // Path to top level props file
  private String propsSource;
  private Set<String> inNodes = new HashSet<String>();
  private Set<String> outNodes = new HashSet<String>();

  private Props inputProps;
  private Props outputProps;

  public static final String ATTEMPT_PARAM = "attempt";
  public static final String PASTATTEMPTS_PARAM = "pastAttempts";

  private int attempt = 0;
  private long delayExecution = 0;
  private ArrayList<ExecutionAttempt> pastAttempts = null;

  // Transient. These values aren't saved, but rediscovered.
  private ExecutableFlowBase parentFlow;

  public ExecutableNode(Node node) {
    this.id = node.getId();
    this.jobSource = node.getJobSource();
    this.propsSource = node.getPropsSource();
  }

  public ExecutableNode(Node node, ExecutableFlowBase parent) {
    this(node.getId(), node.getType(), node.getJobSource(), node
        .getPropsSource(), parent);
  }

  public ExecutableNode(String id, String type, String jobSource,
      String propsSource, ExecutableFlowBase parent) {
    this.id = id;
    this.jobSource = jobSource;
    this.propsSource = propsSource;
    this.type = type;
    setParentFlow(parent);
  }

  public ExecutableNode() {
  }

  public ExecutableFlow getExecutableFlow() {
    if (parentFlow == null) {
      return null;
    }

    return parentFlow.getExecutableFlow();
  }

  public void setParentFlow(ExecutableFlowBase flow) {
    this.parentFlow = flow;
  }

  public ExecutableFlowBase getParentFlow() {
    return parentFlow;
  }

  public String getId() {
    return id;
  }

  public void setId(String id) {
    this.id = id;
  }

  public Status getStatus() {
    return status;
  }

  public String getType() {
    return type;
  }

  public void setType(String type) {
    this.type = type;
  }

  public void setStatus(Status status) {
    this.status = status;
  }

  public long getStartTime() {
    return startTime;
  }

  public void setStartTime(long startTime) {
    this.startTime = startTime;
  }

  public long getEndTime() {
    return endTime;
  }

  public void setEndTime(long endTime) {
    this.endTime = endTime;
  }

  public long getUpdateTime() {
    return updateTime;
  }

  public void setUpdateTime(long updateTime) {
    this.updateTime = updateTime;
  }

  public void addOutNode(String exNode) {
    outNodes.add(exNode);
  }

  public void addInNode(String exNode) {
    inNodes.add(exNode);
  }

  public Set<String> getOutNodes() {
    return outNodes;
  }

  public Set<String> getInNodes() {
    return inNodes;
  }

  public boolean hasJobSource() {
    return jobSource != null;
  }

  public boolean hasPropsSource() {
    return propsSource != null;
  }

  public String getJobSource() {
    return jobSource;
  }

  public String getPropsSource() {
    return propsSource;
  }

  public void setInputProps(Props input) {
    this.inputProps = input;
  }

  public void setOutputProps(Props output) {
    this.outputProps = output;
  }

  public Props getInputProps() {
    return this.inputProps;
  }

  public Props getOutputProps() {
    return outputProps;
  }

  public long getDelayedExecution() {
    return delayExecution;
  }

  public void setDelayedExecution(long delayMs) {
    delayExecution = delayMs;
  }

  public List<ExecutionAttempt> getPastAttemptList() {
    return pastAttempts;
  }

  public int getAttempt() {
    return attempt;
  }

  public void setAttempt(int attempt) {
    this.attempt = attempt;
  }

  public void resetForRetry() {
    ExecutionAttempt pastAttempt = new ExecutionAttempt(attempt, this);
    attempt++;

    synchronized (this) {
      if (pastAttempts == null) {
        pastAttempts = new ArrayList<ExecutionAttempt>();
      }

      pastAttempts.add(pastAttempt);
    }

    this.setStartTime(-1);
    this.setEndTime(-1);
    this.setUpdateTime(System.currentTimeMillis());
    this.setStatus(Status.READY);
  }

  public List<Object> getAttemptObjects() {
    ArrayList<Object> array = new ArrayList<Object>();

    for (ExecutionAttempt attempt : pastAttempts) {
      array.add(attempt.toObject());
    }

    return array;
  }

  public String getNestedId() {
    return getPrintableId(":");
  }

  public String getPrintableId(String delimiter) {
    if (this.getParentFlow() == null
        || this.getParentFlow() instanceof ExecutableFlow) {
      return getId();
    }
    return getParentFlow().getPrintableId(delimiter) + delimiter + getId();
  }

  public Map<String, Object> toObject() {
    Map<String, Object> mapObj = new HashMap<String, Object>();
    fillMapFromExecutable(mapObj);

    return mapObj;
  }

  protected void fillMapFromExecutable(Map<String, Object> objMap) {
    objMap.put(ID_PARAM, this.id);
    objMap.put(STATUS_PARAM, status.toString());
    objMap.put(STARTTIME_PARAM, startTime);
    objMap.put(ENDTIME_PARAM, endTime);
    objMap.put(UPDATETIME_PARAM, updateTime);
    objMap.put(TYPE_PARAM, type);
    objMap.put(ATTEMPT_PARAM, attempt);

    if (inNodes != null && !inNodes.isEmpty()) {
      objMap.put(INNODES_PARAM, inNodes);
    }
    if (outNodes != null && !outNodes.isEmpty()) {
      objMap.put(OUTNODES_PARAM, outNodes);
    }

    if (hasPropsSource()) {
      objMap.put(PROPS_SOURCE_PARAM, this.propsSource);
    }
    if (hasJobSource()) {
      objMap.put(JOB_SOURCE_PARAM, this.jobSource);
    }

    if (outputProps != null && outputProps.size() > 0) {
      objMap.put(OUTPUT_PROPS_PARAM, PropsUtils.toStringMap(outputProps, true));
    }

    if (pastAttempts != null) {
      ArrayList<Object> attemptsList =
          new ArrayList<Object>(pastAttempts.size());
      for (ExecutionAttempt attempts : pastAttempts) {
        attemptsList.add(attempts.toObject());
      }
      objMap.put(PASTATTEMPTS_PARAM, attemptsList);
    }
  }

  @SuppressWarnings("unchecked")
  public void fillExecutableFromMapObject(
      TypedMapWrapper<String, Object> wrappedMap) {
    this.id = wrappedMap.getString(ID_PARAM);
    this.type = wrappedMap.getString(TYPE_PARAM);
    this.status = Status.valueOf(wrappedMap.getString(STATUS_PARAM));
    this.startTime = wrappedMap.getLong(STARTTIME_PARAM);
    this.endTime = wrappedMap.getLong(ENDTIME_PARAM);
    this.updateTime = wrappedMap.getLong(UPDATETIME_PARAM);
    this.attempt = wrappedMap.getInt(ATTEMPT_PARAM, 0);

    this.inNodes = new HashSet<String>();
    this.inNodes.addAll(wrappedMap.getStringCollection(INNODES_PARAM,
        Collections.<String> emptySet()));

    this.outNodes = new HashSet<String>();
    this.outNodes.addAll(wrappedMap.getStringCollection(OUTNODES_PARAM,
        Collections.<String> emptySet()));

    this.propsSource = wrappedMap.getString(PROPS_SOURCE_PARAM);
    this.jobSource = wrappedMap.getString(JOB_SOURCE_PARAM);

    Map<String, String> outputProps =
        wrappedMap.<String, String> getMap(OUTPUT_PROPS_PARAM);
    if (outputProps != null) {
      this.outputProps = new Props(null, outputProps);
    }

    Collection<Object> pastAttempts =
        wrappedMap.<Object> getCollection(PASTATTEMPTS_PARAM);
    if (pastAttempts != null) {
      ArrayList<ExecutionAttempt> attempts = new ArrayList<ExecutionAttempt>();
      for (Object attemptObj : pastAttempts) {
        ExecutionAttempt attempt = ExecutionAttempt.fromObject(attemptObj);
        attempts.add(attempt);
      }

      this.pastAttempts = attempts;
    }
  }

  public void fillExecutableFromMapObject(Map<String, Object> objMap) {
    TypedMapWrapper<String, Object> wrapper =
        new TypedMapWrapper<String, Object>(objMap);
    fillExecutableFromMapObject(wrapper);
  }

  public Map<String, Object> toUpdateObject() {
    Map<String, Object> updatedNodeMap = new HashMap<String, Object>();
    updatedNodeMap.put(ID_PARAM, getId());
    updatedNodeMap.put(STATUS_PARAM, getStatus().getNumVal());
    updatedNodeMap.put(STARTTIME_PARAM, getStartTime());
    updatedNodeMap.put(ENDTIME_PARAM, getEndTime());
    updatedNodeMap.put(UPDATETIME_PARAM, getUpdateTime());

    updatedNodeMap.put(ATTEMPT_PARAM, getAttempt());

    if (getAttempt() > 0) {
      ArrayList<Map<String, Object>> pastAttempts =
          new ArrayList<Map<String, Object>>();
      for (ExecutionAttempt attempt : getPastAttemptList()) {
        pastAttempts.add(attempt.toObject());
      }
      updatedNodeMap.put(PASTATTEMPTS_PARAM, pastAttempts);
    }

    return updatedNodeMap;
  }

  public void applyUpdateObject(TypedMapWrapper<String, Object> updateData) {
    this.status =
        Status.fromInteger(updateData.getInt(STATUS_PARAM,
            this.status.getNumVal()));
    this.startTime = updateData.getLong(STARTTIME_PARAM);
    this.updateTime = updateData.getLong(UPDATETIME_PARAM);
    this.endTime = updateData.getLong(ENDTIME_PARAM);

    if (updateData.containsKey(ATTEMPT_PARAM)) {
      attempt = updateData.getInt(ATTEMPT_PARAM);
      if (attempt > 0) {
        updatePastAttempts(updateData.<Object> getList(PASTATTEMPTS_PARAM,
            Collections.<Object> emptyList()));
      }
    }
  }

  public void applyUpdateObject(Map<String, Object> updateData) {
    TypedMapWrapper<String, Object> wrapper =
        new TypedMapWrapper<String, Object>(updateData);
    applyUpdateObject(wrapper);
  }

  public void cancelNode(long cancelTime) {
    if (this.status == Status.DISABLED) {
      skipNode(cancelTime);
    } else {
      this.setStatus(Status.CANCELLED);
      this.setStartTime(cancelTime);
      this.setEndTime(cancelTime);
      this.setUpdateTime(cancelTime);
    }
  }

  public void skipNode(long skipTime) {
    this.setStatus(Status.SKIPPED);
    this.setStartTime(skipTime);
    this.setEndTime(skipTime);
    this.setUpdateTime(skipTime);
  }

  private void updatePastAttempts(List<Object> pastAttemptsList) {
    if (pastAttemptsList == null) {
      return;
    }

    synchronized (this) {
      if (this.pastAttempts == null) {
        this.pastAttempts = new ArrayList<ExecutionAttempt>();
      }

      // We just check size because past attempts don't change
      if (pastAttemptsList.size() <= this.pastAttempts.size()) {
        return;
      }

      Object[] pastAttemptArray = pastAttemptsList.toArray();
      for (int i = this.pastAttempts.size(); i < pastAttemptArray.length; ++i) {
        ExecutionAttempt attempt =
            ExecutionAttempt.fromObject(pastAttemptArray[i]);
        this.pastAttempts.add(attempt);
      }
    }
  }

  public int getRetries() {
    return inputProps.getInt("retries", 0);
  }

  public long getRetryBackoff() {
    return inputProps.getLong("retry.backoff", 0);
  }
}