This is an automated email from the ASF dual-hosted git repository. gyfora pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
commit a7e6f0c8dd08cecb26822731e058c003ac0fbee5 Author: Zhenqiu Huang <[email protected]> AuthorDate: Fri Oct 21 15:12:33 2022 -0700 [FLINK-29655] Split Flink CRD from flink-kubernates-operator module --- flink-kubernetes-api/pom.xml | 132 ++++++++++++++++ .../config/FlinkOperatorConfiguration.java | 11 +- .../config/KubernetesOperatorConfigOptions.java | 0 .../config}/KubernetesOperatorMetricOptions.java | 2 +- .../flink/kubernetes/operator/config/Mode.java | 0 .../operator/crd/AbstractFlinkResource.java | 0 .../kubernetes/operator/crd/CrdConstants.java | 0 .../kubernetes/operator/crd/FlinkDeployment.java | 0 .../operator/crd/FlinkDeploymentList.java | 0 .../kubernetes/operator/crd/FlinkSessionJob.java | 0 .../operator/crd/FlinkSessionJobList.java | 0 .../operator/crd/spec/AbstractFlinkSpec.java | 0 .../operator/crd/spec/FlinkDeploymentSpec.java | 0 .../operator/crd/spec/FlinkSessionJobSpec.java | 0 .../kubernetes/operator/crd/spec/FlinkVersion.java | 0 .../kubernetes/operator/crd/spec/IngressSpec.java | 0 .../operator/crd/spec/JobManagerSpec.java | 0 .../kubernetes/operator/crd/spec/JobSpec.java | 0 .../kubernetes/operator/crd/spec/JobState.java | 0 .../crd/spec/KubernetesDeploymentMode.java | 0 .../kubernetes/operator/crd/spec/Resource.java | 0 .../operator/crd/spec/TaskManagerSpec.java | 0 .../kubernetes/operator/crd/spec/UpgradeMode.java | 0 .../operator/crd/status/CommonStatus.java | 1 - .../FlinkDeploymentReconciliationStatus.java | 0 .../operator/crd/status/FlinkDeploymentStatus.java | 0 .../FlinkSessionJobReconciliationStatus.java | 0 .../operator/crd/status/FlinkSessionJobStatus.java | 0 .../crd/status/JobManagerDeploymentStatus.java | 10 +- .../kubernetes/operator/crd/status/JobStatus.java | 0 .../crd/status}/ReconciliationMetadata.java | 3 +- .../operator/crd/status/ReconciliationState.java | 0 .../operator/crd/status/ReconciliationStatus.java | 9 +- .../crd/status}/ResourceLifecycleState.java | 2 +- .../kubernetes/operator/crd/status/Savepoint.java | 0 .../operator/crd/status/SavepointFormatType.java | 0 .../operator/crd/status/SavepointInfo.java | 0 .../operator/crd/status/SavepointTriggerType.java | 0 .../operator/crd/status/TaskManagerInfo.java | 0 .../kubernetes/operator/reconciler/diff/Diff.java | 0 .../operator/reconciler/diff/DiffBuilder.java | 0 .../operator/reconciler/diff/DiffResult.java | 0 .../operator/reconciler/diff/DiffType.java | 0 .../operator/reconciler/diff/Diffable.java | 0 .../reconciler/diff/ReflectiveDiffBuilder.java | 0 .../operator/reconciler/diff/SpecDiff.java | 0 .../flink/kubernetes/operator/utils/SpecUtils.java | 119 ++++++++++++++ .../flink/kubernetes/operator/BaseTestUtils.java | 176 +++++++++++++++++++++ .../operator/reconciler/diff/SpecDiffTest.java | 24 +-- .../kubernetes/operator/utils/SpecUtilsTest.java | 60 +++++++ flink-kubernetes-operator/pom.xml | 14 ++ .../metrics/KubernetesOperatorScopeFormat.java | 2 +- .../KubernetesResourceNamespaceScopeFormat.java | 2 +- .../metrics/KubernetesResourceScopeFormat.java | 2 +- .../kubernetes/operator/metrics/MetricManager.java | 1 + .../operator/metrics/OperatorMetricUtils.java | 1 + .../metrics/lifecycle/LifecycleMetrics.java | 17 +- .../lifecycle/ResourceLifecycleMetricTracker.java | 1 + .../operator/reconciler/ReconciliationUtils.java | 92 +---------- .../kubernetes/operator/utils/StatusRecorder.java | 2 +- .../flink/kubernetes/operator/TestUtils.java | 152 +----------------- .../operator/config/FlinkConfigBuilderTest.java | 8 +- .../controller/FlinkDeploymentControllerTest.java | 2 +- .../metrics/FlinkDeploymentMetricsTest.java | 2 +- .../metrics/FlinkSessionJobMetricsTest.java | 2 +- .../metrics/KubernetesClientMetricsTest.java | 1 + .../metrics/KubernetesOperatorMetricGroupTest.java | 2 +- .../lifecycle/ResourceLifecycleMetricsTest.java | 19 +-- .../operator/utils/ReconciliationUtilsTest.java | 38 +---- pom.xml | 5 +- 70 files changed, 582 insertions(+), 332 deletions(-) diff --git a/flink-kubernetes-api/pom.xml b/flink-kubernetes-api/pom.xml new file mode 100644 index 00000000..5a86fb11 --- /dev/null +++ b/flink-kubernetes-api/pom.xml @@ -0,0 +1,132 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.flink</groupId> + <artifactId>flink-kubernetes-operator-parent</artifactId> + <version>1.3-SNAPSHOT</version> + <relativePath>..</relativePath> + </parent> + + <artifactId>flink-kubernetes-api</artifactId> + <name>Flink Kubernetes Operator Api</name> + <packaging>jar</packaging> + + <properties> + <awaitility.version>4.1.0</awaitility.version> + <plugins.tmp.dir>${project.build.directory}/plugins</plugins.tmp.dir> + </properties> + + <dependencies> + <dependency> + <groupId>io.javaoperatorsdk</groupId> + <artifactId>operator-framework</artifactId> + <version>${operator.sdk.version}</version> + </dependency> + + <dependency> + <groupId>io.fabric8</groupId> + <artifactId>kubernetes-client</artifactId> + <version>${fabric8.version}</version> + <!-- FLINK-29384, can be removed when fabric8 pulls snakeyaml 1.32+ --> + <exclusions> + <exclusion> + <groupId>org.yaml</groupId> + <artifactId>snakeyaml</artifactId> + </exclusion> + </exclusions> + </dependency> + + <!-- FLINK-29384, can be removed when fabric8 pulls snakeyaml 1.32+ --> + <dependency> + <groupId>org.yaml</groupId> + <artifactId>snakeyaml</artifactId> + <version>${snakeyaml.version}</version> + </dependency> + + <!-- Flink --> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-core</artifactId> + <version>${flink.version}</version> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-kubernetes</artifactId> + <version>${flink.version}</version> + </dependency> + + <dependency> + <groupId>org.jetbrains</groupId> + <artifactId>annotations</artifactId> + <version>${jetbrains.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>org.projectlombok</groupId> + <artifactId>lombok</artifactId> + <version>${lombok.version}</version> + <scope>provided</scope> + </dependency> + + <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-lang3</artifactId> + <version>${commons-lang3.version}</version> + </dependency> + + <!-- Test --> + + <dependency> + <groupId>org.junit.jupiter</groupId> + <artifactId>junit-jupiter-params</artifactId> + <version>${junit.jupiter.version}</version> + <scope>test</scope> + </dependency> + + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-runtime</artifactId> + <version>${flink.version}</version> + <scope>test</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.maven.plugins</groupId> + <artifactId>maven-jar-plugin</artifactId> + <version>${maven-jar-plugin.version}</version> + <executions> + <execution> + <goals> + <goal>test-jar</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> +</project> diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java similarity index 95% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java index 3a54847a..db1cfef9 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java @@ -19,15 +19,15 @@ package org.apache.flink.kubernetes.operator.config; import org.apache.flink.configuration.Configuration; -import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions; -import org.apache.flink.kubernetes.operator.utils.EnvUtils; import io.javaoperatorsdk.operator.api.config.RetryConfiguration; import lombok.Value; +import org.apache.commons.lang3.StringUtils; import java.time.Duration; import java.util.Arrays; import java.util.HashSet; +import java.util.Optional; import java.util.Set; /** Configuration class for operator. */ @@ -35,6 +35,7 @@ import java.util.Set; public class FlinkOperatorConfiguration { private static final String NAMESPACES_SPLITTER_KEY = "\\s*,\\s*"; + private static final String ENV_KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST"; Duration reconcileInterval; int reconcilerMaxParallelism; @@ -113,7 +114,7 @@ public class FlinkOperatorConfiguration { .OPERATOR_EXCEPTION_THROWABLE_LIST_MAX_COUNT); String flinkServiceHostOverride = null; - if (EnvUtils.get(EnvUtils.ENV_KUBERNETES_SERVICE_HOST).isEmpty()) { + if (getEnv(ENV_KUBERNETES_SERVICE_HOST).isEmpty()) { // not running in k8s, simplify local development flinkServiceHostOverride = "localhost"; } @@ -210,4 +211,8 @@ public class FlinkOperatorConfiguration { return (long) (initialInterval * Math.pow(intervalMultiplier, maxAttempts)); } } + + private static Optional<String> getEnv(String key) { + return Optional.ofNullable(StringUtils.getIfBlank(System.getenv().get(key), () -> null)); + } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorMetricOptions.java similarity index 98% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorMetricOptions.java index dfc876d4..0222eaeb 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorMetricOptions.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.flink.kubernetes.operator.metrics; +package org.apache.flink.kubernetes.operator.config; import org.apache.flink.configuration.ConfigOption; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/Mode.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/Mode.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/Mode.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/config/Mode.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/AbstractFlinkResource.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/AbstractFlinkResource.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/AbstractFlinkResource.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/AbstractFlinkResource.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/CrdConstants.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/CrdConstants.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/CrdConstants.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/CrdConstants.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeployment.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeployment.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeployment.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeployment.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeploymentList.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeploymentList.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeploymentList.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkDeploymentList.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJob.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJob.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJob.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJob.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJobList.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJobList.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJobList.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/FlinkSessionJobList.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/AbstractFlinkSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkDeploymentSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkDeploymentSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkDeploymentSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkDeploymentSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkSessionJobSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/FlinkVersion.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/IngressSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/IngressSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/IngressSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/IngressSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobManagerSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobManagerSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobManagerSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobManagerSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobState.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobState.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobState.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/JobState.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/KubernetesDeploymentMode.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/KubernetesDeploymentMode.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/KubernetesDeploymentMode.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/KubernetesDeploymentMode.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/Resource.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/Resource.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/Resource.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/Resource.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/TaskManagerSpec.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/TaskManagerSpec.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/TaskManagerSpec.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/TaskManagerSpec.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/UpgradeMode.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/UpgradeMode.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/UpgradeMode.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/spec/UpgradeMode.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java similarity index 97% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java index b91efcf3..bdc1cf34 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/CommonStatus.java @@ -20,7 +20,6 @@ package org.apache.flink.kubernetes.operator.crd.status; import org.apache.flink.annotation.Experimental; import org.apache.flink.kubernetes.operator.crd.spec.AbstractFlinkSpec; import org.apache.flink.kubernetes.operator.crd.spec.JobState; -import org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState; import com.fasterxml.jackson.annotation.JsonIgnore; import lombok.AllArgsConstructor; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentReconciliationStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentReconciliationStatus.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentReconciliationStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentReconciliationStatus.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkDeploymentStatus.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobReconciliationStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobReconciliationStatus.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobReconciliationStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobReconciliationStatus.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobStatus.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/FlinkSessionJobStatus.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java similarity index 90% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java index b5ee5a88..79035230 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobManagerDeploymentStatus.java @@ -19,7 +19,8 @@ package org.apache.flink.kubernetes.operator.crd.status; import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration; import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; -import org.apache.flink.kubernetes.operator.utils.SavepointUtils; + +import org.apache.commons.lang3.StringUtils; import java.time.Duration; @@ -51,8 +52,7 @@ public enum JobManagerDeploymentStatus { break; case READY: rescheduleAfter = - SavepointUtils.savepointInProgress( - flinkDeployment.getStatus().getJobStatus()) + savepointInProgress(flinkDeployment.getStatus().getJobStatus()) ? operatorConfiguration.getProgressCheckInterval() : operatorConfiguration.getReconcileInterval(); break; @@ -68,4 +68,8 @@ public enum JobManagerDeploymentStatus { } return rescheduleAfter; } + + private static boolean savepointInProgress(JobStatus jobStatus) { + return StringUtils.isNotEmpty(jobStatus.getSavepointInfo().getTriggerId()); + } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobStatus.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/JobStatus.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationMetadata.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationMetadata.java similarity index 94% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationMetadata.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationMetadata.java index dce62ac8..440d8b8a 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationMetadata.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationMetadata.java @@ -15,10 +15,9 @@ * limitations under the License. */ -package org.apache.flink.kubernetes.operator.reconciler; +package org.apache.flink.kubernetes.operator.crd.status; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; -import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState; import com.fasterxml.jackson.annotation.JsonInclude; import io.fabric8.kubernetes.api.model.ObjectMeta; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationState.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java similarity index 88% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java index 0e18d998..dd0bd9bd 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ReconciliationStatus.java @@ -21,8 +21,7 @@ import org.apache.flink.annotation.Experimental; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; import org.apache.flink.kubernetes.operator.crd.spec.AbstractFlinkSpec; -import org.apache.flink.kubernetes.operator.reconciler.ReconciliationMetadata; -import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; +import org.apache.flink.kubernetes.operator.utils.SpecUtils; import com.fasterxml.jackson.annotation.JsonIgnore; import io.fabric8.kubernetes.model.annotation.PrinterColumn; @@ -71,18 +70,18 @@ public abstract class ReconciliationStatus<SPEC extends AbstractFlinkSpec> { @JsonIgnore public Tuple2<SPEC, ReconciliationMetadata> deserializeLastReconciledSpecWithMeta() { - return ReconciliationUtils.deserializeSpecWithMeta(lastReconciledSpec, getSpecClass()); + return SpecUtils.deserializeSpecWithMeta(lastReconciledSpec, getSpecClass()); } @JsonIgnore public Tuple2<SPEC, ReconciliationMetadata> deserializeLastStableSpecWithMeta() { - return ReconciliationUtils.deserializeSpecWithMeta(lastStableSpec, getSpecClass()); + return SpecUtils.deserializeSpecWithMeta(lastStableSpec, getSpecClass()); } @JsonIgnore public void serializeAndSetLastReconciledSpec( SPEC spec, AbstractFlinkResource<SPEC, ?> resource) { - setLastReconciledSpec(ReconciliationUtils.writeSpecWithMeta(spec, resource)); + setLastReconciledSpec(SpecUtils.writeSpecWithMeta(spec, resource)); } public void markReconciledSpecAsStable() { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleState.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ResourceLifecycleState.java similarity index 97% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleState.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ResourceLifecycleState.java index 324f5346..a5faafc2 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleState.java +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/ResourceLifecycleState.java @@ -15,7 +15,7 @@ * limitations under the License. */ -package org.apache.flink.kubernetes.operator.metrics.lifecycle; +package org.apache.flink.kubernetes.operator.crd.status; import lombok.Getter; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/Savepoint.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointFormatType.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointInfo.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointTriggerType.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointTriggerType.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointTriggerType.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/SavepointTriggerType.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/TaskManagerInfo.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/TaskManagerInfo.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/status/TaskManagerInfo.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/crd/status/TaskManagerInfo.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diff.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diff.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diff.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diff.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffBuilder.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffBuilder.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffBuilder.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffBuilder.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffResult.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffResult.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffResult.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffResult.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffType.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffType.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffType.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/DiffType.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diffable.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diffable.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diffable.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/Diffable.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/ReflectiveDiffBuilder.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/ReflectiveDiffBuilder.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/ReflectiveDiffBuilder.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/ReflectiveDiffBuilder.java diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiff.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiff.java similarity index 100% rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiff.java rename to flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiff.java diff --git a/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/utils/SpecUtils.java b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/utils/SpecUtils.java new file mode 100644 index 00000000..9d21d617 --- /dev/null +++ b/flink-kubernetes-api/src/main/java/org/apache/flink/kubernetes/operator/utils/SpecUtils.java @@ -0,0 +1,119 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.kubernetes.operator.utils; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; +import org.apache.flink.kubernetes.operator.crd.spec.AbstractFlinkSpec; +import org.apache.flink.kubernetes.operator.crd.status.ReconciliationMetadata; +import org.apache.flink.util.Preconditions; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; + +import javax.annotation.Nullable; + +/** Spec utilities. */ +public class SpecUtils { + public static final String INTERNAL_METADATA_JSON_KEY = "resource_metadata"; + private static final ObjectMapper objectMapper = new ObjectMapper(); + + /** + * Deserializes the spec and custom metadata object from JSON. + * + * @param specWithMetaString JSON string. + * @param specClass Spec class for deserialization. + * @param <T> Spec type. + * @return Tuple2 of spec and meta. + */ + public static <T extends AbstractFlinkSpec> + Tuple2<T, ReconciliationMetadata> deserializeSpecWithMeta( + @Nullable String specWithMetaString, Class<T> specClass) { + if (specWithMetaString == null) { + return null; + } + + try { + ObjectNode wrapper = (ObjectNode) objectMapper.readTree(specWithMetaString); + ObjectNode internalMeta = (ObjectNode) wrapper.remove(INTERNAL_METADATA_JSON_KEY); + + if (internalMeta == null) { + // migrating from old format + wrapper.remove("apiVersion"); + return Tuple2.of(objectMapper.treeToValue(wrapper, specClass), null); + } else { + return Tuple2.of( + objectMapper.treeToValue(wrapper.get("spec"), specClass), + objectMapper.convertValue(internalMeta, ReconciliationMetadata.class)); + } + } catch (JsonProcessingException e) { + throw new RuntimeException("Could not deserialize spec, this indicates a bug...", e); + } + } + + /** + * Serializes the spec and custom meta information into a JSON string. + * + * @param spec Flink resource spec. + * @param relatedResource Related Flink resource for creating the meta object. + * @return Serialized json. + */ + public static String writeSpecWithMeta( + AbstractFlinkSpec spec, AbstractFlinkResource<?, ?> relatedResource) { + return writeSpecWithMeta(spec, ReconciliationMetadata.from(relatedResource)); + } + + /** + * Serializes the spec and custom meta information into a JSON string. + * + * @param spec Flink resource spec. + * @param metadata Reconciliation meta object. + * @return Serialized json. + */ + public static String writeSpecWithMeta( + AbstractFlinkSpec spec, ReconciliationMetadata metadata) { + + ObjectNode wrapper = objectMapper.createObjectNode(); + + wrapper.set("spec", objectMapper.valueToTree(Preconditions.checkNotNull(spec))); + wrapper.set( + INTERNAL_METADATA_JSON_KEY, + objectMapper.valueToTree(Preconditions.checkNotNull(metadata))); + + try { + return objectMapper.writeValueAsString(wrapper); + } catch (JsonProcessingException e) { + throw new RuntimeException("Could not serialize spec, this indicates a bug...", e); + } + } + + public static <T> T clone(T object) { + if (object == null) { + return null; + } + try { + return (T) + objectMapper.readValue( + objectMapper.writeValueAsString(object), object.getClass()); + } catch (JsonProcessingException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/BaseTestUtils.java b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/BaseTestUtils.java new file mode 100644 index 00000000..9117ffec --- /dev/null +++ b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/BaseTestUtils.java @@ -0,0 +1,176 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.kubernetes.operator; + +import org.apache.flink.configuration.CheckpointingOptions; +import org.apache.flink.configuration.HighAvailabilityOptions; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; +import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; +import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob; +import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec; +import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec; +import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion; +import org.apache.flink.kubernetes.operator.crd.spec.JobManagerSpec; +import org.apache.flink.kubernetes.operator.crd.spec.JobSpec; +import org.apache.flink.kubernetes.operator.crd.spec.JobState; +import org.apache.flink.kubernetes.operator.crd.spec.Resource; +import org.apache.flink.kubernetes.operator.crd.spec.TaskManagerSpec; +import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode; +import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus; +import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus; + +import io.fabric8.kubernetes.api.model.Container; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.Pod; +import io.fabric8.kubernetes.api.model.PodSpec; + +import java.time.Instant; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** Base Testing utilities. */ +public class BaseTestUtils { + + public static final String TEST_NAMESPACE = "flink-operator-test"; + public static final String TEST_DEPLOYMENT_NAME = "test-cluster"; + public static final String TEST_SESSION_JOB_NAME = "test-session-job"; + public static final String SERVICE_ACCOUNT = "flink-operator"; + public static final String FLINK_VERSION = "latest"; + public static final String IMAGE = String.format("flink:%s", FLINK_VERSION); + public static final String IMAGE_POLICY = "IfNotPresent"; + public static final String SAMPLE_JAR = "local:///tmp/sample.jar"; + + public static FlinkDeployment buildSessionCluster() { + return buildSessionCluster(FlinkVersion.v1_15); + } + + public static FlinkDeployment buildSessionCluster(FlinkVersion version) { + return buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, version); + } + + public static FlinkDeployment buildSessionCluster( + String name, String namespace, FlinkVersion version) { + FlinkDeployment deployment = new FlinkDeployment(); + deployment.setStatus(new FlinkDeploymentStatus()); + deployment.setMetadata( + new ObjectMetaBuilder() + .withName(name) + .withNamespace(namespace) + .withCreationTimestamp(Instant.now().toString()) + .withUid(UUID.randomUUID().toString()) + .build()); + deployment.setSpec(getTestFlinkDeploymentSpec(version)); + return deployment; + } + + public static FlinkDeployment buildApplicationCluster() { + return buildApplicationCluster(FlinkVersion.v1_15); + } + + public static FlinkDeployment buildApplicationCluster(String name, String namespace) { + return buildApplicationCluster(name, namespace, FlinkVersion.v1_15); + } + + public static FlinkDeployment buildApplicationCluster(FlinkVersion version) { + return buildApplicationCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, version); + } + + public static FlinkDeployment buildApplicationCluster( + String name, String namespace, FlinkVersion version) { + FlinkDeployment deployment = buildSessionCluster(name, namespace, version); + deployment + .getSpec() + .setJob( + JobSpec.builder() + .jarURI(SAMPLE_JAR) + .parallelism(1) + .upgradeMode(UpgradeMode.STATELESS) + .state(JobState.RUNNING) + .build()); + deployment.setStatus(deployment.initStatus()); + return deployment; + } + + public static FlinkSessionJob buildSessionJob(String name, String namespace) { + FlinkSessionJob sessionJob = new FlinkSessionJob(); + sessionJob.setStatus(new FlinkSessionJobStatus()); + sessionJob.setMetadata( + new ObjectMetaBuilder() + .withName(name) + .withNamespace(namespace) + .withCreationTimestamp(Instant.now().toString()) + .withUid(UUID.randomUUID().toString()) + .withGeneration(1L) + .build()); + + Map<String, String> conf = new HashMap<>(); + conf.put(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER.key(), "header"); + sessionJob.setSpec( + FlinkSessionJobSpec.builder() + .deploymentName(TEST_DEPLOYMENT_NAME) + .job( + JobSpec.builder() + .jarURI(SAMPLE_JAR) + .parallelism(1) + .upgradeMode(UpgradeMode.STATELESS) + .state(JobState.RUNNING) + .build()) + .flinkConfiguration(conf) + .build()); + return sessionJob; + } + + public static FlinkSessionJob buildSessionJob() { + return buildSessionJob(TEST_SESSION_JOB_NAME, TEST_NAMESPACE); + } + + public static FlinkDeploymentSpec getTestFlinkDeploymentSpec(FlinkVersion version) { + Map<String, String> conf = new HashMap<>(); + conf.put(TaskManagerOptions.NUM_TASK_SLOTS.key(), "2"); + conf.put( + HighAvailabilityOptions.HA_MODE.key(), + KubernetesHaServicesFactory.class.getCanonicalName()); + conf.put(HighAvailabilityOptions.HA_STORAGE_PATH.key(), "test"); + conf.put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(), "test-savepoint-dir"); + conf.put(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key(), "test-checkpoint-dir"); + + return FlinkDeploymentSpec.builder() + .image(IMAGE) + .imagePullPolicy(IMAGE_POLICY) + .serviceAccount(SERVICE_ACCOUNT) + .flinkVersion(version) + .flinkConfiguration(conf) + .jobManager(new JobManagerSpec(new Resource(1.0, "2048m"), 1, null)) + .taskManager(new TaskManagerSpec(new Resource(1.0, "2048m"), null, null)) + .build(); + } + + public static Pod getTestPod(String hostname, String apiVersion, List<Container> containers) { + final PodSpec podSpec = new PodSpec(); + podSpec.setHostname(hostname); + podSpec.setContainers(containers); + final Pod pod = new Pod(); + pod.setApiVersion(apiVersion); + pod.setSpec(podSpec); + return pod; + } +} diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java similarity index 89% rename from flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java rename to flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java index 6f39669a..4c52e238 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java +++ b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/reconciler/diff/SpecDiffTest.java @@ -18,7 +18,7 @@ package org.apache.flink.kubernetes.operator.reconciler.diff; import org.apache.flink.configuration.CoreOptions; -import org.apache.flink.kubernetes.operator.TestUtils; +import org.apache.flink.kubernetes.operator.BaseTestUtils; import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec; import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec; @@ -27,7 +27,7 @@ import org.apache.flink.kubernetes.operator.crd.spec.IngressSpec; import org.apache.flink.kubernetes.operator.crd.spec.JobState; import org.apache.flink.kubernetes.operator.crd.spec.KubernetesDeploymentMode; import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode; -import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; +import org.apache.flink.kubernetes.operator.utils.SpecUtils; import org.junit.jupiter.api.Test; @@ -35,7 +35,7 @@ import java.util.List; import java.util.Map; import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_RECONCILE_INTERVAL; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; import static org.junit.jupiter.api.Assertions.assertEquals; /** Spec diff test. */ @@ -44,22 +44,22 @@ public class SpecDiffTest { @Test public void testFlinkDeploymentSpecChanges() { var left = new FlinkDeploymentSpec(); - var right = ReconciliationUtils.clone(left); + var right = SpecUtils.clone(left); var diff = left.diff(right); assertEquals(DiffType.IGNORE, diff.getType()); assertEquals(0, diff.getNumDiffs()); - left = TestUtils.buildSessionCluster().getSpec(); - right = ReconciliationUtils.clone(left); + left = BaseTestUtils.buildSessionCluster().getSpec(); + right = SpecUtils.clone(left); diff = left.diff(right); assertEquals(DiffType.IGNORE, diff.getType()); assertEquals(0, diff.getNumDiffs()); - left = TestUtils.buildApplicationCluster().getSpec(); - left.setPodTemplate(TestUtils.getTestPod("localhost", "v1", List.of())); + left = BaseTestUtils.buildApplicationCluster().getSpec(); + left.setPodTemplate(BaseTestUtils.getTestPod("localhost", "v1", List.of())); left.setIngress(IngressSpec.builder().template("template").build()); - right = ReconciliationUtils.clone(left); + right = SpecUtils.clone(left); diff = left.diff(right); assertEquals(DiffType.IGNORE, diff.getType()); assertEquals(0, diff.getNumDiffs()); @@ -128,13 +128,13 @@ public class SpecDiffTest { @Test public void testFlinkSessionJobSpecChanges() { var left = new FlinkSessionJobSpec(); - var right = ReconciliationUtils.clone(left); + var right = SpecUtils.clone(left); var diff = left.diff(right); assertEquals(DiffType.IGNORE, diff.getType()); assertEquals(0, diff.getNumDiffs()); - left = TestUtils.buildSessionJob().getSpec(); - right = ReconciliationUtils.clone(left); + left = BaseTestUtils.buildSessionJob().getSpec(); + right = SpecUtils.clone(left); diff = left.diff(right); assertEquals(DiffType.IGNORE, diff.getType()); assertEquals(0, diff.getNumDiffs()); diff --git a/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/utils/SpecUtilsTest.java b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/utils/SpecUtilsTest.java new file mode 100644 index 00000000..9074eca6 --- /dev/null +++ b/flink-kubernetes-api/src/test/java/org/apache/flink/kubernetes/operator/utils/SpecUtilsTest.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.kubernetes.operator.utils; + +import org.apache.flink.kubernetes.operator.BaseTestUtils; +import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; +import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +/** Test for {@link org.apache.flink.kubernetes.operator.utils.SpecUtils}. */ +public class SpecUtilsTest { + + @Test + public void testSpecSerializationWithVersion() throws JsonProcessingException { + FlinkDeployment app = BaseTestUtils.buildApplicationCluster(); + app.getMetadata().setGeneration(12L); + String serialized = SpecUtils.writeSpecWithMeta(app.getSpec(), app); + ObjectNode node = (ObjectNode) new ObjectMapper().readTree(serialized); + + ObjectNode internalMeta = (ObjectNode) node.get(SpecUtils.INTERNAL_METADATA_JSON_KEY); + assertEquals("flink.apache.org/v1beta1", internalMeta.get("apiVersion").asText()); + assertEquals(12L, internalMeta.get("metadata").get("generation").asLong()); + assertEquals( + app.getSpec(), + SpecUtils.deserializeSpecWithMeta(serialized, FlinkDeploymentSpec.class).f0); + + // test backward compatibility + String oldSerialized = + "{\"job\":{\"jarURI\":\"local:///opt/flink/examples/streaming/StateMachineExample.jar\",\"parallelism\":2,\"entryClass\":null,\"args\":[],\"state\":\"running\",\"savepointTriggerNonce\":null,\"initialSavepointPath\":null,\"upgradeMode\":\"stateless\",\"allowNonRestoredState\":null},\"restartNonce\":null,\"flinkConfiguration\":{\"taskmanager.numberOfTaskSlots\":\"2\"},\"image\":\"flink:1.15\",\"imagePullPolicy\":null,\"serviceAccount\":\"flink\",\"flinkVersion\":\"v1_15\", [...] + + var migrated = SpecUtils.deserializeSpecWithMeta(oldSerialized, FlinkDeploymentSpec.class); + assertEquals( + "local:///opt/flink/examples/streaming/StateMachineExample.jar", + migrated.f0.getJob().getJarURI()); + assertNull(migrated.f1); + } +} diff --git a/flink-kubernetes-operator/pom.xml b/flink-kubernetes-operator/pom.xml index 78911d5b..2b130bca 100644 --- a/flink-kubernetes-operator/pom.xml +++ b/flink-kubernetes-operator/pom.xml @@ -56,6 +56,12 @@ under the License. </exclusions> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-kubernetes-api</artifactId> + <version>${project.version}</version> + </dependency> + <!-- FLINK-29384, can be removed when fabric8 pulls snakeyaml 1.32+ --> <dependency> <groupId>org.yaml</groupId> @@ -151,6 +157,14 @@ under the License. <scope>test</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-kubernetes-api</artifactId> + <version>${project.version}</version> + <classifier>tests</classifier> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter-params</artifactId> diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorScopeFormat.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorScopeFormat.java index 81d20561..b13547a2 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorScopeFormat.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorScopeFormat.java @@ -20,7 +20,7 @@ package org.apache.flink.kubernetes.operator.metrics; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.metrics.scope.ScopeFormat; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; /** Format for metrics. * */ public class KubernetesOperatorScopeFormat extends ScopeFormat { diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java index d421832f..6c694e71 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java @@ -20,7 +20,7 @@ package org.apache.flink.kubernetes.operator.metrics; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.metrics.scope.ScopeFormat; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS; import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAME; import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAMESPACE; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java index 0c93e9aa..72a357b5 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java @@ -20,7 +20,7 @@ package org.apache.flink.kubernetes.operator.metrics; import org.apache.flink.configuration.Configuration; import org.apache.flink.runtime.metrics.scope.ScopeFormat; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE; import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAME; import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAMESPACE; import static org.apache.flink.kubernetes.operator.metrics.KubernetesResourceNamespaceScopeFormat.RESOURCE_NS; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java index 490dd03e..3b4d7855 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java @@ -19,6 +19,7 @@ package org.apache.flink.kubernetes.operator.metrics; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java index 155817a8..4f592d25 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java @@ -22,6 +22,7 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.core.plugin.PluginManager; import org.apache.flink.core.plugin.PluginUtils; import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.utils.EnvUtils; import org.apache.flink.metrics.Counter; import org.apache.flink.metrics.Histogram; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java index 8633a8f1..7eb992a8 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java @@ -20,10 +20,11 @@ package org.apache.flink.kubernetes.operator.metrics.lifecycle; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; +import org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState; import org.apache.flink.kubernetes.operator.metrics.CustomResourceMetrics; import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup; -import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils; import org.apache.flink.metrics.Histogram; import org.apache.flink.metrics.MetricGroup; @@ -41,13 +42,13 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.CREATED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.DEPLOYED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.ROLLED_BACK; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.ROLLING_BACK; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.STABLE; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.SUSPENDED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.UPGRADING; /** * Utility for tracking resource lifecycle metrics globally and per namespace. diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricTracker.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricTracker.java index bcba1e2a..894d0799 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricTracker.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricTracker.java @@ -19,6 +19,7 @@ package org.apache.flink.kubernetes.operator.metrics.lifecycle; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState; import org.apache.flink.metrics.Histogram; import org.slf4j.Logger; diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java index 9d9da074..3a689471 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java @@ -17,7 +17,6 @@ package org.apache.flink.kubernetes.operator.reconciler; -import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; @@ -33,12 +32,10 @@ import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType; import org.apache.flink.kubernetes.operator.crd.status.TaskManagerInfo; import org.apache.flink.kubernetes.operator.exception.ValidationException; import org.apache.flink.kubernetes.operator.utils.FlinkUtils; +import org.apache.flink.kubernetes.operator.utils.SpecUtils; import org.apache.flink.kubernetes.operator.utils.StatusRecorder; -import org.apache.flink.util.Preconditions; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.ObjectNode; import io.fabric8.kubernetes.client.CustomResource; import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl; import io.javaoperatorsdk.operator.api.reconciler.RetryInfo; @@ -46,8 +43,6 @@ import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nullable; - import java.util.Optional; import static org.apache.flink.api.common.JobStatus.FINISHED; @@ -59,10 +54,7 @@ public class ReconciliationUtils { private static final Logger LOG = LoggerFactory.getLogger(ReconciliationUtils.class); - public static final String INTERNAL_METADATA_JSON_KEY = "resource_metadata"; - private static final ObjectMapper objectMapper = new ObjectMapper(); - /** * Update status after successful deployment of a new resource spec. Existing reconciliation * errors will be cleared, lastReconciled spec will be updated and for suspended jobs it will @@ -178,16 +170,7 @@ public class ReconciliationUtils { } public static <T> T clone(T object) { - if (object == null) { - return null; - } - try { - return (T) - objectMapper.readValue( - objectMapper.writeValueAsString(object), object.getClass()); - } catch (JsonProcessingException e) { - throw new IllegalStateException(e); - } + return SpecUtils.clone(object); } public static < @@ -260,75 +243,6 @@ public class ReconciliationUtils { || currentReconState == ReconciliationState.UPGRADING; } - /** - * Deserializes the spec and custom metadata object from JSON. - * - * @param specWithMetaString JSON string. - * @param specClass Spec class for deserialization. - * @param <T> Spec type. - * @return Tuple2 of spec and meta. - */ - public static <T extends AbstractFlinkSpec> - Tuple2<T, ReconciliationMetadata> deserializeSpecWithMeta( - @Nullable String specWithMetaString, Class<T> specClass) { - if (specWithMetaString == null) { - return null; - } - - try { - ObjectNode wrapper = (ObjectNode) objectMapper.readTree(specWithMetaString); - ObjectNode internalMeta = (ObjectNode) wrapper.remove(INTERNAL_METADATA_JSON_KEY); - - if (internalMeta == null) { - // migrating from old format - wrapper.remove("apiVersion"); - return Tuple2.of(objectMapper.treeToValue(wrapper, specClass), null); - } else { - return Tuple2.of( - objectMapper.treeToValue(wrapper.get("spec"), specClass), - objectMapper.convertValue(internalMeta, ReconciliationMetadata.class)); - } - } catch (JsonProcessingException e) { - throw new RuntimeException("Could not deserialize spec, this indicates a bug...", e); - } - } - - /** - * Serializes the spec and custom meta information into a JSON string. - * - * @param spec Flink resource spec. - * @param relatedResource Related Flink resource for creating the meta object. - * @return Serialized json. - */ - public static String writeSpecWithMeta( - AbstractFlinkSpec spec, AbstractFlinkResource<?, ?> relatedResource) { - return writeSpecWithMeta(spec, ReconciliationMetadata.from(relatedResource)); - } - - /** - * Serializes the spec and custom meta information into a JSON string. - * - * @param spec Flink resource spec. - * @param metadata Reconciliation meta object. - * @return Serialized json. - */ - public static String writeSpecWithMeta( - AbstractFlinkSpec spec, ReconciliationMetadata metadata) { - - ObjectNode wrapper = objectMapper.createObjectNode(); - - wrapper.set("spec", objectMapper.valueToTree(Preconditions.checkNotNull(spec))); - wrapper.set( - INTERNAL_METADATA_JSON_KEY, - objectMapper.valueToTree(Preconditions.checkNotNull(metadata))); - - try { - return objectMapper.writeValueAsString(wrapper); - } catch (JsonProcessingException e) { - throw new RuntimeException("Could not serialize spec, this indicates a bug...", e); - } - } - public static boolean isJobInTerminalState(CommonStatus<?> status) { var jobState = status.getJobStatus().getState(); return org.apache.flink.api.common.JobStatus.valueOf(jobState).isGloballyTerminalState(); @@ -508,6 +422,6 @@ public class ReconciliationUtils { } reconciliationStatus.setState(ReconciliationState.DEPLOYED); reconciliationStatus.setLastReconciledSpec( - ReconciliationUtils.writeSpecWithMeta(lastSpecWithMeta.f0, lastSpecWithMeta.f1)); + SpecUtils.writeSpecWithMeta(lastSpecWithMeta.f0, lastSpecWithMeta.f1)); } } diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java index c21fe0c8..c9de6e99 100644 --- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java +++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/StatusRecorder.java @@ -24,10 +24,10 @@ import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; import org.apache.flink.kubernetes.operator.crd.status.CommonStatus; import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus; import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus; +import org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState; import org.apache.flink.kubernetes.operator.listener.AuditUtils; import org.apache.flink.kubernetes.operator.listener.FlinkResourceListener; import org.apache.flink.kubernetes.operator.metrics.MetricManager; -import org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ObjectNode; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java index 44fe90b2..32c10191 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java @@ -17,41 +17,21 @@ package org.apache.flink.kubernetes.operator; -import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; -import org.apache.flink.configuration.HighAvailabilityOptions; -import org.apache.flink.configuration.TaskManagerOptions; -import org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory; -import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; -import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; -import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob; -import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec; -import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec; -import org.apache.flink.kubernetes.operator.crd.spec.FlinkVersion; -import org.apache.flink.kubernetes.operator.crd.spec.JobManagerSpec; -import org.apache.flink.kubernetes.operator.crd.spec.JobSpec; -import org.apache.flink.kubernetes.operator.crd.spec.JobState; -import org.apache.flink.kubernetes.operator.crd.spec.Resource; -import org.apache.flink.kubernetes.operator.crd.spec.TaskManagerSpec; -import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode; -import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus; -import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus; import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus; +import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException; import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup; import org.apache.flink.runtime.metrics.MetricRegistry; import org.apache.flink.runtime.metrics.util.TestingMetricRegistry; -import io.fabric8.kubernetes.api.model.Container; import io.fabric8.kubernetes.api.model.ContainerStatus; import io.fabric8.kubernetes.api.model.ContainerStatusBuilder; import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.api.model.ObjectMeta; -import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.PodList; import io.fabric8.kubernetes.api.model.PodListBuilder; -import io.fabric8.kubernetes.api.model.PodSpec; import io.fabric8.kubernetes.api.model.PodStatusBuilder; import io.fabric8.kubernetes.api.model.apps.Deployment; import io.fabric8.kubernetes.api.model.apps.DeploymentCondition; @@ -75,149 +55,23 @@ import java.net.HttpURLConnection; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; -import java.time.Instant; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.Set; -import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import static org.junit.jupiter.api.Assertions.assertTrue; /** Testing utilities. */ -public class TestUtils { - - public static final String TEST_NAMESPACE = "flink-operator-test"; - public static final String TEST_DEPLOYMENT_NAME = "test-cluster"; - public static final String TEST_SESSION_JOB_NAME = "test-session-job"; - public static final String SERVICE_ACCOUNT = "flink-operator"; - public static final String FLINK_VERSION = "latest"; - public static final String IMAGE = String.format("flink:%s", FLINK_VERSION); - public static final String IMAGE_POLICY = "IfNotPresent"; - public static final String SAMPLE_JAR = "local:///tmp/sample.jar"; +public class TestUtils extends BaseTestUtils { private static final String TEST_PLUGINS = "test-plugins"; private static final String PlUGINS_JAR = TEST_PLUGINS + "-test-jar.jar"; - public static FlinkDeployment buildSessionCluster() { - return buildSessionCluster(FlinkVersion.v1_15); - } - - public static FlinkDeployment buildSessionCluster(FlinkVersion version) { - return buildSessionCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, version); - } - - public static FlinkDeployment buildSessionCluster( - String name, String namespace, FlinkVersion version) { - FlinkDeployment deployment = new FlinkDeployment(); - deployment.setStatus(new FlinkDeploymentStatus()); - deployment.setMetadata( - new ObjectMetaBuilder() - .withName(name) - .withNamespace(namespace) - .withCreationTimestamp(Instant.now().toString()) - .withUid(UUID.randomUUID().toString()) - .build()); - deployment.setSpec(getTestFlinkDeploymentSpec(version)); - return deployment; - } - - public static FlinkDeployment buildApplicationCluster() { - return buildApplicationCluster(FlinkVersion.v1_15); - } - - public static FlinkDeployment buildApplicationCluster(String name, String namespace) { - return buildApplicationCluster(name, namespace, FlinkVersion.v1_15); - } - - public static FlinkDeployment buildApplicationCluster(FlinkVersion version) { - return buildApplicationCluster(TEST_DEPLOYMENT_NAME, TEST_NAMESPACE, version); - } - - public static FlinkDeployment buildApplicationCluster( - String name, String namespace, FlinkVersion version) { - FlinkDeployment deployment = buildSessionCluster(name, namespace, version); - deployment - .getSpec() - .setJob( - JobSpec.builder() - .jarURI(SAMPLE_JAR) - .parallelism(1) - .upgradeMode(UpgradeMode.STATELESS) - .state(JobState.RUNNING) - .build()); - deployment.setStatus(deployment.initStatus()); - return deployment; - } - - public static FlinkSessionJob buildSessionJob(String name, String namespace) { - FlinkSessionJob sessionJob = new FlinkSessionJob(); - sessionJob.setStatus(new FlinkSessionJobStatus()); - sessionJob.setMetadata( - new ObjectMetaBuilder() - .withName(name) - .withNamespace(namespace) - .withCreationTimestamp(Instant.now().toString()) - .withUid(UUID.randomUUID().toString()) - .withGeneration(1L) - .build()); - - Map<String, String> conf = new HashMap<>(); - conf.put(KubernetesOperatorConfigOptions.JAR_ARTIFACT_HTTP_HEADER.key(), "header"); - sessionJob.setSpec( - FlinkSessionJobSpec.builder() - .deploymentName(TEST_DEPLOYMENT_NAME) - .job( - JobSpec.builder() - .jarURI(SAMPLE_JAR) - .parallelism(1) - .upgradeMode(UpgradeMode.STATELESS) - .state(JobState.RUNNING) - .build()) - .flinkConfiguration(conf) - .build()); - return sessionJob; - } - - public static FlinkSessionJob buildSessionJob() { - return buildSessionJob(TEST_SESSION_JOB_NAME, TEST_NAMESPACE); - } - - public static FlinkDeploymentSpec getTestFlinkDeploymentSpec(FlinkVersion version) { - Map<String, String> conf = new HashMap<>(); - conf.put(TaskManagerOptions.NUM_TASK_SLOTS.key(), "2"); - conf.put( - HighAvailabilityOptions.HA_MODE.key(), - KubernetesHaServicesFactory.class.getCanonicalName()); - conf.put(HighAvailabilityOptions.HA_STORAGE_PATH.key(), "test"); - conf.put(CheckpointingOptions.SAVEPOINT_DIRECTORY.key(), "test-savepoint-dir"); - conf.put(CheckpointingOptions.CHECKPOINTS_DIRECTORY.key(), "test-checkpoint-dir"); - - return FlinkDeploymentSpec.builder() - .image(IMAGE) - .imagePullPolicy(IMAGE_POLICY) - .serviceAccount(SERVICE_ACCOUNT) - .flinkVersion(version) - .flinkConfiguration(conf) - .jobManager(new JobManagerSpec(new Resource(1.0, "2048m"), 1, null)) - .taskManager(new TaskManagerSpec(new Resource(1.0, "2048m"), null, null)) - .build(); - } - - public static Pod getTestPod(String hostname, String apiVersion, List<Container> containers) { - final PodSpec podSpec = new PodSpec(); - podSpec.setHostname(hostname); - podSpec.setContainers(containers); - final Pod pod = new Pod(); - pod.setApiVersion(apiVersion); - pod.setSpec(podSpec); - return pod; - } - public static PodList createFailedPodList(String crashLoopMessage, String reason) { ContainerStatus cs = new ContainerStatusBuilder() @@ -229,7 +83,7 @@ public class TestUtils { .endState() .build(); - Pod pod = TestUtils.getTestPod("host", "apiVersion", Collections.emptyList()); + Pod pod = getTestPod("host", "apiVersion", Collections.emptyList()); pod.setStatus( new PodStatusBuilder() .withContainerStatuses(Collections.singletonList(cs)) diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java index 4cf3035e..9cfc5d23 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java @@ -64,10 +64,10 @@ import java.util.List; import java.util.Map; import static org.apache.flink.configuration.DeploymentOptions.SHUTDOWN_ON_APPLICATION_FINISH; -import static org.apache.flink.kubernetes.operator.TestUtils.IMAGE; -import static org.apache.flink.kubernetes.operator.TestUtils.IMAGE_POLICY; -import static org.apache.flink.kubernetes.operator.TestUtils.SAMPLE_JAR; -import static org.apache.flink.kubernetes.operator.TestUtils.SERVICE_ACCOUNT; +import static org.apache.flink.kubernetes.operator.BaseTestUtils.IMAGE; +import static org.apache.flink.kubernetes.operator.BaseTestUtils.IMAGE_POLICY; +import static org.apache.flink.kubernetes.operator.BaseTestUtils.SAMPLE_JAR; +import static org.apache.flink.kubernetes.operator.BaseTestUtils.SERVICE_ACCOUNT; import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.DEFAULT_CHECKPOINTING_INTERVAL; import static org.apache.flink.kubernetes.utils.Constants.CONFIG_FILE_LOG4J_NAME; import static org.hamcrest.MatcherAssert.assertThat; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java index 3d38cb83..6600f0c6 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentControllerTest.java @@ -32,9 +32,9 @@ import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode; import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus; import org.apache.flink.kubernetes.operator.crd.status.JobStatus; import org.apache.flink.kubernetes.operator.crd.status.ReconciliationStatus; +import org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState; import org.apache.flink.kubernetes.operator.crd.status.TaskManagerInfo; import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException; -import org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState; import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; import org.apache.flink.kubernetes.operator.reconciler.deployment.AbstractFlinkResourceReconciler; import org.apache.flink.kubernetes.operator.utils.EventRecorder; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkDeploymentMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkDeploymentMetricsTest.java index 81624a57..8ef88b3c 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkDeploymentMetricsTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkDeploymentMetricsTest.java @@ -26,9 +26,9 @@ import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatu import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.OPERATOR_RESOURCE_METRICS_ENABLED; import static org.apache.flink.kubernetes.operator.metrics.FlinkDeploymentMetrics.COUNTER_NAME; import static org.apache.flink.kubernetes.operator.metrics.FlinkDeploymentMetrics.STATUS_GROUP_NAME; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.OPERATOR_RESOURCE_METRICS_ENABLED; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkSessionJobMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkSessionJobMetricsTest.java index 4c813c11..903c496c 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkSessionJobMetricsTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/FlinkSessionJobMetricsTest.java @@ -25,8 +25,8 @@ import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.OPERATOR_RESOURCE_METRICS_ENABLED; import static org.apache.flink.kubernetes.operator.metrics.FlinkSessionJobMetrics.COUNTER_NAME; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.OPERATOR_RESOURCE_METRICS_ENABLED; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java index 9a87f699..387d5ebf 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesClientMetricsTest.java @@ -21,6 +21,7 @@ package org.apache.flink.kubernetes.operator.metrics; import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.operator.TestUtils; import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.utils.KubernetesClientUtils; import io.fabric8.kubernetes.client.KubernetesClientException; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java index 35066040..c9e806ec 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java @@ -26,7 +26,7 @@ import org.junit.jupiter.api.Test; import java.util.Map; -import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; +import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java index cc41bc80..01b96f39 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java @@ -23,13 +23,14 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.kubernetes.operator.TestUtils; import org.apache.flink.kubernetes.operator.config.FlinkConfigManager; import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration; +import org.apache.flink.kubernetes.operator.config.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource; import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob; import org.apache.flink.kubernetes.operator.crd.spec.JobState; import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState; +import org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState; import org.apache.flink.kubernetes.operator.metrics.CustomResourceMetrics; -import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions; import org.apache.flink.kubernetes.operator.metrics.MetricManager; import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils; import org.apache.flink.kubernetes.operator.metrics.TestingMetricListener; @@ -44,14 +45,14 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.LongStream; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.FAILED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED; -import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.CREATED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.DEPLOYED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.FAILED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.ROLLED_BACK; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.ROLLING_BACK; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.STABLE; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.SUSPENDED; +import static org.apache.flink.kubernetes.operator.crd.status.ResourceLifecycleState.UPGRADING; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java index 38b44c66..8445dc57 100644 --- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java +++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ReconciliationUtilsTest.java @@ -18,24 +18,19 @@ package org.apache.flink.kubernetes.operator.utils; import org.apache.flink.configuration.Configuration; -import org.apache.flink.kubernetes.operator.TestUtils; +import org.apache.flink.kubernetes.operator.BaseTestUtils; import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration; import org.apache.flink.kubernetes.operator.crd.FlinkDeployment; -import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec; import org.apache.flink.kubernetes.operator.crd.spec.JobState; import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState; import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils; -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.fasterxml.jackson.databind.node.ObjectNode; import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; -import static org.junit.jupiter.api.Assertions.assertNull; /** Test for {@link org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils}. */ public class ReconciliationUtilsTest { @@ -45,7 +40,7 @@ public class ReconciliationUtilsTest { @Test public void testRescheduleUpgradeImmediately() { - FlinkDeployment app = TestUtils.buildApplicationCluster(); + FlinkDeployment app = BaseTestUtils.buildApplicationCluster(); app.getSpec().getJob().setState(JobState.RUNNING); app.getStatus().getReconciliationStatus().setState(ReconciliationState.DEPLOYED); ReconciliationUtils.updateStatusForDeployedSpec(app, new Configuration()); @@ -67,33 +62,4 @@ public class ReconciliationUtilsTest { assertFalse(updateControl.isUpdateStatus()); assertNotEquals(0, updateControl.getScheduleDelay().get()); } - - @Test - public void testSpecSerializationWithVersion() throws JsonProcessingException { - FlinkDeployment app = TestUtils.buildApplicationCluster(); - app.getMetadata().setGeneration(12L); - String serialized = ReconciliationUtils.writeSpecWithMeta(app.getSpec(), app); - ObjectNode node = (ObjectNode) new ObjectMapper().readTree(serialized); - - ObjectNode internalMeta = - (ObjectNode) node.get(ReconciliationUtils.INTERNAL_METADATA_JSON_KEY); - assertEquals("flink.apache.org/v1beta1", internalMeta.get("apiVersion").asText()); - assertEquals(12L, internalMeta.get("metadata").get("generation").asLong()); - assertEquals( - app.getSpec(), - ReconciliationUtils.deserializeSpecWithMeta(serialized, FlinkDeploymentSpec.class) - .f0); - - // test backward compatibility - String oldSerialized = - "{\"job\":{\"jarURI\":\"local:///opt/flink/examples/streaming/StateMachineExample.jar\",\"parallelism\":2,\"entryClass\":null,\"args\":[],\"state\":\"running\",\"savepointTriggerNonce\":null,\"initialSavepointPath\":null,\"upgradeMode\":\"stateless\",\"allowNonRestoredState\":null},\"restartNonce\":null,\"flinkConfiguration\":{\"taskmanager.numberOfTaskSlots\":\"2\"},\"image\":\"flink:1.15\",\"imagePullPolicy\":null,\"serviceAccount\":\"flink\",\"flinkVersion\":\"v1_15\", [...] - - var migrated = - ReconciliationUtils.deserializeSpecWithMeta( - oldSerialized, FlinkDeploymentSpec.class); - assertEquals( - "local:///opt/flink/examples/streaming/StateMachineExample.jar", - migrated.f0.getJob().getJarURI()); - assertNull(migrated.f1); - } } diff --git a/pom.xml b/pom.xml index edf58d87..f812fcc2 100644 --- a/pom.xml +++ b/pom.xml @@ -51,6 +51,7 @@ under the License. </scm> <modules> + <module>flink-kubernetes-api</module> <module>flink-kubernetes-standalone</module> <module>flink-kubernetes-shaded</module> <module>flink-kubernetes-operator</module> @@ -64,6 +65,7 @@ under the License. <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> <maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version> + <maven-jar-plugin.version>3.3.0</maven-jar-plugin.version> <maven-surefire-plugin.version>3.0.0-M4</maven-surefire-plugin.version> <maven-failsafe-plugin.version>3.0.0-M5</maven-failsafe-plugin.version> <maven-resources-plugin.version>3.2.0</maven-resources-plugin.version> @@ -74,12 +76,13 @@ under the License. <fabric8.version>5.12.3</fabric8.version> <lombok.version>1.18.22</lombok.version> - + <commons-lang3.version>3.12.0</commons-lang3.version> <flink.version>1.15.2</flink.version> <slf4j.version>1.7.36</slf4j.version> <log4j.version>2.17.1</log4j.version> <junit.jupiter.version>5.8.2</junit.jupiter.version> + <jetbrains.version>13.0</jetbrains.version> <spotless.version>2.4.2</spotless.version> <it.skip>true</it.skip>
