Apps: implement force deletion and improve status (#714)

helm/crds/applications.langstream.ai-v1.yml

@@ -27,6 +27,8 @@ spec:
                 type: string
               imagePullPolicy:
                 type: string
+              options:
+                type: string
               tenant:
                 type: string
             required:

langstream-admin-client/src/main/java/ai/langstream/admin/client/AdminClient.java

@@ -292,8 +292,8 @@ public void update(String application, MultiPartBodyPublisher multiPartBodyPubli
 
         @Override
         @SneakyThrows
-        public void delete(String application) {
-            http(newDelete(tenantAppPath("/" + application)));
+        public void delete(String application, boolean force) {
+            http(newDelete(tenantAppPath("/" + application + "?force=" + force)));
         }
 
         @Override

langstream-admin-client/src/main/java/ai/langstream/admin/client/model/Applications.java

@@ -29,7 +29,7 @@ String deploy(
 
     void update(String application, MultiPartBodyPublisher multiPartBodyPublisher);
 
-    void delete(String application);
+    void delete(String application, boolean force);
 
     String get(String application, boolean stats);
 

langstream-api/src/main/java/ai/langstream/api/storage/ApplicationStore.java

@@ -49,7 +49,7 @@ void put(
 
     Secrets getSecrets(String tenant, String applicationId);
 
-    void delete(String tenant, String applicationId);
+    void delete(String tenant, String applicationId, boolean force);
 
     Map<String, StoredApplication> list(String tenant);
 

langstream-cli/src/main/java/ai/langstream/cli/commands/applications/DeleteApplicationCmd.java

@@ -18,16 +18,28 @@
 import lombok.SneakyThrows;
 import picocli.CommandLine;
 
-@CommandLine.Command(name = "delete", header = "Delete an application")
+@CommandLine.Command(
+        name = "delete",
+        header = "Delete an application. The deletion of the application it's asynchronous.")
 public class DeleteApplicationCmd extends BaseApplicationCmd {
 
     @CommandLine.Parameters(description = "ID of the application")
     private String applicationId;
 
+    @CommandLine.Option(
+            names = {"-f", "--force"},
+            description =
+                    "Force application deletion. This could cause orphaned assets and topics.")
+    private boolean force;
+
     @Override
     @SneakyThrows
     public void run() {
-        getClient().applications().delete(applicationId);
-        log(String.format("Application %s deleted", applicationId));
+        getClient().applications().delete(applicationId, force);
+        if (force) {
+            log(String.format("Application '%s' marked for deletion (forced)", applicationId));
+        } else {
+            log(String.format("Application '%s' marked for deletion", applicationId));
+        }
     }
 }

langstream-cli/src/test/java/ai/langstream/cli/commands/applications/AppsCmdTest.java

@@ -352,13 +352,25 @@ public void testGet() throws Exception {
     @Test
     public void testDelete() {
         wireMock.register(
-                WireMock.delete(String.format("/api/applications/%s/my-app", TENANT))
+                WireMock.delete(String.format("/api/applications/%s/my-app?force=false", TENANT))
                         .willReturn(WireMock.ok()));
 
         CommandResult result = executeCommand("apps", "delete", "my-app");
         Assertions.assertEquals(0, result.exitCode());
         Assertions.assertEquals("", result.err());
-        Assertions.assertEquals("Application my-app deleted", result.out());
+        Assertions.assertEquals("Application 'my-app' marked for deletion", result.out());
+    }
+
+    @Test
+    public void testForceDelete() {
+        wireMock.register(
+                WireMock.delete(String.format("/api/applications/%s/my-app?force=true", TENANT))
+                        .willReturn(WireMock.ok()));
+
+        CommandResult result = executeCommand("apps", "delete", "my-app", "-f");
+        Assertions.assertEquals(0, result.exitCode());
+        Assertions.assertEquals("", result.err());
+        Assertions.assertEquals("Application 'my-app' marked for deletion (forced)", result.out());
     }
 
     @Test

langstream-e2e-tests/src/test/java/ai/langstream/tests/AppLifecycleIT.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright DataStax, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package ai.langstream.tests;
+
+import ai.langstream.tests.util.BaseEndToEndTest;
+import ai.langstream.tests.util.TestSuites;
+import java.io.File;
+import java.nio.file.Files;
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@Slf4j
+@ExtendWith(BaseEndToEndTest.class)
+@Tag(TestSuites.CATEGORY_OTHER)
+public class AppLifecycleIT extends BaseEndToEndTest {
+
+    @Test
+    public void testDeleteBrokenApplication() throws Exception {
+        installLangStreamCluster(true);
+        final String tenant = "ten-" + System.currentTimeMillis();
+        setupTenant(tenant);
+        final String applicationId = "my-test-app";
+
+        final Map<String, Map<String, Object>> instanceContent =
+                Map.of(
+                        "instance",
+                        Map.of(
+                                "streamingCluster",
+                                Map.of(
+                                        "type",
+                                        "kafka",
+                                        "configuration",
+                                        Map.of("bootstrapServers", "wrong:9092")),
+                                "computeCluster",
+                                Map.of("type", "kubernetes")));
+
+        final File instanceFile = Files.createTempFile("ls-test", ".yaml").toFile();
+        YAML_MAPPER.writeValue(instanceFile, instanceContent);
+
+        deployLocalApplication(
+                tenant, false, applicationId, "python-processor", instanceFile, Map.of());
+        awaitApplicationInStatus(applicationId, "ERROR_DEPLOYING");
+        executeCommandOnClient(
+                "bin/langstream apps delete %s ".formatted(applicationId).split(" "));
+        awaitApplicationInStatus(applicationId, "ERROR_DELETING");
+        executeCommandOnClient(
+                "bin/langstream apps delete -f %s".formatted(applicationId).split(" "));
+        awaitApplicationCleanup(tenant, applicationId);
+    }
+}

langstream-e2e-tests/src/test/java/ai/langstream/tests/util/BaseEndToEndTest.java

@@ -893,7 +893,8 @@ protected static void dumpPodLogs(String podName, String namespace, String fileP
                     final File outputFile =
                             new File(
                                     TEST_LOGS_DIR,
-                                    "%s.%s.%s.log".formatted(filePrefix, podName, container));
+                                    "%s.%s.%s.%s.log"
+                                            .formatted(filePrefix, namespace, podName, container));
                     try (FileWriter writer = new FileWriter(outputFile)) {
                         writer.write(logs);
                     } catch (IOException e) {
@@ -1070,6 +1071,37 @@ protected static boolean isApplicationReady(
                 expectedRunningTotalExecutors + "/" + expectedRunningTotalExecutors);
     }
 
+    protected static void awaitApplicationInStatus(String applicationId, String status) {
+        Awaitility.await()
+                .atMost(3, TimeUnit.MINUTES)
+                .pollInterval(5, TimeUnit.SECONDS)
+                .until(() -> isApplicationInStatus(applicationId, status));
+    }
+
+    protected static boolean isApplicationInStatus(String applicationId, String expectedStatus) {
+        final String response =
+                executeCommandOnClient(
+                        "bin/langstream apps get %s".formatted(applicationId).split(" "));
+        final List<String> lines = response.lines().collect(Collectors.toList());
+        final String appLine = lines.get(1);
+        final List<String> lineAsList =
+                Arrays.stream(appLine.split(" "))
+                        .filter(s -> !s.isBlank())
+                        .collect(Collectors.toList());
+        final String status = lineAsList.get(3);
+        if (status != null && status.equals(expectedStatus)) {
+            return true;
+        }
+        log.info(
+                "application {} is not in expected status {} but is in {}, dumping:",
+                applicationId,
+                expectedStatus,
+                status);
+        executeCommandOnClient(
+                "bin/langstream apps get %s -o yaml".formatted(applicationId).split(" "));
+        return false;
+    }
+
     @SneakyThrows
     protected static void deployLocalApplicationAndAwaitReady(
             String tenant,
@@ -1100,6 +1132,58 @@ private static void deployLocalApplicationAndAwaitReady(
             String appDirName,
             Map<String, String> env,
             int expectedNumExecutors) {
+        final String tenantNamespace = TENANT_NAMESPACE_PREFIX + tenant;
+        final String podUids =
+                deployLocalApplication(
+                        tenant, isUpdate, applicationId, appDirName, instanceFile, env);
+
+        awaitApplicationReady(applicationId, expectedNumExecutors);
+        Awaitility.await()
+                .atMost(2, TimeUnit.MINUTES)
+                .pollDelay(Duration.ZERO)
+                .pollInterval(5, TimeUnit.SECONDS)
+                .untilAsserted(
+                        () -> {
+                            final List<Pod> pods =
+                                    client.pods()
+                                            .inNamespace(tenantNamespace)
+                                            .withLabels(
+                                                    Map.of(
+                                                            "langstream-application",
+                                                            applicationId,
+                                                            "app",
+                                                            "langstream-runtime"))
+                                            .list()
+                                            .getItems();
+                            log.info(
+                                    "waiting new executors to be ready, found {}, expected {}",
+                                    pods.size(),
+                                    expectedNumExecutors);
+                            if (pods.size() != expectedNumExecutors) {
+                                fail("too many pods: " + pods.size());
+                            }
+                            final String currentUids =
+                                    pods.stream()
+                                            .map(p -> p.getMetadata().getUid())
+                                            .sorted()
+                                            .collect(Collectors.joining(","));
+                            Assertions.assertNotEquals(podUids, currentUids);
+                            for (Pod pod : pods) {
+                                log.info("checking pod readiness {}", pod.getMetadata().getName());
+                                assertTrue(checkPodReadiness(pod));
+                            }
+                        });
+    }
+
+    @SneakyThrows
+    protected static String deployLocalApplication(
+            String tenant,
+            boolean isUpdate,
+            String applicationId,
+            String appDirName,
+            File instanceFile,
+            Map<String, String> env) {
+        final String tenantNamespace = TENANT_NAMESPACE_PREFIX + tenant;
         String testAppsBaseDir = "src/test/resources/apps";
         String testSecretBaseDir = "src/test/resources/secrets";
 
@@ -1126,7 +1210,6 @@ private static void deployLocalApplicationAndAwaitReady(
                             .collect(Collectors.joining(" && "));
             beforeCmd += " && ";
         }
-        final String tenantNamespace = TENANT_NAMESPACE_PREFIX + tenant;
 
         final String podUids;
         if (isUpdate) {
@@ -1153,43 +1236,7 @@ private static void deployLocalApplicationAndAwaitReady(
                 "bin/langstream apps %s %s -app /tmp/app -i /tmp/instance.yaml -s /tmp/secrets.yaml"
                         .formatted(isUpdate ? "update" : "deploy", applicationId);
         executeCommandOnClient((beforeCmd + command).split(" "));
-
-        awaitApplicationReady(applicationId, expectedNumExecutors);
-        Awaitility.await()
-                .atMost(2, TimeUnit.MINUTES)
-                .pollDelay(Duration.ZERO)
-                .pollInterval(5, TimeUnit.SECONDS)
-                .untilAsserted(
-                        () -> {
-                            final List<Pod> pods =
-                                    client.pods()
-                                            .inNamespace(tenantNamespace)
-                                            .withLabels(
-                                                    Map.of(
-                                                            "langstream-application",
-                                                            applicationId,
-                                                            "app",
-                                                            "langstream-runtime"))
-                                            .list()
-                                            .getItems();
-                            log.info(
-                                    "waiting new executors to be ready, found {}, expected {}",
-                                    pods.size(),
-                                    expectedNumExecutors);
-                            if (pods.size() != expectedNumExecutors) {
-                                fail("too many pods: " + pods.size());
-                            }
-                            final String currentUids =
-                                    pods.stream()
-                                            .map(p -> p.getMetadata().getUid())
-                                            .sorted()
-                                            .collect(Collectors.joining(","));
-                            Assertions.assertNotEquals(podUids, currentUids);
-                            for (Pod pod : pods) {
-                                log.info("checking pod readiness {}", pod.getMetadata().getName());
-                                assertTrue(checkPodReadiness(pod));
-                            }
-                        });
+        return podUids;
     }
 
     private static void validateApp(File appDir, File secretFile) throws Exception {
@@ -1265,10 +1312,10 @@ protected static String generateControlPlaneToken(String subject) {
     protected void deleteAppAndAwaitCleanup(String tenant, String applicationId) {
         executeCommandOnClient("bin/langstream apps delete %s".formatted(applicationId).split(" "));
 
-        awaitCleanup(tenant, applicationId);
+        awaitApplicationCleanup(tenant, applicationId);
     }
 
-    private static void awaitCleanup(String tenant, String applicationId) {
+    protected static void awaitApplicationCleanup(String tenant, String applicationId) {
         final String tenantNamespace = TENANT_NAMESPACE_PREFIX + tenant;
         Awaitility.await()
                 .atMost(2, TimeUnit.MINUTES)

langstream-k8s-deployer/langstream-k8s-deployer-api/src/main/java/ai/langstream/deployer/k8s/api/crds/apps/ApplicationSpec.java

@@ -18,7 +18,6 @@
 import ai.langstream.deployer.k8s.api.crds.NamespacedSpec;
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.SneakyThrows;
@@ -42,6 +41,19 @@ public static SerializedApplicationInstance deserializeApplication(
         return mapper.readValue(serializedApplicationInstance, SerializedApplicationInstance.class);
     }
 
+    @SneakyThrows
+    public static ApplicationSpecOptions deserializeOptions(String options) {
+        if (options == null) {
+            return new ApplicationSpecOptions();
+        }
+        return mapper.readValue(options, ApplicationSpecOptions.class);
+    }
+
+    @SneakyThrows
+    public static String serializeOptions(ApplicationSpecOptions options) {
+        return mapper.writeValueAsString(options);
+    }
+
     @Deprecated private String image;
     @Deprecated private String imagePullPolicy;
 
@@ -52,18 +64,5 @@ public static SerializedApplicationInstance deserializeApplication(
     private String application;
 
     private String codeArchiveId;
-
-    @Builder
-    public ApplicationSpec(
-            String tenant,
-            String image,
-            String imagePullPolicy,
-            String application,
-            String codeArchiveId) {
-        super(tenant);
-        this.image = image;
-        this.imagePullPolicy = imagePullPolicy;
-        this.application = application;
-        this.codeArchiveId = codeArchiveId;
-    }
+    private String options;
 }