High level topic mount/unmount APIs

src/v/redpanda/CMakeLists.txt

@@ -54,6 +54,8 @@ v_cc_library(
     admin/kafka.cc
     admin/util.cc
     admin/migrations.cc
+    admin/topics.cc
+    admin/data_migration_utils.cc
     cli_parser.cc
     application.cc
     monitor_unsafe_log_flag.cc

src/v/redpanda/admin/BUILD

@@ -115,6 +115,24 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "data_migration_utils",
+    srcs = [
+        "data_migration_utils.cc",
+    ],
+    hdrs = [
+        "data_migration_utils.h",
+    ],
+    include_prefix = "redpanda/admin",
+    visibility = ["//visibility:public"],
+    deps = [
+        ":migration",
+        "//src/v/cluster",
+        "//src/v/json",
+        "//src/v/model",
+    ],
+)
+
 redpanda_cc_library(
     name = "admin",
     srcs = [
@@ -125,6 +143,7 @@ redpanda_cc_library(
         "recovery.cc",
         "security.cc",
         "server.cc",
+        "topics.cc",
         "transaction.cc",
         "transform.cc",
         "usage.cc",
@@ -159,6 +178,7 @@ redpanda_cc_library(
         ":cluster_config",
         ":cluster_config_schema_util",
         ":config",
+        ":data_migration_utils",
         ":debug",
         ":features",
         ":hbadger",

src/v/redpanda/admin/api-doc/migration.def.json

@@ -147,4 +147,27 @@
             "type": "inbound_migration"
         }
     }
+},
+"mount_configuration": {
+    "type": "object",
+    "required": [
+        "topics"
+    ],
+    "properties": {
+        "topics": {
+            "type": "array",
+            "items": {
+                "type": "inbound_topic"
+            },
+            "description": "List of topics to mount"
+        }
+    }
+},
+"migration_info": {
+    "type": "object",
+    "properties": {
+        "id": {
+            "type": "int"
+        }
+    }
 }

src/v/redpanda/admin/api-doc/migration.json

@@ -89,4 +89,51 @@
             }
         ]
     }
+},
+"/v1/topics/mount": {
+    "post": {
+        "operationId": "mount_topics",
+        "summary": "Mount topic according to provided configuration",
+        "requestBody": {
+            "description": "Mount topic configuration",
+            "required": true,
+            "content": {
+                "application/json": {
+                    "schema": "mount_configuration"
+                }
+            }
+        },
+        "responses": {
+            "200": {
+                "description": "Underlying migration information",
+                "schema": "migration_info"
+            }
+        }
+    }
+},
+"/v1/topics/unmount": {
+    "post": {
+        "operationId": "unmount_topics",
+        "summary": "Unmount the topics provided in the list",
+        "requestBody": {
+            "description": "List of topics to unmount",
+            "required": true,
+            "content": {
+                "application/json": {
+                    "schema": {
+                        "type": "array",
+                        "items": {
+                            "type": "namespaced_topic"
+                        }
+                    }
+                }
+            }
+        },
+        "responses": {
+            "200": {
+                "description": "Underlying migration information",
+                "schema": "migration_info"
+            }
+        }
+    }
 }

src/v/redpanda/admin/data_migration_utils.cc

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2024 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+#include "redpanda/admin/data_migration_utils.h"
+
+#include "model/namespace.h"
+
+model::topic_namespace parse_topic_namespace(json::Value& json) {
+    if (json.HasMember("ns")) {
+        return {
+          model::ns(json["ns"].GetString()),
+          model::topic(json["topic"].GetString())};
+    } else {
+        return {
+          model::kafka_namespace, model::topic(json["topic"].GetString())};
+    }
+}
+
+cluster::data_migrations::inbound_topic parse_inbound_topic(json::Value& json) {
+    cluster::data_migrations::inbound_topic ret;
+    ret.source_topic_name = parse_topic_namespace(json["source_topic"]);
+    if (json.HasMember("alias")) {
+        ret.alias = parse_topic_namespace(json["alias"]);
+    }
+    return ret;
+}
+
+chunked_vector<cluster::data_migrations::inbound_topic>
+parse_inbound_topics(json::Value& json) {
+    chunked_vector<cluster::data_migrations::inbound_topic> ret;
+    auto topics_array = json["topics"].GetArray();
+    ret.reserve(topics_array.Size());
+    std::ranges::transform(
+      topics_array, std::back_inserter(ret), &parse_inbound_topic);
+
+    return ret;
+}
+
+chunked_vector<model::topic_namespace> parse_topics(json::Value& json) {
+    chunked_vector<model::topic_namespace> ret;
+    auto topics_array = json["topics"].GetArray();
+    ret.reserve(topics_array.Size());
+    std::ranges::transform(
+      topics_array, std::back_inserter(ret), &parse_topic_namespace);
+
+    return ret;
+}

