Skip to content

Commit 09d9c5f

Browse files
csviricsviri
committed
wip
1 parent 2eafe34 commit 09d9c5f

File tree

4 files changed

+95
-44
lines changed

4 files changed

+95
-44
lines changed

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ControllerManager.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,8 @@
1212
import io.javaoperatorsdk.operator.processing.LifecycleAware;
1313

1414
/**
15-
* Not confuse with controller manager form go operators. The high level aggregate is Operator in
16-
* JOSDK.
15+
* Not confuse with controller manager form go operators. The highest level aggregate is
16+
* {@link Operator} in JOSDK.
1717
*/
1818
class ControllerManager implements LifecycleAware {
1919

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/LeaderElectionManager.java

+61-1
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,66 @@
11
package io.javaoperatorsdk.operator;
22

3+
import java.util.UUID;
4+
5+
import org.slf4j.Logger;
6+
import org.slf4j.LoggerFactory;
7+
8+
import io.fabric8.kubernetes.client.KubernetesClient;
9+
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
10+
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderCallbacks;
11+
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
12+
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
13+
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectorBuilder;
14+
import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
15+
import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.Lock;
16+
import io.javaoperatorsdk.operator.api.config.LeaderElectionConfiguration;
17+
318
public class LeaderElectionManager {
419

5-
public LeaderElectionManager() {}
20+
private static final Logger log = LoggerFactory.getLogger(LeaderElectionManager.class);
21+
22+
private LeaderElector<NamespacedKubernetesClient> leaderElector = null;
23+
private ControllerManager controllerManager;
24+
25+
public LeaderElectionManager(ControllerManager controllerManager) {
26+
this.controllerManager = controllerManager;
27+
}
28+
29+
public void init(LeaderElectionConfiguration config, KubernetesClient client) {
30+
Lock lock = new LeaseLock(config.getLeaseNamespace(), config.getLeaseName(), identity(config));
31+
// the releaseOnCancel does not seem to be used anywhere
32+
leaderElector = new LeaderElectorBuilder<>((NamespacedKubernetesClient) client)
33+
.withConfig(
34+
new LeaderElectionConfig(lock, config.getLeaseDuration(), config.getRenewDeadline(),
35+
config.getRetryPeriod(), leaderCallbacks(), true, config.getLeaseName()))
36+
.build();
37+
}
38+
39+
40+
public boolean isLeaderElectionOn() {
41+
return leaderElector != null;
42+
}
43+
44+
private LeaderCallbacks leaderCallbacks() {
45+
return new LeaderCallbacks(this::startLeading, this::stopLeading, leader -> {
46+
log.info("New leader with identity: {}", leader);
47+
});
48+
}
49+
50+
private void startLeading() {
51+
52+
}
53+
54+
private void stopLeading() {
55+
56+
}
57+
58+
private String identity(LeaderElectionConfiguration config) {
59+
String identity = config.getIdentity().orElse(System.getenv("HOSTNAME"));
60+
if (identity == null || identity.isBlank()) {
61+
identity = UUID.randomUUID().toString();
62+
}
63+
return identity;
64+
}
65+
666
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/Operator.java

+14-38
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,7 @@
1111
import io.fabric8.kubernetes.api.model.HasMetadata;
1212
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
1313
import io.fabric8.kubernetes.client.KubernetesClient;
14-
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
1514
import io.fabric8.kubernetes.client.Version;
16-
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectionConfig;
17-
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElector;
18-
import io.fabric8.kubernetes.client.extended.leaderelection.LeaderElectorBuilder;
19-
import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock;
20-
import io.fabric8.kubernetes.client.extended.leaderelection.resourcelock.Lock;
2115
import io.javaoperatorsdk.operator.api.config.*;
2216
import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
2317
import io.javaoperatorsdk.operator.processing.Controller;
@@ -27,8 +21,9 @@
2721
public class Operator implements LifecycleAware {
2822
private static final Logger log = LoggerFactory.getLogger(Operator.class);
2923
private final KubernetesClient kubernetesClient;
30-
private final ControllerManager controllers = new ControllerManager();
31-
private LeaderElector leaderElector;
24+
private final ControllerManager controllerManager = new ControllerManager();
25+
private final LeaderElectionManager leaderElectionManager =
26+
new LeaderElectionManager(controllerManager);
3227

3328
public Operator() {
3429
this(new DefaultKubernetesClient(), ConfigurationServiceProvider.instance());
@@ -53,7 +48,8 @@ public Operator(Consumer<ConfigurationServiceOverrider> overrider) {
5348
public Operator(KubernetesClient client, Consumer<ConfigurationServiceOverrider> overrider) {
5449
this.kubernetesClient = client;
5550
ConfigurationServiceProvider.overrideCurrent(overrider);
56-
51+
ConfigurationServiceProvider.instance().getLeaderElectionConfiguration()
52+
.ifPresent(c -> leaderElectionManager.init(c, kubernetesClient));
5753
}
5854

5955
/**
@@ -66,6 +62,8 @@ public Operator(KubernetesClient client, Consumer<ConfigurationServiceOverrider>
6662
public Operator(KubernetesClient kubernetesClient, ConfigurationService configurationService) {
6763
this.kubernetesClient = kubernetesClient;
6864
ConfigurationServiceProvider.set(configurationService);
65+
configurationService.getLeaderElectionConfiguration()
66+
.ifPresent(c -> leaderElectionManager.init(c, kubernetesClient));
6967
}
7068

7169
/** Adds a shutdown hook that automatically calls {@link #stop()} when the app shuts down. */
@@ -85,7 +83,7 @@ public KubernetesClient getKubernetesClient() {
8583
public void start() {
8684
try {
8785

88-
controllers.shouldStart();
86+
controllerManager.shouldStart();
8987

9088
final var version = ConfigurationServiceProvider.instance().getVersion();
9189
log.info(
@@ -98,7 +96,7 @@ public void start() {
9896
log.info("Client version: {}", clientVersion);
9997

10098
ExecutorServiceManager.init();
101-
controllers.start();
99+
controllerManager.start();
102100
} catch (Exception e) {
103101
log.error("Error starting operator", e);
104102
stop();
@@ -112,7 +110,7 @@ public void stop() throws OperatorException {
112110
log.info(
113111
"Operator SDK {} is shutting down...", configurationService.getVersion().getSdkVersion());
114112

115-
controllers.stop();
113+
controllerManager.stop();
116114

117115
ExecutorServiceManager.stop();
118116
if (configurationService.closeClientOnStop()) {
@@ -162,7 +160,7 @@ public <P extends HasMetadata> RegisteredController<P> register(Reconciler<P> re
162160

163161
final var controller = new Controller<>(reconciler, configuration, kubernetesClient);
164162

165-
controllers.add(controller);
163+
controllerManager.add(controller);
166164

167165
final var watchedNS = configuration.watchAllNamespaces() ? "[all namespaces]"
168166
: configuration.getEffectiveNamespaces();
@@ -192,37 +190,15 @@ public <P extends HasMetadata> RegisteredController<P> register(Reconciler<P> re
192190
}
193191

194192
public Optional<RegisteredController> getRegisteredController(String name) {
195-
return controllers.get(name).map(RegisteredController.class::cast);
193+
return controllerManager.get(name).map(RegisteredController.class::cast);
196194
}
197195

198196
public Set<RegisteredController> getRegisteredControllers() {
199-
return new HashSet<>(controllers.controllers());
197+
return new HashSet<>(controllerManager.controllers());
200198
}
201199

202200
public int getRegisteredControllersNumber() {
203-
return controllers.size();
204-
}
205-
206-
private void initLeaderElector(KubernetesClient client) {
207-
208-
var leaderElectionConfig =
209-
ConfigurationServiceProvider.instance().getLeaderElectionConfiguration();
210-
if (leaderElectionConfig.isEmpty()) {
211-
return;
212-
}
213-
var conf = leaderElectionConfig.get();
214-
215-
// todo discuss openshift client not a NamespacedKubernetesClient?
216-
// name of the pod
217-
// todo configurable
218-
String identity = System.getenv("HOSTNAME");
219-
220-
Lock lock = new LeaseLock(conf.getLeaseNamespace(), conf.getLeaseName(), identity);
221-
// todo check release on cancel
222-
leaderElector = new LeaderElectorBuilder<>((NamespacedKubernetesClient) client)
223-
.withConfig(new LeaderElectionConfig(lock, conf.getLeaseDuration(), conf.getRenewDeadline(),
224-
conf.getRetryPeriod(), null, true, conf.getLeaseName()))
225-
.build();
201+
return controllerManager.size();
226202
}
227203

228204
}

‎operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/LeaderElectionConfiguration.java

+18-3
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,8 @@
11
package io.javaoperatorsdk.operator.api.config;
22

33
import java.time.Duration;
4+
import java.util.Optional;
45

5-
// todo discuss leader election with lease vs for life, other options:
6-
// see: https://pkg.go.dev/sigs.k8s.io/controller-runtime/pkg/manager#Options
76
public class LeaderElectionConfiguration {
87

98
public static final Duration LEASE_DURATION_DEFAULT_VALUE = Duration.ofSeconds(15);
@@ -15,6 +14,7 @@ public class LeaderElectionConfiguration {
1514

1615
private final String leaseName;
1716
private final String leaseNamespace;
17+
private final String identity;
1818

1919
private final Duration leaseDuration;
2020
private final Duration renewDeadline;
@@ -26,7 +26,7 @@ public LeaderElectionConfiguration(String leaseName, String leaseNamespace) {
2626
leaseNamespace,
2727
LEASE_DURATION_DEFAULT_VALUE,
2828
RENEW_DEADLINE_DEFAULT_VALUE,
29-
RETRY_PERIOD_DEFAULT_VALUE);
29+
RETRY_PERIOD_DEFAULT_VALUE, null);
3030
}
3131

3232
public LeaderElectionConfiguration(
@@ -35,11 +35,22 @@ public LeaderElectionConfiguration(
3535
Duration leaseDuration,
3636
Duration renewDeadline,
3737
Duration retryPeriod) {
38+
this(leaseName, leaseNamespace, leaseDuration, renewDeadline, retryPeriod, null);
39+
}
40+
41+
public LeaderElectionConfiguration(
42+
String leaseName,
43+
String leaseNamespace,
44+
Duration leaseDuration,
45+
Duration renewDeadline,
46+
Duration retryPeriod,
47+
String identity) {
3848
this.leaseName = leaseName;
3949
this.leaseNamespace = leaseNamespace;
4050
this.leaseDuration = leaseDuration;
4151
this.renewDeadline = renewDeadline;
4252
this.retryPeriod = retryPeriod;
53+
this.identity = identity;
4354
}
4455

4556
public String getLeaseNamespace() {
@@ -61,4 +72,8 @@ public Duration getRenewDeadline() {
6172
public Duration getRetryPeriod() {
6273
return retryPeriod;
6374
}
75+
76+
public Optional<String> getIdentity() {
77+
return Optional.ofNullable(identity);
78+
}
6479
}

0 commit comments

Comments
 (0)