ExtendedJobSanitizer implements EntitySanitizer { @Override public <T> Optional<T> sanitize(T entity) { T sanitized = entitySanitizer.sanitize(entity).orElse(entity); if (sanitized instanceof com.netflix.titus.api.jobmanager.model.job.JobDescriptor) { sanitized = (T) sanitizeJobDescriptor((JobDescriptor) sanitized); } return entity == sanitized ? Optional.empty() : Optional.of(sanitized); } ExtendedJobSanitizer(JobManagerConfiguration jobManagerConfiguration,
                                JobAssertions jobAssertions,
                                @Named(JOB_STRICT_SANITIZER) EntitySanitizer entitySanitizer,
                                DisruptionBudgetSanitizer disruptionBudgetSanitizer,
                                @Named(SECURITY_GROUPS_REQUIRED_FEATURE) Predicate<JobDescriptor> securityGroupsRequiredPredicate,
                                @Named(ENVIRONMENT_VARIABLE_NAMES_STRICT_VALIDATION_FEATURE) Predicate<JobDescriptor> environmentVariableNamesStrictValidationPredicate,
                                TitusRuntime titusRuntime); @Override Set<ValidationError> validate(T entity); @Override Optional<T> sanitize(T entity);  }
@Test public void testDiskSizeIsChangedToMin() { JobDescriptor<BatchJobExt> jobDescriptor = newJobDescriptorWithDiskSize(100); when(configuration.getMinDiskSizeMB()).thenReturn(MIN_DISK_SIZE); when(entitySanitizer.sanitize(any())).thenReturn(Optional.of(jobDescriptor)); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<BatchJobExt>> sanitizedJobDescriptorOpt = sanitizer.sanitize(jobDescriptor); JobDescriptor<BatchJobExt> sanitizedJobDescriptor = sanitizedJobDescriptorOpt.get(); assertThat(sanitizedJobDescriptor).isNotNull(); assertThat(sanitizedJobDescriptor.getContainer().getContainerResources().getDiskMB()).isEqualTo(MIN_DISK_SIZE); String nonCompliant = sanitizedJobDescriptor.getAttributes().get(TITUS_NON_COMPLIANT_FEATURES); assertThat(nonCompliant).contains(FeatureRolloutPlans.MIN_DISK_SIZE_STRICT_VALIDATION_FEATURE); }
@Test public void testDiskSizeIsNotChanged() { JobDescriptor<BatchJobExt> jobDescriptor = newJobDescriptorWithDiskSize(11_000); when(configuration.getMinDiskSizeMB()).thenReturn(MIN_DISK_SIZE); when(entitySanitizer.sanitize(any())).thenReturn(Optional.of(jobDescriptor)); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<BatchJobExt>> sanitizedJobDescriptorOpt = sanitizer.sanitize(jobDescriptor); assertThat(sanitizedJobDescriptorOpt).isEmpty(); }
@Test public void testFlatStringEntryPoint() { JobDescriptor<?> jobDescriptor = newBatchJob() .map(jd -> jd.but(d -> d.getContainer().toBuilder() .withEntryPoint(Collections.singletonList("/bin/sh -c \"sleep 10\"")) .withCommand(null))) .getValue(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<?>> sanitized = sanitizer.sanitize(jobDescriptor); assertThat(sanitized).isPresent(); Map<String, String> attributes = sanitized.get().getAttributes(); assertThat(attributes).containsKey(TITUS_NON_COMPLIANT_FEATURES); List<String> problems = asList(attributes.get(TITUS_NON_COMPLIANT_FEATURES).split(",")); assertThat(problems).contains(FeatureRolloutPlans.ENTRY_POINT_STRICT_VALIDATION_FEATURE); }
@Test public void testValidEntryPoint() { JobDescriptor<?> jobDescriptor = newBatchJob() .map(jd -> jd.but(d -> d.getContainer().toBuilder() .withEntryPoint(asList("/bin/sh", "-c", "sleep 10")))) .getValue(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<?>> sanitized = sanitizer.sanitize(jobDescriptor); assertThat(sanitized).isNotPresent(); }
@Test public void testJobsWithCommandAreNotMarkedNonCompliant() { JobDescriptor<?> jobDescriptor = newBatchJob() .map(jd -> jd.but(d -> d.getContainer().toBuilder() .withEntryPoint(Collections.singletonList("a binary with spaces")) .withCommand(asList("some", "arguments")))) .getValue(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<?>> sanitized = sanitizer.sanitize(jobDescriptor); assertThat(sanitized).isNotPresent(); }
@Test public void testEnvironmentNamesWithInvalidCharactersAndNoValidationFailures() { JobDescriptor<BatchJobExt> jobDescriptor = newJobDescriptorWithEnvironment(";;;", "value"); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<BatchJobExt>> sanitized = sanitizer.sanitize(jobDescriptor); assertThat(sanitized).isNotEmpty(); assertThat(sanitized.get().getAttributes().get(TITUS_NON_COMPLIANT_FEATURES)).isEqualTo(ENVIRONMENT_VARIABLE_NAMES_STRICT_VALIDATION_FEATURE); }
@Test(expected = TitusServiceException.class) public void testEnvironmentNamesWithInvalidCharactersAndWithValidationFailures() { JobDescriptor<BatchJobExt> jobDescriptor = newJobDescriptorWithEnvironment(";;;", "value"); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> true, titusRuntime); sanitizer.sanitize(jobDescriptor); }
@Test public void testTitusAttributesAreResetIfProvidedByUser() { JobDescriptor<BatchJobExt> jobDescriptor = newBatchJob().getValue().toBuilder() .withAttributes(ImmutableMap.<String, String>builder() .put("myApp.a", "b") .put(TITUS_NON_COMPLIANT_FEATURES + "a", "b") .build() ) .build(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<BatchJobExt>> sanitized = sanitizer.sanitize(jobDescriptor); assertThat(sanitized).isNotEmpty(); assertThat(sanitized.get().getAttributes()).containsOnlyKeys("myApp.a"); }
@Test public void testLegacyServiceJobDisruptionBudgetRewrite() { JobDescriptor<ServiceJobExt> jobDescriptor = newServiceJob().getValue().toBuilder() .withDisruptionBudget(DisruptionBudget.none()) .build(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<ServiceJobExt>> sanitizedOpt = sanitizer.sanitize(jobDescriptor); assertThat(sanitizedOpt).isNotEmpty(); JobDescriptor<ServiceJobExt> sanitized = sanitizedOpt.get(); String nonCompliant = sanitized.getAttributes().get(TITUS_NON_COMPLIANT_FEATURES); assertThat(nonCompliant).contains(JobFeatureComplianceChecks.DISRUPTION_BUDGET_FEATURE); SelfManagedDisruptionBudgetPolicy policy = (SelfManagedDisruptionBudgetPolicy) sanitized.getDisruptionBudget().getDisruptionBudgetPolicy(); assertThat(policy.getRelocationTimeMs()).isEqualTo(DisruptionBudgetSanitizer.DEFAULT_SERVICE_RELOCATION_TIME_MS); }
@Test public void testLegacyBatchJobDisruptionBudgetRewrite() { JobDescriptor<BatchJobExt> jobDescriptor = newBatchJob().getValue().toBuilder() .withDisruptionBudget(DisruptionBudget.none()) .build(); ExtendedJobSanitizer sanitizer = new ExtendedJobSanitizer(configuration, jobAssertions, entitySanitizer, disruptionBudgetSanitizer, jd -> false, jd -> false, titusRuntime); Optional<JobDescriptor<BatchJobExt>> sanitizedOpt = sanitizer.sanitize(jobDescriptor); assertThat(sanitizedOpt).isNotEmpty(); JobDescriptor<BatchJobExt> sanitized = sanitizedOpt.get(); String nonCompliant = sanitized.getAttributes().get(TITUS_NON_COMPLIANT_FEATURES); assertThat(nonCompliant).contains(JobFeatureComplianceChecks.DISRUPTION_BUDGET_FEATURE); SelfManagedDisruptionBudgetPolicy policy = (SelfManagedDisruptionBudgetPolicy) sanitized.getDisruptionBudget().getDisruptionBudgetPolicy(); assertThat(policy.getRelocationTimeMs()).isEqualTo( (long) ((jobDescriptor.getExtensions()).getRuntimeLimitMs() * BATCH_RUNTIME_LIMIT_FACTOR) ); }