mnl/VM-Operator
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 4 Wiki Activity Actions

Merge branch 'main' into testing

Browse source
This commit is contained in:
Michael Lipp 2025-03-20 18:49:39 +01:00
commit 6fd7021803
5 changed files with 212 additions and 183 deletions
Download patch file Download diff file Expand all files Collapse all files

23
deploy/vmop-deployment.yaml
View file

@ -22,6 +22,19 @@ spec:
image: >-
registry.mnl.de/org/jdrupes/vm-operator/org.jdrupes.vmoperator.manager:testing
imagePullPolicy: Always
env:
- name: JAVA_OPTS
# The VM operator needs about 25 MB of memory, plus 1 MB for
# each VM. The reason is that for the sake of effeciency, we
# have to keep a parsed representation of the CRD in memory,
# which requires about 512 KB per VM. While handling updates,
# we temporarily have the old and the new version of the CRD
# in memory, so we need another 512 KB per VM.
value: "-Xmx128m"
resources:
requests:
cpu: 100m
memory: 128Mi
volumeMounts:
- name: config
mountPath: /etc/opt/vmoperator
@ -33,16 +46,6 @@ spec:
- ALL
readOnlyRootFilesystem: true
allowPrivilegeEscalation: false
resources:
requests:
cpu: 100m
# The VM operator needs about 25 MB of memory, plus 1 MB for
# each VM. The reason is that for the sake of effeciency, we
# have to keep a parsed representation of the CRD in memory,
# which requires about 512 KB per VM. While handling updates,
# we temporarily have the old and the new version of the CRD
# in memory, so we need another 512 KB per VM.
memory: 256Mi
volumes:
- name: config
configMap:

186
org.jdrupes.vmoperator.manager/src/org/jdrupes/vmoperator/manager/Controller.java
View file

