Skip to content

Commit 48aab47

Browse files
csviricsviri
committed
Dependent Resource Experiment
1 parent 8aa933b commit 48aab47

File tree

9 files changed

+335
-0
lines changed

9 files changed

+335
-0
lines changed

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/DependentResource.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource;
2+
3+
import io.fabric8.kubernetes.client.CustomResource;
4+
import io.javaoperatorsdk.operator.processing.event.EventSource;
5+
6+
public interface DependentResource<T, K extends StatusDescriptor> {
7+
8+
K reconcile(T input);
9+
10+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/SimpleStatus.java

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource;
2+
3+
public class SimpleStatus implements StatusDescriptor {
4+
5+
private Status status;
6+
7+
public SimpleStatus(Status status) {
8+
this.status = status;
9+
}
10+
11+
@Override
12+
public Status getStatus() {
13+
return status;
14+
}
15+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/Status.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource;
2+
3+
public enum Status {
4+
CREATED_SUCCESSFULLY,
5+
IN_PROGRESS,
6+
ERROR
7+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/StatusDescriptor.java

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource;
2+
3+
public interface StatusDescriptor {
4+
5+
Status getStatus();
6+
7+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/deployment/DeploymentDependentResource.java

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource.deployment;
2+
3+
import io.fabric8.kubernetes.api.model.OwnerReference;
4+
import io.fabric8.kubernetes.api.model.apps.Deployment;
5+
import io.fabric8.kubernetes.client.KubernetesClient;
6+
import io.fabric8.kubernetes.client.KubernetesClientException;
7+
import io.fabric8.kubernetes.client.Watcher;
8+
import io.fabric8.kubernetes.client.utils.Serialization;
9+
import io.javaoperatorsdk.operator.processing.dependentresource.DependentResource;
10+
import io.javaoperatorsdk.operator.processing.dependentresource.Status;
11+
import io.javaoperatorsdk.operator.processing.event.AbstractEventSource;
12+
import org.slf4j.Logger;
13+
import org.slf4j.LoggerFactory;
14+
15+
import java.io.IOException;
16+
import java.io.InputStream;
17+
import java.util.Map;
18+
import java.util.Objects;
19+
import java.util.Optional;
20+
import java.util.concurrent.ConcurrentHashMap;
21+
22+
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getUID;
23+
import static io.javaoperatorsdk.operator.processing.KubernetesResourceUtils.getVersion;
24+
import static java.net.HttpURLConnection.HTTP_GONE;
25+
26+
public class DeploymentDependentResource extends AbstractEventSource
27+
implements DependentResource<DeploymentInput, DeploymentStatus>, Watcher<Deployment> {
28+
29+
private final static Logger log = LoggerFactory.getLogger(DeploymentDependentResource.class);
30+
31+
private final KubernetesClient client;
32+
private final Map<String, Deployment> deploymentCache = new ConcurrentHashMap<>();
33+
34+
public DeploymentDependentResource(KubernetesClient kubernetesClient) {
35+
this.client = kubernetesClient;
36+
registerWatch();
37+
}
38+
39+
private void registerWatch() {
40+
client.apps().deployments().inAnyNamespace().withLabel("managed-by", "tomcat-operator").watch(this);
41+
}
42+
43+
44+
@Override
45+
public DeploymentStatus reconcile(DeploymentInput input) {
46+
47+
Optional<Deployment> cachedDeployment = getLatestDeployment(input.getCustomResourceUid());
48+
Deployment deployment;
49+
if (cachedDeployment.isEmpty() || !isDeploymentAccordingToInput(input, cachedDeployment.get())) {
50+
deployment = createOrUpdateDeployment(input);
51+
} else {
52+
deployment = cachedDeployment.get();
53+
}
54+
55+
return new DeploymentStatus(isDeploymentAccordingToInput(input, deployment) ?
56+
Status.CREATED_SUCCESSFULLY : Status.IN_PROGRESS, deployment);
57+
}
58+
59+
private Deployment createOrUpdateDeployment(DeploymentInput input) {
60+
String ns = input.getNamespace();
61+
Deployment existingDeployment = client.apps().deployments()
62+
.inNamespace(ns).withName(input.getName())
63+
.get();
64+
if (existingDeployment == null) {
65+
Deployment deployment = loadYaml(Deployment.class, "deployment.yaml");
66+
deployment.getMetadata().setName(input.getName());
67+
deployment.getMetadata().setNamespace(ns);
68+
deployment.getMetadata().getLabels().put("created-by", input.getName());
69+
deployment.getMetadata().getLabels().put("managed-by", "tomcat-operator");
70+
// set tomcat version
71+
deployment.getSpec().getTemplate().getSpec().getContainers().get(0).setImage(input.image());
72+
deployment.getSpec().setReplicas(input.getReplicaCount());
73+
74+
//make sure label selector matches label (which has to be matched by service selector too)
75+
deployment.getSpec().getTemplate().getMetadata().getLabels().put("app", input.getName());
76+
deployment.getSpec().getSelector().getMatchLabels().put("app",input.getName());
77+
78+
OwnerReference ownerReference = deployment.getMetadata().getOwnerReferences().get(0);
79+
ownerReference.setName(input.getName());
80+
ownerReference.setUid(input.getCustomResourceUid());
81+
82+
log.info("Creating or updating Deployment {} in {}", deployment.getMetadata().getName(), ns);
83+
return client.apps().deployments().inNamespace(ns).create(deployment);
84+
} else {
85+
existingDeployment.getSpec().getTemplate().getSpec().getContainers().get(0).setImage(input.image());
86+
existingDeployment.getSpec().setReplicas(input.getReplicaCount());
87+
return client.apps().deployments().inNamespace(ns).createOrReplace(existingDeployment);
88+
}
89+
}
90+
91+
private boolean isDeploymentAccordingToInput(DeploymentInput input, Deployment deployment) {
92+
io.fabric8.kubernetes.api.model.apps.DeploymentStatus deploymentStatus = Objects.requireNonNullElse(deployment.getStatus(), new io.fabric8.kubernetes.api.model.apps.DeploymentStatus());
93+
String deploymentImage = deployment.getSpec().getTemplate().getSpec()
94+
.getContainers().get(0).getImage();
95+
return input.getReplicaCount() == Objects.requireNonNullElse(deploymentStatus.getReadyReplicas(), 0)
96+
&& input.image().equals(deploymentImage);
97+
98+
}
99+
100+
101+
public Optional<Deployment> getLatestDeployment(String ownerUID) {
102+
return Optional.ofNullable(deploymentCache.get(ownerUID));
103+
}
104+
105+
@Override
106+
public void eventReceived(Watcher.Action action, Deployment deployment) {
107+
log.info("Event received for action: {}, Deployment: {} (rr={})", action.name(), deployment.getMetadata().getName(), deployment.getStatus().getReadyReplicas());
108+
if (action == Action.ERROR) {
109+
log.warn("Skipping {} event for custom resource uid: {}, version: {}", action,
110+
getUID(deployment), getVersion(deployment));
111+
return;
112+
}
113+
deploymentCache.put(deployment.getMetadata().getOwnerReferences().get(0).getUid(), deployment);
114+
eventHandler.handleEvent(new DeploymentEvent(action, deployment, this));
115+
}
116+
117+
@Override
118+
public void onClose(KubernetesClientException e) {
119+
if (e == null) {
120+
return;
121+
}
122+
if (e.getCode() == HTTP_GONE) {
123+
log.warn("Received error for watch, will try to reconnect.", e);
124+
registerWatch();
125+
} else {
126+
// Note that this should not happen normally, since fabric8 client handles reconnect.
127+
// In case it tries to reconnect this method is not called.
128+
log.error("Unexpected error happened with watch. Will exit.", e);
129+
System.exit(1);
130+
}
131+
}
132+
133+
@Override
134+
public void eventSourceDeRegisteredForResource(String ownerUID) {
135+
deploymentCache.remove(ownerUID);
136+
}
137+
138+
private <T> T loadYaml(Class<T> clazz, String yaml) {
139+
try (InputStream is = getClass().getResourceAsStream(yaml)) {
140+
return Serialization.unmarshal(is, clazz);
141+
} catch (IOException ex) {
142+
throw new IllegalStateException("Cannot find yaml on classpath: " + yaml);
143+
}
144+
}
145+
146+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/deployment/DeploymentEvent.java

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource.deployment;
2+
3+
import io.fabric8.kubernetes.api.model.apps.Deployment;
4+
import io.fabric8.kubernetes.client.Watcher;
5+
import io.javaoperatorsdk.operator.processing.event.AbstractEvent;
6+
7+
public class DeploymentEvent extends AbstractEvent {
8+
9+
private final Watcher.Action action;
10+
private final Deployment deployment;
11+
12+
public DeploymentEvent(Watcher.Action action, Deployment resource,
13+
DeploymentDependentResource deploymentEventSource) {
14+
super(resource.getMetadata().getOwnerReferences().get(0).getUid(), deploymentEventSource);
15+
this.action = action;
16+
this.deployment = resource;
17+
}
18+
19+
public Watcher.Action getAction() {
20+
return action;
21+
}
22+
23+
public String resourceUid() {
24+
return getDeployment().getMetadata().getUid();
25+
}
26+
27+
@Override
28+
public String toString() {
29+
return "CustomResourceEvent{" +
30+
"action=" + action +
31+
", resource=[ name=" + getDeployment().getMetadata().getName() + ", kind=" + getDeployment().getKind() +
32+
", apiVersion=" + getDeployment().getApiVersion() + " ,resourceVersion=" + getDeployment().getMetadata().getResourceVersion() +
33+
", markedForDeletion: " + (getDeployment().getMetadata().getDeletionTimestamp() != null
34+
&& !getDeployment().getMetadata().getDeletionTimestamp().isEmpty()) +
35+
" ]" +
36+
'}';
37+
}
38+
39+
public Deployment getDeployment() {
40+
return deployment;
41+
}
42+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/deployment/DeploymentInput.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource.deployment;
2+
3+
public class DeploymentInput {
4+
5+
6+
7+
private final String customResourceUid; // extract this to an interface?
8+
private final String name;
9+
private final String imageName;
10+
private final String imageVersion;
11+
private final String namespace;
12+
private final int replicaCount;
13+
14+
public DeploymentInput(String customResourceUid, String name, String imageName, String imageVersion, String namespace, int replicaCount) {
15+
this.customResourceUid = customResourceUid;
16+
this.name = name;
17+
this.imageName = imageName;
18+
this.imageVersion = imageVersion;
19+
this.namespace = namespace;
20+
this.replicaCount = replicaCount;
21+
}
22+
23+
public String getImageName() {
24+
return imageName;
25+
}
26+
27+
public String getImageVersion() {
28+
return imageVersion;
29+
}
30+
31+
public String getName() {
32+
return name;
33+
}
34+
35+
public String getNamespace() {
36+
return namespace;
37+
}
38+
39+
public int getReplicaCount() {
40+
return replicaCount;
41+
}
42+
43+
public String getCustomResourceUid() {
44+
return customResourceUid;
45+
}
46+
47+
public String image() {
48+
return imageName + ":" + imageVersion;
49+
}
50+
}

‎operator-framework/src/main/java/io/javaoperatorsdk/operator/processing/dependentresource/deployment/DeploymentStatus.java

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
package io.javaoperatorsdk.operator.processing.dependentresource.deployment;
2+
3+
import io.fabric8.kubernetes.api.model.apps.Deployment;
4+
import io.javaoperatorsdk.operator.processing.dependentresource.SimpleStatus;
5+
import io.javaoperatorsdk.operator.processing.dependentresource.Status;
6+
7+
public class DeploymentStatus extends SimpleStatus {
8+
9+
private final Deployment deployment;
10+
11+
public DeploymentStatus(Status status, Deployment deployment) {
12+
super(status);
13+
this.deployment = deployment;
14+
}
15+
16+
public Deployment getDeployment() {
17+
return deployment;
18+
}
19+
}

‎operator-framework/src/test/resources/io/javaoperatorsdk/operator/processing/dependentresource/deployment/deployment.yaml

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
apiVersion: apps/v1
2+
kind: Deployment
3+
metadata:
4+
name: ""
5+
labels:
6+
created-by: ""
7+
managed-by: ""
8+
ownerReferences:
9+
- apiVersion: apps/v1
10+
kind: Tomcat
11+
name: ""
12+
uid: ""
13+
spec:
14+
selector:
15+
matchLabels:
16+
app: ""
17+
replicas: 1
18+
template:
19+
metadata:
20+
labels:
21+
app: ""
22+
spec:
23+
containers:
24+
- name: tomcat
25+
image: tomcat:8.0
26+
ports:
27+
- containerPort: 8080
28+
volumeMounts:
29+
- mountPath: /usr/local/tomcat/webapps
30+
name: webapps-volume
31+
- name: war-downloader
32+
image: busybox:1.28
33+
command: ['tail', '-f', '/dev/null']
34+
volumeMounts:
35+
- name: webapps-volume
36+
mountPath: /data
37+
volumes:
38+
- name: webapps-volume
39+
emptydir: {}

0 commit comments

Comments
 (0)