Wslc events

localization/strings/en-US/Resources.resw

@@ -2291,6 +2291,10 @@ For privacy information about this product please visit https://aka.ms/privacy.<
     <value>Invalid name: '{}'</value>
     <comment>{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated</comment>
   </data>
+  <data name = "MessageWslcEventsInvalidTimeWindow" xml:space = "preserve" >
+    <value>`since` time ({}) cannot be after `until` time ({})</value>
+    <comment>{Locked="`since`"}{Locked="`until`"}{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated</comment>
+  </data>
   <data name = "MessagePathNotAbsolute" xml:space = "preserve" >
     <value>Path is not absolute: '{}'</value>
     <comment>{FixedPlaceholder="{}"}Command line arguments, file names and string inserts should not be translated</comment>

msipackage/package.wix.in

@@ -308,6 +308,14 @@
                     </RegistryKey>
                 </RegistryKey>
 
+                <!-- IWSLCEventStream-->
+                <RegistryKey Root="HKCR" Key="Interface\{7EC66D3B-D098-4D48-B69E-69166F6C4745}">
+                    <RegistryValue Value="IWSLCEventStream" Type="string" />
+                    <RegistryKey Key="ProxyStubClsid32">
+                        <RegistryValue Value="{4EA0C6DD-E9FF-48E7-994E-13A31D10DC60}" Type="string" />
+                    </RegistryKey>
+                </RegistryKey>
+
                 <!-- ICrashDumpCallback-->
                 <RegistryKey Root="HKCR" Key="Interface\{8C5A7B14-9D26-4FAE-AB31-7E5BC23F4801}">
                     <RegistryValue Value="ICrashDumpCallback" Type="string" />

src/windows/WslcSDK/wslcsdk.h

@@ -43,6 +43,8 @@ EXTERN_C_START
 #define WSLC_E_REGISTRY_BLOCKED_BY_POLICY MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 13) /* 0x8004060D */
 #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_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16)                /* 0x80040610 */
+#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17)      /* 0x80040611 */
 
 // Session values
 #define WSLC_SESSION_OPTIONS_SIZE 72

src/windows/inc/wslc_schema.h

@@ -200,4 +200,22 @@ struct Network
     NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(Network, Id, Name, Driver, Scope, Internal, IPAM, Labels);
 };
 
+struct EventActor
+{
+    std::string ID;
+    std::map<std::string, std::string> Attributes;
+
+    NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(EventActor, ID, Attributes);
+};
+
+struct Event
+{
+    std::string Type;
+    std::string Action;
+    EventActor Actor;
+    uint64_t time{};
+
+    NLOHMANN_DEFINE_TYPE_INTRUSIVE_WITH_DEFAULT(Event, Type, Action, Actor, time);
+};
+
 } // namespace wsl::windows::common::wslc_schema

src/windows/service/inc/wslc.idl

