Handle error messages in the watcher (#150)

* Handle error messages in the watcher

* Fix documentation

* Fix typo

Author: qmfrederik

src/KubernetesClient/KubernetesException.cs

@@ -0,0 +1,89 @@
+using k8s.Models;
+using System;
+#if !NETSTANDARD1_4
+using System.Runtime.Serialization;
+#endif
+
+namespace k8s
+{
+    /// <summary>
+    /// Represents an error message returned by the Kubernetes API server.
+    /// </summary>
+    public class KubernetesException : Exception
+    {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="KubernetesException"/> class.
+        /// </summary>
+        public KubernetesException()
+        {
+        }
+
+        /// <summary>
+        /// Initializes a ne winstance of the <see cref="KubernetesException"/> class using
+        /// the data from a <see cref="V1Status"/> object.
+        /// </summary>
+        /// <param name="status">
+        /// A status message which triggered this exception to be thrown.
+        /// </param>
+        public KubernetesException(V1Status status)
+            : this(status?.Message)
+        {
+            this.Status = status;
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="KubernetesException"/> class with an error message.
+        /// </summary>
+        /// <param name="message">
+        /// The error message that explains the reason for the exception.
+        /// </param>
+        public KubernetesException(string message)
+            : base(message)
+        {
+        }
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="KubernetesException"/> class with a specified error
+        /// message and a reference to the inner exception that is the cause of this exception.
+        /// </summary>
+        /// <param name="message">
+        /// The error message that explains the reason for the exception.
+        /// </param>
+        /// <param name="innerException">
+        /// The exception that is the cause of the current exception, or <see langword="null"/>
+        /// if no inner exception is specified.
+        /// </param>
+        public KubernetesException(string message, Exception innerException)
+            : base(message, innerException)
+        {
+        }
+
+#if !NETSTANDARD1_4
+        /// <summary>
+        /// Initializes a new instance of the <see cref="KubernetesException"/> class with serialized data.
+        /// </summary>
+        /// <param name="info">
+        /// The <see cref="SerializationInfo"/> that holds the serialized
+        /// object data about the exception being thrown.
+        /// </param>
+        /// <param name="context">
+        /// The <see cref="StreamingContext"/> that contains contextual information
+        /// about the source or destination.
+        /// </param>
+        protected KubernetesException(SerializationInfo info, StreamingContext context)
+            : base(info, context)
+        {
+        }
+#endif
+
+        /// <summary>
+        /// Gets, when this exception was raised because of a Kubernetes status message, the underlying
+        /// Kubernetes status message.
+        /// </summary>
+        public V1Status Status
+        {
+            get;
+            private set;
+        }
+    }
+}

src/KubernetesClient/KubernetesObject.cs

@@ -0,0 +1,36 @@
+using Newtonsoft.Json;
+
+namespace k8s
+{
+    /// <summary>
+    /// Represents a generic Kubernetes object.
+    /// </summary>
+    /// <remarks>
+    /// You can use the <see cref="KubernetesObject"/> if you receive JSON from a Kubernetes API server but
+    /// are unsure which object the API server is about to return. You can parse the JSON as a <see cref="KubernetesObject"/>
+    /// and use the <see cref="ApiVersion"/> and <see cref="Kind"/> properties to get basic metadata about any Kubernetes object.
+    /// You can then 
+    /// </remarks>
+    public class KubernetesObject
+    {
+        /// <summary>
+        /// Gets or sets aPIVersion defines the versioned schema of this
+        /// representation of an object. Servers should convert recognized
+        /// schemas to the latest internal value, and may reject unrecognized
+        /// values. More info:
+        /// https://git.k8s.io/community/contributors/devel/api-conventions.md#resources
+        /// </summary>
+        [JsonProperty(PropertyName = "apiVersion")]
+        public string ApiVersion { get; set; }
+
+        /// <summary>
+        /// Gets or sets kind is a string value representing the REST resource
+        /// this object represents. Servers may infer this from the endpoint
+        /// the client submits requests to. Cannot be updated. In CamelCase.
+        /// More info:
+        /// https://git.k8s.io/community/contributors/devel/api-conventions.md#types-kinds
+        /// </summary>
+        [JsonProperty(PropertyName = "kind")]
+        public string Kind { get; set; }
+    }
+}

src/KubernetesClient/Watcher.cs

@@ -4,6 +4,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using k8s.Exceptions;
+using k8s.Models;
 using Microsoft.Rest;
 using Microsoft.Rest.Serialization;
 
@@ -30,7 +31,7 @@ public class Watcher<T> : IDisposable
         private readonly CancellationTokenSource _cts;
         private readonly StreamReader _streamReader;
 
-        internal Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, Action<Exception> onError)
+        public Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, Action<Exception> onError)
         {
             _streamReader = streamReader;
             OnEvent += onEvent;
@@ -56,9 +57,20 @@ internal Watcher(StreamReader streamReader, Action<WatchEventType, T> onEvent, A
                         var line = await streamReader.ReadLineAsync();
 
                         try
-                        {
-                            var @event = SafeJsonConvert.DeserializeObject<WatchEvent>(line);
-                            OnEvent?.Invoke(@event.Type, @event.Object);
+                        {
+                            var genericEvent = SafeJsonConvert.DeserializeObject<k8s.Watcher<KubernetesObject>.WatchEvent>(line);
+
+                            if (genericEvent.Object.Kind == "Status")
+                            {
+                                var statusEvent = SafeJsonConvert.DeserializeObject<k8s.Watcher<V1Status>.WatchEvent>(line);
+                                var exception = new KubernetesException(statusEvent.Object);
+                                this.OnError?.Invoke(exception);
+                            }
+                            else
+                            {
+                                var @event = SafeJsonConvert.DeserializeObject<k8s.Watcher<T>.WatchEvent>(line);
+                                this.OnEvent?.Invoke(@event.Type, @event.Object);
+                            }
                         }
                         catch (Exception e)
                         {

tests/KubernetesClient.Tests/WatcherTests.cs

@@ -0,0 +1,45 @@
+using k8s.Models;
+using System;
+using System.IO;
+using System.Text;
+using System.Threading;
+using Xunit;
+
+namespace k8s.tests
+{
+    public class WatcherTests
+    {
+        [Fact]
+        public void ReadError()
+        {
+            byte[] data = Encoding.UTF8.GetBytes("{\"type\":\"ERROR\",\"object\":{\"kind\":\"Status\",\"apiVersion\":\"v1\",\"metadata\":{},\"status\":\"Failure\",\"message\":\"too old resource version: 44982(53593)\",\"reason\":\"Gone\",\"code\":410}}");
+
+            using (MemoryStream stream = new MemoryStream(data))
+            using (StreamReader reader = new StreamReader(stream))
+            {
+                Exception recordedException = null;
+                ManualResetEvent mre = new ManualResetEvent(false);
+
+                Watcher<V1Pod> watcher = new Watcher<V1Pod>(
+                    reader,
+                    null,
+                    (exception) =>
+                    {
+                        recordedException = exception;
+                        mre.Set();
+                    });
+
+                mre.WaitOne();
+
+                Assert.NotNull(recordedException);
+
+                var k8sException = recordedException as KubernetesException;
+
+                Assert.NotNull(k8sException);
+                Assert.NotNull(k8sException.Status);
+                Assert.Equal("too old resource version: 44982(53593)", k8sException.Message);
+                Assert.Equal("too old resource version: 44982(53593)", k8sException.Status.Message);
+            }
+        }
+    }
+}