Events monitoring stream does not seems to always include CR or LF character at the end of every messages

[simardst]

Output of dotnet --info:

.NET Core SDK (reflecting any global.json):
 Version:   3.1.101
 Commit:    b377529961

Runtime Environment:
 OS Name:     Windows
 OS Version:  10.0.18362
 OS Platform: Windows
 RID:         win10-x64
 Base Path:   C:\Program Files\dotnet\sdk\3.1.101\

Host (useful for support):
  Version: 3.1.1
  Commit:  a1388f194c

What version of Docker.DotNet?:

3.125.2

Docker Version:

Client: Docker Engine - Community
 Version:           19.03.5
 API version:       1.40
 Go version:        go1.12.12
 Git commit:        633a0ea
 Built:             Wed Nov 13 07:22:37 2019
 OS/Arch:           windows/amd64
 Experimental:      false

Server: Docker Engine - Community
 Engine:
  Version:          19.03.5
  API version:      1.40 (minimum version 1.12)
  Go version:       go1.12.12
  Git commit:       633a0ea
  Built:            Wed Nov 13 07:29:19 2019
  OS/Arch:          linux/amd64
  Experimental:     false
 containerd:
  Version:          v1.2.10
  GitCommit:        b34a5c8af56e510852c35414db4c1f4fa6172339
 runc:
  Version:          1.0.0-rc8+dev
  GitCommit:        3e425f80a8c931f88e6d94a8c831b9d5aa481657
 docker-init:
  Version:          0.18.0
  GitCommit:        fec3683

Docker Info:

Client:                                                                                                   
 Debug Mode: false                                                                                        
                                                                                                          
Server:                                                                                                   
 Containers: 1                                                                                            
  Running: 1                                                                                              
  Paused: 0                                                                                               
  Stopped: 0                                                                                              
 Images: 10                                                                                               
 Server Version: 19.03.5                                                                                  
 Storage Driver: overlay2                                                                                 
  Backing Filesystem: extfs                                                                               
  Supports d_type: true                                                                                   
  Native Overlay Diff: true                                                                               
 Logging Driver: json-file                                                                                
 Cgroup Driver: cgroupfs                                                                                  
 Plugins:                                                                                                 
  Volume: local                                                                                           
  Network: bridge host ipvlan macvlan null overlay                                                        
  Log: awslogs fluentd gcplogs gelf journald json-file local logentries splunk syslog                     
 Swarm: inactive                                                                                          
 Runtimes: runc                                                                                           
 Default Runtime: runc                                                                                    
 Init Binary: docker-init                                                                                 
 containerd version: b34a5c8af56e510852c35414db4c1f4fa6172339                                             
 runc version: 3e425f80a8c931f88e6d94a8c831b9d5aa481657                                                   
 init version: fec3683                                                                                    
 Security Options:                                                                                        
  seccomp                                                                                                 
   Profile: default                                                                                       
 Kernel Version: 4.19.76-linuxkit                                                                         
 Operating System: Docker Desktop                                                                         
 OSType: linux                                                                                            
 Architecture: x86_64                                                                                     
 CPUs: 2                                                                                                  
 Total Memory: 1.943GiB                                                                                   
 Name: docker-desktop                                                                                     
 ID: XESR:K73P:U465:PX4V:WDXD:HT4Y:H7K3:W7ED:7R2Y:KAEB:VF3G:WM7D                                          
 Docker Root Dir: /var/lib/docker                                                                         
 Debug Mode: true                                                                                         
  File Descriptors: 49                                                                                    
  Goroutines: 67                                                                                          
  System Time: 2020-01-31T17:25:24.631755878Z                                                             
  EventsListeners: 6                                                                                      
 Registry: https://index.docker.io/v1/                                                                    
 Labels:                                                                                                  
 Experimental: false                                                                                      
 Insecure Registries:                                                                                     
  127.0.0.0/8                                                                                             
 Live Restore Enabled: false                                                                              
 Product License: Community Engine                                                                        

Steps to reproduce the issue:

