Skip to content

Fix: Enforce wrangler directive config in wrangler service and pipeline transform - #1039

Open
riyaa14 wants to merge 1 commit into
developfrom
fix/wrangler-config-api
Open

Fix: Enforce wrangler directive config in wrangler service and pipeline transform#1039
riyaa14 wants to merge 1 commit into
developfrom
fix/wrangler-config-api

Conversation

@riyaa14

@riyaa14 riyaa14 commented Jul 14, 2026

Copy link
Copy Markdown

Overview

This PR resolves issues with directive exclusions not being enforced in both the CDAP Wrangler service (wrangler-service) and the CDAP Wrangler pipeline transform (wrangler-transform).

Problem Statement

Directive exclusions and configuration rules were failing to enforce in two key areas:

  1. CDAP Wrangler Service (wrangler-service): Directive execution and validation endpoints were not loading the system-level DirectiveConfig from backend storage (DB/store). Consequently, directive checks and validations performed via the service did not respect restriction rules.
  2. CDAP Wrangler Transform (wrangler-transform): The pipeline transform initialized its directive parser (GrammarBasedParser) with NoOpDirectiveContext, where isExcluded() is hardcoded to return false. As a result, excluded directives were executed silently on pipeline task workers. Furthermore, in distributed execution mode, worker nodes cannot directly invoke HTTP calls to the system CDAP Wrangler service.

Solution

1. Fix for CDAP Wrangler Service

  • Updated service handlers (such as AbstractDirectiveHandler) to fetch DirectiveConfig from backend storage and bind it to the execution context so that UI interactions and directive evaluations via the service respect exclusions.

2. Fix for CDAP Wrangler Transform (Pipeline Runs)

  • Master Node Setup (prepareRun): During pipeline preparation on the master node, Wrangler.prepareRun() calls the system CDAP Wrangler service endpoint (/config) via StageContext.openConnection() to retrieve the active DirectiveConfig.
  • Runtime Argument Propagation: prepareRun() serializes the fetched DirectiveConfig as JSON into a CDAP pipeline runtime argument (pipeline.wrangler.directive.config).
  • Worker Node Execution (initialize & getRecipeParser):
    • On worker initialization, getSystemDirectiveConfigFromRuntimeArgs() reads the DirectiveConfig directly from the CDAP runtime arguments, bypassing the need for worker nodes to perform HTTP calls to the service.
    • Initialized GrammarBasedParser with ConfigDirectiveContext(directiveConfig) instead of NoOpDirectiveContext.
    • Added eager recipe parsing during initialize() to fail fast if an excluded or restricted directive is included in the recipe before processing pipeline records.

Testing

  1. Wrangler Interactive UI
image
  1. Pipeline Preview Run
image
  1. Pipeline Execution
image

Unit Testing

  • Added unit tests in WranglerDirectiveExclusionTest.
  • Ran existing unit test suite across wrangler-api, wrangler-service, and wrangler-transform modules.

@riyaa14 riyaa14 added the build Triggers unit test build label Jul 14, 2026

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request implements a multi-tier configuration resolution for Wrangler directive exclusions, allowing configurations to be passed via pipeline runtime arguments to avoid unreachable HTTP service calls on workers. The feedback recommends refactoring getSystemDirectiveConfigFromRuntimeArgs to accept StageContext as a parameter and fall back to plugin properties. This refactoring completes the three-tier config resolution, enables validation during pipeline configuration, and eliminates the need for the thread-unsafe stageContext instance field.

Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
Comment thread wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java Outdated
@riyaa14
riyaa14 force-pushed the fix/wrangler-config-api branch from 67a3374 to f33c3a9 Compare July 15, 2026 07:00
import io.cdap.cdap.api.data.schema.Schema;
import io.cdap.cdap.api.service.http.SystemHttpServiceContext;
import io.cdap.cdap.features.Feature;
import io.cdap.cdap.spi.data.transaction.TransactionRunners;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SPI shouldn't be used in Handlers.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Moved TransactionRunners to ConfigStore.

ConfigStore store = ConfigStore.get(context);
return store.getConfig();
});
} catch (Exception e) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Catch specific expection

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.


