Add DumpMessages/AlterRole/UpdateUser interfaces

Signed-off-by: yhmo <yihua.mo@zilliz.com>

Author: yhmo

cmake/MilvusProtoGen.cmake

@@ -16,15 +16,14 @@
 
 include_guard(GLOBAL)
 
-set(PROTO_VERSION v2.6.13)
-set(PROTO_URL https://github.com/milvus-io/milvus-proto/archive/refs/tags/${PROTO_VERSION}.tar.gz)
-
+set(PROTO_COMMIT fd141a092113dd7a85ab6dc6c1b037b17a3bf96a)
 
 include(FetchContent)
 
 # download proto
 FetchContent_Declare(milvus_proto
-    URL         ${PROTO_URL}
+    GIT_REPOSITORY https://github.com/milvus-io/milvus-proto.git
+    GIT_TAG        ${PROTO_COMMIT}
 )
 FetchContent_Populate(milvus_proto)
 

examples/README.md

@@ -30,6 +30,7 @@ Examples for MilvusClient:
 
 Examples for MilvusClientV2:
 - `./cmake_build/examples/v2/sdk_array_v2`: example to show the usage of Array field.
+- `./cmake_build/examples/v2/sdk_cdc_v2`: example to show the usage of CDC interfaces including DumpMessages().
 - `./cmake_build/examples/v2/sdk_db_v2`: example to show the usage of databases.
 - `./cmake_build/examples/v2/sdk_default_value_v2`: example to show the usage of default value.
 - `./cmake_build/examples/v2/sdk_dml_v2`: example to show the usage of dml interfaces.

examples/src/v2/cdc.cpp

@@ -0,0 +1,121 @@
+// Licensed to the LF AI & Data foundation under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+#include <cstdint>
+#include <iostream>
+#include <limits>
+#include <string>
+#include <vector>
+
+#include "ExampleUtils.h"
+#include "milvus/MilvusClientV2.h"
+
+namespace {
+
+const std::string kClusterAUri = "http://localhost:19530";
+const std::string kClusterBUri = "http://localhost:29530";
+const std::string kClusterAId = "cdc-test-upstream";
+const std::string kClusterBId = "cdc-test-downstream";
+const int kPChannelNum = 16;
+
+std::vector<std::string>
+GeneratePChannels(const std::string& cluster_id) {
+    std::vector<std::string> pchannels;
+    pchannels.reserve(kPChannelNum);
+    for (int i = 0; i < kPChannelNum; ++i) {
+        pchannels.push_back(cluster_id + "-rootcoord-dml_" + std::to_string(i));
+    }
+    return pchannels;
+}
+
+void
+PrintDumpedMessage(const milvus::DumpedMessage& message) {
+    std::cout << "message_id=" << message.MessageID().ID() << ", wal_name=" << message.MessageID().WalName()
+              << ", payload_size=" << message.Payload().size() << std::endl;
+    if (!message.Properties().empty()) {
+        std::cout << "properties=";
+        util::PrintMap(message.Properties());
+    }
+}
+
+}  // namespace
+
+int
+main(int argc, char* argv[]) {
+    printf("Example start...\n");
+
+    auto cluster_a_client = milvus::MilvusClientV2::Create();
+    auto cluster_b_client = milvus::MilvusClientV2::Create();
+
+    auto status = cluster_a_client->Connect(milvus::ConnectParam{kClusterAUri});
+    util::CheckStatus("connect cluster A", status);
+
+    status = cluster_b_client->Connect(milvus::ConnectParam{kClusterBUri});
+    util::CheckStatus("connect cluster B", status);
+
+    milvus::MilvusCluster cluster_a;
+    cluster_a.WithClusterID(kClusterAId).WithUri(kClusterAUri).WithPChannels(GeneratePChannels(kClusterAId));
+
+    milvus::MilvusCluster cluster_b;
+    cluster_b.WithClusterID(kClusterBId).WithUri(kClusterBUri).WithPChannels(GeneratePChannels(kClusterBId));
+
+    milvus::CrossClusterTopology topology;
+    topology.WithSourceClusterID(kClusterAId).WithTargetClusterID(kClusterBId);
+
+    milvus::ReplicateConfiguration configuration;
+    configuration.AddCluster(std::move(cluster_a));
+    configuration.AddCluster(std::move(cluster_b));
+    configuration.AddCrossClusterTopology(std::move(topology));
+
+    milvus::UpdateReplicateConfigurationRequest update_request;
+    update_request.WithConfiguration(std::move(configuration));
+
+    status = cluster_a_client->UpdateReplicateConfiguration(update_request);
+    util::CheckStatus("update replicate configuration on cluster A", status);
+
+    status = cluster_b_client->UpdateReplicateConfiguration(update_request);
+    util::CheckStatus("update replicate configuration on cluster B", status);
+
+    const std::string target_pchannel = GeneratePChannels(kClusterBId).front();
+
+    milvus::GetReplicateInfoRequest replicate_info_request;
+    replicate_info_request.WithSourceClusterID(kClusterAId).WithTargetPChannel(target_pchannel);
+
+    milvus::GetReplicateInfoResponse replicate_info_response;
+    status = cluster_b_client->GetReplicateInfo(replicate_info_request, replicate_info_response);
+    util::CheckStatus("get replicate info from cluster B", status);
+
+    milvus::ReplicateMessageID start_message_id;
+    start_message_id.WithID(replicate_info_response.Checkpoint().MessageID().ID())
+        .WithWalName(replicate_info_response.Checkpoint().MessageID().WalName());
+
+    milvus::DumpMessagesRequest dump_request;
+    dump_request.WithPChannel(target_pchannel)
+        .WithStartMessageID(std::move(start_message_id))
+        .WithStartTimeTick(replicate_info_response.Checkpoint().TimeTick())
+        .WithEndTimeTick(std::numeric_limits<uint64_t>::max());
+
+    std::cout << "DumpMessages on pchannel: " << target_pchannel << std::endl;
+    status = cluster_a_client->DumpMessages(dump_request, [](const milvus::DumpedMessage& message) {
+        PrintDumpedMessage(message);
+        return milvus::Status::OK();
+    });
+    util::CheckStatus("dump messages from cluster A", status);
+
+    cluster_a_client->Disconnect();
+    cluster_b_client->Disconnect();
+    return 0;
+}

src/impl/MilvusClientV2Impl.cpp

@@ -2507,6 +2507,41 @@ MilvusClientV2Impl::GetReplicateInfo(const GetReplicateInfoRequest& request, Get
         pre, &MilvusConnection::GetReplicateInfo, post);
 }
 