src/v/redpanda/admin/data_migration_utils.h

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2024 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#pragma once
+#include "cluster/data_migration_types.h"
+#include "json/document.h"
+#include "model/metadata.h"
+
+model::topic_namespace parse_topic_namespace(json::Value& json);
+
+cluster::data_migrations::inbound_topic parse_inbound_topic(json::Value& json);
+
+chunked_vector<cluster::data_migrations::inbound_topic>
+parse_inbound_topics(json::Value& json);
+chunked_vector<model::topic_namespace> parse_topics(json::Value& json);

src/v/redpanda/admin/migrations.cc

@@ -19,6 +19,7 @@
 #include "model/metadata.h"
 #include "model/namespace.h"
 #include "redpanda/admin/api-doc/migration.json.hh"
+#include "redpanda/admin/data_migration_utils.h"
 #include "redpanda/admin/server.h"
 #include "redpanda/admin/util.h"
 #include "ssx/async_algorithm.h"
@@ -207,34 +208,10 @@ json::validator make_migration_validator() {
     return json::validator(schema);
 }
 
-model::topic_namespace parse_topic_namespace(json::Value& json) {
-    if (json.HasMember("ns")) {
-        return {
-          model::ns(json["ns"].GetString()),
-          model::topic(json["topic"].GetString())};
-    } else {
-        return {
-          model::kafka_namespace, model::topic(json["topic"].GetString())};
-    }
-}
-
-cluster::data_migrations::inbound_topic parse_inbound_topic(json::Value& json) {
-    cluster::data_migrations::inbound_topic ret;
-    ret.source_topic_name = parse_topic_namespace(json["source_topic"]);
-    if (json.HasMember("alias")) {
-        ret.alias = parse_topic_namespace(json["alias"]);
-    }
-    return ret;
-}
-
 cluster::data_migrations::inbound_migration
 parse_inbound_data_migration(json::Value& json) {
     cluster::data_migrations::inbound_migration ret;
-    auto topics_array = json["topics"].GetArray();
-    ret.topics.reserve(topics_array.Size());
-    std::ranges::transform(
-      topics_array, std::back_inserter(ret.topics), &parse_inbound_topic);
-
+    ret.topics = parse_inbound_topics(json);
     auto consumer_groups_array = json["consumer_groups"].GetArray();
     ret.groups.reserve(consumer_groups_array.Size());
     for (auto& group : consumer_groups_array) {
@@ -246,10 +223,7 @@ parse_inbound_data_migration(json::Value& json) {
 cluster::data_migrations::outbound_migration
 parse_outbound_data_migration(json::Value& json) {
     cluster::data_migrations::outbound_migration ret;
-    auto topics_array = json["topics"].GetArray();
-    ret.topics.reserve(topics_array.Size());
-    std::ranges::transform(
-      topics_array, std::back_inserter(ret.topics), &parse_topic_namespace);
+    ret.topics = parse_topics(json);
 
     auto consumer_groups_array = json["consumer_groups"].GetArray();
     ret.groups.reserve(consumer_groups_array.Size());

src/v/redpanda/admin/server.cc

@@ -408,6 +408,7 @@ void admin_server::configure_admin_routes() {
     register_shadow_indexing_routes();
     register_wasm_transform_routes();
     register_data_migration_routes();
+    register_topic_routes();
     /**
      * Special REST apis active only in recovery mode
      */

src/v/redpanda/admin/server.h

@@ -431,6 +431,7 @@ class admin_server {
     void register_wasm_transform_routes();
     void register_recovery_mode_routes();
     void register_data_migration_routes();
+    void register_topic_routes();
 
     ss::future<ss::json::json_return_type> patch_cluster_config_handler(
       std::unique_ptr<ss::http::request>, const request_auth_result&);
@@ -656,6 +657,12 @@ class admin_server {
     ss::future<ss::json::json_return_type>
       delete_migration(std::unique_ptr<ss::http::request>);
 
+    // Topic routes
+    ss::future<ss::json::json_return_type>
+      mount_topics(std::unique_ptr<ss::http::request>);
+    ss::future<ss::json::json_return_type>
+      unmount_topics(std::unique_ptr<ss::http::request>);
+
     ss::future<> throw_on_error(
       ss::http::request& req,
       std::error_code ec,