Skip to content

Commit 883f597

Browse files
xiancaoxiancao
committed
Merge branch 'develop' of https://github.com/oracle/weblogic-kubernetes-operator into jrftest3
2 parents aa02bf3 + faec58f commit 883f597

File tree

9 files changed

+157
-113
lines changed

9 files changed

+157
-113
lines changed

‎operator/src/main/java/oracle/kubernetes/operator/PodWatcher.java‎

Lines changed: 27 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -8,10 +8,11 @@
88
import io.kubernetes.client.models.V1ObjectMeta;
99
import io.kubernetes.client.models.V1Pod;
1010
import io.kubernetes.client.util.Watch;
11+
import java.util.ArrayList;
12+
import java.util.Collection;
13+
import java.util.HashMap;
1114
import java.util.List;
1215
import java.util.Map;
13-
import java.util.concurrent.ConcurrentHashMap;
14-
import java.util.concurrent.ConcurrentMap;
1516
import java.util.concurrent.ThreadFactory;
1617
import java.util.concurrent.atomic.AtomicBoolean;
1718
import oracle.kubernetes.operator.TuningParameters.WatchTuning;
@@ -39,8 +40,24 @@ public class PodWatcher extends Watcher<V1Pod>
3940
private final WatchListener<V1Pod> listener;
4041

4142
// Map of Pod name to OnReady
42-
private final ConcurrentMap<String, OnReady> readyCallbackRegistrations =
43-
new ConcurrentHashMap<>();
43+
private final Map<String, Collection<OnReady>> readyCallbackRegistrations = new HashMap<>();
44+
45+
private void registerOnReady(String podName, OnReady onReady) {
46+
synchronized (readyCallbackRegistrations) {
47+
Collection<OnReady> col = readyCallbackRegistrations.get(podName);
48+
if (col == null) {
49+
col = new ArrayList<>();
50+
readyCallbackRegistrations.put(podName, col);
51+
}
52+
col.add(onReady);
53+
}
54+
}
55+
56+
private Collection<OnReady> retrieveOnReady(String podName) {
57+
synchronized (readyCallbackRegistrations) {
58+
return readyCallbackRegistrations.remove(podName);
59+
}
60+
}
4461

