feat: agent resource creation with web of things

build.gradle.kts

@@ -174,6 +174,8 @@ dependencies {
 
     implementation("org.semver4j:semver4j:5.3.0")
 
+    implementation("org.eclipse.ditto:ditto-wot-model:3.6.0")
+
     testImplementation("io.javaoperatorsdk:operator-framework-spring-boot-starter-test:5.5.0") {
         exclude(group = "org.apache.logging.log4j", module = "log4j-slf4j2-impl")
     }

src/main/helm/templates/clusterrole.yaml

@@ -16,4 +16,19 @@ rules:
       {{- range .Values.customResources }}
       - "{{ . }}"
       {{- end }}
-    verbs: ["get", "list", "create", "update", "delete", "watch", "patch"]
+    verbs: ["get", "list", "create", "update", "delete", "watch", "patch"]
+
+  # Add permission to access deployments in the apps API group
+  - apiGroups: [ "apps" ]
+    resources: [ "deployments" ]
+    verbs: ["get", "list", "watch", "create", "update", "patch", "delete"]
+
+  # Add permission to access pods
+  - apiGroups: [ "" ]
+    resources: [ "pods" ]
+    verbs: [ "get", "list", "watch" ]
+
+  # Add permission to access services
+  - apiGroups: [ "" ]
+    resources: [ "services" ]
+    verbs: [ "get", "list", "watch" ]

src/main/java/ai/ancf/lmos/operator/reconciler/AgentReconciler.java

@@ -6,28 +6,63 @@
 
 package ai.ancf.lmos.operator.reconciler;
 
-import io.javaoperatorsdk.operator.api.reconciler.Context;
-import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
-import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
-import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
-import ai.ancf.lmos.operator.resources.agent.AgentResource;
+import ai.ancf.lmos.operator.service.AgentDeploymentStatusService;
+import ai.ancf.lmos.operator.service.AgentServiceQuery;
+import ai.ancf.lmos.operator.service.KubernetesResourceManager;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.javaoperatorsdk.operator.api.reconciler.*;
+import org.eclipse.ditto.wot.model.ThingDescription;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
 
+import java.util.concurrent.TimeUnit;
 
+/**
+ * Reconciles Deployment resources by watching associated Pods and registering Services.
+ */
 @Component
-@ControllerConfiguration
-public class AgentReconciler implements Reconciler<AgentResource> {
+@ControllerConfiguration(labelSelector = "wot-agent=true")
+public class AgentReconciler implements Reconciler<Deployment>, Cleaner<Deployment> {
 
     private static final Logger LOG = LoggerFactory.getLogger(AgentReconciler.class);
 
+    private final AgentServiceQuery agentServiceQuery;
+    private final AgentDeploymentStatusService agentDeploymentStatusService;
+    private final KubernetesResourceManager kubernetesResourceManager;
+
+    public AgentReconciler(AgentServiceQuery agentServiceQuery, AgentDeploymentStatusService agentDeploymentStatusService, KubernetesResourceManager kubernetesResourceManager) {
+        this.agentServiceQuery = agentServiceQuery;
+        this.agentDeploymentStatusService = agentDeploymentStatusService;
+        this.kubernetesResourceManager = kubernetesResourceManager;
+    }
 
     @Override
-    public UpdateControl<AgentResource> reconcile(AgentResource agentResource, Context context) {
-        // TODO: fill in logic
-        LOG.debug("Agent reconcile");
+    public UpdateControl<Deployment> reconcile(Deployment deployment, Context context) {
+
+        boolean isDeploymentReady = agentDeploymentStatusService.isDeploymentReady(deployment);
+
+        LOG.info("is Deployment {} ready: {}", deployment.getMetadata().getName(), isDeploymentReady);
 
-        return UpdateControl.noUpdate();
+        if(isDeploymentReady) {
+            try {
+                String serviceUrl = kubernetesResourceManager.getServiceUrl(deployment);
+                ThingDescription thingDescription = agentServiceQuery.queryAgentService(serviceUrl);
+                    kubernetesResourceManager.createOrUpdateAgentResource(thingDescription, deployment);
+                    return UpdateControl.noUpdate();
+            } catch (Exception e) {
+                LOG.error("Error processing td for deployment: {}", deployment.getMetadata().getName(), e);
+                return UpdateControl.<Deployment>noUpdate().rescheduleAfter(10, TimeUnit.SECONDS);
+            }
+        }
+
+        return UpdateControl.<Deployment>noUpdate().rescheduleAfter(10, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public DeleteControl cleanup(Deployment deployment, Context<Deployment> context) {
+        LOG.info("Trigger AgentResource Deletion for deployment: {}", deployment.getMetadata().getName());
+        kubernetesResourceManager.deleteAgentResource(deployment);
+        return DeleteControl.defaultDelete();
     }
 }

src/main/java/ai/ancf/lmos/operator/resources/agent/AgentSpec.java

@@ -14,6 +14,7 @@ public class AgentSpec {
     private Set<String> supportedTenants;
     private Set<String> supportedChannels;
     private Set<ProvidedCapability> providedCapabilities;
+    private String thingDescription;
 
     public AgentSpec() {
     }
@@ -58,6 +59,14 @@ public void setProvidedCapabilities(Set<ProvidedCapability> providedCapabilities
         this.providedCapabilities = providedCapabilities;
     }
 
+    public String getThingDescription() {
+        return thingDescription;
+    }
+
+    public void setThingDescription(String thingDescription) {
+        this.thingDescription = thingDescription;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

src/main/java/ai/ancf/lmos/operator/service/AgentDeploymentStatusService.java

@@ -0,0 +1,31 @@
+/*
+ * SPDX-FileCopyrightText: 2024 Deutsche Telekom AG
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package ai.ancf.lmos.operator.service;
+
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@org.springframework.stereotype.Service
+public class AgentDeploymentStatusService {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AgentDeploymentStatusService.class);
+
+    public boolean isDeploymentReady(Deployment deployment) {
+        String deploymentName = deployment.getMetadata().getName();
+        String deploymentNamespace = deployment.getMetadata().getNamespace();
+        Integer replicas = deployment.getStatus().getReplicas();
+        Integer availableReplicas = deployment.getStatus().getAvailableReplicas();
+        Integer desiredReplicas = deployment.getSpec().getReplicas();
+
+        LOG.info(
+            "Reconciling Deployment: {} in namespace: {}, Replicas, availableReplicas: {}, desiredReplicas: {}",
+            deploymentName, deploymentNamespace, availableReplicas, desiredReplicas);
+
+        return (replicas != null && availableReplicas != null && replicas.equals(desiredReplicas) && availableReplicas.equals(desiredReplicas));
+    }
+}

src/main/java/ai/ancf/lmos/operator/service/AgentServiceQuery.java

@@ -0,0 +1,53 @@
+/*
+ * SPDX-FileCopyrightText: 2024 Deutsche Telekom AG
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package ai.ancf.lmos.operator.service;
+
+import org.eclipse.ditto.json.JsonObject;
+import org.eclipse.ditto.wot.model.ThingDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.client.WebClient;
+
+@org.springframework.stereotype.Service
+public class AgentServiceQuery {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AgentServiceQuery.class);
+
+    private final WebClient webClient;
+
+    public AgentServiceQuery() {
+        this.webClient = WebClient.builder().build();
+    }
+
+    public ThingDescription queryAgentService(String serviceUrl) {
+        LOG.info("Querying Agent for TD: {}", serviceUrl);
+        String thingDescriptionJson = webClient.get()
+                .uri(serviceUrl)
+                .accept(MediaType.APPLICATION_JSON)
+                .retrieve()
+                .bodyToMono(String.class).block();
+
+        if (thingDescriptionJson == null || thingDescriptionJson.isEmpty()) {
+            throw new IllegalStateException("TD Response body from agent is empty");
+        }
+
+        ThingDescription thingDescription = ThingDescription.fromJson(JsonObject.of(thingDescriptionJson));
+        validateTD(thingDescription);
+        return thingDescription;
+    }
+
+    private void validateTD(ThingDescription thingDescription) {
+        if (thingDescription == null) {
+            throw new RuntimeException("ThingDescription is null");
+        }
+
+        thingDescription.getDescription()
+                .filter(desc -> !desc.isEmpty())
+                .orElseThrow(() -> new RuntimeException("Description of TD is not present or is empty"));
+    }
+}

src/main/java/ai/ancf/lmos/operator/service/KubernetesResourceManager.java

@@ -0,0 +1,97 @@
+/*
+ * SPDX-FileCopyrightText: 2024 Deutsche Telekom AG
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package ai.ancf.lmos.operator.service;
+
+import ai.ancf.lmos.operator.resources.agent.AgentResource;
+import ai.ancf.lmos.operator.resources.agent.AgentSpec;
+import io.fabric8.kubernetes.api.model.*;
+import io.fabric8.kubernetes.api.model.apps.Deployment;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.eclipse.ditto.wot.model.ThingDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+
+@org.springframework.stereotype.Service
+public class KubernetesResourceManager {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KubernetesResourceManager.class);
+
+    private final KubernetesClient kubernetesClient;
+
+    public KubernetesResourceManager(KubernetesClient kubernetesClient) {
+        this.kubernetesClient = kubernetesClient;
+    }
+
+    public void createOrUpdateAgentResource(ThingDescription thingDescription, Deployment deployment) {
+        AgentResource agentResource = new AgentResource();
+        agentResource.setMetadata(new ObjectMetaBuilder()
+            .withName(deployment.getMetadata().getName())
+            .withNamespace(deployment.getMetadata().getNamespace())
+            .build());
+
+        AgentSpec spec = new AgentSpec();
+        spec.setDescription(thingDescription.getDescription().get()
+                .toString());
+        spec.setThingDescription(thingDescription.toJsonString());
+
+        agentResource.setSpec(spec);
+
+        AgentResource agentResourceCreated = kubernetesClient.resources(AgentResource.class)
+                .inNamespace(deployment.getMetadata().getNamespace())
+                .withName(deployment.getMetadata().getName())
+                .createOrReplace(agentResource);
+
+        LOG.info("AgentResource {} created/updated for deployment: {}", agentResourceCreated.getFullResourceName(), deployment.getMetadata().getName());
+    }
+
+    public Service findService(Deployment deployment) {
+        Map<String, String> selectorLabels = deployment.getSpec().getSelector().getMatchLabels();
+        String deploymentNamespace = deployment.getMetadata().getNamespace();
+        ServiceList serviceList = kubernetesClient.services()
+                .inNamespace(deploymentNamespace)
+                .withLabels(selectorLabels)
+                .list();
+        if (serviceList.getItems().size() != 1) {
+            LOG.error("Expected exactly one service, but got {}, {}", serviceList.getItems().size(), serviceList.getItems());
+            throw new IllegalStateException("Expected exactly one service, but got " + serviceList.getItems().size());
+        }
+        return serviceList
+                .getItems()
+                .getFirst();
+    }
+
+    public String getBaseServiceUrl(Service service) {
+        ServicePort servicePort = service.getSpec().getPorts().getFirst();
+        String port = servicePort.getPort().toString();
+        boolean isHttps = servicePort.getPort() == 443 || "https".equalsIgnoreCase(servicePort.getName());
+        String protocol = isHttps ? "https://" : "http://";
+        String url = protocol + service.getMetadata().getName() + ":" + port;
+        LOG.info("Service URL is: {}", url);
+        return url;
+    }
+
+    public String getServiceUrl(Deployment deployment) {
+        String agentPath = deployment.getMetadata().getAnnotations().getOrDefault("wot.w3.org/td-endpoint", ".well-known/wot");
+        String baseServiceUrl = getBaseServiceUrl(findService(deployment));
+        if (agentPath.startsWith("/")) {
+            return baseServiceUrl + agentPath;
+        } else {
+            return baseServiceUrl + "/" + agentPath;
+        }
+    }
+
+    public void deleteAgentResource(Deployment deployment) {
+        List<StatusDetails> deleteStatus = kubernetesClient.resources(AgentResource.class)
+                .inNamespace(deployment.getMetadata().getNamespace())
+                .withName(deployment.getMetadata().getName())
+                .delete();
+        LOG.info("AgentResource {} deleted for deployment: {}", deleteStatus, deployment.getMetadata().getName());
+    }
+}