Introduction

This design provides the capability for passing information between the triggering pipeline and triggered pipeline based on the program status based scheduling.

Goals

Provides clear API and backend design for passing program status change event payload between pipelines. 

User Stories 

• Pipeline B is triggered when Pipeline A completes, and Pipeline B also needs the username and password used by Pipeline A
• Pipeline B is triggered when Pipeline A completes, and Pipeline B also needs the stream id Pipeline A reads from
• Pipeline B is triggered when Pipeline A completes, and Pipeline B also needs the schema of a sink Pipeline A writes to.
  
  

Design

A new method will be provided in RuntimeContext to return a new API class TriggeringScheduleInfo to provide information of the schedule which launches the program. All the runtime arguments, stage configurations, user workflow tokens of the pipeline run which triggers the schedule will be included in the . To use these properties, users can define a mapping between a certain property from the triggering program and a runtime argument in the triggered program. Such mapping should be stored in the schedule properties with the key "triggering.properties.mapping". The app containing the triggered program should be able to decode the value contained in the field "triggering.properties.mapping" and get the corresponding properties from TriggeringScheduleInfo and use the values as runtime arguments according to the properties mapping.

For pipeline app, the syntax for referring to different properties in the  filed "triggering.properties.mapping" is defined below:

The syntax for runtime args from the triggering pipeline is: runtime-arg:<namespace>:<pipeline-name>#<runtime-arg-key>

Stage configuration from triggering pipeline: pipeline-config:<namespace>:<pipeline-name>#<pipeline-stage>:<stage-config-key>

User tokens from the triggering pipeline: token:<namespace>:<pipeline-name>#<node-name>:<user-token-key> 

Approach

Here's an example use case to illustrate the approach: Pipeline A in namespace Default triggers Pipeline B when Pipeline A completes. Pipeline B needs the runtime args hostname from Pipeline A as the value of its host field in runtime args. For instance, For instance, Pipeline A contains runtime args pair "hostname" -> "wiki.cask.co" , then in Pipeline B, the value of "host" should be "wiki.cask.co"

When setting up program status schedule, an entry "runtime-arg:default:A#hostname" -> "host" will be stored in a map, and this map will be converted to a JSON String to be stored as a value of the field "triggering.properties.mapping" in the schedule properties. When the notification of Pipeline A's completion triggers the schedule to launch Pipeline B, in the Pipeline app of Pipeline B, the field "triggering.properties.mapping" is extracted from schedule properties. From the decoded map, Pipeline app recognizes the key "runtime-arg:default:A:#hostname" means a runtime argument with key "hostname" in the Pipeline with name A  in the default namespace. Pipeline app thus looks this runtime argument value in Pipeline A from TriggeringScheduleInfo from the RuntimeContext , and provides it as the value for runtime argument "host" in Pipeline B

API changes

New Programmatic APIs

/**
 * This interface represents a context for a processor or elements of a processor.
 */
public interface RuntimeContext {

...

  /**
   * @return {@link TriggeringScheduleInfo} if the program is triggered by a schedule. Otherwise, returns {@code null}.
   */
  @Nullable
  TriggeringScheduleInfo getTriggeringScheduleInfo();
 
...
 
}

/**
 * The information of a schedule to be passed to the program launched by it
 */
public class TriggeringScheduleInfo {

  private final String name;
  private final String description;
  private final Map<String, String> properties;
  private final Map<String, List<ApplicationSpecification>> namespaceApplicationSpecificationsMap;
  private final Map<String, WorkflowToken> userWorkflowTokenMap;
  private final Map<String, Map<String, String>> runtimeArgumentsMap;

  public TriggeringScheduleInfo(String name, String description, Map<String, String> properties, 
                                Map<String, List<ApplicationSpecification>> namespaceApplicationSpecificationsMap, 
                                Map<String, WorkflowToken> userWorkflowTokenMap, 
                                Map<String, Map<String, String>> runtimeArgumentsMap) {
    this.name = name;
    this.description = description;
    this.properties = properties;
    this.namespaceApplicationSpecificationsMap = namespaceApplicationSpecificationsMap;
    this.userWorkflowTokenMap = userWorkflowTokenMap;
    this.runtimeArgumentsMap = runtimeArgumentsMap;
  }

  /**
   * @return Schedule's name, which is unique in an application.
   */
  public String getName() {
    return name;
  }

  /**
   * @return Description of the schedule.
   */
  public String getDescription() {
    return description;
  }

  /**
   * @return Properties of the schedule.
   */
  public Map<String, String> getProperties() {
    return properties;
  }

  /**
   * @return A map with namespace names as keys, and the list of application specification of the parent applications
   *         which contain the programs that trigger the schedule as values.
   */
  Map<String, List<ApplicationSpecification>> getNamespaceApplicationSpecificationsMap() {
    return namespaceApplicationSpecificationsMap;
  }

  /**
   * @return A map with workflow ID String as keys in the format of 
   *         "namespace:application:application-version:workflow-name", or  "namespace:application:workflow-name" 
   *         if the application version is of default version "-SNAPSHOT", and the corresponding user-scope 
   *         workflow token as values.
   */
  public Map<String, WorkflowToken> getUserWorkflowTokenMap() {
    return  userWorkflowTokenMap;
  }

  /**
   * @return A map with program ID String as keys in the format of 
   *         "namespace:application:application-version:program-type:program-name", 
   *         or  "namespace:application:program-type:program-name" if the application version is of 
   *         default version "-SNAPSHOT", and the runtime arguments key-value pairs as values.
   */
  public Map<String, Map<String, String>> getRuntimeArgumentsMap() {
    return runtimeArgumentsMap;
  }
}

UI Impact or Changes

Security Impact 

What's the impact on Authorization and how does the design take care of this aspect

Impact on Infrastructure Outages 

System behavior (if applicable - document impact on downstream [ YARN, HBase etc ] component failures) and how does the design take care of these aspect

Test Scenarios

Releases

Release X.Y.Z

Release X.Y.Z

Related Work

• Work #1
• Work #2
• Work #3

Future work