Skip to content

Commit 9358043

Browse files
rjeberhardrjeberhard
committed
Reconcile watch and recheck with startup
1 parent 5e77fe4 commit 9358043

14 files changed

+258
-158
lines changed

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

Lines changed: 18 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -92,19 +92,21 @@ static void dispatchPodWatch(Watch.Response<V1Pod> item) {
9292
p,
9393
(current, ob) -> {
9494
if (current != null) {
95-
// If the skoPod is null then the operator deleted this pod
96-
// and modifications are to the terminating pod
9795
if (!isOutdated(current.getMetadata(), metadata)) {
9896
return ob;
9997
}
10098
}
99+
// If the skoPod is null then the operator deleted this pod
100+
// and modifications are to the terminating pod
101101
return current;
102102
});
103103
break;
104104
case "DELETED":
105-
info = DomainPresenceInfoManager.lookup(domainUID);
105+
info = DomainPresenceInfoManager.lookup(metadata.getNamespace(), domainUID);
106106
if (info != null) {
107-
sko = ServerKubernetesObjectsManager.lookup(metadata.getName());
107+
sko =
108+
ServerKubernetesObjectsManager.lookup(
109+
metadata.getNamespace(), metadata.getName());
108110
if (sko != null) {
109111
V1Pod oldPod =
110112
sko.getPod()
@@ -236,7 +238,7 @@ static void dispatchServiceWatch(Watch.Response<V1Service> item) {
236238
}
237239
break;
238240
case "DELETED":
239-
info = DomainPresenceInfoManager.lookup(domainUID);
241+
info = DomainPresenceInfoManager.lookup(metadata.getNamespace(), domainUID);
240242
if (info != null) {
241243
if (clusterName != null) {
242244
boolean removed =
@@ -256,7 +258,9 @@ static void dispatchServiceWatch(Watch.Response<V1Service> item) {
256258
makeRightDomainPresence(info, domainUID, info.getDomain(), true, false, true);
257259
}
258260
} else if (serverName != null) {
259-
sko = ServerKubernetesObjectsManager.lookup(metadata.getName());
261+
sko =
262+
ServerKubernetesObjectsManager.lookup(
263+
metadata.getNamespace(), metadata.getName());
260264
if (sko != null) {
261265
if (channelName != null) {
262266
boolean removed =
@@ -349,7 +353,7 @@ static void dispatchIngressWatch(Watch.Response<V1beta1Ingress> item) {
349353
});
350354
break;
351355
case "DELETED":
352-
info = DomainPresenceInfoManager.lookup(domainUID);
356+
info = DomainPresenceInfoManager.lookup(metadata.getNamespace(), domainUID);
353357
if (info != null) {
354358
boolean removed =
355359
removeIfPresentAnd(
@@ -429,7 +433,8 @@ private static void onEvent(V1Event event) {
429433
String message = event.getMessage();
430434
if (message != null) {
431435
if (message.contains(WebLogicConstants.READINESS_PROBE_NOT_READY_STATE)) {
432-
ServerKubernetesObjects sko = ServerKubernetesObjectsManager.lookup(name);
436+
ServerKubernetesObjects sko =
437+
ServerKubernetesObjectsManager.lookup(event.getMetadata().getNamespace(), name);
433438
if (sko != null) {
434439
int idx = message.lastIndexOf(':');
435440
sko.getLastKnownStatus().set(message.substring(idx + 1).trim());
@@ -456,15 +461,15 @@ static void dispatchDomainWatch(Watch.Response<Domain> item) {
456461
d = item.object;
457462
domainUID = d.getSpec().getDomainUID();
458463
LOGGER.info(MessageKeys.WATCH_DOMAIN, domainUID);
459-
existing = DomainPresenceInfoManager.lookup(domainUID);
464+
existing = DomainPresenceInfoManager.lookup(d.getMetadata().getNamespace(), domainUID);
460465
makeRightDomainPresence(existing, domainUID, d, added, false, true);
461466
break;
462467

463468
case "DELETED":
464469
d = item.object;
465470
domainUID = d.getSpec().getDomainUID();
466471
LOGGER.info(MessageKeys.WATCH_DOMAIN_DELETED, domainUID);
467-
existing = DomainPresenceInfoManager.lookup(domainUID);
472+
existing = DomainPresenceInfoManager.lookup(d.getMetadata().getNamespace(), domainUID);
468473
if (existing != null) {
469474
makeRightDomainPresence(existing, domainUID, d, true, true, true);
470475
}
@@ -652,13 +657,14 @@ private static void internalMakeRightDomainPresence(
652657
Step.StepAndPacket plan =
653658
isDeleting ? createDomainDownPlan(existing, ns, domainUID) : createDomainUpPlan(dom, ns);
654659

655-
runDomainPlan(dom, domainUID, plan, isDeleting, isWillInterrupt);
660+
runDomainPlan(dom, domainUID, ns, plan, isDeleting, isWillInterrupt);
656661
}
657662
}
658663

659664
static void runDomainPlan(
660665
Domain dom,
661666
String domainUID,
667+
String ns,
662668
Step.StepAndPacket plan,
663669
boolean isDeleting,
664670
boolean isWillInterrupt) {
@@ -694,7 +700,7 @@ public void onThrowable(Packet packet, Throwable throwable) {
694700
.getExecutor()
695701
.schedule(
696702
() -> {
697-
DomainPresenceInfo existing = DomainPresenceInfoManager.lookup(domainUID);
703+
DomainPresenceInfo existing = DomainPresenceInfoManager.lookup(ns, domainUID);
698704
if (existing != null) {
699705
internalMakeRightDomainPresence(
700706
existing, dom, domainUID, existing.getNamespace(), isDeleting, false);

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

Lines changed: 78 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import java.util.StringTokenizer;
2525
import java.util.TreeSet;
2626
import java.util.concurrent.ConcurrentHashMap;
27+
import java.util.concurrent.ConcurrentMap;
2728
import java.util.concurrent.ScheduledExecutorService;
2829
import java.util.concurrent.ThreadFactory;
2930
import java.util.concurrent.TimeUnit;
@@ -101,7 +102,7 @@ private static ThreadFactory getThreadFactory() {
101102

102103
static final Engine engine = new Engine(wrappedExecutorService);
103104

104-
static final Map<String, AtomicBoolean> isNamespaceStopping = new ConcurrentHashMap<>();
105+
static final ConcurrentMap<String, AtomicBoolean> isNamespaceStopping = new ConcurrentHashMap<>();
105106

106107
private static final Map<String, ConfigMapWatcher> configMapWatchers = new ConcurrentHashMap<>();
107108
private static final Map<String, DomainWatcher> domainWatchers = new ConcurrentHashMap<>();
@@ -121,12 +122,12 @@ private static ThreadFactory getThreadFactory() {
121122
static final String READINESS_PROBE_FAILURE_EVENT_FILTER =
122123
"reason=Unhealthy,type=Warning,involvedObject.fieldPath=spec.containers{weblogic-server}";
123124

124-
static Map<String, DomainPresenceInfo> getDomainPresenceInfos() {
125-
return DomainPresenceInfoManager.getDomainPresenceInfos();
125+
static Map<String, DomainPresenceInfo> getDomainPresenceInfos(String ns) {
126+
return DomainPresenceInfoManager.getDomainPresenceInfos(ns);
126127
}
127128

128-
static ServerKubernetesObjects getKubernetesObjects(String serverLegalName) {
129-
return ServerKubernetesObjectsManager.lookup(serverLegalName);
129+
static Map<String, ServerKubernetesObjects> getKubernetesObjects(String ns) {
130+
return ServerKubernetesObjectsManager.getServerKubernetesObjects(ns);
130131
}
131132

132133
/**
@@ -261,7 +262,7 @@ private static class StartNamespaceBeforeStep extends Step {
261262

262263
@Override
263264
public NextAction apply(Packet packet) {
264-
if (isNamespaceStopping.putIfAbsent(ns, new AtomicBoolean(false)) == null) {
265+
if (isNamespaceStopping(ns).get()) {
265266
try {
266267
HealthCheckHelper.performSecurityChecks(version, operatorNamespace, ns);
267268
} catch (Throwable e) {
@@ -283,10 +284,8 @@ private static void stopNamespaces(Collection<String> namespacesToStop) {
283284
}
284285
}
285286

286-
private static final AtomicBoolean UNINITIALIZED_NS_STOPPING = new AtomicBoolean(true);
287-
288287
static AtomicBoolean isNamespaceStopping(String ns) {
289-
return isNamespaceStopping.getOrDefault(ns, UNINITIALIZED_NS_STOPPING);
288+
return isNamespaceStopping.computeIfAbsent(ns, (key) -> new AtomicBoolean(true));
290289
}
291290

292291
static void runSteps(Step firstStep) {
@@ -527,30 +526,30 @@ public NextAction onFailure(Packet packet, CallResponse<DomainList> callResponse
527526

528527
@Override
529528
public NextAction onSuccess(Packet packet, CallResponse<DomainList> callResponse) {
529+
boolean starting = isNamespaceStopping.putIfAbsent(ns, new AtomicBoolean(false)) == null;
530+
530531
Set<String> domainUIDs = new HashSet<>();
531532
if (callResponse.getResult() != null) {
532533
for (Domain dom : callResponse.getResult().getItems()) {
533534
String domainUID = dom.getSpec().getDomainUID();
534535
domainUIDs.add(domainUID);
535536
DomainPresenceInfo info = DomainPresenceInfoManager.getOrCreate(dom);
536-
if (isNamespaceStopping(dom.getMetadata().getNamespace()).get()) {
537+
if (starting) {
537538
// Update domain here if namespace is not yet running
538539
info.setDomain(dom);
539540
}
540541
DomainProcessor.makeRightDomainPresence(info, domainUID, dom, true, false, false);
541542
}
542543
}
543544

544-
getDomainPresenceInfos()
545+
getDomainPresenceInfos(ns)
545546
.forEach(
546547
(key, value) -> {
547548
if (!domainUIDs.contains(key)) {
548-
if (ns.equals(value.getNamespace())) {
549-
// This is a stranded DomainPresenceInfo.
550-
value.setDeleting(true);
551-
Domain dom = value.getDomain();
552-
DomainProcessor.makeRightDomainPresence(value, key, dom, true, true, false);
553-
}
549+
// This is a stranded DomainPresenceInfo.
550+
value.setDeleting(true);
551+
Domain dom = value.getDomain();
552+
DomainProcessor.makeRightDomainPresence(value, key, dom, true, true, false);
554553
}
555554
});
556555

@@ -592,23 +591,66 @@ public NextAction onFailure(Packet packet, CallResponse<V1ServiceList> callRespo
592591
@Override
593592
public NextAction onSuccess(Packet packet, CallResponse<V1ServiceList> callResponse) {
594593
V1ServiceList result = callResponse.getResult();
594+
595+
Set<String> serviceNames = new HashSet<>();
596+
Set<String> channelNames = new HashSet<>();
597+
Set<String> clusterNames = new HashSet<>();
595598
if (result != null) {
596599
for (V1Service service : result.getItems()) {
597600
String domainUID = ServiceWatcher.getServiceDomainUID(service);
598601
String serverName = ServiceWatcher.getServiceServerName(service);
599602
String channelName = ServiceWatcher.getServiceChannelName(service);
600-
if (domainUID != null && serverName != null) {
603+
String clusterName = ServiceWatcher.getServiceClusterName(service);
604+
if (domainUID != null) {
601605
DomainPresenceInfo info = DomainPresenceInfoManager.getOrCreate(ns, domainUID);
602-
ServerKubernetesObjects sko =
603-
ServerKubernetesObjectsManager.getOrCreate(info, domainUID, serverName);
604-
if (channelName != null) {
605-
sko.getChannels().put(channelName, service);
606-
} else {
607-
sko.getService().set(service);
606+
if (clusterName != null) {
607+
clusterNames.add(clusterName);
608+
info.getClusters().put(clusterName, service);
609+
} else if (serverName != null) {
610+
ServerKubernetesObjects sko =
611+
ServerKubernetesObjectsManager.getOrCreate(info, domainUID, serverName);
612+
if (channelName != null) {
613+
channelNames.add(channelName);
614+
sko.getChannels().put(channelName, service);
615+
} else {
616+
serviceNames.add(service.getMetadata().getName());
617+
sko.getService().set(service);
618+
}
608619
}
609620
}
610621
}
611622
}
623+
624+
getDomainPresenceInfos(ns)
625+
.forEach(
626+
(key, value) -> {
627+
ConcurrentMap<String, V1Service> map = value.getClusters();
628+
map.forEach(
629+
(ckey, cvalue) -> {
630+
map.compute(
631+
ckey,
632+
(k, current) -> {
633+
return clusterNames.contains(ckey) ? current : null;
634+
});
635+
});
636+
});
637+
getKubernetesObjects(ns)
638+
.forEach(
639+
(key, value) -> {
640+
if (!serviceNames.contains(key)) {
641+
value.getService().set(null);
642+
}
643+
ConcurrentMap<String, V1Service> map = value.getChannels();
644+
map.forEach(
645+
(ckey, cvalue) -> {
646+
map.compute(
647+
ckey,
648+
(k, current) -> {
649+
return channelNames.contains(ckey) ? current : null;
650+
});
651+
});
652+
});
653+
612654
if (!serviceWatchers.containsKey(ns)) {
613655
serviceWatchers.put(ns, createServiceWatcher(ns, getInitialResourceVersion(result)));
614656
}
@@ -667,18 +709,30 @@ public NextAction onFailure(Packet packet, CallResponse<V1PodList> callResponse)
667709
@Override
668710
public NextAction onSuccess(Packet packet, CallResponse<V1PodList> callResponse) {
669711
V1PodList result = callResponse.getResult();
712+
713+
Set<String> podNames = new HashSet<>();
670714
if (result != null) {
671715
for (V1Pod pod : result.getItems()) {
672716
String domainUID = PodWatcher.getPodDomainUID(pod);
673717
String serverName = PodWatcher.getPodServerName(pod);
674718
if (domainUID != null && serverName != null) {
719+
podNames.add(pod.getMetadata().getName());
675720
DomainPresenceInfo info = DomainPresenceInfoManager.getOrCreate(ns, domainUID);
676721
ServerKubernetesObjects sko =
677722
ServerKubernetesObjectsManager.getOrCreate(info, domainUID, serverName);
678723
sko.getPod().set(pod);
679724
}
680725
}
681726
}
727+
728+
getKubernetesObjects(ns)
729+
.forEach(
730+
(key, value) -> {
731+
if (!podNames.contains(key)) {
732+
value.getPod().set(null);
733+
}
734+
});
735+
682736
if (!podWatchers.containsKey(ns)) {
683737
podWatchers.put(ns, createPodWatcher(ns, getInitialResourceVersion(result)));
684738
}

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,8 @@ public void receivedResponse(Watch.Response<V1Pod> item) {
9494
Boolean isReady = isReady(pod);
9595
String podName = pod.getMetadata().getName();
9696
Container c = ContainerResolver.getInstance().getContainer();
97-
ServerKubernetesObjects sko = ServerKubernetesObjectsManager.lookup(podName);
97+
ServerKubernetesObjects sko =
98+
ServerKubernetesObjectsManager.lookup(pod.getMetadata().getNamespace(), podName);
9899
if (sko != null) {
99100
sko.getLastKnownStatus().set(isReady ? WebLogicConstants.RUNNING_STATE : null);
100101
}

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,4 +74,13 @@ static String getServiceChannelName(V1Service service) {
7474
}
7575
return null;
7676
}
77+
78+
static String getServiceClusterName(V1Service service) {
79+
V1ObjectMeta meta = service.getMetadata();
80+
Map<String, String> labels = meta.getLabels();
81+
if (labels != null) {
82+
return labels.get(LabelConstants.CLUSTERNAME_LABEL);
83+
}
84+
return null;
85+
}
7786
}

0 commit comments

Comments
 (0)