1. Create DockerClient to local engine (npipe://./pipe/docker_engine)
2. Monitor system events ()

m_dockerClient.System.MonitorEventsAsync(
    new ContainerEventsParameters(),
    new Progress<Message>(m => Console.WriteLine(JsonConvert.SerializeObject(m, Newtonsoft.Json.Formatting.Indented))));

3. Using Docker CLI, monitor events (docker events)
4. Using the docker CLI, start and stops containers
5. Messages are printed on the Docker CLI console, but not on my app console (see 2.)

What actually happened?:
The HTTP request to get the event stream is accepted (200 - OK) and the stream is correctly opened, but there is no CR or LF character to delimit the messages received. Therefore, the stream reader gets stuck (since it is doing ReadLineAsync(...) calls to read messages and parse them)

What did you expect to happen?:
Find a way to extract messages from the event stream without requiring CR or LF charcaters to detect a complete message.

A solution could be to directly parse the stream to extratc JSON objects/messages.

internal static async Task MonitorStreamForMessagesAsync<T>(Task<Stream> streamTask, DockerClient client, CancellationToken cancel, IProgress<T> progress)
{
    using (var stream = await streamTask)
    {
        // ReadLineAsync must be cancelled by closing the whole stream.
        using (cancel.Register(() => stream.Dispose()))
        {
            using (var reader = new StreamReader(stream, new UTF8Encoding(false))) 
            using (var jsonReader = new JsonTextReader(reader) { SupportMultipleContent = true })
            {
                try
                {
                    while (jsonReader.Read())
                    {
                        var prog = client.JsonSerializer.DeserializeObject<T>(jsonReader);
                        if (prog == null) continue;

                        progress.Report(prog);
                    }
                }
                catch (ObjectDisposedException)
                {
                    // The subsequent call to reader.ReadLineAsync() after cancellation
                    // will fail because we disposed the stream. Just ignore here.
                }
            }
        }
    }
}

internal class JsonSerializer
{
    public T DeserializeObject<T>(JsonReader jsonReader)
    {
        var serializer = Newtonsoft.Json.JsonSerializer.Create(this._settings);
        serializer.CheckAdditionalContent = false;
        return serializer.Deserialize<T>(jsonReader);
    }
   ...
}

Additional information:

[apples]

I'm running into this same issue.

The Docker CLI docker events command with JSON formatting seems to guarantee that each event will be on its own line (documented here), however the equivalent query to the HTTP Remote API has no such guarantee (documented here). Indeed, hitting the /events URL with curl -sN reveals that there is nothing at all separating each JSON object.

[stephen-slm]

The implementation done by @simardst does work but hangs when awaiting for another message before completing the process of the current one.

[soerenmuehlbauer]

I'm also hit by that issue. In my case (WSL) I do not receive any events. When manually applying @simardst fix I'm able to get events but sometime I do not receive a container stop immediatly. I receive this later on when another event comes in. Very annoying. For me it's really hard to understand whats going on because there is a pipestream encapsulated inside a BufferedReadStream encapsualted inside a ChunkReadStream. I will no switch to an own implementation (as I only need the events)
My Idea is to ignore any CRLF or such things and only count {} and on equal count to process this.
I debugged this a bit. I doubt this is a problem inside of BufferedReadStream or ChunkReadStream. The problem in the master version is that it uses ReadLineAsync. The problem with the version from @msdeibel is that also the newtonsoft parser waits for smth. I fixed this for now by using

internal static async Task MonitorStreamForMessagesAsync<T>(Task<Stream> streamTask, DockerClient client, CancellationToken cancel, IProgress<T> progress)
{
    using (var stream = await streamTask)
    {
        // ReadLineAsync must be cancelled by closing the whole stream.
        using (cancel.Register(() => stream.Dispose()))
        {
            var startObjectCount = 0;
            using (var reader = new StreamReader(stream, new UTF8Encoding(false)))
            {
                try
                {
                    var ch = new char[1];
                    var sb = new StringBuilder();
                    while (await reader.ReadAsync(ch, 0, 1).ConfigureAwait(false) == 1)
                    {
                        sb.Append(ch[0]);
                        switch (ch[0])
                        {
                            case '}':
                            {
                                startObjectCount--;
                                if (startObjectCount == 0)
                                {
                                    var prog = client.JsonSerializer.DeserializeObject<T>(sb.ToString());
                                    if (prog == null) continue;
                                    progress.Report(prog);
                                    sb.Clear();
                                }
                                break;
                            }
                            case '{':
                                startObjectCount++;
                                break;
                        }
                    }
                }
                catch (ObjectDisposedException)
                {
                    // The subsequent call to reader.ReadLineAsync() after cancellation
                    // will fail because we disposed the stream. Just ignore here.
                }
            }
        }
    }
}

Not fast or beautiful but it works for me at least.

[apples]

Here's the implementation that I've been using:

using Newtonsoft.Json;

public class EventActor
{
    public string Id { get; set; }
    public Dictionary<string, string> Attributes { get; set; }
}

public class Event
{
    public string Type { get; set; }
    public string Action { get; set; }
    public EventActor Actor { get; set; }
    public int Time { get; set; }
    public Int64 TimeNano { get; set; }
}

public class DockerClient
{
    public async Task MonitorEvents(
        Uri endpoint,
        int? since,
        int? until,
        Dictionary<string, Dictionary<string, bool>> filters,
        CancellationToken cancel,
        Action<Event> action)
    {
        var queryParams = new List<string>(3);
        if (since != null) { queryParams.Add($"since={since}"); }
        if (until != null) { queryParams.Add($"until={until}"); }
        if (filters != null) { queryParams.Add($"filters={JsonConvert.SerializeObject(filters)}"); }

        var queryString = String.Join("&", queryParams);

        var client = new HttpClient();
        var uri = new Uri(endpoint, $"events?{queryString}");

        var stream = await client.GetStreamAsync(uri);

        var serializer = new JsonSerializer();

        using (cancel.Register(() => stream.Dispose()))
        using (var reader = new StreamReader(stream, new UTF8Encoding(false)))
        using (var jsonReader = new JsonTextReader(reader) { SupportMultipleContent = true })
        {
            try
            {
                while (jsonReader.Read())
                {
                    var ev = serializer.Deserialize<Event>(jsonReader);
                    action(ev);
                    if (cancel.IsCancellationRequested) {
                        break;
                    }
                }
            }
            catch (ObjectDisposedException)
            {
                // Ignore reads on disposed streams
                return;
            }
        }
    }
}

[jterry75]

Has anyone looked at how the golang client handles this case?