@Test public void testCustomKubernetesManifestCustomizer() throws Exception { InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false); config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false, (fb) -> { return fb.setCustomRuntimeOptions("custom-service-account"); })); factory = createKubernetesRuntimeFactory(null, 10, 1.0, 1.0, Optional.of(new TestKubernetesCustomManifestCustomizer())); verifyJavaInstance(config, pulsarRootDir + "/instances/deps", false); KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l); V1StatefulSet spec = container.createStatefulSet(); assertEquals(spec.getSpec().getTemplate().getSpec().getServiceAccountName(), "my-service-account"); }
@VisibleForTesting V1StatefulSet createStatefulSet() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1StatefulSet statefulSet = new V1StatefulSet(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); statefulSet.metadata(objectMeta); final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec(); statefulSetSpec.serviceName(jobName); statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism()); statefulSetSpec.setPodManagementPolicy("Parallel"); final V1LabelSelector selector = new V1LabelSelector(); selector.matchLabels(getLabels(instanceConfig.getFunctionDetails())); statefulSetSpec.selector(selector); final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec(); final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails())); templateMetaData.annotations(getPrometheusAnnotations()); podTemplateSpec.setMetadata(templateMetaData); final List<String> command = getExecutorCommand(); podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null)); statefulSetSpec.setTemplate(podTemplateSpec); statefulSet.spec(statefulSetSpec); final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet); overridden.getMetadata().name(jobName); return statefulSet; }
KubernetesRuntime implements Runtime { @VisibleForTesting V1StatefulSet createStatefulSet() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1StatefulSet statefulSet = new V1StatefulSet(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); statefulSet.metadata(objectMeta); final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec(); statefulSetSpec.serviceName(jobName); statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism()); statefulSetSpec.setPodManagementPolicy("Parallel"); final V1LabelSelector selector = new V1LabelSelector(); selector.matchLabels(getLabels(instanceConfig.getFunctionDetails())); statefulSetSpec.selector(selector); final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec(); final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails())); templateMetaData.annotations(getPrometheusAnnotations()); podTemplateSpec.setMetadata(templateMetaData); final List<String> command = getExecutorCommand(); podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null)); statefulSetSpec.setTemplate(podTemplateSpec); statefulSet.spec(statefulSetSpec); final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet); overridden.getMetadata().name(jobName); return statefulSet; } }
KubernetesRuntime implements Runtime { @VisibleForTesting V1StatefulSet createStatefulSet() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1StatefulSet statefulSet = new V1StatefulSet(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); statefulSet.metadata(objectMeta); final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec(); statefulSetSpec.serviceName(jobName); statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism()); statefulSetSpec.setPodManagementPolicy("Parallel"); final V1LabelSelector selector = new V1LabelSelector(); selector.matchLabels(getLabels(instanceConfig.getFunctionDetails())); statefulSetSpec.selector(selector); final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec(); final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails())); templateMetaData.annotations(getPrometheusAnnotations()); podTemplateSpec.setMetadata(templateMetaData); final List<String> command = getExecutorCommand(); podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null)); statefulSetSpec.setTemplate(podTemplateSpec); statefulSet.spec(statefulSetSpec); final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet); overridden.getMetadata().name(jobName); return statefulSet; } KubernetesRuntime(AppsV1Api appsClient,
                      CoreV1Api coreClient,
                      String jobNamespace,
                      Map<String, String> customLabels,
                      Boolean installUserCodeDependencies,
                      String pythonDependencyRepository,
                      String pythonExtraDependencyRepository,
                      String pulsarDockerImageName,
                      Map<String, String> functionDockerImages,
                      String imagePullPolicy,
                      String pulsarRootDir,
                      InstanceConfig instanceConfig,
                      String instanceFile,
                      String extraDependenciesDir,
                      String logDirectory,
                      String configAdminCLI,
                      String userCodePkgUrl,
                      String originalCodeFileName,
                      String pulsarServiceUrl,
                      String pulsarAdminUrl,
                      String stateStorageServiceUrl,
                      AuthenticationConfig authConfig,
                      SecretsProviderConfigurator secretsProviderConfigurator,
                      Integer expectedMetricsCollectionInterval,
                      int percentMemoryPadding,
                      double cpuOverCommitRatio,
                      double memoryOverCommitRatio,
                      Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider,
                      boolean authenticationEnabled,
                      Integer grpcPort,
                      Integer metricsPort,
                      String narExtractionDirectory,
                      Optional<KubernetesManifestCustomizer> manifestCustomizer); }
KubernetesRuntime implements Runtime { @VisibleForTesting V1StatefulSet createStatefulSet() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1StatefulSet statefulSet = new V1StatefulSet(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); statefulSet.metadata(objectMeta); final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec(); statefulSetSpec.serviceName(jobName); statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism()); statefulSetSpec.setPodManagementPolicy("Parallel"); final V1LabelSelector selector = new V1LabelSelector(); selector.matchLabels(getLabels(instanceConfig.getFunctionDetails())); statefulSetSpec.selector(selector); final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec(); final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails())); templateMetaData.annotations(getPrometheusAnnotations()); podTemplateSpec.setMetadata(templateMetaData); final List<String> command = getExecutorCommand(); podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null)); statefulSetSpec.setTemplate(podTemplateSpec); statefulSet.spec(statefulSetSpec); final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet); overridden.getMetadata().name(jobName); return statefulSet; } KubernetesRuntime(AppsV1Api appsClient,
                      CoreV1Api coreClient,
                      String jobNamespace,
                      Map<String, String> customLabels,
                      Boolean installUserCodeDependencies,
                      String pythonDependencyRepository,
                      String pythonExtraDependencyRepository,
                      String pulsarDockerImageName,
                      Map<String, String> functionDockerImages,
                      String imagePullPolicy,
                      String pulsarRootDir,
                      InstanceConfig instanceConfig,
                      String instanceFile,
                      String extraDependenciesDir,
                      String logDirectory,
                      String configAdminCLI,
                      String userCodePkgUrl,
                      String originalCodeFileName,
                      String pulsarServiceUrl,
                      String pulsarAdminUrl,
                      String stateStorageServiceUrl,
                      AuthenticationConfig authConfig,
                      SecretsProviderConfigurator secretsProviderConfigurator,
                      Integer expectedMetricsCollectionInterval,
                      int percentMemoryPadding,
                      double cpuOverCommitRatio,
                      double memoryOverCommitRatio,
                      Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider,
                      boolean authenticationEnabled,
                      Integer grpcPort,
                      Integer metricsPort,
                      String narExtractionDirectory,
                      Optional<KubernetesManifestCustomizer> manifestCustomizer); @Override void start(); @Override void reinitialize(); @Override void join(); @Override void stop(); @Override Throwable getDeathException(); @Override CompletableFuture<FunctionStatus> getFunctionStatus(int instanceId); @Override CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics(); @Override CompletableFuture<Void> resetMetrics(); @Override CompletableFuture<InstanceCommunication.MetricsData> getMetrics(int instanceId); @Override String getPrometheusMetrics(); @Override boolean isAlive(); void deleteStatefulSet(); void deleteService(); static String createJobName(Function.FunctionDetails functionDetails); static void doChecks(Function.FunctionDetails functionDetails); }
KubernetesRuntime implements Runtime { @VisibleForTesting V1StatefulSet createStatefulSet() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1StatefulSet statefulSet = new V1StatefulSet(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); statefulSet.metadata(objectMeta); final V1StatefulSetSpec statefulSetSpec = new V1StatefulSetSpec(); statefulSetSpec.serviceName(jobName); statefulSetSpec.setReplicas(instanceConfig.getFunctionDetails().getParallelism()); statefulSetSpec.setPodManagementPolicy("Parallel"); final V1LabelSelector selector = new V1LabelSelector(); selector.matchLabels(getLabels(instanceConfig.getFunctionDetails())); statefulSetSpec.selector(selector); final V1PodTemplateSpec podTemplateSpec = new V1PodTemplateSpec(); final V1ObjectMeta templateMetaData = new V1ObjectMeta().labels(getLabels(instanceConfig.getFunctionDetails())); templateMetaData.annotations(getPrometheusAnnotations()); podTemplateSpec.setMetadata(templateMetaData); final List<String> command = getExecutorCommand(); podTemplateSpec.spec(getPodSpec(command, instanceConfig.getFunctionDetails().hasResources() ? instanceConfig.getFunctionDetails().getResources() : null)); statefulSetSpec.setTemplate(podTemplateSpec); statefulSet.spec(statefulSetSpec); final V1StatefulSet overridden = manifestCustomizer.map((customizer) -> customizer.customizeStatefulSet(instanceConfig.getFunctionDetails(), statefulSet)).orElse(statefulSet); overridden.getMetadata().name(jobName); return statefulSet; } KubernetesRuntime(AppsV1Api appsClient,
                      CoreV1Api coreClient,
                      String jobNamespace,
                      Map<String, String> customLabels,
                      Boolean installUserCodeDependencies,
                      String pythonDependencyRepository,
                      String pythonExtraDependencyRepository,
                      String pulsarDockerImageName,
                      Map<String, String> functionDockerImages,
                      String imagePullPolicy,
                      String pulsarRootDir,
                      InstanceConfig instanceConfig,
                      String instanceFile,
                      String extraDependenciesDir,
                      String logDirectory,
                      String configAdminCLI,
                      String userCodePkgUrl,
                      String originalCodeFileName,
                      String pulsarServiceUrl,
                      String pulsarAdminUrl,
                      String stateStorageServiceUrl,
                      AuthenticationConfig authConfig,
                      SecretsProviderConfigurator secretsProviderConfigurator,
                      Integer expectedMetricsCollectionInterval,
                      int percentMemoryPadding,
                      double cpuOverCommitRatio,
                      double memoryOverCommitRatio,
                      Optional<KubernetesFunctionAuthProvider> functionAuthDataCacheProvider,
                      boolean authenticationEnabled,
                      Integer grpcPort,
                      Integer metricsPort,
                      String narExtractionDirectory,
                      Optional<KubernetesManifestCustomizer> manifestCustomizer); @Override void start(); @Override void reinitialize(); @Override void join(); @Override void stop(); @Override Throwable getDeathException(); @Override CompletableFuture<FunctionStatus> getFunctionStatus(int instanceId); @Override CompletableFuture<InstanceCommunication.MetricsData> getAndResetMetrics(); @Override CompletableFuture<Void> resetMetrics(); @Override CompletableFuture<InstanceCommunication.MetricsData> getMetrics(int instanceId); @Override String getPrometheusMetrics(); @Override boolean isAlive(); void deleteStatefulSet(); void deleteService(); static String createJobName(Function.FunctionDetails functionDetails); static void doChecks(Function.FunctionDetails functionDetails); static final Pattern VALID_POD_NAME_REGEX; }