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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