protected DirectiveConfig getDirectiveConfig() {
try {
return TransactionRunners.run(getContext(), context -> {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

TransactionRunner is not needed here, since ConfigStore already uses it internally.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The ConfigStore in CDAP uses TransactionRunners, but in Wrangler codebase, I didn't find any implementation of ConfigStore where TransactionRunners is being used, thus we need to add it. As suggested, I have removed TransactionRunners from Handler classes, and moved it to ConfigStore.

"precondition", "false",
"workspaceId", workspaceId);
DirectiveConfig directiveConfig = getDirectiveConfig();
Map<String, String> properties = new HashMap<>();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Use ImmutableMap

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

*/
private DirectiveConfig fetchDirectiveConfigFromService(final StageContext context) {
LOG.debug("Fetching directive config from DataPrep service.");
HttpURLConnection connection = null;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Check if RemoteClient can be used?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

RemoteClient is present in io.cdap.cdap.common.internal.remote, which is not exposed to plugins and external applications. According to CDAP Docs on Class Loading, only public API modules are available to plugins.

OpenConnection method present in cdap-api module is thus more suitable for plugin to service connection.

@riyaa14 riyaa14 changed the title Fix: Enforce directive exclusions in CDAP Wrangler service and pipeline transform Fix: Enforce wrangler directive config in wrangler service and pipeline transform Jul 22, 2026
@riyaa14
riyaa14 force-pushed the fix/wrangler-config-api branch from f33c3a9 to 222f064 Compare July 24, 2026 06:44
.withConfigProperty(Config.NAME_DIRECTIVES);
}
DirectiveConfig directiveConfig = getSystemDirectiveConfigFromRuntimeArgs(null);
DirectiveContext directiveContext = new ConfigDirectiveContext(directiveConfig);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

use
DirectiveContext directiveContext = new ConfigDirectiveContext(DirectiveConfig.EMPTY);

for better readability and there is no need to call getSystemDirectiveConfigFromRuntimeArgs

super.prepareRun(context);

DirectiveConfig systemConfig = fetchDirectiveConfigFromService(context);
DirectiveConfig configToSet = systemConfig != null ? systemConfig : DirectiveConfig.EMPTY;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't this risky?

If there are transient network failure or any unintended connectivity errors, the restricted directive will be bypassed and it may have security vulnerability .

In such cases, we should fail the pipeline.

Similar is the case with MACROs:

  • they retry the network call for few minutes, if it is not resolved, we fail the pipeline.

* @param context the stage context
* @return the fetched DirectiveConfig or null if unreachable
*/
private DirectiveConfig fetchDirectiveConfigFromService(final StageContext context) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should be refactoring this into something similar to : AbstractServiceRetryableMacroEvaluator

Need be abstract, can be a impl class. can be DataPrepServiceClient or ServiceRetryableClient

reason :

  • Single responsibility ( instead of adding networking code in wrangler.java )
  • we need to have sufficient retry logic like exponential delay upto 5 minutes.
  • better exception handling like throwing IllegalStateException upon 5 minute failure..which will fail the pipeline and message should appead in pipeline logs.
  • intermediate exceptions in debug level for each try.

This will keep the wrangler file much clean.

please refer io.cdap.cdap.common.service.RetryStrategies for some use cases.


// Configuration specifying the dataprep application and service name.
private static final String APPLICATION_NAME = "dataprep";
private static final String SERVICE_NAME = "service";

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move these to the new class suggest below.

private static final String APPLICATION_NAME = "dataprep";
private static final String SERVICE_NAME = "service";
private static final String CONFIG_METHOD = "config";
private static final int HTTP_CONNECT_TIMEOUT_MS = 5000;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Move these to the new class suggest below. and these should be configurable

Since we do not have access to cconf , we cannot directly use it and let's support runtime arguments to handle this incase we need to modify these values in future.

private static final int DEFAULT_CONNECT_TIMEOUT_MS = 15000; // Matches cdap-default.xml http.client.connection.timeout.ms
private static final int DEFAULT_READ_TIMEOUT_MS = 60000;    // Matches cdap-default.xml http.client.read.timeout.ms

private int getConnectTimeout(StageContext context) {
  String val = context.getArguments().get("wrangler.service.connect.timeout.ms");
  return !Strings.isNullOrEmpty(val) ? Integer.parseInt(val) : DEFAULT_CONNECT_TIMEOUT_MS;
}

private int getReadTimeout(StageContext context) {
  String val = context.getArguments().get("wrangler.service.read.timeout.ms");
  return !Strings.isNullOrEmpty(val) ? Integer.parseInt(val) : DEFAULT_READ_TIMEOUT_MS;
}

@sahusanket

Copy link
Copy Markdown
Contributor

Please add the use case of runtime argument wrangler.directive.config in exclusion-and-aliasing.md with a small example body and usecase.

try {
return ConfigStore.get(getContext()).getConfig();
} catch (IOException | RuntimeException e) {
LOG.warn("Failed to fetch directive exclusions from database. Using empty configuration.", e);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If there was a failure reading from Database, IOException would be thrown. You are ignoring the exception and returning default configuration, which is incorrect.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, my bad, changed - throwing IOException for this method in case of error rather than handling locally.

this.transactionRunner = null;
}

public ConfigStore(TransactionRunner transactionRunner) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a need for separate constructor?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is just to make sure adding TransactionRunners is done in backward compatible way. I didn't want to change the public get method and constructor which exposed by ConfigStore currently. Thus, added new ones rather than updating the existing get method and constructor.

But rethinking this, maybe we can make this change and go forward with updating the current methods because ConfigStore is in wrangler-storage module, which is an internally used module and thus changing the constructor and methods shouldn't be a breaking change externally. If any custom plugin uses wrangler, it should be using wrangler-api module. Any insights of yours would be helpful.

}
}

public static ConfigStore get(TransactionRunner transactionRunner) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a need for a separate overload?

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same reason as mentioned in #1039 (comment)

@riyaa14

riyaa14 commented Jul 27, 2026

Copy link
Copy Markdown
Author

Please add the use case of runtime argument wrangler.directive.config in exclusion-and-aliasing.md with a small example body and usecase.

wrangler.directive.config is strictly an internal detail, for propagating directiveConfig from master to worker nodes so that worker nodes don't have to make HTTP call to fetch it, it not a user facing runtime argument.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

build Triggers unit test build

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants