@Test public void testNoOpKubernetesManifestCustomizer() throws Exception { InstanceConfig config = createJavaInstanceConfig(FunctionDetails.Runtime.JAVA, false); config.setFunctionDetails(createFunctionDetails(FunctionDetails.Runtime.JAVA, false, (fb) -> { JsonObject configObj = new JsonObject(); configObj.addProperty("jobNamespace", "custom-ns"); return fb.setCustomRuntimeOptions(configObj.toString()); })); factory = createKubernetesRuntimeFactory(null, 10, 1.0, 1.0); verifyJavaInstance(config, pulsarRootDir + "/instances/deps", false); KubernetesRuntime container = factory.createContainer(config, userJarFile, userJarFile, 30l); V1Service serviceSpec = container.createService(); assertEquals(serviceSpec.getMetadata().getNamespace(), "default"); }
@VisibleForTesting V1Service createService() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1Service service = new V1Service(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); service.metadata(objectMeta); final V1ServiceSpec serviceSpec = new V1ServiceSpec(); serviceSpec.clusterIP("None"); final V1ServicePort servicePort = new V1ServicePort(); servicePort.name("grpc").port(grpcPort).protocol("TCP"); serviceSpec.addPortsItem(servicePort); serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails())); service.spec(serviceSpec); final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service); overridden.getMetadata().name(jobName); return overridden; }
KubernetesRuntime implements Runtime { @VisibleForTesting V1Service createService() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1Service service = new V1Service(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); service.metadata(objectMeta); final V1ServiceSpec serviceSpec = new V1ServiceSpec(); serviceSpec.clusterIP("None"); final V1ServicePort servicePort = new V1ServicePort(); servicePort.name("grpc").port(grpcPort).protocol("TCP"); serviceSpec.addPortsItem(servicePort); serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails())); service.spec(serviceSpec); final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service); overridden.getMetadata().name(jobName); return overridden; } }
KubernetesRuntime implements Runtime { @VisibleForTesting V1Service createService() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1Service service = new V1Service(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); service.metadata(objectMeta); final V1ServiceSpec serviceSpec = new V1ServiceSpec(); serviceSpec.clusterIP("None"); final V1ServicePort servicePort = new V1ServicePort(); servicePort.name("grpc").port(grpcPort).protocol("TCP"); serviceSpec.addPortsItem(servicePort); serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails())); service.spec(serviceSpec); final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service); overridden.getMetadata().name(jobName); return overridden; } 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 V1Service createService() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1Service service = new V1Service(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); service.metadata(objectMeta); final V1ServiceSpec serviceSpec = new V1ServiceSpec(); serviceSpec.clusterIP("None"); final V1ServicePort servicePort = new V1ServicePort(); servicePort.name("grpc").port(grpcPort).protocol("TCP"); serviceSpec.addPortsItem(servicePort); serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails())); service.spec(serviceSpec); final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service); overridden.getMetadata().name(jobName); return overridden; } 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 V1Service createService() { final String jobName = createJobName(instanceConfig.getFunctionDetails()); final V1Service service = new V1Service(); final V1ObjectMeta objectMeta = new V1ObjectMeta(); objectMeta.name(jobName); objectMeta.setLabels(getLabels(instanceConfig.getFunctionDetails())); objectMeta.setNamespace(jobNamespace); service.metadata(objectMeta); final V1ServiceSpec serviceSpec = new V1ServiceSpec(); serviceSpec.clusterIP("None"); final V1ServicePort servicePort = new V1ServicePort(); servicePort.name("grpc").port(grpcPort).protocol("TCP"); serviceSpec.addPortsItem(servicePort); serviceSpec.selector(getLabels(instanceConfig.getFunctionDetails())); service.spec(serviceSpec); final V1Service overridden = manifestCustomizer.map((customizer) -> customizer.customizeService(instanceConfig.getFunctionDetails(), service)).orElse(service); overridden.getMetadata().name(jobName); return overridden; } 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; }