Goals

Checklist

•  User stories documented (Albert/Vinisha) 
•  User stories reviewed (Nitin)
•  Design documented (AlbertShankar/VinishaKashif)
•  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 table sink. we He wants to configure the name of the database table and name of the table sink at on a per run basis and he gives those values as input before starting the run.  
2. Pipeline A pipeline developer wants to create a pipeline with a custom action at the start of the run, the . The custom action based on a some logic provides the name of the database to use as source and the name of the table to write in sink, the . The next stage in pipeline uses this information to read from the appropriate db 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 get be substituted for 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

ExpandedFunction Syntax : 
${macro-typemacroFunction(macro)}
 
ShorthandProperty lookup notationsyntax:
${macro}
 
Example Usage: 
${runtime-argumentsecure(hostnameaccessKey))} - get hostnameaccess key from secure runtimestore arguments
${wf-tokenlogicalStartTime(hostnametimeFormat))} - get hostname from workflow token
${secure(access_key)) - get access key from secure store 
${function_time(time_format)) - apply apply time function on the time_formattimeFormat provided and use the value. 
 
The Default (short-handshortHand) usage will readsubstitute fromarguments runtimeusing arguments,the havingfollowing anprecedence:
expandedCustom notationAction givesWorkflow-Token user> optionRuntime forArguments using> moreStored macroPreferences
types. 
Examples :
ipConfig : ${hostname}:${port}
JDBC connection string : jdbc:${jdbc-plugin}://${hostname}:${sql-port}/${db-name}
  

...

"stages": [
    {
        "name": "Database",
        "plugin": {
            "name": "Database",
            "type": "batchsource",
            "properties": {
				...
                "user": "${username}",
                "password": "${secure(sql-password)}",
                "jdbcPluginName": "jdbc",
                "jdbcPluginType": "${jdbc-type}",
                "connectionString": "jdbc:${jdbc-type}//
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)
 
Nested macros:
if a macro contains another macro, Example : ${secure(${user-name})} 
In the above example, we want to lookup the user-name in properties first, then use secure store to get the key/password for that user-name. this final key/password will be used for that field. 

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(...)).

Notes:

• The current implementation for macro substitution supports recursive expansion. That is, if a macro such as ${address} expands to ${hostname}:${port

...

• }, then ${hostname} and ${

...

• port} will be evaluated. However, this can lead to an infinite loop from circular macros, so we can add a maximum depth for expansion.

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

Scoping:

If the macro-substitution is performed at the DataPipeline app level, it will be possible to scope at stage name level if the user desires that. 

In our example config of JDBC source to Table sink, there is a common macro "${table-name}", if the user wants to provide a different name for the table-name in Table Sink, he can use scoping.

Example for Scoping:

Provided runtime arguments:
 
Key : table-name, value : employees 
Key : TableSink:table-name, value : employee_sql 
 
table-name is the macro name that is used in both DBSource stage and TableSink stage. 
if user wants to provide a special value for macro "table-name" to be used in TableSink, he will prefix stage-name before the macro name separated by the delimiter (colon).

Hydrator Plugin Changes

Currently when we deploy a pipeline,  configurePipeline is called on each plugin. we perform few validations in configure stage, specifically for schema, syntax for scripts, etc. In some Plugins we also create 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 at configure stage. 

Deferring dataset creation in prepareRun will required changes to BatchSinkContext to have a new method. 

@Beta
public interface BatchSinkContext extends BatchContext {
// new method
void createDataset(String datasetName, String typeName, DatasetProperties properties);
 
//existing methods
@Deprecated
void addOutput(String datasetName);
...
}

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 disallow this, so this addition will only be in BatchSinkContext and not BatchContext.

PluginConfigurer can be made not to extend DatasetConfigurer as it can no longer create dataset.

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. 

There are two way of handling this.

1) At Platform Level

2) At DataPipeline App level

Platform Level Substitution:

Plugins can use "@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 plugin instance is instantiated at configure time, macros cannot be substituted. 