4562
/**
4663
* Factory for PodWatcher.
@@ -96,9 +113,11 @@ public void receivedResponse(Watch.Response<V1Pod> item) {
96113
Boolean isReady = !PodHelper.isDeleting(pod) && PodHelper.isReady(pod);
97114
String podName = pod.getMetadata().getName();
98115
if (isReady) {
99-
OnReady ready = readyCallbackRegistrations.remove(podName);
100-
if (ready != null) {
101-
ready.onReady();
116+
Collection<OnReady> col = retrieveOnReady(podName);
117+
if (col != null) {
118+
for (OnReady ready : col) {
119+
ready.onReady();
120+
}
102121
}
103122
}
104123
break;
@@ -148,7 +167,7 @@ public NextAction apply(Packet packet) {
148167
fiber.resume(packet);
149168
}
150169
};
151-
readyCallbackRegistrations.put(metadata.getName(), ready);
170+
registerOnReady(metadata.getName(), ready);
152171

153172
// Timing window -- pod may have come ready before registration for callback
154173
CallBuilderFactory factory =

‎operator/src/main/java/oracle/kubernetes/operator/Watcher.java‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ private void watchForEvents() {
143143
new WatchBuilder()
144144
.withResourceVersion(resourceVersion.toString())
145145
.withTimeoutSeconds(tuning.watchLifetime))) {
146-
while (watch.hasNext()) {
146+
while (hasNext(watch)) {
147147
Watch.Response<T> item = watch.next();
148148

149149
if (isStopping()) {
@@ -164,6 +164,15 @@ private void watchForEvents() {
164164
}
165165
}
166166

167+
private boolean hasNext(WatchI<T> watch) {
168+
try {
169+
return watch.hasNext();
170+
} catch (Throwable ex) {
171+
// no-op on exception during hasNext
172+
}
173+
return false;
174+
}
175+
167176
/**
168177
* Initiates a watch by using the watch builder to request any updates for the specified watcher.
169178
*

‎operator/src/main/java/oracle/kubernetes/operator/calls/AsyncRequestStep.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -190,7 +190,7 @@ public void onSuccess(
190190
try {
191191
cc.cancel();
192192
} finally {
193-
LOGGER.info(
193+
LOGGER.fine(
194194
MessageKeys.ASYNC_TIMEOUT,
195195
requestParams.call,
196196
requestParams.namespace,

‎operator/src/main/java/oracle/kubernetes/operator/helpers/RollingHelper.java‎

Lines changed: 70 additions & 67 deletions
Original file line numberDiff line numberDiff line change
@@ -176,84 +176,87 @@ public String getDetail() {
176176

177177
@Override
178178
public NextAction apply(Packet packet) {
179-
if (it.hasNext()) {
180-
DomainPresenceInfo info = packet.getSPI(DomainPresenceInfo.class);
181-
WlsDomainConfig config = (WlsDomainConfig) packet.get(ProcessingConstants.DOMAIN_TOPOLOGY);
182-
183-
// Refresh as this is constantly changing
184-
Domain dom = info.getDomain();
185-
// These are presently Ready servers
186-
List<String> availableServers = getReadyServers(info);
187-
188-
List<String> servers = new ArrayList<>();
189-
List<String> readyServers = new ArrayList<>();
190-
List<V1Pod> notReadyServers = new ArrayList<>();
191-
192-
Collection<StepAndPacket> serversThatCanRestartNow = new ArrayList<>();
193-
194-
int countReady = 0;
195-
WlsClusterConfig cluster = config != null ? config.getClusterConfig(clusterName) : null;
196-
if (cluster != null) {
197-
List<WlsServerConfig> serversConfigs = cluster.getServerConfigs();
198-
if (serversConfigs != null) {
199-
for (WlsServerConfig s : serversConfigs) {
200-
// figure out how many servers are currently ready
201-
String name = s.getName();
202-
if (availableServers.contains(name)) {
203-
readyServers.add(s.getName());
204-
countReady++;
205-
} else {
206-
V1Pod pod = info.getServerPod(name);
207-
if (pod != null) {
208-
notReadyServers.add(pod);
179+
synchronized (it) {
180+
if (it.hasNext()) {
181+
DomainPresenceInfo info = packet.getSPI(DomainPresenceInfo.class);
182+
WlsDomainConfig config =
183+
(WlsDomainConfig) packet.get(ProcessingConstants.DOMAIN_TOPOLOGY);
184+
185+
// Refresh as this is constantly changing
186+
Domain dom = info.getDomain();
187+
// These are presently Ready servers
188+
List<String> availableServers = getReadyServers(info);
189+
190+
List<String> servers = new ArrayList<>();
191+
List<String> readyServers = new ArrayList<>();
192+
List<V1Pod> notReadyServers = new ArrayList<>();
193+
194+
Collection<StepAndPacket> serversThatCanRestartNow = new ArrayList<>();
195+
196+
int countReady = 0;
197+
WlsClusterConfig cluster = config != null ? config.getClusterConfig(clusterName) : null;
198+
if (cluster != null) {
199+
List<WlsServerConfig> serversConfigs = cluster.getServerConfigs();
200+
if (serversConfigs != null) {
201+
for (WlsServerConfig s : serversConfigs) {
202+
// figure out how many servers are currently ready
203+
String name = s.getName();
204+
if (availableServers.contains(name)) {
205+
readyServers.add(s.getName());
206+
countReady++;
207+
} else {
208+
V1Pod pod = info.getServerPod(name);
209+
if (pod != null) {
210+
notReadyServers.add(pod);
211+
}
209212
}
210213
}
211214
}
212215
}
213-
}
214216

215-
// then add as many as possible next() entries leaving at least minimum cluster
216-
// availability
217-
while (countReady-- > dom.getMinAvailable(clusterName)) {
218-
StepAndPacket current = it.next();
219-
WlsServerConfig serverConfig =
220-
(WlsServerConfig) current.packet.get(ProcessingConstants.SERVER_SCAN);
221-
String serverName = null;
222-
if (serverConfig != null) {
223-
serverName = serverConfig.getName();
224-
} else if (config != null) {
225-
serverName = config.getAdminServerName();
226-
}
227-
if (serverName != null) {
228-
servers.add(serverName);
229-
}
230-
serversThatCanRestartNow.add(current);
231-
if (!it.hasNext()) {
232-
break;
217+
// then add as many as possible next() entries leaving at least minimum cluster
218+
// availability
219+
while (countReady-- > dom.getMinAvailable(clusterName)) {
220+
StepAndPacket current = it.next();
221+
WlsServerConfig serverConfig =
222+
(WlsServerConfig) current.packet.get(ProcessingConstants.SERVER_SCAN);
223+
String serverName = null;
224+
if (serverConfig != null) {
225+
serverName = serverConfig.getName();
226+
} else if (config != null) {
227+
serverName = config.getAdminServerName();
228+
}
229+
if (serverName != null) {
230+
servers.add(serverName);
231+
}
232+
serversThatCanRestartNow.add(current);
233+
if (!it.hasNext()) {
234+
break;
235+
}
233236
}
234-
}
235237

236-
if (serversThatCanRestartNow.isEmpty()) {
237-
// Not enough servers are ready to let us restart a server now
238-
if (!notReadyServers.isEmpty()) {
239-
PodAwaiterStepFactory pw = PodHelper.getPodAwaiterStepFactory(packet);
240-
Collection<StepAndPacket> waitForUnreadyServers = new ArrayList<>();
241-
for (V1Pod pod : notReadyServers) {
242-
waitForUnreadyServers.add(
243-
new StepAndPacket(pw.waitForReady(pod, null), packet.clone()));
244-
}
238+
if (serversThatCanRestartNow.isEmpty()) {
239+
// Not enough servers are ready to let us restart a server now
240+
if (!notReadyServers.isEmpty()) {
241+
PodAwaiterStepFactory pw = PodHelper.getPodAwaiterStepFactory(packet);
242+
Collection<StepAndPacket> waitForUnreadyServers = new ArrayList<>();
243+
for (V1Pod pod : notReadyServers) {
244+
waitForUnreadyServers.add(
245+
new StepAndPacket(pw.waitForReady(pod, null), packet.clone()));
246+
}
245247

246-
// Wait for at least one of the not-yet-ready servers to become ready
247-
return doForkAtLeastOne(this, packet, waitForUnreadyServers);
248-
} else {
249-
throw new IllegalStateException();
248+
// Wait for at least one of the not-yet-ready servers to become ready
249+
return doForkAtLeastOne(this, packet, waitForUnreadyServers);
250+
} else {
251+
throw new IllegalStateException();
252+
}
250253
}
251-
}
252254

253-
readyServers.removeAll(servers);
254-
LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
255+
readyServers.removeAll(servers);
256+
LOGGER.info(MessageKeys.ROLLING_SERVERS, dom.getDomainUID(), servers, readyServers);
255257

256-
return doNext(new ServersThatCanRestartNowStep(serversThatCanRestartNow, this), packet);
258+
return doNext(new ServersThatCanRestartNowStep(serversThatCanRestartNow, this), packet);
259+
}
257260
}
258261

259262
return doNext(packet);

‎operator/src/main/java/oracle/kubernetes/operator/work/Fiber.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -148,12 +148,12 @@ public void start(Step stepline, Packet packet, CompletionCallback completionCal
148148
this.na.invoke(stepline, packet);
149149
this.completionCallback = completionCallback;
150150

151-
if (LOGGER.isFineEnabled()) {
152-
breadCrumbs = new ArrayList<>();
153-
LOGGER.fine("{0} started", new Object[] {getName()});
154-
}
155-
156151
if (status.get() == NOT_COMPLETE) {
152+
if (LOGGER.isFineEnabled()) {
153+
breadCrumbs = new ArrayList<>();
154+
LOGGER.fine("{0} started", new Object[] {getName()});
155+
}
156+
157157
owner.addRunnable(this);
158158
}
159159
}
@@ -262,7 +262,12 @@ public Fiber createChildFiber() {
262262
children = new ArrayList<>();
263263
}
264264
children.add(child);
265-
addBreadCrumb(child);
265+
if (status.get() == NOT_COMPLETE) {
266+
addBreadCrumb(child);
267+
} else {
268+
// Race condition where child is created after parent is cancelled or done
269+
child.status.set(CANCELLED);
270+
}
266271
}
267272

268273
return child;

‎operator/src/main/resources/scripts/startServer.sh‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ trace "Starting WebLogic Server '${SERVER_NAME}'."
2525
function exitOrLoop {
2626
if [ -f /weblogic-operator/debug/livenessProbeSuccessOverride ]
2727
then
28-
while true ; do sleep 60 ; done
28+
waitForShutdownMarker
2929
else
3030
exit 1
3131
fi
@@ -100,23 +100,21 @@ function waitUntilShutdown() {
100100
#
101101
if [ "${SERVER_OUT_IN_POD_LOG}" == 'true' ] ; then
102102
trace "Showing the server out file from ${SERVER_OUT_FILE}"
103-
tail -F -n +0 ${SERVER_OUT_FILE} || exitOrLoop
104-
else
105-
trace "Wait indefinitely so that the Kubernetes pod does not exit and try to restart"
106-
while true; do sleep 60; done
103+
${SCRIPTPATH}/tailLog.sh ${SERVER_OUT_FILE} &
107104
fi
105+
waitForShutdownMarker
108106
}
109107

110-
function mockWaitUntilShutdown() {
108+
function waitForShutdownMarker() {
111109
#
112110
# Wait forever. Kubernetes will monitor this pod via liveness and readyness probes.
113111
#
114112
trace "Wait indefinitely so that the Kubernetes pod does not exit and try to restart"
115113
while true; do
116-
if [ -e /u01/doShutdown ] ; then
114+
if [ -e /weblogic-operator/doShutdown ] ; then
117115
exit 0
118116
fi
119-
sleep 5
117+
sleep 3
120118
done
121119
}
122120

@@ -219,7 +217,7 @@ copySitCfg /weblogic-operator/introspector ${DOMAIN_HOME}/optconfig/diagnostics
219217

220218
if [ "${MOCK_WLS}" == 'true' ]; then
221219
mockWLS
222-
mockWaitUntilShutdown
220+
waitForShutdownMarker
223221
else
224222
startWLS
225223
waitUntilShutdown

0 commit comments

Comments
 (0)