New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
leaderelection sanity e2e integration tests #1463
Merged
k8s-ci-robot
merged 1 commit into
kubernetes-client:master
from
himanshug:leader_shutdown
Jan 3, 2021
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
337 changes: 337 additions & 0 deletions
337
e2e/src/test/java/io/kubernetes/client/e2e/extended/leaderelection/LeaderElectorTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,337 @@ | ||
/* | ||
Copyright 2020 The Kubernetes Authors. | ||
Licensed 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 io.kubernetes.client.e2e.extended.leaderelection; | ||
|
||
import io.kubernetes.client.extended.leaderelection.LeaderElectionConfig; | ||
import io.kubernetes.client.extended.leaderelection.LeaderElector; | ||
import io.kubernetes.client.extended.leaderelection.Lock; | ||
import io.kubernetes.client.extended.leaderelection.resourcelock.ConfigMapLock; | ||
import io.kubernetes.client.extended.leaderelection.resourcelock.EndpointsLock; | ||
import io.kubernetes.client.extended.leaderelection.resourcelock.LeaseLock; | ||
import io.kubernetes.client.openapi.ApiClient; | ||
import io.kubernetes.client.openapi.ApiException; | ||
import io.kubernetes.client.openapi.apis.CoordinationV1Api; | ||
import io.kubernetes.client.openapi.apis.CoreV1Api; | ||
import io.kubernetes.client.util.ClientBuilder; | ||
import java.io.IOException; | ||
import java.net.HttpURLConnection; | ||
import java.time.Duration; | ||
import java.util.ArrayList; | ||
import java.util.Collection; | ||
import java.util.List; | ||
import java.util.concurrent.BrokenBarrierException; | ||
import java.util.concurrent.CountDownLatch; | ||
import java.util.concurrent.CyclicBarrier; | ||
import java.util.concurrent.atomic.AtomicInteger; | ||
import java.util.concurrent.atomic.AtomicReference; | ||
import org.joda.time.format.*; | ||
import org.junit.Assert; | ||
import org.junit.Before; | ||
import org.junit.Test; | ||
import org.junit.runner.RunWith; | ||
import org.junit.runners.Parameterized; | ||
import org.slf4j.Logger; | ||
import org.slf4j.LoggerFactory; | ||
|
||
@RunWith(Parameterized.class) | ||
public class LeaderElectorTest { | ||
|
||
private static final Logger LOGGER = LoggerFactory.getLogger(LeaderElectorTest.class); | ||
|
||
private static final String LOCK_RESOURCE_NAME = "leader-election-it"; | ||
private static final String NAMESPACE = "default"; | ||
|
||
private enum LockType { | ||
ConfigMap, | ||
Endpoints, | ||
Lease | ||
} | ||
|
||
@Parameterized.Parameters(name = "{0}") | ||
public static Collection<Object[]> constructorFeeder() { | ||
final List<Object[]> args = new ArrayList<>(); | ||
|
||
args.add(new Object[] {LockType.ConfigMap}); | ||
args.add(new Object[] {LockType.Endpoints}); | ||
args.add(new Object[] {LockType.Lease}); | ||
|
||
return args; | ||
} | ||
|
||
private final ApiClient apiClient; | ||
private final LockType lockType; | ||
|
||
public LeaderElectorTest(LockType lockType) { | ||
try { | ||
apiClient = ClientBuilder.defaultClient(); | ||
} catch (IOException ex) { | ||
throw new RuntimeException("Couldn't create ApiClient", ex); | ||
} | ||
this.lockType = lockType; | ||
|
||
// Lease resource requires special care with DateTime | ||
if (lockType == LockType.Lease) { | ||
// TODO: switch date-time library so that micro-sec timestamp can be serialized in RFC3339 | ||
// format w/ correct precision without the hacks | ||
|
||
// This formatter is used for Lease resource spec's acquire/renewTime | ||
DateTimeFormatter isoWithFractionalMicroSecsFormatter = | ||
DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'"); | ||
|
||
DateTimeFormatter formatter = | ||
new DateTimeFormatterBuilder() | ||
.append( | ||
isoWithFractionalMicroSecsFormatter.getPrinter(), | ||
new DateTimeParser[] { | ||
isoWithFractionalMicroSecsFormatter.getParser(), | ||
// need this one to parse "creationTimestamp" format e.g. "2020-12-30T09:29:13Z" | ||
// in Lease resource returned from server | ||
ISODateTimeFormat.dateOptionalTimeParser().getParser(), | ||
}) | ||
.toFormatter(); | ||
|
||
apiClient.setDateTimeFormat(formatter); | ||
} | ||
} | ||
|
||
@Before | ||
public void setup() throws Exception { | ||
// delete the lock resource if it exists, or else first leader candidate might need to wait for | ||
// a whole | ||
// leaseDuration configured | ||
switch (lockType) { | ||
case ConfigMap: | ||
deleteConfigMapLockResource(); | ||
break; | ||
case Endpoints: | ||
deleteEndpointsLockResource(); | ||
break; | ||
case Lease: | ||
deleteLeaseLockResource(); | ||
break; | ||
default: | ||
throw new RuntimeException("Unknown LockType " + lockType); | ||
} | ||
} | ||
|
||
@Test(timeout = 30000L) | ||
public void testSingleCandidateLeaderElection() throws Exception { | ||
CountDownLatch startLeadershipLatch = new CountDownLatch(1); | ||
CountDownLatch stopLeadershipLatch = new CountDownLatch(1); | ||
|
||
LeaderElector leaderElector = | ||
makeAndRunLeaderElectorAsync( | ||
"candidate", | ||
null, | ||
() -> startLeadershipLatch.countDown(), | ||
() -> stopLeadershipLatch.countDown(), | ||
apiClient); | ||
|
||
startLeadershipLatch.await(); | ||
|
||
leaderElector.close(); | ||
|
||
stopLeadershipLatch.await(); | ||
} | ||
|
||
@Test(timeout = 30000L) | ||
public void testMultiCandidateLeaderElection() throws Exception { | ||
CyclicBarrier startBarrier = new CyclicBarrier(2); | ||
|
||
CountDownLatch startBeingLeader = new CountDownLatch(1); | ||
CountDownLatch stopBeingLeader = new CountDownLatch(1); | ||
|
||
AtomicInteger startBeingLeaderCount = new AtomicInteger(); | ||
AtomicInteger stopBeingLeaderCount = new AtomicInteger(); | ||
|
||
AtomicReference<String> leaderRef = new AtomicReference<>(); | ||
|
||
String candidate1 = "candidate1"; | ||
String candidate2 = "candidate2"; | ||
|
||
LeaderElector leaderElector1 = | ||
makeAndRunLeaderElectorAsync( | ||
candidate1, | ||
startBarrier, | ||
() -> { | ||
startBeingLeaderCount.incrementAndGet(); | ||
leaderRef.set(candidate1); | ||
startBeingLeader.countDown(); | ||
}, | ||
() -> { | ||
stopBeingLeaderCount.incrementAndGet(); | ||
stopBeingLeader.countDown(); | ||
}, | ||
apiClient); | ||
|
||
LeaderElector leaderElector2 = | ||
makeAndRunLeaderElectorAsync( | ||
candidate2, | ||
startBarrier, | ||
() -> { | ||
startBeingLeaderCount.incrementAndGet(); | ||
leaderRef.set(candidate2); | ||
startBeingLeader.countDown(); | ||
}, | ||
() -> { | ||
stopBeingLeaderCount.incrementAndGet(); | ||
stopBeingLeader.countDown(); | ||
}, | ||
apiClient); | ||
|
||
// wait till someone becomes leader | ||
startBeingLeader.await(); | ||
Assert.assertNotNull(leaderRef.get()); | ||
Assert.assertTrue(candidate1.equals(leaderRef.get()) || candidate2.equals(leaderRef.get())); | ||
|
||
// stop both LeaderElectors, in order .. non-leader, then leader so that non-leader doesn't get | ||
// to become leader | ||
if (candidate1.equals(leaderRef.get())) { | ||
leaderElector2.close(); | ||
leaderElector1.close(); | ||
} else { | ||
leaderElector1.close(); | ||
leaderElector2.close(); | ||
} | ||
|
||
stopBeingLeader.await(); | ||
|
||
// make sure that only one candidate became leader | ||
Assert.assertEquals(1, startBeingLeaderCount.get()); | ||
Assert.assertEquals(1, stopBeingLeaderCount.get()); | ||
} | ||
|
||
@Test(timeout = 30000L) | ||
public void testLeaderGracefulShutdown() throws Exception { | ||
CountDownLatch startBeingLeader1 = new CountDownLatch(1); | ||
CountDownLatch stopBeingLeader1 = new CountDownLatch(1); | ||
|
||
LeaderElector leaderElector1 = | ||
makeAndRunLeaderElectorAsync( | ||
"candidate1", | ||
null, | ||
() -> startBeingLeader1.countDown(), | ||
() -> stopBeingLeader1.countDown(), | ||
apiClient); | ||
|
||
// wait for candidate1 to become leader | ||
startBeingLeader1.await(); | ||
|
||
CountDownLatch startBeingLeader2 = new CountDownLatch(1); | ||
CountDownLatch stopBeingLeader2 = new CountDownLatch(1); | ||
|
||
LeaderElector leaderElector2 = | ||
makeAndRunLeaderElectorAsync( | ||
"candidate2", | ||
null, | ||
() -> startBeingLeader2.countDown(), | ||
() -> stopBeingLeader2.countDown(), | ||
apiClient); | ||
|
||
leaderElector1.close(); | ||
|
||
// ensure stopBeingLeader hook is called | ||
stopBeingLeader1.await(); | ||
|
||
// wait for candidate2 to become leader | ||
startBeingLeader2.await(); | ||
|
||
leaderElector2.close(); | ||
} | ||
|
||
private LeaderElector makeAndRunLeaderElectorAsync( | ||
String candidateId, | ||
CyclicBarrier startBarrier, | ||
Runnable startBeingLeader, | ||
Runnable stopBeingLeader, | ||
ApiClient apiClient) { | ||
|
||
Lock lock = makeLock(candidateId, NAMESPACE, LOCK_RESOURCE_NAME, apiClient); | ||
|
||
LeaderElectionConfig leaderElectionConfig = | ||
new LeaderElectionConfig( | ||
lock, Duration.ofSeconds(30), Duration.ofSeconds(23), Duration.ofSeconds(3)); | ||
LeaderElector leaderElector = new LeaderElector(leaderElectionConfig); | ||
|
||
Thread thread = | ||
new Thread( | ||
() -> { | ||
if (startBarrier != null) { | ||
try { | ||
startBarrier.await(); | ||
} catch (InterruptedException | BrokenBarrierException ex) { | ||
LOGGER.error("startBarrier.await() failed", ex); | ||
return; | ||
} | ||
} | ||
|
||
leaderElector.run(startBeingLeader, stopBeingLeader); | ||
}, | ||
String.format("%s-leader-elector-main", candidateId)); | ||
thread.setDaemon(true); | ||
thread.start(); | ||
|
||
return leaderElector; | ||
} | ||
|
||
private Lock makeLock( | ||
String candidateId, String namespace, String lockResourceName, ApiClient k8sApiClient) { | ||
|
||
switch (lockType) { | ||
case ConfigMap: | ||
return new ConfigMapLock(namespace, lockResourceName, candidateId, k8sApiClient); | ||
case Endpoints: | ||
return new EndpointsLock(namespace, lockResourceName, candidateId, k8sApiClient); | ||
case Lease: | ||
return new LeaseLock(namespace, lockResourceName, candidateId, k8sApiClient); | ||
default: | ||
throw new RuntimeException("Unknown LockType " + lockType); | ||
} | ||
} | ||
|
||
private void deleteConfigMapLockResource() throws Exception { | ||
try { | ||
CoreV1Api coreV1Api = new CoreV1Api(apiClient); | ||
coreV1Api.deleteNamespacedConfigMap( | ||
LOCK_RESOURCE_NAME, NAMESPACE, null, null, null, null, null, null); | ||
} catch (ApiException ex) { | ||
if (ex.getCode() != HttpURLConnection.HTTP_NOT_FOUND) { | ||
throw ex; | ||
} | ||
} | ||
} | ||
|
||
private void deleteEndpointsLockResource() throws Exception { | ||
try { | ||
CoreV1Api coreV1Api = new CoreV1Api(apiClient); | ||
coreV1Api.deleteNamespacedEndpoints( | ||
LOCK_RESOURCE_NAME, NAMESPACE, null, null, null, null, null, null); | ||
} catch (ApiException ex) { | ||
if (ex.getCode() != HttpURLConnection.HTTP_NOT_FOUND) { | ||
throw ex; | ||
} | ||
} | ||
} | ||
|
||
private void deleteLeaseLockResource() throws Exception { | ||
try { | ||
CoordinationV1Api coordinationV1Api = new CoordinationV1Api(apiClient); | ||
coordinationV1Api.deleteNamespacedLease( | ||
LOCK_RESOURCE_NAME, NAMESPACE, null, null, null, null, null, null); | ||
} catch (ApiException ex) { | ||
if (ex.getCode() != HttpURLConnection.HTTP_NOT_FOUND) { | ||
throw ex; | ||
} | ||
} | ||
} | ||
} |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
dang, the original kubernetes type requires a RFC3339 micro-sec timestamp here however joda-time wont support such precision anymore JodaOrg/joda-time#139. i think we should consider getting #1418 merged so that serialization for micro-sec timestamp can work.
@himanshug can you add a TODO comment here saying:
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
done... yeah
Lease
resource creation etc should just work out of the box :)There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
that said, unrelated to this PR but, "serialization" code by default can either emit to ISO8601 or RFC3339 ... if k8s can accept RFC3339 for all the date-time fields in all the resources then we can switch the default serialization format in
JSON.java
to be RFC3339 or elseLease
object serialization would need to be handled as a special case even after the switch somewhere.