• If we want to keep the field with macro as is,  then the field has to be always a string, this is limiting for plugin developers. as they have to do type casting themselves for using macro on fields with other types than String.
• By having a configure-value we can work-around that, so the plugin developer has to know that this value will be used at configure time. but this might seem unnecessary for the plugin developer as this configure-value isn't very useful except to avoid failure at configure time.

public class TableSinkConfig extends "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"
            }
        }
    }
]

CDAP API Changes

Macro Annotation:

Plugins can use an "@Macro" annotation to specify if a plugin field can be a macro.  

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.

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.FIELD)
/**
 * Annotation for a plugin field to specify if it can support macro
 */
public @interface Macro {
}

/**
 * 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 isMacro;
  ...
}

Example Macro Usage in Plugin:

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.
this field uses@Macro the
value "test" atprivate configString time.name;

  @Macro(enabled=true, configValue="test") 
  private String name;

  @Name(Properties.Table.PROPERTY_SCHEMA)
  @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;
}

...

PluginConfig Changes:

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface Macro@Beta
public abstract class PluginConfig extends Config implements Serializable {

  /**
   * DefaultReturns statusthe if macro is enabled.
   */{@link PluginProperties}.
   */
  public final PluginProperties getProperties() {
  boolean DEFAULT_STATUS =return falseproperties;
  }
 
  /**
   * Returns Returnsfalse if the macrofield is enabled. Default is 'false'.
   */
  boolean enabled() default DEFAULT_STATUS;
  
  /**
   * returns the value to be used at config time
   not annotated as a macro. If field is annotated as macro, 
   * the field value is checked if its a macro, based on the check it returns true/false.  
   */
  String configValue();
}
 

/**
 * 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;
  ...
}

Notes

This will require a CDAP platform level change as its a new annotation.

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

During configure time, if a field is macro enabled, PluginInstantiator would try to use configValue as the value for the field in order to instantiate the plugin.

Pipeline developer should take care that configurePipeline doesn't use the config fields which are macro enabled.  

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.

App Level Substitution:

One possibility is for substitution to be implemented at the app level. This would be ideal if we want to keep the concept of macros Hydrator-specific. If substitution were to occur at the app level, then the user would dictate which fields will be macro-substitutable through the plugin configuration UI. In order to allow non-string properties to be substitutable, the user must provide a default value along with the macro through the UI. For example, it a user enters the "port" property as: ${port}, the UI will provide a way for the user to enter a default port value. Creating a DB batch source would yield the following configuration JSON:

"plugin": {
	"name": "Database",
	"type": "batchsource",
	"properties": {
		"user": "${username}",
		"password": "${secure(sql-password)}",
		"jdbcPluginName": "jdbc",
		...
		"importQuery": "select * from ${table-name};"
		...
		"macroDefaults": "{
			\"user\": \"admin\",
			\"password\": \"pw1234\",
			\"importQuery\": \"select * from test;\"
		}"
	}
}

In this case, the app understands from configuration the fields that are macros and the default values to use for those fields during configure time. 

This would require a new method in PluginContext to accept key and value pairs to substitute for plugin properties.

@Beta
public interface PluginContext {
 
// existing methods
PluginProperties getPluginProperties(String pluginId);
<T> Class<T> loadPluginClass(String pluginId);
<T> T newPluginInstance(String pluginId) throws InstantiationException;
 
/**
 * Creates a new instance of a plugin. The instance returned will have the {@link PluginConfig} setup with
 * {@link PluginProperties} provided at the time when the
 * {@link PluginConfigurer#usePlugin(String, String, String, PluginProperties)} was called during the
 * program configuration time. In addition the parameter pluginProperties can be used to override the existing
 * plugin properties in config, with which the plugin instance will have substituted plugin properties.
 *
 * @param pluginId the unique identifier provide when declaring plugin usage in the program.
 * @param <T> the class type of the plugin
 * @param pluginProperties the properties to override existing plugin properties before instance creation.
 * @return A new instance of the plugin being specified by the arguments
 *
 * @throws InstantiationException if failed create a new instance
 * @throws IllegalArgumentException if pluginId is not found
 * @throws UnsupportedOperationException if the program does not support plugin
 */
