Skip to content

Commit f1a285b

Browse files
therealbobotherealbobo
committed
feat(libsinsp): enable async containerd metadata retrival
Signed-off-by: Roberto Scolaro <roberto.scolaro21@gmail.com>
1 parent 4abd3e7 commit f1a285b

File tree

2 files changed

+174
-24
lines changed

2 files changed

+174
-24
lines changed

‎userspace/libsinsp/container_engine/containerd.cpp‎

Lines changed: 107 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,15 @@ constexpr const std::string_view CONTAINERD_SOCKETS[] = {
3434
"/run/containerd/runtime2/containerd.sock", // tmp
3535
};
3636

37-
bool containerd_interface::is_ok() {
37+
bool containerd_async_source::is_ok() {
3838
return m_stub != nullptr;
3939
}
40-
containerd_interface::containerd_interface(const std::string &socket_path) {
40+
41+
containerd_async_source::containerd_async_source(const std::string &socket_path,
42+
uint64_t max_wait_ms,
43+
uint64_t ttl_ms,
44+
container_cache_interface *cache):
45+
container_async_source(max_wait_ms, ttl_ms, cache) {
4146
grpc::ChannelArguments args;
4247
args.SetInt(GRPC_ARG_ENABLE_HTTP_PROXY, 0);
4348
std::shared_ptr<grpc::Channel> channel =
@@ -70,7 +75,12 @@ containerd_interface::containerd_interface(const std::string &socket_path) {
7075
}
7176
}
7277

73-
grpc::Status containerd_interface::list_container_resp(
78+
containerd_async_source::~containerd_async_source() {
79+
this->stop();
80+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG, "containerd_async: Source destructor");
81+
}
82+
83+
grpc::Status containerd_async_source::list_container_resp(
7484
const std::string &container_id,
7585
ContainerdService::ListContainersResponse &resp) {
7686
ContainerdService::ListContainersRequest req;
@@ -99,20 +109,32 @@ libsinsp::container_engine::containerd::containerd(container_cache_interface &ca
99109
continue;
100110
}
101111

102-
m_interface = std::make_unique<containerd_interface>(socket_path);
103-
if(!m_interface->is_ok()) {
104-
m_interface.reset(nullptr);
112+
container_cache_interface *cache_interface = &container_cache();
113+
m_containerd_info_source =
114+
std::make_unique<containerd_async_source>(socket_path,
115+
containerd_async_source::NO_WAIT_LOOKUP,
116+
10000,
117+
cache_interface);
118+
if(!m_containerd_info_source->is_ok()) {
119+
m_containerd_info_source.reset(nullptr);
105120
continue;
106121
}
107122
}
108123
}
109124

110-
bool libsinsp::container_engine::containerd::parse_containerd(sinsp_container_info &container,
111-
const std::string &container_id) {
125+
bool containerd_async_source::parse(const containerd_lookup_request &request,
126+
sinsp_container_info &container) {
127+
auto container_id = request.container_id;
128+
129+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
130+
"containerd_async (%s): Looking up info for container via socket %s",
131+
request.container_id.c_str(),
132+
m_socket_path);
133+
112134
// given the truncated container id, the full container id needs to be retrivied from
113135
// containerd.
114136
ContainerdService::ListContainersResponse resp;
115-
grpc::Status status = m_interface->list_container_resp(container_id, resp);
137+
grpc::Status status = list_container_resp(container_id, resp);
116138

117139
if(!status.ok()) {
118140
libsinsp_logger()->format(
@@ -126,10 +148,11 @@ bool libsinsp::container_engine::containerd::parse_containerd(sinsp_container_in
126148
auto containers = resp.containers();
127149

128150
if(containers.size() == 0) {
129-
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
130-
"containerd (%s): ListContainerResponse status error message: "
131-
"(container id has no match)",
132-
container.m_id.c_str());
151+
libsinsp_logger()->format(
152+
sinsp_logger::SEV_DEBUG,
153+
"containerd_async (%s): ListContainerResponse status error message: "
154+
"(container id has no match)",
155+
container.m_id.c_str());
133156
return false;
134157
} else if(containers.size() > 1) {
135158
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
@@ -145,7 +168,10 @@ bool libsinsp::container_engine::containerd::parse_containerd(sinsp_container_in
145168
container.m_id = container_id;
146169
container.m_full_id = containers[0].id();
147170
// We assume that the last `/`-separated field is the image
148-
container.m_image = raw_image_splits[0].substr(raw_image_splits[0].rfind("/") + 1);
171+
container.m_image = raw_image_splits[0]
172+
.substr(raw_image_splits[0].rfind("/") + 1)
173+
.append(":")
174+
.append(raw_image_splits.back());
149175
// and the first part is the repo
150176
container.m_imagerepo = raw_image_splits[0].substr(0, raw_image_splits[0].rfind("/"));
151177
container.m_imagetag = raw_image_splits[1];
@@ -189,9 +215,44 @@ bool libsinsp::container_engine::containerd::parse_containerd(sinsp_container_in
189215
container.m_env.emplace_back(env.asString());
190216
}
191217

218+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
219+
"containerd_async (%s): parse returning true",
220+
request.container_id.c_str());
221+
192222
return true;
193223
}
194224

225+
void libsinsp::container_engine::containerd::parse_containerd(
226+
const containerd_lookup_request &request,
227+
container_cache_interface *cache) {
228+
sinsp_container_info result;
229+
230+
bool done;
231+
if(cache->async_allowed()) {
232+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
233+
"containerd_async (%s): Starting asynchronous lookup",
234+
request.container_id.c_str());
235+
done = m_containerd_info_source->lookup(request, result);
236+
} else {
237+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
238+
"containerd_async (%s): Starting synchronous lookup",
239+
request.container_id.c_str());
240+
done = m_containerd_info_source->lookup_sync(request, result);
241+
}
242+
if(done) {
243+
// if a previous lookup call already found the metadata, process it now
244+
m_containerd_info_source->source_callback(request, result);
245+
246+
if(cache->async_allowed()) {
247+
// This should *never* happen, in async mode as ttl is 0 (never wait)
248+
libsinsp_logger()->format(sinsp_logger::SEV_ERROR,
249+
"containerd_async (%s): Unexpected immediate return from "
250+
"containerd_info_source.lookup()",
251+
request.container_id.c_str());
252+
}
253+
}
254+
}
255+
195256
bool libsinsp::container_engine::containerd::resolve(sinsp_threadinfo *tinfo,
196257
bool query_os_for_missing_info) {
197258
auto container = sinsp_container_info();
@@ -201,10 +262,41 @@ bool libsinsp::container_engine::containerd::resolve(sinsp_threadinfo *tinfo,
201262
return false;
202263
}
203264

204-
if(!parse_containerd(container, container_id)) {
265+
containerd_lookup_request request(container_id, CT_CONTAINERD, 0, false);
266+
267+
container_cache_interface *cache = &container_cache();
268+
sinsp_container_info::ptr_t container_info = cache->get_container(request.container_id);
269+
270+
if(!container_info) {
271+
if(!query_os_for_missing_info) {
272+
auto container = sinsp_container_info();
273+
container.m_type = CT_CONTAINERD;
274+
container.m_id = request.container_id;
275+
container.set_lookup_status(sinsp_container_lookup::state::SUCCESSFUL);
276+
cache->notify_new_container(container, tinfo);
277+
return true;
278+
}
279+
280+
if(cache->should_lookup(request.container_id, request.container_type, 0)) {
281+
libsinsp_logger()->format(sinsp_logger::SEV_DEBUG,
282+
"containerd_async (%s): No existing container info",
283+
request.container_id.c_str());
284+
285+
// give containerd a chance to return metadata for this container
286+
cache->set_lookup_status(request.container_id,
287+
request.container_type,
288+
0,
289+
sinsp_container_lookup::state::STARTED);
290+
parse_containerd(request, cache);
291+
}
205292
return false;
206293
}
207294

295+
// Returning true will prevent other container engines from
296+
// trying to resolve the container, so only return true if we
297+
// have complete metadata.
298+
return container_info->is_successful();
299+
208300
tinfo->m_container_id = container_id;
209301

210302
libsinsp::cgroup_limits::cgroup_limits_key key(container.m_id,

‎userspace/libsinsp/container_engine/containerd.h‎

Lines changed: 67 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ class sinsp_container_info;
2222
class sinsp_threadinfo;
2323

2424
#include <libsinsp/container_engine/containerd/containers.grpc.pb.h>
25+
#include <libsinsp/container_engine/container_async_source.h>
2526
#include <libsinsp/container_engine/container_engine_base.h>
2627
#include <libsinsp/container_engine/sinsp_container_type.h>
2728

@@ -30,28 +31,85 @@ namespace ContainerdService = containerd::services::containers::v1;
3031
namespace libsinsp {
3132
namespace container_engine {
3233

33-
class containerd_interface {
34-
public:
35-
containerd_interface(const std::string &socket_path);
34+
struct containerd_lookup_request {
35+
containerd_lookup_request(): container_type(CT_CONTAINERD), uid(0), request_rw_size(false) {}
36+
37+
containerd_lookup_request(const std::string& container_id_value,
38+
sinsp_container_type container_type_value,
39+
unsigned long uid_value,
40+
bool rw_size_value):
41+
container_id(container_id_value),
42+
container_type(container_type_value),
43+
uid(uid_value),
44+
request_rw_size(rw_size_value) {}
45+
46+
bool operator<(const containerd_lookup_request& rhs) const {
47+
if(container_id != rhs.container_id) {
48+
return container_id < rhs.container_id;
49+
}
50+
51+
if(container_type != rhs.container_type) {
52+
return container_type < rhs.container_type;
53+
}
54+
55+
if(uid != rhs.uid) {
56+
return uid < rhs.uid;
57+
}
58+
59+
return request_rw_size < rhs.request_rw_size;
60+
}
61+
62+
bool operator==(const containerd_lookup_request& rhs) const {
63+
return container_id == rhs.container_id && container_type == rhs.container_type &&
64+
uid == rhs.uid && request_rw_size == rhs.request_rw_size;
65+
}
66+
67+
std::string container_id;
68+
sinsp_container_type container_type;
69+
unsigned long uid;
70+
bool request_rw_size;
71+
};
3672

37-
grpc::Status list_container_resp(const std::string &container_id,
38-
ContainerdService::ListContainersResponse &resp);
73+
class containerd_async_source : public container_async_source<containerd_lookup_request> {
74+
using key_type = containerd_lookup_request;
75+
76+
public:
77+
containerd_async_source(const std::string& socket_path,
78+
uint64_t max_wait_ms,
79+
uint64_t ttl_ms,
80+
container_cache_interface* cache);
81+
virtual ~containerd_async_source();
3982

83+
// TODO probably remove
4084
bool is_ok();
4185

4286
private:
87+
bool parse(const containerd_lookup_request& key, sinsp_container_info& container) override;
88+
89+
const char* name() const override { return "containerd"; };
90+
91+
sinsp_container_type container_type(const key_type& key) const override {
92+
return key.container_type;
93+
}
94+
std::string container_id(const key_type& key) const override { return key.container_id; }
95+
96+
grpc::Status list_container_resp(const std::string& container_id,
97+
ContainerdService::ListContainersResponse& resp);
98+
4399
std::unique_ptr<ContainerdService::Containers::Stub> m_stub;
100+
std::string m_socket_path;
44101
};
45102

46103
class containerd : public container_engine_base {
47104
public:
48-
containerd(container_cache_interface &cache);
105+
containerd(container_cache_interface& cache);
49106

50-
bool parse_containerd(sinsp_container_info &container, const std::string &container_id);
51-
bool resolve(sinsp_threadinfo *tinfo, bool query_os_for_missing_info) override;
107+
void parse_containerd(const containerd_lookup_request& request,
108+
container_cache_interface* cache);
109+
bool resolve(sinsp_threadinfo* tinfo, bool query_os_for_missing_info) override;
52110

53111
private:
54-
std::unique_ptr<containerd_interface> m_interface;
112+
std::unique_ptr<containerd_async_source> m_containerd_info_source;
55113
};
56114

57115
} // namespace container_engine

0 commit comments

Comments
 (0)