@samitouri / QOSAMI-WSL-2
Code Commits Issues Pulls Sessions CI/CD
master
master
cpp 327 lines 10.5 KB
Raw
1 /*++
2
3 Copyright (c) Microsoft. All rights reserved.
4
5 Module Name:
6
7 DockerEventTracker.cpp
8
9 Abstract:
10
11 Contains the implementation of DockerEventTracker.
12
13 --*/
14 #include "precomp.h"
15 #include "DockerEventTracker.h"
16 #include "WSLCSession.h"
17 #include "WSLCVirtualMachine.h"
18 #include <nlohmann/json.hpp>
19
20 using wsl::windows::service::wslc::DockerEventTracker;
21 using wsl::windows::service::wslc::DockerHTTPClient;
22 using wsl::windows::service::wslc::WSLCSession;
23 using wsl::windows::service::wslc::WSLCVirtualMachine;
24
25 DockerEventTracker::EventTrackingReference::EventTrackingReference(DockerEventTracker* tracker, size_t id) noexcept :
26 m_tracker(tracker), m_id(id)
27 {
28 }
29
30 DockerEventTracker::EventTrackingReference& DockerEventTracker::EventTrackingReference::operator=(DockerEventTracker::EventTrackingReference&& other) noexcept
31 {
32 Reset();
33 m_id = other.m_id;
34 m_tracker = other.m_tracker;
35
36 other.m_tracker = nullptr;
37 other.m_id = {};
38
39 return *this;
40 }
41
42 void DockerEventTracker::EventTrackingReference::Reset() noexcept
43 {
44 if (m_tracker != nullptr)
45 {
46 m_tracker->UnregisterCallback(m_id);
47 m_tracker = nullptr;
48 m_id = {};
49 }
50 }
51
52 DockerEventTracker::EventTrackingReference::EventTrackingReference(EventTrackingReference&& other) noexcept :
53 m_id(other.m_id), m_tracker(other.m_tracker)
54 {
55 other.m_tracker = nullptr;
56 other.m_id = {};
57 }
58
59 DockerEventTracker::EventTrackingReference::~EventTrackingReference() noexcept
60 {
61 Reset();
62 }
63
64 DockerEventTracker::DockerEventTracker(WSLCSession& session) : m_session(session)
65 {
66 }
67
68 void DockerEventTracker::Connect(DockerHTTPClient& dockerClient, IORelay& relay)
69 {
70 auto onChunk = [this](const gsl::span<char>& buffer) {
71 if (!buffer.empty()) // docker inserts empty lines between events, skip those.
72 {
73 try
74 {
75 OnEvent(std::string_view(buffer.data(), buffer.size()));
76 }
77 catch (...)
78 {
79 WSL_LOG(
80 "DockerEventParseError",
81 TraceLoggingCountedString(
82 buffer.data(), static_cast<UINT16>(std::min(buffer.size(), static_cast<size_t>(USHRT_MAX))), "Data"),
83 TraceLoggingValue(wil::ResultFromCaughtException(), "Error"),
84 TraceLoggingValue(m_session.Id(), "SessionId"));
85 }
86 }
87 };
88
89 auto socket = dockerClient.MonitorEvents();
90
91 relay.AddHandle(std::make_unique<common::io::HTTPChunkBasedReadHandle>(std::move(socket), std::move(onChunk)));
92 }
93
94 DockerEventTracker::~DockerEventTracker()
95 {
96 // N.B. No callback should be left when the tracker is destroyed.
97 WI_ASSERT(m_containerCallbacks.empty());
98 WI_ASSERT(m_volumeCallbacks.empty());
99 WI_ASSERT(m_containerCreateCallbacks.empty());
100 }
101
102 void DockerEventTracker::OnEvent(const std::string_view& event)
103 {
104 WSL_LOG(
105 "DockerEvent",
106 TraceLoggingCountedString(
107 event.data(), static_cast<UINT16>(std::min(event.size(), static_cast<size_t>(USHRT_MAX))), "Data"),
108 TraceLoggingValue(m_session.Id(), "SessionId"));
109
110 auto parsed = nlohmann::json::parse(event);
111
112 auto action = parsed.find("Action");
113 THROW_HR_IF_MSG(E_INVALIDARG, action == parsed.end(), "Failed to parse json: %.*hs", static_cast<int>(event.size()), event.data());
114
115 auto timeEntry = parsed.find("time");
116 THROW_HR_IF_MSG(
117 E_INVALIDARG, timeEntry == parsed.end(), "Failed to parse time from event: %.*hs", static_cast<int>(event.size()), event.data());
118 std::int64_t eventTime = timeEntry->get<std::int64_t>();
119
120 auto actionStr = action->get<std::string>();
121
122 // Route events by Type field. Docker uses "container", "volume", "network", etc.
123 auto type = parsed.find("Type");
124 std::string typeStr = (type != parsed.end()) ? type->get<std::string>() : "container";
125
126 if (typeStr == "container")
127 {
128 OnContainerEvent(parsed, actionStr, eventTime);
129
130 if (actionStr == "create")
131 {
132 OnContainerCreated(parsed, eventTime);
133 }
134 }
135 else if (typeStr == "volume")
136 {
137 OnVolumeEvent(parsed, actionStr, eventTime);
138 }
139 }
140
141 void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime)
142 {
143 static std::map<std::string, ContainerEvent> events{
144 {"start", ContainerEvent::Start},
145 {"die", ContainerEvent::Stop},
146 {"kill", ContainerEvent::Kill},
147 {"destroy", ContainerEvent::Destroy},
148 {"exec_die", ContainerEvent::ExecDied},
149 {"restart", ContainerEvent::Restart}};
150
151 auto actor = parsed.find("Actor");
152 THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in container event");
153
154 auto id = actor->find("ID");
155 THROW_HR_IF_MSG(E_INVALIDARG, id == actor->end(), "Missing Actor.ID in container event");
156
157 auto containerId = id->get<std::string>();
158
159 auto it = events.find(action);
160 if (it == events.end())
161 {
162 return; // Event is not tracked, dropped.
163 }
164
165 std::optional<int> exitCode;
166 std::optional<std::string> execId;
167 auto attributes = actor->find("Attributes");
168 if (attributes != actor->end())
169 {
170 auto exitCodeEntry = attributes->find("exitCode");
171 if (exitCodeEntry != attributes->end())
172 {
173 exitCode = std::stoi(exitCodeEntry->get<std::string>());
174 }
175
176 auto execIdEntry = attributes->find("execID");
177 if (execIdEntry != attributes->end())
178 {
179 execId = execIdEntry->get<std::string>();
180 }
181 }
182
183 // Snapshot the matching callbacks so that they can be invoked without holding m_lock. Callbacks can register and
184 // unregister callbacks (a container that stops releases its exec processes), which would otherwise mutate the
185 // vector being iterated.
186 std::vector<std::shared_ptr<ContainerCallback>> callbacks;
187 {
188 std::lock_guard lock{m_lock};
189
190 for (const auto& e : m_containerCallbacks)
191 {
192 if (e->ContainerId == containerId && (!e->ExecId.has_value() || e->ExecId == execId))
193 {
194 callbacks.emplace_back(e);
195 }
196 }
197 }
198
199 InvokeCallbacks(callbacks, [&](const ContainerCallback& e) { e.Callback(it->second, exitCode, eventTime); });
200 }
201
202 void DockerEventTracker::OnVolumeEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime)
203 {
204 static std::map<std::string, VolumeEvent> events{{"create", VolumeEvent::Create}, {"destroy", VolumeEvent::Destroy}};
205
206 auto it = events.find(action);
207 if (it == events.end())
208 {
209 return; // Event is not tracked, dropped.
210 }
211
212 auto actor = parsed.find("Actor");
213 THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in volume event");
214
215 auto id = actor->find("ID");
216 THROW_HR_IF_MSG(E_INVALIDARG, id == actor->end(), "Missing Actor.ID in volume event");
217
218 auto volumeName = id->get<std::string>();
219
220 std::vector<std::shared_ptr<VolumeCallback>> callbacks;
221 {
222 std::lock_guard lock{m_lock};
223 callbacks = m_volumeCallbacks;
224 }
225
226 InvokeCallbacks(callbacks, [&](const VolumeCallback& e) { e.Callback(volumeName, it->second, eventTime); });
227 }
228
229 void DockerEventTracker::OnContainerCreated(const nlohmann::json& parsed, std::int64_t eventTime)
230 {
231 auto actor = parsed.find("Actor");
232 THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in container event");
233
234 auto id = actor->find("ID");
235 THROW_HR_IF_MSG(E_INVALIDARG, id == actor->end(), "Missing Actor.ID in container event");
236
237 auto containerId = id->get<std::string>();
238
239 std::vector<std::shared_ptr<ContainerCreateCallbackEntry>> callbacks;
240 {
241 std::lock_guard lock{m_lock};
242 callbacks = m_containerCreateCallbacks;
243 }
244
245 InvokeCallbacks(callbacks, [&](const ContainerCreateCallbackEntry& e) { e.Callback(containerId, eventTime); });
246 }
247
248 DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerStateUpdates(
249 const std::string& ContainerId, ContainerStateChangeCallback&& Callback) noexcept
250 {
251 auto id = m_callbackId++;
252 auto entry = std::make_shared<ContainerCallback>(id, std::string{ContainerId}, std::optional<std::string>{}, std::move(Callback));
253
254 std::lock_guard lock{m_lock};
255 m_containerCallbacks.emplace_back(std::move(entry));
256
257 return EventTrackingReference{this, id};
258 }
259
260 DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterExecStateUpdates(
261 const std::string& ContainerId, const std::string& ExecId, ContainerStateChangeCallback&& Callback) noexcept
262 {
263 auto id = m_callbackId++;
264 auto entry = std::make_shared<ContainerCallback>(id, std::string{ContainerId}, std::optional<std::string>{ExecId}, std::move(Callback));
265
266 std::lock_guard lock{m_lock};
267 m_containerCallbacks.emplace_back(std::move(entry));
268
269 return EventTrackingReference{this, id};
270 }
271
272 DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterVolumeUpdates(VolumeEventCallback&& Callback) noexcept
273 {
274 auto id = m_callbackId++;
275 auto entry = std::make_shared<VolumeCallback>(id, std::move(Callback));
276
277 std::lock_guard lock{m_lock};
278 m_volumeCallbacks.emplace_back(std::move(entry));
279
280 return EventTrackingReference{this, id};
281 }
282
283 DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerCreate(ContainerCreateCallback&& Callback) noexcept
284 {
285 auto id = m_callbackId++;
286 auto entry = std::make_shared<ContainerCreateCallbackEntry>(id, std::move(Callback));
287
288 std::lock_guard lock{m_lock};
289 m_containerCreateCallbacks.emplace_back(std::move(entry));
290
291 return EventTrackingReference{this, id};
292 }
293
294 void DockerEventTracker::UnregisterCallback(size_t Id) noexcept
295 {
296 std::shared_ptr<CallbackRegistration> registration;
297
298 {
299 std::lock_guard lock{m_lock};
300
301 auto matches = [Id](const auto& e) { return e->CallbackId == Id; };
302
303 auto take = [&](auto& Callbacks) {
304 auto entry = std::ranges::find_if(Callbacks, matches);
305 if (entry == Callbacks.end())
306 {
307 return false;
308 }
309
310 registration = std::move(*entry);
311 Callbacks.erase(entry);
312 return true;
313 };
314
315 if (!take(m_containerCallbacks) && !take(m_volumeCallbacks) && !take(m_containerCreateCallbacks))
316 {
317 WI_ASSERT(false);
318 }
319 }
320
321 if (registration)
322 {
323 // Wait for any in-flight invocation to complete so the callback can't run once this returns.
324 std::lock_guard invokeLock{registration->InvokeLock};
325 registration->Unregistered = true;
326 }
327 }