Skip to content

Commit 5d90e96

Browse files
rjeberhardrjeberhard
committed
Restore deleted pods, services, and ingress resources
1 parent 8b1a076 commit 5d90e96

14 files changed

+427
-63
lines changed

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,7 @@ public Watch<Domain> initiateWatch(Object context, String resourceVersion) throw
7676
return Watch.createWatch(client.getApiClient(),
7777
client.callBuilder().with($ -> {
7878
$.resourceVersion = resourceVersion;
79-
$.timeoutSeconds = 2;
79+
$.timeoutSeconds = 30;
8080
$.watch = true;
8181
}).listDomainCall(ns),
8282
new TypeToken<Watch.Response<Domain>>() {

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

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
// Copyright 2018, Oracle Corporation and/or its affiliates. All rights reserved.
2+
// Licensed under the Universal Permissive License v 1.0 as shown at http://oss.oracle.com/licenses/upl.
3+
4+
package oracle.kubernetes.operator;
5+
6+
import java.util.concurrent.atomic.AtomicBoolean;
7+
8+
import com.google.gson.reflect.TypeToken;
9+
import io.kubernetes.client.ApiException;
10+
import io.kubernetes.client.models.V1beta1Ingress;
11+
import io.kubernetes.client.util.Watch;
12+
import oracle.kubernetes.operator.helpers.ClientHelper;
13+
import oracle.kubernetes.operator.helpers.ClientHolder;
14+
import oracle.kubernetes.operator.watcher.Watcher;
15+
import oracle.kubernetes.operator.watcher.Watching;
16+
import oracle.kubernetes.operator.watcher.WatchingEventDestination;
17+
18+
/**
19+
* This class handles Ingress watching. It receives Ingress change events and sends
20+
* them into the operator for processing.
21+
*/
22+
public class IngressWatcher implements Runnable {
23+
private final String ns;
24+
private final String initialResourceVersion;
25+
private final WatchingEventDestination<V1beta1Ingress> destination;
26+
private final AtomicBoolean isStopping;
27+
28+
public static IngressWatcher create(String ns, String initialResourceVersion, WatchingEventDestination<V1beta1Ingress> destination, AtomicBoolean isStopping) {
29+
IngressWatcher dlw = new IngressWatcher(ns, initialResourceVersion, destination, isStopping);
30+
Thread thread = new Thread(dlw);
31+
thread.setName("Thread-IngressWatcher-" + ns);
32+
thread.setDaemon(true);
33+
thread.start();
34+
return dlw;
35+
}
36+
37+
private IngressWatcher(String ns, String initialResourceVersion, WatchingEventDestination<V1beta1Ingress> destination, AtomicBoolean isStopping) {
38+
this.ns = ns;
39+
this.initialResourceVersion = initialResourceVersion;
40+
this.destination = destination;
41+
this.isStopping = isStopping;
42+
}
43+
44+
/**
45+
* Polling loop. Get the next Ingress object event and process it.
46+
*/
47+
@Override
48+
public void run() {
49+
ClientHelper helper = ClientHelper.getInstance();
50+
ClientHolder client = helper.take();
51+
try {
52+
Watching<V1beta1Ingress> w = createWatching(client);
53+
Watcher<V1beta1Ingress> watcher = new Watcher<V1beta1Ingress>(w, null, initialResourceVersion);
54+
55+
// invoke watch on current Thread. Won't return until watch stops
56+
watcher.doWatch();
57+
58+
} finally {
59+
helper.recycle(client);
60+
}
61+
}
62+
63+
protected Watching<V1beta1Ingress> createWatching(ClientHolder client) {
64+
return new Watching<V1beta1Ingress>() {
65+
66+
/**
67+
* Watcher callback to issue the list Ingress changes. It is driven by the
68+
* Watcher wrapper to issue repeated watch requests.
69+
* @param context user defined contact object or null
70+
* @param resourceVersion resource version to omit older events
71+
* @return Watch object or null if the operation should end
72+
* @throws ApiException if there is an API error.
73+
*/
74+
@Override
75+
public Watch<V1beta1Ingress> initiateWatch(Object context, String resourceVersion) throws ApiException {
76+
return Watch.createWatch(client.getApiClient(),
77+
client.callBuilder().with($ -> {
78+
$.resourceVersion = resourceVersion;
79+
$.labelSelector = LabelConstants.DOMAINUID_LABEL; // Any Ingress with a domainUID label
80+
$.timeoutSeconds = 30;
81+
$.watch = true;
82+
}).listIngressCall(ns),
83+
new TypeToken<Watch.Response<V1beta1Ingress>>() {
84+
}.getType());
85+
}
86+
87+
@Override
88+
public void eventCallback(Watch.Response<V1beta1Ingress> item) {
89+
processEventCallback(item);
90+
}
91+
92+
@Override
93+
public boolean isStopping() {
94+
return isStopping.get();
95+
}
96+
};
97+
}
98+
99+
public void processEventCallback(Watch.Response<V1beta1Ingress> item) {
100+
destination.eventCallback(item);
101+
}
102+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ public interface LabelConstants {
88
public static final String DOMAINUID_LABEL = "weblogic.domainUID";
99
public static final String DOMAINNAME_LABEL = "weblogic.domainName";
1010
public static final String SERVERNAME_LABEL = "weblogic.serverName";
11+
public static final String CHANNELNAME_LABEL = "weblogic.channelName";
1112
public static final String CLUSTERNAME_LABEL = "weblogic.clusterName";
1213

1314
}

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

Lines changed: 111 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -83,10 +83,13 @@ public class Main {
8383
private static final ConcurrentMap<String, Fiber> domainUpdaters = new ConcurrentHashMap<String, Fiber>();
8484
private static final ConcurrentMap<String, DomainPresenceInfo> domains = new ConcurrentHashMap<String, DomainPresenceInfo>();
8585
private static final AtomicBoolean stopping = new AtomicBoolean(false);
86+
private static String principal;
8687
private static RestServer restServer = null;
8788
private static Thread livenessThread = null;
8889
private static Map<String, DomainWatcher> domainWatchers = new HashMap<>();
8990
private static Map<String, PodWatcher> podWatchers = new HashMap<>();
91+
private static Map<String, ServiceWatcher> serviceWatchers = new HashMap<>();
92+
private static Map<String, IngressWatcher> ingressWatchers = new HashMap<>();
9093
private static KubernetesVersion version = null;
9194

9295
private static final Engine engine = new Engine("operator");
@@ -118,7 +121,7 @@ public static void main(String[] args) {
118121
if (serviceAccountName == null) {
119122
serviceAccountName = "default";
120123
}
121-
String principal = "system:serviceaccount:" + namespace + ":" + serviceAccountName;
124+
principal = "system:serviceaccount:" + namespace + ":" + serviceAccountName;
122125

123126
LOGGER.info(MessageKeys.OP_CONFIG_NAMESPACE, namespace);
124127
StringBuilder tns = new StringBuilder();
@@ -163,8 +166,9 @@ public static void main(String[] args) {
163166
// this would happen when the Domain was running BEFORE the Operator starts up
164167
LOGGER.info(MessageKeys.LISTING_DOMAINS);
165168
for (String ns : targetNamespaces) {
166-
PodWatcher pw = createPodWatcher(ns);
167-
podWatchers.put(ns, pw);
169+
podWatchers.put(ns, createPodWatcher(ns));
170+
serviceWatchers.put(ns, createServiceWatcher(ns));
171+
ingressWatchers.put(ns, createIngressWatcher(ns));
168172

169173
Step domainList = CallBuilder.create().listDomainAsync(ns, new ResponseStep<DomainList>(null) {
170174
@Override
@@ -181,12 +185,12 @@ public NextAction onSuccess(Packet packet, DomainList result, int statusCode,
181185
Map<String, List<String>> responseHeaders) {
182186
if (result != null) {
183187
for (Domain dom : result.getItems()) {
184-
doCheckAndCreateDomainPresence(principal, dom);
188+
doCheckAndCreateDomainPresence(dom);
185189
}
186190
}
187191

188192
// main logic now happens in the watch handlers
189-
domainWatchers.put(ns, createDomainWatcher(principal, ns, result != null ? result.getMetadata().getResourceVersion() : ""));
193+
domainWatchers.put(ns, createDomainWatcher(ns, result != null ? result.getMetadata().getResourceVersion() : ""));
190194
return doNext(packet);
191195
}
192196
});
@@ -360,7 +364,7 @@ public static void doRestartAdmin(String principal, String domainUID) {
360364
if (info != null) {
361365
Domain dom = info.getDomain();
362366
if (dom != null) {
363-
doCheckAndCreateDomainPresence(principal, dom, true, null, null);
367+
doCheckAndCreateDomainPresence(dom, true, null, null);
364368
}
365369
}
366370
}
@@ -377,7 +381,7 @@ public static void doRollingRestartServers(String principal, String domainUID, L
377381
if (info != null) {
378382
Domain dom = info.getDomain();
379383
if (dom != null) {
380-
doCheckAndCreateDomainPresence(principal, dom, false, servers, null);
384+
doCheckAndCreateDomainPresence(dom, false, servers, null);
381385
}
382386
}
383387
}
@@ -394,17 +398,17 @@ public static void doRollingRestartClusters(String principal, String domainUID,
394398
if (info != null) {
395399
Domain dom = info.getDomain();
396400
if (dom != null) {
397-
doCheckAndCreateDomainPresence(principal, dom, false, null, clusters);
401+
doCheckAndCreateDomainPresence(dom, false, null, clusters);
398402
}
399403
}
400404
}
401405

402-
private static void doCheckAndCreateDomainPresence(String principal, Domain dom) {
403-
doCheckAndCreateDomainPresence(principal, dom, false, null, null);
406+
private static void doCheckAndCreateDomainPresence(Domain dom) {
407+
doCheckAndCreateDomainPresence(dom, false, null, null);
404408
}
405409

406410
private static void doCheckAndCreateDomainPresence(
407-
String principal, Domain dom, boolean explicitRestartAdmin,
411+
Domain dom, boolean explicitRestartAdmin,
408412
List<String> explicitRestartServers, List<String> explicitRestartClusters) {
409413
LOGGER.entering();
410414

@@ -933,29 +937,14 @@ public ManagedServerDownStep(String serverName, ServerKubernetesObjects sko, Ste
933937

934938
@Override
935939
public NextAction apply(Packet packet) {
936-
V1ObjectMeta meta = sko.getPod().getMetadata();
937-
V1DeleteOptions deleteOptions = new V1DeleteOptions();
938940
List<V1Service> services = new ArrayList<V1Service>();
939941
services.add(sko.getService());
940942
services.addAll(sko.getChannels().values());
941943

942944
return doNext(IngressHelper.createRemoveServerStep(serverName, sko.getService(),
943-
new DeleteServiceListStep(services, CallBuilder.create().deletePodAsync(meta.getName(), meta.getNamespace(), deleteOptions, new ResponseStep<V1Status>(next) {
944-
@Override
945-
public NextAction onFailure(Packet packet, ApiException e, int statusCode,
946-
Map<String, List<String>> responseHeaders) {
947-
if (statusCode == CallBuilder.NOT_FOUND) {
948-
return onSuccess(packet, null, statusCode, responseHeaders);
949-
}
950-
return super.onFailure(packet, e, statusCode, responseHeaders);
951-
}
952-
953-
@Override
954-
public NextAction onSuccess(Packet packet, V1Status result, int statusCode,
955-
Map<String, List<String>> responseHeaders) {
956-
return doNext(new ManagedServerDownFinalizeStep(serverName, next), packet);
957-
}
958-
}))), packet);
945+
new DeleteServiceListStep(services,
946+
PodHelper.deletePodStep(sko,
947+
new ManagedServerDownFinalizeStep(serverName, next)))), packet);
959948
}
960949
}
961950

@@ -1262,22 +1251,109 @@ public static boolean getStopping() {
12621251
return stopping.get();
12631252
}
12641253

1265-
private static DomainWatcher createDomainWatcher(String principal, String namespace, String initialResourceVersion) {
1266-
return DomainWatcher.create(namespace, initialResourceVersion, (item) -> { dispatchDomainWatch(item, principal); }, stopping);
1254+
private static DomainWatcher createDomainWatcher(String namespace, String initialResourceVersion) {
1255+
return DomainWatcher.create(namespace, initialResourceVersion, Main::dispatchDomainWatch, stopping);
12671256
}
12681257

12691258
private static PodWatcher createPodWatcher(String namespace) {
1270-
return PodWatcher.create(namespace, "", stopping);
1259+
return PodWatcher.create(namespace, "", Main::dispatchPodWatch, stopping);
12711260
}
12721261

1262+
private static void dispatchPodWatch(Watch.Response<V1Pod> item) {
1263+
switch (item.type) {
1264+
case "DELETED":
1265+
V1Pod p = item.object;
1266+
V1ObjectMeta metadata = p.getMetadata();
1267+
String domainUID = metadata.getLabels().get(LabelConstants.DOMAINUID_LABEL);
1268+
String serverName = metadata.getLabels().get(LabelConstants.SERVERNAME_LABEL);
1269+
if (domainUID != null) {
1270+
DomainPresenceInfo info = domains.get(domainUID);
1271+
if (info != null && serverName != null) {
1272+
ServerKubernetesObjects sko = info.getServers().get(serverName);
1273+
if (sko != null) {
1274+
if (sko.getPod() != null) {
1275+
// Pod was deleted, but sko still contains a non-null entry
1276+
LOGGER.info(MessageKeys.POD_DELETED, domainUID, metadata.getNamespace(), serverName);
1277+
doCheckAndCreateDomainPresence(info.getDomain());
1278+
}
1279+
}
1280+
}
1281+
}
1282+
break;
1283+
1284+
case "ERROR":
1285+
default:
1286+
}
1287+
}
1288+
1289+
1290+
private static ServiceWatcher createServiceWatcher(String namespace) {
1291+
return ServiceWatcher.create(namespace, "", Main::dispatchServiceWatch, stopping);
1292+
}
1293+
1294+
private static void dispatchServiceWatch(Watch.Response<V1Service> item) {
1295+
switch (item.type) {
1296+
case "DELETED":
1297+
V1Service s = item.object;
1298+
V1ObjectMeta metadata = s.getMetadata();
1299+
String domainUID = metadata.getLabels().get(LabelConstants.DOMAINUID_LABEL);
1300+
String serverName = metadata.getLabels().get(LabelConstants.SERVERNAME_LABEL);
1301+
String channelName = metadata.getLabels().get(LabelConstants.CHANNELNAME_LABEL);
1302+
if (domainUID != null) {
1303+
DomainPresenceInfo info = domains.get(domainUID);
1304+
if (info != null && serverName != null) {
1305+
ServerKubernetesObjects sko = info.getServers().get(serverName);
1306+
if (sko != null) {
1307+
if ((channelName != null ? sko.getChannels().get(channelName) : sko.getService()) != null) {
1308+
// Service was deleted, but sko still contains a non-null entry
1309+
LOGGER.info(MessageKeys.SERVICE_DELETED, domainUID, metadata.getNamespace(), serverName);
1310+
doCheckAndCreateDomainPresence(info.getDomain());
1311+
}
1312+
}
1313+
}
1314+
}
1315+
break;
1316+
1317+
case "ERROR":
1318+
default:
1319+
}
1320+
}
1321+
1322+
private static IngressWatcher createIngressWatcher(String namespace) {
1323+
return IngressWatcher.create(namespace, "", Main::dispatchIngressWatch, stopping);
1324+
}
1325+
1326+
private static void dispatchIngressWatch(Watch.Response<V1beta1Ingress> item) {
1327+
switch (item.type) {
1328+
case "DELETED":
1329+
V1beta1Ingress i = item.object;
1330+
V1ObjectMeta metadata = i.getMetadata();
1331+
String domainUID = metadata.getLabels().get(LabelConstants.DOMAINUID_LABEL);
1332+
String clusterName = metadata.getLabels().get(LabelConstants.CLUSTERNAME_LABEL);
1333+
if (domainUID != null) {
1334+
DomainPresenceInfo info = domains.get(domainUID);
1335+
if (info != null && clusterName != null) {
1336+
if (clusterName != null && info.getIngresses().get(clusterName) != null) {
1337+
// Ingress was deleted, but sko still contains a non-null entry
1338+
LOGGER.info(MessageKeys.INGRESS_DELETED, domainUID, metadata.getNamespace(), clusterName);
1339+
doCheckAndCreateDomainPresence(info.getDomain());
1340+
}
1341+
}
1342+
}
1343+
break;
1344+
1345+
case "ERROR":
1346+
default:
1347+
}
1348+
}
1349+
12731350
/**
12741351
* Dispatch the Domain event to the appropriate handler.
12751352
*
12761353
* @param item An item received from a Watch response.
12771354
* @param principal The name of the principal that will be used in this watch.
12781355
*/
1279-
public static void dispatchDomainWatch(Watch.Response<Domain> item, String principal) {
1280-
1356+
private static void dispatchDomainWatch(Watch.Response<Domain> item) {
12811357
try {
12821358
Domain d;
12831359
String domainUID;
@@ -1287,7 +1363,7 @@ public static void dispatchDomainWatch(Watch.Response<Domain> item, String princ
12871363
d = item.object;
12881364
domainUID = d.getSpec().getDomainUID();
12891365
LOGGER.info(MessageKeys.WATCH_DOMAIN, domainUID);
1290-
doCheckAndCreateDomainPresence(principal, d);
1366+
doCheckAndCreateDomainPresence(d);
12911367
break;
12921368

12931369
case "DELETED":

0 commit comments

Comments
 (0)