packages/iSulad
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
iSulad/0040-add-support-for-GetContainerEvents.patch

2602 lines
96 KiB
Diff
Raw Permalink Normal View History

upgrade from upstream Signed-off-by: jikai <jikai11@huawei.com>
2024-04-02 11:50:19 +00:00
From 745497bdc5c5192709ecc7b3edc91a5170f5b30e Mon Sep 17 00:00:00 2001
From: jikai <jikai11@huawei.com>
Date: Fri, 29 Mar 2024 09:33:38 +0000
upgrade from upstream
2024-04-20 10:00:53 +08:00
Subject: [PATCH 40/69] add support for GetContainerEvents
upgrade from upstream Signed-off-by: jikai <jikai11@huawei.com>
2024-04-02 11:50:19 +00:00
Signed-off-by: jikai <jikai11@huawei.com>
---
src/daemon/CMakeLists.txt | 3 +
src/daemon/common/cri/v1/v1_cri_helpers.cc | 205 +++++++++++++++
src/daemon/common/cri/v1/v1_cri_helpers.h | 5 +
.../{entry => common}/cri/v1/v1_naming.cc | 0
.../{entry => common}/cri/v1/v1_naming.h | 0
src/daemon/config/isulad_config.c | 1 +
.../entry/connect/grpc/cri/cri_service.cc | 3 +-
.../entry/connect/grpc/cri/cri_service.h | 1 +
.../cri/v1/cri_v1_runtime_runtime_service.cc | 147 ++++++++++-
.../cri/v1/cri_v1_runtime_runtime_service.h | 11 +-
src/daemon/entry/connect/grpc/grpc_service.cc | 6 +-
.../v1/v1_cri_container_manager_service.cc | 203 +--------------
.../cri/v1/v1_cri_container_manager_service.h | 13 -
.../v1/v1_cri_pod_sandbox_manager_service.cc | 92 ++++++-
.../v1/v1_cri_pod_sandbox_manager_service.h | 10 +-
.../entry/cri/v1/v1_cri_runtime_service.h | 4 +-
.../cri/v1/v1_cri_runtime_service_impl.cc | 10 +-
.../cri/v1/v1_cri_runtime_service_impl.h | 7 +-
src/daemon/executor/container_cb/execution.c | 25 ++
.../executor/container_cb/execution_create.c | 12 +
src/daemon/mailbox/CMakeLists.txt | 11 +
src/daemon/mailbox/mailbox.c | 167 +++++++++++++
src/daemon/mailbox/mailbox.h | 82 ++++++
src/daemon/mailbox/mailbox_message.c | 94 +++++++
src/daemon/mailbox/mailbox_message.h | 50 ++++
src/daemon/mailbox/message_queue.c | 234 ++++++++++++++++++
src/daemon/mailbox/message_queue.h | 57 +++++
src/daemon/mailbox/message_subscriber.c | 85 +++++++
src/daemon/mailbox/message_subscriber.h | 41 +++
src/daemon/modules/api/container_api.h | 5 +
.../modules/container/supervisor/supervisor.c | 18 ++
src/daemon/sandbox/sandbox.cc | 9 +
src/utils/cutils/blocking_queue.c | 185 ++++++++++++++
src/utils/cutils/blocking_queue.h | 66 +++++
test/mocks/mailbox_mock.cc | 30 +++
test/mocks/mailbox_mock.h | 30 +++
test/sandbox/controller/shim/CMakeLists.txt | 1 +
test/sandbox/sandbox/CMakeLists.txt | 2 +
38 files changed, 1681 insertions(+), 244 deletions(-)
rename src/daemon/{entry => common}/cri/v1/v1_naming.cc (100%)
rename src/daemon/{entry => common}/cri/v1/v1_naming.h (100%)
create mode 100644 src/daemon/mailbox/CMakeLists.txt
create mode 100644 src/daemon/mailbox/mailbox.c
create mode 100644 src/daemon/mailbox/mailbox.h
create mode 100644 src/daemon/mailbox/mailbox_message.c
create mode 100644 src/daemon/mailbox/mailbox_message.h
create mode 100644 src/daemon/mailbox/message_queue.c
create mode 100644 src/daemon/mailbox/message_queue.h
create mode 100644 src/daemon/mailbox/message_subscriber.c
create mode 100644 src/daemon/mailbox/message_subscriber.h
create mode 100644 src/utils/cutils/blocking_queue.c
create mode 100644 src/utils/cutils/blocking_queue.h
create mode 100644 test/mocks/mailbox_mock.cc
create mode 100644 test/mocks/mailbox_mock.h
diff --git a/src/daemon/CMakeLists.txt b/src/daemon/CMakeLists.txt
index d5280c88..29af3dca 100644
--- a/src/daemon/CMakeLists.txt
+++ b/src/daemon/CMakeLists.txt
@@ -3,6 +3,7 @@
aux_source_directory(${CMAKE_CURRENT_SOURCE_DIR} daemon_top_srcs)
add_subdirectory(executor)
add_subdirectory(entry)
+add_subdirectory(mailbox)
add_subdirectory(modules)
add_subdirectory(config)
add_subdirectory(common)
@@ -11,6 +12,7 @@ set(local_daemon_srcs
${daemon_top_srcs}
${EXECUTOR_SRCS}
${ENTRY_SRCS}
+ ${MAILBOX_SRCS}
${MODULES_SRCS}
${CONFIG_SRCS}
${DAEMON_COMMON_SRCS}
@@ -20,6 +22,7 @@ set(local_daemon_incs
${CMAKE_CURRENT_SOURCE_DIR}
${EXECUTOR_INCS}
${ENTRY_INCS}
+ ${MAILBOX_INCS}
${MODULES_INCS}
${CONFIG_INCS}
${DAEMON_COMMON_INCS}
diff --git a/src/daemon/common/cri/v1/v1_cri_helpers.cc b/src/daemon/common/cri/v1/v1_cri_helpers.cc
index c57301ce..a3488894 100644
--- a/src/daemon/common/cri/v1/v1_cri_helpers.cc
+++ b/src/daemon/common/cri/v1/v1_cri_helpers.cc
@@ -32,6 +32,7 @@
#include "service_container_api.h"
#include "isulad_config.h"
#include "sha256.h"
+#include "v1_naming.h"
namespace CRIHelpersV1 {
@@ -458,4 +459,208 @@ void ApplySandboxSecurityContextToHostConfig(const runtime::v1::LinuxSandboxSecu
}
}
+void PackContainerImageToStatus(
+ container_inspect *inspect, std::unique_ptr<runtime::v1::ContainerStatus> &contStatus, Errors &error)
+{
+ if (inspect->config == nullptr) {
+ return;
+ }
+
+ if (inspect->config->image != nullptr) {
+ contStatus->mutable_image()->set_image(inspect->config->image);
+ }
+
+ contStatus->set_image_ref(CRIHelpers::ToPullableImageID(inspect->config->image, inspect->config->image_ref));
+}
+
+void UpdateBaseStatusFromInspect(
+ container_inspect *inspect, int64_t &createdAt, int64_t &startedAt, int64_t &finishedAt,
+ std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
+{
+ runtime::v1::ContainerState state { runtime::v1::CONTAINER_UNKNOWN };
+ std::string reason;
+ std::string message;
+ int32_t exitCode { 0 };
+
+ if (inspect->state == nullptr) {
+ goto pack_status;
+ }
+
+ if (inspect->state->running) {
+ // Container is running
+ state = runtime::v1::CONTAINER_RUNNING;
+ } else {
+ // Container is not running.
+ if (finishedAt != 0) { // Case 1
+ state = runtime::v1::CONTAINER_EXITED;
+ if (inspect->state->exit_code == 0) {
+ reason = "Completed";
+ } else {
+ reason = "Error";
+ }
+ } else if (inspect->state->exit_code != 0) { // Case 2
+ state = runtime::v1::CONTAINER_EXITED;
+ finishedAt = createdAt;
+ startedAt = createdAt;
+ reason = "ContainerCannotRun";
+ } else { // Case 3
+ state = runtime::v1::CONTAINER_CREATED;
+ }
+ if (inspect->state->error != nullptr) {
+ message = inspect->state->error;
+ }
+ exitCode = (int32_t)inspect->state->exit_code;
+ }
+
+pack_status:
+ contStatus->set_exit_code(exitCode);
+ contStatus->set_state(state);
+ contStatus->set_created_at(createdAt);
+ contStatus->set_started_at(startedAt);
+ contStatus->set_finished_at(finishedAt);
+ contStatus->set_reason(reason);
+ contStatus->set_message(message);
+}
+
+void PackLabelsToStatus(container_inspect *inspect,
+ std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
+{
+ if (inspect->config == nullptr || inspect->config->labels == nullptr) {
+ return;
+ }
+ CRIHelpers::ExtractLabels(inspect->config->labels, *contStatus->mutable_labels());
+ CRIHelpers::ExtractAnnotations(inspect->config->annotations, *contStatus->mutable_annotations());
+ for (size_t i = 0; i < inspect->config->labels->len; i++) {
+ if (strcmp(inspect->config->labels->keys[i], CRIHelpers::Constants::CONTAINER_LOGPATH_LABEL_KEY.c_str()) == 0) {
+ contStatus->set_log_path(inspect->config->labels->values[i]);
+ break;
+ }
+ }
+}
+
+void ConvertMountsToStatus(container_inspect *inspect,
+ std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
+{
+ for (size_t i = 0; i < inspect->mounts_len; i++) {
+ runtime::v1::Mount *mount = contStatus->add_mounts();
+ mount->set_host_path(inspect->mounts[i]->source);
+ mount->set_container_path(inspect->mounts[i]->destination);
+ mount->set_readonly(!inspect->mounts[i]->rw);
+ if (inspect->mounts[i]->propagation == nullptr || strcmp(inspect->mounts[i]->propagation, "rprivate") == 0) {
+ mount->set_propagation(runtime::v1::PROPAGATION_PRIVATE);
+ } else if (strcmp(inspect->mounts[i]->propagation, "rslave") == 0) {
+ mount->set_propagation(runtime::v1::PROPAGATION_HOST_TO_CONTAINER);
+ } else if (strcmp(inspect->mounts[i]->propagation, "rshared") == 0) {
+ mount->set_propagation(runtime::v1::PROPAGATION_BIDIRECTIONAL);
+ }
+ // Note: Can't set SeLinuxRelabel
+ }
+}
+
+void ConvertResourcesToStatus(container_inspect *inspect,
+ std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
+{
+ if (inspect->resources == nullptr) {
+ return;
+ }
+ runtime::v1::LinuxContainerResources *resources = contStatus->mutable_resources()->mutable_linux();
+ if (inspect->resources->cpu_shares != 0) {
+ resources->set_cpu_shares(inspect->resources->cpu_shares);
+ }
+ if (inspect->resources->cpu_period != 0) {
+ resources->set_cpu_period(inspect->resources->cpu_period);
+ }
+ if (inspect->resources->cpu_quota != 0) {
+ resources->set_cpu_quota(inspect->resources->cpu_quota);
+ }
+ if (inspect->resources->memory != 0) {
+ resources->set_memory_limit_in_bytes(inspect->resources->memory);
+ }
+ if (inspect->resources->memory_swap != 0) {
+ resources->set_memory_swap_limit_in_bytes(inspect->resources->memory_swap);
+ }
+ for (size_t i = 0; i < inspect->resources->hugetlbs_len; i++) {
+ runtime::v1::HugepageLimit *hugepage = resources->add_hugepage_limits();
+ hugepage->set_page_size(inspect->resources->hugetlbs[i]->page_size);
+ hugepage->set_limit(inspect->resources->hugetlbs[i]->limit);
+ }
+ if (inspect->resources->unified != nullptr) {
+ for (size_t i = 0; i < inspect->resources->unified->len; i++) {
+ auto &resUnified = *(resources->mutable_unified());
+ resUnified[inspect->resources->unified->keys[i]] = inspect->resources->unified->values[i];
+ }
+ }
+}
+
+void ContainerStatusToGRPC(container_inspect *inspect,
+ std::unique_ptr<runtime::v1::ContainerStatus> &contStatus,
+ Errors &error)
+{
+ if (inspect->id != nullptr) {
+ contStatus->set_id(inspect->id);
+ }
+
+ int64_t createdAt {};
+ int64_t startedAt {};
+ int64_t finishedAt {};
+ CRIHelpers::GetContainerTimeStamps(inspect, &createdAt, &startedAt, &finishedAt, error);
+ if (error.NotEmpty()) {
+ return;
+ }
+ contStatus->set_created_at(createdAt);
+ contStatus->set_started_at(startedAt);
+ contStatus->set_finished_at(finishedAt);
+
+ PackContainerImageToStatus(inspect, contStatus, error);
+ UpdateBaseStatusFromInspect(inspect, createdAt, startedAt, finishedAt, contStatus);
+ PackLabelsToStatus(inspect, contStatus);
+ CRINamingV1::ParseContainerName(contStatus->annotations(), contStatus->mutable_metadata(), error);
+ if (error.NotEmpty()) {
+ return;
+ }
+ ConvertMountsToStatus(inspect, contStatus);
+ ConvertResourcesToStatus(inspect, contStatus);
+}
+
+std::unique_ptr<runtime::v1::ContainerStatus> GetContainerStatus(service_executor_t *m_cb, const std::string &containerID, Errors &error)
+{
+ if (m_cb == nullptr) {
+ error.SetError("Invalid input arguments: empty service executor");
+ return nullptr;
+ }
+
+ if (containerID.empty()) {
+ error.SetError("Empty container id");
+ return nullptr;
+ }
+
+ std::string realContainerID = CRIHelpers::GetRealContainerOrSandboxID(m_cb, containerID, false, error);
+ if (error.NotEmpty()) {
+ ERROR("Failed to find container id %s: %s", containerID.c_str(), error.GetCMessage());
+ error.Errorf("Failed to find container id %s: %s", containerID.c_str(), error.GetCMessage());
+ return nullptr;
+ }
+
+ container_inspect *inspect = CRIHelpers::InspectContainer(realContainerID, error, false);
+ if (error.NotEmpty()) {
+ return nullptr;
+ }
+ if (inspect == nullptr) {
+ error.SetError("Get null inspect");
+ return nullptr;
+ }
+ using ContainerStatusPtr = std::unique_ptr<runtime::v1::ContainerStatus>;
+ ContainerStatusPtr contStatus(new (std::nothrow) runtime::v1::ContainerStatus);
+ if (contStatus == nullptr) {
+ error.SetError("Out of memory");
+ free_container_inspect(inspect);
+ return nullptr;
+ }
+
+ ContainerStatusToGRPC(inspect, contStatus, error);
+
+ free_container_inspect(inspect);
+ return contStatus;
+}
+
} // v1 namespace CRIHelpers
diff --git a/src/daemon/common/cri/v1/v1_cri_helpers.h b/src/daemon/common/cri/v1/v1_cri_helpers.h
index b6e6aec6..1578c428 100644
--- a/src/daemon/common/cri/v1/v1_cri_helpers.h
+++ b/src/daemon/common/cri/v1/v1_cri_helpers.h
@@ -27,6 +27,8 @@
#include "checkpoint_handler.h"
#include "constants.h"
#include "errors.h"
+#include "callback.h"
+#include "cstruct_wrapper.h"
namespace CRIHelpersV1 {
@@ -78,6 +80,9 @@ std::string CRISandboxerConvert(const std::string &runtime);
void ApplySandboxSecurityContextToHostConfig(const runtime::v1::LinuxSandboxSecurityContext &context, host_config *hc,
Errors &error);
+auto GetContainerStatus(service_executor_t *m_cb, const std::string &containerID, Errors &error)
+-> std::unique_ptr<runtime::v1::ContainerStatus>;
+
}; // namespace CRIHelpers
#endif // DAEMON_ENTRY_CRI_V1ALPHA_CRI_HELPERS_H
diff --git a/src/daemon/entry/cri/v1/v1_naming.cc b/src/daemon/common/cri/v1/v1_naming.cc
similarity index 100%
rename from src/daemon/entry/cri/v1/v1_naming.cc
rename to src/daemon/common/cri/v1/v1_naming.cc
diff --git a/src/daemon/entry/cri/v1/v1_naming.h b/src/daemon/common/cri/v1/v1_naming.h
similarity index 100%
rename from src/daemon/entry/cri/v1/v1_naming.h
rename to src/daemon/common/cri/v1/v1_naming.h
diff --git a/src/daemon/config/isulad_config.c b/src/daemon/config/isulad_config.c
index 8179558e..778ff921 100644
--- a/src/daemon/config/isulad_config.c
+++ b/src/daemon/config/isulad_config.c
@@ -1760,6 +1760,7 @@ int merge_json_confs_into_global(struct service_arguments *args)
args->json_confs->cri_sandboxers = tmp_json_confs->cri_sandboxers;
tmp_json_confs->cri_sandboxers = NULL;
args->json_confs->enable_cri_v1 = tmp_json_confs->enable_cri_v1;
+ args->json_confs->enable_pod_events = tmp_json_confs->enable_pod_events;
#endif
args->json_confs->systemd_cgroup = tmp_json_confs->systemd_cgroup;
diff --git a/src/daemon/entry/connect/grpc/cri/cri_service.cc b/src/daemon/entry/connect/grpc/cri/cri_service.cc
index c1986c44..d10a60b5 100644
--- a/src/daemon/entry/connect/grpc/cri/cri_service.cc
+++ b/src/daemon/entry/connect/grpc/cri/cri_service.cc
@@ -89,8 +89,9 @@ int CRIService::Init(const isulad_daemon_configs *config)
#ifdef ENABLE_CRI_API_V1
m_enableCRIV1 = config->enable_cri_v1;
+ m_enablePodEvents = config->enable_pod_events;
if (m_enableCRIV1) {
- m_runtimeV1RuntimeService.Init(m_podSandboxImage, m_pluginManager, err);
+ m_runtimeV1RuntimeService.Init(m_podSandboxImage, m_pluginManager, m_enablePodEvents, err);
if (err.NotEmpty()) {
ERROR("Init CRI v1 runtime service failed: %s", err.GetCMessage());
return -1;
diff --git a/src/daemon/entry/connect/grpc/cri/cri_service.h b/src/daemon/entry/connect/grpc/cri/cri_service.h
index 77b2eb72..041c7c63 100644
--- a/src/daemon/entry/connect/grpc/cri/cri_service.h
+++ b/src/daemon/entry/connect/grpc/cri/cri_service.h
@@ -56,6 +56,7 @@ private:
std::string m_podSandboxImage;
std::shared_ptr<Network::PluginManager> m_pluginManager;
bool m_enableCRIV1;
+ bool m_enablePodEvents;
};
}
diff --git a/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.cc b/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.cc
index 76e393f3..bc5ab591 100644
--- a/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.cc
+++ b/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.cc
@@ -22,11 +22,37 @@
#include "callback.h"
#include "network_plugin.h"
#include "v1_cri_runtime_service_impl.h"
+#include "mailbox.h"
+#include "mailbox_message.h"
using namespace CRIV1;
+static void *cri_container_topic_handler(void *context, void *arg)
+{
+ if (context == nullptr || arg == nullptr) {
+ ERROR("Invalid input arguments");
+ return nullptr;
+ }
+
+ auto v1runtimeService = static_cast<RuntimeV1RuntimeServiceImpl *>(context);
+ auto msg = static_cast<cri_container_message_t *>(arg);
+ return v1runtimeService->GenerateCRIContainerEvent(msg->container_id, msg->sandbox_id,
+ static_cast<runtime::v1::ContainerEventType>(msg->type));
+}
+
+static void cri_container_topic_release(void *arg)
+{
+ if (arg == nullptr) {
+ return;
+ }
+
+ auto resp = static_cast<runtime::v1::ContainerEventResponse *>(arg);
+ delete resp;
+}
+
void RuntimeV1RuntimeServiceImpl::Init(std::string &podSandboxImage,
- std::shared_ptr<Network::PluginManager> networkPlugin, Errors &err)
+ std::shared_ptr<Network::PluginManager> networkPlugin,
+ bool enablePodEvents, Errors &err)
{
// Assembly implementation for CRIRuntimeServiceImpl
service_executor_t *cb = get_service_executor();
@@ -36,7 +62,18 @@ void RuntimeV1RuntimeServiceImpl::Init(std::string &podSandboxImage,
return;
}
- m_rService = std::unique_ptr<CRIV1::CRIRuntimeService>(new CRIRuntimeServiceImpl(podSandboxImage, cb, networkPlugin));
+ if (enablePodEvents) {
+ if (mailbox_register_topic_handler(MAILBOX_TOPIC_CRI_CONTAINER, cri_container_topic_handler,
+ this, cri_container_topic_release, true) != 0) {
+ ERROR("Failed to register container topic handler");
+ err.SetError("Failed to register container topic handler");
+ return;
+ }
+ m_enablePodEvents = enablePodEvents;
+ }
+
+
+ m_rService = std::unique_ptr<CRIV1::CRIRuntimeService>(new CRIRuntimeServiceImpl(podSandboxImage, cb, networkPlugin, m_enablePodEvents));
}
void RuntimeV1RuntimeServiceImpl::Wait()
@@ -45,6 +82,54 @@ void RuntimeV1RuntimeServiceImpl::Wait()
void RuntimeV1RuntimeServiceImpl::Shutdown()
{
+ mailbox_unregister_topic_handler(MAILBOX_TOPIC_CRI_CONTAINER);
+}
+
+auto RuntimeV1RuntimeServiceImpl::GenerateCRIContainerEvent(const char *container_id, const char *sandbox_id,
+ runtime::v1::ContainerEventType type) -> runtime::v1::ContainerEventResponse *
+{
+ if (container_id == nullptr || sandbox_id == nullptr) {
+ ERROR("Invalid input arguments");
+ return nullptr;
+ }
+
+ if (type < runtime::v1::ContainerEventType::CONTAINER_CREATED_EVENT ||
+ type > runtime::v1::ContainerEventType::CONTAINER_DELETED_EVENT) {
+ ERROR("Invalid container event type %d", type);
+ return nullptr;
+ }
+
+ std::string containerID(container_id), sandboxID(sandbox_id);
+ Errors error;
+ runtime::v1::ContainerEventResponse *response = new (std::nothrow) runtime::v1::ContainerEventResponse();
+ if (response == nullptr) {
+ ERROR("Out of memory");
+ return nullptr;
+ }
+
+ runtime::v1::PodSandboxStatusResponse *statusReply = new (std::nothrow) runtime::v1::PodSandboxStatusResponse();
+ if (statusReply == nullptr) {
+ ERROR("Out of memory");
+ delete response;
+ return nullptr;
+ }
+
+ m_rService->PodSandboxStatus(sandboxID, statusReply, error);
+ if (!error.Empty()) {
+ WARN("Object: CRI, Type: Failed to status pod:%s due to %s", sandboxID.c_str(),
+ error.GetMessage().c_str());
+ } else {
+ *(response->mutable_pod_sandbox_status()) = *(statusReply->mutable_status());
+ for (auto &containerStatus : statusReply->containers_statuses()) {
+ *(response->add_containers_statuses()) = containerStatus;
+ }
+ }
+
+ response->set_container_event_type((runtime::v1::ContainerEventType)type);
+ response->set_container_id(containerID);
+ response->set_created_at(util_get_now_time_nanos());
+
+ return response;
}
grpc::Status RuntimeV1RuntimeServiceImpl::Version(grpc::ServerContext *context,
@@ -398,14 +483,12 @@ grpc::Status RuntimeV1RuntimeServiceImpl::PodSandboxStatus(grpc::ServerContext *
INFO("Event: {Object: CRI, Type: Status Pod: %s}", request->pod_sandbox_id().c_str());
- std::unique_ptr<runtime::v1::PodSandboxStatus> podStatus;
- podStatus = m_rService->PodSandboxStatus(request->pod_sandbox_id(), error);
- if (!error.Empty() || !podStatus) {
+ m_rService->PodSandboxStatus(request->pod_sandbox_id(), reply, error);
+ if (!error.Empty()) {
ERROR("Object: CRI, Type: Failed to status pod:%s due to %s", request->pod_sandbox_id().c_str(),
error.GetMessage().c_str());
return grpc::Status(grpc::StatusCode::UNKNOWN, error.GetMessage());
}
- *(reply->mutable_status()) = *podStatus;
INFO("Event: {Object: CRI, Type: Statused Pod: %s}", request->pod_sandbox_id().c_str());
@@ -657,3 +740,55 @@ RuntimeV1RuntimeServiceImpl::RuntimeConfig(grpc::ServerContext *context,
return grpc::Status::OK;
}
+
+grpc::Status RuntimeV1RuntimeServiceImpl::GetContainerEvents(grpc::ServerContext *context,
+ const runtime::v1::GetEventsRequest *request,
+ grpc::ServerWriter<runtime::v1::ContainerEventResponse> *writer)
+{
+ Errors error;
+
+ if (context == nullptr || request == nullptr || writer == nullptr) {
+ ERROR("Invalid input arguments");
+ return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, "Invalid input arguments");
+ }
+
+ if (!m_enablePodEvents) {
+ ERROR("Pod events is not enabled");
+ return grpc::Status(grpc::StatusCode::UNIMPLEMENTED, "Pod events is not enabled");
+ }
+
+ INFO("Event: {Object: CRI, Type: Getting Container Events}");
+
+ __isula_auto_subscriber auto sub = mailbox_subscribe(MAILBOX_TOPIC_CRI_CONTAINER);
+ if (sub == nullptr) {
+ ERROR("Object: CRI, Type: Failed to subscribe container events");
+ return grpc::Status(grpc::StatusCode::UNKNOWN, "Failed to subscribe container events");
+ }
+
+ for (;;) {
+ __isula_auto_mailbox_message mailbox_message *msg = NULL;
+ int ret = message_subscriber_pop(sub, &msg);
+ if (ret == 0) {
+ if (msg == nullptr) {
+ // nullptr response indicates eventqueue being shutdown, not need to unscribe now
+ return grpc::Status(grpc::StatusCode::UNKNOWN, "Event queue is shutdown");
+ }
+ auto *response = static_cast<runtime::v1::ContainerEventResponse *>(msg->data);
+ if (!writer->Write(*response)) {
+ break;
+ }
+ } else if (ret != ETIMEDOUT) {
+ ERROR("Failed to pop message from subscriber");
+ break;
+ }
+ if (context->IsCancelled()) {
+ INFO("Object: CRI, Type: GetContainerEvents is cancelled");
+ break;
+ }
+ }
+
+ mailbox_unsubscribe(MAILBOX_TOPIC_CRI_CONTAINER, sub);
+ INFO("Event: {Object: CRI, Type: Got Container Events}");
+
+ return grpc::Status::OK;
+}
diff --git a/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.h b/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.h
index 52cc6b99..842d1811 100644
--- a/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.h
+++ b/src/daemon/entry/connect/grpc/cri/v1/cri_v1_runtime_runtime_service.h
@@ -26,9 +26,13 @@
// Implement of runtime RuntimeService
class RuntimeV1RuntimeServiceImpl : public runtime::v1::RuntimeService::Service {
public:
- void Init(std::string &podSandboxImage, std::shared_ptr<Network::PluginManager> networkPlugin, Errors &err);
+ void Init(std::string &podSandboxImage, std::shared_ptr<Network::PluginManager> networkPlugin,
+ bool enablePodEvents, Errors &err);
void Wait();
void Shutdown();
+ auto GenerateCRIContainerEvent(const char *container_id, const char *sandbox_id, runtime::v1::ContainerEventType type)
+ -> runtime::v1::ContainerEventResponse *;
+
grpc::Status Version(grpc::ServerContext *context, const runtime::v1::VersionRequest *request,
runtime::v1::VersionResponse *reply) override;
@@ -105,8 +109,13 @@ public:
const runtime::v1::RuntimeConfigRequest *request,
runtime::v1::RuntimeConfigResponse *reply) override;
+ grpc::Status GetContainerEvents(grpc::ServerContext *context,
+ const runtime::v1::GetEventsRequest *request,
+ grpc::ServerWriter<runtime::v1::ContainerEventResponse> *writer) override;
+
private:
std::unique_ptr<CRIV1::CRIRuntimeService> m_rService;
+ bool m_enablePodEvents;
};
#endif // DAEMON_ENTRY_CONNECT_GRPC_CRI_V1_RUNTIME_RUNTIME_SERVICE_H
diff --git a/src/daemon/entry/connect/grpc/grpc_service.cc b/src/daemon/entry/connect/grpc/grpc_service.cc
index 61e284f3..1d8de922 100644
--- a/src/daemon/entry/connect/grpc/grpc_service.cc
+++ b/src/daemon/entry/connect/grpc/grpc_service.cc
@@ -108,11 +108,11 @@ public:
void Shutdown(void)
{
- m_server->Shutdown();
-
- // call CRI to shutdown stream server
+ // call CRI to shutdown stream server, shutdown cri first to notify events thread to exit
m_criService.Shutdown();
+ m_server->Shutdown();
+
// Shutdown daemon, this operation should remove socket file.
for (const auto &address : m_socketPath) {
if (address.find(UNIX_SOCKET_PREFIX) == 0) {
diff --git a/src/daemon/entry/cri/v1/v1_cri_container_manager_service.cc b/src/daemon/entry/cri/v1/v1_cri_container_manager_service.cc
index cac5c0ba..e86dafae 100644
--- a/src/daemon/entry/cri/v1/v1_cri_container_manager_service.cc
+++ b/src/daemon/entry/cri/v1/v1_cri_container_manager_service.cc
@@ -1007,208 +1007,9 @@ cleanup:
return contStats;
}
-void ContainerManagerService::PackContainerImageToStatus(
- container_inspect *inspect, std::unique_ptr<runtime::v1::ContainerStatus> &contStatus, Errors &error)
+std::unique_ptr<runtime::v1::ContainerStatus> ContainerManagerService::ContainerStatus(const std::string &containerID, Errors &error)
{
- if (inspect->config == nullptr) {
- return;
- }
-
- if (inspect->config->image != nullptr) {
- contStatus->mutable_image()->set_image(inspect->config->image);
- }
-
- contStatus->set_image_ref(CRIHelpers::ToPullableImageID(inspect->config->image, inspect->config->image_ref));
- return;
-}
-
-void ContainerManagerService::UpdateBaseStatusFromInspect(
- container_inspect *inspect, int64_t &createdAt, int64_t &startedAt, int64_t &finishedAt,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
-{
- runtime::v1::ContainerState state { runtime::v1::CONTAINER_UNKNOWN };
- std::string reason;
- std::string message;
- int32_t exitCode { 0 };
-
- if (inspect->state == nullptr) {
- goto pack_status;
- }
-
- if (inspect->state->running) {
- // Container is running
- state = runtime::v1::CONTAINER_RUNNING;
- } else {
- // Container is not running.
- if (finishedAt != 0) { // Case 1
- state = runtime::v1::CONTAINER_EXITED;
- if (inspect->state->exit_code == 0) {
- reason = "Completed";
- } else {
- reason = "Error";
- }
- } else if (inspect->state->exit_code != 0) { // Case 2
- state = runtime::v1::CONTAINER_EXITED;
- finishedAt = createdAt;
- startedAt = createdAt;
- reason = "ContainerCannotRun";
- } else { // Case 3
- state = runtime::v1::CONTAINER_CREATED;
- }
- if (inspect->state->oom_killed) {
- reason = "OOMKilled";
- }
- if (inspect->state->error != nullptr) {
- message = inspect->state->error;
- }
- exitCode = (int32_t)inspect->state->exit_code;
- }
-
-pack_status:
- contStatus->set_exit_code(exitCode);
- contStatus->set_state(state);
- contStatus->set_created_at(createdAt);
- contStatus->set_started_at(startedAt);
- contStatus->set_finished_at(finishedAt);
- contStatus->set_reason(reason);
- contStatus->set_message(message);
-}
-
-void ContainerManagerService::PackLabelsToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
-{
- if (inspect->config == nullptr || inspect->config->labels == nullptr) {
- return;
- }
- CRIHelpers::ExtractLabels(inspect->config->labels, *contStatus->mutable_labels());
- CRIHelpers::ExtractAnnotations(inspect->config->annotations, *contStatus->mutable_annotations());
- for (size_t i = 0; i < inspect->config->labels->len; i++) {
- if (strcmp(inspect->config->labels->keys[i], CRIHelpers::Constants::CONTAINER_LOGPATH_LABEL_KEY.c_str()) == 0) {
- contStatus->set_log_path(inspect->config->labels->values[i]);
- break;
- }
- }
-}
-
-void ContainerManagerService::ConvertMountsToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
-{
- for (size_t i = 0; i < inspect->mounts_len; i++) {
- runtime::v1::Mount *mount = contStatus->add_mounts();
- mount->set_host_path(inspect->mounts[i]->source);
- mount->set_container_path(inspect->mounts[i]->destination);
- mount->set_readonly(!inspect->mounts[i]->rw);
- if (inspect->mounts[i]->propagation == nullptr || strcmp(inspect->mounts[i]->propagation, "rprivate") == 0) {
- mount->set_propagation(runtime::v1::PROPAGATION_PRIVATE);
- } else if (strcmp(inspect->mounts[i]->propagation, "rslave") == 0) {
- mount->set_propagation(runtime::v1::PROPAGATION_HOST_TO_CONTAINER);
- } else if (strcmp(inspect->mounts[i]->propagation, "rshared") == 0) {
- mount->set_propagation(runtime::v1::PROPAGATION_BIDIRECTIONAL);
- }
- // Note: Can't set SeLinuxRelabel
- }
-}
-
-void ContainerManagerService::ConvertResourcesToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus)
-{
- if (inspect->resources == nullptr) {
- return;
- }
- runtime::v1::LinuxContainerResources *resources = contStatus->mutable_resources()->mutable_linux();
- if (inspect->resources->cpu_shares != 0) {
- resources->set_cpu_shares(inspect->resources->cpu_shares);
- }
- if (inspect->resources->cpu_period != 0) {
- resources->set_cpu_period(inspect->resources->cpu_period);
- }
- if (inspect->resources->cpu_quota != 0) {
- resources->set_cpu_quota(inspect->resources->cpu_quota);
- }
- if (inspect->resources->memory != 0) {
- resources->set_memory_limit_in_bytes(inspect->resources->memory);
- }
- if (inspect->resources->memory_swap != 0) {
- resources->set_memory_swap_limit_in_bytes(inspect->resources->memory_swap);
- }
- for (size_t i = 0; i < inspect->resources->hugetlbs_len; i++) {
- runtime::v1::HugepageLimit *hugepage = resources->add_hugepage_limits();
- hugepage->set_page_size(inspect->resources->hugetlbs[i]->page_size);
- hugepage->set_limit(inspect->resources->hugetlbs[i]->limit);
- }
- if (inspect->resources->unified != nullptr) {
- for (size_t i = 0; i < inspect->resources->unified->len; i++) {
- auto &resUnified = *(resources->mutable_unified());
- resUnified[inspect->resources->unified->keys[i]] = inspect->resources->unified->values[i];
- }
- }
-}
-
-void ContainerManagerService::ContainerStatusToGRPC(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus,
- Errors &error)
-{
- if (inspect->id != nullptr) {
- contStatus->set_id(inspect->id);
- }
-
- int64_t createdAt {};
- int64_t startedAt {};
- int64_t finishedAt {};
- CRIHelpers::GetContainerTimeStamps(inspect, &createdAt, &startedAt, &finishedAt, error);
- if (error.NotEmpty()) {
- return;
- }
- contStatus->set_created_at(createdAt);
- contStatus->set_started_at(startedAt);
- contStatus->set_finished_at(finishedAt);
-
- PackContainerImageToStatus(inspect, contStatus, error);
- UpdateBaseStatusFromInspect(inspect, createdAt, startedAt, finishedAt, contStatus);
- PackLabelsToStatus(inspect, contStatus);
- CRINamingV1::ParseContainerName(contStatus->annotations(), contStatus->mutable_metadata(), error);
- if (error.NotEmpty()) {
- return;
- }
- ConvertMountsToStatus(inspect, contStatus);
- ConvertResourcesToStatus(inspect, contStatus);
-}
-
-std::unique_ptr<runtime::v1::ContainerStatus>
-ContainerManagerService::ContainerStatus(const std::string &containerID, Errors &error)
-{
- if (containerID.empty()) {
- error.SetError("Empty container id");
- return nullptr;
- }
-
- std::string realContainerID = CRIHelpers::GetRealContainerOrSandboxID(m_cb, containerID, false, error);
- if (error.NotEmpty()) {
- ERROR("Failed to find container id %s: %s", containerID.c_str(), error.GetCMessage());
- error.Errorf("Failed to find container id %s: %s", containerID.c_str(), error.GetCMessage());
- return nullptr;
- }
-
- container_inspect *inspect = CRIHelpers::InspectContainer(realContainerID, error, false);
- if (error.NotEmpty()) {
- return nullptr;
- }
- if (inspect == nullptr) {
- error.SetError("Get null inspect");
- return nullptr;
- }
- using ContainerStatusPtr = std::unique_ptr<runtime::v1::ContainerStatus>;
- ContainerStatusPtr contStatus(new (std::nothrow) runtime::v1::ContainerStatus);
- if (contStatus == nullptr) {
- error.SetError("Out of memory");
- free_container_inspect(inspect);
- return nullptr;
- }
-
- ContainerStatusToGRPC(inspect, contStatus, error);
-
- free_container_inspect(inspect);
- return contStatus;
+ return CRIHelpersV1::GetContainerStatus(m_cb, containerID, error);
}
void ContainerManagerService::UpdateContainerResources(const std::string &containerID,
diff --git a/src/daemon/entry/cri/v1/v1_cri_container_manager_service.h b/src/daemon/entry/cri/v1/v1_cri_container_manager_service.h
index 4e772bda..50f5ed69 100644
--- a/src/daemon/entry/cri/v1/v1_cri_container_manager_service.h
+++ b/src/daemon/entry/cri/v1/v1_cri_container_manager_service.h
@@ -111,19 +111,6 @@ private:
std::unique_ptr<runtime::v1::ContainerStats> &container);
void SetFsUsage(const imagetool_fs_info *fs_usage, int64_t timestamp,
std::unique_ptr<runtime::v1::ContainerStats> &container);
- void ContainerStatusToGRPC(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus, Errors &error);
- void PackContainerImageToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus, Errors &error);
- void UpdateBaseStatusFromInspect(container_inspect *inspect, int64_t &createdAt, int64_t &startedAt,
- int64_t &finishedAt,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus);
- void PackLabelsToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus);
- void ConvertMountsToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus);
- void ConvertResourcesToStatus(container_inspect *inspect,
- std::unique_ptr<runtime::v1::ContainerStatus> &contStatus);
void ExecSyncFromGRPC(const std::string &containerID, const google::protobuf::RepeatedPtrField<std::string> &cmd,
int64_t timeout, container_exec_request **request, Errors &error);
auto ValidateExecRequest(const runtime::v1::ExecRequest &req, Errors &error) -> int;
diff --git a/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.cc b/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.cc
index f125e714..4291d8a0 100644
--- a/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.cc
+++ b/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.cc
@@ -36,6 +36,7 @@
#include "sandbox_manager.h"
#include "transform.h"
#include "isulad_config.h"
+#include "mailbox.h"
namespace CRIV1 {
void PodSandboxManagerService::PrepareSandboxData(const runtime::v1::PodSandboxConfig &config,
@@ -302,6 +303,7 @@ auto PodSandboxManagerService::RunPodSandbox(const runtime::v1::PodSandboxConfig
std::string jsonCheckpoint;
std::string network_setting_json;
runtime::v1::PodSandboxConfig copyConfig = config;
+ cri_container_message_t msg = { 0 };
// Step 1: Parepare sandbox name, runtime and networkMode
PrepareSandboxData(config, runtimeHandler, sandboxName, runtimeInfo, networkMode, error);
@@ -372,6 +374,11 @@ auto PodSandboxManagerService::RunPodSandbox(const runtime::v1::PodSandboxConfig
goto cleanup_network;
}
+ msg.container_id = sandbox->GetId().c_str();
+ msg.sandbox_id = sandbox->GetId().c_str();
+ msg.type = CRI_CONTAINER_MESSAGE_TYPE_CREATED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &msg);
+
// Step 10: Save network settings json to disk
// Update network settings before start sandbox since sandbox container will use the sandbox key
if (namespace_is_cni(networkMode.c_str())) {
@@ -391,6 +398,9 @@ auto PodSandboxManagerService::RunPodSandbox(const runtime::v1::PodSandboxConfig
return response_id;
}
+ msg.type = CRI_CONTAINER_MESSAGE_TYPE_STARTED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &msg);
+
return sandbox->GetId();
cleanup_network:
@@ -700,6 +710,13 @@ void PodSandboxManagerService::RemovePodSandbox(const std::string &podSandboxID,
ERROR("Failed to delete sandbox %s: %s", podSandboxID.c_str(), error.GetCMessage());
}
+ if (error.Empty()) {
+ cri_container_message_t msg = { 0 };
+ msg.container_id = sandbox->GetId().c_str();
+ msg.sandbox_id = sandbox->GetId().c_str();
+ msg.type = CRI_CONTAINER_MESSAGE_TYPE_DELETED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &msg);
+ }
}
auto PodSandboxManagerService::SharesHostNetwork(const container_inspect *inspect) -> runtime::v1::NamespaceMode
@@ -800,10 +817,29 @@ void PodSandboxManagerService::SetSandboxStatusNetwork(std::shared_ptr<sandbox::
}
}
-std::unique_ptr<runtime::v1::PodSandboxStatus>
-PodSandboxManagerService::PodSandboxStatus(const std::string &podSandboxID, Errors &error)
+void PodSandboxManagerService::GetContainerStatuses(const std::string &podSandboxID,
+ std::vector<std::unique_ptr<runtime::v1::ContainerStatus>> &containerStatuses,
+ std::vector<std::string> &errors) {
+ auto list_response_wrapper = GetContainerListResponse(podSandboxID, errors);
+ if (list_response_wrapper == nullptr) {
+ return;
+ }
+
+ auto list_response = list_response_wrapper->get();
+ // Remove all containers in the sandbox.
+ for (size_t i = 0; i < list_response->containers_len; i++) {
+ Errors stError;
+ containerStatuses.push_back(CRIHelpersV1::GetContainerStatus(m_cb, list_response->containers[i]->id, stError));
+ if (stError.NotEmpty()) {
+ ERROR("Error get container status: %s: %s", list_response->containers[i]->id, stError.GetCMessage());
+ errors.push_back(stError.GetMessage());
+ }
+ }
+}
+
+std::unique_ptr<runtime::v1::PodSandboxStatus> PodSandboxManagerService::GetPodSandboxStatus(const std::string &podSandboxID, Errors &error)
{
- std::unique_ptr<runtime::v1::PodSandboxStatus> podStatus(new runtime::v1::PodSandboxStatus);
+ std::unique_ptr<runtime::v1::PodSandboxStatus> podStatus(new (std::nothrow) runtime::v1::PodSandboxStatus);
if (podStatus == nullptr) {
ERROR("Out of memory");
error.SetError("Out of memory");
@@ -831,6 +867,50 @@ PodSandboxManagerService::PodSandboxStatus(const std::string &podSandboxID, Erro
return podStatus;
}
+void PodSandboxManagerService::PodSandboxStatus(const std::string &podSandboxID,
+ runtime::v1::PodSandboxStatusResponse *reply, Errors &error)
+{
+ if (reply == nullptr) {
+ ERROR("Invalid NULL reply");
+ error.SetError("Invalid NULL reply");
+ return;
+ }
+
+
+ auto podStatus = GetPodSandboxStatus(podSandboxID, error);
+ if (error.NotEmpty()) {
+ ERROR("Failed to get pod sandbox status: %s", error.GetCMessage());
+ return;
+ }
+
+ auto sandbox = sandbox::SandboxManager::GetInstance()->GetSandbox(podSandboxID);
+ if (sandbox == nullptr) {
+ ERROR("Failed to find sandbox id %s", podSandboxID.c_str());
+ error.Errorf("Failed to find sandbox id %s", podSandboxID.c_str());
+ return;
+ }
+
+ *(reply->mutable_status()) = *podStatus;
+
+
+ if (!m_enablePodEvents) {
+ return;
+ }
+
+ std::vector<std::unique_ptr<runtime::v1::ContainerStatus>> containerStatuses;
+ std::vector<std::string> errors;
+ GetContainerStatuses(sandbox->GetId(), containerStatuses, errors);
+ if (errors.size() != 0) {
+ error.SetAggregate(errors);
+ return;
+ }
+
+ for (auto &containerStatus : containerStatuses) {
+ *(reply->add_containers_statuses()) = *containerStatus;
+ }
+ return;
+}
+
void PodSandboxManagerService::ListPodSandbox(const runtime::v1::PodSandboxFilter &filter,
std::vector<std::unique_ptr<runtime::v1::PodSandbox>> &pods,
Errors &error)
@@ -944,7 +1024,7 @@ void PodSandboxManagerService::GetPodSandboxNetworkMetrics(const std::string &ne
void PodSandboxManagerService::PackagePodSandboxStatsAttributes(
const std::string &id, std::unique_ptr<runtime::v1::PodSandboxStats> &podStatsPtr, Errors &error)
{
- auto status = PodSandboxStatus(id, error);
+ auto status = GetPodSandboxStatus(id, error);
if (error.NotEmpty()) {
return;
}
@@ -1111,8 +1191,8 @@ auto PodSandboxManagerService::PodSandboxStats(const std::string &podSandboxID,
auto &config = sandbox->GetSandboxConfig();
auto oldStatsRec = sandbox->GetStatsInfo();
- auto status = PodSandboxStatus(sandbox->GetId(), tmpErr);
- if (error.NotEmpty()) {
+ auto status = GetPodSandboxStatus(sandbox->GetId(), tmpErr);
+ if (tmpErr.NotEmpty()) {
ERROR("Failed to get podsandbox %s status: %s", sandbox->GetId().c_str(), tmpErr.GetCMessage());
error.Errorf("Failed to get podsandbox %s status", sandbox->GetId().c_str());
return nullptr;
diff --git a/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.h b/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.h
index c3d98b8c..3872c4c9 100644
--- a/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.h
+++ b/src/daemon/entry/cri/v1/v1_cri_pod_sandbox_manager_service.h
@@ -38,10 +38,11 @@ namespace CRIV1 {
class PodSandboxManagerService {
public:
PodSandboxManagerService(const std::string &podSandboxImage, service_executor_t *cb,
- std::shared_ptr<Network::PluginManager> pluginManager)
+ std::shared_ptr<Network::PluginManager> pluginManager, bool enablePodEvents)
: m_podSandboxImage(podSandboxImage)
, m_cb(cb)
, m_pluginManager(pluginManager)
+ , m_enablePodEvents(enablePodEvents)
{
}
PodSandboxManagerService(const PodSandboxManagerService &) = delete;
@@ -55,8 +56,7 @@ public:
void RemovePodSandbox(const std::string &podSandboxID, Errors &error);
- auto PodSandboxStatus(const std::string &podSandboxID, Errors &error)
- -> std::unique_ptr<runtime::v1::PodSandboxStatus>;
+ void PodSandboxStatus(const std::string &podSandboxID, runtime::v1::PodSandboxStatusResponse *reply, Errors &error);
void ListPodSandbox(const runtime::v1::PodSandboxFilter &filter,
std::vector<std::unique_ptr<runtime::v1::PodSandbox>> &pods, Errors &error);
@@ -129,6 +129,9 @@ private:
std::vector<std::string> &podSandboxIDs, Errors &error);
void ApplySandboxLinuxOptions(const runtime::v1::LinuxPodSandboxConfig &lc, host_config *hc,
container_config *custom_config, Errors &error);
+ auto GetPodSandboxStatus(const std::string &podSandboxID, Errors &error) -> std::unique_ptr<runtime::v1::PodSandboxStatus>;
+ void GetContainerStatuses(const std::string &podSandboxID, std::vector<std::unique_ptr<runtime::v1::ContainerStatus>> &containerStatuses,
+ std::vector<std::string> &errors);
private:
std::string m_podSandboxImage;
@@ -136,6 +139,7 @@ private:
std::map<std::string, bool> m_networkReady;
service_executor_t *m_cb { nullptr };
std::shared_ptr<Network::PluginManager> m_pluginManager { nullptr };
+ bool m_enablePodEvents;
};
} // namespace CRI
diff --git a/src/daemon/entry/cri/v1/v1_cri_runtime_service.h b/src/daemon/entry/cri/v1/v1_cri_runtime_service.h
index 839f6724..4521e3df 100644
--- a/src/daemon/entry/cri/v1/v1_cri_runtime_service.h
+++ b/src/daemon/entry/cri/v1/v1_cri_runtime_service.h
@@ -70,8 +70,8 @@ public:
virtual void RemovePodSandbox(const std::string &podSandboxID, Errors &error) = 0;
- virtual auto PodSandboxStatus(const std::string &podSandboxID,
- Errors &error) -> std::unique_ptr<runtime::v1::PodSandboxStatus> = 0;
+ virtual void PodSandboxStatus(const std::string &podSandboxID, runtime::v1::PodSandboxStatusResponse *reply,
+ Errors &error) = 0;
virtual void ListPodSandbox(const runtime::v1::PodSandboxFilter &filter,
std::vector<std::unique_ptr<runtime::v1::PodSandbox>> &pods, Errors &error) = 0;
diff --git a/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.cc b/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.cc
index aa5ae516..7b40e29d 100644
--- a/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.cc
+++ b/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.cc
@@ -19,11 +19,12 @@
namespace CRIV1 {
CRIRuntimeServiceImpl::CRIRuntimeServiceImpl(const std::string &podSandboxImage, service_executor_t *cb,
- std::shared_ptr<Network::PluginManager> pluginManager)
+ std::shared_ptr<Network::PluginManager> pluginManager, bool enablePodEvents)
: m_runtimeVersioner(new RuntimeVersionerService(cb))
, m_containerManager(new ContainerManagerService(cb))
- , m_podSandboxManager(new PodSandboxManagerService(podSandboxImage, cb, pluginManager))
+ , m_podSandboxManager(new PodSandboxManagerService(podSandboxImage, cb, pluginManager, enablePodEvents))
, m_runtimeManager(new RuntimeManagerService(cb, pluginManager))
+ , m_enablePodEvents(enablePodEvents)
{
}
@@ -124,10 +125,9 @@ void CRIRuntimeServiceImpl::RemovePodSandbox(const std::string &podSandboxID, Er
m_podSandboxManager->RemovePodSandbox(podSandboxID, error);
}
-auto CRIRuntimeServiceImpl::PodSandboxStatus(const std::string &podSandboxID, Errors &error)
--> std::unique_ptr<runtime::v1::PodSandboxStatus>
+void CRIRuntimeServiceImpl::PodSandboxStatus(const std::string &podSandboxID, runtime::v1::PodSandboxStatusResponse *reply, Errors &error)
{
- return m_podSandboxManager->PodSandboxStatus(podSandboxID, error);
+ m_podSandboxManager->PodSandboxStatus(podSandboxID, reply, error);
}
void CRIRuntimeServiceImpl::ListPodSandbox(const runtime::v1::PodSandboxFilter &filter,
diff --git a/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.h b/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.h
index 0a25749f..6ae59bfa 100644
--- a/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.h
+++ b/src/daemon/entry/cri/v1/v1_cri_runtime_service_impl.h
@@ -26,7 +26,8 @@ namespace CRIV1 {
class CRIRuntimeServiceImpl : public CRIRuntimeService {
public:
CRIRuntimeServiceImpl(const std::string &podSandboxImage, service_executor_t *cb,
- std::shared_ptr<Network::PluginManager> pluginManager);
+ std::shared_ptr<Network::PluginManager> pluginManager,
+ bool enablePodEvents);
CRIRuntimeServiceImpl(const CRIRuntimeServiceImpl &) = delete;
auto operator=(const CRIRuntimeServiceImpl &) -> CRIRuntimeServiceImpl & = delete;
virtual ~CRIRuntimeServiceImpl() = default;
@@ -72,8 +73,7 @@ public:
void RemovePodSandbox(const std::string &podSandboxID, Errors &error) override;
- auto PodSandboxStatus(const std::string &podSandboxID, Errors &error)
- -> std::unique_ptr<runtime::v1::PodSandboxStatus> override;
+ void PodSandboxStatus(const std::string &podSandboxID, runtime::v1::PodSandboxStatusResponse *reply, Errors &error) override;
void ListPodSandbox(const runtime::v1::PodSandboxFilter &filter,
std::vector<std::unique_ptr<runtime::v1::PodSandbox>> &pods, Errors &error) override;
@@ -103,6 +103,7 @@ protected:
private:
std::string m_podSandboxImage;
std::shared_ptr<Network::PluginManager> m_pluginManager { nullptr };
+ bool m_enablePodEvents;
};
} // namespace CRIV1
#endif // DAEMON_ENTRY_CRI_V1_CRI_RUNTIME_SERVICE_IMPL_H
diff --git a/src/daemon/executor/container_cb/execution.c b/src/daemon/executor/container_cb/execution.c
index 88c6b354..e5c96628 100644
--- a/src/daemon/executor/container_cb/execution.c
+++ b/src/daemon/executor/container_cb/execution.c
@@ -62,6 +62,7 @@
#include "event_type.h"
#include "utils_timestamp.h"
#include "utils_verify.h"
+#include "mailbox.h"
#ifdef ENABLE_NATIVE_NETWORK
#include "service_network_api.h"
@@ -542,6 +543,9 @@ static int container_start_cb(const container_start_request *request, container_
container_t *cont = NULL;
int sync_fd = -1;
pthread_t thread_id = 0;
+#ifdef ENABLE_CRI_API_V1
+ cri_container_message_t message;
+#endif
DAEMON_CLEAR_ERRMSG();
@@ -596,6 +600,15 @@ static int container_start_cb(const container_start_request *request, container_
EVENT("Event: {Object: %s, Type: Running}", id);
(void)isulad_monitor_send_container_event(id, START, -1, 0, NULL, NULL);
+#ifdef ENABLE_CRI_API_V1
+ if (is_container_in_sandbox(cont->common_config->sandbox_info)) {
+ message.container_id = id;
+ message.sandbox_id = cont->common_config->sandbox_info->id;
+ message.type = CRI_CONTAINER_MESSAGE_TYPE_STARTED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &message);
+ }
+#endif
+
pack_response:
handle_start_io_thread_by_cc(cc, sync_fd, thread_id);
delete_daemon_fifos(fifopath, (const char **)fifos);
@@ -1009,6 +1022,9 @@ static int container_delete_cb(const container_delete_request *request, containe
char *name = NULL;
char *id = NULL;
container_t *cont = NULL;
+#ifdef ENABLE_CRI_API_V1
+ cri_container_message_t message;
+#endif
DAEMON_CLEAR_ERRMSG();
if (request == NULL || response == NULL) {
@@ -1063,6 +1079,15 @@ static int container_delete_cb(const container_delete_request *request, containe
EVENT("Event: {Object: %s, Type: Deleted}", id);
+#ifdef ENABLE_CRI_API_V1
+ if (is_container_in_sandbox(cont->common_config->sandbox_info)) {
+ message.container_id = cont->common_config->id;
+ message.sandbox_id = cont->common_config->sandbox_info->id;
+ message.type = CRI_CONTAINER_MESSAGE_TYPE_DELETED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &message);
+ }
+#endif
+
pack_response:
pack_delete_response(*response, cc, id);
container_unref(cont);
diff --git a/src/daemon/executor/container_cb/execution_create.c b/src/daemon/executor/container_cb/execution_create.c
index e00afb68..a9102226 100644
--- a/src/daemon/executor/container_cb/execution_create.c
+++ b/src/daemon/executor/container_cb/execution_create.c
@@ -62,6 +62,7 @@
#include "opt_log.h"
#include "runtime_api.h"
#include "id_name_manager.h"
+#include "mailbox.h"
#ifdef ENABLE_CRI_API_V1
static bool validate_sandbox_info(const container_sandbox_info *sandbox)
@@ -1389,6 +1390,9 @@ int container_create_cb(const container_create_request *request, container_creat
bool skip_id_name_manage = false;
bool skip_sandbox_key_manage = false;
__isula_auto_sysinfo_t sysinfo_t *sysinfo = NULL;
+#ifdef ENABLE_CRI_API_V1
+ cri_container_message_t message;
+#endif
DAEMON_CLEAR_ERRMSG();
@@ -1572,6 +1576,14 @@ int container_create_cb(const container_create_request *request, container_creat
EVENT("Event: {Object: %s, Type: Created %s}", id, name);
(void)isulad_monitor_send_container_event(id, CREATE, -1, 0, NULL, NULL);
+#ifdef ENABLE_CRI_API_V1
+ if (is_container_in_sandbox(request->sandbox)) {
+ message.container_id = id;
+ message.sandbox_id = request->sandbox->id;
+ message.type = CRI_CONTAINER_MESSAGE_TYPE_CREATED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &message);
+ }
+#endif
goto pack_response;
umount_channel:
diff --git a/src/daemon/mailbox/CMakeLists.txt b/src/daemon/mailbox/CMakeLists.txt
new file mode 100644
index 00000000..984f9acb
--- /dev/null
+++ b/src/daemon/mailbox/CMakeLists.txt
@@ -0,0 +1,11 @@
+# get current directory sources files
+aux_source_directory(${CMAKE_CURRENT_SOURCE_DIR} mailbox_top_srcs)
+
+set(MAILBOX_SRCS
+ ${mailbox_top_srcs}
+ PARENT_SCOPE
+ )
+set(MAILBOX_INCS
+ ${CMAKE_CURRENT_SOURCE_DIR}
+ PARENT_SCOPE
+ )
\ No newline at end of file
diff --git a/src/daemon/mailbox/mailbox.c b/src/daemon/mailbox/mailbox.c
new file mode 100644
index 00000000..732b91b9
--- /dev/null
+++ b/src/daemon/mailbox/mailbox.c
@@ -0,0 +1,167 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide common event definition
+ ******************************************************************************/
+
+#include "mailbox.h"
+
+#include <isula_libutils/log.h>
+
+#include "message_queue.h"
+#include "mailbox_message.h"
+#include "message_subscriber.h"
+
+mailbox_topic_handler_t mailbox_topic_handlers[MAILBOX_TOPIC_MAX] = { 0 };
+
+static bool mailbox_topic_valid(mailbox_topic topic) {
+ return topic > MAILBOX_TOPIC_INVALID && topic < MAILBOX_TOPIC_MAX;
+}
+
+static bool mailbox_should_publish(mailbox_topic topic)
+{
+ if (!mailbox_topic_valid(topic)) {
+ ERROR("Invalid topic %d", topic);
+ return false;
+ }
+
+ if (!mailbox_topic_handlers[topic].registered) {
+ return false;
+ }
+
+ if (mailbox_topic_handlers[topic].queue == NULL) {
+ return true;
+ }
+
+ // for async queues, only publish if anyone subscribe
+ return message_queue_have_subscribers(mailbox_topic_handlers[topic].queue);
+}
+
+// only register once when iSulad start, no need to free the queue
+int mailbox_register_topic_handler(mailbox_topic topic, message_generator_t generator, void *context,
+ message_release_t release, bool async)
+{
+ if (!mailbox_topic_valid(topic)) {
+ ERROR("Invalid topic %d", topic);
+ return -1;
+ }
+
+ if (generator == NULL) {
+ ERROR("Invalid generator for topic %d", topic);
+ return -1;
+ }
+
+ mailbox_topic_handlers[topic].generator = generator;
+ mailbox_topic_handlers[topic].context = context;
+ mailbox_topic_handlers[topic].release = release;
+ if (async) {
+ mailbox_topic_handlers[topic].queue = message_queue_create(release);
+ if (mailbox_topic_handlers[topic].queue == NULL) {
+ ERROR("Failed to create message queue for topic %d", topic);
+ return -1;
+ }
+ }
+ mailbox_topic_handlers[topic].registered = true;
+ return 0;
+}
+
+// unregister only when iSulad shutdown, no need to free the queue
+void mailbox_unregister_topic_handler(mailbox_topic topic)
+{
+ if (!mailbox_topic_valid(topic)) {
+ ERROR("Invalid topic %d", topic);
+ return;
+ }
+
+ if (mailbox_topic_handlers[topic].queue != NULL) {
+ message_queue_shutdown(mailbox_topic_handlers[topic].queue);
+ }
+ mailbox_topic_handlers[topic].registered = false;
+}
+
+void mailbox_publish(mailbox_topic topic, void *data)
+{
+ if (!mailbox_should_publish(topic)) {
+ return;
+ }
+
+ message_generator_t generator = mailbox_topic_handlers[topic].generator;
+ void *context = mailbox_topic_handlers[topic].context;
+ message_release_t release = mailbox_topic_handlers[topic].release;
+ message_queue *queue = mailbox_topic_handlers[topic].queue;
+
+ if (generator == NULL) {
+ ERROR("No handler for topic %d", topic);
+ return;
+ }
+
+ void *middle = generator(context, data);
+ if (middle == NULL) {
+ return;
+ }
+
+ if (queue != NULL) {
+ mailbox_message *msg = mailbox_message_create(middle, release);
+ if (msg == NULL) {
+ ERROR("Failed to create mailbox message");
+ if (release) {
+ release(middle);
+ }
+ return;
+ }
+ if (message_queue_publish(queue, msg) != 0) {
+ ERROR("Failed to publish event");
+ mailbox_message_unref(msg);
+ return;
+ }
+ }
+}
+
+message_subscriber *mailbox_subscribe(mailbox_topic topic)
+{
+ if (!mailbox_topic_valid(topic)) {
+ ERROR("Invalid topic %d", topic);
+ return NULL;
+ }
+
+ if (!mailbox_topic_handlers[topic].registered) {
+ ERROR("Handler for topic %d not registered", topic);
+ return NULL;
+ }
+
+ if (mailbox_topic_handlers[topic].queue != NULL) {
+ return message_queue_subscribe(mailbox_topic_handlers[topic].queue,
+ mailbox_topic_handlers[topic].release);
+ }
+
+ // For sync queues, there is no need to subscribe, just return
+ return NULL;
+}
+
+void mailbox_unsubscribe(mailbox_topic topic, message_subscriber *sub)
+{
+ if (!mailbox_topic_valid(topic)) {
+ ERROR("Invalid topic %d", topic);
+ return;
+ }
+
+ if (!mailbox_topic_handlers[topic].registered) {
+ ERROR("Handler for topic %d not registered", topic);
+ return;
+ }
+
+ if (mailbox_topic_handlers[topic].queue != NULL) {
+ return message_queue_unsubscribe(mailbox_topic_handlers[topic].queue, sub);
+ }
+
+ return;
+}
diff --git a/src/daemon/mailbox/mailbox.h b/src/daemon/mailbox/mailbox.h
new file mode 100644
index 00000000..1dc2e934
--- /dev/null
+++ b/src/daemon/mailbox/mailbox.h
@@ -0,0 +1,82 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide common event definition
+ ******************************************************************************/
+
+#ifndef DAEMON_MAILBOX_MAILBOX_H
+#define DAEMON_MAILBOX_MAILBOX_H
+
+#include "daemon_arguments.h"
+#include "blocking_queue.h"
+#include "message_queue.h"
+#include "message_subscriber.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+typedef enum {
+ MAILBOX_TOPIC_INVALID = -1,
+ MAILBOX_TOPIC_CRI_CONTAINER,
+ MAILBOX_TOPIC_MAX
+} mailbox_topic;
+
+// for async message, it generates a true message to publish
+// for sync message, it is the callback function to handle the data to publish
+typedef void *(*message_generator_t)(void *, void *);
+// release function of message generated by generator, if any
+typedef void (*message_release_t)(void *);
+
+typedef struct {
+ // to generate a message
+ message_generator_t generator;
+ // context of handler
+ void *context;
+ // release function of message, if any
+ message_release_t release;
+ // message queue
+ message_queue *queue;
+ // if registered
+ bool registered;
+} mailbox_topic_handler_t;
+
+typedef enum {
+ CRI_CONTAINER_MESSAGE_TYPE_INVALID = -1,
+ CRI_CONTAINER_MESSAGE_TYPE_CREATED,
+ CRI_CONTAINER_MESSAGE_TYPE_STARTED,
+ CRI_CONTAINER_MESSAGE_TYPE_STOPPED,
+ CRI_CONTAINER_MESSAGE_TYPE_DELETED,
+ CRI_CONTAINER_MESSAGE_TYPE_MAX
+} cri_container_message_type;
+
+typedef struct {
+ const char *container_id;
+ const char *sandbox_id;
+ cri_container_message_type type;
+} cri_container_message_t;
+
+int mailbox_register_topic_handler(mailbox_topic topic, message_generator_t handle, void *context,
+ message_release_t release, bool async);
+
+void mailbox_unregister_topic_handler(mailbox_topic topic);
+
+void mailbox_publish(mailbox_topic topic, void *data);
+
+message_subscriber *mailbox_subscribe(mailbox_topic topic);
+
+void mailbox_unsubscribe(mailbox_topic, message_subscriber *sub);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/daemon/mailbox/mailbox_message.c b/src/daemon/mailbox/mailbox_message.c
new file mode 100644
index 00000000..b16a1bdd
--- /dev/null
+++ b/src/daemon/mailbox/mailbox_message.c
@@ -0,0 +1,94 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide mailbox message definition
+ ******************************************************************************/
+
+#include "mailbox_message.h"
+
+#include <isula_libutils/log.h>
+
+#include "utils.h"
+
+// Once the create succeeds, the ownership is transferred to the mailbox_message.
+mailbox_message *mailbox_message_create(void *data, void (*destroy)(void *)) {
+ __isula_auto_free mailbox_message *msg = NULL;
+ msg = util_common_calloc_s(sizeof(mailbox_message));
+ if (msg == NULL) {
+ ERROR("Out of memory");
+ return NULL;
+ }
+
+ msg->data = data;
+ msg->destroy = destroy;
+ msg->ref_count = 1;
+
+ if (pthread_mutex_init(&msg->lock, NULL) != 0) {
+ ERROR("Failed to init mutex");
+ return NULL;
+ }
+
+ return isula_transfer_ptr(msg);
+}
+
+int mailbox_message_ref(mailbox_message *dest) {
+ __isula_auto_pm_unlock pthread_mutex_t *lock = NULL;
+ if (dest == NULL) {
+ ERROR("Invalid mailbox_message");
+ return -1;
+ }
+
+ if (pthread_mutex_lock(&dest->lock) != 0) {
+ ERROR("Failed to lock mutex");
+ return -1;
+ }
+ lock = &dest->lock;
+
+ if (dest->ref_count == INT_MAX) {
+ ERROR("Reference count overflow");
+ return -1;
+ }
+
+ dest->ref_count++;
+
+ return 0;
+}
+
+void mailbox_message_unref(mailbox_message *dest) {
+ __isula_auto_pm_unlock pthread_mutex_t *lock = NULL;
+ if (dest == NULL) {
+ return;
+ }
+
+ if (pthread_mutex_lock(&dest->lock) != 0) {
+ ERROR("Failed to lock mutex");
+ return;
+ }
+ lock = &dest->lock;
+
+ if (dest->ref_count == 0) {
+ ERROR("Reference count underflow, should not reach here");
+ return;
+ }
+
+ dest->ref_count--;
+ if (dest->ref_count == 0) {
+ if (dest->destroy) {
+ dest->destroy(dest->data);
+ }
+ lock = NULL;
+ (void)pthread_mutex_unlock(&dest->lock);
+ (void)pthread_mutex_destroy(&dest->lock);
+ free(dest);
+ }
+ return;
+}
diff --git a/src/daemon/mailbox/mailbox_message.h b/src/daemon/mailbox/mailbox_message.h
new file mode 100644
index 00000000..39e40b70
--- /dev/null
+++ b/src/daemon/mailbox/mailbox_message.h
@@ -0,0 +1,50 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide ref counted ptr definition
+ ******************************************************************************/
+
+#ifndef DAEMON_MAILBOX_MAILBOX_MESSAGE_H
+#define DAEMON_MAILBOX_MAILBOX_MESSAGE_H
+
+#include <pthread.h>
+#include <stdbool.h>
+
+#include <isula_libutils/auto_cleanup.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct mailbox_message {
+ void *data;
+ size_t ref_count;
+ pthread_mutex_t lock;
+ void (*destroy)(void *);
+} mailbox_message;
+
+mailbox_message *mailbox_message_create(void *ptr, void (*destroy)(void *));
+
+int mailbox_message_ref(mailbox_message *p);
+
+void mailbox_message_unref(mailbox_message *p);
+
+// define auto free function callback for mailbox_message
+define_auto_cleanup_callback(mailbox_message_unref, mailbox_message);
+// define auto free macro for char *
+#define __isula_auto_mailbox_message auto_cleanup_tag(mailbox_message_unref)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/daemon/mailbox/message_queue.c b/src/daemon/mailbox/message_queue.c
new file mode 100644
index 00000000..7fe044f2
--- /dev/null
+++ b/src/daemon/mailbox/message_queue.c
@@ -0,0 +1,234 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide message queue definition
+ ******************************************************************************/
+
+#include "message_queue.h"
+
+#include <sys/prctl.h>
+#include <isula_libutils/log.h>
+
+#include "utils.h"
+
+// default set subscriber timeout to 1000ms, maybe could be configured later
+const int64_t subscribe_timeout = 1000;
+
+static void message_queue_subscriber_free(void *key, void *val)
+{
+ return;
+}
+
+static void *message_queue_thread(void *arg)
+{
+ int ret = 0;
+
+ ret = pthread_detach(pthread_self());
+ if (ret != 0) {
+ CRIT("Set thread detach fail");
+ return NULL;
+ }
+
+ prctl(PR_SET_NAME, "Message Queue");
+
+ message_queue *mq = (message_queue *)arg;
+ if (mq == NULL) {
+ ERROR("Invalid argument");
+ return NULL;
+ }
+
+ for (;;) {
+ void *data = NULL;
+ if (blocking_queue_pop(mq->messages, &data) != 0) {
+ ERROR("Fail to get message, message queue thread exit");
+ break;
+ }
+
+ __isula_auto_mailbox_message mailbox_message *msg = (mailbox_message *)data;
+ // an empty msg indicates shutdown
+ if (pthread_rwlock_rdlock(&mq->rwlock) != 0) {
+ ERROR("Failed to lock rwlock");
+ continue;
+ }
+
+ bool should_shutdown = (msg == NULL);
+ map_itor *itor = map_itor_new(mq->subscribers);
+ if (itor == NULL) {
+ ERROR("Out of memory");
+ if (pthread_rwlock_unlock(&mq->rwlock) != 0) {
+ ERROR("Failed to lock rwlock");
+ }
+ break;
+ }
+
+ for (; map_itor_valid(itor); map_itor_next(itor)) {
+ void *sub = map_itor_key(itor);
+ if (should_shutdown) {
+ message_subscriber_shutdown(sub);
+ } else {
+ if (message_subscriber_push(sub, msg) != 0) {
+ ERROR("Failed to push event to subscriber");
+ }
+ }
+ }
+ map_itor_free(itor);
+
+ if (pthread_rwlock_unlock(&mq->rwlock) != 0) {
+ ERROR("Failed to unlock rwlock");
+ }
+
+ // if msg is NULL, it is a shutdown signal
+ if (should_shutdown) {
+ break;
+ }
+ }
+
+ return NULL;
+}
+
+message_queue *message_queue_create(void (*release)(void *))
+{
+ __isula_auto_free message_queue *mq = NULL;
+ __isula_auto_blocking_queue blocking_queue *bq = NULL;
+ pthread_t message_queue_tid;
+ mq = util_common_calloc_s(sizeof(message_queue));
+ if (mq == NULL) {
+ ERROR("Out of memory");
+ return NULL;
+ }
+
+ bq = blocking_queue_create(BLOCKING_QUEUE_NO_TIMEOUT, release);
+ if (bq == NULL) {
+ ERROR("Failed to create events queue");
+ return NULL;
+ }
+
+ mq->subscribers = map_new(MAP_PTR_INT, MAP_DEFAULT_CMP_FUNC, message_queue_subscriber_free);
+ if (mq->subscribers == NULL) {
+ ERROR("Failed to create subscribers map");
+ return NULL;
+ }
+
+ if (pthread_rwlock_init(&mq->rwlock, NULL) != 0) {
+ ERROR("Failed to init rwlock");
+ map_free(mq->subscribers);
+ return NULL;
+ }
+
+ if (pthread_create(&message_queue_tid, NULL, message_queue_thread, mq) != 0) {
+ ERROR("Failed to create message queue thread");
+ pthread_rwlock_destroy(&mq->rwlock);
+ map_free(mq->subscribers);
+ return NULL;
+ }
+
+ mq->messages = isula_transfer_ptr(bq);
+ return isula_transfer_ptr(mq);
+}
+
+// message queue should be global value, it will be destroyed when daemon exit
+void message_queue_shutdown(message_queue *mq)
+{
+ if (mq == NULL) {
+ return;
+ }
+
+ blocking_queue_clear(mq->messages);
+
+ // push a nullptr to notify the thread to exit
+ if (blocking_queue_push(mq->messages, NULL) != 0) {
+ ERROR("Failed to push nullptr to message queue");
+ }
+}
+
+message_subscriber *message_queue_subscribe(message_queue *mq, void (*release)(void *))
+{
+ __isula_auto_subscriber message_subscriber *sub = NULL;
+ __isula_auto_prw_unlock pthread_rwlock_t *lock = NULL;
+ int val = 0;
+ if (mq == NULL) {
+ ERROR("Invalid argument");
+ return NULL;
+ }
+
+ sub = message_subscriber_create(subscribe_timeout, release);
+ if (sub == NULL) {
+ ERROR("Failed to create subscriber");
+ return NULL;
+ }
+
+ if (pthread_rwlock_wrlock(&mq->rwlock) != 0) {
+ ERROR("Failed to lock rwlock");
+ return NULL;
+ }
+ lock = &mq->rwlock;
+
+ if (map_insert(mq->subscribers, sub, (void *)&val) == false) {
+ ERROR("Failed to insert subscriber");
+ return NULL;
+ }
+
+ return isula_transfer_ptr(sub);
+}
+
+void message_queue_unsubscribe(message_queue *mq, message_subscriber *sub)
+{
+ __isula_auto_prw_unlock pthread_rwlock_t *lock = NULL;
+ if (mq == NULL) {
+ ERROR("Invalid argument");
+ return;
+ }
+
+ if (pthread_rwlock_wrlock(&mq->rwlock) != 0) {
+ ERROR("Failed to lock rwlock");
+ return;
+ }
+ lock = &mq->rwlock;
+
+ if (map_remove(mq->subscribers, sub) == false) {
+ ERROR("Failed to remove subscriber");
+ return;
+ }
+
+ return;
+}
+
+int message_queue_publish(message_queue *mq, mailbox_message *msg)
+{
+ if (mq == NULL || msg == NULL) {
+ ERROR("Invalid argument");
+ return -1;
+ }
+
+ if (blocking_queue_push(mq->messages, msg) != 0) {
+ ERROR("Failed to push message");
+ return -1;
+ }
+ return 0;
+}
+
+bool message_queue_have_subscribers(message_queue *mq)
+{
+ __isula_auto_prw_unlock pthread_rwlock_t *lock = NULL;
+ if (mq == NULL) {
+ ERROR("Invalid argument");
+ return false;
+ }
+
+ if (pthread_rwlock_wrlock(&mq->rwlock) != 0) {
+ ERROR("Failed to lock rwlock");
+ return false;
+ }
+ lock = &mq->rwlock;
+
+ return map_size(mq->subscribers) > 0;
+}
diff --git a/src/daemon/mailbox/message_queue.h b/src/daemon/mailbox/message_queue.h
new file mode 100644
index 00000000..7905840f
--- /dev/null
+++ b/src/daemon/mailbox/message_queue.h
@@ -0,0 +1,57 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide message queue definition
+ ******************************************************************************/
+
+#ifndef DAEMON_MESSAGE_MESSAGE_QUEUE_H
+#define DAEMON_MESSAGE_MESSAGE_QUEUE_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <pthread.h>
+
+#include "blocking_queue.h"
+#include "mailbox_message.h"
+#include "map.h"
+#include "message_subscriber.h"
+
+typedef struct message_queue {
+ blocking_queue *messages;
+
+ // lock for set of subscribers
+ pthread_rwlock_t rwlock;
+
+ map_t *subscribers;
+
+ int64_t sub_timeout;
+} message_queue;
+
+message_queue *message_queue_create(void (*release)(void *));
+
+void message_queue_shutdown(message_queue *mq);
+
+message_subscriber *message_queue_subscribe(message_queue *mq, void (*release)(void *));
+
+void message_queue_unsubscribe(message_queue *mq, message_subscriber *sub);
+
+int message_queue_publish(message_queue *mq, mailbox_message *msg);
+
+bool message_queue_have_subscribers(message_queue *mq);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/src/daemon/mailbox/message_subscriber.c b/src/daemon/mailbox/message_subscriber.c
new file mode 100644
index 00000000..8ef3cb58
--- /dev/null
+++ b/src/daemon/mailbox/message_subscriber.c
@@ -0,0 +1,85 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide message subscriber definition
+ ******************************************************************************/
+
+#include "message_subscriber.h"
+
+#include <isula_libutils/log.h>
+
+#include "utils.h"
+
+message_subscriber *message_subscriber_create(int64_t timeout, void (*release)(void *))
+{
+ message_subscriber *sub = (message_subscriber *)util_common_calloc_s(sizeof(message_subscriber));
+ if (sub == NULL) {
+ ERROR("Out of memory");
+ return NULL;
+ }
+ sub->queue = blocking_queue_create(timeout, release);
+ if (sub->queue == NULL) {
+ ERROR("Failed to create blocking queue");
+ free(sub);
+ return NULL;
+ }
+ return sub;
+}
+
+int message_subscriber_push(message_subscriber *sub, mailbox_message *msg)
+{
+ if (sub == NULL || msg == NULL) {
+ ERROR("Invalid argument");
+ return -1;
+ }
+
+ if (mailbox_message_ref(msg) != 0) {
+ ERROR("Failed to get message");
+ return -1;
+ }
+
+ if (blocking_queue_push(sub->queue, msg) != 0) {
+ ERROR("Failed to push message to queue");
+ mailbox_message_unref(msg);
+ return -1;
+ }
+
+ return 0;
+}
+
+int message_subscriber_pop(message_subscriber *sub, mailbox_message **msg)
+{
+ if (sub == NULL) {
+ ERROR("Invalid argument");
+ return -1;
+ }
+ return blocking_queue_pop(sub->queue, (void **)msg);
+}
+
+void message_subscriber_shutdown(message_subscriber *sub)
+{
+ if (sub == NULL) {
+ return;
+ }
+
+ blocking_queue_clear(sub->queue);
+ (void)blocking_queue_push(sub->queue, NULL);
+}
+
+void message_subscriber_destroy(message_subscriber *sub)
+{
+ if (sub == NULL) {
+ return;
+ }
+ blocking_queue_destroy(sub->queue);
+ free(sub);
+}
diff --git a/src/daemon/mailbox/message_subscriber.h b/src/daemon/mailbox/message_subscriber.h
new file mode 100644
index 00000000..de4574d9
--- /dev/null
+++ b/src/daemon/mailbox/message_subscriber.h
@@ -0,0 +1,41 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide message subscriber definition
+ ******************************************************************************/
+
+#ifndef SRC_DAEMON_MAILBOX_MESSAGE_SUBSCRIBER_H
+#define SRC_DAEMON_MAILBOX_MESSAGE_SUBSCRIBER_H
+
+#include "blocking_queue.h"
+#include "mailbox_message.h"
+
+typedef struct {
+ blocking_queue *queue;
+} message_subscriber;
+
+message_subscriber *message_subscriber_create(int64_t timeout, void (*release)(void *));
+
+void message_subscriber_shutdown(message_subscriber *sub);
+
+void message_subscriber_destroy(message_subscriber *sub);
+
+int message_subscriber_push(message_subscriber *sub, mailbox_message *msg);
+
+int message_subscriber_pop(message_subscriber *sub, mailbox_message **msg);
+
+// define auto free function callback for blocking queue
+define_auto_cleanup_callback(message_subscriber_destroy, message_subscriber);
+// define auto free macro for blocking queue
+#define __isula_auto_subscriber auto_cleanup_tag(message_subscriber_destroy)
+
+#endif
diff --git a/src/daemon/modules/api/container_api.h b/src/daemon/modules/api/container_api.h
index 830fd696..55c59980 100644
--- a/src/daemon/modules/api/container_api.h
+++ b/src/daemon/modules/api/container_api.h
@@ -289,6 +289,11 @@ static inline bool is_sandbox_container(container_sandbox_info *sandbox)
{
return sandbox != NULL && sandbox->is_sandbox_container;
}
+
+static inline bool is_container_in_sandbox(container_sandbox_info *sandbox)
+{
+ return sandbox != NULL && !sandbox->is_sandbox_container;
+}
#endif
#if defined(__cplusplus) || defined(c_plusplus)
diff --git a/src/daemon/modules/container/supervisor/supervisor.c b/src/daemon/modules/container/supervisor/supervisor.c
index 1b7da383..83d46268 100644
--- a/src/daemon/modules/container/supervisor/supervisor.c
+++ b/src/daemon/modules/container/supervisor/supervisor.c
@@ -38,6 +38,7 @@
#include "container_api.h"
#include "event_type.h"
#include "utils_file.h"
+#include "mailbox.h"
#ifdef ENABLE_CRI_API_V1
#include "sandbox_ops.h"
#endif
@@ -51,6 +52,7 @@ struct supervisor_handler_data {
int fd;
int exit_code;
char *name;
+ char *sandbox_name;
char *runtime;
bool is_sandbox_container;
pid_ppid_info_t pid_info;
@@ -152,6 +154,9 @@ static void supervisor_handler_data_free(struct supervisor_handler_data *data)
free(data->name);
data->name = NULL;
+ free(data->sandbox_name);
+ data->sandbox_name = NULL;
+
free(data->runtime);
data->runtime = NULL;
@@ -172,6 +177,9 @@ static void *clean_resources_thread(void *arg)
pid_t pid = data->pid_info.pid;
int retry_count = 0;
int max_retry = 10;
+#ifdef ENABLE_CRI_API_V1
+ cri_container_message_t msg;
+#endif
ret = pthread_detach(pthread_self());
if (ret != 0) {
@@ -218,6 +226,13 @@ retry:
(void)isulad_monitor_send_container_event(name, STOPPED, (int)pid, data->exit_code, NULL, NULL);
#ifdef ENABLE_CRI_API_V1
+ if (data->sandbox_name) {
+ msg.container_id = name;
+ msg.sandbox_id = data->sandbox_name;
+ msg.type = CRI_CONTAINER_MESSAGE_TYPE_STOPPED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &msg);
+ }
+
if (data->is_sandbox_container) {
if (sandbox_on_sandbox_exit(name, data->exit_code) < 0) {
ERROR("Failed to handle sandbox %s exit", name);
@@ -329,6 +344,9 @@ int container_supervisor_add_exit_monitor(int fd, const char *exit_fifo, const p
data->runtime = util_strdup_s(cont->runtime);
#ifdef ENABLE_CRI_API_V1
data->is_sandbox_container = is_sandbox_container(cont->common_config->sandbox_info);
+ if (is_container_in_sandbox(cont->common_config->sandbox_info)) {
+ data->sandbox_name = util_strdup_s(cont->common_config->sandbox_info->id);
+ }
#endif
data->pid_info.pid = pid_info->pid;
data->pid_info.start_time = pid_info->start_time;
diff --git a/src/daemon/sandbox/sandbox.cc b/src/daemon/sandbox/sandbox.cc
index 7b6496ed..c70116c1 100644
--- a/src/daemon/sandbox/sandbox.cc
+++ b/src/daemon/sandbox/sandbox.cc
@@ -37,6 +37,7 @@
#include "cxxutils.h"
#include "controller_manager.h"
#include "utils_timestamp.h"
+#include "mailbox.h"
#define SANDBOX_READY_STATE_STR "SANDBOX_READY"
#define SANDBOX_NOTREADY_STATE_STR "SANDBOX_NOTREADY"
@@ -527,6 +528,14 @@ void Sandbox::OnSandboxExit(const ControllerExitInfo &exitInfo)
if (!SaveState(error)) {
ERROR("Failed to save sandbox state, %s", m_id.c_str());
}
+
+ if (error.Empty()) {
+ cri_container_message_t msg = { 0 };
+ msg.container_id = GetId().c_str();
+ msg.sandbox_id = GetId().c_str();
+ msg.type = CRI_CONTAINER_MESSAGE_TYPE_STOPPED;
+ mailbox_publish(MAILBOX_TOPIC_CRI_CONTAINER, &msg);
+ }
}
auto Sandbox::UpdateStatus(Errors &error) -> bool
diff --git a/src/utils/cutils/blocking_queue.c b/src/utils/cutils/blocking_queue.c
new file mode 100644
index 00000000..7c9c5f50
--- /dev/null
+++ b/src/utils/cutils/blocking_queue.c
@@ -0,0 +1,185 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide blocking queue definition
+ ******************************************************************************/
+
+#include "blocking_queue.h"
+
+#include <pthread.h>
+#include <time.h>
+#include <isula_libutils/log.h>
+
+#include "utils.h"
+#include "utils_timestamp.h"
+
+// create blocking queue with timeout(ms), if timeout < 0, then with no timeout
+blocking_queue *blocking_queue_create(int64_t timeout, void (*release)(void *))
+{
+ __isula_auto_free blocking_queue *queue = NULL;
+ __isula_auto_free blocking_node *node = NULL;
+ queue = (blocking_queue *)util_common_calloc_s(sizeof(blocking_queue));
+ if (queue == NULL) {
+ ERROR("Out of memory");
+ return NULL;
+ }
+ node = (blocking_node *)util_common_calloc_s(sizeof(blocking_node));
+ if (node == NULL) {
+ ERROR("Out of memory");
+ return NULL;
+ }
+
+ if (pthread_mutex_init(&queue->lock, NULL) != 0) {
+ ERROR("Failed to init mutex");
+ return NULL;
+ }
+
+ if (pthread_cond_init(&queue->not_empty, NULL) != 0) {
+ ERROR("Failed to init cond");
+ (void)pthread_mutex_destroy(&queue->lock);
+ return NULL;
+ }
+
+ queue->head = node;
+ queue->tail = node;
+ node = NULL;
+ queue->release = release;
+
+ if (timeout >= 0) {
+ queue->timeout.tv_sec = timeout / (Time_Second / Time_Milli);
+ queue->timeout.tv_nsec = (timeout % (Time_Second / Time_Milli) ) * Time_Milli;
+ } else {
+ queue->timeout.tv_sec = -1;
+ }
+
+ return isula_transfer_ptr(queue);
+}
+
+int blocking_queue_push(blocking_queue *queue, void *data)
+{
+ __isula_auto_free blocking_node *new_node = NULL;
+ __isula_auto_pm_unlock pthread_mutex_t *lock = NULL;
+ if (queue == NULL) {
+ ERROR("Invalid NULL arguments");
+ return -1;
+ }
+
+ new_node = (blocking_node *)util_common_calloc_s(sizeof(blocking_node));
+ if (new_node == NULL) {
+ ERROR("Out of memory");
+ return -1;
+ }
+ new_node->data = data;
+ new_node->next = NULL;
+
+ if (pthread_mutex_lock(&queue->lock) != 0) {
+ ERROR("Failed to lock mutex");
+ return -1;
+ }
+ lock = &queue->lock;
+
+ queue->tail->next = new_node;
+ queue->tail = new_node;
+ new_node = NULL;
+
+ if (pthread_cond_broadcast(&queue->not_empty) != 0) {
+ ERROR("Failed to broadcast cond");
+ }
+
+ return 0;
+}
+
+int blocking_queue_pop(blocking_queue *queue, void **data) {
+ if (queue == NULL || data == NULL) {
+ ERROR("Invalid NULL arguments");
+ return -1;
+ }
+
+ __isula_auto_pm_unlock pthread_mutex_t *lock = NULL;
+ if (pthread_mutex_lock(&queue->lock) != 0) {
+ ERROR("Failed to lock mutex");
+ return -1;
+ }
+ lock = &queue->lock;
+
+ while (queue->head->next == NULL) {
+ if (queue->timeout.tv_sec >= 0) {
+ int ret = pthread_cond_timedwait(&queue->not_empty, &queue->lock, &queue->timeout);
+ if (ret != 0) {
+ if (ret != ETIMEDOUT) {
+ ERROR("Failed to wait cond");
+ }
+ return ret;
+ }
+ } else {
+ int ret = pthread_cond_wait(&queue->not_empty, &queue->lock);
+ if (ret != 0) {
+ ERROR("Failed to wait cond");
+ return ret;
+ }
+ }
+ }
+
+ blocking_node *old_head = queue->head;
+ blocking_node *new_head = old_head->next;
+ *data = new_head->data;
+ queue->head = new_head;
+
+ free(old_head);
+ return 0;
+}
+
+void blocking_queue_clear(blocking_queue *queue)
+{
+ if (queue == NULL) {
+ return;
+ }
+
+ __isula_auto_pm_unlock pthread_mutex_t *lock = NULL;
+ // clear all nodes in queue
+ if (queue == NULL) {
+ ERROR("Invalid NULL arguments");
+ return;
+ }
+
+ if (pthread_mutex_lock(&queue->lock) != 0) {
+ ERROR("Failed to lock mutex");
+ return;
+ }
+ lock = &queue->lock;
+
+ while (queue->head->next != NULL) {
+ blocking_node *old_head = queue->head;
+ blocking_node *new_head = old_head->next;
+ if (queue->release) {
+ queue->release(old_head->data);
+ }
+ free(old_head);
+ queue->head = new_head;
+ }
+}
+
+// ensure there is no other thread executing enqueue or dequeue operation
+void blocking_queue_destroy(blocking_queue *queue)
+{
+ if (queue == NULL) {
+ return;
+ }
+
+ blocking_queue_clear(queue);
+
+ (void)pthread_mutex_destroy(&queue->lock);
+
+ (void)pthread_cond_destroy(&queue->not_empty);
+
+ free(queue);
+}
diff --git a/src/utils/cutils/blocking_queue.h b/src/utils/cutils/blocking_queue.h
new file mode 100644
index 00000000..1c52a9d3
--- /dev/null
+++ b/src/utils/cutils/blocking_queue.h
@@ -0,0 +1,66 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-03-25
+ * Description: provide blocking queue definition
+ ******************************************************************************/
+
+#ifndef DAEMON_UTILS_CUTILS_BLOCKING_QUEUE_H
+#define DAEMON_UTILS_CUTILS_BLOCKING_QUEUE_H
+
+#include <pthread.h>
+#include <time.h>
+#include <isula_libutils/auto_cleanup.h>
+
+#include "utils_timestamp.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define BLOCKING_QUEUE_NO_TIMEOUT -1
+
+typedef struct blocking_node {
+ void *data;
+ struct blocking_node *next;
+} blocking_node;
+
+typedef struct blocking_queue {
+ blocking_node *head;
+ blocking_node *tail;
+ pthread_mutex_t lock;
+ struct timespec timeout;
+ pthread_cond_t not_empty;
+ void (*release)(void *);
+} blocking_queue;
+
+// create blocking queue with timeout(ms), if timeout < 0, then with no timeout
+blocking_queue *blocking_queue_create(int64_t timeout, void (*release)(void *));
+
+int blocking_queue_push(blocking_queue *queue, void *data);
+
+int blocking_queue_pop(blocking_queue *queue, void **data);
+
+void blocking_queue_clear(blocking_queue *queue);
+
+// ensure there is no other thread executing enqueue or dequeue operation
+void blocking_queue_destroy(blocking_queue *queue);
+
+// define auto free function callback for blocking queue
+define_auto_cleanup_callback(blocking_queue_destroy, blocking_queue);
+// define auto free macro for blocking queue
+#define __isula_auto_blocking_queue auto_cleanup_tag(blocking_queue_destroy)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/test/mocks/mailbox_mock.cc b/test/mocks/mailbox_mock.cc
new file mode 100644
index 00000000..601b804e
--- /dev/null
+++ b/test/mocks/mailbox_mock.cc
@@ -0,0 +1,30 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2020-2022. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-04-02
+ * Description: mailbox mock
+ ******************************************************************************/
+
+#include "mailbox_mock.h"
+
+MockMailbox *g_mailbox_mock = nullptr;
+
+void Mailbox_SetMock(MockMailbox* mock)
+{
+ g_mailbox_mock = mock;
+}
+
+void mailbox_publish(mailbox_topic topic, void *data)
+{
+ if (g_mailbox_mock != nullptr) {
+ g_mailbox_mock->MailboxPublish(topic, data);
+ }
+}
diff --git a/test/mocks/mailbox_mock.h b/test/mocks/mailbox_mock.h
new file mode 100644
index 00000000..ce48f0fc
--- /dev/null
+++ b/test/mocks/mailbox_mock.h
@@ -0,0 +1,30 @@
+/******************************************************************************
+ * Copyright (c) Huawei Technologies Co., Ltd. 2020-2022. All rights reserved.
+ * iSulad licensed under the Mulan PSL v2.
+ * You can use this software according to the terms and conditions of the Mulan PSL v2.
+ * You may obtain a copy of Mulan PSL v2 at:
+ * http://license.coscl.org.cn/MulanPSL2
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
+ * PURPOSE.
+ * See the Mulan PSL v2 for more details.
+ * Author: jikai
+ * Create: 2024-04-02
+ * Description: mailbox mock
+ ******************************************************************************/
+
+#ifndef _ISULAD_TEST_MOCKS_MAILBOX_MOCK_H
+#define _ISULAD_TEST_MOCKS_MAILBOX_MOCK_H
+
+#include <gmock/gmock.h>
+#include "mailbox.h"
+
+class MockMailbox {
+public:
+ virtual ~MockMailbox() = default;
+ MOCK_METHOD2(MailboxPublish, void(mailbox_topic topic, void *data));
+};
+
+void Mailbox_SetMock(MockMailbox* mock);
+
+#endif
diff --git a/test/sandbox/controller/shim/CMakeLists.txt b/test/sandbox/controller/shim/CMakeLists.txt
index 6423bb80..26a66e51 100644
--- a/test/sandbox/controller/shim/CMakeLists.txt
+++ b/test/sandbox/controller/shim/CMakeLists.txt
@@ -7,6 +7,7 @@ add_executable(${EXE}
${CMAKE_BINARY_DIR}/grpc/src/api/services/cri/gogo.pb.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/v1/v1_cri_helpers.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/v1/v1_cri_security_context.cc
+ ${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/v1/v1_naming.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/checkpoint_handler.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/cri_constants.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../../src/daemon/common/cri/cri_helpers.cc
diff --git a/test/sandbox/sandbox/CMakeLists.txt b/test/sandbox/sandbox/CMakeLists.txt
index 138d4d8d..2a35388f 100644
--- a/test/sandbox/sandbox/CMakeLists.txt
+++ b/test/sandbox/sandbox/CMakeLists.txt
@@ -23,6 +23,7 @@ add_executable(${EXE}
${CMAKE_CURRENT_SOURCE_DIR}/../../../test/mocks/grpc_sandboxer_client_mock.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../test/mocks/controller_stub_mock.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../test/mocks/shim_controller_mock.cc
+ ${CMAKE_CURRENT_SOURCE_DIR}/../../../test/mocks/mailbox_mock.cc
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/common/err_msg.c
${CMAKE_CURRENT_SOURCE_DIR}/sandbox_ut.cc)
@@ -33,6 +34,7 @@ target_include_directories(${EXE} PUBLIC
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/config
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/entry/cri
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/executor
+ ${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/mailbox
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/sandbox/controller
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/sandbox/controller/shim
${CMAKE_CURRENT_SOURCE_DIR}/../../../src/daemon/sandbox/controller/sandboxer
--
2.34.1
Reference in New Issue Copy Permalink