Goals

Checklist

• User stories documented (Albert/Vinisha) 
• User stories reviewed (Nitin)
• Design documented (Albert/Vinisha)
• Design reviewed (Terence/Andreas)
• Feature merged ()
• Examples and guides ()
• Integration tests () 
• Documentation for feature ()
• Blog post

Use Cases

1. A pipeline developer wants to create a pipeline that has several configuration settings that are not known at pipeline creation time, but that are set at the start of the each pipeline run. 
   1. A pipeline developer wants to create a pipeline that reads from a database source and writes to a table sink. He wants to configure the name of the database table and name of the table sink on a per run basis and he gives those values as input before starting the run.  
2. A pipeline developer wants to create a pipeline with a custom action at the start of the run. The custom action based on some logic provides the name of the database to use as source and the name of the table to write in sink. The next stage in pipeline uses this information to read from the appropriate database source and write to the table sink.  
   

User Stories

1. As a pipeline developer, I want to be able to configure a plugin property to some value that will be substituted each run based on the runtime arguments.
2. As a pipeline operator, I want to be able to set arguments for the entire pipeline that will be used for substitution.
3. As a pipeline operator, I want to be able to set arguments for a specific stage in the pipeline that will be used for substitution.
4. As a plugin developer, I want to be able to write a code that is executed at the start of the pipeline and sets arguments for the rest of the run.

Design

Macros Syntax

Function Syntax : 
${macroFunction(macro)}
 
Shorthand notation:
${macro}
 
Example Usage: 
${secure(accessKey)} - get access key from secure store 
${functionTime(timeFormat)} - apply time function on the time_format provided and use the value. 
 
The Default (shortHand) usage will substitute arguments using the following precedence:
Custom Action Workflow-Token > Runtime Arguments > Stored Preferences
 
Examples:
ipConfig: ${hostname}:${port}
JDBC connection string : jdbc:${jdbc-plugin}://${hostname}:${sql-port}/${db-name}
 
Using the expanded syntax allows additional logic to be applied to the macro arguments through a macro function.
Escaping can be supported using the \ (backslash) character (e.g. \${hostname} will not be substituted)

The shorthand notation supports retrieval precedence to limit the exposure of underlying workflow-tokens and runtime-arguments to pipeline operators. The "functionTime" macro function uses the logical start time of a run to perform the substitution. This is an example of a macro function that is not just a key-value lookup but allows for extra logic to be performed before a value is returned. For now, the implementation will only support the following macro functions: runtime-arguments. Once the secure store API is available, it will also support secure store. In the future, we can see if we will allow developers to create custom macro functions (similar to functionTime(...)).