+Status
+MilvusClientV2Impl::DumpMessages(const DumpMessagesRequest& request,
+                                 const std::function<Status(const DumpedMessage&)>& on_message) {
+    if (!on_message) {
+        return {StatusCode::INVALID_ARGUMENT, "DumpMessages callback cannot be empty"};
+    }
+
+    auto connection = connection_.GetConnection();
+    if (connection == nullptr) {
+        return {StatusCode::NOT_CONNECTED, "Connection is not created!"};
+    }
+
+    proto::milvus::DumpMessagesRequest rpc_request;
+    rpc_request.set_pchannel(request.PChannel());
+    rpc_request.mutable_start_message_id()->set_id(request.StartMessageID().ID());
+    const auto& wal_name_str = request.StartMessageID().WalName();
+    if (!wal_name_str.empty()) {
+        proto::common::WALName wal_name;
+        if (!proto::common::WALName_Parse(wal_name_str, &wal_name)) {
+            return {StatusCode::INVALID_ARGUMENT, "Unknown WAL name: " + wal_name_str};
+        }
+        rpc_request.mutable_start_message_id()->set_wal_name(wal_name);
+    }
+    rpc_request.set_start_timetick(request.StartTimeTick());
+    rpc_request.set_end_timetick(request.EndTimeTick());
+
+    auto callback = [&on_message](const proto::common::ImmutableMessage& rpc_message) {
+        DumpedMessage message;
+        ConvertImmutableMessage(rpc_message, message);
+        return on_message(message);
+    };
+
+    return connection->DumpMessages(rpc_request, GrpcOpts{}, callback);
+}
+
 Status
 MilvusClientV2Impl::CreateResourceGroup(const CreateResourceGroupRequest& request) {
     auto pre = [&request](proto::milvus::CreateResourceGroupRequest& rpc_request) {
@@ -2656,6 +2691,18 @@ MilvusClientV2Impl::UpdatePassword(const UpdatePasswordRequest& request) {
         pre, &MilvusConnection::UpdateCredential, nullptr);
 }
 
+Status
+MilvusClientV2Impl::UpdateUser(const UpdateUserRequest& request) {
+    auto pre = [&request](proto::milvus::UpdateCredentialRequest& rpc_request) {
+        rpc_request.set_username(request.UserName());
+        rpc_request.set_description(request.Description());
+        return Status::OK();
+    };
+
+    return connection_.Invoke<proto::milvus::UpdateCredentialRequest, proto::common::Status>(
+        pre, &MilvusConnection::UpdateCredential, nullptr);
+}
+
 Status
 MilvusClientV2Impl::DropUser(const DropUserRequest& request) {
     auto pre = [&request](proto::milvus::DeleteCredentialRequest& rpc_request) {
@@ -2680,6 +2727,8 @@ MilvusClientV2Impl::DescribeUser(const DescribeUserRequest& request, DescribeUse
         desc.SetName(request.UserName());
         if (rpc_response.results().size() > 0) {
             auto result = rpc_response.results().at(0);
+            desc.SetName(result.user().name());
+            desc.SetDescription(result.description());
             for (const auto& role : result.roles()) {
                 desc.AddRole(role.name());
             }
@@ -2719,6 +2768,18 @@ MilvusClientV2Impl::CreateRole(const CreateRoleRequest& request) {
         pre, &MilvusConnection::CreateRole, nullptr);
 }
 
+Status
+MilvusClientV2Impl::AlterRole(const AlterRoleRequest& request) {
+    auto pre = [&request](proto::milvus::AlterRoleRequest& rpc_request) {
+        rpc_request.set_role_name(request.RoleName());
+        rpc_request.set_description(request.Description());
+        return Status::OK();
+    };
+
+    return connection_.Invoke<proto::milvus::AlterRoleRequest, proto::common::Status>(pre, &MilvusConnection::AlterRole,
+                                                                                      nullptr);
+}
+
 Status
 MilvusClientV2Impl::DropRole(const DropRoleRequest& request) {
     auto pre = [&request](proto::milvus::DropRoleRequest& rpc_request) {
@@ -2740,13 +2801,36 @@ MilvusClientV2Impl::DescribeRole(const DescribeRoleRequest& request, DescribeRol
         return Status::OK();
     };
 
-    auto post = [&request, &response](const proto::milvus::SelectGrantResponse& rpc_response) {
+    auto post = [this, &request, &response](const proto::milvus::SelectGrantResponse& rpc_response) {
         RoleDesc desc;
         desc.SetName(request.RoleName());
         for (const auto& entity : rpc_response.entities()) {
             desc.AddGrantItem({entity.object().name(), entity.object_name(), entity.db_name(), entity.role().name(),
                                entity.grantor().user().name(), entity.grantor().privilege().name()});
         }
+
+        proto::milvus::SelectRoleRequest role_request;
+        role_request.set_include_user_info(false);
+        role_request.mutable_role()->set_name(request.RoleName());
+        proto::milvus::SelectRoleResponse role_response;
+
+        auto connection = connection_.GetConnection();
+        if (connection == nullptr) {
+            return Status{StatusCode::NOT_CONNECTED, "Connection is not created!"};
+        }
+        auto retry_param = connection_.GetRetryParam();
+        auto caller = [&]() {
+            return connection->SelectRole(role_request, role_response, GrpcOpts{connection_.GetRpcDeadlineMs()});
+        };
+        auto status = Retry(caller, retry_param);
+        if (!status.IsOk()) {
+            return status;
+        }
+        if (role_response.results_size() > 0) {
+            desc.SetName(role_response.results(0).role().name());
+            desc.SetDescription(role_response.results(0).role().description());
+        }
+
         response.SetDesc(std::move(desc));
         return Status::OK();
     };

src/impl/MilvusClientV2Impl.h

@@ -268,6 +268,10 @@ class MilvusClientV2Impl : public MilvusClientV2, public std::enable_shared_from
     Status
     GetReplicateInfo(const GetReplicateInfoRequest& request, GetReplicateInfoResponse& response) final;
 
+    Status
+    DumpMessages(const DumpMessagesRequest& request,
+                 const std::function<Status(const DumpedMessage&)>& on_message) final;
+
     Status
     CreateResourceGroup(const CreateResourceGroupRequest& request) final;
 
@@ -295,6 +299,9 @@ class MilvusClientV2Impl : public MilvusClientV2, public std::enable_shared_from
     Status
     UpdatePassword(const UpdatePasswordRequest& request) final;
 
+    Status
+    UpdateUser(const UpdateUserRequest& request) final;
+
     Status
     DropUser(const DropUserRequest& request) final;
 
@@ -307,6 +314,9 @@ class MilvusClientV2Impl : public MilvusClientV2, public std::enable_shared_from
     Status
     CreateRole(const CreateRoleRequest& request) final;
 
+    Status
+    AlterRole(const AlterRoleRequest& request) final;
+
     Status
     DropRole(const DropRoleRequest& request) final;
 

src/impl/MilvusConnection.cpp

@@ -683,6 +683,55 @@ MilvusConnection::GetReplicateInfo(const proto::milvus::GetReplicateInfoRequest&
     return grpcCall("GetReplicateInfo", &Stub::GetReplicateInfo, request, response, options);
 }
 
+Status
+MilvusConnection::DumpMessages(const proto::milvus::DumpMessagesRequest& request, const GrpcContextOptions& options,
+                               const std::function<Status(const proto::common::ImmutableMessage&)>& on_message) {
+    std::shared_ptr<proto::milvus::MilvusService::Stub> stub;
+    {
+        std::lock_guard<std::mutex> lock(stub_mtx_);
+        stub = stub_;
+    }
+    if (stub == nullptr) {
+        return {StatusCode::NOT_CONNECTED, "Connection is not ready!"};
+    }
+
+    ::grpc::ClientContext context;
+    if (options.timeout > 0) {
+        auto deadline = std::chrono::system_clock::now() + std::chrono::milliseconds{options.timeout};
+        context.set_deadline(deadline);
+    }
+
+    auto reader = stub->DumpMessages(&context, request);
+    proto::milvus::DumpMessagesResponse response;
+    while (reader->Read(&response)) {
+        switch (response.response_case()) {
+            case proto::milvus::DumpMessagesResponse::kStatus: {
+                auto status = StatusByProtoResponse(response.status());
+                if (!status.IsOk()) {
+                    return status;
+                }
+                break;
+            }
+            case proto::milvus::DumpMessagesResponse::kMessage: {
+                auto status = on_message(response.message());
+                if (!status.IsOk()) {
+                    return status;
+                }
+                break;
+            }
+            case proto::milvus::DumpMessagesResponse::RESPONSE_NOT_SET:
+            default:
+                return {StatusCode::SERVER_FAILED, "Unexpected empty dumpMessages response"};
+        }
+    }
+
+    auto grpc_status = reader->Finish();
+    if (!grpc_status.ok()) {
+        return StatusCodeFromGrpcStatus(grpc_status);
+    }
+    return Status::OK();
+}
+
 Status
 MilvusConnection::SelectUser(const proto::milvus::SelectUserRequest& request,
                              proto::milvus::SelectUserResponse& response, const GrpcContextOptions& options) {
@@ -707,6 +756,12 @@ MilvusConnection::CreateRole(const proto::milvus::CreateRoleRequest& request, pr
     return grpcCall("CreateRole", &Stub::CreateRole, request, response, options);
 }
 
+Status
+MilvusConnection::AlterRole(const proto::milvus::AlterRoleRequest& request, proto::common::Status& response,
+                            const GrpcContextOptions& options) {
+    return grpcCall("AlterRole", &Stub::AlterRole, request, response, options);
+}
+
 Status
 MilvusConnection::DropRole(const proto::milvus::DropRoleRequest& request, proto::common::Status& response,
                            const GrpcContextOptions& options) {