@ -20,29 +20,34 @@ package org.jdrupes.vmoperator.manager;
import com.google.gson.JsonObject;
import io.kubernetes.client.apimachinery.GroupVersionKind;
import io.kubernetes.client.custom.V1Patch;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.Configuration;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.Comparator;
import java.util.Optional;
import java.util.logging.Level;
import org.jdrupes.vmoperator.common.Constants.Crd;
import org.jdrupes.vmoperator.common.Constants.Status;
import org.jdrupes.vmoperator.common.K8sClient;
import org.jdrupes.vmoperator.common.K8sDynamicStub;
import org.jdrupes.vmoperator.common.K8sObserver.ResponseType;
import org.jdrupes.vmoperator.common.VmDefinition.Assignment;
import org.jdrupes.vmoperator.common.VmDefinitionStub;
import org.jdrupes.vmoperator.common.VmPool;
import org.jdrupes.vmoperator.manager.events.AssignVm;
import org.jdrupes.vmoperator.manager.events.ChannelManager;
import org.jdrupes.vmoperator.manager.events.Exit;
import org.jdrupes.vmoperator.manager.events.GetPools;
import org.jdrupes.vmoperator.manager.events.GetVms;
import org.jdrupes.vmoperator.manager.events.GetVms.VmData;
import org.jdrupes.vmoperator.manager.events.ModifyVm;
import org.jdrupes.vmoperator.manager.events.PodChanged;
import org.jdrupes.vmoperator.manager.events.UpdateAssignment;
import org.jdrupes.vmoperator.manager.events.VmChannel;
import org.jdrupes.vmoperator.manager.events.VmDefChanged;
import org.jdrupes.vmoperator.util.GsonPtr;
import org.jdrupes.vmoperator.manager.events.VmPoolChanged;
import org.jgrapes.core.Channel;
import org.jgrapes.core.Component;
import org.jgrapes.core.annotation.Handler;
@ -89,6 +94,7 @@ import org.jgrapes.util.events.ConfigurationUpdate;
public class Controller extends Component {
private String namespace;
private final ChannelManager<String, VmChannel, ?> chanMgr;
/**
* Creates a new instance.
@ -97,17 +103,16 @@ public class Controller extends Component {
public Controller(Channel componentChannel) {
super(componentChannel);
// Prepare component tree
ChannelManager<String, VmChannel, ?> chanMgr
= new ChannelManager<>(name -> {
try {
return new VmChannel(channel(), newEventPipeline(),
new K8sClient());
} catch (IOException e) {
logger.log(Level.SEVERE, e, () -> "Failed to create client"
+ " for handling changes: " + e.getMessage());
return null;
}
});
chanMgr = new ChannelManager<>(name -> {
try {
return new VmChannel(channel(), newEventPipeline(),
new K8sClient());
} catch (IOException e) {
logger.log(Level.SEVERE, e, () -> "Failed to create client"
+ " for handling changes: " + e.getMessage());
return null;
}
});
attach(new VmMonitor(channel(), chanMgr));
attach(new DisplaySecretMonitor(channel(), chanMgr));
// Currently, we don't use the IP assigned by the load balancer
@ -181,74 +186,119 @@ public class Controller extends Component {
}
/**
* On modify vm.
* Returns the VM data.
*
* @param event the event
* @throws ApiException the api exception
* @throws IOException Signals that an I/O exception has occurred.
*/
@Handler
public void onModifyVm(ModifyVm event, VmChannel channel)
throws ApiException, IOException {
patchVmDef(channel.client(), event.name(), "spec/vm/" + event.path(),
event.value());
}
private void patchVmDef(K8sClient client, String name, String path,
Object value) throws ApiException, IOException {
var vmStub = K8sDynamicStub.get(client,
new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM), namespace,
name);
// Patch running
String valueAsText = value instanceof String
? "\"" + value + "\""
: value.toString();
var res = vmStub.patch(V1Patch.PATCH_FORMAT_JSON_PATCH,
new V1Patch("[{\"op\": \"replace\", \"path\": \"/"
+ path + "\", \"value\": " + valueAsText + "}]"),
client.defaultPatchOptions());
if (!res.isPresent()) {
logger.warning(
() -> "Cannot patch definition for Vm " + vmStub.name());
}
public void onGetVms(GetVms event) {
event.setResult(chanMgr.channels().stream()
.filter(c -> event.name().isEmpty()
|| c.vmDefinition().name().equals(event.name().get()))
.filter(c -> event.user().isEmpty() && event.roles().isEmpty()
|| !c.vmDefinition().permissionsFor(event.user().orElse(null),
event.roles()).isEmpty())
.filter(c -> event.fromPool().isEmpty()
|| c.vmDefinition().assignment().map(Assignment::pool)
.map(p -> p.equals(event.fromPool().get())).orElse(false))
.filter(c -> event.toUser().isEmpty()
|| c.vmDefinition().assignment().map(Assignment::user)
.map(u -> u.equals(event.toUser().get())).orElse(false))
.map(c -> new VmData(c.vmDefinition(), c))
.toList());
}
/**
* Attempt to Update the assignment information in the status of the
* VM CR. Returns true if successful. The handler does not attempt
* retries, because in case of failure it will be necessary to
* re-evaluate the chosen VM.
* Assign a VM if not already assigned.
*
* @param event the event
* @param channel the channel
* @throws ApiException the api exception
* @throws InterruptedException
*/
@Handler
public void onUpdatedAssignment(UpdateAssignment event, VmChannel channel)
throws ApiException {
try {
var vmDef = channel.vmDefinition();
var vmStub = VmDefinitionStub.get(channel.client(),
new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM),
vmDef.namespace(), vmDef.name());
if (vmStub.updateStatus(vmDef, from -> {
JsonObject status = from.statusJson();
var assignment = GsonPtr.to(status).to(Status.ASSIGNMENT);
assignment.set("pool", event.fromPool().name());
assignment.set("user", event.toUser());
assignment.set("lastUsed", Instant.now().toString());
return status;
}).isPresent()) {
event.setResult(true);
@SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
public void onAssignVm(AssignVm event)
throws ApiException, InterruptedException {
while (true) {
// Search for existing assignment.
var vmQuery = chanMgr.channels().stream()
.filter(c -> c.vmDefinition().assignment().map(Assignment::pool)
.map(p -> p.equals(event.fromPool())).orElse(false))
.filter(c -> c.vmDefinition().assignment().map(Assignment::user)
.map(u -> u.equals(event.toUser())).orElse(false))
.findFirst();
if (vmQuery.isPresent()) {
var vmDef = vmQuery.get().vmDefinition();
event.setResult(new VmData(vmDef, vmQuery.get()));
return;
}
} catch (ApiException e) {
// Log exceptions except for conflict, which can be expected
if (HttpURLConnection.HTTP_CONFLICT != e.getCode()) {
throw e;
// Get the pool definition for checking possible assignment
VmPool vmPool = newEventPipeline().fire(new GetPools()
.withName(event.fromPool())).get().stream().findFirst()
.orElse(null);
if (vmPool == null) {
return;
}
// Find available VM.
vmQuery = chanMgr.channels().stream()
.filter(c -> vmPool.isAssignable(c.vmDefinition()))
.sorted(Comparator.comparing((VmChannel c) -> c.vmDefinition()
.assignment().map(Assignment::lastUsed)
.orElse(Instant.ofEpochSecond(0)))
.thenComparing(preferRunning))
.findFirst();
// None found
if (vmQuery.isEmpty()) {
return;
}
// Assign to user
var chosenVm = vmQuery.get();
if (Optional.ofNullable(chosenVm.fire(new UpdateAssignment(
vmPool, event.toUser())).get()).orElse(false)) {
var vmDef = chosenVm.vmDefinition();
event.setResult(new VmData(vmDef, chosenVm));
// Make sure that a newly assigned VM is running.
chosenVm.fire(new ModifyVm(vmDef.name(), "state", "Running"));
return;
}
}
event.setResult(false);
}
private static Comparator<VmChannel> preferRunning
= new Comparator<>() {
@Override
public int compare(VmChannel ch1, VmChannel ch2) {
if (ch1.vmDefinition().conditionStatus("Running").orElse(false)
&& !ch2.vmDefinition().conditionStatus("Running")
.orElse(false)) {
return -1;
}
return 0;
}
};
/**
* When s pool is deleted, remove all related assignments.
*
* @param event the event
* @throws InterruptedException
*/
@Handler
@SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
public void onPoolChanged(VmPoolChanged event) throws InterruptedException {
if (!event.deleted()) {
return;
}
var vms = newEventPipeline()
.fire(new GetVms().assignedFrom(event.vmPool().name())).get();
for (var vm : vms) {
vm.channel().fire(new UpdateAssignment(event.vmPool(), null));
}
}
/**

159
org.jdrupes.vmoperator.manager/src/org/jdrupes/vmoperator/manager/VmMonitor.java
View file

@ -18,21 +18,25 @@
package org.jdrupes.vmoperator.manager;
import com.google.gson.JsonObject;
import io.kubernetes.client.apimachinery.GroupVersionKind;
import io.kubernetes.client.custom.V1Patch;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.models.V1ObjectMeta;
import io.kubernetes.client.util.Watch;
import io.kubernetes.client.util.generic.options.ListOptions;
import java.io.IOException;
import java.net.HttpURLConnection;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import org.jdrupes.vmoperator.common.Constants.Crd;
import org.jdrupes.vmoperator.common.Constants.Status;
import org.jdrupes.vmoperator.common.K8s;
import org.jdrupes.vmoperator.common.K8sClient;
import org.jdrupes.vmoperator.common.K8sDynamicStub;
@ -40,23 +44,18 @@ import org.jdrupes.vmoperator.common.K8sObserver.ResponseType;
import org.jdrupes.vmoperator.common.K8sV1ConfigMapStub;
import org.jdrupes.vmoperator.common.K8sV1StatefulSetStub;
import org.jdrupes.vmoperator.common.VmDefinition;
import org.jdrupes.vmoperator.common.VmDefinition.Assignment;
import org.jdrupes.vmoperator.common.VmDefinitionStub;
import org.jdrupes.vmoperator.common.VmDefinitions;
import org.jdrupes.vmoperator.common.VmExtraData;
import org.jdrupes.vmoperator.common.VmPool;
import static org.jdrupes.vmoperator.manager.Constants.APP_NAME;
import static org.jdrupes.vmoperator.manager.Constants.VM_OP_NAME;
import org.jdrupes.vmoperator.manager.events.AssignVm;
import org.jdrupes.vmoperator.manager.events.ChannelManager;
import org.jdrupes.vmoperator.manager.events.GetPools;
import org.jdrupes.vmoperator.manager.events.GetVms;
import org.jdrupes.vmoperator.manager.events.GetVms.VmData;
import org.jdrupes.vmoperator.manager.events.ModifyVm;
import org.jdrupes.vmoperator.manager.events.PodChanged;
import org.jdrupes.vmoperator.manager.events.UpdateAssignment;
import org.jdrupes.vmoperator.manager.events.VmChannel;
import org.jdrupes.vmoperator.manager.events.VmDefChanged;
import org.jdrupes.vmoperator.util.GsonPtr;
import org.jgrapes.core.Channel;
import org.jgrapes.core.Event;
import org.jgrapes.core.annotation.Handler;
@ -233,101 +232,79 @@ public class VmMonitor extends
}
/**
* Returns the VM data.
*
* @param event the event
*/
@Handler
public void onGetVms(GetVms event) {
event.setResult(channelManager.channels().stream()
.filter(c -> event.name().isEmpty()
|| c.vmDefinition().name().equals(event.name().get()))
.filter(c -> event.user().isEmpty() && event.roles().isEmpty()
|| !c.vmDefinition().permissionsFor(event.user().orElse(null),
event.roles()).isEmpty())
.filter(c -> event.fromPool().isEmpty()
|| c.vmDefinition().assignment().map(Assignment::pool)
.map(p -> p.equals(event.fromPool().get())).orElse(false))
.filter(c -> event.toUser().isEmpty()
|| c.vmDefinition().assignment().map(Assignment::user)
.map(u -> u.equals(event.toUser().get())).orElse(false))
.map(c -> new VmData(c.vmDefinition(), c))
.toList());
}
/**
* Assign a VM if not already assigned.
* On modify vm.
*
* @param event the event
* @throws ApiException the api exception
* @throws InterruptedException
* @throws IOException Signals that an I/O exception has occurred.
*/
@Handler
@SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
public void onAssignVm(AssignVm event)
throws ApiException, InterruptedException {
while (true) {
// Search for existing assignment.
var vmQuery = channelManager.channels().stream()
.filter(c -> c.vmDefinition().assignment().map(Assignment::pool)
.map(p -> p.equals(event.fromPool())).orElse(false))
.filter(c -> c.vmDefinition().assignment().map(Assignment::user)
.map(u -> u.equals(event.toUser())).orElse(false))
.findFirst();
if (vmQuery.isPresent()) {
var vmDef = vmQuery.get().vmDefinition();
event.setResult(new VmData(vmDef, vmQuery.get()));
return;
}
public void onModifyVm(ModifyVm event, VmChannel channel)
throws ApiException, IOException {
patchVmDef(channel.client(), event.name(), "spec/vm/" + event.path(),
event.value());
}
// Get the pool definition for checking possible assignment
VmPool vmPool = newEventPipeline().fire(new GetPools()
.withName(event.fromPool())).get().stream().findFirst()
.orElse(null);
if (vmPool == null) {
return;
}
private void patchVmDef(K8sClient client, String name, String path,
Object value) throws ApiException, IOException {
var vmStub = K8sDynamicStub.get(client,
new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM), namespace(),
name);
// Find available VM.
vmQuery = channelManager.channels().stream()
.filter(c -> vmPool.isAssignable(c.vmDefinition()))
.sorted(Comparator.comparing((VmChannel c) -> c.vmDefinition()
.assignment().map(Assignment::lastUsed)
.orElse(Instant.ofEpochSecond(0)))
.thenComparing(preferRunning))
.findFirst();
// None found
if (vmQuery.isEmpty()) {
return;
}
// Assign to user
var chosenVm = vmQuery.get();
if (Optional.ofNullable(chosenVm.fire(new UpdateAssignment(
vmPool, event.toUser())).get()).orElse(false)) {
var vmDef = chosenVm.vmDefinition();
event.setResult(new VmData(vmDef, chosenVm));
// Make sure that a newly assigned VM is running.
chosenVm.pipeline().fire(new ModifyVm(vmDef.name(),
"state", "Running", chosenVm));
return;
}
// Patch running
String valueAsText = value instanceof String
? "\"" + value + "\""
: value.toString();
var res = vmStub.patch(V1Patch.PATCH_FORMAT_JSON_PATCH,
new V1Patch("[{\"op\": \"replace\", \"path\": \"/"
+ path + "\", \"value\": " + valueAsText + "}]"),
client.defaultPatchOptions());
if (!res.isPresent()) {
logger.warning(
() -> "Cannot patch definition for Vm " + vmStub.name());
}
}
private static Comparator<VmChannel> preferRunning
= new Comparator<>() {
@Override
public int compare(VmChannel ch1, VmChannel ch2) {
if (ch1.vmDefinition().conditionStatus("Running").orElse(false)
&& !ch2.vmDefinition().conditionStatus("Running")
.orElse(false)) {
return -1;
/**
* Attempt to Update the assignment information in the status of the
* VM CR. Returns true if successful. The handler does not attempt
* retries, because in case of failure it will be necessary to
* re-evaluate the chosen VM.
*
* @param event the event
* @param channel the channel
* @throws ApiException the api exception
*/
@Handler
public void onUpdatedAssignment(UpdateAssignment event, VmChannel channel)
throws ApiException {
try {
var vmDef = channel.vmDefinition();
var vmStub = VmDefinitionStub.get(channel.client(),
new GroupVersionKind(Crd.GROUP, "", Crd.KIND_VM),
vmDef.namespace(), vmDef.name());
if (vmStub.updateStatus(vmDef, from -> {
JsonObject status = from.statusJson();
if (event.toUser() == null) {
((JsonObject) GsonPtr.to(status).get())
.remove(Status.ASSIGNMENT);
} else {
var assignment = GsonPtr.to(status).to(Status.ASSIGNMENT);
assignment.set("pool", event.fromPool().name());
assignment.set("user", event.toUser());
assignment.set("lastUsed", Instant.now().toString());
}
return 0;
return status;
}).isPresent()) {
event.setResult(true);
}
};
} catch (ApiException e) {
// Log exceptions except for conflict, which can be expected
if (HttpURLConnection.HTTP_CONFLICT != e.getCode()) {
throw e;
}
}
event.setResult(false);
}
}

9
org.jdrupes.vmoperator.vmaccess/src/org/jdrupes/vmoperator/vmaccess/VmAccess.java
View file

@ -129,6 +129,7 @@ public class VmAccess extends FreeMarkerConlet<VmAccess.ResourceModel> {
private EventPipeline appPipeline;
private static ObjectMapper objectMapper
= new ObjectMapper().registerModule(new JavaTimeModule());
private Class<?> preferredIpVersion = Inet4Address.class;
private Set<String> syncUsers = Collections.emptySet();
private Set<String> syncRoles = Collections.emptySet();
@ -785,12 +786,12 @@ public class VmAccess extends FreeMarkerConlet<VmAccess.ResourceModel> {
switch (event.method()) {
case "start":
if (perms.contains(VmDefinition.Permission.START)) {
fire(new ModifyVm(vmName, "state", "Running", vmChannel));
vmChannel.fire(new ModifyVm(vmName, "state", "Running"));
}
break;
case "stop":
if (perms.contains(VmDefinition.Permission.STOP)) {
fire(new ModifyVm(vmName, "state", "Stopped", vmChannel));
vmChannel.fire(new ModifyVm(vmName, "state", "Stopped"));
}
break;
case "reset":
@ -800,7 +801,7 @@ public class VmAccess extends FreeMarkerConlet<VmAccess.ResourceModel> {
break;
case "resetConfirmed":
if (perms.contains(VmDefinition.Permission.RESET)) {
fire(new ResetVm(vmName), vmChannel);
vmChannel.fire(new ResetVm(vmName));
}
break;
case "openConsole":
@ -838,7 +839,7 @@ public class VmAccess extends FreeMarkerConlet<VmAccess.ResourceModel> {
}
var pwQuery = Event.onCompletion(new GetDisplaySecret(vmDef, user),
e -> gotPassword(channel, model, vmDef, e));
fire(pwQuery, vmChannel);
vmChannel.fire(pwQuery);
}
private void gotPassword(ConsoleConnection channel, ResourceModel model,

18
org.jdrupes.vmoperator.vmmgmt/src/org/jdrupes/vmoperator/vmmgmt/VmMgmt.java
View file

@ -423,12 +423,12 @@ public class VmMgmt extends FreeMarkerConlet<VmMgmt.VmsModel> {
switch (event.method()) {
case "start":
if (perms.contains(VmDefinition.Permission.START)) {
fire(new ModifyVm(vmName, "state", "Running", vmChannel));
vmChannel.fire(new ModifyVm(vmName, "state", "Running"));
}
break;
case "stop":
if (perms.contains(VmDefinition.Permission.STOP)) {
fire(new ModifyVm(vmName, "state", "Stopped", vmChannel));
vmChannel.fire(new ModifyVm(vmName, "state", "Stopped"));
}
break;
case "reset":
@ -438,22 +438,20 @@ public class VmMgmt extends FreeMarkerConlet<VmMgmt.VmsModel> {
break;
case "resetConfirmed":
if (perms.contains(VmDefinition.Permission.RESET)) {
fire(new ResetVm(vmName), vmChannel);
vmChannel.fire(new ResetVm(vmName));
}
break;
case "openConsole":
openConsole(channel, model, vmChannel, vmDef, user, perms);
break;
case "cpus":
fire(new ModifyVm(vmName, "currentCpus",
new BigDecimal(event.param(1).toString()).toBigInteger(),
vmChannel));
vmChannel.fire(new ModifyVm(vmName, "currentCpus",
new BigDecimal(event.param(1).toString()).toBigInteger()));
break;
case "ram":
fire(new ModifyVm(vmName, "currentRam",
vmChannel.fire(new ModifyVm(vmName, "currentRam",
new Quantity(new BigDecimal(event.param(1).toString()),
Format.BINARY_SI).toSuffixedString(),
vmChannel));
Format.BINARY_SI).toSuffixedString()));
break;
default:// ignore
break;
@ -488,7 +486,7 @@ public class VmMgmt extends FreeMarkerConlet<VmMgmt.VmsModel> {
}
var pwQuery = Event.onCompletion(new GetDisplaySecret(vmDef, user),
e -> gotPassword(channel, model, vmDef, e));
fire(pwQuery, vmChannel);
vmChannel.fire(pwQuery);
}
private void gotPassword(ConsoleConnection channel, VmsModel model,