@@ -561,6 +561,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
 {
@@ -597,6 +609,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. 0 means
+    // unbounded on that end.
+    // Filters are key/value pair. Values sharing a key are OR'd, distinct keys are AND'd.
+    HRESULT GetEvents(
+        [in] ULONGLONG SinceTime,
+        [in] ULONGLONG 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);
@@ -749,3 +773,5 @@ cpp_quote("#define WSLC_E_CONTAINER_DISABLED MAKE_HRESULT(SEVERITY_ERROR, FACILI
 cpp_quote("#define WSLC_E_REGISTRY_BLOCKED_BY_POLICY MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 13) /* 0x8004060D */")
 cpp_quote("#define WSLC_E_VOLUME_NOT_AVAILABLE MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 14) /* 0x8004060E */")
 cpp_quote("#define WSLC_E_SESSION_NOT_FOUND MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 15) /* 0x8004060F */")
+cpp_quote("#define WSLC_E_EVENTS_LOST MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 16) /* 0x80040610 */")
+cpp_quote("#define WSLC_E_EVENT_STREAM_FINISHED MAKE_HRESULT(SEVERITY_ERROR, FACILITY_ITF, WSLC_E_BASE + 17) /* 0x80040611 */")

src/windows/wslcsession/CMakeLists.txt

@@ -25,6 +25,7 @@ set(SOURCES
     # Supporting classes
     DockerEventTracker.cpp
     DockerHTTPClient.cpp
+    EventStore.cpp
     IORelay.cpp
     OptionParser.cpp
     ServiceProcessLauncher.cpp
@@ -34,6 +35,7 @@ set(SOURCES
 set(HEADERS
     DockerEventTracker.h
     DockerHTTPClient.h
+    EventStore.h
     IORelay.h
     OptionParser.h
     ServiceProcessLauncher.h

src/windows/wslcsession/DockerEventTracker.cpp

@@ -153,7 +153,11 @@ void DockerEventTracker::OnEvent(const std::string_view& event)
 void DockerEventTracker::OnContainerEvent(const nlohmann::json& parsed, const std::string& action, std::uint64_t eventTime)
 {
     static std::map<std::string, ContainerEvent> events{
-        {"start", ContainerEvent::Start}, {"die", ContainerEvent::Stop}, {"destroy", ContainerEvent::Destroy}, {"exec_die", ContainerEvent::ExecDied}};
+        {"start", ContainerEvent::Start},
+        {"die", ContainerEvent::Stop},
+        {"kill", ContainerEvent::Kill},
+        {"destroy", ContainerEvent::Destroy},
+        {"exec_die", ContainerEvent::ExecDied}};
 
     auto actor = parsed.find("Actor");
     THROW_HR_IF_MSG(E_INVALIDARG, actor == parsed.end(), "Missing Actor in container event");

src/windows/wslcsession/DockerEventTracker.h

@@ -29,7 +29,8 @@ enum class ContainerEvent
     Stop,
     Exit,
     Destroy,
-    ExecDied
+    ExecDied,
+    Kill
 };
 
 enum class VolumeEvent

src/windows/wslcsession/EventStore.cpp

@@ -0,0 +1,252 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+
+#include "precomp.h"
+#include "EventStore.h"
+#include "WSLCSession.h"
+#include "WSLCExecutionContext.h"
+#include <chrono>
+#include "wslc_schema.h"
+
+using wsl::shared::Localization;
+
+namespace wsl::windows::service::wslc {
+
+namespace {
+
+    // Normalizes a seconds-since-epoch window bound from the COM boundary, treating zero as "unset".
+    std::optional<uint64_t> ToTimeBound(uint64_t TimeSeconds)
+    {
+        if (TimeSeconds == 0)
+        {
+            return std::nullopt;
+        }
+
+        return TimeSeconds;
+    }
+
+} // namespace
+
+void EventStore::Append(wsl::windows::common::wslc_schema::Event Event)
+{
+    std::lock_guard lock(m_lock);
+
+    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::optional<uint64_t> TimeSeconds) noexcept
+try
+{
+    wsl::windows::common::wslc_schema::Event event;
+    event.Type = std::move(Type);
+    event.Action = std::move(Action);
+    event.Actor.ID = ActorId;
+
+    event.time = TimeSeconds.value_or(static_cast<uint64_t>(
+        std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch()).count()));
+
+    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<std::string, std::vector<std::string>>& 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<IWSLCEventStream> EventStore::CreateStream(
+    Microsoft::WRL::ComPtr<WSLCSession> Session, uint64_t SinceTime, uint64_t UntilTime, std::map<std::string, std::vector<std::string>> Filters)
+{
+    // A non-zero until earlier than since describes a backwards, empty window.
+    THROW_HR_WITH_USER_ERROR_IF(
+        E_INVALIDARG, Localization::MessageWslcEventsInvalidTimeWindow(SinceTime, UntilTime), UntilTime != 0 && SinceTime > UntilTime);
+
+    Microsoft::WRL::ComPtr<EventStream> stream;
+    THROW_IF_FAILED(Microsoft::WRL::MakeAndInitialize<EventStream>(&stream, std::move(Session), this, SinceTime, UntilTime, std::move(Filters)));
+
+    return stream;
+}
+
+std::optional<wsl::windows::common::wslc_schema::Event> 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<std::mutex>& Lock, uint64_t SequenceNumber, std::optional<uint64_t> 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())
+    {
+        // Until is Unix seconds and system_clock shares that epoch, so it doubles as the wait deadline.
+        const std::chrono::sys_seconds deadline{std::chrono::seconds{static_cast<int64_t>(Until.value())}};
+        if (!m_updated.wait_until(Lock, deadline, ready))
+        {
+            return false;
+        }
+    }
+    else
+    {
+        m_updated.wait(Lock, ready);
+    }
+
+    THROW_HR_IF(E_ABORT, m_terminating);
+    return true;
+}
+
+std::optional<wsl::windows::common::wslc_schema::Event> EventStore::Get(
+    std::optional<uint64_t>& SequenceNumber,
+    std::optional<uint64_t> Since,
+    std::optional<uint64_t> Until,
+    const std::map<std::string, std::vector<std::string>>& 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();
+
+        // An event past the until-bound closes the window: the stream is finished.
+        if (Until.has_value() && event.time > Until.value())
+        {
+            return std::nullopt;
+        }
+
+        // Advance past this event so the next read resumes at the following one.
+        SequenceNumber.value()++;
+
+        // 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() || event.time >= 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<WSLCSession> Session,
+    EventStore* Store,
+    uint64_t SinceTime,
+    uint64_t UntilTime,
+    std::map<std::string, std::vector<std::string>> 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;
+
+    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<wil::unique_cotaskmem_ansistring>(wsl::shared::ToJson(event.value()).c_str()).release();
+    return S_OK;
+}
+CATCH_RETURN();
+
+} // namespace wsl::windows::service::wslc