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
The following commit(s) were added to refs/heads/main by this push:
new bca630c3 [FLINK-29401] Refactor observer structure
bca630c3 is described below
commit bca630c3a003149683b74a54ccd376f9a9691028
Author: Gyula Fora <[email protected]>
AuthorDate: Wed Oct 12 09:10:08 2022 +0200
[FLINK-29401] Refactor observer structure
---
.../flink/kubernetes/operator/FlinkOperator.java | 9 +-
.../controller/FlinkDeploymentController.java | 6 +-
.../observer/AbstractFlinkResourceObserver.java | 129 +++++++++++
.../operator/observer/JobStatusObserver.java | 27 ++-
...idObserverContext.java => ObserverContext.java} | 11 +-
...r.java => AbstractFlinkDeploymentObserver.java} | 85 +++----
.../observer/deployment/ApplicationObserver.java | 117 +++++-----
.../FlinkDeploymentObserverContext.java} | 26 ++-
...ry.java => FlinkDeploymentObserverFactory.java} | 4 +-
.../observer/deployment/SessionObserver.java | 11 +-
.../sessionjob/FlinkSessionJobObserver.java | 230 +++++++++++++++++++
.../sessionjob/FlinkSessionJobObserverContext.java | 58 +++++
.../observer/sessionjob/SessionJobObserver.java | 246 ---------------------
.../TestingFlinkDeploymentController.java | 4 +-
...rTest.java => FlinkSessionJobObserverTest.java} | 8 +-
15 files changed, 568 insertions(+), 403 deletions(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index d4a97124..adfd7590 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -32,8 +32,8 @@ import
org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGrou
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics;
import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils;
-import
org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
-import
org.apache.flink.kubernetes.operator.observer.sessionjob.SessionJobObserver;
+import
org.apache.flink.kubernetes.operator.observer.deployment.FlinkDeploymentObserverFactory;
+import
org.apache.flink.kubernetes.operator.observer.sessionjob.FlinkSessionJobObserver;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobReconciler;
import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
@@ -129,7 +129,7 @@ public class FlinkOperator {
new ReconcilerFactory(
client, flinkServiceFactory, configManager,
eventRecorder, statusRecorder);
var observerFactory =
- new ObserverFactory(
+ new FlinkDeploymentObserverFactory(
flinkServiceFactory, configManager, statusRecorder,
eventRecorder);
var controller =
@@ -152,7 +152,8 @@ public class FlinkOperator {
var reconciler =
new SessionJobReconciler(
client, flinkServiceFactory, configManager,
eventRecorder, statusRecorder);
- var observer = new SessionJobObserver(flinkServiceFactory,
configManager, eventRecorder);
+ var observer =
+ new FlinkSessionJobObserver(flinkServiceFactory,
configManager, eventRecorder);
var controller =
new FlinkSessionJobController(
configManager,
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
index 0026ac54..b08493ab 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
@@ -24,7 +24,7 @@ import
org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
-import
org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
+import
org.apache.flink.kubernetes.operator.observer.deployment.FlinkDeploymentObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -63,7 +63,7 @@ public class FlinkDeploymentController
private final Set<FlinkResourceValidator> validators;
private final ReconcilerFactory reconcilerFactory;
- private final ObserverFactory observerFactory;
+ private final FlinkDeploymentObserverFactory observerFactory;
private final StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>
statusRecorder;
private final EventRecorder eventRecorder;
@@ -71,7 +71,7 @@ public class FlinkDeploymentController
FlinkConfigManager configManager,
Set<FlinkResourceValidator> validators,
ReconcilerFactory reconcilerFactory,
- ObserverFactory observerFactory,
+ FlinkDeploymentObserverFactory observerFactory,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>
statusRecorder,
EventRecorder eventRecorder) {
this.configManager = configManager;
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
new file mode 100644
index 00000000..9191eac4
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/AbstractFlinkResourceObserver.java
@@ -0,0 +1,129 @@
+/*
+ * 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.observer;
+
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Base observer for all Flink resources. */
+public abstract class AbstractFlinkResourceObserver<
+ CR extends AbstractFlinkResource<?, ?>, CTX extends
ObserverContext>
+ implements Observer<CR> {
+
+ protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ protected final FlinkConfigManager configManager;
+ protected final EventRecorder eventRecorder;
+
+ public AbstractFlinkResourceObserver(
+ FlinkConfigManager configManager, EventRecorder eventRecorder) {
+ this.configManager = configManager;
+ this.eventRecorder = eventRecorder;
+ }
+
+ @Override
+ public final void observe(CR resource, Context<?> context) {
+ var observerContext = getObserverContext(resource, context);
+
+ if (!isResourceReadyToBeObserved(resource, context, observerContext)) {
+ return;
+ }
+
+ // Trigger resource specific observe logic
+ observeInternal(resource, context, observerContext);
+
+ SavepointUtils.resetTriggerIfJobNotRunning(resource, eventRecorder);
+ }
+
+ /**
+ * Get the observer context for the current resource.
+ *
+ * @param resource Resource being observed
+ * @param context Resource context
+ * @return Observer context
+ */
+ protected abstract CTX getObserverContext(CR resource, Context<?> context);
+
+ /**
+ * Check whether the resource should be observed. In certain states such
as suspended
+ * applications or in-progress upgrades and rollbacks, observing is not
necessary.
+ *
+ * @param resource Current resource
+ * @param resourceContext Resource context
+ * @param observerContext Observer context
+ * @return True if we should observe the resource
+ */
+ protected boolean isResourceReadyToBeObserved(
+ CR resource, Context<?> resourceContext, CTX observerContext) {
+ var reconciliationStatus =
resource.getStatus().getReconciliationStatus();
+
+ if (reconciliationStatus.isBeforeFirstDeployment()) {
+ logger.debug("Skipping observe before first deployment");
+ return false;
+ }
+
+ if (reconciliationStatus.getState() ==
ReconciliationState.ROLLING_BACK) {
+ logger.debug("Skipping observe during rollback operation");
+ return false;
+ }
+
+ // We are in the middle or possibly right after an upgrade
+ if (reconciliationStatus.getState() == ReconciliationState.UPGRADING) {
+ // We must check if the upgrade went through without the status
upgrade for some reason
+ updateStatusToDeployedIfAlreadyUpgraded(resource, resourceContext,
observerContext);
+ if (reconciliationStatus.getState() ==
ReconciliationState.UPGRADING) {
+
ReconciliationUtils.clearLastReconciledSpecIfFirstDeploy(resource);
+ logger.debug("Skipping observe before resource is deployed
during upgrade");
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * Internal observer logic specific to each resource type.
+ *
+ * @param resource Resource to be observed
+ * @param resourceContext Resource context
+ * @param observerContext Observer context
+ */
+ protected abstract void observeInternal(
+ CR resource, Context<?> resourceContext, CTX observerContext);
+
+ /**
+ * Checks a resource that is currently in the UPGRADING state whether it
was already deployed
+ * but we simply miss the status information. After comparing the target
resource generation
+ * with the one from the possible deployment if they match we update the
status to the already
+ * DEPLOYED state.
+ *
+ * @param resource Flink resource.
+ * @param resourceContext Context for resource.
+ * @param observerContext Context for observer.
+ */
+ protected abstract void updateStatusToDeployedIfAlreadyUpgraded(
+ CR resource, Context<?> resourceContext, CTX observerContext);
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
index 816b82af..b6aa3283 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobStatusObserver.java
@@ -25,6 +25,7 @@ import
org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.runtime.client.JobStatusMessage;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,15 +35,16 @@ import java.util.Optional;
import java.util.concurrent.TimeoutException;
/** An observer to observe the job status. */
-public abstract class JobStatusObserver<R extends AbstractFlinkResource<?, ?>,
CTX> {
+public abstract class JobStatusObserver<
+ R extends AbstractFlinkResource<?, ?>, CTX extends ObserverContext> {
private static final Logger LOG =
LoggerFactory.getLogger(JobStatusObserver.class);
private static final int MAX_ERROR_STRING_LENGTH = 512;
public static final String MISSING_SESSION_JOB_ERR = "Missing Session Job";
- private final FlinkService flinkService;
- private final EventRecorder eventRecorder;
+ protected final FlinkService flinkService;
+ protected final EventRecorder eventRecorder;
public JobStatusObserver(FlinkService flinkService, EventRecorder
eventRecorder) {
this.flinkService = flinkService;
@@ -53,11 +55,10 @@ public abstract class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>, C
* Observe the status of the flink job.
*
* @param resource The custom resource to be observed.
- * @param deployedConfig Deployed job config.
* @param ctx Observe context.
* @return If job found return true, otherwise return false.
*/
- public boolean observe(R resource, Configuration deployedConfig, CTX ctx) {
+ public boolean observe(R resource, Context resourceContext, CTX ctx) {
var jobStatus = resource.getStatus().getJobStatus();
LOG.info("Observing job status");
var previousJobStatus = jobStatus.getState();
@@ -65,13 +66,13 @@ public abstract class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>, C
List<JobStatusMessage> clusterJobStatuses;
try {
// Query job list from the cluster
- clusterJobStatuses = new
ArrayList<>(flinkService.listJobs(deployedConfig));
+ clusterJobStatuses = new
ArrayList<>(flinkService.listJobs(ctx.getDeployedConfig()));
} catch (Exception e) {
// Error while accessing the rest api, will try again later...
LOG.error("Exception while listing jobs", e);
ifRunningMoveToReconciling(jobStatus, previousJobStatus);
if (e instanceof TimeoutException) {
- onTimeout(ctx);
+ onTimeout(resource, resourceContext, ctx);
}
return false;
}
@@ -84,10 +85,10 @@ public abstract class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>, C
if (targetJobStatusMessage.isEmpty()) {
LOG.warn("No matching jobs found on the cluster");
ifRunningMoveToReconciling(jobStatus, previousJobStatus);
- onTargetJobNotFound(resource, deployedConfig);
+ onTargetJobNotFound(resource, ctx.getDeployedConfig());
return false;
} else {
- updateJobStatus(resource, targetJobStatusMessage.get(),
deployedConfig);
+ updateJobStatus(resource, targetJobStatusMessage.get(),
ctx.getDeployedConfig());
}
ReconciliationUtils.checkAndUpdateStableSpec(resource.getStatus());
return true;
@@ -95,7 +96,7 @@ public abstract class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>, C
LOG.debug("No jobs found on the cluster");
// No jobs found on the cluster, it is possible that the
jobmanager is still starting up
ifRunningMoveToReconciling(jobStatus, previousJobStatus);
- onNoJobsFound(resource, deployedConfig);
+ onNoJobsFound(resource, ctx.getDeployedConfig());
return false;
}
}
@@ -132,9 +133,11 @@ public abstract class JobStatusObserver<R extends
AbstractFlinkResource<?, ?>, C
/**
* Callback when list jobs timeout.
*
- * @param ctx Observe context.
+ * @param resource The Flink resource.
+ * @param resourceContext Resource context.
+ * @param observerContext Observe context.
*/
- protected abstract void onTimeout(CTX ctx);
+ protected abstract void onTimeout(R resource, Context<?> resourceContext,
CTX observerContext);
/**
* Filter the target job status message by the job list from the cluster.
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/VoidObserverContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ObserverContext.java
similarity index 77%
rename from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/VoidObserverContext.java
rename to
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ObserverContext.java
index 3ed3940f..874aba8d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/VoidObserverContext.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/ObserverContext.java
@@ -15,9 +15,12 @@
* limitations under the License.
*/
-package org.apache.flink.kubernetes.operator.observer.context;
+package org.apache.flink.kubernetes.operator.observer;
-/** An empty observer context. */
-public class VoidObserverContext {
- public static final VoidObserverContext INSTANCE = new
VoidObserverContext();
+import org.apache.flink.configuration.Configuration;
+
+/** Context for the currently observed resource. */
+public interface ObserverContext {
+
+ Configuration getDeployedConfig();
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
similarity index 79%
rename from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
rename to
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
index 7319e923..d13068d6 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractDeploymentObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/AbstractFlinkDeploymentObserver.java
@@ -26,14 +26,12 @@ 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.status.FlinkDeploymentStatus;
import
org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
-import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
-import org.apache.flink.kubernetes.operator.observer.Observer;
+import
org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
-import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
import io.fabric8.kubernetes.api.model.ContainerStatus;
@@ -51,67 +49,49 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
-/** The base observer. */
-public abstract class AbstractDeploymentObserver implements
Observer<FlinkDeployment> {
+/** Base observer for session and application clusters. */
+public abstract class AbstractFlinkDeploymentObserver
+ extends AbstractFlinkResourceObserver<FlinkDeployment,
FlinkDeploymentObserverContext> {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected final FlinkService flinkService;
- protected final FlinkConfigManager configManager;
- protected final EventRecorder eventRecorder;
- public AbstractDeploymentObserver(
+ public AbstractFlinkDeploymentObserver(
FlinkService flinkService,
FlinkConfigManager configManager,
EventRecorder eventRecorder) {
+ super(configManager, eventRecorder);
this.flinkService = flinkService;
- this.configManager = configManager;
- this.eventRecorder = eventRecorder;
}
@Override
- public void observe(FlinkDeployment flinkApp, Context<?> context) {
- var status = flinkApp.getStatus();
- var reconciliationStatus = status.getReconciliationStatus();
-
- if (reconciliationStatus.isBeforeFirstDeployment()) {
- logger.debug("Skipping observe before first deployment");
- return;
- }
-
- if (reconciliationStatus.getState() ==
ReconciliationState.ROLLING_BACK) {
- logger.debug("Skipping observe during rollback operation");
- return;
- }
+ protected FlinkDeploymentObserverContext getObserverContext(
+ FlinkDeployment resource, Context<?> context) {
+ return new FlinkDeploymentObserverContext(resource, configManager);
+ }
- // We are in the middle or possibly right after an upgrade
- if (reconciliationStatus.getState() == ReconciliationState.UPGRADING) {
- // We must check if the upgrade went through without the status
upgrade for some reason
- checkIfAlreadyUpgraded(flinkApp, context);
- if (reconciliationStatus.getState() ==
ReconciliationState.UPGRADING) {
-
ReconciliationUtils.clearLastReconciledSpecIfFirstDeploy(flinkApp);
- logger.debug("Skipping observe before resource is deployed
during upgrade");
- return;
- }
- }
+ @Override
+ public void observeInternal(
+ FlinkDeployment flinkDep,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext) {
- Configuration observeConfig = configManager.getObserveConfig(flinkApp);
- if (!isJmDeploymentReady(flinkApp)) {
+ if (!isJmDeploymentReady(flinkDep)) {
// Only observe the JM if we think it's in bad state
- observeJmDeployment(flinkApp, context, observeConfig);
+ observeJmDeployment(flinkDep, context,
observerContext.getDeployedConfig());
}
- if (isJmDeploymentReady(flinkApp)) {
+ if (isJmDeploymentReady(flinkDep)) {
// Only observe session/application if JM is ready
- observeFlinkCluster(flinkApp, context, observeConfig);
+ observeFlinkCluster(flinkDep, context, observerContext);
}
- if (isJmDeploymentReady(flinkApp)) {
- observeClusterInfo(flinkApp, observeConfig);
+ if (isJmDeploymentReady(flinkDep)) {
+ observeClusterInfo(flinkDep, observerContext.getDeployedConfig());
}
- SavepointUtils.resetTriggerIfJobNotRunning(flinkApp, eventRecorder);
- clearErrorsIfDeploymentIsHealthy(flinkApp);
+ clearErrorsIfDeploymentIsHealthy(flinkDep);
}
private void observeClusterInfo(FlinkDeployment flinkApp, Configuration
configuration) {
@@ -256,16 +236,11 @@ public abstract class AbstractDeploymentObserver
implements Observer<FlinkDeploy
err);
}
- /**
- * Checks a deployment that is currently in the UPGRADING state whether it
was already deployed
- * but we simply miss the status information. After comparing the target
resource generation
- * with the one from the possible deployment if they match we update the
status to the already
- * DEPLOYED state.
- *
- * @param flinkDep Flink resource to check.
- * @param context Context for reconciliation.
- */
- private void checkIfAlreadyUpgraded(FlinkDeployment flinkDep, Context<?>
context) {
+ @Override
+ protected void updateStatusToDeployedIfAlreadyUpgraded(
+ FlinkDeployment flinkDep,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext) {
var status = flinkDep.getStatus();
Optional<Deployment> depOpt =
context.getSecondaryResource(Deployment.class);
depOpt.ifPresent(
@@ -307,8 +282,10 @@ public abstract class AbstractDeploymentObserver
implements Observer<FlinkDeploy
*
* @param flinkApp the target flinkDeployment resource
* @param context the context with which the operation is executed
- * @param deployedConfig config that is deployed on the Flink cluster
+ * @param observerContext Observer context
*/
protected abstract void observeFlinkCluster(
- FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig);
+ FlinkDeployment flinkApp,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext);
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
index c41ee389..abb156fb 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserver.java
@@ -25,7 +25,6 @@ import
org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.observer.ClusterHealthObserver;
import org.apache.flink.kubernetes.operator.observer.JobStatusObserver;
import org.apache.flink.kubernetes.operator.observer.SavepointObserver;
-import
org.apache.flink.kubernetes.operator.observer.context.ApplicationObserverContext;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -39,10 +38,11 @@ import java.util.Optional;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED;
/** The observer of {@link
org.apache.flink.kubernetes.operator.config.Mode#APPLICATION} cluster. */
-public class ApplicationObserver extends AbstractDeploymentObserver {
+public class ApplicationObserver extends AbstractFlinkDeploymentObserver {
private final SavepointObserver<FlinkDeployment, FlinkDeploymentStatus>
savepointObserver;
- private final JobStatusObserver<FlinkDeployment,
ApplicationObserverContext> jobStatusObserver;
+ private final JobStatusObserver<FlinkDeployment,
FlinkDeploymentObserverContext>
+ jobStatusObserver;
private final ClusterHealthObserver clusterHealthObserver;
@@ -53,70 +53,77 @@ public class ApplicationObserver extends
AbstractDeploymentObserver {
super(flinkService, configManager, eventRecorder);
this.savepointObserver =
new SavepointObserver<>(flinkService, configManager,
eventRecorder);
- this.jobStatusObserver =
- new JobStatusObserver<>(flinkService, eventRecorder) {
- @Override
- public void onTimeout(ApplicationObserverContext ctx) {
- observeJmDeployment(ctx.flinkApp, ctx.context,
ctx.deployedConfig);
- }
-
- @Override
- protected Optional<JobStatusMessage> filterTargetJob(
- JobStatus status, List<JobStatusMessage>
clusterJobStatuses) {
- if (!clusterJobStatuses.isEmpty()) {
- return Optional.of(clusterJobStatuses.get(0));
- }
- return Optional.empty();
- }
-
- @Override
- protected void onTargetJobNotFound(
- FlinkDeployment resource, Configuration config) {
- // This should never happen for application clusters,
there is something
- // wrong
- setUnknownJobError(resource);
- }
-
- /**
- * We found a job on an application cluster that doesn't
match the expected job.
- * Trigger error.
- *
- * @param deployment Application deployment.
- */
- private void setUnknownJobError(FlinkDeployment
deployment) {
- deployment
- .getStatus()
- .getJobStatus()
-
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
- String err = "Unrecognized Job for Application
deployment";
- logger.error(err);
-
ReconciliationUtils.updateForReconciliationError(deployment, err);
- eventRecorder.triggerEvent(
- deployment,
- EventRecorder.Type.Warning,
- EventRecorder.Reason.Missing,
- EventRecorder.Component.Job,
- err);
- }
- };
+ this.jobStatusObserver = new ApplicationJobObserver(flinkService,
eventRecorder);
this.clusterHealthObserver = new ClusterHealthObserver(flinkService);
}
@Override
protected void observeFlinkCluster(
- FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig) {
+ FlinkDeployment flinkApp,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext) {
logger.debug("Observing application cluster");
- boolean jobFound =
- jobStatusObserver.observe(
- flinkApp,
- deployedConfig,
- new ApplicationObserverContext(flinkApp, context,
deployedConfig));
+ boolean jobFound = jobStatusObserver.observe(flinkApp, context,
observerContext);
if (jobFound) {
+ var deployedConfig = observerContext.getDeployedConfig();
savepointObserver.observeSavepointStatus(flinkApp, deployedConfig);
if
(deployedConfig.getBoolean(OPERATOR_CLUSTER_HEALTH_CHECK_ENABLED)) {
clusterHealthObserver.observe(flinkApp, deployedConfig);
}
}
}
+
+ private class ApplicationJobObserver
+ extends JobStatusObserver<FlinkDeployment,
FlinkDeploymentObserverContext> {
+ public ApplicationJobObserver(FlinkService flinkService, EventRecorder
eventRecorder) {
+ super(flinkService, eventRecorder);
+ }
+
+ @Override
+ public void onTimeout(
+ FlinkDeployment flinkDep,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext) {
+ observeJmDeployment(flinkDep, context,
observerContext.getDeployedConfig());
+ }
+
+ @Override
+ protected Optional<JobStatusMessage> filterTargetJob(
+ JobStatus status, List<JobStatusMessage> clusterJobStatuses) {
+ if (!clusterJobStatuses.isEmpty()) {
+ return Optional.of(clusterJobStatuses.get(0));
+ }
+ return Optional.empty();
+ }
+
+ @Override
+ protected void onTargetJobNotFound(FlinkDeployment resource,
Configuration config) {
+ // This should never happen for application clusters, there is
something
+ // wrong
+ setUnknownJobError(resource);
+ }
+
+ /**
+ * We found a job on an application cluster that doesn't match the
expected job. Trigger
+ * error.
+ *
+ * @param deployment Application deployment.
+ */
+ private void setUnknownJobError(FlinkDeployment deployment) {
+ deployment
+ .getStatus()
+ .getJobStatus()
+
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+ String err = "Unrecognized Job for Application deployment";
+ logger.error(err);
+ ReconciliationUtils.updateForReconciliationError(deployment, err);
+ eventRecorder.triggerEvent(
+ deployment,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.Missing,
+ EventRecorder.Component.Job,
+ err);
+ }
+ }
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverContext.java
similarity index 55%
rename from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
rename to
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverContext.java
index 6582d792..782b081d 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/context/ApplicationObserverContext.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverContext.java
@@ -15,24 +15,26 @@
* limitations under the License.
*/
-package org.apache.flink.kubernetes.operator.observer.context;
+package org.apache.flink.kubernetes.operator.observer.deployment;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.observer.ObserverContext;
-import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Value;
-/** A context for application observer. */
-public class ApplicationObserverContext {
+/** Context for observing {@link FlinkDeployment} resources. */
+@Value
+public class FlinkDeploymentObserverContext implements ObserverContext {
- public final FlinkDeployment flinkApp;
- public final Context<?> context;
- public final Configuration deployedConfig;
+ Configuration deployedConfig;
- public ApplicationObserverContext(
- FlinkDeployment flinkApp, Context<?> context, Configuration
deployedConfig) {
- this.flinkApp = flinkApp;
- this.context = context;
- this.deployedConfig = deployedConfig;
+ protected FlinkDeploymentObserverContext(
+ FlinkDeployment resource, FlinkConfigManager configManager) {
+ this.deployedConfig =
+
resource.getStatus().getReconciliationStatus().isBeforeFirstDeployment()
+ ? null
+ : configManager.getObserveConfig(resource);
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverFactory.java
similarity index 97%
rename from
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
rename to
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverFactory.java
index b2145622..e0471d1c 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/ObserverFactory.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/FlinkDeploymentObserverFactory.java
@@ -32,7 +32,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/** The factory to create the observer based on the {@link FlinkDeployment}
mode. */
-public class ObserverFactory {
+public class FlinkDeploymentObserverFactory {
private final FlinkServiceFactory flinkServiceFactory;
private final FlinkConfigManager configManager;
@@ -41,7 +41,7 @@ public class ObserverFactory {
private final Map<Tuple2<Mode, KubernetesDeploymentMode>,
Observer<FlinkDeployment>>
observerMap;
- public ObserverFactory(
+ public FlinkDeploymentObserverFactory(
FlinkServiceFactory flinkServiceFactory,
FlinkConfigManager configManager,
StatusRecorder<FlinkDeployment, FlinkDeploymentStatus>
statusRecorder,
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
index 74054ea7..1f79003f 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/deployment/SessionObserver.java
@@ -17,7 +17,6 @@
package org.apache.flink.kubernetes.operator.observer.deployment;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
@@ -29,7 +28,7 @@ import io.javaoperatorsdk.operator.api.reconciler.Context;
import java.util.concurrent.TimeoutException;
/** The observer of the {@link
org.apache.flink.kubernetes.operator.config.Mode#SESSION} cluster. */
-public class SessionObserver extends AbstractDeploymentObserver {
+public class SessionObserver extends AbstractFlinkDeploymentObserver {
public SessionObserver(
FlinkService flinkService,
@@ -40,11 +39,13 @@ public class SessionObserver extends
AbstractDeploymentObserver {
@Override
public void observeFlinkCluster(
- FlinkDeployment deployment, Context<?> context, Configuration
deployedConfig) {
+ FlinkDeployment deployment,
+ Context<?> context,
+ FlinkDeploymentObserverContext observerContext) {
// Check if session cluster can serve rest calls following our
practice in JobObserver
try {
logger.debug("Observing session cluster");
- flinkService.listJobs(deployedConfig);
+ flinkService.listJobs(observerContext.getDeployedConfig());
var rs = deployment.getStatus().getReconciliationStatus();
if (rs.getState() == ReconciliationState.DEPLOYED) {
rs.markReconciledSpecAsStable();
@@ -53,7 +54,7 @@ public class SessionObserver extends
AbstractDeploymentObserver {
logger.error("REST service in session cluster is bad now", e);
if (e instanceof TimeoutException) {
// check for problems with the underlying deployment
- observeJmDeployment(deployment, context, deployedConfig);
+ observeJmDeployment(deployment, context,
observerContext.getDeployedConfig());
}
}
}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
new file mode 100644
index 00000000..1b9414e8
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserver.java
@@ -0,0 +1,230 @@
+/*
+ * 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.observer.sessionjob;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
+import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
+import
org.apache.flink.kubernetes.operator.observer.AbstractFlinkResourceObserver;
+import org.apache.flink.kubernetes.operator.observer.JobStatusObserver;
+import org.apache.flink.kubernetes.operator.observer.SavepointObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** The observer of {@link FlinkSessionJob}. */
+public class FlinkSessionJobObserver
+ extends AbstractFlinkResourceObserver<FlinkSessionJob,
FlinkSessionJobObserverContext> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(FlinkSessionJobObserver.class);
+ private final FlinkServiceFactory flinkServiceFactory;
+
+ public FlinkSessionJobObserver(
+ FlinkServiceFactory flinkServiceFactory,
+ FlinkConfigManager configManager,
+ EventRecorder eventRecorder) {
+ super(configManager, eventRecorder);
+ this.flinkServiceFactory = flinkServiceFactory;
+ }
+
+ @Override
+ protected FlinkSessionJobObserverContext getObserverContext(
+ FlinkSessionJob resource, Context<?> context) {
+ return new FlinkSessionJobObserverContext(
+ resource, context, flinkServiceFactory, configManager);
+ }
+
+ @Override
+ protected boolean isResourceReadyToBeObserved(
+ FlinkSessionJob resource,
+ Context<?> context,
+ FlinkSessionJobObserverContext observerContext) {
+ return super.isResourceReadyToBeObserved(resource, context,
observerContext)
+ && observerContext.isReadyToReconcile();
+ }
+
+ @Override
+ protected void observeInternal(
+ FlinkSessionJob flinkSessionJob,
+ Context<?> ctx,
+ FlinkSessionJobObserverContext observerContext) {
+
+ var jobStatusObserver = new SessionJobStatusObserver(observerContext,
eventRecorder);
+ var jobFound = jobStatusObserver.observe(flinkSessionJob, ctx,
observerContext);
+
+ if (jobFound) {
+ var savepointObserver =
+ new SavepointObserver<FlinkSessionJob,
FlinkSessionJobStatus>(
+ observerContext.getFlinkService(), configManager,
eventRecorder);
+ savepointObserver.observeSavepointStatus(
+ flinkSessionJob, observerContext.getDeployedConfig());
+ }
+ }
+
+ @Override
+ protected void updateStatusToDeployedIfAlreadyUpgraded(
+ FlinkSessionJob flinkSessionJob,
+ Context<?> ctx,
+ FlinkSessionJobObserverContext observerContext) {
+ var uid = flinkSessionJob.getMetadata().getUid();
+ Collection<JobStatusMessage> jobStatusMessages;
+ try {
+ jobStatusMessages =
+
observerContext.getFlinkService().listJobs(observerContext.getDeployedConfig());
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to list jobs", e);
+ }
+ var matchedJobs = new ArrayList<JobID>();
+ for (JobStatusMessage jobStatusMessage : jobStatusMessages) {
+ var jobId = jobStatusMessage.getJobId();
+ if (jobId.getLowerPart() == uid.hashCode()
+ &&
!jobStatusMessage.getJobState().isGloballyTerminalState()) {
+ matchedJobs.add(jobId);
+ }
+ }
+
+ if (matchedJobs.isEmpty()) {
+ return;
+ } else if (matchedJobs.size() > 1) {
+ // this indicates a bug, which means we have more than one running
job for a single
+ // SessionJob CR.
+ throw new RuntimeException(
+ String.format(
+ "Unexpected case: %d job found for the resource
with uid: %s",
+ matchedJobs.size(),
flinkSessionJob.getMetadata().getUid()));
+ } else {
+ var matchedJobID = matchedJobs.get(0);
+ Long upgradeTargetGeneration =
+
ReconciliationUtils.getUpgradeTargetGeneration(flinkSessionJob);
+ long deployedGeneration = matchedJobID.getUpperPart();
+ var oldJobID =
flinkSessionJob.getStatus().getJobStatus().getJobId();
+
+ if (upgradeTargetGeneration == deployedGeneration) {
+ LOG.info(
+ "Pending upgrade is already deployed, updating status.
Old jobID:{}, new jobID:{}",
+ oldJobID,
+ matchedJobID.toHexString());
+
ReconciliationUtils.updateStatusForAlreadyUpgraded(flinkSessionJob);
+ flinkSessionJob
+ .getStatus()
+ .getJobStatus()
+
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+
flinkSessionJob.getStatus().getJobStatus().setJobId(matchedJobID.toHexString());
+ } else {
+ var msg =
+ String.format(
+ "Running job %s's generation %s doesn't match
upgrade target generation %s.",
+ matchedJobID.toHexString(),
+ deployedGeneration,
+ upgradeTargetGeneration);
+ throw new RuntimeException(msg);
+ }
+ }
+ }
+
+ private static class SessionJobStatusObserver
+ extends JobStatusObserver<FlinkSessionJob,
FlinkSessionJobObserverContext> {
+
+ public SessionJobStatusObserver(
+ FlinkSessionJobObserverContext observerContext, EventRecorder
eventRecorder) {
+ super(observerContext.getFlinkService(), eventRecorder);
+ }
+
+ @Override
+ protected void onTimeout(
+ FlinkSessionJob sessionJob,
+ Context<?> ctx,
+ FlinkSessionJobObserverContext sessionJobObserverContext) {}
+
+ @Override
+ protected Optional<JobStatusMessage> filterTargetJob(
+ JobStatus status, List<JobStatusMessage> clusterJobStatuses) {
+ var jobId =
+ Preconditions.checkNotNull(
+ status.getJobId(), "The jobID to be observed
should not be null");
+ var matchedList =
+ clusterJobStatuses.stream()
+ .filter(job ->
job.getJobId().toHexString().equals(jobId))
+ .collect(Collectors.toList());
+ Preconditions.checkArgument(
+ matchedList.size() <= 1,
+ String.format(
+ "Expected one job for JobID: %s, but %d founded",
+ status.getJobId(), matchedList.size()));
+
+ if (matchedList.size() == 0) {
+ LOG.warn("No job found for JobID: {}", jobId);
+ return Optional.empty();
+ } else {
+ return Optional.of(matchedList.get(0));
+ }
+ }
+
+ @Override
+ protected void onTargetJobNotFound(FlinkSessionJob resource,
Configuration config) {
+ ifHaDisabledMarkSessionJobMissing(resource, config);
+ }
+
+ @Override
+ protected void onNoJobsFound(FlinkSessionJob resource, Configuration
config) {
+ ifHaDisabledMarkSessionJobMissing(resource, config);
+ }
+
+ /**
+ * When HA is disabled the session job will not recover on JM
restarts. If the JM goes down
+ * / restarted the session job should be marked missing.
+ *
+ * @param sessionJob Flink session job.
+ * @param conf Flink config.
+ */
+ private void ifHaDisabledMarkSessionJobMissing(
+ FlinkSessionJob sessionJob, Configuration conf) {
+ if (HighAvailabilityMode.isHighAvailabilityModeActivated(conf)) {
+ return;
+ }
+ sessionJob
+ .getStatus()
+ .getJobStatus()
+
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
+ LOG.error(MISSING_SESSION_JOB_ERR);
+ ReconciliationUtils.updateForReconciliationError(sessionJob,
MISSING_SESSION_JOB_ERR);
+ eventRecorder.triggerEvent(
+ sessionJob,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.Missing,
+ EventRecorder.Component.Job,
+ MISSING_SESSION_JOB_ERR);
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverContext.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverContext.java
new file mode 100644
index 00000000..9b0cfab1
--- /dev/null
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverContext.java
@@ -0,0 +1,58 @@
+/*
+ * 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.observer.sessionjob;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.observer.ObserverContext;
+import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkService;
+import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
+
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Value;
+
+/** Context for observing {@link FlinkSessionJob} resources. */
+@Value
+public class FlinkSessionJobObserverContext implements ObserverContext {
+
+ boolean readyToReconcile;
+
+ FlinkService flinkService;
+
+ Configuration deployedConfig;
+
+ public FlinkSessionJobObserverContext(
+ FlinkSessionJob resource,
+ Context<?> resourceContext,
+ FlinkServiceFactory flinkServiceFactory,
+ FlinkConfigManager configManager) {
+ var flinkDepOpt =
resourceContext.getSecondaryResource(FlinkDeployment.class);
+ this.readyToReconcile =
SessionJobReconciler.sessionClusterReady(flinkDepOpt);
+ if (readyToReconcile) {
+ var flinkDeployment = flinkDepOpt.get();
+ flinkService = flinkServiceFactory.getOrCreate(flinkDeployment);
+ deployedConfig =
configManager.getSessionJobConfig(flinkDeployment, resource.getSpec());
+ } else {
+ flinkService = null;
+ deployedConfig = null;
+ }
+ }
+}
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
deleted file mode 100644
index 17580dc0..00000000
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserver.java
+++ /dev/null
@@ -1,246 +0,0 @@
-/*
- * 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.observer.sessionjob;
-
-import org.apache.flink.api.common.JobID;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
-import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
-import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
-import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
-import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
-import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
-import org.apache.flink.kubernetes.operator.observer.JobStatusObserver;
-import org.apache.flink.kubernetes.operator.observer.Observer;
-import org.apache.flink.kubernetes.operator.observer.SavepointObserver;
-import
org.apache.flink.kubernetes.operator.observer.context.VoidObserverContext;
-import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
-import
org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobReconciler;
-import org.apache.flink.kubernetes.operator.service.FlinkService;
-import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
-import org.apache.flink.kubernetes.operator.utils.EventRecorder;
-import org.apache.flink.kubernetes.operator.utils.SavepointUtils;
-import org.apache.flink.runtime.client.JobStatusMessage;
-import org.apache.flink.runtime.jobmanager.HighAvailabilityMode;
-import org.apache.flink.util.Preconditions;
-
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Optional;
-import java.util.stream.Collectors;
-
-/** The observer of {@link FlinkSessionJob}. */
-public class SessionJobObserver implements Observer<FlinkSessionJob> {
-
- private static final Logger LOG =
LoggerFactory.getLogger(SessionJobObserver.class);
- private final FlinkServiceFactory flinkServiceFactory;
- private final FlinkConfigManager configManager;
- private final EventRecorder eventRecorder;
-
- public SessionJobObserver(
- FlinkServiceFactory flinkServiceFactory,
- FlinkConfigManager configManager,
- EventRecorder eventRecorder) {
- this.flinkServiceFactory = flinkServiceFactory;
- this.configManager = configManager;
- this.eventRecorder = eventRecorder;
- }
-
- private JobStatusObserver<FlinkSessionJob, VoidObserverContext>
getJobStatusObserver(
- FlinkService flinkService) {
- return new JobStatusObserver<>(flinkService, eventRecorder) {
- @Override
- protected void onTimeout(VoidObserverContext
sessionJobObserverContext) {}
-
- @Override
- protected Optional<JobStatusMessage> filterTargetJob(
- JobStatus status, List<JobStatusMessage>
clusterJobStatuses) {
- var jobId =
- Preconditions.checkNotNull(
- status.getJobId(), "The jobID to be observed
should not be null");
- var matchedList =
- clusterJobStatuses.stream()
- .filter(job ->
job.getJobId().toHexString().equals(jobId))
- .collect(Collectors.toList());
- Preconditions.checkArgument(
- matchedList.size() <= 1,
- String.format(
- "Expected one job for JobID: %s, but %d
founded",
- status.getJobId(), matchedList.size()));
-
- if (matchedList.size() == 0) {
- LOG.warn("No job found for JobID: {}", jobId);
- return Optional.empty();
- } else {
- return Optional.of(matchedList.get(0));
- }
- }
-
- @Override
- protected void onTargetJobNotFound(FlinkSessionJob resource,
Configuration config) {
- ifHaDisabledMarkSessionJobMissing(resource, config);
- }
-
- @Override
- protected void onNoJobsFound(FlinkSessionJob resource,
Configuration config) {
- ifHaDisabledMarkSessionJobMissing(resource, config);
- }
-
- /**
- * When HA is disabled the session job will not recover on JM
restarts. If the JM goes
- * down / restarted the session job should be marked missing.
- *
- * @param sessionJob Flink session job.
- * @param conf Flink config.
- */
- private void ifHaDisabledMarkSessionJobMissing(
- FlinkSessionJob sessionJob, Configuration conf) {
- if
(HighAvailabilityMode.isHighAvailabilityModeActivated(conf)) {
- return;
- }
- sessionJob
- .getStatus()
- .getJobStatus()
-
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
- LOG.error(MISSING_SESSION_JOB_ERR);
- ReconciliationUtils.updateForReconciliationError(
- sessionJob, MISSING_SESSION_JOB_ERR);
- eventRecorder.triggerEvent(
- sessionJob,
- EventRecorder.Type.Warning,
- EventRecorder.Reason.Missing,
- EventRecorder.Component.Job,
- MISSING_SESSION_JOB_ERR);
- }
- };
- }
-
- @Override
- public void observe(FlinkSessionJob flinkSessionJob, Context<?> context) {
- var status = flinkSessionJob.getStatus();
- var reconciliationStatus = status.getReconciliationStatus();
-
- if (reconciliationStatus.isBeforeFirstDeployment()) {
- LOG.debug("Skipping observe before first deployment");
- return;
- }
-
- if (reconciliationStatus.getState() ==
ReconciliationState.ROLLING_BACK) {
- LOG.debug("Skipping observe during rollback operation");
- return;
- }
-
- Optional<FlinkDeployment> flinkDepOpt =
context.getSecondaryResource(FlinkDeployment.class);
-
- if (!SessionJobReconciler.sessionClusterReady(flinkDepOpt)) {
- return;
- }
-
- FlinkService flinkService =
flinkServiceFactory.getOrCreate(flinkDepOpt.get());
- var jobStatusObserver = getJobStatusObserver(flinkService);
- var deployedConfig =
- configManager.getSessionJobConfig(flinkDepOpt.get(),
flinkSessionJob.getSpec());
-
- // We are in the middle or possibly right after an upgrade
- if (reconciliationStatus.getState() == ReconciliationState.UPGRADING) {
- // We must check if the upgrade went through without the status
upgrade for some reason
- checkIfAlreadyUpgraded(flinkSessionJob, deployedConfig,
flinkService);
- if (reconciliationStatus.getState() ==
ReconciliationState.UPGRADING) {
- LOG.debug("Skipping observe before resource is deployed during
upgrade");
-
ReconciliationUtils.clearLastReconciledSpecIfFirstDeploy(flinkSessionJob);
- return;
- }
- }
-
- var jobFound =
- jobStatusObserver.observe(
- flinkSessionJob, deployedConfig,
VoidObserverContext.INSTANCE);
-
- if (jobFound) {
- var savepointObserver =
- new SavepointObserver<FlinkSessionJob,
FlinkSessionJobStatus>(
- flinkService, configManager, eventRecorder);
- savepointObserver.observeSavepointStatus(flinkSessionJob,
deployedConfig);
- }
- SavepointUtils.resetTriggerIfJobNotRunning(flinkSessionJob,
eventRecorder);
- }
-
- private void checkIfAlreadyUpgraded(
- FlinkSessionJob flinkSessionJob,
- Configuration deployedConfig,
- FlinkService flinkService) {
- var uid = flinkSessionJob.getMetadata().getUid();
- Collection<JobStatusMessage> jobStatusMessages;
- try {
- jobStatusMessages = flinkService.listJobs(deployedConfig);
- } catch (Exception e) {
- throw new RuntimeException("Failed to list jobs", e);
- }
- var matchedJobs = new ArrayList<JobID>();
- for (JobStatusMessage jobStatusMessage : jobStatusMessages) {
- var jobId = jobStatusMessage.getJobId();
- if (jobId.getLowerPart() == uid.hashCode()
- &&
!jobStatusMessage.getJobState().isGloballyTerminalState()) {
- matchedJobs.add(jobId);
- }
- }
-
- if (matchedJobs.isEmpty()) {
- return;
- } else if (matchedJobs.size() > 1) {
- // this indicates a bug, which means we have more than one running
job for a single
- // SessionJob CR.
- throw new RuntimeException(
- String.format(
- "Unexpected case: %d job found for the resource
with uid: %s",
- matchedJobs.size(),
flinkSessionJob.getMetadata().getUid()));
- } else {
- var matchedJobID = matchedJobs.get(0);
- Long upgradeTargetGeneration =
-
ReconciliationUtils.getUpgradeTargetGeneration(flinkSessionJob);
- long deployedGeneration = matchedJobID.getUpperPart();
- var oldJobID =
flinkSessionJob.getStatus().getJobStatus().getJobId();
-
- if (upgradeTargetGeneration == deployedGeneration) {
- LOG.info(
- "Pending upgrade is already deployed, updating status.
Old jobID:{}, new jobID:{}",
- oldJobID,
- matchedJobID.toHexString());
-
ReconciliationUtils.updateStatusForAlreadyUpgraded(flinkSessionJob);
- flinkSessionJob
- .getStatus()
- .getJobStatus()
-
.setState(org.apache.flink.api.common.JobStatus.RECONCILING.name());
-
flinkSessionJob.getStatus().getJobStatus().setJobId(matchedJobID.toHexString());
- } else {
- var msg =
- String.format(
- "Running job %s's generation %s doesn't match
upgrade target generation %s.",
- matchedJobID.toHexString(),
- deployedGeneration,
- upgradeTargetGeneration);
- throw new RuntimeException(msg);
- }
- }
- }
-}
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
index 14ba9671..046130e3 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -23,7 +23,7 @@ import
org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
-import
org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
+import
org.apache.flink.kubernetes.operator.observer.deployment.FlinkDeploymentObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import
org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
import org.apache.flink.kubernetes.operator.service.FlinkServiceFactory;
@@ -83,7 +83,7 @@ public class TestingFlinkDeploymentController
configManager,
eventRecorder,
statusRecorder),
- new ObserverFactory(
+ new FlinkDeploymentObserverFactory(
flinkServiceFactory, configManager,
statusRecorder, eventRecorder),
statusRecorder,
eventRecorder);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
similarity index 98%
rename from
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
rename to
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
index f42287f4..d082ac74 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/FlinkSessionJobObserverTest.java
@@ -53,14 +53,14 @@ import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
-/** Tests for {@link SessionJobObserver}. */
+/** Tests for {@link FlinkSessionJobObserver}. */
@EnableKubernetesMockClient(crud = true)
-public class SessionJobObserverTest {
+public class FlinkSessionJobObserverTest {
private KubernetesClient kubernetesClient;
private final FlinkConfigManager configManager = new
FlinkConfigManager(new Configuration());
private TestingFlinkService flinkService;
- private SessionJobObserver observer;
+ private FlinkSessionJobObserver observer;
private SessionJobReconciler reconciler;
private EventCollector eventCollector = new EventCollector();
@@ -72,7 +72,7 @@ public class SessionJobObserverTest {
var statusRecorder = new TestingStatusRecorder<FlinkSessionJob,
FlinkSessionJobStatus>();
flinkService = new TestingFlinkService();
FlinkServiceFactory flinkServiceFactory = new
TestingFlinkServiceFactory(flinkService);
- observer = new SessionJobObserver(flinkServiceFactory, configManager,
eventRecorder);
+ observer = new FlinkSessionJobObserver(flinkServiceFactory,
configManager, eventRecorder);
reconciler =
new SessionJobReconciler(
kubernetesClient,
