| 1 | // Copyright (C) Microsoft Corporation. All rights reserved. |
| 2 | |
| 3 | #include "precomp.h" |
| 4 | #include "EventStore.h" |
| 5 | #include "WSLCSession.h" |
| 6 | #include "WSLCExecutionContext.h" |
| 7 | #include <chrono> |
| 8 | |
| 9 | using wsl::shared::Localization; |
| 10 | |
| 11 | namespace wsl::windows::service::wslc { |
| 12 | |
| 13 | namespace { |
| 14 | |
| 15 | std::optional<std::chrono::sys_seconds> ToTimeBound(int64_t TimeSeconds) |
| 16 | { |
| 17 | if (TimeSeconds == 0) |
| 18 | { |
| 19 | return std::nullopt; |
| 20 | } |
| 21 | |
| 22 | // Waiting on a bound converts it to the system clock's 100ns ticks, which a far-future second would overflow. |
| 23 | constexpr auto c_maxBound = std::chrono::floor<std::chrono::seconds>(std::chrono::system_clock::time_point::max()); |
| 24 | return std::min(std::chrono::sys_seconds{std::chrono::seconds{TimeSeconds}}, c_maxBound); |
| 25 | } |
| 26 | |
| 27 | } // namespace |
| 28 | |
| 29 | void EventStore::Append(wsl::windows::common::wslc_schema::Event Event) |
| 30 | { |
| 31 | std::lock_guard lock(m_lock); |
| 32 | |
| 33 | // Events are recorded in Docker's delivery order, which is also timestamp order. Subscribers rely on |
| 34 | // this: they resume from a sequence number, so an out-of-order event could never be inserted where it |
| 35 | // belongs without hiding it from readers that already moved past that point. |
| 36 | WI_ASSERT(m_events.empty() || m_events.back().time <= Event.time); |
| 37 | |
| 38 | m_events.push_back(std::move(Event)); |
| 39 | |
| 40 | if (m_events.size() > c_eventRingCapacity) |
| 41 | { |
| 42 | m_events.pop_front(); |
| 43 | ++m_firstSequenceNumber; |
| 44 | } |
| 45 | |
| 46 | m_updated.notify_all(); |
| 47 | } |
| 48 | |
| 49 | void EventStore::Record(std::string&& Type, std::string&& Action, const std::string& ActorId, std::map<std::string, std::string> ActorAttributes, std::int64_t Time) noexcept |
| 50 | try |
| 51 | { |
| 52 | wsl::windows::common::wslc_schema::Event event; |
| 53 | event.Type = std::move(Type); |
| 54 | event.Action = std::move(Action); |
| 55 | event.Actor.ID = ActorId; |
| 56 | event.Actor.Attributes = std::move(ActorAttributes); |
| 57 | event.time = Time; |
| 58 | |
| 59 | Append(std::move(event)); |
| 60 | } |
| 61 | CATCH_LOG() |
| 62 | |
| 63 | namespace { |
| 64 | |
| 65 | // Values sharing a key are OR'd, distinct keys are AND'd. Unrecognized keys are ignored. |
| 66 | bool EventMatchesFilters(const wsl::windows::common::wslc_schema::Event& event, const std::map<std::string, std::vector<std::string>>& filters) |
| 67 | { |
| 68 | for (const auto& [key, values] : filters) |
| 69 | { |
| 70 | if (key == "type") |
| 71 | { |
| 72 | if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Type == v; })) |
| 73 | { |
| 74 | return false; |
| 75 | } |
| 76 | } |
| 77 | else if (key == "event") |
| 78 | { |
| 79 | if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Action == v; })) |
| 80 | { |
| 81 | return false; |
| 82 | } |
| 83 | } |
| 84 | else if (key == "container") |
| 85 | { |
| 86 | if (event.Type != "container" || |
| 87 | !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; })) |
| 88 | { |
| 89 | return false; |
| 90 | } |
| 91 | } |
| 92 | else if (key == "image") |
| 93 | { |
| 94 | if (event.Type != "image" || !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; })) |
| 95 | { |
| 96 | return false; |
| 97 | } |
| 98 | } |
| 99 | } |
| 100 | return true; |
| 101 | } |
| 102 | |
| 103 | } // namespace |
| 104 | |
| 105 | Microsoft::WRL::ComPtr<IWSLCEventStream> EventStore::CreateStream( |
| 106 | Microsoft::WRL::ComPtr<WSLCSession> Session, int64_t SinceTime, int64_t UntilTime, std::map<std::string, std::vector<std::string>> Filters) |
| 107 | { |
| 108 | // Zero means unbounded on that end, so it never makes the window run backwards. |
| 109 | THROW_HR_WITH_USER_ERROR_IF( |
| 110 | E_INVALIDARG, |
| 111 | Localization::MessageWslcEventsInvalidTimeWindow(SinceTime, UntilTime), |
| 112 | SinceTime < 0 || UntilTime < 0 || (SinceTime != 0 && UntilTime != 0 && SinceTime > UntilTime)); |
| 113 | |
| 114 | Microsoft::WRL::ComPtr<EventStream> stream; |
| 115 | THROW_IF_FAILED(Microsoft::WRL::MakeAndInitialize<EventStream>(&stream, std::move(Session), this, SinceTime, UntilTime, std::move(Filters))); |
| 116 | |
| 117 | return stream; |
| 118 | } |
| 119 | |
| 120 | std::optional<wsl::windows::common::wslc_schema::Event> EventStore::GetLockHeld(uint64_t SequenceNumber) |
| 121 | { |
| 122 | // Callers resync a lagging reader before reaching here, so the requested event is never evicted. |
| 123 | WI_ASSERT(SequenceNumber >= m_firstSequenceNumber); |
| 124 | |
| 125 | const uint64_t index = SequenceNumber - m_firstSequenceNumber; |
| 126 | if (index >= m_events.size()) |
| 127 | { |
| 128 | return std::nullopt; |
| 129 | } |
| 130 | |
| 131 | return m_events[index]; |
| 132 | } |
| 133 | |
| 134 | bool EventStore::WaitForEvent(std::unique_lock<std::mutex>& Lock, uint64_t SequenceNumber, std::optional<std::chrono::sys_seconds> Until) |
| 135 | { |
| 136 | // Ready once the reader's event is buffered, its slot is evicted, or the session terminates. |
| 137 | // Eviction while parked wakes us too, so the caller reports the gap on its next pass. |
| 138 | const auto ready = [&] { return m_terminating || SequenceNumber < m_firstSequenceNumber + m_events.size(); }; |
| 139 | |
| 140 | if (Until.has_value()) |
| 141 | { |
| 142 | if (!m_updated.wait_until(Lock, Until.value(), ready)) |
| 143 | { |
| 144 | return false; |
| 145 | } |
| 146 | } |
| 147 | else |
| 148 | { |
| 149 | m_updated.wait(Lock, ready); |
| 150 | } |
| 151 | |
| 152 | THROW_HR_IF(E_ABORT, m_terminating); |
| 153 | return true; |
| 154 | } |
| 155 | |
| 156 | std::optional<wsl::windows::common::wslc_schema::Event> EventStore::Get( |
| 157 | std::optional<uint64_t>& SequenceNumber, |
| 158 | std::optional<std::chrono::sys_seconds> Since, |
| 159 | std::optional<std::chrono::sys_seconds> Until, |
| 160 | const std::map<std::string, std::vector<std::string>>& Filters) |
| 161 | { |
| 162 | std::unique_lock lock(m_lock); |
| 163 | |
| 164 | // Position the reader. A first read (no sequence number yet) starts at the oldest buffered |
| 165 | // event |
| 166 | SequenceNumber = SequenceNumber.value_or(m_firstSequenceNumber); |
| 167 | |
| 168 | while (true) |
| 169 | { |
| 170 | // A reader that has fallen behind the ring missed events to eviction: reset it so the |
| 171 | // next call starts fresh at the oldest buffered event, and report the gap. |
| 172 | if (SequenceNumber.value() < m_firstSequenceNumber) |
| 173 | { |
| 174 | SequenceNumber = std::nullopt; |
| 175 | THROW_HR(WSLC_E_EVENTS_LOST); |
| 176 | } |
| 177 | |
| 178 | if (!WaitForEvent(lock, SequenceNumber.value(), Until)) |
| 179 | { |
| 180 | // The until window elapsed with no further event: the stream is finished. |
| 181 | return std::nullopt; |
| 182 | } |
| 183 | |
| 184 | // Evicted while parked: loop back to reset and report the gap. |
| 185 | // TODO: A burst of more than c_eventRingCapacity events between the wake and reacquiring the |
| 186 | // lock can evict this reader's event before it is read, forcing a WSLC_E_EVENTS_LOST. Redesign |
| 187 | // so that every parked reader is guaranteed to observe an event before the next write can evict |
| 188 | // it. |
| 189 | if (SequenceNumber.value() < m_firstSequenceNumber) |
| 190 | { |
| 191 | continue; |
| 192 | } |
| 193 | |
| 194 | const auto event = GetLockHeld(SequenceNumber.value()).value(); |
| 195 | const std::chrono::sys_seconds eventTime{std::chrono::seconds{event.time}}; |
| 196 | |
| 197 | // Advance in delivery order before applying the time window. |
| 198 | SequenceNumber.value()++; |
| 199 | |
| 200 | // Events are appended in non-decreasing timestamp order (see Append()), so once we reach the |
| 201 | // exclusive Until bound, the stream is finished. |
| 202 | if (Until.has_value() && eventTime >= Until.value()) |
| 203 | { |
| 204 | return std::nullopt; |
| 205 | } |
| 206 | |
| 207 | // Return the event if it falls within the since-bound and matches the caller's filters; |
| 208 | // otherwise loop to skip it. |
| 209 | if ((!Since.has_value() || eventTime >= Since.value()) && EventMatchesFilters(event, Filters)) |
| 210 | { |
| 211 | return event; |
| 212 | } |
| 213 | } |
| 214 | } |
| 215 | |
| 216 | void EventStore::OnSessionTerminating() |
| 217 | { |
| 218 | { |
| 219 | std::lock_guard lock(m_lock); |
| 220 | m_terminating = true; |
| 221 | } |
| 222 | |
| 223 | m_updated.notify_all(); |
| 224 | } |
| 225 | |
| 226 | HRESULT EventStream::RuntimeClassInitialize( |
| 227 | Microsoft::WRL::ComPtr<WSLCSession> Session, |
| 228 | EventStore* Store, |
| 229 | int64_t SinceTime, |
| 230 | int64_t UntilTime, |
| 231 | std::map<std::string, std::vector<std::string>> Filters) |
| 232 | { |
| 233 | m_session = std::move(Session); |
| 234 | m_store = Store; |
| 235 | m_since = ToTimeBound(SinceTime); |
| 236 | m_until = ToTimeBound(UntilTime); |
| 237 | m_filters = std::move(Filters); |
| 238 | return S_OK; |
| 239 | } |
| 240 | |
| 241 | HRESULT EventStream::GetNext(LPSTR* EventJson) |
| 242 | try |
| 243 | { |
| 244 | RETURN_HR_IF_NULL(E_POINTER, EventJson); |
| 245 | *EventJson = nullptr; |
| 246 | |
| 247 | std::lock_guard lock(m_lock); |
| 248 | const auto event = m_store->Get(m_nextSequenceNumber, m_since, m_until, m_filters); |
| 249 | if (!event.has_value()) |
| 250 | { |
| 251 | return WSLC_E_EVENT_STREAM_FINISHED; |
| 252 | } |
| 253 | |
| 254 | *EventJson = wil::make_unique_ansistring<wil::unique_cotaskmem_ansistring>(wsl::shared::ToJson(event.value()).c_str()).release(); |
| 255 | return S_OK; |
| 256 | } |
| 257 | CATCH_RETURN(); |
| 258 | |
| 259 | } // namespace wsl::windows::service::wslc |