diff --git a/doc/docs/api-reference/c/error-codes.md b/doc/docs/api-reference/c/error-codes.md index 91e22551..e704cf8c 100644 --- a/doc/docs/api-reference/c/error-codes.md +++ b/doc/docs/api-reference/c/error-codes.md @@ -18,6 +18,8 @@ #define WSLC_E_VOLUME_NOT_AVAILABLE MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 14) /* 0x8004060E */ #define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 15) /* 0x8004060F */ #define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */ +#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */ +#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */ ``` | Symbol | Hex Value | @@ -39,5 +41,7 @@ | `WSLC_E_VOLUME_NOT_AVAILABLE` | `0x8004060E` | | `WSLC_E_SESSION_NOT_FOUND` | `0x8004060F` | | `WSLC_E_VM_NOT_RUNNING` | `0x80040610` | +| `WSLC_E_EVENTS_LOST` | `0x80040611` | +| `WSLC_E_EVENT_STREAM_FINISHED` | `0x80040612` | --- diff --git a/localization/strings/en-US/Resources.resw b/localization/strings/en-US/Resources.resw index 8db58645..24c395e3 100644 --- a/localization/strings/en-US/Resources.resw +++ b/localization/strings/en-US/Resources.resw @@ -2371,6 +2371,10 @@ For privacy information about this product please visit https://aka.ms/privacy.< Invalid name: '{}' {FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated + + `since` time ({}) cannot be after `until` time ({}) + {FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated{Locked="`since`"}{Locked="`until`"} + Path is not absolute: '{}' {FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated diff --git a/msipackage/package.wix.in b/msipackage/package.wix.in index 38b01cd2..aea50478 100644 --- a/msipackage/package.wix.in +++ b/msipackage/package.wix.in @@ -359,6 +359,14 @@ + + + + + + + + diff --git a/src/windows/WslcSDK/wslcsdk.h b/src/windows/WslcSDK/wslcsdk.h index 56f05a61..5b86444f 100644 --- a/src/windows/WslcSDK/wslcsdk.h +++ b/src/windows/WslcSDK/wslcsdk.h @@ -44,6 +44,8 @@ EXTERN_C_START #define WSLC_E_VOLUME_NOT_AVAILABLE MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 14) /* 0x8004060E */ #define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 15) /* 0x8004060F */ #define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */ +#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */ +#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */ // Session values #define WSLC_SESSION_OPTIONS_SIZE 72 diff --git a/src/windows/common/wslutil.cpp b/src/windows/common/wslutil.cpp index 50065721..027304fc 100644 --- a/src/windows/common/wslutil.cpp +++ b/src/windows/common/wslutil.cpp @@ -180,6 +180,8 @@ static const std::map g_commonErrors{ X(WSLC_E_NETWORK_NOT_FOUND), X(WSLC_E_SESSION_NOT_FOUND), X(WSLC_E_VM_NOT_RUNNING), + X(WSLC_E_EVENTS_LOST), + X(WSLC_E_EVENT_STREAM_FINISHED), X(WSLC_E_WU_SEARCH_FAILED), X_WIN32(RPC_S_SERVER_UNAVAILABLE), X_WIN32(ERROR_ELEVATION_REQUIRED), diff --git a/src/windows/inc/wslc_schema.h b/src/windows/inc/wslc_schema.h index 0df438a7..023e0334 100644 --- a/src/windows/inc/wslc_schema.h +++ b/src/windows/inc/wslc_schema.h @@ -348,4 +348,22 @@ struct VolumeListEntry NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(VolumeListEntry, Name, Driver, Mountpoint, Scope, Labels); }; +struct EventActor +{ + std::string ID; + std::map Attributes; + + NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(EventActor, ID, Attributes); +}; + +struct Event +{ + std::string Type; + std::string Action; + EventActor Actor; + std::int64_t time{}; + + NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(Event, Type, Action, Actor, time); +}; + } // namespace wsl::windows::common::wslc_schema diff --git a/src/windows/service/inc/wslc.idl b/src/windows/service/inc/wslc.idl index c4ee8321..dd2c165f 100644 --- a/src/windows/service/inc/wslc.idl +++ b/src/windows/service/inc/wslc.idl @@ -671,6 +671,18 @@ typedef struct _WSLCListContainersOptions ULONG FiltersCount; } WSLCListContainersOptions; +[ + uuid(7EC66D3B-D098-4D48-B69E-69166F6C4745), + pointer_default(unique), + object +] +interface IWSLCEventStream : IUnknown +{ + // Blocks until the next matching event, the until-time is reached, or the session terminates, + // then returns the event as a JSON object following the wslc_schema::Event format. + HRESULT GetNext([out, string] LPSTR* EventJson); +} + // Settings for IWSLCSession::Initialize - passed from service to per-user process typedef struct _WSLCSessionInitSettings { @@ -708,6 +720,18 @@ interface IWSLCSession : IUnknown // termination event has been signaled; before that the call fails. HRESULT GetTerminationReason([out] WSLCVirtualMachineTerminationReason* Reason, [out] LPWSTR* Details); + // Opens an event stream that mirrors `docker events`. The window and filters are captured by the + // returned stream object; events are then pulled one at a time via IWSLCEventStream::GetNext. + // SinceTime / UntilTime bound the window by event time, in seconds since the Unix epoch. SinceTime + // is inclusive and UntilTime is exclusive. 0 means unbounded on that end. + // Filters are key/value pairs. Values sharing a key are OR'd, distinct keys are AND'd. + HRESULT GetEvents( + [in] LONGLONG SinceTime, + [in] LONGLONG UntilTime, + [in, unique, size_is(FiltersCount)] const WSLCFilter* Filters, + [in] ULONG FiltersCount, + [out] IWSLCEventStream** Stream); + // Image management. HRESULT PullImage([in] LPCSTR Image, [in, unique] LPCSTR RegistryAuthenticationInformation, [in, unique] IProgressCallback* ProgressCallback, [in, unique] IWarningCallback* WarningCallback); HRESULT BuildImage([in] const WSLCBuildImageOptions* Options, [in, unique] IProgressCallback* ProgressCallback, [in, unique, system_handle(sh_event)] HANDLE CancelEvent); @@ -885,3 +909,5 @@ cpp_quote("#define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILIT // N.B. WSLC_E_VM_NOT_RUNNING is part of the plugin API contract and is also defined in WslPluginApi.h. // The two definitions must stay in sync. cpp_quote("#define WSLC_E_VM_NOT_RUNNING MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */") +cpp_quote("#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */") +cpp_quote("#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 18) /* 0x80040612 */") diff --git a/src/windows/wslcsession/CMakeLists.txt b/src/windows/wslcsession/CMakeLists.txt index 50c80365..da076197 100644 --- a/src/windows/wslcsession/CMakeLists.txt +++ b/src/windows/wslcsession/CMakeLists.txt @@ -26,6 +26,7 @@ set(SOURCES # Supporting classes DockerEventTracker.cpp DockerHTTPClient.cpp + EventStore.cpp IORelay.cpp OptionParser.cpp ServiceProcessLauncher.cpp @@ -35,6 +36,7 @@ set(SOURCES set(HEADERS DockerEventTracker.h DockerHTTPClient.h + EventStore.h IORelay.h OptionParser.h ServiceProcessLauncher.h diff --git a/src/windows/wslcsession/DockerEventTracker.cpp b/src/windows/wslcsession/DockerEventTracker.cpp index a809ddc7..9a5868f8 100644 --- a/src/windows/wslcsession/DockerEventTracker.cpp +++ b/src/windows/wslcsession/DockerEventTracker.cpp @@ -96,6 +96,7 @@ DockerEventTracker::~DockerEventTracker() // N.B. No callback should be left when the tracker is destroyed. WI_ASSERT(m_containerCallbacks.empty()); WI_ASSERT(m_volumeCallbacks.empty()); + WI_ASSERT(m_containerCreateCallbacks.empty()); } void DockerEventTracker::OnEvent(const std::string_view& event) @@ -125,33 +126,16 @@ void DockerEventTracker::OnEvent(const std::string_view& event) if (typeStr == "container") { OnContainerEvent(parsed, actionStr, eventTime); + + if (actionStr == "create") + { + OnContainerCreated(parsed, eventTime); + } } else if (typeStr == "volume") { OnVolumeEvent(parsed, actionStr, eventTime); } - - // Track object creation for WaitForObjectCreated. - auto actor = parsed.find("Actor"); - if (actor != parsed.end()) - { - auto id = actor->find("ID"); - if (id != actor->end()) - { - auto objectId = id->get(); - if (actionStr == "create") - { - std::lock_guard lock{m_lock}; - m_createdObjects.insert(objectId); - m_objectCreated.SetEvent(); - } - else if (actionStr == "destroy") - { - std::lock_guard lock{m_lock}; - m_createdObjects.erase(objectId); - } - } - } } void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime) @@ -159,6 +143,7 @@ void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const st static std::map events{ {"start", ContainerEvent::Start}, {"die", ContainerEvent::Stop}, + {"kill", ContainerEvent::Kill}, {"destroy", ContainerEvent::Destroy}, {"exec_die", ContainerEvent::ExecDied}, {"restart", ContainerEvent::Restart}}; @@ -241,30 +226,23 @@ void DockerEventTracker::OnVolumeEvent(const nlohmann::json& parsed, const std:: InvokeCallbacks(callbacks, [&](const VolumeCallback& e) { e.Callback(volumeName, it->second, eventTime); }); } -void DockerEventTracker::WaitForObjectCreated(const std::string& ObjectId) +void DockerEventTracker::OnContainerCreated(const nlohmann::json& parsed, std::int64_t eventTime) { - constexpr auto c_timeout = std::chrono::seconds{60}; + auto actor = parsed.find("Actor"); + THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in container event"); - while (true) + auto id = actor->find("ID"); + THROW_HR_IF_MSG(E_INVALIDARG, id == actor->end(), "Missing Actor.ID in container event"); + + auto containerId = id->get(); + + std::vector> callbacks; { - { - std::lock_guard lock{m_lock}; - if (m_createdObjects.contains(ObjectId)) - { - return; - } - - // Reset under the lock so a concurrent OnEvent() that runs after we release the lock - // and before the wait can re-signal the event and unblock us. - m_objectCreated.ResetEvent(); - } - - THROW_HR_IF_MSG( - HRESULT_FROM_WIN32(ERROR_TIMEOUT), - !m_session.WaitForEventOrSessionTerminating(m_objectCreated.get(), c_timeout), - "Timed out waiting for Docker create event for object '%hs'", - ObjectId.c_str()); + std::lock_guard lock{m_lock}; + callbacks = m_containerCreateCallbacks; } + + InvokeCallbacks(callbacks, [&](const ContainerCreateCallbackEntry& e) { e.Callback(containerId, eventTime); }); } DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerStateUpdates( @@ -302,6 +280,17 @@ DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterVolumeUpd return EventTrackingReference{this, id}; } +DockerEventTracker::EventTrackingReference DockerEventTracker::RegisterContainerCreate(ContainerCreateCallback&& Callback) noexcept +{ + auto id = m_callbackId++; + auto entry = std::make_shared(id, std::move(Callback)); + + std::lock_guard lock{m_lock}; + m_containerCreateCallbacks.emplace_back(std::move(entry)); + + return EventTrackingReference{this, id}; +} + void DockerEventTracker::UnregisterCallback(size_t Id) noexcept { std::shared_ptr registration; @@ -311,22 +300,21 @@ void DockerEventTracker::UnregisterCallback(size_t Id) noexcept auto matches = [Id](const auto& e) { return e->CallbackId == Id; }; - // Try container callbacks first, then volume callbacks. - if (auto container = std::ranges::find_if(m_containerCallbacks, matches); container != m_containerCallbacks.end()) - { - registration = std::move(*container); - m_containerCallbacks.erase(container); - } - else - { - auto volume = std::ranges::find_if(m_volumeCallbacks, matches); - WI_ASSERT(volume != m_volumeCallbacks.end()); - - if (volume != m_volumeCallbacks.end()) + auto take = [&](auto& Callbacks) { + auto entry = std::ranges::find_if(Callbacks, matches); + if (entry == Callbacks.end()) { - registration = std::move(*volume); - m_volumeCallbacks.erase(volume); + return false; } + + registration = std::move(*entry); + Callbacks.erase(entry); + return true; + }; + + if (!take(m_containerCallbacks) && !take(m_volumeCallbacks) && !take(m_containerCreateCallbacks)) + { + WI_ASSERT(false); } } diff --git a/src/windows/wslcsession/DockerEventTracker.h b/src/windows/wslcsession/DockerEventTracker.h index f65fdea0..812a62f5 100644 --- a/src/windows/wslcsession/DockerEventTracker.h +++ b/src/windows/wslcsession/DockerEventTracker.h @@ -30,7 +30,8 @@ enum class ContainerEvent Stop, Exit, Destroy, - ExecDied + ExecDied, + Kill }; enum class VolumeEvent @@ -64,6 +65,7 @@ public: using ContainerStateChangeCallback = std::function, std::int64_t)>; using VolumeEventCallback = std::function; + using ContainerCreateCallback = std::function; explicit DockerEventTracker(WSLCSession& session); ~DockerEventTracker(); @@ -76,13 +78,16 @@ public: EventTrackingReference RegisterContainerStateUpdates(const std::string& ContainerId, ContainerStateChangeCallback&& Callback) noexcept; EventTrackingReference RegisterExecStateUpdates(const std::string& ContainerId, const std::string& ExecId, ContainerStateChangeCallback&& Callback) noexcept; EventTrackingReference RegisterVolumeUpdates(VolumeEventCallback&& Callback) noexcept; - void UnregisterCallback(size_t Id) noexcept; - void WaitForObjectCreated(const std::string& ObjectId); + // Invoked for every container create event, after the per-container state callbacks. Unlike those, + // this isn't keyed by container id, because the id isn't known until Docker assigns it. + EventTrackingReference RegisterContainerCreate(ContainerCreateCallback&& Callback) noexcept; + void UnregisterCallback(size_t Id) noexcept; private: void OnEvent(const std::string_view& event); void OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime); + void OnContainerCreated(const nlohmann::json& parsed, std::int64_t eventTime); void OnVolumeEvent(const nlohmann::json& parsed, const std::string& action, std::int64_t eventTime); // Callbacks are invoked without holding m_lock so that a callback can register or unregister callbacks, and so @@ -125,8 +130,19 @@ private: const VolumeEventCallback Callback; }; + struct ContainerCreateCallbackEntry : CallbackRegistration + { + ContainerCreateCallbackEntry(size_t Id, ContainerCreateCallback&& Callback) : + CallbackRegistration(Id), Callback(std::move(Callback)) + { + } + + const ContainerCreateCallback Callback; + }; + _Guarded_by_(m_lock) std::vector> m_containerCallbacks; _Guarded_by_(m_lock) std::vector> m_volumeCallbacks; + _Guarded_by_(m_lock) std::vector> m_containerCreateCallbacks; // Invokes a snapshot of callbacks taken under m_lock, skipping registrations that have since been unregistered. template @@ -142,9 +158,6 @@ private: } } - _Guarded_by_(m_lock) std::unordered_set m_createdObjects; - _Guarded_by_(m_lock) wil::unique_event m_objectCreated { wil::EventOptions::ManualReset }; - WSLCSession& m_session; std::mutex m_lock; std::atomic m_callbackId{0}; diff --git a/src/windows/wslcsession/EventStore.cpp b/src/windows/wslcsession/EventStore.cpp new file mode 100644 index 00000000..6a2b19cc --- /dev/null +++ b/src/windows/wslcsession/EventStore.cpp @@ -0,0 +1,259 @@ +// Copyright (C) Microsoft Corporation. All rights reserved. + +#include "precomp.h" +#include "EventStore.h" +#include "WSLCSession.h" +#include "WSLCExecutionContext.h" +#include + +using wsl::shared::Localization; + +namespace wsl::windows::service::wslc { + +namespace { + + std::optional ToTimeBound(int64_t TimeSeconds) + { + if (TimeSeconds == 0) + { + return std::nullopt; + } + + // Waiting on a bound converts it to the system clock's 100ns ticks, which a far-future second would overflow. + constexpr auto c_maxBound = std::chrono::floor(std::chrono::system_clock::time_point::max()); + return std::min(std::chrono::sys_seconds{std::chrono::seconds{TimeSeconds}}, c_maxBound); + } + +} // namespace + +void EventStore::Append(wsl::windows::common::wslc_schema::Event Event) +{ + std::lock_guard lock(m_lock); + + // Events are recorded in Docker's delivery order, which is also timestamp order. Subscribers rely on + // this: they resume from a sequence number, so an out-of-order event could never be inserted where it + // belongs without hiding it from readers that already moved past that point. + WI_ASSERT(m_events.empty() || m_events.back().time <= Event.time); + + m_events.push_back(std::move(Event)); + + if (m_events.size() > c_eventRingCapacity) + { + m_events.pop_front(); + ++m_firstSequenceNumber; + } + + m_updated.notify_all(); +} + +void EventStore::Record(std::string&& Type, std::string&& Action, const std::string& ActorId, std::map ActorAttributes, std::int64_t Time) noexcept +try +{ + wsl::windows::common::wslc_schema::Event event; + event.Type = std::move(Type); + event.Action = std::move(Action); + event.Actor.ID = ActorId; + event.Actor.Attributes = std::move(ActorAttributes); + event.time = Time; + + Append(std::move(event)); +} +CATCH_LOG() + +namespace { + + // Values sharing a key are OR'd, distinct keys are AND'd. Unrecognized keys are ignored. + bool EventMatchesFilters(const wsl::windows::common::wslc_schema::Event& event, const std::map>& filters) + { + for (const auto& [key, values] : filters) + { + if (key == "type") + { + if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Type == v; })) + { + return false; + } + } + else if (key == "event") + { + if (!std::ranges::any_of(values, [&](const std::string& v) { return event.Action == v; })) + { + return false; + } + } + else if (key == "container") + { + if (event.Type != "container" || + !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; })) + { + return false; + } + } + else if (key == "image") + { + if (event.Type != "image" || !std::ranges::any_of(values, [&](const std::string& v) { return event.Actor.ID == v; })) + { + return false; + } + } + } + return true; + } + +} // namespace + +Microsoft::WRL::ComPtr EventStore::CreateStream( + Microsoft::WRL::ComPtr Session, int64_t SinceTime, int64_t UntilTime, std::map> Filters) +{ + // Zero means unbounded on that end, so it never makes the window run backwards. + THROW_HR_WITH_USER_ERROR_IF( + E_INVALIDARG, + Localization::MessageWslcEventsInvalidTimeWindow(SinceTime, UntilTime), + SinceTime < 0 || UntilTime < 0 || (SinceTime != 0 && UntilTime != 0 && SinceTime > UntilTime)); + + Microsoft::WRL::ComPtr stream; + THROW_IF_FAILED(Microsoft::WRL::MakeAndInitialize(&stream, std::move(Session), this, SinceTime, UntilTime, std::move(Filters))); + + return stream; +} + +std::optional EventStore::GetLockHeld(uint64_t SequenceNumber) +{ + // Callers resync a lagging reader before reaching here, so the requested event is never evicted. + WI_ASSERT(SequenceNumber >= m_firstSequenceNumber); + + const uint64_t index = SequenceNumber - m_firstSequenceNumber; + if (index >= m_events.size()) + { + return std::nullopt; + } + + return m_events[index]; +} + +bool EventStore::WaitForEvent(std::unique_lock& Lock, uint64_t SequenceNumber, std::optional Until) +{ + // Ready once the reader's event is buffered, its slot is evicted, or the session terminates. + // Eviction while parked wakes us too, so the caller reports the gap on its next pass. + const auto ready = [&] { return m_terminating || SequenceNumber < m_firstSequenceNumber + m_events.size(); }; + + if (Until.has_value()) + { + if (!m_updated.wait_until(Lock, Until.value(), ready)) + { + return false; + } + } + else + { + m_updated.wait(Lock, ready); + } + + THROW_HR_IF(E_ABORT, m_terminating); + return true; +} + +std::optional EventStore::Get( + std::optional& SequenceNumber, + std::optional Since, + std::optional Until, + const std::map>& Filters) +{ + std::unique_lock lock(m_lock); + + // Position the reader. A first read (no sequence number yet) starts at the oldest buffered + // event + SequenceNumber = SequenceNumber.value_or(m_firstSequenceNumber); + + while (true) + { + // A reader that has fallen behind the ring missed events to eviction: reset it so the + // next call starts fresh at the oldest buffered event, and report the gap. + if (SequenceNumber.value() < m_firstSequenceNumber) + { + SequenceNumber = std::nullopt; + THROW_HR(WSLC_E_EVENTS_LOST); + } + + if (!WaitForEvent(lock, SequenceNumber.value(), Until)) + { + // The until window elapsed with no further event: the stream is finished. + return std::nullopt; + } + + // Evicted while parked: loop back to reset and report the gap. + // TODO: A burst of more than c_eventRingCapacity events between the wake and reacquiring the + // lock can evict this reader's event before it is read, forcing a WSLC_E_EVENTS_LOST. Redesign + // so that every parked reader is guaranteed to observe an event before the next write can evict + // it. + if (SequenceNumber.value() < m_firstSequenceNumber) + { + continue; + } + + const auto event = GetLockHeld(SequenceNumber.value()).value(); + const std::chrono::sys_seconds eventTime{std::chrono::seconds{event.time}}; + + // Advance in delivery order before applying the time window. + SequenceNumber.value()++; + + // Events are appended in non-decreasing timestamp order (see Append()), so once we reach the + // exclusive Until bound, the stream is finished. + if (Until.has_value() && eventTime >= Until.value()) + { + return std::nullopt; + } + + // Return the event if it falls within the since-bound and matches the caller's filters; + // otherwise loop to skip it. + if ((!Since.has_value() || eventTime >= Since.value()) && EventMatchesFilters(event, Filters)) + { + return event; + } + } +} + +void EventStore::OnSessionTerminating() +{ + { + std::lock_guard lock(m_lock); + m_terminating = true; + } + + m_updated.notify_all(); +} + +HRESULT EventStream::RuntimeClassInitialize( + Microsoft::WRL::ComPtr Session, + EventStore* Store, + int64_t SinceTime, + int64_t UntilTime, + std::map> Filters) +{ + m_session = std::move(Session); + m_store = Store; + m_since = ToTimeBound(SinceTime); + m_until = ToTimeBound(UntilTime); + m_filters = std::move(Filters); + return S_OK; +} + +HRESULT EventStream::GetNext(LPSTR* EventJson) +try +{ + RETURN_HR_IF_NULL(E_POINTER, EventJson); + *EventJson = nullptr; + + std::lock_guard lock(m_lock); + const auto event = m_store->Get(m_nextSequenceNumber, m_since, m_until, m_filters); + if (!event.has_value()) + { + return WSLC_E_EVENT_STREAM_FINISHED; + } + + *EventJson = wil::make_unique_ansistring(wsl::shared::ToJson(event.value()).c_str()).release(); + return S_OK; +} +CATCH_RETURN(); + +} // namespace wsl::windows::service::wslc diff --git a/src/windows/wslcsession/EventStore.h b/src/windows/wslcsession/EventStore.h new file mode 100644 index 00000000..b47aed6b --- /dev/null +++ b/src/windows/wslcsession/EventStore.h @@ -0,0 +1,88 @@ +// Copyright (C) Microsoft Corporation. All rights reserved. + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include "wslc.h" +#include "wslc_schema.h" + +namespace wsl::windows::service::wslc { + +class WSLCSession; + +class EventStore +{ +public: + static constexpr size_t c_eventRingCapacity = 256; + + void Record(std::string&& Type, std::string&& Action, const std::string& ActorId, std::map ActorAttributes, std::int64_t Time) noexcept; + + Microsoft::WRL::ComPtr CreateStream( + Microsoft::WRL::ComPtr Session, int64_t SinceTime, int64_t UntilTime, std::map> Filters); + + // Returns the next event at or after SequenceNumber that falls within [Since, Until) and matches + // Filters, advancing SequenceNumber past it. A nullopt SequenceNumber starts a fresh reader at the + // oldest buffered event (no gap is reported). If the reader has since fallen behind the ring, + // resyncs SequenceNumber to the oldest buffered event and throws WSLC_E_EVENTS_LOST. Returns + // nullopt once the Until window has closed. + std::optional Get( + std::optional& SequenceNumber, + std::optional Since, + std::optional Until, + const std::map>& Filters); + + void OnSessionTerminating(); + +private: + void Append(wsl::windows::common::wslc_schema::Event Event); + + // Blocks until the event at SequenceNumber is buffered, its slot is evicted, or the session + // terminates. Returns false only when Until elapsed with no event ready. Throws E_ABORT if the + // session terminated while waiting. + bool WaitForEvent(std::unique_lock& Lock, uint64_t SequenceNumber, std::optional Until); + + std::optional GetLockHeld(uint64_t SequenceNumber); + + std::mutex m_lock; + std::condition_variable m_updated; + + _Guarded_by_(m_lock) std::deque m_events; + _Guarded_by_(m_lock) uint64_t m_firstSequenceNumber = 1; + + _Guarded_by_(m_lock) bool m_terminating = false; +}; + +class EventStream + : public Microsoft::WRL::RuntimeClass, IWSLCEventStream, IFastRundown> +{ +public: + HRESULT RuntimeClassInitialize( + Microsoft::WRL::ComPtr Session, + EventStore* Store, + int64_t SinceTime, + int64_t UntilTime, + std::map> Filters); + + IFACEMETHOD(GetNext)(_Outptr_result_z_ LPSTR* EventJson) override; + +private: + Microsoft::WRL::ComPtr m_session; + EventStore* m_store = nullptr; + + std::optional m_since; + std::optional m_until; + std::map> m_filters; + + std::mutex m_lock; + _Guarded_by_(m_lock) std::optional m_nextSequenceNumber; +}; + +} // namespace wsl::windows::service::wslc diff --git a/src/windows/wslcsession/WSLCContainer.cpp b/src/windows/wslcsession/WSLCContainer.cpp index ca447556..ed248d15 100644 --- a/src/windows/wslcsession/WSLCContainer.cpp +++ b/src/windows/wslcsession/WSLCContainer.cpp @@ -474,6 +474,21 @@ WSLCContainerState DockerStateToWSLCState(ContainerState state) } } +std::string WSLCStateToEventAction(WSLCContainerState state) +{ + switch (state) + { + case WslcContainerStateRunning: + return "start"; + case WslcContainerStateExited: + return "stop"; + case WslcContainerStateDeleted: + return "destroy"; + default: + WI_ASSERT(false); + return "unknown"; + } +} std::string CleanContainerName(const std::string& name) { // Docker container names have a leading '/', strip it. @@ -814,6 +829,7 @@ WSLCContainerImpl::WSLCContainerImpl( std::vector&& ports, std::map&& labels, std::function&& onDeleted, + EventStore& eventStore, WSLCContainerState InitialState, std::int64_t CreatedAt, WSLCProcessFlags InitProcessFlags, @@ -832,6 +848,7 @@ WSLCContainerImpl::WSLCContainerImpl( m_comWrapper(wil::MakeOrThrow(wslcSession, std::move(onDeleted))), m_containerEvents(runtime.Events().RegisterContainerStateUpdates( m_id, std::bind(&WSLCContainerImpl::OnEvent, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))), + m_eventStore(eventStore), m_state(InitialState), m_createdAt(CreatedAt), m_initProcessFlags(InitProcessFlags), @@ -1224,12 +1241,34 @@ __requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::CompleteTransitio transition->Completed.SetEvent(); } +void WSLCContainerImpl::RecordEvent(std::string&& Action, std::int64_t Time, std::optional ExitCode) noexcept +try +{ + auto attributes = StripInternalLabels(m_labels); + attributes["name"] = m_name; + attributes["image"] = m_image; + + if (ExitCode.has_value()) + { + attributes["exitCode"] = std::to_string(ExitCode.value()); + } + + m_eventStore.Record("container", std::move(Action), m_id, std::move(attributes), Time); +} +CATCH_LOG() + void WSLCContainerImpl::OnEvent(ContainerEvent event, std::optional exitCode, std::int64_t eventTime) noexcept { // Either owner may disconnect the COM wrapper, so both must outlive m_lock. unique_com_disconnect comWrapper; std::shared_ptr transition; + if (event == ContainerEvent::Kill) + { + RecordEvent("kill", eventTime); + return; + } + { auto lifecycleLock = m_lifecycleLock.lock_exclusive(); auto lock = m_lock.lock_exclusive(); @@ -1396,7 +1435,7 @@ void WSLCContainerImpl::Stop(WSLCSignal Signal, LONG TimeoutSeconds, bool Kill) } } -__requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exitCode, std::optional stopTimestamp) +__requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exitCode, std::int64_t stopTime) { auto transition = m_transition; @@ -1427,7 +1466,7 @@ __requires_exclusive_lock_held(m_lock) void WSLCContainerImpl::OnStopped(int exi // Ignore duplicate or late Stop events so they do not overwrite an already committed state. if (m_state == WslcContainerStateRunning) { - CommitState(WslcContainerStateExited, stopTimestamp); + CommitState(WslcContainerStateExited, stopTime, exitCode); } std::exception_ptr transitionException; @@ -2048,11 +2087,11 @@ std::shared_ptr WSLCContainerImpl::Create( WSLCSessionRuntime& runtime, IWSLCPluginNotifier* pluginNotifier, const std::unordered_map& sessionNetworks, - std::function&& OnDeleted) + std::function&& OnDeleted, + EventStore& eventStore) { auto& virtualMachine = runtime.Vm(); auto& DockerClient = runtime.Docker(); - auto& EventTracker = runtime.Events(); const auto mounts = ConvertAndValidateMounts(containerOptions); common::docker_schema::CreateContainer request; @@ -2486,11 +2525,6 @@ std::shared_ptr WSLCContainerImpl::Create( name.c_str()); } - // Wait for the container create event to be delivered on the Docker event stream so that - // any events for objects created for the container (e.g. volumes) are delivered before we return - // from this function. - EventTracker.WaitForObjectCreated(result.Id); - // Collect the names of referenced docker named volumes so Start() can verify // they are available before running the container. std::vector namedVolumes; @@ -2509,6 +2543,7 @@ std::shared_ptr WSLCContainerImpl::Create( } auto mergedLabels = StripInternalLabels(std::move(inspectData.Config.Labels)); + const auto createdAt = wsl::windows::common::timestamp::Rfc3339ToEpoch(inspectData.Created); auto container = std::make_shared( wslcSession, @@ -2523,8 +2558,9 @@ std::shared_ptr WSLCContainerImpl::Create( std::move(mappedPorts), std::move(mergedLabels), std::move(OnDeleted), + eventStore, WslcContainerStateCreated, - wsl::windows::common::timestamp::Rfc3339ToEpoch(inspectData.Created), + createdAt, containerOptions.InitProcessOptions.Flags, containerOptions.Flags); @@ -2539,7 +2575,8 @@ std::shared_ptr WSLCContainerImpl::Open( WSLCSession& wslcSession, WSLCSessionRuntime& runtime, IWSLCPluginNotifier* pluginNotifier, - std::function&& OnDeleted) + std::function&& OnDeleted, + EventStore& eventStore) { auto& virtualMachine = runtime.Vm(); auto& DockerClient = runtime.Docker(); @@ -2611,6 +2648,7 @@ std::shared_ptr WSLCContainerImpl::Open( std::move(ports), std::move(labels), std::move(OnDeleted), + eventStore, DockerStateToWSLCState(dockerContainer.State), dockerContainer.Created, metadata.InitProcessFlags, @@ -2917,7 +2955,7 @@ __requires_exclusive_lock_held(m_lock) unique_com_disconnect WSLCContainerImpl:: return unique_com_disconnect{std::exchange(m_comWrapper, nullptr)}; } -__requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerState State, std::optional stateChangedAt) noexcept +__requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerState State, std::int64_t Time, std::optional ExitCode) noexcept { // N.B. A deleted container cannot transition back to any other state. WI_ASSERT(m_state != WslcContainerStateDeleted); @@ -2930,7 +2968,9 @@ __requires_lock_held(m_lock) void WSLCContainerImpl::CommitState(WSLCContainerSt m_state = State; m_stateGeneration++; - m_stateChangedAt = stateChangedAt.value_or(static_cast(std::time(nullptr))); + m_stateChangedAt = Time; + + RecordEvent(WSLCStateToEventAction(State), Time, ExitCode); // Keep the VM alive while this container is Running and release the hold once it leaves that // state, even when no client holds the wrapper (e.g. a detached `run -d` container). Dropping diff --git a/src/windows/wslcsession/WSLCContainer.h b/src/windows/wslcsession/WSLCContainer.h index f145aaae..75fb094a 100644 --- a/src/windows/wslcsession/WSLCContainer.h +++ b/src/windows/wslcsession/WSLCContainer.h @@ -35,6 +35,7 @@ class WSLCContainer; class WSLCSession; class WSLCSessionRuntime; class WSLCVolumes; +class EventStore; class unique_com_disconnect { @@ -85,6 +86,7 @@ public: std::vector&& ports, std::map&& labels, std::function&& OnDeleted, + EventStore& eventStore, WSLCContainerState InitialState, std::int64_t CreatedAt, WSLCProcessFlags InitProcessFlags, @@ -123,7 +125,7 @@ public: // Re-registers a stopped container's VM-scoped port allocations against the restarted VM. void RecoverPorts(const common::docker_schema::ContainerInfo& dockerContainer); - __requires_lock_held(m_lock) void CommitState(WSLCContainerState State, std::optional stateChangedAt = std::nullopt) noexcept; + __requires_lock_held(m_lock) void CommitState(WSLCContainerState State, std::int64_t Time, std::optional ExitCode = std::nullopt) noexcept; const std::string& ID() const noexcept; @@ -141,14 +143,20 @@ public: WSLCSessionRuntime& runtime, IWSLCPluginNotifier* pluginNotifier, const std::unordered_map& SessionNetworks, - std::function&& OnDeleted); + std::function&& OnDeleted, + EventStore& eventStore); static std::shared_ptr Open( const common::docker_schema::ContainerInfo& DockerContainer, WSLCSession& wslcSession, WSLCSessionRuntime& runtime, IWSLCPluginNotifier* pluginNotifier, - std::function&& OnDeleted); + std::function&& OnDeleted, + EventStore& eventStore); + + // Appends an event for this container to the session's event stream. Must be called from the Docker + // event stream thread so that recorded events keep Docker's delivery order. + void RecordEvent(std::string&& Action, std::int64_t Time, std::optional ExitCode = std::nullopt) noexcept; private: enum class TransitionKind @@ -197,7 +205,7 @@ private: __requires_exclusive_lock_held(m_lock) void ReleaseProcesses(); __requires_exclusive_lock_held(m_lock) [[nodiscard]] unique_com_disconnect PrepareDisconnectComWrapper(); - __requires_exclusive_lock_held(m_lock) void OnStopped(int exitCode, std::optional stopTimestamp); + __requires_exclusive_lock_held(m_lock) void OnStopped(int exitCode, std::int64_t stopTime); void SetExitCode(int ExitCode) noexcept; void SignalInitProcessExit() noexcept; @@ -252,6 +260,7 @@ private: std::map m_labels; Microsoft::WRL::ComPtr m_comWrapper; DockerEventTracker::EventTrackingReference m_containerEvents; + EventStore& m_eventStore; std::string m_networkMode; // Held (non-empty) exactly while the container is Running so the session's VM stays alive even diff --git a/src/windows/wslcsession/WSLCProcessControl.cpp b/src/windows/wslcsession/WSLCProcessControl.cpp index 0d78286c..cfacac0e 100644 --- a/src/windows/wslcsession/WSLCProcessControl.cpp +++ b/src/windows/wslcsession/WSLCProcessControl.cpp @@ -172,7 +172,7 @@ void DockerExecProcessControl::SetExitCode(int ExitCode) } } -void DockerExecProcessControl::OnEvent(ContainerEvent Event, std::optional ExitCode, std::int64_t /*eventTime*/) +void DockerExecProcessControl::OnEvent(ContainerEvent Event, std::optional ExitCode, std::int64_t) { if (Event == ContainerEvent::ExecDied && !m_exitEvent.is_signaled()) { diff --git a/src/windows/wslcsession/WSLCSession.cpp b/src/windows/wslcsession/WSLCSession.cpp index ee4d16fc..25a336d7 100644 --- a/src/windows/wslcsession/WSLCSession.cpp +++ b/src/windows/wslcsession/WSLCSession.cpp @@ -41,6 +41,7 @@ using wsl::windows::service::wslc::WSLCVirtualMachine; constexpr auto c_containerdSocket = "/run/containerd/containerd.sock"; constexpr auto c_storageVhdFilename = wsl::windows::wslc::DefaultStorageVhdName; constexpr uint32_t c_progressPrecision = 4; +constexpr auto c_containerCreateEventTimeout = std::chrono::seconds{60}; // Default grace period to keep an otherwise-idle VM running before tearing it down (used when the // session's IdleTimeoutSec setting is 0/unset). This avoids thrashing the VM (repeated @@ -499,6 +500,9 @@ try m_runtime.Initialize(m_vmFactoryGitCookie, m_git, &m_settings, idleGracePeriod, std::move(sessionContext), std::move(hooks)); + m_containerEventTracking = m_runtime.Events().RegisterContainerCreate( + std::bind(&WSLCSession::OnContainerCreated, this, std::placeholders::_1, std::placeholders::_2)); + return S_OK; } CATCH_RETURN() @@ -2312,7 +2316,9 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio try { - std::scoped_lock lock(m_containersLock, m_networksLock); + std::unique_lock containersLock{m_containersLock}; + WaitForConflictingCreateToComplete(containersLock); + std::unique_lock networksLock{m_networksLock}; // Generate a unique container name if the user didn't provide one. std::string containerName; @@ -2350,13 +2356,24 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio m_runtime, m_pluginNotifier.get(), m_networks, - std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1)); + std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1), + m_eventStore); - // Key the map by Docker's container ID, which is set in the WSLCContainerImpl constructor and stable for its lifetime. - auto [it, inserted] = m_containers.emplace(container->ID(), std::move(container)); - WI_ASSERT(inserted); + auto pendingCreate = StartPendingCreate(container); - it->second->CopyTo(Container); + containersLock.unlock(); + networksLock.unlock(); + + // m_pendingCreate is published under m_containersLock before the event thread can observe it, so + // OnContainerCreated() is guaranteed to complete this create unless the session tears down first. + WaitForPendingCreateCompletion(pendingCreate); + + if (pendingCreate->Exception) + { + std::rethrow_exception(pendingCreate->Exception); + } + + container->CopyTo(Container); } catch (const DockerHTTPException& e) { @@ -2372,6 +2389,97 @@ void WSLCSession::CreateContainerImpl(const WSLCContainerOptions* containerOptio } } +__requires_lock_held(m_containersLock) std::shared_ptr WSLCSession::StartPendingCreate(std::shared_ptr Container) +{ + WI_ASSERT(!m_pendingCreate); + + m_pendingCreate = std::make_shared(); + m_pendingCreate->Container = std::move(Container); + + return m_pendingCreate; +} + +void WSLCSession::WaitForPendingCreateCompletion(const std::shared_ptr& PendingCreate) +{ + auto io = CreateIOContext(); + io.AddHandle(std::make_unique(PendingCreate->Completed.get())); + + try + { + io.Run(c_containerCreateEventTimeout); + } + catch (...) + { + if (wil::ResultFromCaughtException() != HRESULT_FROM_WIN32(ERROR_TIMEOUT)) + { + throw; + } + + // Fail this create rather than leaving m_pendingCreate set, which would wedge every later one. + // Any container docker did manage to create is left behind; the same broken event stream makes + // deleting it unreliable, and a late create event is ignored once m_pendingCreate is cleared. + std::lock_guard containersLock{m_containersLock}; + if (m_pendingCreate == PendingCreate) + { + CompletePendingCreate(PendingCreate, std::current_exception()); + } + } + + WI_ASSERT(PendingCreate->Completed.is_signaled()); +} + +__requires_lock_held(m_containersLock) void WSLCSession::CompletePendingCreate( + const std::shared_ptr& PendingCreate, std::exception_ptr Exception) noexcept +{ + WI_ASSERT(m_pendingCreate == PendingCreate); + PendingCreate->Exception = std::move(Exception); + m_pendingCreate.reset(); + PendingCreate->Completed.SetEvent(); +} + +void WSLCSession::WaitForConflictingCreateToComplete(std::unique_lock& ContainersLock) +{ + while (m_pendingCreate) + { + auto pendingCreate = m_pendingCreate; + ContainersLock.unlock(); + + WaitForPendingCreateCompletion(pendingCreate); + + ContainersLock.lock(); + } +} + +void WSLCSession::OnContainerCreated(const std::string& ContainerId, std::int64_t Time) noexcept +try +{ + std::lock_guard containersLock{m_containersLock}; + + // Containers created behind our back (BuildKit, for instance) have no pending create to match. + if (!m_pendingCreate || m_pendingCreate->Container->ID() != ContainerId) + { + return; + } + + auto pendingCreate = m_pendingCreate; + std::exception_ptr exception; + + try + { + // Key the map by Docker's container ID, which is set in the WSLCContainerImpl constructor and stable for its lifetime. + WI_VERIFY(m_containers.emplace(ContainerId, pendingCreate->Container).second); + pendingCreate->Container->RecordEvent("create", Time); + } + catch (...) + { + // Hand the failure to the waiting create rather than letting it return a container the session isn't tracking. + exception = std::current_exception(); + } + + CompletePendingCreate(pendingCreate, std::move(exception)); +} +CATCH_LOG() + HRESULT WSLCSession::OpenContainer(LPCSTR Id, IWSLCContainer** Container) try { @@ -3342,6 +3450,9 @@ try if (!m_sessionTerminatingEvent.is_signaled()) { m_sessionTerminatingEvent.SetEvent(); + + // Wake any readers parked in an event stream so they abort instead of waiting forever. + m_eventStore.OnSessionTerminating(); } // Cancel any pending IO on user-provided handles to unblock operations @@ -3907,6 +4018,23 @@ try } CATCH_RETURN(); +HRESULT WSLCSession::GetEvents(LONGLONG SinceTime, LONGLONG UntilTime, const WSLCFilter* Filters, ULONG FiltersCount, IWSLCEventStream** Stream) +try +{ + WSLCExecutionContext context(this); + + RETURN_HR_IF_NULL(E_POINTER, Stream); + + *Stream = nullptr; + + auto filters = wsl::windows::common::wslutil::ParseKeyMultiValuePairs(Filters, FiltersCount); + auto stream = m_eventStore.CreateStream(Microsoft::WRL::ComPtr{this}, SinceTime, UntilTime, std::move(filters)); + + *Stream = stream.Detach(); + return S_OK; +} +CATCH_RETURN(); + void WSLCSession::RecoverExistingContainers() { WI_ASSERT(m_runtime.HasDocker()); @@ -3940,7 +4068,7 @@ void WSLCSession::RecoverExistingContainers() try { auto container = WSLCContainerImpl::Open( - dockerContainer, *this, m_runtime, m_pluginNotifier.get(), std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1)); + dockerContainer, *this, m_runtime, m_pluginNotifier.get(), std::bind(&WSLCSession::OnContainerDeleted, this, std::placeholders::_1), m_eventStore); auto [it, inserted] = m_containers.emplace(container->ID(), std::move(container)); WI_ASSERT(inserted); diff --git a/src/windows/wslcsession/WSLCSession.h b/src/windows/wslcsession/WSLCSession.h index ae1bb5e3..0499b3e2 100644 --- a/src/windows/wslcsession/WSLCSession.h +++ b/src/windows/wslcsession/WSLCSession.h @@ -24,6 +24,7 @@ Abstract: #include "WSLCNetworkMetadata.h" #include "DockerEventTracker.h" #include "DockerHTTPClient.h" +#include "EventStore.h" #include "IORelay.h" #include #include @@ -114,6 +115,15 @@ public: IFACEMETHOD(GetTerminationEvent)(_Out_ HANDLE* Event) override; IFACEMETHOD(GetTerminationReason)(_Out_ WSLCVirtualMachineTerminationReason* Reason, _Out_ LPWSTR* Details) override; + // Event streaming. Opens a stream object that yields matching events one at a time via + // IWSLCEventStream::GetNext. + IFACEMETHOD(GetEvents)( + _In_ LONGLONG SinceTime, + _In_ LONGLONG UntilTime, + _In_reads_opt_(FiltersCount) const WSLCFilter* Filters, + _In_ ULONG FiltersCount, + _Outptr_ IWSLCEventStream** Stream) override; + // Image management. IFACEMETHOD(PullImage)( _In_ LPCSTR Image, @@ -262,11 +272,6 @@ public: UserCOMCallback RegisterUserCOMCallback(); void UnregisterUserCOMCallback(DWORD ThreadId); - HANDLE SessionTerminatingEvent() const noexcept - { - return m_sessionTerminatingEvent.get(); - } - ULONG Id() const noexcept { return m_id; @@ -319,6 +324,30 @@ private: void CreateContainerImpl(const WSLCContainerOptions* Options, IWSLCContainer** Container); + // A create RPC hands the container off to the Docker event stream thread here and waits, so the + // container is committed and its create event recorded from that thread. Recording it from the RPC + // thread instead would let another container's event be recorded first and regress the event + // stream's timestamps. + struct PendingContainerCreate + { + wil::unique_event Completed{wil::EventOptions::ManualReset}; + std::shared_ptr Container; + std::exception_ptr Exception; + }; + + __requires_lock_held(m_containersLock) std::shared_ptr StartPendingCreate(std::shared_ptr Container); + + __requires_lock_held(m_containersLock) void CompletePendingCreate( + const std::shared_ptr& PendingCreate, std::exception_ptr Exception) noexcept; + + void WaitForPendingCreateCompletion(const std::shared_ptr& PendingCreate); + + // Returns with the lock held once no create is in flight. Only one fits: the slot is unkeyed, + // because the container ID isn't known until Docker assigns it. + void WaitForConflictingCreateToComplete(std::unique_lock& ContainersLock); + + void OnContainerCreated(const std::string& ContainerId, std::int64_t Time) noexcept; + void ConfigureStorage(const WSLCSessionInitSettings& Settings, PSID UserSid); void Ext4Format(const std::string& Device); @@ -388,6 +417,14 @@ private: wil::com_ptr m_pluginNotifier; + // Bounded in-memory ring of recent lifecycle events, shared by all event-stream subscribers. + EventStore m_eventStore; + + __guarded_by(m_containersLock) std::shared_ptr m_pendingCreate; + + // N.B. Declared after everything OnContainerCreated() touches so the callback is unregistered first. + DockerEventTracker::EventTrackingReference m_containerEventTracking; + // User-provided handles that the session is currently doing IO on. std::mutex m_userHandlesLock; __guarded_by(m_userHandlesLock) std::vector m_userHandles; diff --git a/test/windows/WSLCTests.cpp b/test/windows/WSLCTests.cpp index f43857f6..6e2b0dd4 100644 --- a/test/windows/WSLCTests.cpp +++ b/test/windows/WSLCTests.cpp @@ -27,6 +27,7 @@ Abstract: #include "WSLCSessionDefaults.h" #include +using namespace std::chrono; using namespace std::literals::chrono_literals; using namespace wsl::windows::common::registry; using wsl::windows::common::ClientRunningWSLCProcess; @@ -6857,6 +6858,294 @@ class WSLCTests } } + WSLC_TEST_METHOD(EventStream) + { + constexpr auto c_containerName = "wslc-test-events"; + constexpr auto c_imageName = "debian:latest"; + constexpr auto c_labelKey = "event-label"; + constexpr auto c_labelValue = "event-value"; + const auto expectedExitCode = std::to_string(128 + WSLCSignalSIGKILL); + + auto now = [] { return duration_cast(system_clock::now().time_since_epoch()).count(); }; + + // Drains a bounded event stream to completion (GetNext returns WSLC_E_EVENT_STREAM_FINISHED + // once the until-time has passed and the backlog is exhausted), parsing each event's JSON. + auto drain = [](IWSLCEventStream* stream) { + std::vector events; + + wil::unique_cotaskmem_ansistring eventJson; + HRESULT result; + while (SUCCEEDED(result = stream->GetNext(&eventJson))) + { + events.push_back(wsl::shared::FromJson(eventJson.get())); + } + + VERIFY_ARE_EQUAL(WSLC_E_EVENT_STREAM_FINISHED, result); + return events; + }; + + // Verifies the given events match the expected actions in order for a given actor. + auto verifyEvents = [&](const std::vector& events, + const std::string& actorId, + const std::vector& expectedActions) { + VERIFY_ARE_EQUAL(events.size(), expectedActions.size()); + + for (size_t i = 0; i < expectedActions.size(); ++i) + { + const auto& action = expectedActions[i]; + const auto& event = events[i]; + + VERIFY_ARE_EQUAL(action, event.Action); + VERIFY_ARE_EQUAL(actorId, event.Actor.ID); + VERIFY_ARE_EQUAL(c_containerName, event.Actor.Attributes.at("name")); + VERIFY_ARE_EQUAL(c_imageName, event.Actor.Attributes.at("image")); + VERIFY_ARE_EQUAL(c_labelValue, event.Actor.Attributes.at(c_labelKey)); + VERIFY_IS_FALSE(event.Actor.Attributes.contains("com.microsoft.wsl.container.metadata")); + + if (action == "stop") + { + VERIFY_ARE_EQUAL(expectedExitCode, event.Actor.Attributes.at("exitCode")); + } + else + { + VERIFY_IS_FALSE(event.Actor.Attributes.contains("exitCode")); + } + } + }; + + // Run a container through its create/start/kill/stop lifecycle inside a bounded time window. + const LONGLONG since = now(); + std::string id; + { + WSLCContainerLauncher launcher(c_imageName, c_containerName, {"sleep", "99999"}); + launcher.AddLabel(c_labelKey, c_labelValue); + auto container = launcher.Launch(*m_defaultSession); + id = container.Id(); + + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateRunning); + + // Kill (rather than Stop) so Docker emits a 'kill' event ahead of the 'die' that stops it. + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGKILL)); + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateExited); + } + + const LONGLONG until = now() + 1; + std::vector lifecycleEvents; + + // The container's create, start, kill, stop, then destroy events are reported in order, each carrying + // the container's 64-hex id as the actor. + { + WSLCFilter filter{"container", id.c_str()}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream)); + + lifecycleEvents = drain(stream.get()); + verifyEvents(lifecycleEvents, id, {"create", "start", "kill", "stop", "destroy"}); + + // The whole lifecycle falls inside the requested window. + VERIFY_IS_TRUE(lifecycleEvents[0].time >= since); + VERIFY_IS_TRUE(lifecycleEvents[4].time < until); + } + + // Each lifecycle action is independently selectable: an 'event=' filter, AND'd with + // the container filter, returns exactly that one event out of the five recorded above. + auto verifyEventFilter = [&](const char* action) { + WSLCFilter filters[]{{"container", id.c_str()}, {"event", action}}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, filters, ARRAYSIZE(filters), &stream)); + + verifyEvents(drain(stream.get()), id, {action}); + }; + + verifyEventFilter("create"); + verifyEventFilter("start"); + verifyEventFilter("kill"); + verifyEventFilter("stop"); + verifyEventFilter("destroy"); + + // Values sharing a filter key are OR'd. + { + WSLCFilter filters[]{{"container", id.c_str()}, {"event", "create"}, {"event", "destroy"}}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, filters, ARRAYSIZE(filters), &stream)); + + verifyEvents(drain(stream.get()), id, {"create", "destroy"}); + } + + // Image events are not recorded yet, so a 'type=image' filter excludes the container's + // events and leaves the stream empty. + { + WSLCFilter filter{"type", "image"}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream)); + + VERIFY_IS_TRUE(drain(stream.get()).empty()); + } + + // An unmatched container id yields an empty stream, and GetNext validates its out-pointer. + { + WSLCFilter filter{"container", "0000000000000000000000000000000000000000000000000000000000000000"}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, until, &filter, 1, &stream)); + + VERIFY_IS_TRUE(drain(stream.get()).empty()); + } + + // A since-time later than a non-zero until-time describes a backwards window and is rejected. + { + wil::com_ptr stream; + VERIFY_ARE_EQUAL(E_INVALIDARG, m_defaultSession->GetEvents(since + 1, since, nullptr, 0, &stream)); + ValidateCOMErrorMessage(wsl::shared::Localization::MessageWslcEventsInvalidTimeWindow(since + 1, since)); + } + } + + WSLC_TEST_METHOD(EventStreamReportsLostEvents) + { + // One more than the store's ring capacity, so the reader's next slot is guaranteed evicted. + constexpr size_t c_signalsToEvictReader = 257; + + WSLCContainerLauncher launcher("debian:latest", "wslc-test-event-stream-overrun", {"sleep", "99999"}); + auto container = launcher.Launch(*m_defaultSession); + const auto id = container.Id(); + + WSLCFilter filter{"container", id.c_str()}; + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(0, 0, &filter, 1, &stream)); + + // Read one event to place the reader's cursor inside the ring. + wil::unique_cotaskmem_ansistring eventJson; + VERIFY_SUCCEEDED(stream->GetNext(&eventJson)); + const auto firstEvent = wsl::shared::FromJson(eventJson.get()); + VERIFY_ARE_EQUAL("create", firstEvent.Action); + VERIFY_ARE_EQUAL(id, firstEvent.Actor.ID); + + // Docker emits a 'kill' event per signal. SIGWINCH is ignored by an unhandling init process, so the + // container keeps running and each signal costs only one event. + for (size_t i = 0; i < c_signalsToEvictReader; ++i) + { + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH)); + } + + // Stopping waits for the 'die' event, which Docker delivers after every preceding 'kill'. Without this + // barrier the reader could be checked before the ring has overrun it. + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGKILL)); + VERIFY_ARE_EQUAL(container.State(), WslcContainerStateExited); + + VERIFY_ARE_EQUAL(WSLC_E_EVENTS_LOST, stream->GetNext(&eventJson)); + + // Reporting the gap resyncs the reader, so it resumes from the oldest event still buffered. + VERIFY_SUCCEEDED(stream->GetNext(&eventJson)); + const auto resumedEvent = wsl::shared::FromJson(eventJson.get()); + VERIFY_ARE_EQUAL("kill", resumedEvent.Action); + VERIFY_ARE_EQUAL(id, resumedEvent.Actor.ID); + } + + WSLC_TEST_METHOD(EventStreamSerializesConcurrentReaders) + { + constexpr auto c_containerName = "wslc-test-concurrent-event-readers"; + + WSLCContainerLauncher launcher("debian:latest", c_containerName, {"sleep", "99999"}); + auto container = launcher.Launch(*m_defaultSession); + const auto id = container.Id(); + + WSLCFilter filters[]{{"container", id.c_str()}, {"event", "kill"}}; + wil::com_ptr stream; + + // The window doubles as a hang guard, so it must comfortably outlast the waits below. + const LONGLONG until = duration_cast(system_clock::now().time_since_epoch()).count() + 120; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(0, until, filters, ARRAYSIZE(filters), &stream)); + + wil::unique_cotaskmem_ansistring firstEventJson; + wil::unique_cotaskmem_ansistring secondEventJson; + HRESULT firstResult{}; + HRESULT secondResult{}; + wil::unique_event firstReaderStarted{wil::EventOptions::ManualReset}; + wil::unique_event secondReaderStarted{wil::EventOptions::ManualReset}; + std::thread firstReader; + std::thread secondReader; + + auto cleanup = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&]() { + LOG_IF_FAILED(container.Get().Kill(WSLCSignalSIGWINCH)); + LOG_IF_FAILED(container.Get().Kill(WSLCSignalSIGWINCH)); + + if (firstReader.joinable()) + { + firstReader.join(); + } + + if (secondReader.joinable()) + { + secondReader.join(); + } + }); + + firstReader = std::thread([&]() { + firstReaderStarted.SetEvent(); + firstResult = stream->GetNext(&firstEventJson); + }); + VERIFY_IS_TRUE(firstReaderStarted.wait(30 * 1000)); + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(firstReader.native_handle(), 100)); + + secondReader = std::thread([&]() { + secondReaderStarted.SetEvent(); + secondResult = stream->GetNext(&secondEventJson); + }); + VERIFY_IS_TRUE(secondReaderStarted.wait(30 * 1000)); + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(secondReader.native_handle(), 100)); + + HANDLE readers[]{firstReader.native_handle(), secondReader.native_handle()}; + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH)); + + const DWORD completedReader = WaitForMultipleObjects(ARRAYSIZE(readers), readers, FALSE, 30 * 1000); + VERIFY_IS_TRUE(completedReader == WAIT_OBJECT_0 || completedReader == WAIT_OBJECT_0 + 1); + + // One event completes exactly one call; the other stays serialized until another event arrives. + const DWORD pendingReader = completedReader == WAIT_OBJECT_0 ? 1 : 0; + VERIFY_ARE_EQUAL(WAIT_TIMEOUT, WaitForSingleObject(readers[pendingReader], 100)); + + VERIFY_SUCCEEDED(container.Get().Kill(WSLCSignalSIGWINCH)); + VERIFY_ARE_EQUAL(WAIT_OBJECT_0, WaitForMultipleObjects(ARRAYSIZE(readers), readers, TRUE, 30 * 1000)); + + firstReader.join(); + secondReader.join(); + cleanup.release(); + + VERIFY_SUCCEEDED(firstResult); + VERIFY_SUCCEEDED(secondResult); + + const auto firstEvent = wsl::shared::FromJson(firstEventJson.get()); + const auto secondEvent = wsl::shared::FromJson(secondEventJson.get()); + VERIFY_ARE_EQUAL("kill", firstEvent.Action); + VERIFY_ARE_EQUAL(id, firstEvent.Actor.ID); + VERIFY_ARE_EQUAL("kill", secondEvent.Action); + VERIFY_ARE_EQUAL(id, secondEvent.Actor.ID); + } + + WSLC_TEST_METHOD(EventStreamSessionTerminationAbortsReader) + { + WSLCFilter filter{"type", "container"}; + const LONGLONG since = duration_cast(system_clock::now().time_since_epoch()).count(); + wil::com_ptr stream; + VERIFY_SUCCEEDED(m_defaultSession->GetEvents(since, 0, &filter, 1, &stream)); + + std::promise getNextResult; + std::thread readerThread([&]() { + wil::unique_cotaskmem_ansistring eventJson; + getNextResult.set_value(stream->GetNext(&eventJson)); + }); + auto threadCleanup = wil::scope_exit_log(WI_DIAGNOSTICS_INFO, [&]() { readerThread.join(); }); + + auto future = getNextResult.get_future(); + + VERIFY_SUCCEEDED(m_defaultSession->Terminate()); + auto restore = ResetTestSession(); + + // Termination wakes the parked reader; it must finish quickly and report E_ABORT. + FAIL_FAST_IF_MSG( + future.wait_for(10s) != std::future_status::ready, "event stream reader did not abort after session termination"); + VERIFY_ARE_EQUAL(E_ABORT, future.get()); + } + WSLC_TEST_METHOD(OpenContainer) { auto expectOpen = [&](const char* Id, HRESULT expectedResult = S_OK) {