BasicWorkflowValidator implements WorkflowValidator { @Override public List<String> validateWorkflow(Workflow workflow) { var workflowId = workflow.id(); var cfg = workflow.configuration(); final List<String> e = new ArrayList<>(); var componentId = workflowId.componentId(); empty(e, "component id", workflowId.componentId()); if (componentId.contains("#")) { e.add("component id cannot contain #"); } empty(e, "workflow id", workflowId.id()); if (!workflowId.id().equals(cfg.id())) { e.add("workflow id mismatch"); } upperLimit(e, cfg.id().length(), MAX_ID_LENGTH, "id too long"); upperLimit(e, cfg.commitSha().map(String::length).orElse(0), MAX_COMMIT_SHA_LENGTH, "commitSha too long"); upperLimit(e, cfg.secret().map(s -> s.name().length()).orElse(0), MAX_SECRET_NAME_LENGTH, "secret name too long"); upperLimit(e, cfg.secret().map(s -> s.mountPath().length()).orElse(0), MAX_SECRET_MOUNT_PATH_LENGTH, "secret mount path too long"); upperLimit(e, cfg.serviceAccount().map(String::length).orElse(0), MAX_SERVICE_ACCOUNT_LENGTH, "service account too long"); upperLimit(e, cfg.resources().size(), MAX_RESOURCES, "too many resources"); upperLimit(e, cfg.env().size(), MAX_ENV_VARS, "too many env vars"); upperLimit(e, cfg.env().entrySet().stream() .mapToInt(entry -> entry.getKey().length() + entry.getValue().length()).sum(), MAX_ENV_SIZE, "env too big"); upperLimit(e, cfg.retryCondition().map(String::length).orElse(0), MAX_RETRY_CONDITION_LENGTH, "retry condition too long"); cfg.resources().stream().map(String::length).forEach(v -> upperLimit(e, v, MAX_RESOURCE_LENGTH, "resource name too long")); cfg.offset().ifPresent(offset -> { try { TimeUtil.addOffset(ZonedDateTime.now(ZoneOffset.UTC), offset); } catch (DateTimeParseException ex) { e.add(format("invalid offset: %s", ex.getMessage())); } }); try { TimeUtil.cron(cfg.schedule()); } catch (IllegalArgumentException ex) { e.add("invalid schedule"); } cfg.runningTimeout().ifPresent(timeout -> lowerLimit(e, timeout, MIN_RUNNING_TIMEOUT, "running timeout is too small")); cfg.serviceAccount().ifPresent(serviceAccount -> { if (!validateServiceAccount(serviceAccount)) { e.add("service account is not a valid email address: " + serviceAccount); } }); cfg.retryCondition().ifPresent(retryCondition -> { try { new SpelExpressionParser().parseRaw(retryCondition); } catch (RuntimeException ex) { e.add(format("invalid retry condition: %s", ex.getMessage())); } }); validateConflictingExecConf(e, cfg); validateDockerConf(e, cfg); validateFlyteExecConf(e, cfg); return e; } BasicWorkflowValidator(DockerImageValidator dockerImageValidator); @Override List<String> validateWorkflow(Workflow workflow);  }
@Test public void validateInvalidOffset() { final List<String> errors = sut.validateWorkflow(Workflow.create( "test", TestData.HOURLY_WORKFLOW_CONFIGURATION_WITH_INVALID_OFFSET)); assertThat(errors, hasSize(1)); assertThat(errors.get(0), startsWith("invalid offset")); }
@Test public void validateInvalidDockerImage() { var errors = sut.validateWorkflow(Workflow.create("test", INVALID_DOCKER_WORKFLOW_CONFIGURATION)); assertThat(errors, containsInAnyOrder("invalid image: error")); }
@Test public void validateInvalidWorkflow() { final String id = Strings.repeat("id", 1024); final String schedule = Strings.repeat("schedule", 1024); final String offset = Strings.repeat("offset", 1024); final String commitSha = Strings.repeat("sha", 1024); final List<String> args = IntStream.range(0, 100).mapToObj(i -> "arg-" + i).collect(toList()); final Secret secret = Secret.create(Strings.repeat("foo", 1024), Strings.repeat("bar", 4711)); final String serviceAccount = Strings.repeat("account@abc.com", 1024); final List<String> resources = IntStream.range(0, 10) .mapToObj(i -> Strings.repeat("res-" + i, 100)).collect(toList()); final Map<String, String> env = IntStream.range(0, 2000).boxed() .collect(toMap(i -> "env-var-" + i, i -> "env-val-" + i)); final long envSize = env.entrySet().stream().mapToLong(e -> e.getKey().length() + e.getValue().length()).sum(); final Duration runningTimeout = Duration.ofSeconds(59L); var retryCondition = Strings.repeat("foo -> bar", 1024); final WorkflowConfiguration invalidConfiguration = WorkflowConfiguration.builder() .id(id) .schedule(Schedule.parse(schedule)) .offset(offset) .commitSha(commitSha) .dockerArgs(args) .secret(secret) .serviceAccount(serviceAccount) .resources(resources) .serviceAccount(serviceAccount) .env(env) .runningTimeout(runningTimeout) .retryCondition(retryCondition) .build(); final List<String> errors = sut.validateWorkflow(Workflow.create("test", invalidConfiguration)); final List<String> expectedErrors = ImmutableList.<String>builder() .add(limit("id too long", id.length(), MAX_ID_LENGTH)) .add("invalid schedule") .add(limit("commitSha too long", commitSha.length(), MAX_COMMIT_SHA_LENGTH)) .add(limit("secret name too long", secret.name().length(), MAX_SECRET_NAME_LENGTH)) .add(limit("secret mount path too long", secret.mountPath().length(), MAX_SECRET_MOUNT_PATH_LENGTH)) .add(limit("service account too long", serviceAccount.length(), MAX_SERVICE_ACCOUNT_LENGTH)) .add(limit("too many resources", resources.size(), MAX_RESOURCES)) .add(resources.stream().map(r -> limit("resource name too long", r.length(), MAX_RESOURCE_LENGTH)).toArray(String[]::new)) .add("invalid offset: Unable to parse offset period") .add(limit("too many env vars", env.size(), MAX_ENV_VARS)) .add(limit("env too big", envSize, MAX_ENV_SIZE)) .add(limit("running timeout is too small", runningTimeout, MIN_RUNNING_TIMEOUT)) .add("service account is not a valid email address: " + serviceAccount) .add(limit("retry condition too long", retryCondition.length(), MAX_RETRY_CONDITION_LENGTH)) .add("invalid retry condition: Expression [" + retryCondition + "] @4: EL1042E: Problem parsing right operand") .build(); assertThat(errors, containsInAnyOrder(expectedErrors.toArray())); }
@Test public void shouldNotAllowWorkflowIdMismatch() { var component = "test"; var workflowConfiguration = WorkflowConfiguration.builder() .id("foo") .schedule(Schedule.HOURS) .build(); @SuppressWarnings("ExtendsAutoValue") var workflow = new Workflow() { @Override public String componentId() { return component; } @Override public String workflowId() { return "bar"; } @Override public WorkflowConfiguration configuration() { return workflowConfiguration; } }; var errors = sut.validateWorkflow(workflow); assertThat(errors, contains("workflow id mismatch")); }
@Test public void shouldNotAllowEmptyWorkflowId() { var workflowConfiguration = WorkflowConfiguration.builder() .id("") .schedule(Schedule.HOURS) .build(); var errors = sut.validateWorkflow(Workflow.create("test", workflowConfiguration)); assertThat(errors, contains("workflow id cannot be empty")); }
@Test public void shouldNotAllowEmptyComponent() { var errors = sut.validateWorkflow(Workflow.create("", FULL_WORKFLOW_CONFIGURATION)); assertThat(errors, contains("component id cannot be empty")); }
@Parameters({"foo#bar", "#", "##"}) @Test public void shouldNotAllowComponentWithHash(String component) { assertThat(sut.validateWorkflow(Workflow.create(component, FULL_WORKFLOW_CONFIGURATION)), contains("component id cannot contain #")); }
@Parameters({"sa@.abc.com", "sa#@abc.com"}) @Test public void shouldRejectInvalidServiceAccount(String serviceAccount) { WorkflowConfiguration configuration = WorkflowConfigurationBuilder.from(FULL_WORKFLOW_CONFIGURATION) .serviceAccount(serviceAccount) .build(); assertThat(sut.validateWorkflow(Workflow.create("test", configuration)), contains("service account is not a valid email address: " + serviceAccount)); }
@Test public void shouldRejectSpaceServiceAccountWithTailingSpace() { String[] invalidServiceAccounts = {"sa@abc.com ", "sa@abc.com\n"}; for (String serviceAccount : invalidServiceAccounts) { WorkflowConfiguration configuration = WorkflowConfigurationBuilder.from(FULL_WORKFLOW_CONFIGURATION) .serviceAccount(serviceAccount) .build(); assertThat( sut.validateWorkflow(Workflow.create("test", configuration)), contains("service account is not a valid email address: " + serviceAccount)); } }
@Test @TestCaseName("{method} - {1}") @Parameters(source = InvalidFlyteConfExecArgsProvider.class) public void shouldRejectInvalidFlyteExecConf(WorkflowConfiguration configuration, String expectedError) { assertThat( sut.validateWorkflow(Workflow.create("test", configuration)), Matchers.allOf( Matchers.<String>hasSize(1), Matchers.contains(expectedError) ) ); }
@Test public void shouldRejectConflictingExecConf() { assertThat( sut.validateWorkflow(Workflow.create("test", DOCKER_AND_FLYTE_CONFLICTING_CONFIGURATION)), containsInAnyOrder("configuration cannot specify both docker and flyte parameters") ); }