"stages": [
    {
        "name": "Database",
        "plugin": {
            "name": "Database",
            "type": "batchsource",
            "properties": {
				...
                "user": "${username}",
                "password": "${secure(sql-password)}",
                "jdbcPluginName": "jdbc",
                "jdbcPluginType": "${jdbc-type}",
                "connectionString": "jdbc:${jdbc-type}//${hostname}:${port}/${db-name}",
                "importQuery": "select * from ${table-name};"
            }
        }
    },
    {
        "name": "Table",
        "plugin": {
            "name": "Table",
            "type": "batchsink",                                        
            "properties": {
                "schema": "{\"type\":\"record\",\"name\":\"etlSchemaBody\",
                \"fields\":[{\"name\":\"name\",\"type\":\"string\"},
                {\"name\":\"age\",\"type\":\"int\"},{\"name\":\"emp_id\",\"type\":\"long\"}]}",
                "name": "${table-name}",
                "schema.row.field": "name"
            }
        }
    }
]

Hydrator Plugin Changes

Currently when we deploy a pipeline,  configurePipeline is called on each plugin. we perform few validations in the configure stage, specifically for schema strings, syntax for scripts, etc. In some Plugins we also create a dataset if the dataset doesn't already exist. 

The dataset to write to can be macro-substituted, so we have to defer dataset creation to prepareRun rather than doing this in the configure stage. 

Deferring dataset creation in prepareRun will require adding a new method to BatchContext. 

@Beta
public interface BatchContext extends DatasetContext, TransformContext {

// new method
void createDataset(String datasetName, String typeName, DatasetProperties properties);
 
// existing methods
long getLogicalStartTime();

/**
 * Returns runtime arguments of the Batch Job.
 *
 * @return runtime arguments of the Batch Job.
 */
Map<String, String> getRuntimeArguments();

/**
 * Updates an entry in the runtime arguments.
 *
 * @param key key to update
 * @param value value to update to
 * @param overwrite if {@code true} and if the key exists in the runtime arguments, it will get overwritten to
 *                  the given value; if {@code false}, the existing value of the key won't get updated.
 */
void setRuntimeArgument(String key, String value, boolean overwrite);

/**
 * Returns the hadoop job.
 * @deprecated this method will be removed.
 */
@Deprecated
<T> T getHadoopJob();
...
}

Currently if a stream given in stream source or table given in table source doesn't exist, we create a new stream/table. We want to allow table creation as we want to create external dataset for sources, but disallow stream creation, so we are adding only createDataset to the BatchContext.

PluginConfigurer can be made not to extend DatasetConfigurer later as we no longer want to allow creating a dataset in configure.

However there are certain fields which are used to determine the schema in the plugin and those cannot be macro-substituted as schema validation is essential during configure time and we want to disallow macro usage for them. 

Platform Level Substitution:

Plugins can use an "@Macro" annotation to specify if a plugin field can be a macro and also provides a configure-value to use at configure time to instantiate the plugin.  

When a plugin instance is instantiated at configure time, macros cannot be substituted as the values to substitute have not been specified yet. By default, macros will be disabled for properties. This is to prevent new plugin developers from having to worry about undefined behavior if they did not consider or are not familiar with macros.

public class TableSinkConfig extends PluginConfig {
  @Name(Properties.Table.NAME)
  @Description("Name of the table. If the table does not already exist, one will be created.")
  // The name of the table can be specified by a runtime macro, by default macros are disabled for fields.
  @Macro 
  private String name;

  @Name(Properties.Table.PROPERTY_SCHEMA)
  @Description("schema of the table as a JSON Object. If the table does not already exist, one will be " +
    "created with this schema, which will allow the table to be explored through Hive. If no schema is given, the " +
    "table created will not be explorable.")
  @Nullable
  private String schemaStr;

  @Name(Properties.Table.PROPERTY_SCHEMA_ROW_FIELD)
  @Description("The name of the record field that should be used as the row key when writing to the table.")
  private String rowField;
}

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
public @interface Macro {

  /**
   * Default status if macro is enabled.
   */
  boolean DEFAULT_STATUS = false;

  /**
   * Returns if macro is enabled. Default is 'false'.
   */
  boolean enabled() default DEFAULT_STATUS;

}
 

/**
 * Contains information about a property used by a plugin.
 */
@Beta
public class PluginPropertyField {

  private final String name;
  private final String description;
  private final String type;
  private final boolean required;
  // returns true if this field can accept macro
  private final boolean macroEnabled;
  ...
}

This will require a CDAP platform level change as it's a new annotation.

PluginInstantiator has to understand if macro is enabled and set fields appropriately.. 

During configure time in the configurePipeline method, if a field is macro enabled, the property should not be validated as the macro has not been provided a substitutable value.

PluginConfigurer Changes:

PluginConfigurer will expose the following field

public interface PluginConfigurer extends DatasetConfigurer {
  // ...
 
  boolean isMacro(String fieldName);
}

The method will return whether or not the property with the provided fieldName contained a macro at configure time. We don't want to force deferring macros to runtime in the case that a field is macroable but actually has no macro provided in its configuration. This allows optional checking of properties at configure time:

@Override
void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  if (!pipelineConfigurer.isMacro("name")) {
    // perform some operation using the "name" property
  }
  // ...
}

 To allow macro substitution of non-String properties, any properties configured with a macro will have a placeholder default value at configure time. For primitive types, this would be Java's default value. For objects, this would be null. This is ok because we are exposing a method to check whether a property is safe to validate at configure time through the pipelineConfigurer object.

BatchContext Changes

Given that a dataset could be created at configure time if no fields are provided macros, a check should be available for plugin developers to see whether the dataset already exists at runtime. We can do this by altering the runtime context object passed in to the prepareRun method. As the object extends BatchContext which extends DatasetContext, we can create a new method in BatchContext that checks for the existence of a dataset.

public interface BatchContext extends DatasetContext, TransformContext {
	// ...
 
	boolean datasetExists(String datasetName);
}

This method will return whether or not the dataset with the provided datasetName already exists and can be used in prepareRun:

@Override
public void prepareRun(BatchSinkContext context) {
  if (!context.datasetExists(config.getName())) {
    pipelineConfigurer.createDataset(config.getName(), ...);
  }
  // ...
}

During runtime, PipelineInsantiator would get config fields and values to substitute and can use that information to substitute macro appropriately and return an instantiated plugin.

"plugin": {
	"name": "Database",
	"type": "batchsource",
	"properties": {
		"user": "${wf-token(username)}",
		"password": "${secure(sql-password)}",
		"jdbcPluginName": "jdbc",         
		"importQuery": "select * from ${wf-token(table-name)};"
	}
}

interface MacroContext {	
	/**
	 * Given the macro key, return the substituted value
     */ 
	String getValue(String macroKey);
}

Based on the macro type, one of the below MacroContext's will be used to get the value for macro. 
 
DefaultMacroContext implements MacroContext {
	Map<String, String> runtimeArguments;
	String getValue(String macroKey) {
		return runtimeArguments.get(macroKey);
	}
}

SecureMacroContext implements MacroContext {
	SecureStore secureStore;
	String getValue(String macroKey) {
		return secureStore.get(macroKey);
	}
}

RuntimeFunctionMacro implements MacroContext {	
	long logicalStartTime;
	Function<String, String> timezoneFunction;
	String getValue(String arguments) {
		return timezoneFunction.apply(arguments);
	}
}