morhidi commented on code in PR #268:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/268#discussion_r896612072
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java:
##########
@@ -81,59 +80,121 @@ public FlinkDeploymentController(
ReconcilerFactory reconcilerFactory,
ObserverFactory observerFactory,
MetricManager<FlinkDeployment> metricManager,
+ Metrics metrics,
StatusHelper<FlinkDeploymentStatus> statusHelper) {
this.configManager = configManager;
this.kubernetesClient = kubernetesClient;
this.validators = validators;
this.reconcilerFactory = reconcilerFactory;
this.observerFactory = observerFactory;
this.metricManager = metricManager;
+ this.metrics = metrics;
this.statusHelper = statusHelper;
}
@Override
public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
- LOG.info("Deleting FlinkDeployment");
- statusHelper.updateStatusFromCache(flinkApp);
try {
- observerFactory.getOrCreate(flinkApp).observe(flinkApp, context);
- } catch (DeploymentFailedException dfe) {
- // ignore during cleanup
+ return metrics.timeControllerExecution(
+ new Metrics.ControllerExecution<>() {
+ @Override
+ public String name() {
+ return "cleanup";
+ }
+
+ @Override
+ public String controllerName() {
+ return
FlinkDeploymentController.class.getSimpleName();
+ }
+
+ @Override
+ public String successTypeName(DeleteControl
deleteControl) {
+ return deleteControl.isRemoveFinalizer()
+ ? "delete"
+ : "finalizerNotRemoved";
+ }
+
+ @Override
+ public DeleteControl execute() {
+ LOG.info("Deleting FlinkDeployment");
+ statusHelper.updateStatusFromCache(flinkApp);
+ try {
+
observerFactory.getOrCreate(flinkApp).observe(flinkApp, context);
+ } catch (DeploymentFailedException dfe) {
+ // ignore during cleanup
+ }
+ metricManager.onRemove(flinkApp);
+ statusHelper.removeCachedStatus(flinkApp);
+ return reconcilerFactory
+ .getOrCreate(flinkApp)
+ .cleanup(flinkApp, context);
+ }
+ });
+ } catch (Exception e) {
+ throw new OperatorException(e);
}
- metricManager.onRemove(flinkApp);
- statusHelper.removeCachedStatus(flinkApp);
- return reconcilerFactory.getOrCreate(flinkApp).cleanup(flinkApp,
context);
}
@Override
public UpdateControl<FlinkDeployment> reconcile(FlinkDeployment flinkApp,
Context context)
Review Comment:
ditto:
https://github.com/java-operator-sdk/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java#L151
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java:
##########
@@ -70,37 +70,36 @@ public class FlinkOperator {
private final FlinkConfigManager configManager;
private final Set<FlinkResourceValidator> validators;
private final MetricGroup metricGroup;
+ private final Metrics metrics;
+
+ private static final String OPERATOR_SDK_GROUP = "operator.sdk";
public FlinkOperator(@Nullable Configuration conf) {
this.client = new DefaultKubernetesClient();
this.configManager = conf != null ? new FlinkConfigManager(conf) : new
FlinkConfigManager();
- this.operator =
- new Operator(
- client,
- getConfigurationServiceOverriderConsumer(
- configManager.getOperatorConfiguration()));
- this.flinkService = new FlinkService(client, configManager);
- this.validators = ValidatorUtils.discoverValidators(configManager);
this.metricGroup =
OperatorMetricUtils.initOperatorMetrics(configManager.getDefaultConfig());
+ this.metrics = new
FlinkOperatorMetrics(metricGroup.addGroup(OPERATOR_SDK_GROUP));
+ this.operator = new Operator(client,
getConfigurationServiceOverriderConsumer());
+ this.flinkService = new FlinkService(client, configManager);
+ this.validators = ValidatorUtils.discoverValidators(configManager);
PluginManager pluginManager =
PluginUtils.createPluginManagerFromRootFolder(configManager.getDefaultConfig());
FileSystem.initialize(configManager.getDefaultConfig(), pluginManager);
}
- @VisibleForTesting
- protected static Consumer<ConfigurationServiceOverrider>
- getConfigurationServiceOverriderConsumer(
- FlinkOperatorConfiguration operatorConfiguration) {
+ private Consumer<ConfigurationServiceOverrider>
getConfigurationServiceOverriderConsumer() {
return overrider -> {
- int parallelism =
operatorConfiguration.getReconcilerMaxParallelism();
+ int parallelism =
+
configManager.getOperatorConfiguration().getReconcilerMaxParallelism();
if (parallelism == -1) {
LOG.info("Configuring operator with unbounded reconciliation
thread pool.");
overrider.withExecutorService(Executors.newCachedThreadPool());
} else {
LOG.info("Configuring operator with {} reconciliation
threads.", parallelism);
overrider.withConcurrentReconciliationThreads(parallelism);
}
+ overrider.withMetrics(metrics);
Review Comment:
I only expected that we would need to implement the Metrics interface and
register it here, and the rest is handle by JOSDK framework.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkDeploymentController.java:
##########
@@ -81,59 +80,121 @@ public FlinkDeploymentController(
ReconcilerFactory reconcilerFactory,
ObserverFactory observerFactory,
MetricManager<FlinkDeployment> metricManager,
+ Metrics metrics,
StatusHelper<FlinkDeploymentStatus> statusHelper) {
this.configManager = configManager;
this.kubernetesClient = kubernetesClient;
this.validators = validators;
this.reconcilerFactory = reconcilerFactory;
this.observerFactory = observerFactory;
this.metricManager = metricManager;
+ this.metrics = metrics;
this.statusHelper = statusHelper;
}
@Override
public DeleteControl cleanup(FlinkDeployment flinkApp, Context context) {
Review Comment:
I don't really understand this part. This must be handled at JOSDK level,
isn't it?
https://github.com/java-operator-sdk/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/Controller.java#L109
--
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]
