You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by GitBox <gi...@apache.org> on 2022/03/08 06:17:38 UTC
[GitHub] [flink-kubernetes-operator] gyfora commented on a change in pull request #46: [FLINK-26261] Introduce status and create event for terminal deployment error
gyfora commented on a change in pull request #46:
URL: https://github.com/apache/flink-kubernetes-operator/pull/46#discussion_r821349489
##########
File path: flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/JobManagerDeploymentStatus.java
##########
@@ -54,8 +58,9 @@
operatorConfiguration.getPortCheckIntervalInSec(),
TimeUnit.SECONDS);
case MISSING:
+ case ERROR:
default:
- return null;
+ return UpdateControl.noUpdate();
Review comment:
An error will trigger a status update, is `noUpdate` correct here?
##########
File path: flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
##########
@@ -113,29 +118,70 @@ public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
Configuration effectiveConfig =
FlinkUtils.getEffectiveConfig(flinkApp, defaultConfig.getFlinkConfig());
- boolean readyToReconcile = observer.observe(flinkApp, context, effectiveConfig);
- if (!readyToReconcile) {
- return flinkApp.getStatus()
- .getJobManagerDeploymentStatus()
- .toUpdateControl(flinkApp, operatorConfiguration);
- }
-
try {
+ boolean readyToReconcile = observer.observe(flinkApp, context, effectiveConfig);
+ if (!readyToReconcile) {
+ return flinkApp.getStatus()
+ .getJobManagerDeploymentStatus()
+ .toUpdateControl(flinkApp, operatorConfiguration);
+ }
+
UpdateControl<FlinkDeployment> updateControl =
reconcilerFactory
.getOrCreate(flinkApp)
.reconcile(operatorNamespace, flinkApp, context, effectiveConfig);
updateForReconciliationSuccess(flinkApp);
return updateControl;
- } catch (InvalidDeploymentException ide) {
- LOG.error("Reconciliation failed", ide);
- updateForReconciliationError(flinkApp, ide.getMessage());
+ } catch (DeploymentFailedException dfe) {
+ updateForDeploymentFailed(flinkApp, dfe);
return UpdateControl.updateStatus(flinkApp);
} catch (Exception e) {
throw new ReconciliationException(e);
}
}
+ private void updateForDeploymentFailed(
+ FlinkDeployment flinkApp, DeploymentFailedException dfe) {
+ LOG.error(
+ "Deployment {}/{} failed with {}",
+ flinkApp.getMetadata().getNamespace(),
+ flinkApp.getMetadata().getName(),
+ dfe.getMessage());
+ flinkApp.getStatus().setJobManagerDeploymentStatus(JobManagerDeploymentStatus.ERROR);
+ updateForReconciliationError(flinkApp, dfe.getMessage());
+
+ // TODO: avoid repeated event
+ EventBuilder evtb =
+ new EventBuilder()
+ .withApiVersion("v1")
+ .withNewInvolvedObject()
+ .withKind(flinkApp.getKind())
+ .withName(flinkApp.getMetadata().getName())
+ .withNamespace(flinkApp.getMetadata().getNamespace())
+ .withUid(flinkApp.getMetadata().getUid())
+ .endInvolvedObject()
+ .withType(dfe.deployCondition.getType())
+ .withReason(dfe.deployCondition.getReason())
+ // TODO: timestamp
+ // .withEventTime(new MicroTime("2006-01-02T15:04:05.000000Z"))
+ // .withNewEventTime(timeFormat)
+ .withFirstTimestamp(dfe.deployCondition.getLastTransitionTime())
+ .withLastTimestamp(dfe.deployCondition.getLastUpdateTime())
+ .withMessage(dfe.getMessage())
+ .withNewMetadata()
+ .withGenerateName(flinkApp.getMetadata().getName())
+ .withNamespace(flinkApp.getMetadata().getNamespace())
+ .endMetadata()
+ .withNewSource()
+ .withComponent(dfe.component)
+ .endSource();
Review comment:
This should be extracted to a utility method that returns `Event`
##########
File path: flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java
##########
@@ -113,29 +118,70 @@ public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
Configuration effectiveConfig =
FlinkUtils.getEffectiveConfig(flinkApp, defaultConfig.getFlinkConfig());
- boolean readyToReconcile = observer.observe(flinkApp, context, effectiveConfig);
- if (!readyToReconcile) {
- return flinkApp.getStatus()
- .getJobManagerDeploymentStatus()
- .toUpdateControl(flinkApp, operatorConfiguration);
- }
-
try {
+ boolean readyToReconcile = observer.observe(flinkApp, context, effectiveConfig);
+ if (!readyToReconcile) {
+ return flinkApp.getStatus()
+ .getJobManagerDeploymentStatus()
+ .toUpdateControl(flinkApp, operatorConfiguration);
+ }
+
UpdateControl<FlinkDeployment> updateControl =
reconcilerFactory
.getOrCreate(flinkApp)
.reconcile(operatorNamespace, flinkApp, context, effectiveConfig);
updateForReconciliationSuccess(flinkApp);
return updateControl;
- } catch (InvalidDeploymentException ide) {
- LOG.error("Reconciliation failed", ide);
- updateForReconciliationError(flinkApp, ide.getMessage());
+ } catch (DeploymentFailedException dfe) {
+ updateForDeploymentFailed(flinkApp, dfe);
return UpdateControl.updateStatus(flinkApp);
} catch (Exception e) {
throw new ReconciliationException(e);
}
}
+ private void updateForDeploymentFailed(
Review comment:
I think the method name should say something about the event being fired. This is not really just a status update as the others.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org