<T> T newPluginInstance(String pluginId, Map<String, String> pluginProperties) throws InstantiationException;
 

Configure time:

The app can all this new method with macroDefault values, so plugin instance creation will use macro default values for those config fields.

Run time:

The app performs substitution for the properties with macros using the value from runtime arguments ( or workflow token) and calls the method with the field name's and substitution value's. 

Custom Action Setting Config values:

Custom action can use workflow token to set values for field names.  

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

If pipeline builder wants to use workflow token sent from custom action to be used as value for fields, then he uses macro-type token in his fields as above.

Context has access to workflow token and we should be able to use Workflow tokens similar to runtime arguments for substitution. 

Documentation Changes (wip)

Regardless of where the substitution occurs, the guidelines for creating Hydrator plugins would have to change. For existing plugins, any validation for properties that are macro-substitutable that exists in configurePipeline must be moved to prepareRun.

Reference

Many plugins have properties that are used in constructing or validating a schema at configure time. These fields need to have macros disabled to allow this. The following plugins and fields would be affected:

...

Copybook contents are converted to an InputStream and used to get external records, which are in turn used to add fields to the schema.

...

public final boolean isMacro(String fieldName) {
       ...
   }
}

The method will return whether or not the property with the provided fieldName contains 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 for simple pipelines.  

private final TableSinkConfig sinkConfig;
@Override
void configurePipeline(PipelineConfigurer pipelineConfigurer) {
  if (!sinkConfig.isMacro("name")) {
 	// create dataset if the datasetName field is not a macro
    pipelineConfigurer.createDataset(name, datasetType, DatasetProperties.builder().addAll(properties).build());
  }
  ...
}

Notes:

• To allow macro substitution of non-String properties, any properties configured with a macro will have a placeholder default value at configure time which is different from runtime.
• For primitive types, this would be Java's default value, Example: integer field will have a value of 0. objects, would be null.
• This is fine because we are exposing a method to check whether a field is a macro or not in pipelineConfigurer. so plugin developers can check at configure time if a field is macro before performing appropriate action using that field. 
• During runtime, macro fields will be substituted and plugin instance will have fields with substituted values.

@Beta
public interface PluginContext {
 
/**
 * Creates a new instance of a plugin. The instance returned will have the {@link PluginConfig} setup with
 * {@link PluginProperties} provided at the time when the
 * {@link PluginConfigurer#usePlugin(String, String, String, PluginProperties)} was called during the
 * program configuration time. If a plugin field has a macro, 
 * the parameter evaluator is used to evaluate the macro and the evaluated value is used for the plugin field.
 *
 * @param pluginId the unique identifier provide when declaring plugin usage in the program.
 * @param evaluator the macro evaluator that's used to evaluate macro for plugin field if macro is supported on those fields. 
 * @param <T> the class type of the plugin
 * @return A new instance of the plugin being specified by the arguments
 *
 * @throws InstantiationException if failed create a new instance
 * @throws IllegalArgumentException if pluginId is not found
 * @throws UnsupportedOperationException if the program does not support plugin
 */
<T> T newPluginInstance(String pluginId, MacroEvaluator evaluator) throws InstantiationException;

@Beta
interface MacroEvaluator {  
 
/**
 * lookup the property and return the value corresponding to the property.
 */ 
 String evaluate(String property);  
/**
 * Use the macro function and call the function with provided arguments, 
 * apply appropriate function or use properties in default case and return the evaluated value
 */ 
 String evaluate(String macroFunction, String... arguments);  
} 

@Beta
public interface BatchContext extends DatasetContext, TransformContext {

/**
 * create dataset identified by datasetName, typeName with properties.
 */
void createDataset(String datasetName, String typeName, DatasetProperties properties);

/**
 * return true if dataset with datasetName exists 
 */
boolean datasetExists(String datasetName);

...
}

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

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);
	}
} 

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 {	
	TimeZone timeZone;
	long logicalStartTime;
	Function<String, String> timezoneFunction;
	String getValue(String arguments) {
		return timezoneFunction.apply(arguments);
	}
}